Because of copy-paste the temporary directory used for this test is called "fakesysdir". That's probably misleading. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- tests/fdstreamtest.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/fdstreamtest.c b/tests/fdstreamtest.c index 4605845..5e82dac 100644 --- a/tests/fdstreamtest.c +++ b/tests/fdstreamtest.c @@ -314,7 +314,7 @@ static int testFDStreamWriteNonblock(const void *data) return testFDStreamWriteCommon(data, false); } -#define SCRATCHDIRTEMPLATE abs_builddir "/fakesysfsdir-XXXXXX" +#define SCRATCHDIRTEMPLATE abs_builddir "/fdstreamdir-XXXXXX" static int mymain(void) @@ -323,7 +323,7 @@ mymain(void) int ret = 0; if (!mkdtemp(scratchdir)) { - virFilePrintf(stderr, "Cannot create fakesysfsdir"); + virFilePrintf(stderr, "Cannot create fdstreamdir"); abort(); } -- 2.10.2

It helps with debugging if we know what's the return value of saferead(). Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- tests/fdstreamtest.c | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/fdstreamtest.c b/tests/fdstreamtest.c index 5e82dac..68a5bc3 100644 --- a/tests/fdstreamtest.c +++ b/tests/fdstreamtest.c @@ -250,14 +250,16 @@ static int testFDStreamWriteCommon(const char *scratchdir, bool blocking) goto cleanup; for (i = 0; i < 10; i++) { - size_t want; + size_t want, got; if (i == 9) want = PATTERN_LEN / 2; else want = PATTERN_LEN; - if (saferead(fd, buf, want) != want) { - virFilePrintf(stderr, "Short read from data\n"); + if ((got = saferead(fd, buf, want)) != want) { + virFilePrintf(stderr, + "Short read from data, i=%zu got=%zu want=%zu\n", + i, got, want); goto cleanup; } -- 2.10.2

