One big downside of using the pipe to transfer the data is that we can really transfer just bare data. No metadata can be carried through unless some formatted messages are introduced. That would be quite painful to achieve so let's use a message queue. It's fairly easy to exchange info between threads now that iohelper is no longer used. The reason why we cannot use the FD for plain files directly is that despite us setting noblock flag on the FD, any read()/write() blocks regardless (which is a show stopper since those parts of the code are run from the event loop) and poll() reports such FD as always readable/writable - even though the subsequent operation might block. The pipe is still not gone though. It is used to signal to even loop that an event occurred (e.g. data are available for reading in the queue, or vice versa). Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/util/virfdstream.c | 402 ++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 350 insertions(+), 52 deletions(-) diff --git a/src/util/virfdstream.c b/src/util/virfdstream.c index 5ce78fe58..4b42939e7 100644 --- a/src/util/virfdstream.c +++ b/src/util/virfdstream.c @@ -49,6 +49,27 @@ VIR_LOG_INIT("fdstream"); +typedef enum { + VIR_FDSTREAM_MSG_TYPE_DATA, +} virFDStreamMsgType; + +typedef struct _virFDStreamMsg virFDStreamMsg; +typedef virFDStreamMsg *virFDStreamMsgPtr; +struct _virFDStreamMsg { + virFDStreamMsgPtr next; + + virFDStreamMsgType type; + + union { + struct { + char *buf; + size_t len; + size_t offset; + } data; + } stream; +}; + + /* Tunnelled migration stream support */ typedef struct virFDStreamData virFDStreamData; typedef virFDStreamData *virFDStreamDataPtr; @@ -80,18 +101,25 @@ struct virFDStreamData { /* Thread data */ virThreadPtr thread; + virCond threadCond; int threadErr; bool threadQuit; + bool threadAbort; + bool threadDoRead; + virFDStreamMsgPtr msg; }; static virClassPtr virFDStreamDataClass; +static void virFDStreamMsgQueueFree(virFDStreamMsgPtr *queue); + static void virFDStreamDataDispose(void *obj) { virFDStreamDataPtr fdst = obj; VIR_DEBUG("obj=%p", fdst); + virFDStreamMsgQueueFree(&fdst->msg); } static int virFDStreamDataOnceInit(void) @@ -108,6 +136,89 @@ static int virFDStreamDataOnceInit(void) VIR_ONCE_GLOBAL_INIT(virFDStreamData) +static int +virFDStreamMsgQueuePush(virFDStreamDataPtr fdst, + virFDStreamMsgPtr msg, + int fd, + const char *fdname) +{ + virFDStreamMsgPtr *tmp = &fdst->msg; + char c = '1'; + + while (*tmp) + tmp = &(*tmp)->next; + + *tmp = msg; + virCondSignal(&fdst->threadCond); + + if (safewrite(fd, &c, sizeof(c)) != sizeof(c)) { + virReportSystemError(errno, + _("Unable to write to %s"), + fdname); + return -1; + } + + return 0; +} + + +static virFDStreamMsgPtr +virFDStreamMsgQueuePop(virFDStreamDataPtr fdst, + int fd, + const char *fdname) +{ + virFDStreamMsgPtr tmp = fdst->msg; + char c; + + if (tmp) { + fdst->msg = tmp->next; + tmp->next = NULL; + } + + virCondSignal(&fdst->threadCond); + + if (saferead(fd, &c, sizeof(c)) != sizeof(c)) { + virReportSystemError(errno, + _("Unable to read from %s"), + fdname); + return NULL; + } + + return tmp; +} + + +static void +virFDStreamMsgFree(virFDStreamMsgPtr msg) +{ + if (!msg) + return; + + switch (msg->type) { + case VIR_FDSTREAM_MSG_TYPE_DATA: + VIR_FREE(msg->stream.data.buf); + break; + } + + VIR_FREE(msg); +} + + +static void +virFDStreamMsgQueueFree(virFDStreamMsgPtr *queue) +{ + virFDStreamMsgPtr tmp = *queue; + + while (tmp) { + virFDStreamMsgPtr next = tmp->next; + virFDStreamMsgFree(tmp); + tmp = next; + } + + *queue = NULL; +} + + static int virFDStreamRemoveCallback(virStreamPtr stream) { virFDStreamDataPtr fdst = stream->privateData; @@ -273,6 +384,7 @@ typedef virFDStreamThreadData *virFDStreamThreadDataPtr; struct _virFDStreamThreadData { virStreamPtr st; size_t length; + bool doRead; int fdin; char *fdinname; int fdout; @@ -293,6 +405,86 @@ virFDStreamThreadDataFree(virFDStreamThreadDataPtr data) } +static ssize_t +virFDStreamThreadDoRead(virFDStreamDataPtr fdst, + const int fdin, + const int fdout, + const char *fdinname, + const char *fdoutname, + size_t buflen) +{ + virFDStreamMsgPtr msg = NULL; + char *buf = NULL; + ssize_t got; + + if (VIR_ALLOC(msg) < 0) + goto error; + + if (VIR_ALLOC_N(buf, buflen) < 0) + goto error; + + if ((got = saferead(fdin, buf, buflen)) < 0) { + virReportSystemError(errno, + _("Unable to read %s"), + fdinname); + goto error; + } + + msg->type = VIR_FDSTREAM_MSG_TYPE_DATA; + msg->stream.data.buf = buf; + msg->stream.data.len = got; + buf = NULL; + + virFDStreamMsgQueuePush(fdst, msg, fdout, fdoutname); + msg = NULL; + + return got; + + error: + VIR_FREE(buf); + virFDStreamMsgFree(msg); + return -1; +} + + +static ssize_t +virFDStreamThreadDoWrite(virFDStreamDataPtr fdst, + const int fdin, + const int fdout, + const char *fdinname, + const char *fdoutname) +{ + ssize_t got; + virFDStreamMsgPtr msg = fdst->msg; + bool pop = false; + + switch (msg->type) { + case VIR_FDSTREAM_MSG_TYPE_DATA: + got = safewrite(fdout, + msg->stream.data.buf + msg->stream.data.offset, + msg->stream.data.len - msg->stream.data.offset); + if (got < 0) { + virReportSystemError(errno, + _("Unable to write %s"), + fdoutname); + return -1; + } + + msg->stream.data.offset += got; + + pop = msg->stream.data.offset == msg->stream.data.len; + break; + } + + if (pop) { + virFDStreamMsgQueuePop(fdst, fdin, fdinname); + virFDStreamMsgFree(msg); + } + + return got; +} + + static void virFDStreamThread(void *opaque) { @@ -304,14 +496,12 @@ virFDStreamThread(void *opaque) int fdout = data->fdout; char *fdoutname = data->fdoutname; virFDStreamDataPtr fdst = st->privateData; - char *buf = NULL; + bool doRead = fdst->threadDoRead; size_t buflen = 256 * 1024; size_t total = 0; virObjectRef(fdst); - - if (VIR_ALLOC_N(buf, buflen) < 0) - goto error; + virObjectLock(fdst); while (1) { ssize_t got; @@ -323,39 +513,56 @@ virFDStreamThread(void *opaque) if (buflen == 0) break; /* End of requested data from client */ - if ((got = saferead(fdin, buf, buflen)) < 0) { - virReportSystemError(errno, - _("Unable to read %s"), - fdinname); + while (doRead == (fdst->msg != NULL) && + !fdst->threadQuit) { + if (virCondWait(&fdst->threadCond, &fdst->parent.lock)) { + virReportSystemError(errno, "%s", + _("failed to wait on condition")); + goto error; + } + } + + if (fdst->threadQuit) { + /* If stream abort was requested, quit early. */ + if (fdst->threadAbort) + goto cleanup; + + /* Otherwise flush buffers and quit gracefully. */ + if (doRead == (fdst->msg != NULL)) + break; + } + + if (doRead) + got = virFDStreamThreadDoRead(fdst, + fdin, fdout, + fdinname, fdoutname, + buflen); + else + got = virFDStreamThreadDoWrite(fdst, + fdin, fdout, + fdinname, fdoutname); + + if (got < 0) goto error; - } if (got == 0) break; total += got; - - if (safewrite(fdout, buf, got) < 0) { - virReportSystemError(errno, - _("Unable to write %s"), - fdoutname); - goto error; - } } cleanup: + fdst->threadQuit = true; + virObjectUnlock(fdst); if (!virObjectUnref(fdst)) st->privateData = NULL; VIR_FORCE_CLOSE(fdin); VIR_FORCE_CLOSE(fdout); virFDStreamThreadDataFree(data); - VIR_FREE(buf); return; error: - virObjectLock(fdst); fdst->threadErr = errno; - virObjectUnlock(fdst); goto cleanup; } @@ -368,6 +575,10 @@ virFDStreamJoinWorker(virFDStreamDataPtr fdst, if (!fdst->thread) return 0; + fdst->threadAbort = streamAbort; + fdst->threadQuit = true; + virCondSignal(&fdst->threadCond); + /* Give the thread a chance to lock the FD stream object. */ virObjectUnlock(fdst); virThreadJoin(fdst->thread); @@ -381,6 +592,7 @@ virFDStreamJoinWorker(virFDStreamDataPtr fdst, ret = 0; cleanup: VIR_FREE(fdst->thread); + virCondDestroy(&fdst->threadCond); return ret; } @@ -427,11 +639,14 @@ virFDStreamCloseInt(virStreamPtr st, bool streamAbort) fdst->abortCallbackDispatching = false; } - /* mutex locked */ - ret = VIR_CLOSE(fdst->fd); if (virFDStreamJoinWorker(fdst, streamAbort) < 0) ret = -1; + /* mutex locked */ + if ((ret = VIR_CLOSE(fdst->fd)) < 0) + virReportSystemError(errno, "%s", + _("Unable to close")); + st->privateData = NULL; /* call the internal stream closing callback */ @@ -468,7 +683,8 @@ virFDStreamAbort(virStreamPtr st) static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) { virFDStreamDataPtr fdst = st->privateData; - int ret; + virFDStreamMsgPtr msg = NULL; + int ret = -1; if (nbytes > INT_MAX) { virReportSystemError(ERANGE, "%s", @@ -496,25 +712,51 @@ static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) nbytes = fdst->length - fdst->offset; } - retry: - ret = write(fdst->fd, bytes, nbytes); - if (ret < 0) { - VIR_WARNINGS_NO_WLOGICALOP_EQUAL_EXPR - if (errno == EAGAIN || errno == EWOULDBLOCK) { - VIR_WARNINGS_RESET - ret = -2; - } else if (errno == EINTR) { - goto retry; - } else { - ret = -1; - virReportSystemError(errno, "%s", + if (fdst->thread) { + char *buf; + + if (fdst->threadQuit) { + virReportSystemError(EBADF, "%s", _("cannot write to stream")); + return -1; + } + + if (VIR_ALLOC(msg) < 0 || + VIR_ALLOC_N(buf, nbytes) < 0) + goto cleanup; + + memcpy(buf, bytes, nbytes); + msg->type = VIR_FDSTREAM_MSG_TYPE_DATA; + msg->stream.data.buf = buf; + msg->stream.data.len = nbytes; + + virFDStreamMsgQueuePush(fdst, msg, fdst->fd, "pipe"); + msg = NULL; + ret = nbytes; + } else { + retry: + ret = write(fdst->fd, bytes, nbytes); + if (ret < 0) { + VIR_WARNINGS_NO_WLOGICALOP_EQUAL_EXPR + if (errno == EAGAIN || errno == EWOULDBLOCK) { + VIR_WARNINGS_RESET + ret = -2; + } else if (errno == EINTR) { + goto retry; + } else { + ret = -1; + virReportSystemError(errno, "%s", + _("cannot write to stream")); + } } - } else if (fdst->length) { - fdst->offset += ret; } + if (fdst->length) + fdst->offset += ret; + + cleanup: virObjectUnlock(fdst); + virFDStreamMsgFree(msg); return ret; } @@ -522,7 +764,7 @@ static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) { virFDStreamDataPtr fdst = st->privateData; - int ret; + int ret = -1; if (nbytes > INT_MAX) { virReportSystemError(ERANGE, "%s", @@ -548,24 +790,70 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) nbytes = fdst->length - fdst->offset; } - retry: - ret = read(fdst->fd, bytes, nbytes); - if (ret < 0) { - VIR_WARNINGS_NO_WLOGICALOP_EQUAL_EXPR - if (errno == EAGAIN || errno == EWOULDBLOCK) { - VIR_WARNINGS_RESET - ret = -2; - } else if (errno == EINTR) { - goto retry; - } else { - ret = -1; - virReportSystemError(errno, "%s", - _("cannot read from stream")); + if (fdst->thread) { + virFDStreamMsgPtr msg = NULL; + + while (!(msg = fdst->msg)) { + if (fdst->threadQuit) { + if (nbytes) { + virReportSystemError(EBADF, "%s", + _("stream is not open")); + } else { + ret = 0; + } + goto cleanup; + } else { + virObjectUnlock(fdst); + virCondSignal(&fdst->threadCond); + virObjectLock(fdst); + } + } + + if (msg->type != VIR_FDSTREAM_MSG_TYPE_DATA) { + /* Nope, nope, I'm outta here */ + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("unexpected message type")); + goto cleanup; + } + + if (nbytes > msg->stream.data.len - msg->stream.data.offset) + nbytes = msg->stream.data.len - msg->stream.data.offset; + + memcpy(bytes, + msg->stream.data.buf + msg->stream.data.offset, + nbytes); + + msg->stream.data.offset += nbytes; + if (msg->stream.data.offset == msg->stream.data.len) { + virFDStreamMsgQueuePop(fdst, fdst->fd, "pipe"); + virFDStreamMsgFree(msg); + } + + ret = nbytes; + + } else { + retry: + ret = read(fdst->fd, bytes, nbytes); + if (ret < 0) { + VIR_WARNINGS_NO_WLOGICALOP_EQUAL_EXPR + if (errno == EAGAIN || errno == EWOULDBLOCK) { + VIR_WARNINGS_RESET + ret = -2; + } else if (errno == EINTR) { + goto retry; + } else { + ret = -1; + virReportSystemError(errno, "%s", + _("cannot read from stream")); + } + goto cleanup; } - } else if (fdst->length) { - fdst->offset += ret; } + if (fdst->length) + fdst->offset += ret; + + cleanup: virObjectUnlock(fdst); return ret; } @@ -610,11 +898,19 @@ static int virFDStreamOpenInternal(virStreamPtr st, st->privateData = fdst; if (threadData) { + fdst->threadDoRead = threadData->doRead; + /* Create the thread after fdst and st were initialized. * The thread worker expects them to be that way. */ if (VIR_ALLOC(fdst->thread) < 0) goto error; + if (virCondInit(&fdst->threadCond) < 0) { + virReportSystemError(errno, "%s", + _("cannot initialize condition variable")); + goto error; + } + if (virThreadCreate(fdst->thread, true, virFDStreamThread, @@ -783,6 +1079,7 @@ virFDStreamOpenFileInternal(virStreamPtr st, VIR_STRDUP(threadData->fdoutname, "pipe") < 0) goto error; tmpfd = pipefds[0]; + threadData->doRead = true; } else { threadData->fdin = pipefds[0]; threadData->fdout = fd; @@ -790,6 +1087,7 @@ virFDStreamOpenFileInternal(virStreamPtr st, VIR_STRDUP(threadData->fdoutname, path) < 0) goto error; tmpfd = pipefds[1]; + threadData->doRead = false; } } -- 2.13.0

On 05/16/2017 10:26 PM, John Ferlan wrote: A-ha! So this function might return uninitialized value (variable?), which is propagated to virFDStreamThread where it hits an error. Well, one one hand compiler tries to be smart (neither of those checks including @got in the parent function cause the compilation error since in that case @got is initialized); but on the other hand compiler fails to see there's no way with the current code for msg->type to be something different than MSG_TYPE_DATA in which case @got is set a value. Sigh. Actually, it doesn't matter. At this point, @fdst should have at least one reference held by parent process (I mean the other thread that spawned this thread). Not even in the case when unfair thread scheduling occurs. I mean, imagine streamOpen() & streamClose() to be called immediately one after another. steamOpen() spawns the thread, but lets assume that the scheduler is unfair and does not schedule the thread for a while. Well, streamClose() waits for the thread to join anyway, therefore it doesn't really matter on ordering of lock() & ref() operations. Michal

On 05/16/2017 10:03 AM, Michal Privoznik wrote: Still undocumented. I know it got discussed a few times... I guess the odd thing I find about "long long" is that all the stream mgmt lengths/sizes, etc. are size_t. You also only test for smaller values only. Just saying... Please just add the function description *and* why a long long is being used here. Also a brief summary of what's being done and of course that lseek @cleanup: Reviewed-by: John Ferlan <jferlan(a)redhat.com&gt; John [...]

On 05/17/2017 08:41 AM, Michal Privoznik wrote: Works for me with the one adjustment below: Thanks - Reviewed-by: John Ferlan <jferlan(a)redhat.com&gt; John function

On 05/16/2017 10:03 AM, Michal Privoznik wrote: [...] Perhaps some preventative programming: virCheckNonNegativeArgGoto(length, error); Although that would mean calling virDispatchError unless there was some sort of virCheck*ArgReturn(length, -1) Reviewed-by: John Ferlan <jferlan(a)redhat.com&gt; John [...]

This function is basically a counterpart for virStreamSendHole(). If one side of a stream called virStreamSendHole() the other should call virStreamRecvHole() to get the size of the hole. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- include/libvirt/libvirt-stream.h | 4 ++++ src/driver-stream.h | 6 ++++++ src/libvirt-stream.c | 44 ++++++++++++++++++++++++++++++++++++++++ src/libvirt_public.syms | 1 + 4 files changed, 55 insertions(+) diff --git a/include/libvirt/libvirt-stream.h b/include/libvirt/libvirt-stream.h index 14c9af142..feaa8ad64 100644 --- a/include/libvirt/libvirt-stream.h +++ b/include/libvirt/libvirt-stream.h @@ -54,6 +54,10 @@ int virStreamSendHole(virStreamPtr st, long long length, unsigned int flags); +int virStreamRecvHole(virStreamPtr, + long long *length, + unsigned int flags); + /** * virStreamSourceFunc: diff --git a/src/driver-stream.h b/src/driver-stream.h index 0a5201431..0fb56ebd2 100644 --- a/src/driver-stream.h +++ b/src/driver-stream.h @@ -46,6 +46,11 @@ typedef int long long length, unsigned int flags); +typedef int +(*virDrvStreamRecvHole)(virStreamPtr st, + long long *length, + unsigned int flags); + typedef int (*virDrvStreamEventAddCallback)(virStreamPtr stream, int events, @@ -74,6 +79,7 @@ struct _virStreamDriver { virDrvStreamRecv streamRecv; virDrvStreamRecvFlags streamRecvFlags; virDrvStreamSendHole streamSendHole; + virDrvStreamRecvHole streamRecvHole; virDrvStreamEventAddCallback streamEventAddCallback; virDrvStreamEventUpdateCallback streamEventUpdateCallback; virDrvStreamEventRemoveCallback streamEventRemoveCallback; diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c index a09896dcd..dc0dc9ea3 100644 --- a/src/libvirt-stream.c +++ b/src/libvirt-stream.c @@ -405,6 +405,50 @@ virStreamSendHole(virStreamPtr stream, } +/** + * virStreamRecvHole: + * @stream: pointer to the stream object + * @length: number of bytes to skip + * @flags: extra flags; not used yet, so callers should always pass 0 + * + * This API is used to determine the @length in bytes of the + * empty space to be created in a @stream's target file when + * uploading or downloading sparsely populated files. This is the + * counterpart to virStreamSendHole(). + * + * Returns 0 on success, + * -1 on error or when there's currently no hole in the stream + */ +int +virStreamRecvHole(virStreamPtr stream, + long long *length, + unsigned int flags) +{ + VIR_DEBUG("stream=%p, length=%p flags=%x", + stream, length, flags); + + virResetLastError(); + + virCheckStreamReturn(stream, -1); + virCheckNonNullArgReturn(length, -1); + + if (stream->driver && + stream->driver->streamRecvHole) { + int ret; + ret = (stream->driver->streamRecvHole)(stream, length, flags); + if (ret < 0) + goto error; + return ret; + } + + virReportUnsupportedError(); + + error: + virDispatchError(stream->conn); + return -1; +} + + /** * virStreamSendAll: * @stream: pointer to the stream object diff --git a/src/libvirt_public.syms b/src/libvirt_public.syms index 3be7cc6a0..b73cc8af1 100644 --- a/src/libvirt_public.syms +++ b/src/libvirt_public.syms @@ -762,6 +762,7 @@ LIBVIRT_3.1.0 { LIBVIRT_3.4.0 { global: virStreamRecvFlags; + virStreamRecvHole; virStreamSendHole; } LIBVIRT_3.1.0; -- 2.13.0

Add a new flag to virStreamRecvFlags in order to handle being able to stop reading from the stream so that the consumer can generate a "hole" in stream target. Generation of a hole replaces the need to receive and handle a sequence of zero bytes for sparse stream targets. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- include/libvirt/libvirt-stream.h | 4 ++++ src/libvirt-stream.c | 34 +++++++++++++++++++++++++++++++++- 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/include/libvirt/libvirt-stream.h b/include/libvirt/libvirt-stream.h index feaa8ad64..c4baaf7a3 100644 --- a/include/libvirt/libvirt-stream.h +++ b/include/libvirt/libvirt-stream.h @@ -45,6 +45,10 @@ int virStreamRecv(virStreamPtr st, char *data, size_t nbytes); +typedef enum { + VIR_STREAM_RECV_STOP_AT_HOLE = (1 << 0), +} virStreamRecvFlagsValues; + int virStreamRecvFlags(virStreamPtr st, char *data, size_t nbytes, diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c index dc0dc9ea3..bedb6159a 100644 --- a/src/libvirt-stream.c +++ b/src/libvirt-stream.c @@ -290,7 +290,7 @@ virStreamRecv(virStreamPtr stream, * @stream: pointer to the stream object * @data: buffer to read into from stream * @nbytes: size of @data buffer - * @flags: extra flags; not used yet, so callers should always pass 0 + * @flags: bitwise-OR of virStreamRecvFlagsValues * * Reads a series of bytes from the stream. This method may * block the calling application for an arbitrary amount @@ -300,6 +300,33 @@ virStreamRecv(virStreamPtr stream, * @flags. Calling this function with no @flags set (equal to * zero) is equivalent to calling virStreamRecv(stream, data, nbytes). * + * If flag VIR_STREAM_RECV_STOP_AT_HOLE is set, this function + * will stop reading from stream if it has reached a hole. In + * that case, -3 is returned and virStreamRecvHole() should be + * called to get the hole size. An example using this flag might + * look like this: + * + * while (1) { + * char buf[4096]; + * + * int ret = virStreamRecvFlags(st, buf, len, VIR_STREAM_STOP_AT_HOLE); + * if (ret < 0) { + * if (ret == -3) { + * long long len; + * ret = virStreamRecvHole(st, &len, 0); + * if (ret < 0) { + * ...error.. + * } else { + * ...seek len bytes in target... + * } + * } else { + * return -1; + * } + * } else { + * ...write buf to target... + * } + * } + * * Returns 0 when the end of the stream is reached, at * which time the caller should invoke virStreamFinish() * to get confirmation of stream completion. @@ -310,6 +337,9 @@ virStreamRecv(virStreamPtr stream, * * Returns -2 if there is no data pending to be read & the * stream is marked as non-blocking. + * + * Returns -3 if there is a hole in stream and caller requested + * to stop at a hole. */ int virStreamRecvFlags(virStreamPtr stream, @@ -331,6 +361,8 @@ virStreamRecvFlags(virStreamPtr stream, ret = (stream->driver->streamRecvFlags)(stream, data, nbytes, flags); if (ret == -2) return -2; + if (ret == -3) + return -3; if (ret < 0) goto error; return ret; -- 2.13.0

On 05/16/2017 11:22 PM, John Ferlan wrote: Not really. This is just an example - anybody can understand that. Michal

This is just a wrapper over new function that have been just introduced: virStreamSendHole() . It's very similar to virStreamSendAll() except it handles sparse streams well. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- include/libvirt/libvirt-stream.h | 65 +++++++++++++++- src/libvirt-stream.c | 159 +++++++++++++++++++++++++++++++++++++++ src/libvirt_public.syms | 1 + 3 files changed, 222 insertions(+), 3 deletions(-) diff --git a/include/libvirt/libvirt-stream.h b/include/libvirt/libvirt-stream.h index a5e69a1c1..d18d43140 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 6bf4c4f29..4cbe5eee1 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); + * + * virConnectUploadFile(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, &sectionLen, 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 37fc4e224..fac77fbea 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 .... -- 2.13.0

This is just an internal API, that calls corresponding function in stream driver. This function will set @data=1 if the underlying file is in data section, or @data=0 if it is in a hole. At any rate, @length is set to number of bytes remaining in the section the file currently is. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/driver-stream.h | 6 ++++++ src/libvirt-stream.c | 48 ++++++++++++++++++++++++++++++++++++++++++++++++ src/libvirt_internal.h | 4 ++++ src/libvirt_private.syms | 1 + 4 files changed, 59 insertions(+) diff --git a/src/driver-stream.h b/src/driver-stream.h index 0fb56ebd2..f207bf0eb 100644 --- a/src/driver-stream.h +++ b/src/driver-stream.h @@ -51,6 +51,11 @@ typedef int long long *length, unsigned int flags); +typedef int +(*virDrvStreamInData)(virStreamPtr st, + int *data, + long long *length); + typedef int (*virDrvStreamEventAddCallback)(virStreamPtr stream, int events, @@ -80,6 +85,7 @@ struct _virStreamDriver { virDrvStreamRecvFlags streamRecvFlags; virDrvStreamSendHole streamSendHole; virDrvStreamRecvHole streamRecvHole; + virDrvStreamInData streamInData; virDrvStreamEventAddCallback streamEventAddCallback; virDrvStreamEventUpdateCallback streamEventUpdateCallback; virDrvStreamEventRemoveCallback streamEventRemoveCallback; diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c index 4cbe5eee1..30c305035 100644 --- a/src/libvirt-stream.c +++ b/src/libvirt-stream.c @@ -481,6 +481,54 @@ virStreamRecvHole(virStreamPtr stream, } +/** + * virStreamInData: + * @stream: stream + * @data: are we in data or hole + * @length: length to next section + * + * This function checks the underlying stream (typically a file) + * to learn whether the current stream position lies within a + * data section or a hole. Upon return @data is set to a nonzero + * value if former is the case, or to zero if @stream is in a + * hole. Moreover, @length is updated to tell caller how many + * bytes can be read from @stream until current section changes + * (from data to a hole or vice versa). + * + * NB: there's an implicit hole at EOF. In this situation this + * function should set @data = false, @length = 0 and return 0. + * + * To sum it up: + * + * data section: @data = true, @length > 0 + * hole: @data = false, @length > 0 + * EOF: @data = false, @length = 0 + * + * Returns 0 on success, + * -1 otherwise + */ +int +virStreamInData(virStreamPtr stream, + int *data, + long long *length) +{ + VIR_DEBUG("stream=%p, data=%p, length=%p", stream, data, length); + + /* No checks of arguments or error resetting. This is an + * internal function that just happen to live next to some + * public functions of ours. */ + + if (stream->driver->streamInData) { + int ret; + ret = (stream->driver->streamInData)(stream, data, length); + return ret; + } + + virReportUnsupportedError(); + return -1; +} + + /** * virStreamSendAll: * @stream: pointer to the stream object diff --git a/src/libvirt_internal.h b/src/libvirt_internal.h index 96439d840..62f490a7d 100644 --- a/src/libvirt_internal.h +++ b/src/libvirt_internal.h @@ -294,4 +294,8 @@ virTypedParameterValidateSet(virConnectPtr conn, virTypedParameterPtr params, int nparams); +int virStreamInData(virStreamPtr stream, + int *data, + long long *length); + #endif diff --git a/src/libvirt_private.syms b/src/libvirt_private.syms index 4102a002b..a1447eb44 100644 --- a/src/libvirt_private.syms +++ b/src/libvirt_private.syms @@ -1124,6 +1124,7 @@ virStateCleanup; virStateInitialize; virStateReload; virStateStop; +virStreamInData; # locking/domain_lock.h -- 2.13.0

On 05/17/2017 12:00 AM, John Ferlan wrote: It's not publicly accessible. It's private. But sure, checks would be harmless. I can add them. Michal

On 05/16/2017 10:03 AM, Michal Privoznik wrote: [...] Should this be a virObjectRef(stream); ? Reviewed-by: John Ferlan <jferlan(a)redhat.com&gt; John [...]

Add a new argument to daemonCreateClientStream in order to allow for future expansion to mark that a specific stream can be used to skip data, such as the case with sparsely populated files. The new flag will be the eventual decision point between client/server to decide whether both ends can support and want to use sparse streams. A new bool 'allowSkip' is added to both _virNetClientStream and daemonClientStream in order to perform the tracking. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- daemon/remote.c | 2 +- daemon/stream.c | 6 +++++- daemon/stream.h | 3 ++- src/remote/remote_driver.c | 6 ++++-- src/rpc/gendispatch.pl | 4 ++-- src/rpc/virnetclientstream.c | 6 +++++- src/rpc/virnetclientstream.h | 3 ++- 7 files changed, 21 insertions(+), 9 deletions(-) diff --git a/daemon/remote.c b/daemon/remote.c index 0dbb250ff..fd8542120 100644 --- a/daemon/remote.c +++ b/daemon/remote.c @@ -5323,7 +5323,7 @@ remoteDispatchDomainMigratePrepareTunnel3Params(virNetServerPtr server ATTRIBUTE if (!(st = virStreamNew(priv->conn, VIR_STREAM_NONBLOCK)) || !(stream = daemonCreateClientStream(client, st, remoteProgram, - &msg->header))) + &msg->header, false))) goto cleanup; if (virDomainMigratePrepareTunnel3Params(priv->conn, st, params, nparams, diff --git a/daemon/stream.c b/daemon/stream.c index 11c0a469d..6465463ff 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -52,6 +52,8 @@ struct daemonClientStream { virNetMessagePtr rx; bool tx; + bool allowSkip; + daemonClientStreamPtr next; }; @@ -321,7 +323,8 @@ daemonClientStream * daemonCreateClientStream(virNetServerClientPtr client, virStreamPtr st, virNetServerProgramPtr prog, - virNetMessageHeaderPtr header) + virNetMessageHeaderPtr header, + bool allowSkip) { daemonClientStream *stream; daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client); @@ -339,6 +342,7 @@ daemonCreateClientStream(virNetServerClientPtr client, stream->serial = header->serial; stream->filterID = -1; stream->st = st; + stream->allowSkip = allowSkip; return stream; } diff --git a/daemon/stream.h b/daemon/stream.h index cf76e717a..e1f106759 100644 --- a/daemon/stream.h +++ b/daemon/stream.h @@ -30,7 +30,8 @@ daemonClientStream * daemonCreateClientStream(virNetServerClientPtr client, virStreamPtr st, virNetServerProgramPtr prog, - virNetMessageHeaderPtr hdr); + virNetMessageHeaderPtr hdr, + bool allowSkip); int daemonFreeClientStream(virNetServerClientPtr client, daemonClientStream *stream); diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index b152be523..aebdd47c9 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -6173,7 +6173,8 @@ remoteDomainMigratePrepareTunnel3(virConnectPtr dconn, if (!(netst = virNetClientStreamNew(st, priv->remoteProgram, REMOTE_PROC_DOMAIN_MIGRATE_PREPARE_TUNNEL3, - priv->counter))) + priv->counter, + false))) goto done; if (virNetClientAddStream(priv->client, netst) < 0) { @@ -7098,7 +7099,8 @@ remoteDomainMigratePrepareTunnel3Params(virConnectPtr dconn, if (!(netst = virNetClientStreamNew(st, priv->remoteProgram, REMOTE_PROC_DOMAIN_MIGRATE_PREPARE_TUNNEL3_PARAMS, - priv->counter))) + priv->counter, + false))) goto cleanup; if (virNetClientAddStream(priv->client, netst) < 0) { diff --git a/src/rpc/gendispatch.pl b/src/rpc/gendispatch.pl index e608812ce..98625983a 100755 --- a/src/rpc/gendispatch.pl +++ b/src/rpc/gendispatch.pl @@ -1024,7 +1024,7 @@ elsif ($mode eq "server") { print " if (!(st = virStreamNew(priv->conn, VIR_STREAM_NONBLOCK)))\n"; print " goto cleanup;\n"; print "\n"; - print " if (!(stream = daemonCreateClientStream(client, st, remoteProgram, &msg->header)))\n"; + print " if (!(stream = daemonCreateClientStream(client, st, remoteProgram, &msg->header, false)))\n"; print " goto cleanup;\n"; print "\n"; } @@ -1738,7 +1738,7 @@ elsif ($mode eq "client") { if ($call->{streamflag} ne "none") { print "\n"; - print " if (!(netst = virNetClientStreamNew(st, priv->remoteProgram, $call->{constname}, priv->counter)))\n"; + print " if (!(netst = virNetClientStreamNew(st, priv->remoteProgram, $call->{constname}, priv->counter, false)))\n"; print " goto done;\n"; print "\n"; print " if (virNetClientAddStream(priv->client, netst) < 0) {\n"; diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index 01761cf8d..4c27f308e 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -54,6 +54,8 @@ struct _virNetClientStream { virNetMessagePtr rx; bool incomingEOF; + bool allowSkip; + virNetClientStreamEventCallback cb; void *cbOpaque; virFreeCallback cbFree; @@ -138,7 +140,8 @@ virNetClientStreamEventTimer(int timer ATTRIBUTE_UNUSED, void *opaque) virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, virNetClientProgramPtr prog, int proc, - unsigned serial) + unsigned serial, + bool allowSkip) { virNetClientStreamPtr st; @@ -152,6 +155,7 @@ virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, st->prog = virObjectRef(prog); st->proc = proc; st->serial = serial; + st->allowSkip = allowSkip; return st; } diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index e278dab85..f3bc0672b 100644 --- a/src/rpc/virnetclientstream.h +++ b/src/rpc/virnetclientstream.h @@ -35,7 +35,8 @@ typedef void (*virNetClientStreamEventCallback)(virNetClientStreamPtr stream, virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, virNetClientProgramPtr prog, int proc, - unsigned serial); + unsigned serial, + bool allowSkip); bool virNetClientStreamRaiseError(virNetClientStreamPtr st); -- 2.13.0

This is going to be RPC representation for virStreamSendHole. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/virnetprotocol.x | 5 +++++ src/virnetprotocol-structs | 4 ++++ 2 files changed, 9 insertions(+) diff --git a/src/rpc/virnetprotocol.x b/src/rpc/virnetprotocol.x index 9ce33b073..cab047cb0 100644 --- a/src/rpc/virnetprotocol.x +++ b/src/rpc/virnetprotocol.x @@ -236,3 +236,8 @@ struct virNetMessageError { int int2; virNetMessageNetwork net; /* unused */ }; + +struct virNetStreamHole { + hyper length; + unsigned int flags; +}; diff --git a/src/virnetprotocol-structs b/src/virnetprotocol-structs index af4526c90..aa6e0602a 100644 --- a/src/virnetprotocol-structs +++ b/src/virnetprotocol-structs @@ -42,3 +42,7 @@ struct virNetMessageError { int int2; virNetMessageNetwork net; }; +struct virNetStreamHole { + int64_t length; + u_int flags; +}; -- 2.13.0

This is a special type of stream packet, that is bidirectional and will contain information on how much bytes are both sides skipping in the stream. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- daemon/stream.c | 3 ++- src/rpc/virnetclient.c | 1 + src/rpc/virnetprotocol.x | 12 +++++++++++- src/virnetprotocol-structs | 1 + 4 files changed, 15 insertions(+), 2 deletions(-) diff --git a/daemon/stream.c b/daemon/stream.c index 6465463ff..f44c21278 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -287,7 +287,8 @@ daemonStreamFilter(virNetServerClientPtr client ATTRIBUTE_UNUSED, virMutexLock(&stream->priv->lock); - if (msg->header.type != VIR_NET_STREAM) + if (msg->header.type != VIR_NET_STREAM && + msg->header.type != VIR_NET_STREAM_HOLE) goto cleanup; if (!virNetServerProgramMatches(stream->prog, msg)) diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index 837a8a707..95cd9a6c7 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -1284,6 +1284,7 @@ virNetClientCallDispatch(virNetClientPtr client) return virNetClientCallDispatchMessage(client); case VIR_NET_STREAM: /* Stream protocol */ + case VIR_NET_STREAM_HOLE: /* Sparse stream protocol*/ return virNetClientCallDispatchStream(client); default: diff --git a/src/rpc/virnetprotocol.x b/src/rpc/virnetprotocol.x index cab047cb0..ee9899059 100644 --- a/src/rpc/virnetprotocol.x +++ b/src/rpc/virnetprotocol.x @@ -143,6 +143,14 @@ const VIR_NET_MESSAGE_NUM_FDS_MAX = 32; * * status == VIR_NET_ERROR * remote_error Error information * + * - type == VIR_NET_STREAM_HOLE + * * status == VIR_NET_CONTINUE + * byte[] hole data + * * status == VIR_NET_ERROR + * remote_error error information + * * status == VIR_NET_OK + * <empty> + * */ enum virNetMessageType { /* client -> server. args from a method call */ @@ -156,7 +164,9 @@ enum virNetMessageType { /* client -> server. args from a method call, with passed FDs */ VIR_NET_CALL_WITH_FDS = 4, /* server -> client. reply/error from a method call, with passed FDs */ - VIR_NET_REPLY_WITH_FDS = 5 + VIR_NET_REPLY_WITH_FDS = 5, + /* either direction, stream hole data packet */ + VIR_NET_STREAM_HOLE = 6 }; enum virNetMessageStatus { diff --git a/src/virnetprotocol-structs b/src/virnetprotocol-structs index aa6e0602a..b36581f86 100644 --- a/src/virnetprotocol-structs +++ b/src/virnetprotocol-structs @@ -6,6 +6,7 @@ enum virNetMessageType { VIR_NET_STREAM = 3, VIR_NET_CALL_WITH_FDS = 4, VIR_NET_REPLY_WITH_FDS = 5, + VIR_NET_STREAM_HOLE = 6, }; enum virNetMessageStatus { VIR_NET_OK = 0, -- 2.13.0

Ideally, this would be generated, but to achieve that corresponding XDR definitions needed to go into a different .x file. But they belong just to the one that they are right now. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- tools/wireshark/src/packet-libvirt.c | 52 ++++++++++++++++++++++++++++++++++++ tools/wireshark/src/packet-libvirt.h | 2 ++ 2 files changed, 54 insertions(+) diff --git a/tools/wireshark/src/packet-libvirt.c b/tools/wireshark/src/packet-libvirt.c index 260161e98..a1f5a34f4 100644 --- a/tools/wireshark/src/packet-libvirt.c +++ b/tools/wireshark/src/packet-libvirt.c @@ -50,8 +50,12 @@ static int hf_libvirt_serial = -1; static int hf_libvirt_status = -1; static int hf_libvirt_stream = -1; static int hf_libvirt_num_of_fds = -1; +static int hf_libvirt_stream_hole_length = -1; +static int hf_libvirt_stream_hole_flags = -1; +static int hf_libvirt_stream_hole = -1; int hf_libvirt_unknown = -1; static gint ett_libvirt = -1; +static gint ett_libvirt_stream_hole = -1; #define XDR_PRIMITIVE_DISSECTOR(xtype, ctype, ftype) \ static gboolean \ @@ -326,6 +330,33 @@ dissect_libvirt_payload_xdr_data(tvbuff_t *tvb, proto_tree *tree, gint payload_l dissect_libvirt_fds(tvb, start + payload_length, nfds); } +static gboolean +dissect_xdr_stream_hole(tvbuff_t *tvb, proto_tree *tree, XDR *xdrs, int hf) +{ + goffset start; + proto_item *ti; + + start = xdr_getpos(xdrs); + if (hf == -1) { + ti = proto_tree_add_item(tree, hf_libvirt_stream_hole, tvb, start, -1, ENC_NA); + } else { + header_field_info *hfinfo; + hfinfo = proto_registrar_get_nth(hf_libvirt_stream_hole); + ti = proto_tree_add_item(tree, hf, tvb, start, -1, ENC_NA); + proto_item_append_text(ti, " :: %s", hfinfo->name); + } + tree = proto_item_add_subtree(ti, ett_libvirt_stream_hole); + + hf = hf_libvirt_stream_hole_length; + if (!dissect_xdr_hyper(tvb, tree, xdrs, hf)) return FALSE; + + hf = hf_libvirt_stream_hole_flags; + if (!dissect_xdr_u_int(tvb, tree, xdrs, hf)) return FALSE; + + proto_item_set_len(ti, xdr_getpos(xdrs) - start); + return TRUE; +} + static void dissect_libvirt_payload(tvbuff_t *tvb, proto_tree *tree, guint32 prog, guint32 proc, guint32 type, guint32 status) @@ -346,6 +377,8 @@ dissect_libvirt_payload(tvbuff_t *tvb, proto_tree *tree, dissect_libvirt_payload_xdr_data(tvb, tree, payload_length, status, VIR_ERROR_MESSAGE_DISSECTOR); } else if (type == VIR_NET_STREAM) { /* implicitly, status == VIR_NET_CONTINUE */ dissect_libvirt_stream(tvb, tree, payload_length); + } else if (type == VIR_NET_STREAM_HOLE) { + dissect_libvirt_payload_xdr_data(tvb, tree, payload_length, status, dissect_xdr_stream_hole); } else { goto unknown; } @@ -525,6 +558,24 @@ proto_register_libvirt(void) NULL, 0x0, NULL, HFILL} }, + { &hf_libvirt_stream_hole, + { "stream_hole", "libvirt.stream_hole", + FT_NONE, BASE_NONE, + NULL, 0x0, + NULL, HFILL} + }, + { &hf_libvirt_stream_hole_length, + { "length", "libvirt.stream_hole.length", + FT_INT64, BASE_DEC, + NULL, 0x0, + NULL, HFILL} + }, + { &hf_libvirt_stream_hole_flags, + { "flags", "libvirt.stream_hole.flags", + FT_UINT32, BASE_DEC, + NULL, 0x0, + NULL, HFILL} + }, { &hf_libvirt_unknown, { "unknown", "libvirt.unknown", FT_BYTES, BASE_NONE, @@ -535,6 +586,7 @@ proto_register_libvirt(void) static gint *ett[] = { VIR_DYNAMIC_ETTSET + &ett_libvirt_stream_hole, &ett_libvirt }; diff --git a/tools/wireshark/src/packet-libvirt.h b/tools/wireshark/src/packet-libvirt.h index 5f99fdfae..9874a8cbf 100644 --- a/tools/wireshark/src/packet-libvirt.h +++ b/tools/wireshark/src/packet-libvirt.h @@ -53,6 +53,7 @@ enum vir_net_message_type { VIR_NET_STREAM = 3, VIR_NET_CALL_WITH_FDS = 4, VIR_NET_REPLY_WITH_FDS = 5, + VIR_NET_STREAM_HOLE = 6, }; enum vir_net_message_status { @@ -76,6 +77,7 @@ static const value_string type_strings[] = { { VIR_NET_STREAM, "STREAM" }, { VIR_NET_CALL_WITH_FDS, "CALL_WITH_FDS" }, { VIR_NET_REPLY_WITH_FDS, "REPLY_WITH_FDS" }, + { VIR_NET_STREAM_HOLE, "STREAM_HOLE" }, { -1, NULL } }; -- 2.13.0

On 05/17/2017 01:44 PM, John Ferlan wrote: 'int hf' is not there to represent LL @offset. @hf stands for 'header field'. It's an index into an array where info on the field is to be found, for instance: { &hf_libvirt_stream_hole, { "stream_hole", "libvirt.stream_hole", FT_NONE, BASE_NONE, NULL, 0x0, NULL, HFILL} }, "stream_hole" - abbreviated name of the field "libvirt.stream_hole" - full name of the field FT_NONE - field type (char, bool, integer of all sorts, ...) BASE_NONE - base for decoding (decimal, hexa, octa, ...) and so on. Therefore, the 'int hf' in the dissect_xdr_stream_hole is not 'long long offset' contained in the packed, rather than an index into an array. This index is then used when telling wireshark how to display particular piece of a packet. Good enough for me :-) Michal

On 05/16/2017 10:03 AM, Michal Privoznik wrote: Perhaps this is what got me confused in the previous review of that series patch 30. Here it "feels like" having two holes in a row is not allowed, but then there was that logic and discussion that implied it was possible. So unless math is off, this is a "shouldn't happen" type condition. I'm fine with leaving the check in, but just trying to convince myself. Do we still not have an answer? Nothing in this module does such an abort, so rather than leave this dangling, I'd lean towards no. Still one way or another this does need to be cleaned up before pushing... Still, consider this: Reviewed-by: John Ferlan <jferlan(a)redhat.com&gt; John

On 05/16/2017 10:04 AM, Michal Privoznik wrote: This should be int virNetClient... Although, yes, different than other functions here. Reviewed-by: John Ferlan <jferlan(a)redhat.com&gt; John [...]

On 05/16/2017 10:04 AM, Michal Privoznik wrote: Of course this is what I was referring to in patch 19... The only way to call virNetClientStreamHandleHole is if holeLength = 0, but since all the other validity checks (e.g. st->rx/msg and msg->header.type == HOLE) are present - what's one more later on... Your call you could also modify patch 19 to not have any extraneous checks that wouldn't be possible given the above lines... Your call, IDC. holeLength == 0 Reviewed-by: John Ferlan <jferlan(a)redhat.com&gt; John

On 05/16/2017 10:04 AM, Michal Privoznik wrote: empty between paragraphs s/it an inform/it and inform/ Oh, right virStreamInData is not publicly available... (patch 10 comments)... dataLen == 0 ? (I know, same thing - I guess it's just one of those visual things for me...) Extra space between "the other" Since dataLen is only "set" if stream->allowSkip - should this be fenced similarly? Not that I see ->dataLen being used for anything other than sparse stream mgmt... Reviewed-by: John Ferlan <jferlan(a)redhat.com&gt; John

These flags to APIs will tell if caller wants to use sparse stream for storage transfer. At the same time, it's safe to enable them in storage driver frontend and rely on our backends checking the flags. This way we can enable specific flags only on some specific backends, e.g. enable VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM for filesystem backend but not iSCSI backend. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- include/libvirt/libvirt-storage.h | 9 +++++++++ src/libvirt-storage.c | 4 ++-- src/remote/remote_protocol.x | 2 ++ src/storage/storage_driver.c | 4 ++-- src/storage/storage_util.c | 10 ++++++---- 5 files changed, 21 insertions(+), 8 deletions(-) diff --git a/include/libvirt/libvirt-storage.h b/include/libvirt/libvirt-storage.h index 45ec72065..4517f713c 100644 --- a/include/libvirt/libvirt-storage.h +++ b/include/libvirt/libvirt-storage.h @@ -346,11 +346,20 @@ virStorageVolPtr virStorageVolCreateXMLFrom (virStoragePoolPtr pool, const char *xmldesc, virStorageVolPtr clonevol, unsigned int flags); + +typedef enum { + VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM = 1 << 0, /* Use sparse stream */ +} virStorageVolDownloadFlags; + int virStorageVolDownload (virStorageVolPtr vol, virStreamPtr stream, unsigned long long offset, unsigned long long length, unsigned int flags); +typedef enum { + VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM = 1 << 0, /* Use sparse stream */ +} virStorageVolUploadFlags; + int virStorageVolUpload (virStorageVolPtr vol, virStreamPtr stream, unsigned long long offset, diff --git a/src/libvirt-storage.c b/src/libvirt-storage.c index 05eec8a9d..64202998b 100644 --- a/src/libvirt-storage.c +++ b/src/libvirt-storage.c @@ -1549,7 +1549,7 @@ virStorageVolCreateXMLFrom(virStoragePoolPtr pool, * @stream: stream to use as output * @offset: position in @vol to start reading from * @length: limit on amount of data to download - * @flags: extra flags; not used yet, so callers should always pass 0 + * @flags: bitwise-OR of virStorageVolDownloadFlags * * Download the content of the volume as a stream. If @length * is zero, then the remaining contents of the volume after @@ -1613,7 +1613,7 @@ virStorageVolDownload(virStorageVolPtr vol, * @stream: stream to use as input * @offset: position to start writing to * @length: limit on amount of data to upload - * @flags: extra flags; not used yet, so callers should always pass 0 + * @flags: bitwise-OR of virStorageVolUploadFlags * * Upload new content to the volume from a stream. This call * will fail if @offset + @length exceeds the size of the diff --git a/src/remote/remote_protocol.x b/src/remote/remote_protocol.x index 87b2bd365..25e62a181 100644 --- a/src/remote/remote_protocol.x +++ b/src/remote/remote_protocol.x @@ -4896,6 +4896,7 @@ enum remote_procedure { /** * @generate: both * @writestream: 1 + * @sparseflag: VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM * @acl: storage_vol:data_write */ REMOTE_PROC_STORAGE_VOL_UPLOAD = 208, @@ -4903,6 +4904,7 @@ enum remote_procedure { /** * @generate: both * @readstream: 1 + * @sparseflag: VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM * @acl: storage_vol:data_read */ REMOTE_PROC_STORAGE_VOL_DOWNLOAD = 209, diff --git a/src/storage/storage_driver.c b/src/storage/storage_driver.c index 2103ed11d..1b0d776c7 100644 --- a/src/storage/storage_driver.c +++ b/src/storage/storage_driver.c @@ -2117,7 +2117,7 @@ storageVolDownload(virStorageVolPtr obj, virStorageVolDefPtr vol = NULL; int ret = -1; - virCheckFlags(0, -1); + virCheckFlags(VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM, -1); if (!(vol = virStorageVolDefFromVol(obj, &pool, &backend))) return -1; @@ -2285,7 +2285,7 @@ storageVolUpload(virStorageVolPtr obj, virStorageVolStreamInfoPtr cbdata = NULL; int ret = -1; - virCheckFlags(0, -1); + virCheckFlags(VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM, -1); if (!(vol = virStorageVolDefFromVol(obj, &pool, &backend))) return -1; diff --git a/src/storage/storage_util.c b/src/storage/storage_util.c index 908cad874..493c651b7 100644 --- a/src/storage/storage_util.c +++ b/src/storage/storage_util.c @@ -2401,8 +2401,9 @@ virStorageBackendVolUploadLocal(virConnectPtr conn ATTRIBUTE_UNUSED, char *target_path = vol->target.path; int ret = -1; int has_snap = 0; + bool sparse = flags & VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM; - virCheckFlags(0, -1); + virCheckFlags(VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM, -1); /* if volume has target format VIR_STORAGE_FILE_PLOOP * we need to restore DiskDescriptor.xml, according to * new contents of volume. This operation will be perfomed @@ -2427,7 +2428,7 @@ virStorageBackendVolUploadLocal(virConnectPtr conn ATTRIBUTE_UNUSED, /* Not using O_CREAT because the file is required to already exist at * this point */ ret = virFDStreamOpenBlockDevice(stream, target_path, - offset, len, false, O_WRONLY); + offset, len, sparse, O_WRONLY); cleanup: VIR_FREE(path); @@ -2447,8 +2448,9 @@ virStorageBackendVolDownloadLocal(virConnectPtr conn ATTRIBUTE_UNUSED, char *target_path = vol->target.path; int ret = -1; int has_snap = 0; + bool sparse = flags & VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM; - virCheckFlags(0, -1); + virCheckFlags(VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM, -1); if (vol->target.format == VIR_STORAGE_FILE_PLOOP) { has_snap = storageBackendPloopHasSnapshots(vol->target.path); if (has_snap < 0) { @@ -2465,7 +2467,7 @@ virStorageBackendVolDownloadLocal(virConnectPtr conn ATTRIBUTE_UNUSED, } ret = virFDStreamOpenBlockDevice(stream, target_path, - offset, len, false, O_RDONLY); + offset, len, sparse, O_RDONLY); cleanup: VIR_FREE(path); -- 2.13.0

On 05/17/2017 05:42 PM, John Ferlan wrote: How about this? iff --git i/src/libvirt-storage.c w/src/libvirt-storage.c index 64202998b..35f9926d5 100644 --- i/src/libvirt-storage.c +++ w/src/libvirt-storage.c @@ -1555,6 +1555,13 @@ virStorageVolCreateXMLFrom(virStoragePoolPtr pool, * is zero, then the remaining contents of the volume after * @offset will be downloaded. * + * If VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM is set in @flags + * effective transmission of holes is enabled. This assumes using + * the @stream with combination of virStreamSparseRecvAll() or + * virStreamRecvFlags(stream, ..., flags = + * VIR_STREAM_RECV_STOP_AT_HOLE) for honouring holes sent by + * server. + * * This call sets up an asynchronous stream; subsequent use of * stream APIs is necessary to transfer the actual data, * determine how much data is successfully transferred, and @@ -1621,6 +1628,11 @@ virStorageVolDownload(virStorageVolPtr vol, * will be raised if an attempt is made to upload greater * than @length bytes of data. * + * If VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM is set in @flags + * effective transmission of holes is enabled. This assumes using + * the @stream with combination of virStreamSparseSendAll() or + * virStreamSendHole() to preserve source file sparseness. + * * This call sets up an asynchronous stream; subsequent use of * stream APIs is necessary to transfer the actual data, * determine how much data is successfully transferred, and Michal

The command grew new --sparse switch that does nothing more than enables the sparse streams feature for this command. Among with the switch new helper function is introduced: virshStreamSkip(). This is the callback that is called whenever daemon sends us a hole. In the callback we reflect the hole in underlying file by seeking as many bytes as told. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- tools/virsh-util.c | 18 ++++++++++++++++++ tools/virsh-util.h | 5 +++++ tools/virsh-volume.c | 12 ++++++++++-- tools/virsh.pod | 3 ++- 4 files changed, 35 insertions(+), 3 deletions(-) diff --git a/tools/virsh-util.c b/tools/virsh-util.c index 4b86e29cb..198625bdb 100644 --- a/tools/virsh-util.c +++ b/tools/virsh-util.c @@ -153,6 +153,24 @@ virshStreamSink(virStreamPtr st ATTRIBUTE_UNUSED, } +int +virshStreamSkip(virStreamPtr st ATTRIBUTE_UNUSED, + long long offset, + void *opaque) +{ + int *fd = opaque; + off_t cur; + + if ((cur = lseek(*fd, offset, SEEK_CUR)) == (off_t) -1) + return -1; + + if (ftruncate(*fd, cur) < 0) + return -1; + + return 0; +} + + void virshDomainFree(virDomainPtr dom) { diff --git a/tools/virsh-util.h b/tools/virsh-util.h index 64cef23c0..0aba247f6 100644 --- a/tools/virsh-util.h +++ b/tools/virsh-util.h @@ -57,6 +57,11 @@ virshStreamSink(virStreamPtr st, size_t nbytes, void *opaque); +int +virshStreamSkip(virStreamPtr st, + long long offset, + void *opaque); + int virshDomainGetXMLFromDom(vshControl *ctl, virDomainPtr dom, diff --git a/tools/virsh-volume.c b/tools/virsh-volume.c index 66fe70ea7..3d19b745e 100644 --- a/tools/virsh-volume.c +++ b/tools/virsh-volume.c @@ -763,6 +763,10 @@ static const vshCmdOptDef opts_vol_download[] = { .type = VSH_OT_INT, .help = N_("amount of data to download") }, + {.name = "sparse", + .type = VSH_OT_BOOL, + .help = N_("preserve sparseness of volume") + }, {.name = NULL} }; @@ -778,6 +782,7 @@ cmdVolDownload(vshControl *ctl, const vshCmd *cmd) unsigned long long offset = 0, length = 0; bool created = false; virshControlPtr priv = ctl->privData; + unsigned int flags = 0; if (vshCommandOptULongLong(ctl, cmd, "offset", &offset) < 0) return false; @@ -791,6 +796,9 @@ cmdVolDownload(vshControl *ctl, const vshCmd *cmd) if (vshCommandOptStringReq(ctl, cmd, "file", &file) < 0) goto cleanup; + if (vshCommandOptBool(cmd, "sparse")) + flags |= VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM; + if ((fd = open(file, O_WRONLY|O_CREAT|O_EXCL, 0666)) < 0) { if (errno != EEXIST || (fd = open(file, O_WRONLY|O_TRUNC, 0666)) < 0) { @@ -806,12 +814,12 @@ cmdVolDownload(vshControl *ctl, const vshCmd *cmd) goto cleanup; } - if (virStorageVolDownload(vol, st, offset, length, 0) < 0) { + if (virStorageVolDownload(vol, st, offset, length, flags) < 0) { vshError(ctl, _("cannot download from volume %s"), name); goto cleanup; } - if (virStreamRecvAll(st, virshStreamSink, &fd) < 0) { + if (virStreamSparseRecvAll(st, virshStreamSink, virshStreamSkip, &fd) < 0) { vshError(ctl, _("cannot receive data from volume %s"), name); goto cleanup; } diff --git a/tools/virsh.pod b/tools/virsh.pod index 727acf6e6..dcaa0c170 100644 --- a/tools/virsh.pod +++ b/tools/virsh.pod @@ -3942,12 +3942,13 @@ regarding possible target volume and pool changes as a result of the pool refresh when the upload is attempted. =item B<vol-download> [I<--pool> I<pool-or-uuid>] [I<--offset> I<bytes>] -[I<--length> I<bytes>] I<vol-name-or-key-or-path> I<local-file> +[I<--length> I<bytes>] [I<--sparse>] I<vol-name-or-key-or-path> I<local-file> Download the contents of a storage volume to I<local-file>. I<--pool> I<pool-or-uuid> is the name or UUID of the storage pool the volume is in. I<vol-name-or-key-or-path> is the name or key or path of the volume to download. +If I<--sparse> is specified, this command will preserve volume sparseness. I<--offset> is the position in the storage volume at which to start reading the data. The value must be 0 or larger. I<--length> is an upper bound of the amount of data to be downloaded. A negative value is interpreted as -- 2.13.0

Similarly to previous commit, implement sparse streams feature for vol-upload. This is, however, slightly different approach, because we must implement a function that will tell us whether we are in a data section or in a hole. But there's no magic hidden in here. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- tools/virsh-util.c | 47 +++++++++++++++++++++++++++++++++++++++++++++++ tools/virsh-util.h | 24 ++++++++++++++++++++++++ tools/virsh-volume.c | 38 +++++++++++++++++++++++++------------- tools/virsh.pod | 3 ++- 4 files changed, 98 insertions(+), 14 deletions(-) diff --git a/tools/virsh-util.c b/tools/virsh-util.c index 198625bdb..44be3ad64 100644 --- a/tools/virsh-util.c +++ b/tools/virsh-util.c @@ -153,6 +153,35 @@ virshStreamSink(virStreamPtr st ATTRIBUTE_UNUSED, } +int +virshStreamSource(virStreamPtr st ATTRIBUTE_UNUSED, + char *bytes, + size_t nbytes, + void *opaque) +{ + virshStreamCallbackDataPtr cbData = opaque; + int fd = cbData->fd; + + return saferead(fd, bytes, nbytes); +} + + +int +virshStreamSourceSkip(virStreamPtr st ATTRIBUTE_UNUSED, + long long offset, + void *opaque) +{ + virshStreamCallbackDataPtr cbData = opaque; + int fd = cbData->fd; + off_t cur; + + if ((cur = lseek(fd, offset, SEEK_CUR)) == (off_t) -1) + return -1; + + return 0; +} + + int virshStreamSkip(virStreamPtr st ATTRIBUTE_UNUSED, long long offset, @@ -171,6 +200,24 @@ virshStreamSkip(virStreamPtr st ATTRIBUTE_UNUSED, } +int +virshStreamInData(virStreamPtr st ATTRIBUTE_UNUSED, + int *inData, + long long *offset, + void *opaque) +{ + virshStreamCallbackDataPtr cbData = opaque; + vshControl *ctl = cbData->ctl; + int fd = cbData->fd; + int ret; + + if ((ret = virFileInData(fd, inData, offset)) < 0) + vshError(ctl, "%s", _("Unable to get current position in stream")); + + return ret; +} + + void virshDomainFree(virDomainPtr dom) { diff --git a/tools/virsh-util.h b/tools/virsh-util.h index 0aba247f6..9a0af3513 100644 --- a/tools/virsh-util.h +++ b/tools/virsh-util.h @@ -57,11 +57,35 @@ virshStreamSink(virStreamPtr st, size_t nbytes, void *opaque); +typedef struct _virshStreamCallbackData virshStreamCallbackData; +typedef virshStreamCallbackData *virshStreamCallbackDataPtr; +struct _virshStreamCallbackData { + vshControl *ctl; + int fd; +}; + +int +virshStreamSource(virStreamPtr st, + char *bytes, + size_t nbytes, + void *opaque); + +int +virshStreamSourceSkip(virStreamPtr st, + long long offset, + void *opaque); + int virshStreamSkip(virStreamPtr st, long long offset, void *opaque); +int +virshStreamInData(virStreamPtr st, + int *inData, + long long *offset, + void *opaque); + int virshDomainGetXMLFromDom(vshControl *ctl, virDomainPtr dom, diff --git a/tools/virsh-volume.c b/tools/virsh-volume.c index 3d19b745e..0736bdcdb 100644 --- a/tools/virsh-volume.c +++ b/tools/virsh-volume.c @@ -660,18 +660,13 @@ static const vshCmdOptDef opts_vol_upload[] = { .type = VSH_OT_INT, .help = N_("amount of data to upload") }, + {.name = "sparse", + .type = VSH_OT_BOOL, + .help = N_("preserve sparseness of volume") + }, {.name = NULL} }; -static int -cmdVolUploadSource(virStreamPtr st ATTRIBUTE_UNUSED, - char *bytes, size_t nbytes, void *opaque) -{ - int *fd = opaque; - - return saferead(*fd, bytes, nbytes); -} - static bool cmdVolUpload(vshControl *ctl, const vshCmd *cmd) { @@ -683,6 +678,8 @@ cmdVolUpload(vshControl *ctl, const vshCmd *cmd) const char *name = NULL; unsigned long long offset = 0, length = 0; virshControlPtr priv = ctl->privData; + unsigned int flags = 0; + virshStreamCallbackData cbData; if (vshCommandOptULongLong(ctl, cmd, "offset", &offset) < 0) return false; @@ -701,19 +698,34 @@ cmdVolUpload(vshControl *ctl, const vshCmd *cmd) goto cleanup; } + cbData.ctl = ctl; + cbData.fd = fd; + + if (vshCommandOptBool(cmd, "sparse")) + flags |= VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM; + if (!(st = virStreamNew(priv->conn, 0))) { vshError(ctl, _("cannot create a new stream")); goto cleanup; } - if (virStorageVolUpload(vol, st, offset, length, 0) < 0) { + if (virStorageVolUpload(vol, st, offset, length, flags) < 0) { vshError(ctl, _("cannot upload to volume %s"), name); goto cleanup; } - if (virStreamSendAll(st, cmdVolUploadSource, &fd) < 0) { - vshError(ctl, _("cannot send data to volume %s"), name); - goto cleanup; + if (flags & VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM) { + if (virStreamSparseSendAll(st, virshStreamSource, + virshStreamInData, + virshStreamSourceSkip, &cbData) < 0) { + vshError(ctl, _("cannot send data to volume %s"), name); + goto cleanup; + } + } else { + if (virStreamSendAll(st, virshStreamSource, &cbData) < 0) { + vshError(ctl, _("cannot send data to volume %s"), name); + goto cleanup; + } } if (VIR_CLOSE(fd) < 0) { diff --git a/tools/virsh.pod b/tools/virsh.pod index dcaa0c170..42769170d 100644 --- a/tools/virsh.pod +++ b/tools/virsh.pod @@ -3924,13 +3924,14 @@ the storage volume should be deleted as well. Not all storage drivers support this option, presently only rbd. =item B<vol-upload> [I<--pool> I<pool-or-uuid>] [I<--offset> I<bytes>] -[I<--length> I<bytes>] I<vol-name-or-key-or-path> I<local-file> +[I<--length> I<bytes>] [I<--sparse>] I<vol-name-or-key-or-path> I<local-file> Upload the contents of I<local-file> to a storage volume. I<--pool> I<pool-or-uuid> is the name or UUID of the storage pool the volume is in. I<vol-name-or-key-or-path> is the name or key or path of the volume where the file will be uploaded. +If I<--sparse> is specified, this command will preserve volume sparseness. I<--offset> is the position in the storage volume at which to start writing the data. The value must be 0 or larger. I<--length> is an upper bound of the amount of data to be uploaded. A negative value is interpreted -- 2.13.0