In virFDStreamRead(), fill buffer from this and next messages

Before this change, buffers returned from virFDStreamRead() would
alternate in size (262120 and 24), because it only consumed the
bytes remaining from the current background thread message.

As the background thread reads 262144 bytes (256kB) of data in
each chunk, where the maximum size returned from virFDStreamRead()
to be transferred over the remote protocol is only 262120, 24 bytes
would be left in the buffer on each iteration. The next iteration
leaves 24 bytes, which used to be returned without considering
messages waiting in the queue.

Reviewed-by: Daniel P. Berrangé <berrange@redhat.com>
Signed-off-by: Erik Huelsmann <ehuels@gmail.com>

diff --git a/src/util/virfdstream.c b/src/util/virfdstream.c
index 26a1f00316db5afb5c383ccd86946d9dd0888eca..4c974ba5d7f626dc92b6d42399186a0bb54628db 100644 (file)
--- a/src/util/virfdstream.c
+++ b/src/util/virfdstream.c
@@ -905,7 +905,10 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes)
 
     if (fdst->thread) {
         virFDStreamMsg *msg = NULL;
+        size_t got = 0;
+        size_t bsz = 0;
 
+    more:
         while (!(msg = fdst->msg)) {
             if (fdst->threadQuit || fdst->threadErr) {
                 if (nbytes) {
@@ -917,7 +920,7 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes)
                         virReportSystemError(EBADF, "%s",
                                              _("stream is not open"));
                 } else {
-                    ret = 0;
+                    ret = got;
                 }
                 goto cleanup;
             } else {
@@ -931,7 +934,7 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes)
          * return 0 immediately. */
         if (msg->type == VIR_FDSTREAM_MSG_TYPE_HOLE &&
             msg->stream.hole.len == 0) {
-            ret = 0;
+            ret = got;
             goto cleanup;
         }
 
@@ -942,21 +945,26 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes)
             goto cleanup;
         }
 
-        if (nbytes > msg->stream.data.len - msg->stream.data.offset)
-            nbytes = msg->stream.data.len - msg->stream.data.offset;
+        bsz = msg->stream.data.len - msg->stream.data.offset;
+        if (nbytes < bsz)
+            bsz = nbytes;
 
-        memcpy(bytes,
+        memcpy(bytes + got,
                msg->stream.data.buf + msg->stream.data.offset,
-               nbytes);
+               bsz);
+        got += bsz;
+        nbytes -= bsz;
 
-        msg->stream.data.offset += nbytes;
+        msg->stream.data.offset += bsz;
         if (msg->stream.data.offset == msg->stream.data.len) {
             virFDStreamMsgQueuePop(fdst, fdst->fd, "pipe");
             virFDStreamMsgFree(msg);
         }
 
-        ret = nbytes;
-
+        ret = got;
+        if (nbytes > 0) {
+            goto more;
+        }
     } else {
      retry:
         ret = read(fdst->fd, bytes, nbytes);