There is really no reason why we should have to have 'struct' everywhere. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/util/virfdstream.c | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/src/util/virfdstream.c b/src/util/virfdstream.c index 75b69b6..e6670c5 100644 --- a/src/util/virfdstream.c +++ b/src/util/virfdstream.c @@ -50,6 +50,8 @@ VIR_LOG_INIT("fdstream"); /* Tunnelled migration stream support */ +typedef struct virFDStreamData virFDStreamData; +typedef virFDStreamData *virFDStreamDataPtr; struct virFDStreamData { int fd; int errfd; @@ -82,7 +84,7 @@ struct virFDStreamData { static int virFDStreamRemoveCallback(virStreamPtr stream) { - struct virFDStreamData *fdst = stream->privateData; + virFDStreamDataPtr fdst = stream->privateData; int ret = -1; if (!fdst) { @@ -119,7 +121,7 @@ static int virFDStreamRemoveCallback(virStreamPtr stream) static int virFDStreamUpdateCallback(virStreamPtr stream, int events) { - struct virFDStreamData *fdst = stream->privateData; + virFDStreamDataPtr fdst = stream->privateData; int ret = -1; if (!fdst) { @@ -151,7 +153,7 @@ static void virFDStreamEvent(int watch ATTRIBUTE_UNUSED, void *opaque) { virStreamPtr stream = opaque; - struct virFDStreamData *fdst = stream->privateData; + virFDStreamDataPtr fdst = stream->privateData; virStreamEventCallback cb; void *cbopaque; virFreeCallback ff; @@ -200,7 +202,7 @@ virFDStreamAddCallback(virStreamPtr st, void *opaque, virFreeCallback ff) { - struct virFDStreamData *fdst = st->privateData; + virFDStreamDataPtr fdst = st->privateData; int ret = -1; if (!fdst) { @@ -242,7 +244,7 @@ virFDStreamAddCallback(virStreamPtr st, } static int -virFDStreamCloseCommand(struct virFDStreamData *fdst, bool streamAbort) +virFDStreamCloseCommand(virFDStreamDataPtr fdst, bool streamAbort) { char buf[1024]; ssize_t len; @@ -295,7 +297,7 @@ virFDStreamCloseCommand(struct virFDStreamData *fdst, bool streamAbort) static int virFDStreamCloseInt(virStreamPtr st, bool streamAbort) { - struct virFDStreamData *fdst; + virFDStreamDataPtr fdst; virStreamEventCallback cb; void *opaque; int ret; @@ -378,7 +380,7 @@ virFDStreamAbort(virStreamPtr st) static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) { - struct virFDStreamData *fdst = st->privateData; + virFDStreamDataPtr fdst = st->privateData; int ret; if (nbytes > INT_MAX) { @@ -432,7 +434,7 @@ static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) { - struct virFDStreamData *fdst = st->privateData; + virFDStreamDataPtr fdst = st->privateData; int ret; if (nbytes > INT_MAX) { @@ -498,7 +500,7 @@ static int virFDStreamOpenInternal(virStreamPtr st, int errfd, unsigned long long length) { - struct virFDStreamData *fdst; + virFDStreamDataPtr fdst; VIR_DEBUG("st=%p fd=%d cmd=%p errfd=%d length=%llu", st, fd, cmd, errfd, length); @@ -754,7 +756,7 @@ int virFDStreamOpenPTY(virStreamPtr st, unsigned long long length, int oflags) { - struct virFDStreamData *fdst = NULL; + virFDStreamDataPtr fdst = NULL; struct termios rawattr; if (virFDStreamOpenFileInternal(st, path, @@ -817,7 +819,7 @@ int virFDStreamSetInternalCloseCb(virStreamPtr st, void *opaque, virFDStreamInternalCloseCbFreeOpaque fcb) { - struct virFDStreamData *fdst = st->privateData; + virFDStreamDataPtr fdst = st->privateData; virMutexLock(&fdst->lock); -- 2.10.2

While this is no functional change, it makes the code look a bit nicer. Moreover, it prepares ground for future work. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/util/virfdstream.c | 97 +++++++++++++++++++++++++++++--------------------- 1 file changed, 57 insertions(+), 40 deletions(-) diff --git a/src/util/virfdstream.c b/src/util/virfdstream.c index e6670c5..9a4a7ff 100644 --- a/src/util/virfdstream.c +++ b/src/util/virfdstream.c @@ -53,6 +53,8 @@ VIR_LOG_INIT("fdstream"); typedef struct virFDStreamData virFDStreamData; typedef virFDStreamData *virFDStreamDataPtr; struct virFDStreamData { + virObjectLockable parent; + int fd; int errfd; virCommandPtr cmd; @@ -77,10 +79,31 @@ struct virFDStreamData { virFDStreamInternalCloseCb icbCb; virFDStreamInternalCloseCbFreeOpaque icbFreeOpaque; void *icbOpaque; - - virMutex lock; }; +static virClassPtr virFDStreamDataClass; + +static void +virFDStreamDataDispose(void *obj) +{ + virFDStreamDataPtr fdst = obj; + + VIR_DEBUG("obj=%p", fdst); +} + +static int virFDStreamDataOnceInit(void) +{ + if (!(virFDStreamDataClass = virClassNew(virClassForObjectLockable(), + "virFDStreamData", + sizeof(virFDStreamData), + virFDStreamDataDispose))) + return -1; + + return 0; +} + +VIR_ONCE_GLOBAL_INIT(virFDStreamData) + static int virFDStreamRemoveCallback(virStreamPtr stream) { @@ -93,7 +116,7 @@ static int virFDStreamRemoveCallback(virStreamPtr stream) return -1; } - virMutexLock(&fdst->lock); + virObjectLock(fdst); if (fdst->watch == 0) { virReportError(VIR_ERR_INTERNAL_ERROR, "%s", _("stream does not have a callback registered")); @@ -115,7 +138,7 @@ static int virFDStreamRemoveCallback(virStreamPtr stream) ret = 0; cleanup: - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return ret; } @@ -130,7 +153,7 @@ static int virFDStreamUpdateCallback(virStreamPtr stream, int events) return -1; } - virMutexLock(&fdst->lock); + virObjectLock(fdst); if (fdst->watch == 0) { virReportError(VIR_ERR_INTERNAL_ERROR, "%s", _("stream does not have a callback registered")); @@ -143,7 +166,7 @@ static int virFDStreamUpdateCallback(virStreamPtr stream, int events) ret = 0; cleanup: - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return ret; } @@ -162,9 +185,9 @@ static void virFDStreamEvent(int watch ATTRIBUTE_UNUSED, if (!fdst) return; - virMutexLock(&fdst->lock); + virObjectLock(fdst); if (!fdst->cb) { - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return; } @@ -172,21 +195,19 @@ static void virFDStreamEvent(int watch ATTRIBUTE_UNUSED, cbopaque = fdst->opaque; ff = fdst->ff; fdst->dispatching = true; - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); cb(stream, events, cbopaque); - virMutexLock(&fdst->lock); + virObjectLock(fdst); fdst->dispatching = false; if (fdst->cbRemoved && ff) (ff)(cbopaque); closed = fdst->closed; - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); - if (closed) { - virMutexDestroy(&fdst->lock); - VIR_FREE(fdst); - } + if (closed) + virObjectUnref(fdst); } static void virFDStreamCallbackFree(void *opaque) @@ -211,7 +232,7 @@ virFDStreamAddCallback(virStreamPtr st, return -1; } - virMutexLock(&fdst->lock); + virObjectLock(fdst); if (fdst->watch != 0) { virReportError(VIR_ERR_INTERNAL_ERROR, "%s", _("stream already has a callback registered")); @@ -239,7 +260,7 @@ virFDStreamAddCallback(virStreamPtr st, ret = 0; cleanup: - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return ret; } @@ -307,7 +328,7 @@ virFDStreamCloseInt(virStreamPtr st, bool streamAbort) if (!st || !(fdst = st->privateData) || fdst->abortCallbackDispatching) return 0; - virMutexLock(&fdst->lock); + virObjectLock(fdst); /* aborting the stream, ensure the callback is called if it's * registered for stream error event */ @@ -317,7 +338,7 @@ virFDStreamCloseInt(virStreamPtr st, bool streamAbort) VIR_STREAM_EVENT_WRITABLE))) { /* don't enter this function accidentally from the callback again */ if (fdst->abortCallbackCalled) { - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return 0; } @@ -327,12 +348,12 @@ virFDStreamCloseInt(virStreamPtr st, bool streamAbort) /* cache the pointers */ cb = fdst->cb; opaque = fdst->opaque; - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); /* call failure callback, poll reports nothing on closed fd */ (cb)(st, VIR_STREAM_EVENT_ERROR, opaque); - virMutexLock(&fdst->lock); + virObjectLock(fdst); fdst->abortCallbackDispatching = false; } @@ -356,11 +377,10 @@ virFDStreamCloseInt(virStreamPtr st, bool streamAbort) if (fdst->dispatching) { fdst->closed = true; - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); } else { - virMutexUnlock(&fdst->lock); - virMutexDestroy(&fdst->lock); - VIR_FREE(fdst); + virObjectUnlock(fdst); + virObjectUnref(fdst); } return ret; @@ -395,13 +415,13 @@ static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) return -1; } - virMutexLock(&fdst->lock); + virObjectLock(fdst); if (fdst->length) { if (fdst->length == fdst->offset) { virReportSystemError(ENOSPC, "%s", _("cannot write to stream")); - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return -1; } @@ -427,7 +447,7 @@ static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) fdst->offset += ret; } - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return ret; } @@ -449,11 +469,11 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) return -1; } - virMutexLock(&fdst->lock); + virObjectLock(fdst); if (fdst->length) { if (fdst->length == fdst->offset) { - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return 0; } @@ -479,7 +499,7 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) fdst->offset += ret; } - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return ret; } @@ -505,25 +525,22 @@ static int virFDStreamOpenInternal(virStreamPtr st, VIR_DEBUG("st=%p fd=%d cmd=%p errfd=%d length=%llu", st, fd, cmd, errfd, length); + if (virFDStreamDataInitialize() < 0) + return -1; + if ((st->flags & VIR_STREAM_NONBLOCK) && virSetNonBlock(fd) < 0) { virReportSystemError(errno, "%s", _("Unable to set non-blocking mode")); return -1; } - if (VIR_ALLOC(fdst) < 0) + if (!(fdst = virObjectLockableNew(virFDStreamDataClass))) return -1; fdst->fd = fd; fdst->cmd = cmd; fdst->errfd = errfd; fdst->length = length; - if (virMutexInit(&fdst->lock) < 0) { - VIR_FREE(fdst); - virReportError(VIR_ERR_INTERNAL_ERROR, "%s", - _("Unable to initialize mutex")); - return -1; - } st->driver = &virFDStreamDrv; st->privateData = fdst; @@ -821,7 +838,7 @@ int virFDStreamSetInternalCloseCb(virStreamPtr st, { virFDStreamDataPtr fdst = st->privateData; - virMutexLock(&fdst->lock); + virObjectLock(fdst); if (fdst->icbFreeOpaque) (fdst->icbFreeOpaque)(fdst->icbOpaque); @@ -830,6 +847,6 @@ int virFDStreamSetInternalCloseCb(virStreamPtr st, fdst->icbOpaque = opaque; fdst->icbFreeOpaque = fcb; - virMutexUnlock(&fdst->lock); + virObjectUnlock(fdst); return 0; } -- 2.10.2

Currently we use iohelper for virFDStream implementation. This is because UNIX I/O can lie sometimes: even though a FD for a file/block device is set as unblocking, actual read()/write() can block. To avoid this, a pipe is created and one end is kept for read/write while the other is handed over to iohelper to write/read the data for us. Thus it's iohelper which gets blocked and not our event loop. This approach has two problems: 1) we are spawning a new process. 2) any exchange of information between daemon and iohelper can be done only through the pipe. Therefore, iohelper is replaced with an implementation in thread which is created just for the stream lifetime. The data are still transferred through pipe (for now), but both problems described above are solved. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/util/virfdstream.c | 245 +++++++++++++++++++++++++++++++------------------ src/util/virfdstream.h | 1 - 2 files changed, 158 insertions(+), 88 deletions(-) diff --git a/src/util/virfdstream.c b/src/util/virfdstream.c index 9a4a7ff..7a8d65d 100644 --- a/src/util/virfdstream.c +++ b/src/util/virfdstream.c @@ -56,8 +56,6 @@ struct virFDStreamData { virObjectLockable parent; int fd; - int errfd; - virCommandPtr cmd; unsigned long long offset; unsigned long long length; @@ -79,6 +77,11 @@ struct virFDStreamData { virFDStreamInternalCloseCb icbCb; virFDStreamInternalCloseCbFreeOpaque icbFreeOpaque; void *icbOpaque; + + /* Thread data */ + virThreadPtr thread; + int threadErr; + bool threadQuit; }; static virClassPtr virFDStreamDataClass; @@ -264,57 +267,123 @@ virFDStreamAddCallback(virStreamPtr st, return ret; } + +typedef struct _virFDStreamThreadData virFDStreamThreadData; +typedef virFDStreamThreadData *virFDStreamThreadDataPtr; +struct _virFDStreamThreadData { + virStreamPtr st; + size_t length; + int fdin; + char *fdinname; + int fdout; + char *fdoutname; +}; + + +static void +virFDStreamThreadDataFree(virFDStreamThreadDataPtr data) +{ + if (!data) + return; + + virObjectUnref(data->st); + VIR_FREE(data->fdinname); + VIR_FREE(data->fdoutname); + VIR_FREE(data); +} + + +static void +virFDStreamThread(void *opaque) +{ + virFDStreamThreadDataPtr data = opaque; + virStreamPtr st = data->st; + size_t length = data->length; + int fdin = data->fdin; + char *fdinname = data->fdinname; + int fdout = data->fdout; + char *fdoutname = data->fdoutname; + virFDStreamDataPtr fdst = st->privateData; + char *buf = NULL; + size_t buflen = 256 * 1024; + size_t total = 0; + + virObjectRef(fdst); + + if (VIR_ALLOC_N(buf, buflen) < 0) + goto error; + + while (1) { + ssize_t got; + + if (length && + (length - total) < buflen) + buflen = length - total; + + if (buflen == 0) + break; /* End of requested data from client */ + + if ((got = saferead(fdin, buf, buflen)) < 0) { + virReportSystemError(errno, + _("Unable to read %s"), + fdinname); + goto error; + } + + if (got == 0) + break; + + total += got; + + if (safewrite(fdout, buf, got) < 0) { + virReportSystemError(errno, + _("Unable to write %s"), + fdoutname); + goto error; + } + } + + cleanup: + 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; +} + + static int -virFDStreamCloseCommand(virFDStreamDataPtr fdst, bool streamAbort) +virFDStreamJoinWorker(virFDStreamDataPtr fdst, bool streamAbort) { - char buf[1024]; - ssize_t len; - int status; int ret = -1; - - if (!fdst->cmd) + if (!fdst->thread) return 0; - if ((len = saferead(fdst->errfd, buf, sizeof(buf)-1)) < 0) - buf[0] = '\0'; - else - buf[len] = '\0'; + /* Give the thread a chance to lock the FD stream object. */ + virObjectUnlock(fdst); + virThreadJoin(fdst->thread); + virObjectLock(fdst); - virCommandRawStatus(fdst->cmd); - if (virCommandWait(fdst->cmd, &status) < 0) - goto cleanup; - - if (status != 0) { - if (buf[0] != '\0') { - virReportError(VIR_ERR_INTERNAL_ERROR, "%s", buf); - } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGPIPE) { - if (streamAbort) { - /* Explicit abort request means the caller doesn't care - if there's data left over, so skip the error */ - goto out; - } - - virReportError(VIR_ERR_INTERNAL_ERROR, "%s", - _("I/O helper exited " - "before all data was processed")); - } else { - char *str = virProcessTranslateStatus(status); - virReportError(VIR_ERR_INTERNAL_ERROR, - _("I/O helper exited with %s"), - NULLSTR(str)); - VIR_FREE(str); - } + if (fdst->threadErr && !streamAbort) { + /* errors are expected on streamAbort */ goto cleanup; } - out: ret = 0; cleanup: - virCommandFree(fdst->cmd); - fdst->cmd = NULL; + VIR_FREE(fdst->thread); return ret; } + static int virFDStreamCloseInt(virStreamPtr st, bool streamAbort) { @@ -359,12 +428,9 @@ virFDStreamCloseInt(virStreamPtr st, bool streamAbort) /* mutex locked */ ret = VIR_CLOSE(fdst->fd); - if (virFDStreamCloseCommand(fdst, streamAbort) < 0) + if (virFDStreamJoinWorker(fdst, streamAbort) < 0) ret = -1; - if (VIR_CLOSE(fdst->errfd) < 0) - VIR_DEBUG("ignoring failed close on fd %d", fdst->errfd); - st->privateData = NULL; /* call the internal stream closing callback */ @@ -516,14 +582,13 @@ static virStreamDriver virFDStreamDrv = { static int virFDStreamOpenInternal(virStreamPtr st, int fd, - virCommandPtr cmd, - int errfd, + virFDStreamThreadDataPtr threadData, unsigned long long length) { virFDStreamDataPtr fdst; - VIR_DEBUG("st=%p fd=%d cmd=%p errfd=%d length=%llu", - st, fd, cmd, errfd, length); + VIR_DEBUG("st=%p fd=%d threadData=%p length=%llu", + st, fd, threadData, length); if (virFDStreamDataInitialize() < 0) return -1; @@ -538,21 +603,39 @@ static int virFDStreamOpenInternal(virStreamPtr st, return -1; fdst->fd = fd; - fdst->cmd = cmd; - fdst->errfd = errfd; fdst->length = length; st->driver = &virFDStreamDrv; st->privateData = fdst; + if (threadData) { + /* 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 (virThreadCreate(fdst->thread, + true, + virFDStreamThread, + threadData) < 0) + goto error; + } + return 0; + + error: + VIR_FREE(fdst->thread); + st->driver = NULL; + st->privateData = NULL; + virObjectUnref(fdst); + return -1; } int virFDStreamOpen(virStreamPtr st, int fd) { - return virFDStreamOpenInternal(st, fd, NULL, -1, 0); + return virFDStreamOpenInternal(st, fd, NULL, 0); } @@ -598,7 +681,7 @@ int virFDStreamConnectUNIX(virStreamPtr st, goto error; } - if (virFDStreamOpenInternal(st, fd, NULL, -1, 0) < 0) + if (virFDStreamOpenInternal(st, fd, NULL, 0) < 0) goto error; return 0; @@ -627,11 +710,10 @@ virFDStreamOpenFileInternal(virStreamPtr st, bool forceIOHelper) { int fd = -1; - int childfd = -1; + int pipefds[2] = { -1, -1 }; + int tmpfd = -1; struct stat sb; - virCommandPtr cmd = NULL; - int errfd = -1; - char *iohelper_path = NULL; + virFDStreamThreadDataPtr threadData = NULL; VIR_DEBUG("st=%p path=%s oflags=%x offset=%llu length=%llu mode=%o", st, path, oflags, offset, length, mode); @@ -648,6 +730,7 @@ virFDStreamOpenFileInternal(virStreamPtr st, path); return -1; } + tmpfd = fd; if (fstat(fd, &sb) < 0) { virReportSystemError(errno, @@ -672,7 +755,6 @@ virFDStreamOpenFileInternal(virStreamPtr st, if ((st->flags & VIR_STREAM_NONBLOCK) && ((!S_ISCHR(sb.st_mode) && !S_ISFIFO(sb.st_mode)) || forceIOHelper)) { - int fds[2] = { -1, -1 }; if ((oflags & O_ACCMODE) == O_RDWR) { virReportError(VIR_ERR_INTERNAL_ERROR, @@ -681,58 +763,47 @@ virFDStreamOpenFileInternal(virStreamPtr st, goto error; } - if (pipe(fds) < 0) { + if (pipe(pipefds) < 0) { virReportSystemError(errno, "%s", _("Unable to create pipe")); goto error; } - if (!(iohelper_path = virFileFindResource("libvirt_iohelper", - abs_topbuilddir "/src", - LIBEXECDIR))) + if (VIR_ALLOC(threadData) < 0) goto error; - cmd = virCommandNewArgList(iohelper_path, - path, - NULL); - - VIR_FREE(iohelper_path); - - virCommandAddArgFormat(cmd, "%llu", length); - virCommandPassFD(cmd, fd, - VIR_COMMAND_PASS_FD_CLOSE_PARENT); - virCommandAddArgFormat(cmd, "%d", fd); + threadData->st = virObjectRef(st); + threadData->length = length; if ((oflags & O_ACCMODE) == O_RDONLY) { - childfd = fds[1]; - fd = fds[0]; - virCommandSetOutputFD(cmd, &childfd); + threadData->fdin = fd; + threadData->fdout = pipefds[1]; + if (VIR_STRDUP(threadData->fdinname, path) < 0 || + VIR_STRDUP(threadData->fdoutname, "pipe") < 0) + goto error; + tmpfd = pipefds[0]; } else { - childfd = fds[0]; - fd = fds[1]; - virCommandSetInputFD(cmd, childfd); + threadData->fdin = pipefds[0]; + threadData->fdout = fd; + if (VIR_STRDUP(threadData->fdinname, "pipe") < 0 || + VIR_STRDUP(threadData->fdoutname, path) < 0) + goto error; + tmpfd = pipefds[1]; } - virCommandSetErrorFD(cmd, &errfd); - - if (virCommandRunAsync(cmd, NULL) < 0) - goto error; - - VIR_FORCE_CLOSE(childfd); } - if (virFDStreamOpenInternal(st, fd, cmd, errfd, length) < 0) + if (virFDStreamOpenInternal(st, tmpfd, threadData, length) < 0) goto error; return 0; error: - virCommandFree(cmd); VIR_FORCE_CLOSE(fd); - VIR_FORCE_CLOSE(childfd); - VIR_FORCE_CLOSE(errfd); - VIR_FREE(iohelper_path); + VIR_FORCE_CLOSE(pipefds[0]); + VIR_FORCE_CLOSE(pipefds[1]); if (oflags & O_CREAT) unlink(path); + virFDStreamThreadDataFree(threadData); return -1; } diff --git a/src/util/virfdstream.h b/src/util/virfdstream.h index 32a741e..34c4c3f 100644 --- a/src/util/virfdstream.h +++ b/src/util/virfdstream.h @@ -24,7 +24,6 @@ # define __VIR_FDSTREAM_H_ # include "internal.h" -# include "vircommand.h" /* internal callback, the generic one is used up by daemon stream driver */ /* the close callback is called with fdstream private data locked */ -- 2.10.2

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 7a8d65d..0350494 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; } @@ -367,6 +574,10 @@ virFDStreamJoinWorker(virFDStreamDataPtr fdst, bool streamAbort) 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); @@ -380,6 +591,7 @@ virFDStreamJoinWorker(virFDStreamDataPtr fdst, bool streamAbort) ret = 0; cleanup: VIR_FREE(fdst->thread); + virCondDestroy(&fdst->threadCond); return ret; } @@ -426,11 +638,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 */ @@ -467,7 +682,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", @@ -495,25 +711,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; } @@ -521,7 +763,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", @@ -547,24 +789,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; } @@ -609,11 +897,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, @@ -782,6 +1078,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; @@ -789,6 +1086,7 @@ virFDStreamOpenFileInternal(virStreamPtr st, VIR_STRDUP(threadData->fdoutname, path) < 0) goto error; tmpfd = pipefds[1]; + threadData->doRead = false; } } -- 2.10.2

On 04/27/2017 07:48 PM, Daniel P. Berrange wrote: I hear what you're saying but I don't think this is a problem. Exactly because of the way we wait on the condition. Assume this thread is doing reads. That is it reads from fdin (an actual file on a disk) and feeds the message queue with the data. Now, it reads some data, push it at the end of the message queue (fdst->msg != NULL at that point) and goes to the while loop above. Effectively it eats all the spurious wake ups for as long as fdst->msg != NULL. Now assume the thread is doing writes (read data from the queue and writes into a file a disk). Again, as long as there are some messages to be written (that is fdst->msg != NULL) the control won't even reach virCondWait. And if it does, it's because fdst->msg == NULL in which case there is no data to be written. But something has just came up to my mind whilst writing these lines - I wonder if we can ditch the condition entirely and rely on the pipe + poll(). I mean, this worker would use pipe to signalize to the event loop that there is a message waiting for it in the queue. Question is how thread safe this approach would be. Michal

After 1eb6647979f8c nobody calls the iohelper with 6 arguments. Everybody uses the other mode. Well, the only user of iohelper after the previous commit is virFileWrapperFd really. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/util/iohelper.c | 72 ++++------------------------------------------------- 1 file changed, 5 insertions(+), 67 deletions(-) diff --git a/src/util/iohelper.c b/src/util/iohelper.c index 00f31e7..d7bf5c7 100644 --- a/src/util/iohelper.c +++ b/src/util/iohelper.c @@ -44,35 +44,6 @@ #define VIR_FROM_THIS VIR_FROM_STORAGE static int -prepare(const char *path, int oflags, int mode, - unsigned long long offset) -{ - int fd = -1; - - if (oflags & O_CREAT) { - fd = open(path, oflags, mode); - } else { - fd = open(path, oflags); - } - if (fd < 0) { - virReportSystemError(errno, _("Unable to open %s"), path); - goto cleanup; - } - - if (offset) { - if (lseek(fd, offset, SEEK_SET) < 0) { - virReportSystemError(errno, _("Unable to seek %s to %llu"), - path, offset); - VIR_FORCE_CLOSE(fd); - goto cleanup; - } - } - - cleanup: - return fd; -} - -static int runIO(const char *path, int fd, int oflags, unsigned long long length) { void *base = NULL; /* Location to be freed */ @@ -207,9 +178,7 @@ usage(int status) if (status) { fprintf(stderr, _("%s: try --help for more details"), program_name); } else { - printf(_("Usage: %s FILENAME OFLAGS MODE OFFSET LENGTH DELETE\n" - " or: %s FILENAME LENGTH FD\n"), - program_name, program_name); + printf(_("Usage: %s FILENAME LENGTH FD\n"), program_name); } exit(status); } @@ -218,13 +187,9 @@ int main(int argc, char **argv) { const char *path; - unsigned long long offset; unsigned long long length; int oflags = -1; - int mode; - unsigned int delete = 0; int fd = -1; - int lengthIndex = 0; program_name = argv[0]; @@ -239,31 +204,13 @@ main(int argc, char **argv) if (argc > 1 && STREQ(argv[1], "--help")) usage(EXIT_SUCCESS); - if (argc == 7) { /* FILENAME OFLAGS MODE OFFSET LENGTH DELETE */ - lengthIndex = 5; - if (virStrToLong_i(argv[2], NULL, 10, &oflags) < 0) { - fprintf(stderr, _("%s: malformed file flags %s"), + if (argc == 4) { /* FILENAME LENGTH FD */ + if (virStrToLong_ull(argv[2], NULL, 10, &length) < 0) { + fprintf(stderr, _("%s: malformed file length %s"), program_name, argv[2]); exit(EXIT_FAILURE); } - if (virStrToLong_i(argv[3], NULL, 10, &mode) < 0) { - fprintf(stderr, _("%s: malformed file mode %s"), - program_name, argv[3]); - exit(EXIT_FAILURE); - } - if (virStrToLong_ull(argv[4], NULL, 10, &offset) < 0) { - fprintf(stderr, _("%s: malformed file offset %s"), - program_name, argv[4]); - exit(EXIT_FAILURE); - } - if (argc == 7 && virStrToLong_ui(argv[6], NULL, 10, &delete) < 0) { - fprintf(stderr, _("%s: malformed delete flag %s"), - program_name, argv[6]); - exit(EXIT_FAILURE); - } - fd = prepare(path, oflags, mode, offset); - } else if (argc == 4) { /* FILENAME LENGTH FD */ - lengthIndex = 2; + if (virStrToLong_i(argv[3], NULL, 10, &fd) < 0) { fprintf(stderr, _("%s: malformed fd %s"), program_name, argv[3]); @@ -287,18 +234,9 @@ main(int argc, char **argv) usage(EXIT_FAILURE); } - if (virStrToLong_ull(argv[lengthIndex], NULL, 10, &length) < 0) { - fprintf(stderr, _("%s: malformed file length %s"), - program_name, argv[lengthIndex]); - exit(EXIT_FAILURE); - } - if (fd < 0 || runIO(path, fd, oflags, length) < 0) goto error; - if (delete) - unlink(path); - return 0; error: -- 2.10.2

This function takes a FD and determines whether the current position is in data section or in a hole. In addition to that, it also determines how much bytes are there remaining till the current section ends. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/libvirt_private.syms | 1 + src/util/virfile.c | 81 +++++++++++++++++++ src/util/virfile.h | 3 + tests/virfiletest.c | 203 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 288 insertions(+) diff --git a/src/libvirt_private.syms b/src/libvirt_private.syms index 181e178..7132b3a 100644 --- a/src/libvirt_private.syms +++ b/src/libvirt_private.syms @@ -1620,6 +1620,7 @@ virFileGetHugepageSize; virFileGetMountReverseSubtree; virFileGetMountSubtree; virFileHasSuffix; +virFileInData; virFileIsAbsPath; virFileIsDir; virFileIsExecutable; diff --git a/src/util/virfile.c b/src/util/virfile.c index cbfa384..093125c 100644 --- a/src/util/virfile.c +++ b/src/util/virfile.c @@ -3793,6 +3793,87 @@ virFileComparePaths(const char *p1, const char *p2) cleanup: VIR_FREE(res1); VIR_FREE(res2); + + return ret; +} + + +int virFileInData(int fd, + int *inData, + unsigned long long *length) +{ + int ret = -1; + off_t cur, data, hole, end; + + /* Get current position */ + cur = lseek(fd, 0, SEEK_CUR); + if (cur == (off_t) -1) { + virReportSystemError(errno, "%s", + _("Unable to get current position in file")); + goto cleanup; + } + + /* Now try to get data and hole offsets */ + data = lseek(fd, cur, SEEK_DATA); + + /* There are four options: + * 1) data == cur; @cur is in data + * 2) data > cur; @cur is in a hole, next data at @data + * 3) data < 0, errno = ENXIO; either @cur is in trailing hole, or @cur is beyond EOF. + * 4) data < 0, errno != ENXIO; we learned nothing + */ + + if (data == (off_t) -1) { + /* cases 3 and 4 */ + if (errno != ENXIO) { + virReportSystemError(errno, "%s", + _("Unable to seek to data")); + goto cleanup; + } + + *inData = 0; + /* There are two situations now. There is always an + * implicit hole at EOF. However, there might be a + * trailing hole just before EOF too. If that's the case + * report it. */ + if ((end = lseek(fd, 0, SEEK_END)) == (off_t) -1) { + virReportSystemError(errno, "%s", + _("Unable to seek to EOF")); + goto cleanup; + } + *length = end - cur; + } else if (data > cur) { + /* case 2 */ + *inData = 0; + *length = data - cur; + } else { + /* case 1 */ + *inData = 1; + + /* We don't know where does the next hole start. Let's + * find out. Here we get the same 4 possibilities as + * described above.*/ + hole = lseek(fd, data, SEEK_HOLE); + if (hole == (off_t) -1 || hole == data) { + /* cases 1, 3 and 4 */ + /* Wait a second. The reason why we are here is + * because we are in data. But at the same time we + * are in a trailing hole? Wut!? Do the best what we + * can do here. */ + virReportSystemError(errno, "%s", + _("unable to seek to hole")); + goto cleanup; + } else { + /* case 2 */ + *length = (hole - data); + } + } + + ret = 0; + cleanup: + /* At any rate, reposition back to where we started. */ + if (cur != (off_t) -1) + ignore_value(lseek(fd, cur, SEEK_SET)); return ret; } diff --git a/src/util/virfile.h b/src/util/virfile.h index 41c25a2..32a1d3c 100644 --- a/src/util/virfile.h +++ b/src/util/virfile.h @@ -340,4 +340,7 @@ int virFileReadValueInt(const char *path, int *value); int virFileReadValueUint(const char *path, unsigned int *value); int virFileReadValueBitmap(const char *path, int maxlen, virBitmapPtr *value); +int virFileInData(int fd, + int *inData, + unsigned long long *length); #endif /* __VIR_FILE_H */ diff --git a/tests/virfiletest.c b/tests/virfiletest.c index 702a76a..3e298dc 100644 --- a/tests/virfiletest.c +++ b/tests/virfiletest.c @@ -21,6 +21,7 @@ #include <config.h> #include <stdlib.h> +#include <fcntl.h> #include "testutils.h" #include "virfile.h" @@ -119,6 +120,190 @@ testFileSanitizePath(const void *opaque) static int +makeSparseFile(const off_t offsets[], + const bool startData); + +#ifdef __linux__ +/* Create a sparse file. @offsets in KiB. */ +static int +makeSparseFile(const off_t offsets[], + const bool startData) +{ + int fd = -1; + char path[] = abs_builddir "fileInData.XXXXXX"; + off_t len = 0; + size_t i; + + if ((fd = mkostemp(path, O_CLOEXEC|O_RDWR)) < 0) + goto error; + + if (unlink(path) < 0) + goto error; + + for (i = 0; offsets[i] != (off_t) -1; i++) + len += offsets[i] * 1024; + + while (len) { + const char buf[] = "abcdefghijklmnopqrstuvwxyz"; + off_t toWrite = sizeof(buf); + + if (toWrite > len) + toWrite = len; + + if (safewrite(fd, buf, toWrite) < 0) { + fprintf(stderr, "unable to write to %s (errno=%d)\n", path, errno); + goto error; + } + + len -= toWrite; + } + + len = 0; + for (i = 0; offsets[i] != (off_t) -1; i++) { + bool inData = startData; + + if (i % 2) + inData = !inData; + + if (!inData && + fallocate(fd, + FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE, + len, offsets[i] * 1024) < 0) { + fprintf(stderr, "unable to punch a hole at offset %lld length %lld\n", + (long long) len, (long long) offsets[i]); + goto error; + } + + len += offsets[i] * 1024; + } + + if (lseek(fd, 0, SEEK_SET) == (off_t) -1) { + fprintf(stderr, "unable to lseek (errno=%d)\n", errno); + goto error; + } + + return fd; + error: + VIR_FORCE_CLOSE(fd); + return -1; +} + +#else /* !__linux__ */ + +static int +makeSparseFile(const off_t offsets[] ATTRIBUTE_UNUSED, + const bool startData ATTRIBUTE_UNUSED) +{ + return -1; +} + +#endif /* !__linux__ */ + + +#define EXTENT 4 +static bool +holesSupported(void) +{ + off_t offsets[] = {EXTENT, EXTENT, EXTENT, -1}; + off_t tmp; + int fd; + bool ret = false; + + if ((fd = makeSparseFile(offsets, true)) < 0) + goto cleanup; + + /* The way this works is: there are 4K of data followed by 4K hole followed + * by 4K hole again. Check if the filesystem we are running the test suite + * on supports holes. */ + if ((tmp = lseek(fd, 0, SEEK_DATA)) == (off_t) -1) + goto cleanup; + + if (tmp != 0) + goto cleanup; + + if ((tmp = lseek(fd, tmp, SEEK_HOLE)) == (off_t) -1) + goto cleanup; + + if (tmp != EXTENT * 1024) + goto cleanup; + + if ((tmp = lseek(fd, tmp, SEEK_DATA)) == (off_t) -1) + goto cleanup; + + if (tmp != 2 * EXTENT * 1024) + goto cleanup; + + if ((tmp = lseek(fd, tmp, SEEK_HOLE)) == (off_t) -1) + goto cleanup; + + if (tmp != 3 * EXTENT * 1024) + goto cleanup; + + ret = true; + cleanup: + VIR_FORCE_CLOSE(fd); + return ret; +} + + +struct testFileInData { + bool startData; /* whether the list of offsets starts with data section */ + off_t *offsets; +}; + + +static int +testFileInData(const void *opaque) +{ + const struct testFileInData *data = opaque; + int fd = -1; + int ret = -1; + size_t i; + + if ((fd = makeSparseFile(data->offsets, data->startData)) < 0) + goto cleanup; + + for (i = 0; data->offsets[i] != (off_t) -1; i++) { + bool shouldInData = data->startData; + int realInData; + unsigned long long shouldLen; + unsigned long long realLen; + + if (i % 2) + shouldInData = !shouldInData; + + if (virFileInData(fd, &realInData, &realLen) < 0) + goto cleanup; + + if (realInData != shouldInData) { + fprintf(stderr, "Unexpected data/hole. Expected %s got %s\n", + shouldInData ? "data" : "hole", + realInData ? "data" : "hole"); + goto cleanup; + } + + shouldLen = data->offsets[i] * 1024; + if (realLen != shouldLen) { + fprintf(stderr, "Unexpected section length. Expected %lld got %lld\n", + shouldLen, realLen); + goto cleanup; + } + + if (lseek(fd, shouldLen, SEEK_CUR) < 0) { + fprintf(stderr, "Unable to seek\n"); + goto cleanup; + } + } + + ret = 0; + + cleanup: + VIR_FORCE_CLOSE(fd); + return ret; +} + + +static int mymain(void) { int ret = 0; @@ -186,6 +371,24 @@ mymain(void) DO_TEST_SANITIZE_PATH_SAME("gluster://bar.baz/fooo//hoo"); DO_TEST_SANITIZE_PATH_SAME("gluster://bar.baz/fooo///////hoo"); +#define DO_TEST_IN_DATA(inData, ...) \ + do { \ + off_t offsets[] = {__VA_ARGS__, -1}; \ + struct testFileInData data = { \ + .startData = inData, .offsets = offsets, \ + }; \ + if (virTestRun(virTestCounterNext(), testFileInData, &data) < 0) \ + ret = -1; \ + } while (0) + + if (holesSupported()) { + DO_TEST_IN_DATA(true, 4, 4, 4); + DO_TEST_IN_DATA(false, 4, 4, 4); + DO_TEST_IN_DATA(true, 8, 8, 8); + DO_TEST_IN_DATA(false, 8, 8, 8); + DO_TEST_IN_DATA(true, 8, 16, 32, 64, 128, 256, 512); + DO_TEST_IN_DATA(false, 8, 16, 32, 64, 128, 256, 512); + } return ret != 0 ? EXIT_FAILURE : EXIT_SUCCESS; } -- 2.10.2

On 04/29/2017 06:46 PM, John Ferlan wrote: We shouldn't fail. All the tools that I've seen and do support sparse files do accept fully allocated files gracefully, e.g. cp --sparse allways or rsync -S. I mean, even though this function you're suggesting here would fail or something, the whole stream transmission should just carry on. No. I will not. But it doesn't matter if the first hole is at EOF or before it, if we know how to transfer holes effectively. Firstly, this is merely what I'm implementing later in the series (see patch 33/38 [and I just realized that due to a mistake patch 34/38 has wrong commit message]). Secondly, getting current position in a file should be as quick as this (imagine me snapping my fingers :-)). The VFS has to keep track of the current position in a file for sake of all file operations. /me goes and looks into kernel sources... Yep, it looks like so: generic_file_llseek() from fs/read_write.c. But the main point is .. ... that here, in this branch, we should send the data from the data section to client, right? But how can we do that if we lseek()-ed away (to the end of the section)?. I mean virFileStreamIsInHole() would need to seek back to the original current position in the file anyway. Just like virFileInData() does. Otherwise, if a file is in data section we would not read it. Okay, I'll add some description. Yeah, it can be a bool. It's for both. So the whole point of sparse streams is to preserve sparseness as volume images are copied from daemon to client (virsh vol-download) or from client to daemon (virsh vol-upload). Currently what we have is something among the following lines: while (true) { got = read(file, buf, sizeof(buf)); if (got < 0 || got == 0) break; if (virStreamSend(stream, buf, got) < 0) break; } (Simplified very much, look at virStreamSendAll + virStreamRecvAll). As you can see, we don't care about the holes. We don't know about holes at all. So the point of this whole feature is to still copy the data AND preserve sparseness at the same time: while (true) { virFileInData(fd, &inData, &offset); if (inData) { while (offset) { got = read(fd, buf, sizeof(buf)); virStreamSend(stream, buf, got); offset -= got; } } else { virStreamSendHole(stream, offset); lseek(fd, offset, SEEK_CUR); } } Here we can clearly see that virFileInData() must leave the position in @fd unchanged upon its return. Because of file size. Imagine a file with the following structure |- data 1MB -||- hole 1MB -||- EOF -| The file is clearly 2MBs in size, right? Now, my function would report 2 holes: the first one of 1MB size and the second implicit one at EOF. If we were to merge those two holes into one, the file would shrink in size and basically become a different file. The problem is we don't transfer any kind of metadata throughout the copying process. The other side doesn't know how big is the file we're copying; what is its data/hole structure, nothing. What do you mean? What error checks? This code can be simpler because when entering this branch we know that we are in a data section. And we just want to know its length. So we call SEEK_HOLE. Well, that's the point. The phrasing of the question is very important. How can we know there's no hole from @cur to EOF if we don't go through this branch? And after we've gone through it (and after 33/38) we will no longer call this function at all. Not sure. If the kernel is giving me spurious data how can I believe any subsequent read()? Probably the other condition hole == data is never going to happen, but I thought better be sure than sorry. Yep, that one is valid too. I can add it to the set of tests below, Ha! my compiler doesn't complain. Fixed. Fair enough. This is not a trivial feature. It touches parts of libvirt that perhaps we all are scared to touch. So I expect some misconceptions need to be cleared. Michal

On 05/03/2017 11:13 PM, John Ferlan wrote: Sure. And that is what happens. Well, sort of. With this patches the following can happen: Sparse Stream Req Yes | No FS -----------+------------ supports Yes Success | Fallback holes No Error | Fallback A problem might be if FS doesn't support holes (=lseek(SEEK_HOLE) returns an error) as we error out in that case. On the other hand, that's libvirt philosophy - if user requested a feature we cannot meet, we should error out. They can always retry with the feature turned off. Or, we can introduce another flag: VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM_OPT which silently falls back to non-sparse streams if underlying FS doesn't support holes. BTW: I don't think there's a filesystem now that wouldn't support holes. Right, ext2 doesn't but is there a machine that still uses ext2 and runs latest libvirt? I don't think so. Therefore, the following table is more important: Sparse Stream Req Yes | No -----------+------------ sparse Yes Success | Fallback file No Success* | Fallback * - This is where we effectively fallback to old method; simply because VIR_NET_STREAM_SKIP is never sent and thus the new code paths are never triggered. Basically, at the beginning of the transfer we detect that there's X bytes long data section. So we send bare data (no skip/hole). After we've sent all the data we ask again and find out that we're at EOF so the stream is finished. Well, it's up to caller to ensure that. But since this function is called from basically one place when transferring files, we are safe here - the FD that is passed to this function is dealt with from just one threat. We don't support interrupting streams. While this is what 99.9% users do, we still offer possibility to transfer just [X:Y] bytes of a file. Again, we are not getting @cur each time we want to read from the file. In the end, this function is called at section boundaries. So if you have: 0 -> data -> hole -> data -> hole -> data -> hole -> data -> EOF this function is going to be called 8 times in total. Regardless of size of each section. That is not bad. I still don't think lseek() is a problem. I've ran a test on my laptop and was able to do as much as 20M lseek()-s per second. IOW, one lseek() call was about 50ns. So I don't see any performance issue there. There are places in libvirt where we sleep for seconds. Compare that to 50ns spent here. I don't think so. Consider the following: kernel already has to keep a map of data/hole sections. I don't see a reason for us to duplicate that in userspace. We would have keep track of position in the file. Well, duplicate - kernel already does that. Then keep track of data/hole sections. Again, duplicate. And for what? Saving couple of nanoseconds? Well, we would lose them while maintaining our own copies of information that kernel already has. Well, as I've said above, it's not always 0->EOF copy of a file. Users can specify "I want copy of this volume form offset O, X bytes long". At the first call of virFileInData, the @fd is positioned at offset O. What if users chose O beyond EOF? Sure, we cannot. It's a file. Nothing stops you from overwriting it under libvirt's hands. At that point, any track (copy of metadata) we have is stale. Perhaps no. But I wanted to be safe than sorry. Yes. Not now. There might be one in the future. But user can already assemble this information. I mean, they can fetch volume info and learn how big the volume is. Then when they are doing the copy they know how much data has been transferred in the stream (the sum of all virStreamRecv() values). You can do that with the current APIs. I don't think we will ever need to expose that kind of information. But never say never. But that has nothing to do with this feature anyway. The latter. Michal

On 04/20/2017 06:01 AM, Michal Privoznik wrote: Could just simply state this patch is adding the virStreamRecvFlags as a variant to the virStreamRecv function in order to allow for future expansion of functionality for processing sparse streams using a @flags argument. [...] Instead, consider: Calling this function with no @flags set (equal to zero) is equivalent to calling virStreamRecv(stream, data, nbytes). Should 'zi' be 'zu'? I realize this is cut-n-paste, so perhaps other usages could be fixed in a trivial patch as well. This will be (at least) 3.4.0... ACK w/ a couple of adjustments. Feels strange to not see remote_protocol.x changed at the same time though... John

On 04/20/2017 06:01 AM, Michal Privoznik wrote: s/can be/is (unless it can be used for something else ;-)) lseek is an implementation detail... However, it could be stated that the skipping would be from the current point in the file forward by some number of bytes. It's expected to be used in conjunction with code that is copying over the real (or non-zero) data and should be considered an optimization over sending zere data segments. While it would be unused for now, should @flags be added. Who knows what use it could have, but avoids a new Flags API, but does cause a few other wording changes here. Perhaps it's just me - but "Skip" and "HoleSize" just seem awkward. Would "virStreamSetSkip" and "virStreamGetSkip" be more apropos? (or s/Skip/HoleSize/ - ewww). Names would then follow our more recent function naming guidelines. I think I dislike the HoleSize much more than the Skip. Was there consideration for using 'off_t' instead of ULL? I know it's an implementation detail of virFDStreamData and lseek() usage, but it does hide things... IDC either way. Consider something like: Rather than transmitting empty file space, this API directs the @stream target to create @length bytes of empty space. This API would be used when uploading or downloading sparsely populated files to avoid the needless copy of empty file space. Although I suppose there could be more applications than file targets. I'd say ACK, but I think the function name discussion needs to be complete first... John

