The stream serial number is the serial number of the RPC call that initiated a data transfer. And as such can never be negative. Moreover, when looking up internal state for a stream, the serial numbers are compared. But hey, the serial number in message header is unsigned too! Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- daemon/stream.c | 22 +++++++++++----------- src/rpc/virnetserverprogram.c | 12 ++++++------ src/rpc/virnetserverprogram.h | 4 ++-- 3 files changed, 19 insertions(+), 19 deletions(-) diff --git a/daemon/stream.c b/daemon/stream.c index dfe0bf9..ce1e054 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -42,7 +42,7 @@ struct daemonClientStream { virStreamPtr st; int procedure; - int serial; + unsigned int serial; unsigned int recvEOF : 1; unsigned int closed : 1; @@ -92,11 +92,11 @@ daemonStreamUpdateEvents(daemonClientStream *stream) * fast stream, but slow client */ static void -daemonStreamMessageFinished(virNetMessagePtr msg ATTRIBUTE_UNUSED, +daemonStreamMessageFinished(virNetMessagePtr msg, void *opaque) { daemonClientStream *stream = opaque; - VIR_DEBUG("stream=%p proc=%d serial=%d", + VIR_DEBUG("stream=%p proc=%d serial=%u", stream, msg->header.proc, msg->header.serial); stream->tx = 1; @@ -293,7 +293,7 @@ daemonStreamFilter(virNetServerClientPtr client ATTRIBUTE_UNUSED, msg->header.serial != stream->serial) goto cleanup; - VIR_DEBUG("Incoming client=%p, rx=%p, serial=%d, proc=%d, status=%d", + VIR_DEBUG("Incoming client=%p, rx=%p, serial=%u, proc=%d, status=%d", client, stream->rx, msg->header.proc, msg->header.serial, msg->header.status); @@ -324,7 +324,7 @@ daemonCreateClientStream(virNetServerClientPtr client, daemonClientStream *stream; daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client); - VIR_DEBUG("client=%p, proc=%d, serial=%d, st=%p", + VIR_DEBUG("client=%p, proc=%d, serial=%u, st=%p", client, header->proc, header->serial, st); if (VIR_ALLOC(stream) < 0) @@ -360,7 +360,7 @@ int daemonFreeClientStream(virNetServerClientPtr client, if (stream->refs) return 0; - VIR_DEBUG("client=%p, proc=%d, serial=%d", + VIR_DEBUG("client=%p, proc=%d, serial=%u", client, stream->procedure, stream->serial); virObjectUnref(stream->prog); @@ -398,7 +398,7 @@ int daemonAddClientStream(virNetServerClientPtr client, daemonClientStream *stream, bool transmit) { - VIR_DEBUG("client=%p, proc=%d, serial=%d, st=%p, transmit=%d", + VIR_DEBUG("client=%p, proc=%d, serial=%u, st=%p, transmit=%d", client, stream->procedure, stream->serial, stream->st, transmit); daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client); @@ -448,7 +448,7 @@ int daemonRemoveClientStream(virNetServerClientPtr client, daemonClientStream *stream) { - VIR_DEBUG("client=%p, proc=%d, serial=%d, st=%p", + VIR_DEBUG("client=%p, proc=%d, serial=%u, st=%p", client, stream->procedure, stream->serial, stream->st); daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client); daemonClientStream *curr = priv->streams; @@ -515,7 +515,7 @@ daemonStreamHandleWriteData(virNetServerClientPtr client, { int ret; - VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%d, len=%zu, offset=%zu", + VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%u, len=%zu, offset=%zu", client, stream, msg->header.proc, msg->header.serial, msg->bufferLength, msg->bufferOffset); @@ -565,7 +565,7 @@ daemonStreamHandleFinish(virNetServerClientPtr client, { int ret; - VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%d", + VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%u", client, stream, msg->header.proc, msg->header.serial); stream->closed = 1; @@ -602,7 +602,7 @@ daemonStreamHandleAbort(virNetServerClientPtr client, daemonClientStream *stream, virNetMessagePtr msg) { - VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%d", + VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%u", client, stream, msg->header.proc, msg->header.serial); virNetMessageError rerr; diff --git a/src/rpc/virnetserverprogram.c b/src/rpc/virnetserverprogram.c index a4d9295..311e344 100644 --- a/src/rpc/virnetserverprogram.c +++ b/src/rpc/virnetserverprogram.c @@ -147,9 +147,9 @@ virNetServerProgramSendError(unsigned program, virNetMessageErrorPtr rerr, int procedure, int type, - int serial) + unsigned int serial) { - VIR_DEBUG("prog=%d ver=%d proc=%d type=%d serial=%d msg=%p rerr=%p", + VIR_DEBUG("prog=%d ver=%d proc=%d type=%d serial=%u msg=%p rerr=%p", program, version, procedure, type, serial, msg, rerr); virNetMessageSaveError(rerr); @@ -217,7 +217,7 @@ int virNetServerProgramSendStreamError(virNetServerProgramPtr prog, virNetMessagePtr msg, virNetMessageErrorPtr rerr, int procedure, - int serial) + unsigned int serial) { return virNetServerProgramSendError(prog->program, prog->version, @@ -282,7 +282,7 @@ int virNetServerProgramDispatch(virNetServerProgramPtr prog, memset(&rerr, 0, sizeof(rerr)); - VIR_DEBUG("prog=%d ver=%d type=%d status=%d serial=%d proc=%d", + VIR_DEBUG("prog=%d ver=%d type=%d status=%d serial=%u proc=%d", msg->header.prog, msg->header.vers, msg->header.type, msg->header.status, msg->header.serial, msg->header.proc); @@ -312,7 +312,7 @@ int virNetServerProgramDispatch(virNetServerProgramPtr prog, * stream packets after we closed down a stream. Just drop & ignore * these. */ - VIR_INFO("Ignoring unexpected stream data serial=%d proc=%d status=%d", + VIR_INFO("Ignoring unexpected stream data serial=%u proc=%d status=%d", msg->header.serial, msg->header.proc, msg->header.status); /* Send a dummy reply to free up 'msg' & unblock client rx */ virNetMessageClear(msg); @@ -510,7 +510,7 @@ int virNetServerProgramSendStreamData(virNetServerProgramPtr prog, virNetServerClientPtr client, virNetMessagePtr msg, int procedure, - int serial, + unsigned int serial, const char *data, size_t len) { diff --git a/src/rpc/virnetserverprogram.h b/src/rpc/virnetserverprogram.h index 0ccc372..531fca0 100644 --- a/src/rpc/virnetserverprogram.h +++ b/src/rpc/virnetserverprogram.h @@ -90,7 +90,7 @@ int virNetServerProgramSendStreamError(virNetServerProgramPtr prog, virNetMessagePtr msg, virNetMessageErrorPtr rerr, int procedure, - int serial); + unsigned int serial); int virNetServerProgramUnknownError(virNetServerClientPtr client, virNetMessagePtr msg, @@ -100,7 +100,7 @@ int virNetServerProgramSendStreamData(virNetServerProgramPtr prog, virNetServerClientPtr client, virNetMessagePtr msg, int procedure, - int serial, + unsigned int serial, const char *data, size_t len); -- 2.7.3

This structure item is used as pure boolean. There's no need to hold whole integer for it. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- daemon/stream.c | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/daemon/stream.c b/daemon/stream.c index 2a9e6e4..a2a370c 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -50,7 +50,7 @@ struct daemonClientStream { int filterID; virNetMessagePtr rx; - int tx; + bool tx; daemonClientStreamPtr next; }; @@ -99,7 +99,7 @@ daemonStreamMessageFinished(virNetMessagePtr msg, VIR_DEBUG("stream=%p proc=%d serial=%u", stream, msg->header.proc, msg->header.serial); - stream->tx = 1; + stream->tx = true; daemonStreamUpdateEvents(stream); daemonFreeClientStream(NULL, stream); @@ -197,7 +197,7 @@ daemonStreamEvent(virStreamPtr st, int events, void *opaque) (events & VIR_STREAM_EVENT_HANGUP)) { virNetMessagePtr msg; events &= ~(VIR_STREAM_EVENT_HANGUP); - stream->tx = 0; + stream->tx = false; stream->recvEOF = true; if (!(msg = virNetMessageNew(false))) { daemonRemoveClientStream(client, stream); @@ -422,7 +422,7 @@ int daemonAddClientStream(virNetServerClientPtr client, } if (transmit) - stream->tx = 1; + stream->tx = true; virMutexLock(&priv->lock); stream->next = priv->streams; @@ -753,7 +753,7 @@ daemonStreamHandleRead(virNetServerClientPtr client, stream->serial); } else { virNetMessagePtr msg; - stream->tx = 0; + stream->tx = false; if (ret == 0) stream->recvEOF = true; if (!(msg = virNetMessageNew(false))) -- 2.7.3

After 434de30da545aea137 the status values are prefixed VIR_NET_ rather than REMOTE_. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/virnetclient.c | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index c68da6d..781e74c 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -1063,9 +1063,9 @@ static int virNetClientCallDispatchStream(virNetClientPtr client) VIR_DEBUG("Found call %p", thecall); /* Status is either - * - REMOTE_OK - no payload for streams - * - REMOTE_ERROR - followed by a remote_error struct - * - REMOTE_CONTINUE - followed by a raw data packet + * - VIR_NET_OK - no payload for streams + * - VIR_NET_ERROR - followed by a remote_error struct + * - VIR_NET_CONTINUE - followed by a raw data packet */ switch (client->msg.header.status) { case VIR_NET_CONTINUE: { -- 2.7.3

Usually, we have this 'if() goto cleanup;' pattern in our new code. It is going to be useful here too. Thing is, there was a memleak. If there has been an error in virNetServerProgramSendStreamError() or virNetServerProgramSendStreamData() created message was never freed. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- daemon/stream.c | 68 ++++++++++++++++++++++++++++----------------------------- 1 file changed, 34 insertions(+), 34 deletions(-) diff --git a/daemon/stream.c b/daemon/stream.c index a2a370c..cf42a10 100644 --- a/daemon/stream.c +++ b/daemon/stream.c @@ -709,9 +709,12 @@ static int daemonStreamHandleRead(virNetServerClientPtr client, daemonClientStream *stream) { + virNetMessagePtr msg = NULL; + virNetMessageError rerr; char *buffer; size_t bufferLen = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX; - int ret; + int ret = -1; + int rv; VIR_DEBUG("client=%p, stream=%p tx=%d closed=%d", client, stream, stream->tx, stream->closed); @@ -728,50 +731,47 @@ daemonStreamHandleRead(virNetServerClientPtr client, if (!stream->tx) return 0; + memset(&rerr, 0, sizeof(rerr)); + if (VIR_ALLOC_N(buffer, bufferLen) < 0) return -1; - ret = virStreamRecv(stream->st, buffer, bufferLen); - if (ret == -2) { + if (!(msg = virNetMessageNew(false))) + goto cleanup; + + rv = virStreamRecv(stream->st, buffer, bufferLen); + if (rv == -2) { /* Should never get this, since we're only called when we know * we're readable, but hey things change... */ - ret = 0; - } else if (ret < 0) { - virNetMessagePtr msg; - virNetMessageError rerr; - - memset(&rerr, 0, sizeof(rerr)); - - if (!(msg = virNetMessageNew(false))) - ret = -1; - else - ret = virNetServerProgramSendStreamError(remoteProgram, - client, - msg, - &rerr, - stream->procedure, - stream->serial); + } else if (rv < 0) { + if (virNetServerProgramSendStreamError(remoteProgram, + client, + msg, + &rerr, + stream->procedure, + stream->serial) < 0) + goto cleanup; } else { - virNetMessagePtr msg; stream->tx = false; - if (ret == 0) + if (rv == 0) stream->recvEOF = true; - if (!(msg = virNetMessageNew(false))) - ret = -1; - if (msg) { - msg->cb = daemonStreamMessageFinished; - msg->opaque = stream; - stream->refs++; - ret = virNetServerProgramSendStreamData(remoteProgram, - client, - msg, - stream->procedure, - stream->serial, - buffer, ret); - } + msg->cb = daemonStreamMessageFinished; + msg->opaque = stream; + stream->refs++; + if (virNetServerProgramSendStreamData(remoteProgram, + client, + msg, + stream->procedure, + stream->serial, + buffer, rv) < 0) + goto cleanup; } + msg = NULL; + ret = 0; + cleanup: VIR_FREE(buffer); + virNetMessageFree(msg); return ret; } -- 2.7.3

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.7.3

On 04/15/2016 09:51 AM, Michal Privoznik wrote: s/then/than the/ Can I "assume" this testing covers the reverted patch scenario. IOW: I think this needs https://bugzilla.redhat.com/show_bug.cgi?id=1026136 to be reopened... Might have been "nice" to indicate/summarize what this algorithm does as opposed to the other. I think you started at the end of the first paragraph, but I'm not 100% sure - I guess it's easier for me if it's explicitly said, such as: In virNetClientStreamQueuePacket instead of ... In virNetClientStreamRecvPacket instead of ... instead of implicitly said if you know the code. The functions are just tough to read without (more) knowledge (than I have about them) of how they are designed to function. Since he had a hand in the above bug, hopefully Martin can take a look at this patch. So if 'st->rx == NULL', then 'rv = nbytes - want;' or 0 - I assume that is 'expected'... Nothing needs to be done with want here? I guess this shows my lack of depth of understanding of these algorithms... Big black box that I hope works without me needing to intervene! John

On 04/21/2016 10:28 AM, Michal Privoznik wrote: ...just processes ... and stores ... New paragraph before "Change" ...a new receive (rx) queue... ... are queue... ... data is copied... The only reason I suggested is I think technically the revert makes the previous changes essentially NULL and void. Since that commit was connected with a bug #, I just wanted to be sure "process wise" we're covered... It's not that important though. OK... I see , instead of allocating and copying data from incoming stream socket into a buffer to only be copied again into the client buffer, we'll "steal" the entire buffer destined for the client from the incoming stream socket and then create a queue for the client to copy - seems OK to me... ACK for 7-8 John BTW: I know it's existing, but virNetMessageQueueServe caused me to go look and see it's really a virNetMessageQueuePop to complement the virNetMessageQueuePush (sigh)

On 04/15/2016 09:51 AM, Michal Privoznik wrote: ^^^ I assume you mean, "s/for/four" John