diff --git a/include/libvirt/libvirt-stream.h b/include/libvirt/libvirt-stream.h index a5e69a1c1a30f3b836abb81a86ef3fde9c10f848..d18d4314058ad2ff785b1c2f7b33629f4891ed4b 100644 --- a/include/libvirt/libvirt-stream.h +++ b/include/libvirt/libvirt-stream.h @@ -71,9 +71,9 @@ int virStreamRecvHole(virStreamPtr, * @nbytes: size of the data array * @opaque: optional application provided data * - * The virStreamSourceFunc callback is used together - * with the virStreamSendAll function for libvirt to - * obtain the data that is to be sent. + * The virStreamSourceFunc callback is used together with + * the virStreamSendAll and virStreamSparseSendAll functions + * for libvirt to obtain the data that is to be sent. * * The callback will be invoked multiple times, * fetching data in small chunks. The application @@ -96,6 +96,65 @@ int virStreamSendAll(virStreamPtr st, virStreamSourceFunc handler, void *opaque); +/** + * virStreamSourceHoleFunc: + * @st: the stream object + * @inData: are we in data section + * @length: how long is the section we are currently in + * @opaque: optional application provided data + * + * The virStreamSourceHoleFunc callback is used together with the + * virStreamSparseSendAll function for libvirt to obtain the + * length of section stream is currently in. + * + * Moreover, upon successful return, @length should be updated + * with how many bytes are left until the current section ends + * (either data section or hole section). Also the stream is + * currently in data section, @inData should be set to a non-zero + * value and vice versa. + * + * NB: there's an implicit hole at the end of each file. If + * that's the case, @inData and @length should be both set to 0. + * + * This function should not adjust the current position within + * the file. + * + * Returns 0 on success, + * -1 upon error + */ +typedef int (*virStreamSourceHoleFunc)(virStreamPtr st, + int *inData, + long long *length, + void *opaque); + +/** + * virStreamSourceSkipFunc: + * @st: the stream object + * @length: stream hole size + * @opaque: optional application provided data + * + * This callback is used together with the virStreamSparseSendAll + * to skip holes in the underlying file as reported by + * virStreamSourceHoleFunc. + * + * The callback may be invoked multiple times as holes are found + * during processing a stream. The application should skip + * processing the hole in the stream source and then return. + * A return value of -1 at any time will abort the send operation. + * + * Returns 0 on success, + * -1 upon error. + */ +typedef int (*virStreamSourceSkipFunc)(virStreamPtr st, + long long length, + void *opaque); + +int virStreamSparseSendAll(virStreamPtr st, + virStreamSourceFunc handler, + virStreamSourceHoleFunc holeHandler, + virStreamSourceSkipFunc skipHandler, + void *opaque); + /** * virStreamSinkFunc: * diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c index 6bf4c4f29f5218319279e06d3b346f69074a9bb7..c7ca1cc47776ce65cdbba3302ee810d716544860 100644 --- a/src/libvirt-stream.c +++ b/src/libvirt-stream.c @@ -574,6 +574,165 @@ virStreamSendAll(virStreamPtr stream, } +/** + * virStreamSparseSendAll: + * @stream: pointer to the stream object + * @handler: source callback for reading data from application + * @holeHandler: source callback for determining holes + * @skipHandler: skip holes as reported by @holeHandler + * @opaque: application defined data + * + * Send the entire data stream, reading the data from the + * requested data source. This is simply a convenient alternative + * to virStreamSend, for apps that do blocking-I/O. + * + * An example using this with a hypothetical file upload + * API looks like + * + * int mysource(virStreamPtr st, char *buf, int nbytes, void *opaque) { + * int *fd = opaque; + * + * return read(*fd, buf, nbytes); + * } + * + * int myskip(virStreamPtr st, long long offset, void *opaque) { + * int *fd = opaque; + * + * return lseek(*fd, offset, SEEK_CUR) == (off_t) -1 ? -1 : 0; + * } + * + * int myindata(virStreamPtr st, int *inData, + * long long *offset, void *opaque) { + * int *fd = opaque; + * + * if (@fd in hole) { + * *inData = 0; + * *offset = holeSize; + * } else { + * *inData = 1; + * *offset = dataSize; + * } + * + * return 0; + * } + * + * virStreamPtr st = virStreamNew(conn, 0); + * int fd = open("demo.iso", O_RDONLY); + * + * virConnectUploadSparseFile(conn, st); + * if (virStreamSparseSendAll(st, + * mysource, + * myindata, + * myskip, + * &fd) < 0) { + * ...report an error ... + * goto done; + * } + * if (virStreamFinish(st) < 0) + * ...report an error... + * virStreamFree(st); + * close(fd); + * + * Note that @opaque data are shared between @handler, @holeHandler and @skipHandler. + * + * Returns 0 if all the data was successfully sent. The caller + * should invoke virStreamFinish(st) to flush the stream upon + * success and then virStreamFree. + * + * Returns -1 upon any error, with virStreamAbort() already + * having been called, so the caller need only call + * virStreamFree(). + */ +int virStreamSparseSendAll(virStreamPtr stream, + virStreamSourceFunc handler, + virStreamSourceHoleFunc holeHandler, + virStreamSourceSkipFunc skipHandler, + void *opaque) +{ + char *bytes = NULL; + size_t want = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX; + int ret = -1; + unsigned long long dataLen = 0; + + VIR_DEBUG("stream=%p handler=%p holeHandler=%p opaque=%p", + stream, handler, holeHandler, opaque); + + virResetLastError(); + + virCheckStreamReturn(stream, -1); + virCheckNonNullArgGoto(handler, cleanup); + virCheckNonNullArgGoto(holeHandler, cleanup); + virCheckNonNullArgGoto(skipHandler, cleanup); + + if (stream->flags & VIR_STREAM_NONBLOCK) { + virReportError(VIR_ERR_OPERATION_INVALID, "%s", + _("data sources cannot be used for non-blocking streams")); + goto cleanup; + } + + if (VIR_ALLOC_N(bytes, want) < 0) + goto cleanup; + + for (;;) { + int inData, got, offset = 0; + long long sectionLen; + const unsigned int skipFlags = 0; + + if (!dataLen) { + if (holeHandler(stream, &inData, §ionLen, opaque) < 0) { + virStreamAbort(stream); + goto cleanup; + } + + if (!inData && sectionLen) { + if (virStreamSendHole(stream, sectionLen, skipFlags) < 0) { + virStreamAbort(stream); + goto cleanup; + } + + if (skipHandler(stream, sectionLen, opaque) < 0) { + virReportSystemError(errno, "%s", + _("unable to skip hole")); + virStreamAbort(stream); + goto cleanup; + } + continue; + } else { + dataLen = sectionLen; + } + } + + if (want > dataLen) + want = dataLen; + + got = (handler)(stream, bytes, want, opaque); + if (got < 0) { + virStreamAbort(stream); + goto cleanup; + } + if (got == 0) + break; + while (offset < got) { + int done; + done = virStreamSend(stream, bytes + offset, got - offset); + if (done < 0) + goto cleanup; + offset += done; + dataLen -= done; + } + } + ret = 0; + + cleanup: + VIR_FREE(bytes); + + if (ret != 0) + virDispatchError(stream->conn); + + return ret; +} + + /** * virStreamRecvAll: * @stream: pointer to the stream object diff --git a/src/libvirt_public.syms b/src/libvirt_public.syms index 37fc4e2243f4165b0c114bcca2b312fddf8b4874..fac77fbea88139328e7a646fc30a1329d2b1db44 100644 --- a/src/libvirt_public.syms +++ b/src/libvirt_public.syms @@ -765,6 +765,7 @@ LIBVIRT_3.4.0 { virStreamRecvHole; virStreamSendHole; virStreamSparseRecvAll; + virStreamSparseSendAll; } LIBVIRT_3.1.0; # .... define new API here using predicted next version number ....