nbd: Use common read/write-all qio functions

The new qio_channel_{read,write}{,v}_all functions are documented as yielding until data is available. When used on a blocking channel, this yield is done via qio_channel_wait() which spawns a new coroutine under the hood (so it is the new entry point that yields as needed); but if we are already in a coroutine (at which

point QIO_CHANNEL_ERR_BLOCK is only possible if we are a non-blocking channel), we want to yield the current coroutine instead of spawning a new one.

Signed-off-by Eric Blake <[email protected]>

---
 io/channel.c | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

io/channel.c +5 -1

index 5e8c2f0a91..9e62794cab 100644
@@ -105,7 +105,11 @@ int qio_channel_readv_all(QIOChannel *ioc,
         ssize_t len;
         len = qio_channel_readv(ioc, local_iov, nlocal_iov, errp);
         if (len == QIO_CHANNEL_ERR_BLOCK) {
-            qio_channel_wait(ioc, G_IO_IN);
+            if (qemu_in_coroutine()) {
+                qio_channel_yield(ioc, G_IO_IN);
+            } else {
+                qio_channel_wait(ioc, G_IO_IN);
+            }
             continue;
         } else if (len < 0) {
             goto cleanup;

@@ -143,7 +147,11 @@ int qio_channel_writev_all(QIOChannel *ioc,
         ssize_t len;
         len = qio_channel_writev(ioc, local_iov, nlocal_iov, errp);

The new qio_channel_{read,write}{,v}_all functions are documented as yielding until data is available. When used on a blocking channel, this yield is done via qio_channel_wait() which spawns a new coroutine under the hood (so it is the new entry point that yields as needed); but if we are already in a coroutine (at which

point QIO_CHANNEL_ERR_BLOCK is only possible if we are a non-blocking channel), we want to yield the current coroutine instead of spawning a new one.

Signed-off-by Eric Blake <[email protected]>

---
 io/channel.c | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

io/channel.c +5 -1

index 5e8c2f0a91..9e62794cab 100644
@@ -105,7 +105,11 @@ int qio_channel_readv_all(QIOChannel *ioc,
         ssize_t len;
         len = qio_channel_readv(ioc, local_iov, nlocal_iov, errp);
         if (len == QIO_CHANNEL_ERR_BLOCK) {
-            qio_channel_wait(ioc, G_IO_IN);
+            if (qemu_in_coroutine()) {
+                qio_channel_yield(ioc, G_IO_IN);
+            } else {
+                qio_channel_wait(ioc, G_IO_IN);
+            }
             continue;
         } else if (len < 0) {
             goto cleanup;

@@ -143,7 +147,11 @@ int qio_channel_writev_all(QIOChannel *ioc,
         ssize_t len;
         len = qio_channel_writev(ioc, local_iov, nlocal_iov, errp);

Quoted 5 lines · Eric Blake

The new qio_channel_{read,write}{,v}_all functions are documented as yielding until data is available. When used on a blocking channel, this yield is done via qio_channel_wait() which spawns a new coroutine under the hood (so it is the new entry point that yields as needed); but if we are already in a coroutine (at which

qio_channel_wait doesn't spawn any coroutine - it simply rnus a nested event loop to wait for the channel...

Quoted 3 lines

point QIO_CHANNEL_ERR_BLOCK is only possible if we are a non-blocking channel), we want to yield the current coroutine instead of spawning a new one.

...none the less, I think this is ok.

Acked-by Daniel P. Berrange <[email protected]>

Regards,
Daniel

Some callers want to distinguish between clean EOF (no bytes read) vs. a short read (at least one byte read, but EOF encountered before reaching the desired length), as it allows clients the ability to do a graceful shutdown when a server shuts down at defined safe points in the protocol, rather than treating all shutdown scenarios as an error due to EOF. However, we don't want to require all callers to have to check for early EOF. So add another wrapper function that can be used by the callers that care about the distinction.

Signed-off-by Eric Blake <[email protected]>

---
 include/io/channel.h | 53 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 io/channel.c         | 48 +++++++++++++++++++++++++++++++++++++++--------
 2 files changed, 93 insertions(+), 8 deletions(-)

include/io/channel.h +53 -0

index 8f25893c45..3995e243a3 100644
@@ -269,6 +269,36 @@ ssize_t qio_channel_writev_full(QIOChannel *ioc,
                                 Error **errp);
 
 /**
+ * qio_channel_readv_all_eof:
+ * @ioc: the channel object
+ * @iov: the array of memory regions to read data into
+ * @niov: the length of the @iov array
+ * @errp: pointer to a NULL-initialized error object
+ *
+ * Read data from the IO channel, storing it in the
+ * memory regions referenced by @iov. Each element
+ * in the @iov will be fully populated with data
+ * before the next one is used. The @niov parameter
+ * specifies the total number of elements in @iov.
+ *
+ * The function will wait for all requested data
+ * to be read, yielding from the current coroutine
+ * if required.
+ *
+ * If end-of-file occurs before any data is read,
+ * no error is reported; otherwise, if it occurs
+ * before all requested data has been read, an error
+ * will be reported.
+ *
+ * Returns: 1 if all bytes were read, 0 if end-of-file
+ *          occurs without data, or -1 on error
+ */
+int qio_channel_readv_all_eof(QIOChannel *ioc,
+                              const struct iovec *iov,
+                              size_t niov,
+                              Error **errp);
+
+/**
  * qio_channel_readv_all:
  * @ioc: the channel object
  * @iov: the array of memory regions to read data into

@@ -383,6 +413,28 @@ ssize_t qio_channel_write(QIOChannel *ioc,
                           Error **errp);
 
 /**
+ * qio_channel_read_all_eof:
+ * @ioc: the channel object
+ * @buf: the memory region to read data into
+ * @buflen: the number of bytes to @buf
+ * @errp: pointer to a NULL-initialized error object
+ *
+ * Reads @buflen bytes into @buf, possibly blocking or (if the
+ * channel is non-blocking) yielding from the current coroutine
+ * multiple times until the entire content is read. If end-of-file
+ * occurs immediately it is not an error, but if it occurs after
+ * data has been read it will return an error rather than a
+ * short-read. Otherwise behaves as qio_channel_read().
+ *
+ * Returns: 1 if all bytes were read, 0 if end-of-file occurs
+ *          without data, or -1 on error
+ */
+int qio_channel_read_all_eof(QIOChannel *ioc,
+                             char *buf,
+                             size_t buflen,
+                             Error **errp);
+
+/**
  * qio_channel_read_all:
  * @ioc: the channel object
  * @buf: the memory region to read data into

@@ -401,6 +453,7 @@ int qio_channel_read_all(QIOChannel *ioc,
                          char *buf,
                          size_t buflen,
                          Error **errp);
+
 /**
  * qio_channel_write_all:
  * @ioc: the channel object

io/channel.c +33 -8

index 9e62794cab..ec4b86de7c 100644
@@ -86,16 +86,16 @@ ssize_t qio_channel_writev_full(QIOChannel *ioc,
 }
 
 
-
-int qio_channel_readv_all(QIOChannel *ioc,
-                          const struct iovec *iov,
-                          size_t niov,
-                          Error **errp)
+int qio_channel_readv_all_eof(QIOChannel *ioc,
+                              const struct iovec *iov,
+                              size_t niov,
+                              Error **errp)
 {
     int ret = -1;
     struct iovec *local_iov = g_new(struct iovec, niov);
     struct iovec *local_iov_head = local_iov;
     unsigned int nlocal_iov = niov;
+    bool partial = false;
 
     nlocal_iov = iov_copy(local_iov, nlocal_iov,
                           iov, niov,

@@ -114,21 +114,43 @@ int qio_channel_readv_all(QIOChannel *ioc,
         } else if (len < 0) {
             goto cleanup;
         } else if (len == 0) {
-            error_setg(errp,
-                       "Unexpected end-of-file before all bytes were read");
+            if (partial) {
+                error_setg(errp,
+                           "Unexpected end-of-file before all bytes were read");
+            } else {
+                ret = 0;
+            }
             goto cleanup;
         }
 
+        partial = true;
         iov_discard_front(&local_iov, &nlocal_iov, len);
     }
 
-    ret = 0;
+    ret = 1;
 
  cleanup:
     g_free(local_iov_head);
     return ret;
 }
 
+int qio_channel_readv_all(QIOChannel *ioc,
+                          const struct iovec *iov,
+                          size_t niov,
+                          Error **errp)
+{
+    int ret = qio_channel_readv_all_eof(ioc, iov, niov, errp);
+
+    if (ret == 0) {
+        ret = -1;
+        error_setg(errp,
+                   "Unexpected end-of-file before all bytes were read");
+    } else if (ret == 1) {
+        ret = 0;
+    }
+    return ret;
+}
+
 int qio_channel_writev_all(QIOChannel *ioc,
                            const struct iovec *iov,
                            size_t niov,

@@ -205,6 +227,16 @@ ssize_t qio_channel_write(QIOChannel *ioc,
 }
 
 
+int qio_channel_read_all_eof(QIOChannel *ioc,
+                             char *buf,
+                             size_t buflen,

Rather than open-coding our own read/write-all functions, we can make use of the recently-added qio code. It slightly changes the error message in one of the iotests.

Signed-off-by Eric Blake <[email protected]>

---
 include/block/nbd.h        |  2 --
 nbd/nbd-internal.h         | 41 +++++++----------------------------------
 block/nbd-client.c         | 15 +++++++--------
 nbd/common.c               | 45 ---------------------------------------------
 tests/qemu-iotests/083.out |  8 ++++----
 5 files changed, 18 insertions(+), 93 deletions(-)

include/block/nbd.h +0 -2

index 040cdd2e60..707fd37575 100644
@@ -155,8 +155,6 @@ struct NBDExportInfo {
 };
 typedef struct NBDExportInfo NBDExportInfo;
 
-ssize_t nbd_rwv(QIOChannel *ioc, struct iovec *iov, size_t niov, size_t length,
-                bool do_read, Error **errp);
 int nbd_receive_negotiate(QIOChannel *ioc, const char *name,
                           QCryptoTLSCreds *tlscreds, const char *hostname,
                           QIOChannel **outioc, NBDExportInfo *info,

nbd/nbd-internal.h +7 -34

index 03549e3f39..8a609a227f 100644
@@ -85,28 +85,14 @@
 static inline int nbd_read_eof(QIOChannel *ioc, void *buffer, size_t size,
                                Error **errp)
 {
-    struct iovec iov = { .iov_base = buffer, .iov_len = size };
-    ssize_t ret;
-
-    /* Sockets are kept in blocking mode in the negotiation phase.  After
-     * that, a non-readable socket simply means that another thread stole
-     * our request/reply.  Synchronization is done with recv_coroutine, so
-     * that this is coroutine-safe.
-     */
+    int ret;
 
     assert(size);
-
-    ret = nbd_rwv(ioc, &iov, 1, size, true, errp);
-    if (ret <= 0) {
-        return ret;
+    ret = qio_channel_read_all_eof(ioc, buffer, size, errp);
+    if (ret < 0) {
+        ret = -EIO;
     }
-
-    if (ret != size) {
-        error_setg(errp, "End of file");
-        return -EINVAL;
-    }
-
-    return 1;
+    return ret;
 }
 
 /* nbd_read

@@ -115,14 +101,7 @@ static inline int nbd_read_eof(QIOChannel *ioc, void *buffer, size_t size,
 static inline int nbd_read(QIOChannel *ioc, void *buffer, size_t size,
                            Error **errp)
 {
-    int ret = nbd_read_eof(ioc, buffer, size, errp);
-
-    if (ret == 0) {
-        ret = -EINVAL;
-        error_setg(errp, "End of file");
-    }
-
-    return ret < 0 ? ret : 0;
+    return qio_channel_read_all(ioc, buffer, size, errp) < 0 ? -EIO : 0;
 }
 
 /* nbd_write

@@ -131,13 +110,7 @@ static inline int nbd_read(QIOChannel *ioc, void *buffer, size_t size,
 static inline int nbd_write(QIOChannel *ioc, const void *buffer, size_t size,
                             Error **errp)
 {
-    struct iovec iov = { .iov_base = (void *) buffer, .iov_len = size };
-
-    ssize_t ret = nbd_rwv(ioc, &iov, 1, size, false, errp);
-
-    assert(ret < 0 || ret == size);
-
-    return ret < 0 ? ret : 0;
+    return qio_channel_write_all(ioc, buffer, size, errp) < 0 ? -EIO : 0;
 }
 
 struct NBDTLSHandshakeData {

block/nbd-client.c +7 -8

index f0dbea24d3..ee7f758e68 100644
@@ -121,7 +121,7 @@ static int nbd_co_send_request(BlockDriverState *bs,
                                QEMUIOVector *qiov)
 {
     NBDClientSession *s = nbd_get_client_session(bs);
-    int rc, ret, i;
+    int rc, i;
 
     qemu_co_mutex_lock(&s->send_mutex);
     while (s->in_flight == MAX_NBD_REQUESTS) {

@@ -156,9 +156,9 @@ static int nbd_co_send_request(BlockDriverState *bs,
         qio_channel_set_cork(s->ioc, true);
         rc = nbd_send_request(s->ioc, request);
         if (rc >= 0 && !s->quit) {
-            ret = nbd_rwv(s->ioc, qiov->iov, qiov->niov, request->len, false,
-                          NULL);
-            if (ret != request->len) {
+            assert(request->len == iov_size(qiov->iov, qiov->niov));
+            if (qio_channel_writev_all(s->ioc, qiov->iov, qiov->niov,
+                                       NULL) < 0) {
                 rc = -EIO;
             }
         }

@@ -184,7 +184,6 @@ static void nbd_co_receive_reply(NBDClientSession *s,
                                  QEMUIOVector *qiov)
 {
     int i = HANDLE_TO_INDEX(s, request->handle);
-    int ret;
 
     /* Wait until we're woken up by nbd_read_reply_entry.  */
     s->requests[i].receiving = true;

@@ -195,9 +194,9 @@ static void nbd_co_receive_reply(NBDClientSession *s,
         reply->error = EIO;
     } else {
         if (qiov && reply->error == 0) {
-            ret = nbd_rwv(s->ioc, qiov->iov, qiov->niov, request->len, true,
-                          NULL);
-            if (ret != request->len) {
+            assert(request->len == iov_size(qiov->iov, qiov->niov));
+            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
+                                      NULL) < 0) {
                 reply->error = EIO;
                 s->quit = true;
             }

nbd/common.c +0 -45

index e288d1b972..59a5316be9 100644
@@ -20,51 +20,6 @@
 #include "qapi/error.h"
 #include "nbd-internal.h"
 
-/* nbd_wr_syncv
- * The function may be called from coroutine or from non-coroutine context.
- * When called from non-coroutine context @ioc must be in blocking mode.
- */
-ssize_t nbd_rwv(QIOChannel *ioc, struct iovec *iov, size_t niov, size_t length,
-                bool do_read, Error **errp)
-{
-    ssize_t done = 0;
-    struct iovec *local_iov = g_new(struct iovec, niov);
-    struct iovec *local_iov_head = local_iov;
-    unsigned int nlocal_iov = niov;
-
-    nlocal_iov = iov_copy(local_iov, nlocal_iov, iov, niov, 0, length);
-
-    while (nlocal_iov > 0) {
-        ssize_t len;
-        if (do_read) {
-            len = qio_channel_readv(ioc, local_iov, nlocal_iov, errp);
-        } else {
-            len = qio_channel_writev(ioc, local_iov, nlocal_iov, errp);
-        }
-        if (len == QIO_CHANNEL_ERR_BLOCK) {
-            /* errp should not be set */
-            assert(qemu_in_coroutine());
-            qio_channel_yield(ioc, do_read ? G_IO_IN : G_IO_OUT);
-            continue;
-        }
-        if (len < 0) {
-            done = -EIO;
-            goto cleanup;
-        }
-
-        if (do_read && len == 0) {
-            break;
-        }
-
-        iov_discard_front(&local_iov, &nlocal_iov, len);
-        done += len;
-    }
-
- cleanup:
-    g_free(local_iov_head);
-    return done;
-}
-
 /* Discard length bytes from channel.  Return -errno on failure and 0 on
  * success */
 int nbd_drop(QIOChannel *ioc, size_t size, Error **errp)

tests/qemu-iotests/083.out +2 -3

index fb71b6f8ad..25dde519e3 100644
@@ -69,12 +69,12 @@ read failed: Input/output error
 
 === Check disconnect 4 reply ===
 
-End of file
+Unexpected end-of-file before all bytes were read
 read failed: Input/output error
 
 === Check disconnect 8 reply ===
 
-End of file
+Unexpected end-of-file before all bytes were read
 read failed: Input/output error
 
 === Check disconnect before data ===

@@ -180,12 +180,12 @@ read failed: Input/output error
 
 === Check disconnect 4 reply ===
 
-End of file

Now that Daniel just added convenience qio channel looping functions, we might as well use it in NBD instead of open-coding our own. In doing so, fix a couple of short-falls noticed in the qio code.

If Dan is happy with this, I can take the series through the NBD queue.

[The diffstat is slightly misleading - it's a net reduction in lines of code, but the added comments hide that fact]

Eric Blake (3):
io: Yield rather than wait when already in coroutine
io: Add new qio_channel_read{,v}_all_eof functions
nbd: Use new qio_channel_*_all() functions