This reverts commit d9c9e138f22c48626f719f880920e04c639e0177. Unfortunately, things are going to be handled differently so this commit must go. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/virnetclientstream.c | 152 +++++++++++++++---------------------------- 1 file changed, 53 insertions(+), 99 deletions(-) diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index 64e9cd2..b428f4b 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -49,9 +49,9 @@ struct _virNetClientStream { * time by stopping consuming any incoming data * off the socket.... */ - struct iovec *incomingVec; /* I/O Vector to hold data */ - size_t writeVec; /* Vectors produced */ - size_t readVec; /* Vectors consumed */ + char *incoming; + size_t incomingOffset; + size_t incomingLength; bool incomingEOF; virNetClientStreamEventCallback cb; @@ -86,9 +86,9 @@ virNetClientStreamEventTimerUpdate(virNetClientStreamPtr st) if (!st->cb) return; - VIR_DEBUG("Check timer readVec %zu writeVec %zu %d", st->readVec, st->writeVec, st->cbEvents); + VIR_DEBUG("Check timer offset=%zu %d", st->incomingOffset, st->cbEvents); - if ((((st->readVec < st->writeVec) || st->incomingEOF) && + if (((st->incomingOffset || st->incomingEOF) && (st->cbEvents & VIR_STREAM_EVENT_READABLE)) || (st->cbEvents & VIR_STREAM_EVENT_WRITABLE)) { VIR_DEBUG("Enabling event timer"); @@ -110,14 +110,13 @@ virNetClientStreamEventTimer(int timer ATTRIBUTE_UNUSED, void *opaque) if (st->cb && (st->cbEvents & VIR_STREAM_EVENT_READABLE) && - ((st->readVec < st->writeVec) || st->incomingEOF)) + (st->incomingOffset || st->incomingEOF)) events |= VIR_STREAM_EVENT_READABLE; if (st->cb && (st->cbEvents & VIR_STREAM_EVENT_WRITABLE)) events |= VIR_STREAM_EVENT_WRITABLE; - VIR_DEBUG("Got Timer dispatch %d %d readVec %zu writeVec %zu", events, st->cbEvents, - st->readVec, st->writeVec); + VIR_DEBUG("Got Timer dispatch %d %d offset=%zu", events, st->cbEvents, st->incomingOffset); if (events) { virNetClientStreamEventCallback cb = st->cb; void *cbOpaque = st->cbOpaque; @@ -162,7 +161,7 @@ void virNetClientStreamDispose(void *obj) virNetClientStreamPtr st = obj; virResetError(&st->err); - VIR_FREE(st->incomingVec); + VIR_FREE(st->incoming); virObjectUnref(st->prog); } @@ -266,50 +265,38 @@ int virNetClientStreamQueuePacket(virNetClientStreamPtr st, virNetMessagePtr msg) { int ret = -1; - struct iovec iov; - char *base; - size_t piece, pieces, length, offset = 0, size = 1024*1024; + size_t need; virObjectLock(st); + need = msg->bufferLength - msg->bufferOffset; + if (need) { + size_t avail = st->incomingLength - st->incomingOffset; + if (need > avail) { + size_t extra = need - avail; + if (VIR_REALLOC_N(st->incoming, + st->incomingLength + extra) < 0) { + VIR_DEBUG("Out of memory handling stream data"); + goto cleanup; + } + st->incomingLength += extra; + } - length = msg->bufferLength - msg->bufferOffset; - - if (length == 0) { + memcpy(st->incoming + st->incomingOffset, + msg->buffer + msg->bufferOffset, + msg->bufferLength - msg->bufferOffset); + st->incomingOffset += (msg->bufferLength - msg->bufferOffset); + } else { st->incomingEOF = true; - goto end; } - pieces = VIR_DIV_UP(length, size); - for (piece = 0; piece < pieces; piece++) { - if (size > length - offset) - size = length - offset; - - if (VIR_ALLOC_N(base, size)) { - VIR_DEBUG("Allocation failed"); - goto cleanup; - } - - memcpy(base, msg->buffer + msg->bufferOffset + offset, size); - iov.iov_base = base; - iov.iov_len = size; - offset += size; - - if (VIR_APPEND_ELEMENT(st->incomingVec, st->writeVec, iov) < 0) { - VIR_DEBUG("Append failed"); - VIR_FREE(base); - goto cleanup; - } - VIR_DEBUG("Wrote piece of vector. readVec %zu, writeVec %zu size %zu", - st->readVec, st->writeVec, size); - } - - end: + VIR_DEBUG("Stream incoming data offset %zu length %zu EOF %d", + st->incomingOffset, st->incomingLength, + st->incomingEOF); virNetClientStreamEventTimerUpdate(st); + ret = 0; cleanup: - VIR_DEBUG("Stream incoming data readVec %zu writeVec %zu EOF %d", - st->readVec, st->writeVec, st->incomingEOF); virObjectUnlock(st); return ret; } @@ -374,21 +361,17 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, size_t nbytes, bool nonblock) { - int ret = -1; - size_t partial, offset; - - virObjectLock(st); - + int rv = -1; VIR_DEBUG("st=%p client=%p data=%p nbytes=%zu nonblock=%d", st, client, data, nbytes, nonblock); - - if ((st->readVec >= st->writeVec) && !st->incomingEOF) { + virObjectLock(st); + if (!st->incomingOffset && !st->incomingEOF) { virNetMessagePtr msg; - int rv; + int ret; if (nonblock) { VIR_DEBUG("Non-blocking mode and no data available"); - ret = -2; + rv = -2; goto cleanup; } @@ -404,66 +387,37 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, VIR_DEBUG("Dummy packet to wait for stream data"); virObjectUnlock(st); - rv = virNetClientSendWithReplyStream(client, msg, st); + ret = virNetClientSendWithReplyStream(client, msg, st); virObjectLock(st); virNetMessageFree(msg); - if (rv < 0) + if (ret < 0) goto cleanup; } - offset = 0; - partial = nbytes; - - while (st->incomingVec && (st->readVec < st->writeVec)) { - struct iovec *iov = st->incomingVec + st->readVec; - - if (!iov || !iov->iov_base) { - virReportError(VIR_ERR_INTERNAL_ERROR, - "%s", _("NULL pointer encountered")); - goto cleanup; - } - - if (partial < iov->iov_len) { - memcpy(data+offset, iov->iov_base, partial); - memmove(iov->iov_base, (char*)iov->iov_base+partial, - iov->iov_len-partial); - iov->iov_len -= partial; - offset += partial; - VIR_DEBUG("Consumed %zu, left %zu", partial, iov->iov_len); - break; + VIR_DEBUG("After IO %zu", st->incomingOffset); + if (st->incomingOffset) { + int want = st->incomingOffset; + if (want > nbytes) + want = nbytes; + memcpy(data, st->incoming, want); + if (want < st->incomingOffset) { + memmove(st->incoming, st->incoming + want, st->incomingOffset - want); + st->incomingOffset -= want; + } else { + VIR_FREE(st->incoming); + st->incomingOffset = st->incomingLength = 0; } - - memcpy(data+offset, iov->iov_base, iov->iov_len); - VIR_DEBUG("Consumed %zu. Moving to next piece", iov->iov_len); - partial -= iov->iov_len; - offset += iov->iov_len; - VIR_FREE(iov->iov_base); - iov->iov_len = 0; - st->readVec++; - - VIR_DEBUG("Read piece of vector. read %zu, readVec %zu, writeVec %zu", - offset, st->readVec, st->writeVec); - } - - /* Shrink the I/O Vector buffer to free up memory. Do the - shrinking only when there is selected amount or more buffers to - free so it doesn't constantly memmove() and realloc() buffers. - */ - if (st->readVec >= 16) { - memmove(st->incomingVec, st->incomingVec + st->readVec, - sizeof(*st->incomingVec)*(st->writeVec - st->readVec)); - VIR_SHRINK_N(st->incomingVec, st->writeVec, st->readVec); - VIR_DEBUG("shrink removed %zu, left %zu", st->readVec, st->writeVec); - st->readVec = 0; + rv = want; + } else { + rv = 0; } - ret = offset; virNetClientStreamEventTimerUpdate(st); cleanup: virObjectUnlock(st); - return ret; + return rv; } -- 2.8.1

These are wrappers over virStreamRecv and virStreamSend so that users have to care about nothing but writing data into / reading data from a sink (typically a file). Note, that these wrappers are used exclusively on client side as the daemon has slightly different approach. Anyway, the wrappers allocate this buffer and use it for intermediate data storage until the data is passed to stream to send, or to the client application. So far, we are using 64KB buffer. This is enough, but suboptimal because server can send messages up to VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX bytes big (262120B, roughly 256KB). So if we make the buffer this big, a single message containing the data is sent instead of four, which is current situation. This means lower overhead, because each message contains a header which needs to be processed, each message is processed roughly same amount of time regardless of its size, less bytes need to be sent through the wire, and so on. Note that since server will never sent us a stream message bigger than VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX there's no point in sizing up the client buffer past this threshold. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/libvirt-stream.c | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c index c16f586..8384b37 100644 --- a/src/libvirt-stream.c +++ b/src/libvirt-stream.c @@ -23,6 +23,7 @@ #include "datatypes.h" #include "viralloc.h" #include "virlog.h" +#include "rpc/virnetprotocol.h" VIR_LOG_INIT("libvirt.stream"); @@ -330,7 +331,7 @@ virStreamSendAll(virStreamPtr stream, void *opaque) { char *bytes = NULL; - int want = 1024*64; + size_t want = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX; int ret = -1; VIR_DEBUG("stream=%p, handler=%p, opaque=%p", stream, handler, opaque); @@ -423,7 +424,7 @@ virStreamRecvAll(virStreamPtr stream, void *opaque) { char *bytes = NULL; - int want = 1024*64; + size_t want = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX; int ret = -1; VIR_DEBUG("stream=%p, handler=%p, opaque=%p", stream, handler, opaque); -- 2.8.1

On 04/28/2016 04:04 AM, Michal Privoznik wrote: 'offset' is a bit misleading - you're not skipping _to_ the given offset, so much as _over_ length bytes. I'd name it 'length'. Otherwise looks okay. -- Eric Blake eblake redhat com +1-919-301-3266 Libvirt virtualization library http://libvirt.org

On 05.05.2016 16:37, Eric Blake wrote: Ah, good point. Whilst implementing this I've balanced between Seek and Skip back and forth. That's why I call streem seekable even if it's really just skip what is implemented here. Michal

On 04/28/2016 04:04 AM, Michal Privoznik wrote: Again, naming this 'length' makes more sense (you're not skipping TO a particular offset, but OVER a given length). s/callback/a callback/ I guess that means we allow passing skipCb=NULL to deregister a callback; does it need to be specifically documented? Are there scenarios where you WANT to deregister before closing something else, to make sure that a stale callback is not called during a race scenario? I might have omitted the () around stream->skipCb, but I don't know if we have a consistent style, and yours makes it obvious that we know we are dereferencing a function pointer. Worth keeping sorted? -- Eric Blake eblake redhat com +1-919-301-3266 Libvirt virtualization library http://libvirt.org

These functions will be called to determine whether underlying file that stream is transferring is currently in a data or hole. While virStreamRegisterInData is exposed, virStreamInData does not need to be made a public API as it will be called just internally. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- include/libvirt/libvirt-stream.h | 23 +++++++++++ src/datatypes.h | 8 ++-- src/libvirt-stream.c | 88 ++++++++++++++++++++++++++++++++++++++++ src/libvirt_internal.h | 3 ++ src/libvirt_private.syms | 1 + src/libvirt_public.syms | 1 + 6 files changed, 121 insertions(+), 3 deletions(-) diff --git a/include/libvirt/libvirt-stream.h b/include/libvirt/libvirt-stream.h index 1a5286a..3a0c986 100644 --- a/include/libvirt/libvirt-stream.h +++ b/include/libvirt/libvirt-stream.h @@ -69,6 +69,29 @@ int virStreamRegisterSkip(virStreamPtr stream, int virStreamSkip(virStreamPtr st, unsigned long long offset); +/** + * virStreamInDataFunc: + * @stream: stream + * @data: are we in data or hole + * @offset: offset to next section + * @opaque: optional application provided data + * + * This callback is called whenever virStreamInData needs to + * check whether @stream is in data section or in hole. Check + * description for virStreamInData for more detailed description. + * + * Returns 0 on success (with @data and @offset updated) + * -1 otherwise (with @data and @offset untouched) + */ +typedef int (*virStreamInDataFunc)(virStreamPtr stream, + int *data, + unsigned long long *offset, + void *opaque); + +int virStreamRegisterInData(virStreamPtr stream, + virStreamInDataFunc inDataCb, + void *opaque); + /** * virStreamSourceFunc: diff --git a/src/datatypes.h b/src/datatypes.h index 169fc46..41f1536 100644 --- a/src/datatypes.h +++ b/src/datatypes.h @@ -569,11 +569,13 @@ struct _virStream { virStreamDriverPtr driver; void *privateData; - /* Unfortunately, this can't go into virStreamDriver because - * when register function for skipCb is called, @driver - * is not populated yet. */ + /* Unfortunately, these can't go into virStreamDriver because + * when register function for skipCb or inDataFunc is called, + * @driver is not populated yet. */ virStreamSkipFunc skipCb; void *skipCbOpaque; + virStreamInDataFunc inDataCb; + void *inDataCbOpaque; }; /** diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c index 58665f1..371efed 100644 --- a/src/libvirt-stream.c +++ b/src/libvirt-stream.c @@ -396,6 +396,94 @@ virStreamSkip(virStreamPtr stream, /** + * virStreamRegisterInData: + * @stream: stream + * @inDataCb: callback function + * @opaque: optional application provided data + * + * This function registers callback that will be called whenever + * virStreamInData needs to check whether @stream is currently in + * data or in a hole. This is to be used purely with sparse + * streams. + * + * Returns 0 on success, + * -1 otherwise. + */ +int +virStreamRegisterInData(virStreamPtr stream, + virStreamInDataFunc inDataCb, + void *opaque) +{ + VIR_DEBUG("stream=%p, inDataCb=%p opaque=%p", stream, inDataCb, opaque); + + virResetLastError(); + + virCheckStreamReturn(stream, -1); + virCheckNonNullArgReturn(inDataCb, -1); + + if (stream->inDataCb) { + virReportError(VIR_ERR_OPERATION_INVALID, "%s", + _("A inData callback is already registered")); + return -1; + } + + stream->inDataCb = inDataCb; + stream->inDataCbOpaque = opaque; + return 0; +} + + +/** + * virStreamInData: + * @stream: stream + * @data: are we in data or hole + * @offset: offset to next section + * + * This function will check underlying stream (typically a file) + * whether the current position the stream is in lies in a data + * section or in a hole. Upon return @data is set to a nonzero + * value if former is the case, or to zero if @stream is in a + * hole. Moreover, @offset it updated to tell caller how much + * 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, @offset = 0 + * and return 0. + * + * Returns 0 on success, + * -1 otherwise + */ +int +virStreamInData(virStreamPtr stream, + int *data, + unsigned long long *offset) +{ + VIR_DEBUG("stream=%p, data=%p, offset=%p", stream, data, offset); + + virResetLastError(); + + virCheckStreamReturn(stream, -1); + virCheckNonNullArgReturn(data, -1); + virCheckNonNullArgReturn(offset, -1); + + if (stream->inDataCb) { + int ret; + ret = (stream->inDataCb)(stream, data, offset, stream->inDataCbOpaque); + if (ret < 0) + goto error; + return ret; + } + + virReportUnsupportedError(); + + error: + virDispatchError(stream->conn); + 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 7a75491..d20acbc 100644 --- a/src/libvirt_internal.h +++ b/src/libvirt_internal.h @@ -298,4 +298,7 @@ int virStreamSkipCallback(virStreamPtr stream, unsigned long long offset); +int virStreamInData(virStreamPtr stream, + int *data, + unsigned long long *offset); #endif diff --git a/src/libvirt_private.syms b/src/libvirt_private.syms index e83d5d6..e506d1c 100644 --- a/src/libvirt_private.syms +++ b/src/libvirt_private.syms @@ -1015,6 +1015,7 @@ virStateCleanup; virStateInitialize; virStateReload; virStateStop; +virStreamInData; virStreamSkipCallback; diff --git a/src/libvirt_public.syms b/src/libvirt_public.syms index 0b80d27..396fea5 100644 --- a/src/libvirt_public.syms +++ b/src/libvirt_public.syms @@ -736,6 +736,7 @@ LIBVIRT_1.3.5 { global: virStreamSkip; virStreamRegisterSkip; + virStreamRegisterInData; } LIBVIRT_1.3.3; # .... define new API here using predicted next version number .... -- 2.8.1

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 6bed2c5..b7fd869 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5594,7 +5594,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; @@ -6518,7 +6519,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 624049b..10fde25 100755 --- a/src/rpc/gendispatch.pl +++ b/src/rpc/gendispatch.pl @@ -1654,7 +1654,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 34989a9..8920395 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 = 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.8.1

Even though there's no way how to make stream seekable 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 fde029d..e2e08aa 100644 --- a/daemon/remote.c +++ b/daemon/remote.c @@ -4948,7 +4948,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 c892dcb..02729e4 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -52,6 +52,8 @@ struct daemonClientStream { virNetMessagePtr rx; bool tx; + bool seekable; + daemonClientStreamPtr next; }; @@ -319,7 +321,8 @@ daemonClientStream * daemonCreateClientStream(virNetServerClientPtr client, virStreamPtr st, virNetServerProgramPtr prog, - virNetMessageHeaderPtr header) + virNetMessageHeaderPtr header, + bool seekable) { daemonClientStream *stream; daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client); @@ -337,6 +340,7 @@ daemonCreateClientStream(virNetServerClientPtr client, stream->serial = header->serial; stream->filterID = -1; stream->st = st; + stream->seekable = seekable; 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 b7fd869..a3c331e 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5597,7 +5597,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) { @@ -6522,7 +6523,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 10fde25..f2fa467 100755 --- a/src/rpc/gendispatch.pl +++ b/src/rpc/gendispatch.pl @@ -976,7 +976,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"; } @@ -1654,7 +1654,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 8920395..06d91d1 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -54,6 +54,8 @@ struct _virNetClientStream { virNetMessagePtr rx; bool incomingEOF; + bool seekable; /* User requested seekable 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 seekable) { virNetClientStreamPtr st; @@ -152,6 +155,7 @@ virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, st->prog = prog; st->proc = proc; st->serial = serial; + st->seekable = seekable; virObjectRef(prog); 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.8.1

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 327a334..f73e1aa 100644 --- a/src/rpc/virnetprotocol.x +++ b/src/rpc/virnetprotocol.x @@ -233,3 +233,7 @@ struct virNetMessageError { int int2; virNetMessageNetwork net; /* unused */ }; + +struct virNetStreamSkip { + unsigned hyper offset; +}; diff --git a/src/virnetprotocol-structs b/src/virnetprotocol-structs index af4526c..96b984e 100644 --- a/src/virnetprotocol-structs +++ b/src/virnetprotocol-structs @@ -42,3 +42,6 @@ struct virNetMessageError { int int2; virNetMessageNetwork net; }; +struct virNetStreamSkip { + uint64_t offset; +}; -- 2.8.1

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 | 40 ++++++++++++++++++++++++++++++++++++ tools/wireshark/src/packet-libvirt.h | 2 ++ 2 files changed, 42 insertions(+) diff --git a/tools/wireshark/src/packet-libvirt.c b/tools/wireshark/src/packet-libvirt.c index aa1c323..fb44135 100644 --- a/tools/wireshark/src/packet-libvirt.c +++ b/tools/wireshark/src/packet-libvirt.c @@ -52,8 +52,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_offset = -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 \ @@ -328,6 +331,28 @@ 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; + + ti = proto_tree_add_item(tree, hf_libvirt_stream_skip, tvb, start, -1, ENC_NA); + tree = proto_item_add_subtree(ti, ett_libvirt_stream_skip); + + hf = hf_libvirt_stream_skip_offset; + 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_offset, 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) @@ -348,6 +373,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; } @@ -517,6 +544,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_offset, + { "offset", "libvirt.stream_skip.offset", + FT_UINT64, BASE_DEC, + NULL, 0x0, + NULL, HFILL} + }, { &hf_libvirt_unknown, { "unknown", "libvirt.unknown", FT_BYTES, BASE_NONE, @@ -527,6 +566,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.8.1

This is just a helper function that takes in an offset 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 66f9383..18d3f2d 100644 --- a/src/libvirt_remote.syms +++ b/src/libvirt_remote.syms @@ -168,6 +168,7 @@ virNetServerProgramNew; virNetServerProgramSendReplyError; virNetServerProgramSendStreamData; virNetServerProgramSendStreamError; +virNetServerProgramSendStreamSkip; virNetServerProgramUnknownError; diff --git a/src/rpc/virnetserverprogram.c b/src/rpc/virnetserverprogram.c index d1597f4..fbdb179 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 offset) +{ + virNetStreamSkip data; + + VIR_DEBUG("client=%p msg=%p offset=%llu", client, msg, offset); + + memset(&data, 0, sizeof(data)); + data.offset = offset; + + 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..545aa3d 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 offset); + #endif /* __VIR_NET_SERVER_PROGRAM_H__ */ -- 2.8.1

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 finally call the virStreamSkipCallback(). 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 bfd76d7..e4dee9b 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 @@ -358,6 +359,68 @@ int virNetClientStreamSendPacket(virNetClientStreamPtr st, return -1; } + +static int ATTRIBUTE_UNUSED +virNetClientStreamHandleSkip(virNetClientPtr client, + virNetClientStreamPtr st) +{ + virNetMessagePtr msg; + virNetStreamSkip data; + int rv, 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 || + 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->seekable) { + virReportError(VIR_ERR_RPC, "%s", + _("Unexpected stream seek")); + goto cleanup; + } + + if (virNetMessageDecodePayload(msg, + (xdrproc_t) xdr_virNetStreamSkip, + &data) < 0) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("Malformed stream seek packet")); + goto cleanup; + } + + virNetMessageQueueServe(&st->rx); + virNetMessageFree(msg); + + virObjectUnlock(st); + rv = virStreamSkipCallback(st->stream, data.offset); + virObjectLock(st); + + if (rv < 0) + goto cleanup; + + ret = 0; + cleanup: + if (ret < 0) { + /* Abort stream? */ + } + return ret; +} + + int virNetClientStreamRecvPacket(virNetClientStreamPtr st, virNetClientPtr client, char *data, -- 2.8.1

Whenever client is able to receive some data from stream daemonStreamHandleRead is called. But now the behaviour of this function needs to be changed a bit. Previously it just read data from underlying file (of chardev or whatever) and sent those through the stream to client. This model will not work any longer because it does not differentiate whether underlying file is in data or hole section. Therefore, at the beginning of this function add code that checks this situation and acts accordingly. So after the this, when wanting to send some data we always check whether we are not in a hole and if so, skip it an inform client about its size. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- daemon/stream.c | 51 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/daemon/stream.c b/daemon/stream.c index 1a7d334..5bf6f23 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -766,6 +766,8 @@ daemonStreamHandleRead(virNetServerClientPtr client, size_t bufferLen = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX; int ret = -1; int rv; + int inData = 0; + unsigned long long offset; VIR_DEBUG("client=%p, stream=%p tx=%d closed=%d", client, stream, stream->tx, stream->closed); @@ -790,6 +792,54 @@ daemonStreamHandleRead(virNetServerClientPtr client, if (!(msg = virNetMessageNew(false))) goto cleanup; + if (stream->seekable) { + /* 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. */ + + rv = virStreamInData(stream->st, &inData, &offset); + VIR_DEBUG("rv=%d inData=%d offset=%llu", rv, inData, offset); + + if (rv < 0) { + if (virNetServerProgramSendStreamError(remoteProgram, + client, + msg, + &rerr, + stream->procedure, + stream->serial) < 0) + goto cleanup; + msg = NULL; + + /* We're done with this call */ + goto done; + } else { + if (!inData && offset) { + stream->tx = false; + msg->cb = daemonStreamMessageFinished; + msg->opaque = stream; + stream->refs++; + if (virNetServerProgramSendStreamSkip(remoteProgram, + client, + msg, + stream->procedure, + stream->serial, + offset) < 0) + goto cleanup; + + msg = NULL; + + /* We have successfully sent stream skip to the other side. + * To keep streams in sync seek locally too. */ + virStreamSkipCallback(stream->st, offset); + /* We're done with this call */ + goto done; + } + } + + if (offset < bufferLen) + bufferLen = offset; + } + rv = virStreamRecv(stream->st, buffer, bufferLen); if (rv == -2) { /* Should never get this, since we're only called when we know @@ -821,6 +871,7 @@ daemonStreamHandleRead(virNetServerClientPtr client, msg = NULL; } + done: ret = 0; cleanup: VIR_FREE(buffer); -- 2.8.1

Plenty of clients use virStreamSendAll() instead of using virStreamSend() directly. We should allow those clients to utilize the new feature too. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/libvirt-stream.c | 37 +++++++++++++++++++++++++++++++------ 1 file changed, 31 insertions(+), 6 deletions(-) diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c index 371efed..989c0b2 100644 --- a/src/libvirt-stream.c +++ b/src/libvirt-stream.c @@ -528,9 +528,10 @@ virStreamSendAll(virStreamPtr stream, virStreamSourceFunc handler, void *opaque) { - char *bytes = NULL; - size_t want = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX; + char *buf = NULL; + size_t bufSize = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX; int ret = -1; + unsigned long long dataLen = 0; VIR_DEBUG("stream=%p, handler=%p, opaque=%p", stream, handler, opaque); virResetLastError(); @@ -544,12 +545,35 @@ virStreamSendAll(virStreamPtr stream, goto cleanup; } - if (VIR_ALLOC_N(bytes, want) < 0) + if (VIR_ALLOC_N(buf, bufSize) < 0) goto cleanup; for (;;) { + size_t want = bufSize; int got, offset = 0; - got = (handler)(stream, bytes, want, opaque); + + if (stream->inDataCb && !dataLen) { + int inData = 0; + int rv = virStreamInData(stream, &inData, &dataLen); + + if (rv < 0) { + virStreamAbort(stream); + goto cleanup; + } + + if (!inData && dataLen) { + if (virStreamSkip(stream, dataLen) < 0 || + virStreamSkipCallback(stream, dataLen) < 0) { + virStreamAbort(stream); + goto cleanup; + } + } + } + + if (dataLen && want > dataLen) + want = dataLen; + + got = (handler)(stream, buf, want, opaque); if (got < 0) { virStreamAbort(stream); goto cleanup; @@ -558,16 +582,17 @@ virStreamSendAll(virStreamPtr stream, break; while (offset < got) { int done; - done = virStreamSend(stream, bytes + offset, got - offset); + done = virStreamSend(stream, buf + offset, got - offset); if (done < 0) goto cleanup; offset += done; + dataLen -= done; } } ret = 0; cleanup: - VIR_FREE(bytes); + VIR_FREE(buf); if (ret != 0) virDispatchError(stream->conn); -- 2.8.1

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 f2fa467..2abe186 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}; @@ -934,6 +941,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"; @@ -976,7 +988,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"; } @@ -1643,6 +1655,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"; @@ -1654,7 +1671,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.8.1

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-volume.c | 15 ++++++++++++++- tools/virsh.c | 8 ++++++++ tools/virsh.h | 3 +++ tools/virsh.pod | 3 ++- 4 files changed, 27 insertions(+), 2 deletions(-) diff --git a/tools/virsh-volume.c b/tools/virsh-volume.c index 9cc8e52..018e0f3 100644 --- a/tools/virsh-volume.c +++ b/tools/virsh-volume.c @@ -762,6 +762,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} }; @@ -777,6 +781,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; @@ -790,6 +795,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) { @@ -805,7 +813,12 @@ cmdVolDownload(vshControl *ctl, const vshCmd *cmd) goto cleanup; } - if (virStorageVolDownload(vol, st, offset, length, 0) < 0) { + if (virStreamRegisterSkip(st, virshStreamSkip, &fd) < 0) { + vshError(ctl, _("cannot register stream skip handling function")); + goto cleanup; + } + + if (virStorageVolDownload(vol, st, offset, length, flags) < 0) { vshError(ctl, _("cannot download from volume %s"), name); goto cleanup; } diff --git a/tools/virsh.c b/tools/virsh.c index af07251..af4f9d1 100644 --- a/tools/virsh.c +++ b/tools/virsh.c @@ -257,6 +257,14 @@ int virshStreamSink(virStreamPtr st ATTRIBUTE_UNUSED, return safewrite(*fd, bytes, nbytes); } +int virshStreamSkip(virStreamPtr st ATTRIBUTE_UNUSED, + unsigned long long offset, void *opaque) +{ + int *fd = opaque; + + return lseek(*fd, offset, SEEK_CUR) == (off_t) -1 ? -1 : 0; +} + /* --------------- * Command Connect * --------------- diff --git a/tools/virsh.h b/tools/virsh.h index fd552bb..5c382ef 100644 --- a/tools/virsh.h +++ b/tools/virsh.h @@ -150,4 +150,7 @@ int virshDomainState(vshControl *ctl, virDomainPtr dom, int *reason); int virshStreamSink(virStreamPtr st, const char *bytes, size_t nbytes, void *opaque); +int virshStreamSkip(virStreamPtr st, + unsigned long long offset, void *opaque); + #endif /* VIRSH_H */ diff --git a/tools/virsh.pod b/tools/virsh.pod index 90ab47d..64f0f5f 100644 --- a/tools/virsh.pod +++ b/tools/virsh.pod @@ -3575,12 +3575,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.8.1

This is kind of a hacky approach to the following problem, but so far I am unable to come up with anything better. On some occasions (esp. when dealing with regular files) libvirt_iohelper is spawned to prefetch data for us. We will then have a pipe then for reading the data from it. This does not fit in our sparse stream implementation as one simply doesn't lseek() over a pipe. Until this is resolved, let's suppress use of the IO helper and read data from FD directly. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/fdstream.c | 58 +++++++++++++++++++++++++++---------------- src/fdstream.h | 3 ++- src/storage/storage_backend.c | 6 +++-- 3 files changed, 43 insertions(+), 24 deletions(-) diff --git a/src/fdstream.c b/src/fdstream.c index 38342a7..41e9e06 100644 --- a/src/fdstream.c +++ b/src/fdstream.c @@ -78,6 +78,11 @@ struct virFDStreamData { virMutex lock; }; +enum { + VIR_FDSTREAM_OPEN_FORCE_IOHELPER = 1 << 0, + VIR_FDSTREAM_OPEN_SPARSE = 1 << 1, +}; + static int virFDStreamRemoveCallback(virStreamPtr stream) { @@ -574,7 +579,8 @@ static int virFDStreamOpenInternal(virStreamPtr st, int fd, virCommandPtr cmd, int errfd, - unsigned long long length) + unsigned long long length, + unsigned int flags) { struct virFDStreamData *fdst; @@ -603,8 +609,10 @@ static int virFDStreamOpenInternal(virStreamPtr st, st->driver = &virFDStreamDrv; st->privateData = fdst; - st->skipCb = virFDStreamSkip; - st->inDataCb = virFDStreamInData; + if (flags & VIR_FDSTREAM_OPEN_SPARSE) { + st->skipCb = virFDStreamSkip; + st->inDataCb = virFDStreamInData; + } return 0; } @@ -613,7 +621,7 @@ static int virFDStreamOpenInternal(virStreamPtr st, int virFDStreamOpen(virStreamPtr st, int fd) { - return virFDStreamOpenInternal(st, fd, NULL, -1, 0); + return virFDStreamOpenInternal(st, fd, NULL, -1, 0, 0); } @@ -659,7 +667,7 @@ int virFDStreamConnectUNIX(virStreamPtr st, goto error; } - if (virFDStreamOpenInternal(st, fd, NULL, -1, 0) < 0) + if (virFDStreamOpenInternal(st, fd, NULL, -1, 0, 0) < 0) goto error; return 0; @@ -685,7 +693,7 @@ virFDStreamOpenFileInternal(virStreamPtr st, unsigned long long length, int oflags, int mode, - bool forceIOHelper) + unsigned int flags) { int fd = -1; int childfd = -1; @@ -694,8 +702,8 @@ virFDStreamOpenFileInternal(virStreamPtr st, int errfd = -1; char *iohelper_path = NULL; - VIR_DEBUG("st=%p path=%s oflags=%x offset=%llu length=%llu mode=%o", - st, path, oflags, offset, length, mode); + VIR_DEBUG("st=%p path=%s oflags=%x offset=%llu length=%llu mode=%o flags=%x", + st, path, oflags, offset, length, mode, flags); oflags |= O_NOCTTY | O_BINARY; @@ -729,10 +737,15 @@ virFDStreamOpenFileInternal(virStreamPtr st, * non-blocking I/O on block devs/regular files. To * support those we need to fork a helper process to do * the I/O so we just have a fifo. Or use AIO :-( + * Moreover, when opening a file for a sparse stream, make + * sure we end up with something seekable which a FIFO is + * not. */ - if ((st->flags & VIR_STREAM_NONBLOCK) && - ((!S_ISCHR(sb.st_mode) && - !S_ISFIFO(sb.st_mode)) || forceIOHelper)) { + if (!(flags & VIR_FDSTREAM_OPEN_SPARSE) && + ((st->flags & VIR_STREAM_NONBLOCK) && + ((!S_ISCHR(sb.st_mode) && + !S_ISFIFO(sb.st_mode)) || + flags & VIR_FDSTREAM_OPEN_FORCE_IOHELPER))) { int fds[2] = { -1, -1 }; if ((oflags & O_ACCMODE) == O_RDWR) { @@ -781,7 +794,7 @@ virFDStreamOpenFileInternal(virStreamPtr st, VIR_FORCE_CLOSE(childfd); } - if (virFDStreamOpenInternal(st, fd, cmd, errfd, length) < 0) + if (virFDStreamOpenInternal(st, fd, cmd, errfd, length, flags) < 0) goto error; return 0; @@ -811,7 +824,7 @@ int virFDStreamOpenFile(virStreamPtr st, } return virFDStreamOpenFileInternal(st, path, offset, length, - oflags, 0, false); + oflags, 0, 0); } int virFDStreamCreateFile(virStreamPtr st, @@ -823,8 +836,7 @@ int virFDStreamCreateFile(virStreamPtr st, { return virFDStreamOpenFileInternal(st, path, offset, length, - oflags | O_CREAT, mode, - false); + oflags | O_CREAT, mode, 0); } #ifdef HAVE_CFMAKERAW @@ -839,8 +851,7 @@ int virFDStreamOpenPTY(virStreamPtr st, if (virFDStreamOpenFileInternal(st, path, offset, length, - oflags | O_CREAT, 0, - false) < 0) + oflags | O_CREAT, 0, 0) < 0) return -1; fdst = st->privateData; @@ -876,8 +887,7 @@ int virFDStreamOpenPTY(virStreamPtr st, { return virFDStreamOpenFileInternal(st, path, offset, length, - oflags | O_CREAT, 0, - false); + oflags | O_CREAT, 0, 0); } #endif /* !HAVE_CFMAKERAW */ @@ -885,11 +895,17 @@ int virFDStreamOpenBlockDevice(virStreamPtr st, const char *path, unsigned long long offset, unsigned long long length, - int oflags) + int oflags, + bool sparse) { + unsigned int flags = VIR_FDSTREAM_OPEN_FORCE_IOHELPER; + + if (sparse) + flags |= VIR_FDSTREAM_OPEN_SPARSE; + return virFDStreamOpenFileInternal(st, path, offset, length, - oflags, 0, true); + oflags, 0, flags); } int virFDStreamSetInternalCloseCb(virStreamPtr st, diff --git a/src/fdstream.h b/src/fdstream.h index 2c913ea..bfdebc2 100644 --- a/src/fdstream.h +++ b/src/fdstream.h @@ -60,7 +60,8 @@ int virFDStreamOpenBlockDevice(virStreamPtr st, const char *path, unsigned long long offset, unsigned long long length, - int oflags); + int oflags, + bool sparse); int virFDStreamSetInternalCloseCb(virStreamPtr st, virFDStreamInternalCloseCb cb, diff --git a/src/storage/storage_backend.c b/src/storage/storage_backend.c index 3a23cd7..f92d074 100644 --- a/src/storage/storage_backend.c +++ b/src/storage/storage_backend.c @@ -2030,7 +2030,8 @@ virStorageBackendVolUploadLocal(virConnectPtr conn ATTRIBUTE_UNUSED, /* Not using O_CREAT because the file is required to already exist at * this point */ ret = virFDStreamOpenBlockDevice(stream, target_path, - offset, len, O_WRONLY); + offset, len, O_WRONLY, + flags & VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM); cleanup: VIR_FREE(path); @@ -2068,7 +2069,8 @@ virStorageBackendVolDownloadLocal(virConnectPtr conn ATTRIBUTE_UNUSED, } ret = virFDStreamOpenBlockDevice(stream, target_path, - offset, len, O_RDONLY); + offset, len, O_RDONLY, + flags & VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM); cleanup: VIR_FREE(path); -- 2.8.1

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 5bf6f23..179c5eb 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -54,6 +54,7 @@ struct daemonClientStream { bool tx; bool seekable; + size_t dataLen; /* How much data is there remaining until we see a hole */ daemonClientStreamPtr next; }; @@ -792,7 +793,7 @@ daemonStreamHandleRead(virNetServerClientPtr client, if (!(msg = virNetMessageNew(false))) goto cleanup; - if (stream->seekable) { + if (stream->seekable && !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. */ @@ -836,10 +837,13 @@ daemonStreamHandleRead(virNetServerClientPtr client, } } - if (offset < bufferLen) - bufferLen = offset; + stream->dataLen = offset; } + if (stream->seekable && + 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 @@ -854,6 +858,8 @@ daemonStreamHandleRead(virNetServerClientPtr client, goto cleanup; msg = NULL; } else { + stream->dataLen -= rv; + stream->tx = false; if (rv == 0) stream->recvEOF = true; -- 2.8.1