On 05/05/2017 07:25 AM, Michal Privoznik wrote: I understand completely (in more ways than one)! However, I still have the hacking/coding style guide to fallback upon that requires certain syntax ;-)... While Skip "qualifies" as a verb and perhaps squeaks by on that technicality - the "HoleSize" has no verb telling me what HoleSize is doing. The dichotomy that "Skip" doesn't have a Size object, but Hole does isn't lost either. I actually think Size is a "given", but as someone who has trouble creating names that pass muster when my code is reviewed - who am I to give too much advice here! Also now that I'm further down the road - I keep having to attempt to remind myself whether this is a we're setting/performing the set skip operation or this is a we're getting the set hole size operation. Function, RPC, structures, etc. names would help unmuddle things. The thing is the implementation uses a function that expects an off_t. When I see ULL I'm usually considering memory sizes which can be large. Not that a file couldn't be, but it usually isn't. The concern is - could we run into a situation where a coding error somewhere supplies a negative value that will now appear to be some inordinately large positive value; whereas, we could avoid that by stipulating 'virStreamSkip' (or whatever it gets called) expects a positive and/or greater than 0 value (e.g. virCheckPositiveArgGoto). Also as I see it, virFileInData calculates *length from two off_t's and in none of those calculations is it possible to generate a negative number. If we did, something is wrong. Still since virStreamSkip can be called and not necessarily use that calculation. I'll also point out the coding example provided for virStreamSkip: + * while (1) { + * char buf[4096]; + * size_t len; <===== + * if (..in hole...) { + * ..get hole size... + * virStreamSkip(st, len); ^^^^ <=== Not a ULL... Although the corollary did use the ULL for virStreamHoleSize in its example. Similar argument for virStreamHoleSize could be made - we shouldn't receive a negative value, but we have no way of differentiating a coding mistake and an inordinately large value. At least by enforcing usage of 'off_t' we're saying - go forward... John

On 05/05/2017 04:48 PM, Daniel P. Berrange wrote: So I've given this some though and ran some experiments. On a 32bit arch I've found this: long long 8 signed size_t 4 unsigned ssize_t 4 signed off_t 4 signed So size_t is 4 bytes long and long long is 8 bytes. This got me thinking, size_t type makes sense for those APIs where we need to address individual bytes. But what would happen if I have the following file on a 32 bit arch: [2MB data] -> [5GB hole] -> [2M data] The hole does not fit into size_t, but it would fit into long long. On the other hand, we are very likely to hit lseek() error as off_t is 4 bytes also (WTF?!). On a 64 bit arch everything is as expected: long long 8 signed size_t 8 unsigned ssize_t 8 signed off_t 8 signed So after all, I think I can switch to size_t. I'm just really surprised to learn that off_t is 4 bytes on a 32 bit arch. Michal

On 05/12/2017 12:56 PM, John Ferlan wrote: Sure it can. Kernel might use long long internally. It definitely has to use bigger type than size_t otherwise we couldn't have >4GB files. But we can. BTW there are other ways to detect holes in files. For instance 'cp' uses ioctl(fd, FS_IOC_FIEMAP, fiemap) to get map of extents on the disk. The fiemap struct then uses uint64_t type to represent addresses and lengths for extents. Therefore I'm gonna stick with my decision and have this unsigned long long. The other argument for using that is: we want ULL to be used at RPC level, right? However, this might clash when mixing 32 and 64 bit client/server as decoding into size_t might fail - ULL does not fit into size_t on 32 bits. Here's an example of a big sparse file: rpi ~ # truncate -s 5G bla.raw rpi ~ # ls -lhs bla.raw 0 -rw-r--r-- 1 root root 5.0G May 12 14:57 bla.raw rpi ~ # uname -a Linux rpi 4.1.15-v7+ #830 SMP Tue Dec 15 17:02:45 GMT 2015 armv7l ARMv7 Processor rev 5 (v7l) BCM2709 GNU/Linux rpi ~ # /home/zippy/tmp/sparse/sparse_map bla.raw hole: 0 len: 5368709120 end: 5368709120 sparse_map is a program I've written for getting data/hole map of a given file. You can find it here: https://pastebin.com/SSjAUi3q BTW: try changing those (long long) typecasts to (size_t) and you'll immediately see why we need long long instead of size_t. Michal

On 05/15/2017 10:25 AM, Daniel P. Berrange wrote: Standalone demo program, that basically does this: #define PRINT_SIZE(x) printf( #x " %d %s\n", sizeof(x), (x)-1 < 0 ? "signed" : "unsigned") PRINT_SIZE(long long); PRINT_SIZE(size_t); PRINT_SIZE(off_t); I think we really need just long long. Consider that even though libvirt enables large file support internally, we don't enable it at the header files level. It's responsibility of developers of mgmt application to do that. Having said that, if we had off_t in public API we can't possibly know its size as it depends on something out of our control. Therefore, I see long long as our only option. Now, question is whether we want signed or unsigned long long. I don't have an opinion about that. On one hand, off_t is signed, but that's because lseek() can seek backwards. We don't have that in our streams. Yet. On the other hand, our streams are different to regular files. I view them as a unidirectional pipe. With some extensions (e.g. sparse messages). lseek() doesn't work over pipes, does it. But then again, long long might be more future proof, if we will ever want to assign a meaning to negative seeks. Michal

On 04/20/2017 06:01 AM, Michal Privoznik wrote: s/should/would need to/ Again the naming doesn't necessarily conform to our newer standard, but that's something I'd expect we've already discussed in patch 11... Likewise should a @flags argument be added? Hmmm.. this last sentence belongs in the "Returns" space. Consider 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 virStream....(). s/on error/on error or when there's currently no hole in the stream Again, similar to last patch - ACK after the API naming discussion... John

On 04/20/2017 06:01 AM, Michal Privoznik wrote: Consider: 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. Obvious API name impact from patch 11/12... Consider perhaps VIR_STREAM_RECV_HANDLE_SPARSE - or something that indicates sparse management. ...if ((ret = vir*()) < 0) report error ? IDC exactly, just noting. The entire code snippet needs an extra space indent; otherwise, it doesn't look right on the generated web page. I would think this is ACK-able w/ the minor adjustments - there could be side effects from previous adjustments though. John

On 04/20/2017 06:01 AM, Michal Privoznik wrote: You could have some API name adjustments not only in the commit message but also within code. The callback may be invoked multiple times as holes are found during processing a stream. The application should create the hole in the stream target and then return. A return value of -1 at any time will abort the receive operation. s/sink./sink @handler and calling the skip @holeHandler to generate holes for sparse stream targets. s/virStreamRecv/virStreamRecvAll/ s/Free/Free(). s/, so/, so/ s/Free()/Free()./ (e.g. lose the extra space and add a period) Could virStreamFree(). go on the previous line? We could continue; here, right? Although, I suppose (0 < -3) would fail anyway, but if someone changed that someday and didn't read this part... An ACK ends up being a bit dependent upon previous patches... What's here seems fine to me though. John

On 04/20/2017 06:01 AM, Michal Privoznik wrote: You could have some changes here due to previous API name changes. Still the commit msg is a bit terse needs some cleanup anyway. s/much/many s/there// s/until/until the/ s/be it/either/ s/) and if/). Also if/ Need extra line between paragraphs s/As a corner case,/NB:/ ... @inData and @length should both be set to 0. (reads better to me) Need extra line between paragraphs Consider instead: This function is not expected to adjust the current position within the file. (or more explicitly - should not adjust!) 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. Spurious extra line... Now there's 3 between functions. Some smart-aleck comment here ;-) So if !inData and sectionLen == 0, we don't go here, but We go here indicating 'dataLen = 0' which is a misnomer since we're not inData, rather we at EOF, right? Or did I miss something a bit subtle. Need to clean up the spacing here... e.g. } /** Another one close for an ACK - some cleanup here as well as seeing the fallout from previous patches. John

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 | 43 +++++++++++++++++++++++++++++++++++++++++++ src/libvirt_internal.h | 3 +++ src/libvirt_private.syms | 1 + 4 files changed, 53 insertions(+) diff --git a/src/driver-stream.h b/src/driver-stream.h index e196b6d..5d84b9a 100644 --- a/src/driver-stream.h +++ b/src/driver-stream.h @@ -50,6 +50,11 @@ typedef int unsigned long long *length); typedef int +(*virDrvStreamInData)(virStreamPtr st, + int *data, + unsigned long long *length); + +typedef int (*virDrvStreamEventAddCallback)(virStreamPtr stream, int events, virStreamEventCallback cb, @@ -78,6 +83,7 @@ struct _virStreamDriver { virDrvStreamRecvFlags streamRecvFlags; virDrvStreamSkip streamSkip; virDrvStreamHoleSize streamHoleSize; + virDrvStreamInData streamInData; virDrvStreamEventAddCallback streamEventAddCallback; virDrvStreamEventUpdateCallback streamEventUpdateCallback; virDrvStreamEventRemoveCallback streamEventRemoveCallback; diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c index 7ad5a38..975315e 100644 --- a/src/libvirt-stream.c +++ b/src/libvirt-stream.c @@ -473,6 +473,49 @@ virStreamHoleSize(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 hold. 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). + * + * As a special case, there's an implicit hole at EOF. In this + * situation this function should set @data = false, @length = 0 + * and return 0. + * + * Returns 0 on success, + * -1 otherwise + */ +int +virStreamInData(virStreamPtr stream, + int *data, + unsigned 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 * @handler: source callback for reading data from application diff --git a/src/libvirt_internal.h b/src/libvirt_internal.h index 96439d8..0e945aa 100644 --- a/src/libvirt_internal.h +++ b/src/libvirt_internal.h @@ -294,4 +294,7 @@ virTypedParameterValidateSet(virConnectPtr conn, virTypedParameterPtr params, int nparams); +int virStreamInData(virStreamPtr stream, + int *data, + unsigned long long *lengtht); #endif diff --git a/src/libvirt_private.syms b/src/libvirt_private.syms index 7132b3a..07633f2 100644 --- a/src/libvirt_private.syms +++ b/src/libvirt_private.syms @@ -1117,6 +1117,7 @@ virStateCleanup; virStateInitialize; virStateReload; virStateStop; +virStreamInData; # locking/domain_lock.h -- 2.10.2

This has no real added value right now, but is going to be very helpful later. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 6 ++++-- src/rpc/gendispatch.pl | 2 +- src/rpc/virnetclientstream.c | 6 +++++- src/rpc/virnetclientstream.h | 3 ++- 4 files changed, 12 insertions(+), 5 deletions(-) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 718e322..7024464 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -6170,7 +6170,8 @@ remoteDomainMigratePrepareTunnel3(virConnectPtr dconn, memset(&args, 0, sizeof(args)); memset(&ret, 0, sizeof(ret)); - if (!(netst = virNetClientStreamNew(priv->remoteProgram, + if (!(netst = virNetClientStreamNew(st, + priv->remoteProgram, REMOTE_PROC_DOMAIN_MIGRATE_PREPARE_TUNNEL3, priv->counter))) goto done; @@ -7094,7 +7095,8 @@ remoteDomainMigratePrepareTunnel3Params(virConnectPtr dconn, goto cleanup; } - if (!(netst = virNetClientStreamNew(priv->remoteProgram, + if (!(netst = virNetClientStreamNew(st, + priv->remoteProgram, REMOTE_PROC_DOMAIN_MIGRATE_PREPARE_TUNNEL3_PARAMS, priv->counter))) goto cleanup; diff --git a/src/rpc/gendispatch.pl b/src/rpc/gendispatch.pl index 173189c..e608812 100755 --- a/src/rpc/gendispatch.pl +++ b/src/rpc/gendispatch.pl @@ -1738,7 +1738,7 @@ elsif ($mode eq "client") { if ($call->{streamflag} ne "none") { print "\n"; - print " if (!(netst = virNetClientStreamNew(priv->remoteProgram, $call->{constname}, priv->counter)))\n"; + print " if (!(netst = virNetClientStreamNew(st, priv->remoteProgram, $call->{constname}, priv->counter)))\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 2105bd0..01761cf 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -36,6 +36,8 @@ VIR_LOG_INIT("rpc.netclientstream"); struct _virNetClientStream { virObjectLockable parent; + virStreamPtr stream; /* Reverse pointer to parent stream */ + virNetClientProgramPtr prog; int proc; unsigned serial; @@ -133,7 +135,8 @@ virNetClientStreamEventTimer(int timer ATTRIBUTE_UNUSED, void *opaque) } -virNetClientStreamPtr virNetClientStreamNew(virNetClientProgramPtr prog, +virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, + virNetClientProgramPtr prog, int proc, unsigned serial) { @@ -145,6 +148,7 @@ virNetClientStreamPtr virNetClientStreamNew(virNetClientProgramPtr prog, if (!(st = virObjectLockableNew(virNetClientStreamClass))) return NULL; + st->stream = stream; st->prog = virObjectRef(prog); st->proc = proc; st->serial = serial; diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index a0d2be9..e278dab 100644 --- a/src/rpc/virnetclientstream.h +++ b/src/rpc/virnetclientstream.h @@ -32,7 +32,8 @@ typedef virNetClientStream *virNetClientStreamPtr; typedef void (*virNetClientStreamEventCallback)(virNetClientStreamPtr stream, int events, void *opaque); -virNetClientStreamPtr virNetClientStreamNew(virNetClientProgramPtr prog, +virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, + virNetClientProgramPtr prog, int proc, unsigned serial); -- 2.10.2

Even though there's no way to make stream skippable right now, it is going to be soon. We need to track this info so that we don't send virStreamSkip to a client that did not want it or vice versa. 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 1610fea..fbd5ba2 100644 --- a/daemon/remote.c +++ b/daemon/remote.c @@ -5373,7 +5373,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 11c0a46..624a626 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -52,6 +52,8 @@ struct daemonClientStream { virNetMessagePtr rx; bool tx; + bool skippable; + daemonClientStreamPtr next; }; @@ -321,7 +323,8 @@ daemonClientStream * daemonCreateClientStream(virNetServerClientPtr client, virStreamPtr st, virNetServerProgramPtr prog, - virNetMessageHeaderPtr header) + virNetMessageHeaderPtr header, + bool skippable) { daemonClientStream *stream; daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client); @@ -339,6 +342,7 @@ daemonCreateClientStream(virNetServerClientPtr client, stream->serial = header->serial; stream->filterID = -1; stream->st = st; + stream->skippable = skippable; return stream; } diff --git a/daemon/stream.h b/daemon/stream.h index cf76e71..bf5dc24 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 seekable); int daemonFreeClientStream(virNetServerClientPtr client, daemonClientStream *stream); diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 7024464..6a2c6f6 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 e608812..9862598 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 01761cf..42619bf 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -54,6 +54,8 @@ struct _virNetClientStream { virNetMessagePtr rx; bool incomingEOF; + bool skippable; /* User requested skippable stream */ + 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 skippable) { virNetClientStreamPtr st; @@ -152,6 +155,7 @@ virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, st->prog = virObjectRef(prog); st->proc = proc; st->serial = serial; + st->skippable = skippable; return st; } diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index e278dab..0a5aafd 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 seekable); bool virNetClientStreamRaiseError(virNetClientStreamPtr st); -- 2.10.2

This is going to be RPC representation for virStreamSkip. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/virnetprotocol.x | 4 ++++ src/virnetprotocol-structs | 3 +++ 2 files changed, 7 insertions(+) diff --git a/src/rpc/virnetprotocol.x b/src/rpc/virnetprotocol.x index 9ce33b0..3623588 100644 --- a/src/rpc/virnetprotocol.x +++ b/src/rpc/virnetprotocol.x @@ -236,3 +236,7 @@ struct virNetMessageError { int int2; virNetMessageNetwork net; /* unused */ }; + +struct virNetStreamSkip { + unsigned hyper length; +}; diff --git a/src/virnetprotocol-structs b/src/virnetprotocol-structs index af4526c..a8cc603 100644 --- a/src/virnetprotocol-structs +++ b/src/virnetprotocol-structs @@ -42,3 +42,6 @@ struct virNetMessageError { int int2; virNetMessageNetwork net; }; +struct virNetStreamSkip { + uint64_t length; +}; -- 2.10.2

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 624a626..6899c18 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_SKIP) goto cleanup; if (!virNetServerProgramMatches(stream->prog, msg)) diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index c959747..6cff3c4 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_SKIP: /* Stream skip protocol */ return virNetClientCallDispatchStream(client); default: diff --git a/src/rpc/virnetprotocol.x b/src/rpc/virnetprotocol.x index 3623588..3530694 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_SKIP + * * status == VIR_NET_CONTINUE + * byte[] skip 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 skip data packet */ + VIR_NET_STREAM_SKIP = 6 }; enum virNetMessageStatus { diff --git a/src/virnetprotocol-structs b/src/virnetprotocol-structs index a8cc603..c31683f 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_SKIP = 6, }; enum virNetMessageStatus { VIR_NET_OK = 0, -- 2.10.2

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 | 48 ++++++++++++++++++++++++++++++++++++ tools/wireshark/src/packet-libvirt.h | 2 ++ 2 files changed, 50 insertions(+) diff --git a/tools/wireshark/src/packet-libvirt.c b/tools/wireshark/src/packet-libvirt.c index 260161e..be82a23 100644 --- a/tools/wireshark/src/packet-libvirt.c +++ b/tools/wireshark/src/packet-libvirt.c @@ -50,8 +50,11 @@ 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_skip_length = -1; +static int hf_libvirt_stream_skip = -1; int hf_libvirt_unknown = -1; static gint ett_libvirt = -1; +static gint ett_libvirt_stream_skip = -1; #define XDR_PRIMITIVE_DISSECTOR(xtype, ctype, ftype) \ static gboolean \ @@ -326,6 +329,36 @@ 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_skip(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_skip, tvb, start, -1, ENC_NA); + } else { + header_field_info *hfinfo; + hfinfo = proto_registrar_get_nth(hf_libvirt_stream_skip); + 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_skip); + + hf = hf_libvirt_stream_skip_length; + if (!dissect_xdr_u_hyper(tvb, tree, xdrs, hf)) return FALSE; + proto_item_set_len(ti, xdr_getpos(xdrs) - start); + return TRUE; +} + + +static void +dissect_libvirt_stream_skip(tvbuff_t *tvb, proto_tree *tree, gint payload_length, guint32 status) +{ + proto_tree_add_item(tree, hf_libvirt_stream_skip_length, tvb, VIR_HEADER_LEN, -1, ENC_NA); +} + static void dissect_libvirt_payload(tvbuff_t *tvb, proto_tree *tree, guint32 prog, guint32 proc, guint32 type, guint32 status) @@ -346,6 +379,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_SKIP) { + dissect_libvirt_payload_xdr_data(tvb, tree, payload_length, status, dissect_xdr_stream_skip); } else { goto unknown; } @@ -525,6 +560,18 @@ proto_register_libvirt(void) NULL, 0x0, NULL, HFILL} }, + { &hf_libvirt_stream_skip, + { "stream_skip", "libvirt.stream_skip", + FT_BYTES, BASE_NONE, + NULL, 0x0, + NULL, HFILL} + }, + { &hf_libvirt_stream_skip_length, + { "length", "libvirt.stream_skip.length", + FT_UINT64, BASE_DEC, + NULL, 0x0, + NULL, HFILL} + }, { &hf_libvirt_unknown, { "unknown", "libvirt.unknown", FT_BYTES, BASE_NONE, @@ -535,6 +582,7 @@ proto_register_libvirt(void) static gint *ett[] = { VIR_DYNAMIC_ETTSET + &ett_libvirt_stream_skip, &ett_libvirt }; diff --git a/tools/wireshark/src/packet-libvirt.h b/tools/wireshark/src/packet-libvirt.h index 5f99fdf..006aa6d 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_SKIP = 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_SKIP, "STREAM_SKIP" }, { -1, NULL } }; -- 2.10.2

This is just a helper function that takes in a length value, encodes it into XDR and sends to client. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/libvirt_remote.syms | 1 + src/rpc/virnetserverprogram.c | 33 +++++++++++++++++++++++++++++++++ src/rpc/virnetserverprogram.h | 7 +++++++ 3 files changed, 41 insertions(+) diff --git a/src/libvirt_remote.syms b/src/libvirt_remote.syms index ca1f3ac..29dceab 100644 --- a/src/libvirt_remote.syms +++ b/src/libvirt_remote.syms @@ -178,6 +178,7 @@ virNetServerProgramNew; virNetServerProgramSendReplyError; virNetServerProgramSendStreamData; virNetServerProgramSendStreamError; +virNetServerProgramSendStreamSkip; virNetServerProgramUnknownError; diff --git a/src/rpc/virnetserverprogram.c b/src/rpc/virnetserverprogram.c index d1597f4..6d84056 100644 --- a/src/rpc/virnetserverprogram.c +++ b/src/rpc/virnetserverprogram.c @@ -548,6 +548,39 @@ int virNetServerProgramSendStreamData(virNetServerProgramPtr prog, } +int virNetServerProgramSendStreamSkip(virNetServerProgramPtr prog, + virNetServerClientPtr client, + virNetMessagePtr msg, + int procedure, + unsigned int serial, + unsigned long long length) +{ + virNetStreamSkip data; + + VIR_DEBUG("client=%p msg=%p length=%llu", client, msg, length); + + memset(&data, 0, sizeof(data)); + data.length = length; + + msg->header.prog = prog->program; + msg->header.vers = prog->version; + msg->header.proc = procedure; + msg->header.type = VIR_NET_STREAM_SKIP; + msg->header.serial = serial; + msg->header.status = VIR_NET_CONTINUE; + + if (virNetMessageEncodeHeader(msg) < 0) + return -1; + + if (virNetMessageEncodePayload(msg, + (xdrproc_t) xdr_virNetStreamSkip, + &data) < 0) + return -1; + + return virNetServerClientSendMessage(client, msg); +} + + void virNetServerProgramDispose(void *obj ATTRIBUTE_UNUSED) { } diff --git a/src/rpc/virnetserverprogram.h b/src/rpc/virnetserverprogram.h index 531fca0..eba2168 100644 --- a/src/rpc/virnetserverprogram.h +++ b/src/rpc/virnetserverprogram.h @@ -104,4 +104,11 @@ int virNetServerProgramSendStreamData(virNetServerProgramPtr prog, const char *data, size_t len); +int virNetServerProgramSendStreamSkip(virNetServerProgramPtr prog, + virNetServerClientPtr client, + virNetMessagePtr msg, + int procedure, + unsigned int serial, + unsigned long long length); + #endif /* __VIR_NET_SERVER_PROGRAM_H__ */ -- 2.10.2

On 05/05/2017 05:26 PM, John Ferlan wrote: Exactly. For me Skip and HoleSize have clear meaning. So your suggestion is to have: SetSkip and GetSkip? Well, I don't like it that much but if that's the only thing that should prevent this from merging ... Also, as I've said earlier, I'm gonna send v2 without any name change for the time being. As you've correctly noticed, a lot of functions, RPC calls, and other stuff have their name derived from current naming. Therefore changing that would be a non-trivial amount of work and therefore I'd like to do it just once. After we have a clear agreement on the naming. Also, now that we seem to be using Reviewed-by tags, should I include those in my commit messages for v2? I mean, you've reviewed patches 00-30. And you'll definitely deserve the tag. But what I'm asking is if I should put it there already for v2 or before pushing the patches. To extend my thinking further, review is not the same as ACK. So Reviewed-by should include all the reviewers even though just one of them gave ACK? Or if a patch is respin in version X, should it include all the past reviewers? So many questions and not many answers. Michal

On 05/10/2017 01:53 PM, Michal Privoznik wrote: For some unknown reason I failed to see the obvious. virStreamSendHole() <-- for public APIs virStreamRecvHole() VIR_NET_STREAM_HOLE <-- for RPC packet What do you think? Michal

While the previous commit implemented a helper for sending a STREAM_SKIP packet for daemon, this is a client's counterpart. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/libvirt_remote.syms | 1 + src/rpc/virnetclientstream.c | 52 ++++++++++++++++++++++++++++++++++++++++++++ src/rpc/virnetclientstream.h | 4 ++++ 3 files changed, 57 insertions(+) diff --git a/src/libvirt_remote.syms b/src/libvirt_remote.syms index 29dceab..6093613 100644 --- a/src/libvirt_remote.syms +++ b/src/libvirt_remote.syms @@ -54,6 +54,7 @@ virNetClientStreamQueuePacket; virNetClientStreamRaiseError; virNetClientStreamRecvPacket; virNetClientStreamSendPacket; +virNetClientStreamSendSkip; virNetClientStreamSetError; diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index 42619bf..1e30080 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -429,6 +429,58 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, } +int +virNetClientStreamSendSkip(virNetClientStreamPtr st, + virNetClientPtr client, + unsigned long long length) +{ + virNetMessagePtr msg = NULL; + virNetStreamSkip data; + int ret = -1; + + VIR_DEBUG("st=%p length=%llu", st, length); + + if (!st->skippable) { + virReportError(VIR_ERR_OPERATION_INVALID, "%s", + _("Skipping is not supported with this stream")); + return -1; + } + + memset(&data, 0, sizeof(data)); + data.length = length; + + if (!(msg = virNetMessageNew(false))) + return -1; + + virObjectLock(st); + + msg->header.prog = virNetClientProgramGetProgram(st->prog); + msg->header.vers = virNetClientProgramGetVersion(st->prog); + msg->header.status = VIR_NET_CONTINUE; + msg->header.type = VIR_NET_STREAM_SKIP; + msg->header.serial = st->serial; + msg->header.proc = st->proc; + + virObjectUnlock(st); + + if (virNetMessageEncodeHeader(msg) < 0) + goto cleanup; + + if (virNetMessageEncodePayload(msg, + (xdrproc_t) xdr_virNetStreamSkip, + &data) < 0) + goto cleanup; + + if (virNetClientSendNoReply(client, msg) < 0) + goto cleanup; + + ret = 0; + cleanup: + virNetMessageFree(msg); + return ret; +} + + int virNetClientStreamEventAddCallback(virNetClientStreamPtr st, int events, virNetClientStreamEventCallback cb, diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index 0a5aafd..a648b7c 100644 --- a/src/rpc/virnetclientstream.h +++ b/src/rpc/virnetclientstream.h @@ -61,6 +61,10 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, size_t nbytes, bool nonblock); +int virNetClientStreamSendSkip(virNetClientStreamPtr st, + virNetClientPtr client, + unsigned long long length); + int virNetClientStreamEventAddCallback(virNetClientStreamPtr st, int events, virNetClientStreamEventCallback cb, -- 2.10.2

Basically, whenever the new type of stream packet arrives to the daemon call this function that decodes it and calls virStreamSkipCallback(). Otherwise a regular data stream packet has arrived and therefore continue its processing. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- daemon/stream.c | 81 +++++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 70 insertions(+), 11 deletions(-) diff --git a/daemon/stream.c b/daemon/stream.c index 6899c18..84709da 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -29,6 +29,7 @@ #include "virlog.h" #include "virnetserverclient.h" #include "virerror.h" +#include "libvirt_internal.h" #define VIR_FROM_THIS VIR_FROM_STREAMS @@ -653,6 +654,52 @@ daemonStreamHandleAbort(virNetServerClientPtr client, } +static int +daemonStreamHandleSkip(virNetServerClientPtr client, + daemonClientStream *stream, + virNetMessagePtr msg) +{ + int ret; + virNetStreamSkip data; + + VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%u", + client, stream, msg->header.proc, msg->header.serial); + + /* Let's check if client plays nicely and advertised usage of + * sparse stream upfront. */ + if (!stream->skippable) { + virReportError(VIR_ERR_RPC, "%s", + _("Unexpected stream skip")); + return -1; + } + + if (virNetMessageDecodePayload(msg, + (xdrproc_t) xdr_virNetStreamSkip, + &data) < 0) + return -1; + + ret = virStreamSkip(stream->st, data.length); + + if (ret < 0) { + virNetMessageError rerr; + + memset(&rerr, 0, sizeof(rerr)); + + VIR_INFO("Stream skip failed"); + stream->closed = true; + virStreamEventRemoveCallback(stream->st); + virStreamAbort(stream->st); + + return virNetServerProgramSendReplyError(stream->prog, + client, + msg, + &rerr, + &msg->header); + } + + return 0; +} + /* * Called when the stream is signalled has being able to accept @@ -671,19 +718,31 @@ daemonStreamHandleWrite(virNetServerClientPtr client, virNetMessagePtr msg = stream->rx; int ret; - switch (msg->header.status) { - case VIR_NET_OK: - ret = daemonStreamHandleFinish(client, stream, msg); - break; + if (msg->header.type == VIR_NET_STREAM_SKIP) { + /* Handle special case when client sent us skip. + * Otherwise just carry on with processing stream + * data. */ + ret = daemonStreamHandleSkip(client, stream, msg); + } else if (msg->header.type == VIR_NET_STREAM) { + switch (msg->header.status) { + case VIR_NET_OK: + ret = daemonStreamHandleFinish(client, stream, msg); + break; - case VIR_NET_CONTINUE: - ret = daemonStreamHandleWriteData(client, stream, msg); - break; + case VIR_NET_CONTINUE: + ret = daemonStreamHandleWriteData(client, stream, msg); + break; - case VIR_NET_ERROR: - default: - ret = daemonStreamHandleAbort(client, stream, msg); - break; + case VIR_NET_ERROR: + default: + ret = daemonStreamHandleAbort(client, stream, msg); + break; + } + } else { + virReportError(VIR_ERR_RPC, + _("Unexpected message type: %d"), + msg->header.type); + ret = -1; } if (ret > 0) -- 2.10.2

This is a function that handles an incoming STREAM_SKIP packet. Even though it is not wired up yet, it will be soon. At the beginning do couple of checks whether server plays nicely and sent us a STREAM_SKIP packed only after we've enabled sparse streams. Then decodes the message payload to see how big the hole is and stores it in passed @length argument. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/virnetclientstream.c | 63 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index 1e30080..027ffde 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -28,6 +28,7 @@ #include "virerror.h" #include "virlog.h" #include "virthread.h" +#include "libvirt_internal.h" #define VIR_FROM_THIS VIR_FROM_RPC @@ -55,6 +56,7 @@ struct _virNetClientStream { bool incomingEOF; bool skippable; /* User requested skippable stream */ + unsigned long long skipLength; /* Size of incoming hole in stream. */ virNetClientStreamEventCallback cb; void *cbOpaque; @@ -356,6 +358,67 @@ int virNetClientStreamSendPacket(virNetClientStreamPtr st, return -1; } + +static int ATTRIBUTE_UNUSED +virNetClientStreamHandleSkip(virNetClientPtr client, + virNetClientStreamPtr st) +{ + virNetMessagePtr msg; + virNetStreamSkip data; + int ret = -1; + + VIR_DEBUG("client=%p st=%p", client, st); + + msg = st->rx; + memset(&data, 0, sizeof(data)); + + /* We should not be called unless there's VIR_NET_STREAM_SKIP + * message at the head of the list. But doesn't hurt to check */ + if (!msg) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("No message in the queue")); + goto cleanup; + } + + if (msg->header.type != VIR_NET_STREAM_SKIP) { + virReportError(VIR_ERR_INTERNAL_ERROR, + _("Invalid message prog=%d type=%d serial=%u proc=%d"), + msg->header.prog, + msg->header.type, + msg->header.serial, + msg->header.proc); + goto cleanup; + } + + /* Server should not send us VIR_NET_STREAM_SKIP unless we + * have requested so. But does not hurt to check ... */ + if (!st->skippable) { + virReportError(VIR_ERR_RPC, "%s", + _("Unexpected stream skip")); + goto cleanup; + } + + if (virNetMessageDecodePayload(msg, + (xdrproc_t) xdr_virNetStreamSkip, + &data) < 0) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("Malformed stream skip packet")); + goto cleanup; + } + + virNetMessageQueueServe(&st->rx); + virNetMessageFree(msg); + st->skipLength += data.length; + + ret = 0; + cleanup: + if (ret < 0) { + /* Abort stream? */ + } + return ret; +} + + int virNetClientStreamRecvPacket(virNetClientStreamPtr st, virNetClientPtr client, char *data, -- 2.10.2

Now that we have RPC wrappers over VIR_NET_STREAM_SKIP we can start wiring them up. This commit wires up situation when a client wants to send a hole to daemon. To keep stream offsets synchronous, upon successful call on the daemon skip the same hole in local part of the stream. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 6a2c6f6..6037e08 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5683,6 +5683,34 @@ remoteStreamRecv(virStreamPtr st, return remoteStreamRecvFlags(st, data, nbytes, 0); } + +static int +remoteStreamSkip(virStreamPtr st, + unsigned long long length) +{ + VIR_DEBUG("st=%p length=%llu", st, length); + struct private_data *priv = st->conn->privateData; + virNetClientStreamPtr privst = st->privateData; + int rv; + + if (virNetClientStreamRaiseError(privst)) + return -1; + + remoteDriverLock(priv); + priv->localUses++; + remoteDriverUnlock(priv); + + rv = virNetClientStreamSendSkip(privst, + priv->client, + length); + + remoteDriverLock(priv); + priv->localUses--; + remoteDriverUnlock(priv); + return rv; +} + + struct remoteStreamCallbackData { virStreamPtr st; virStreamEventCallback cb; @@ -5857,6 +5885,7 @@ static virStreamDriver remoteStreamDrv = { .streamRecv = remoteStreamRecv, .streamRecvFlags = remoteStreamRecvFlags, .streamSend = remoteStreamSend, + .streamSkip = remoteStreamSkip, .streamFinish = remoteStreamFinish, .streamAbort = remoteStreamAbort, .streamEventAddCallback = remoteStreamEventAddCallback, -- 2.10.2

Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 3 ++- src/rpc/virnetclientstream.c | 10 +++++++--- src/rpc/virnetclientstream.h | 3 ++- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 6037e08..0512f14 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5665,7 +5665,8 @@ remoteStreamRecvFlags(virStreamPtr st, priv->client, data, nbytes, - (st->flags & VIR_STREAM_NONBLOCK)); + (st->flags & VIR_STREAM_NONBLOCK), + flags); VIR_DEBUG("Done %d", rv); diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index 027ffde..f687bfa 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -423,13 +423,17 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, virNetClientPtr client, char *data, size_t nbytes, - bool nonblock) + bool nonblock, + unsigned int flags) { int rv = -1; size_t want; - VIR_DEBUG("st=%p client=%p data=%p nbytes=%zu nonblock=%d", - st, client, data, nbytes, nonblock); + VIR_DEBUG("st=%p client=%p data=%p nbytes=%zu nonblock=%d flags=%x", + st, client, data, nbytes, nonblock, flags); + + virCheckFlags(0, -1); + virObjectLock(st); if (!st->rx && !st->incomingEOF) { virNetMessagePtr msg; diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index a648b7c..2835066 100644 --- a/src/rpc/virnetclientstream.h +++ b/src/rpc/virnetclientstream.h @@ -59,7 +59,8 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, virNetClientPtr client, char *data, size_t nbytes, - bool nonblock); + bool nonblock, + unsigned int flags); int virNetClientStreamSendSkip(virNetClientStreamPtr st, virNetClientPtr client, -- 2.10.2

This function will fetch previously processed stream holes and return their sum. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/libvirt_remote.syms | 1 + src/rpc/virnetclientstream.c | 15 +++++++++++++++ src/rpc/virnetclientstream.h | 4 ++++ 3 files changed, 20 insertions(+) diff --git a/src/libvirt_remote.syms b/src/libvirt_remote.syms index 6093613..f546647 100644 --- a/src/libvirt_remote.syms +++ b/src/libvirt_remote.syms @@ -48,6 +48,7 @@ virNetClientStreamEOF; virNetClientStreamEventAddCallback; virNetClientStreamEventRemoveCallback; virNetClientStreamEventUpdateCallback; +virNetClientStreamHoleSize; virNetClientStreamMatches; virNetClientStreamNew; virNetClientStreamQueuePacket; diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index f687bfa..c773524 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -548,6 +548,21 @@ virNetClientStreamSendSkip(virNetClientStreamPtr st, } +int virNetClientStreamHoleSize(virNetClientPtr client ATTRIBUTE_UNUSED, + virNetClientStreamPtr st, + unsigned long long *length) +{ + int ret = st->skipLength; + + if (length) { + *length = st->skipLength; + st->skipLength = 0; + } + + return ret; +} + + int virNetClientStreamEventAddCallback(virNetClientStreamPtr st, int events, virNetClientStreamEventCallback cb, diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index 2835066..9caa091 100644 --- a/src/rpc/virnetclientstream.h +++ b/src/rpc/virnetclientstream.h @@ -66,6 +66,10 @@ int virNetClientStreamSendSkip(virNetClientStreamPtr st, virNetClientPtr client, unsigned long long length); +int virNetClientStreamHoleSize(virNetClientPtr client, + virNetClientStreamPtr st, + unsigned long long *length); + int virNetClientStreamEventAddCallback(virNetClientStreamPtr st, int events, virNetClientStreamEventCallback cb, -- 2.10.2

Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 0512f14..376e9ba 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5712,6 +5712,29 @@ remoteStreamSkip(virStreamPtr st, } +static int +remoteStreamHoleSize(virStreamPtr st, + unsigned long long *length) +{ + VIR_DEBUG("st=%p length=%p", st, length); + struct private_data *priv = st->conn->privateData; + virNetClientStreamPtr privst = st->privateData; + int rv; + + if (virNetClientStreamRaiseError(privst)) + return -1; + + remoteDriverLock(priv); + priv->localUses++; + remoteDriverUnlock(priv); + + rv = virNetClientStreamHoleSize(priv->client, privst, length); + + remoteDriverLock(priv); + priv->localUses--; + remoteDriverUnlock(priv); + return rv; +} struct remoteStreamCallbackData { virStreamPtr st; virStreamEventCallback cb; @@ -5887,6 +5910,7 @@ static virStreamDriver remoteStreamDrv = { .streamRecvFlags = remoteStreamRecvFlags, .streamSend = remoteStreamSend, .streamSkip = remoteStreamSkip, + .streamHoleSize = remoteStreamHoleSize, .streamFinish = remoteStreamFinish, .streamAbort = remoteStreamAbort, .streamEventAddCallback = remoteStreamEventAddCallback, -- 2.10.2

Whenever server sends a client stream packet (either regular with actual data or stream skip one) it is queued on @st->rx. So the list is a mixture of both types of stream packets. So now that we have all the helpers needed we can wire their processing up. But since virNetClientStreamRecvPacket doesn't support VIR_STREAM_RECV_STOP_AT_HOLE flag yet, let's turn all received skips into zeroes repeating requested times. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/virnetclientstream.c | 44 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 42 insertions(+), 2 deletions(-) diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index c773524..ff35137 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -296,6 +296,8 @@ int virNetClientStreamQueuePacket(virNetClientStreamPtr st, virObjectLock(st); + /* Don't distinguish VIR_NET_STREAM and VIR_NET_STREAM_SKIP + * here just yet. We want in order processing! */ virNetMessageQueuePush(&st->rx, tmp_msg); virNetClientStreamEventTimerUpdate(st); @@ -359,7 +361,7 @@ int virNetClientStreamSendPacket(virNetClientStreamPtr st, } -static int ATTRIBUTE_UNUSED +static int virNetClientStreamHandleSkip(virNetClientPtr client, virNetClientStreamPtr st) { @@ -435,6 +437,8 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, virCheckFlags(0, -1); virObjectLock(st); + + reread: if (!st->rx && !st->incomingEOF) { virNetMessagePtr msg; int ret; @@ -466,8 +470,44 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, } VIR_DEBUG("After IO rx=%p", st->rx); + + while (st->rx && + st->rx->header.type == VIR_NET_STREAM_SKIP) { + /* Handle skip sent to us by server. */ + + if (virNetClientStreamHandleSkip(client, st) < 0) + goto cleanup; + } + + if (!st->rx && !st->incomingEOF && !st->skipLength) { + if (nonblock) { + VIR_DEBUG("Non-blocking mode and no data available"); + rv = -2; + goto cleanup; + } + + /* We have consumed all packets from incoming queue but those + * were only skip packets, no data. Read the stream again. */ + goto reread; + } + want = nbytes; - while (want && st->rx) { + + if (st->skipLength) { + /* Pretend skipLength zeroes was read from stream. */ + size_t len = want; + + if (len > st->skipLength) + len = st->skipLength; + + memset(data, 0, len); + st->skipLength -= len; + want -= len; + } + + while (want && + st->rx && + st->rx->header.type == VIR_NET_STREAM) { virNetMessagePtr msg = st->rx; size_t len = want; -- 2.10.2

On 05/05/2017 05:43 PM, John Ferlan wrote: Because our streams are slightly more generic than just a file transfer. I mean, we use it for just any data transfer. Therefore it could be used even for this start -> data -> skip -> skip -> data -> end. You won't have this with regular files stored on a disk, but then again, virStream can be used (and is used) for generic data transfer. vol-download an vol-upload is just a subset of its usage. Now, with that example, it makes sense to merge those skips into one big, doesn't it? But I agree, it looks a bit weird. I can drop the merging - it's not a performance issue. I was just trying to save caller couple of calls to our APIs. Fair enough. I'm gonna send v2 with most of your review points worked in. Except, for now I'm still gonna use Skip() and HoleSize() - changing that now would be a non-trivial piece of work and thus I'd like to do it just once, when we have clear agreement on naming. Hope you understand that. You can see my reasoning from above in effect here. Michal

Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 2 +- src/rpc/virnetclientstream.c | 9 ++++++++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 376e9ba..fd76811 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5652,7 +5652,7 @@ remoteStreamRecvFlags(virStreamPtr st, virNetClientStreamPtr privst = st->privateData; int rv; - virCheckFlags(0, -1); + virCheckFlags(VIR_STREAM_RECV_STOP_AT_HOLE, -1); if (virNetClientStreamRaiseError(privst)) return -1; diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index ff35137..18d7cbc 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -434,7 +434,7 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, VIR_DEBUG("st=%p client=%p data=%p nbytes=%zu nonblock=%d flags=%x", st, client, data, nbytes, nonblock, flags); - virCheckFlags(0, -1); + virCheckFlags(VIR_STREAM_RECV_STOP_AT_HOLE, -1); virObjectLock(st); @@ -497,6 +497,13 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, /* Pretend skipLength zeroes was read from stream. */ size_t len = want; + /* Yes, pretend unless we are asked not to. */ + if (flags & VIR_STREAM_RECV_STOP_AT_HOLE) { + /* No error reporting here. Caller knows what they are doing. */ + rv = -3; + goto cleanup; + } + if (len > st->skipLength) len = st->skipLength; -- 2.10.2

While virStreamInData() should be a small and quick function, in our implementation it seeks multiple times. Moreover, it is called even if we know that we are in data. Well, quite. If we track its return values and do some basic math with them, we can end up calling virStreamInData right at the boundaries of a data section or a hole and nowhere else. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- daemon/stream.c | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/daemon/stream.c b/daemon/stream.c index a21f1bf..0901d82 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -54,6 +54,7 @@ struct daemonClientStream { bool tx; bool skippable; + size_t dataLen; /* How much data is there remaining until we see a hole */ daemonClientStreamPtr next; }; @@ -823,7 +824,7 @@ daemonStreamHandleRead(virNetServerClientPtr client, if (!(msg = virNetMessageNew(false))) goto cleanup; - if (stream->skippable) { + if (stream->skippable && !stream->dataLen) { /* Handle skip. We want to send some data to the client. But we might * be in a hole. Seek to next data. But if we are in data already, just * carry on. */ @@ -867,10 +868,13 @@ daemonStreamHandleRead(virNetServerClientPtr client, } } - if (length < bufferLen) - bufferLen = length; + stream->dataLen = length; } + if (stream->skippable && + bufferLen > stream->dataLen) + bufferLen = stream->dataLen; + rv = virStreamRecv(stream->st, buffer, bufferLen); if (rv == -2) { /* Should never get this, since we're only called when we know @@ -885,6 +889,8 @@ daemonStreamHandleRead(virNetServerClientPtr client, goto cleanup; msg = NULL; } else { + stream->dataLen -= rv; + stream->tx = false; if (rv == 0) stream->recvEOF = true; -- 2.10.2

Now, not all APIs are going to support sparse streams. To some it makes no sense at all, e.g. virDomainOpenConsole() or virDomainOpenChannel(). To others, we will need a special flag to indicate that client wants to enable sparse streams. Instead of having to write RPC dispatchers by hand we can just annotate in our .x files that a certain flag to certain RPC call enables this feature. For instance: /** * @generate: both * @readstream: 1 * @sparseflag: VIR_SPARSE_STREAM * @acl: storage_vol:data_read */ REMOTE_PROC_DOMAIN_SOME_API = XXX, Therefore, whenever client calls virDomainSomeAPI(.., VIR_SPARSE_STREAM); daemon will mark that down and send stream skips when possible. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/gendispatch.pl | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/src/rpc/gendispatch.pl b/src/rpc/gendispatch.pl index 9862598..def88d4 100755 --- a/src/rpc/gendispatch.pl +++ b/src/rpc/gendispatch.pl @@ -281,6 +281,13 @@ while (<PROTOCOL>) { $calls{$name}->{streamflag} = "none"; } + if (exists $opts{sparseflag}) { + die "\@sparseflag requires stream" unless $calls{$name}->{streamflag} ne "none"; + $calls{$name}->{sparseflag} = $opts{sparseflag}; + } else { + $calls{$name}->{sparseflag} = "none"; + } + $calls{$name}->{acl} = $opts{acl}; $calls{$name}->{aclfilter} = $opts{aclfilter}; @@ -982,6 +989,11 @@ elsif ($mode eq "server") { if ($call->{streamflag} ne "none") { print " virStreamPtr st = NULL;\n"; print " daemonClientStreamPtr stream = NULL;\n"; + if ($call->{sparseflag} ne "none") { + print " const bool sparse = args->flags & $call->{sparseflag};\n" + } else { + print " const bool sparse = false;\n"; + } } print "\n"; @@ -1024,7 +1036,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, false)))\n"; + print " if (!(stream = daemonCreateClientStream(client, st, remoteProgram, &msg->header, sparse)))\n"; print " goto cleanup;\n"; print "\n"; } @@ -1727,6 +1739,11 @@ elsif ($mode eq "client") { if ($call->{streamflag} ne "none") { print " virNetClientStreamPtr netst = NULL;\n"; + if ($call->{sparseflag} ne "none") { + print " const bool sparse = flags & $call->{sparseflag};\n" + } else { + print " const bool sparse = false;\n"; + } } print "\n"; @@ -1738,7 +1755,7 @@ elsif ($mode eq "client") { if ($call->{streamflag} ne "none") { print "\n"; - print " if (!(netst = virNetClientStreamNew(st, priv->remoteProgram, $call->{constname}, priv->counter, false)))\n"; + print " if (!(netst = virNetClientStreamNew(st, priv->remoteProgram, $call->{constname}, priv->counter, sparse)))\n"; print " goto done;\n"; print "\n"; print " if (virNetClientAddStream(priv->client, netst) < 0) {\n"; -- 2.10.2

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 4b86e29..4119d7c 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, + unsigned 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 64cef23..756546a 100644 --- a/tools/virsh-util.h +++ b/tools/virsh-util.h @@ -58,6 +58,11 @@ virshStreamSink(virStreamPtr st, void *opaque); int +virshStreamSkip(virStreamPtr st, + unsigned long long offset, + void *opaque); + +int virshDomainGetXMLFromDom(vshControl *ctl, virDomainPtr dom, unsigned int flags, diff --git a/tools/virsh-volume.c b/tools/virsh-volume.c index 66fe70e..3d19b74 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 e16f62f..ce87c19 100644 --- a/tools/virsh.pod +++ b/tools/virsh.pod @@ -3916,12 +3916,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.10.2