[Qemu-devel] [PATCH 0/3] nbd: Use common read/write-all qio functions

[Qemu-devel] [PATCH 0/3] nbd: Use common read/write-all qio functions

[Eric Blake]

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

 include/block/nbd.h        |  2 --
 include/io/channel.h       | 53 ++++++++++++++++++++++++++++++++++++++++
 nbd/nbd-internal.h         | 41 ++++++-------------------------
 block/nbd-client.c         | 15 ++++++------
 io/channel.c               | 60 ++++++++++++++++++++++++++++++++++++++--------
 nbd/common.c               | 45 ----------------------------------
 tests/qemu-iotests/083.out |  8 +++----
 7 files changed, 121 insertions(+), 103 deletions(-)

-- 
2.13.5

[Qemu-devel] [PATCH 1/3] io: Yield rather than wait when already in coroutine

[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
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 <eblake@redhat.com>
---
 io/channel.c | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/io/channel.c b/io/channel.c
index 5e8c2f0a91..9e62794cab 100644
--- a/io/channel.c
+++ b/io/channel.c
@@ -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);
         if (len == QIO_CHANNEL_ERR_BLOCK) {
-            qio_channel_wait(ioc, G_IO_OUT);
+            if (qemu_in_coroutine()) {
+                qio_channel_yield(ioc, G_IO_OUT);
+            } else {
+                qio_channel_wait(ioc, G_IO_OUT);
+            }
             continue;
         }
         if (len < 0) {
-- 
2.13.5

[Qemu-devel] [PATCH 2/3] io: Add new qio_channel_read{, v}_all_eof functions

[Eric Blake]

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 <eblake@redhat.com>
---
 include/io/channel.h | 53 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 io/channel.c         | 48 +++++++++++++++++++++++++++++++++++++++--------
 2 files changed, 93 insertions(+), 8 deletions(-)

diff --git a/include/io/channel.h b/include/io/channel.h
index 8f25893c45..3995e243a3 100644
--- a/include/io/channel.h
+++ b/include/io/channel.h
@@ -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
diff --git a/io/channel.c b/io/channel.c
index 9e62794cab..ec4b86de7c 100644
--- a/io/channel.c
+++ b/io/channel.c
@@ -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,
+                             Error **errp)
+{
+    struct iovec iov = { .iov_base = buf, .iov_len = buflen };
+    return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
+}
+
+
 int qio_channel_read_all(QIOChannel *ioc,
                          char *buf,
                          size_t buflen,
-- 
2.13.5

[Qemu-devel] [PATCH 3/3] nbd: Use new qio_channel_*_all() functions

[Eric Blake]

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 <eblake@redhat.com>
---
 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(-)

diff --git a/include/block/nbd.h b/include/block/nbd.h
index 040cdd2e60..707fd37575 100644
--- a/include/block/nbd.h
+++ b/include/block/nbd.h
@@ -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,
diff --git a/nbd/nbd-internal.h b/nbd/nbd-internal.h
index 03549e3f39..8a609a227f 100644
--- a/nbd/nbd-internal.h
+++ b/nbd/nbd-internal.h
@@ -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 {
diff --git a/block/nbd-client.c b/block/nbd-client.c
index f0dbea24d3..ee7f758e68 100644
--- a/block/nbd-client.c
+++ b/block/nbd-client.c
@@ -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;
             }
diff --git a/nbd/common.c b/nbd/common.c
index e288d1b972..59a5316be9 100644
--- a/nbd/common.c
+++ b/nbd/common.c
@@ -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)
diff --git a/tests/qemu-iotests/083.out b/tests/qemu-iotests/083.out
index fb71b6f8ad..25dde519e3 100644
--- a/tests/qemu-iotests/083.out
+++ b/tests/qemu-iotests/083.out
@@ -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
+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 ===
-- 
2.13.5

Re: [Qemu-devel] [PATCH 1/3] io: Yield rather than wait when already in coroutine

[Daniel P. Berrange]

On Tue, Sep 05, 2017 at 02:11:12PM -0500, Eric Blake wrote:
> 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...

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

> 
> Signed-off-by: Eric Blake <eblake@redhat.com>
> ---
>  io/channel.c | 12 ++++++++++--
>  1 file changed, 10 insertions(+), 2 deletions(-)
> 
> diff --git a/io/channel.c b/io/channel.c
> index 5e8c2f0a91..9e62794cab 100644
> --- a/io/channel.c
> +++ b/io/channel.c
> @@ -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);
>          if (len == QIO_CHANNEL_ERR_BLOCK) {
> -            qio_channel_wait(ioc, G_IO_OUT);
> +            if (qemu_in_coroutine()) {
> +                qio_channel_yield(ioc, G_IO_OUT);
> +            } else {
> +                qio_channel_wait(ioc, G_IO_OUT);
> +            }
>              continue;
>          }
>          if (len < 0) {

Acked-by: Daniel P. Berrange <berrange@redhat.com>

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|

Re: [Qemu-devel] [PATCH 2/3] io: Add new qio_channel_read{, v}_all_eof functions

[Daniel P. Berrange]

On Tue, Sep 05, 2017 at 02:11:13PM -0500, Eric Blake wrote:
> 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 <eblake@redhat.com>
> ---
>  include/io/channel.h | 53 ++++++++++++++++++++++++++++++++++++++++++++++++++++
>  io/channel.c         | 48 +++++++++++++++++++++++++++++++++++++++--------
>  2 files changed, 93 insertions(+), 8 deletions(-)
> 
> diff --git a/include/io/channel.h b/include/io/channel.h
> index 8f25893c45..3995e243a3 100644
> --- a/include/io/channel.h
> +++ b/include/io/channel.h
> @@ -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
> diff --git a/io/channel.c b/io/channel.c
> index 9e62794cab..ec4b86de7c 100644
> --- a/io/channel.c
> +++ b/io/channel.c
> @@ -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,
> +                             Error **errp)
> +{
> +    struct iovec iov = { .iov_base = buf, .iov_len = buflen };
> +    return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
> +}
> +
> +
>  int qio_channel_read_all(QIOChannel *ioc,
>                           char *buf,
>                           size_t buflen,

Acked-by: Daniel P. Berrange <berrange@redhat.com>

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|

Re: [Qemu-devel] [PATCH 3/3] nbd: Use new qio_channel_*_all() functions

[Daniel P. Berrange]

On Tue, Sep 05, 2017 at 02:11:14PM -0500, Eric Blake wrote:
> 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 <eblake@redhat.com>
> ---
>  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(-)
> 
> diff --git a/include/block/nbd.h b/include/block/nbd.h
> index 040cdd2e60..707fd37575 100644
> --- a/include/block/nbd.h
> +++ b/include/block/nbd.h
> @@ -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,
> diff --git a/nbd/nbd-internal.h b/nbd/nbd-internal.h
> index 03549e3f39..8a609a227f 100644
> --- a/nbd/nbd-internal.h
> +++ b/nbd/nbd-internal.h
> @@ -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 {
> diff --git a/block/nbd-client.c b/block/nbd-client.c
> index f0dbea24d3..ee7f758e68 100644
> --- a/block/nbd-client.c
> +++ b/block/nbd-client.c
> @@ -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;
>              }
> diff --git a/nbd/common.c b/nbd/common.c
> index e288d1b972..59a5316be9 100644
> --- a/nbd/common.c
> +++ b/nbd/common.c
> @@ -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)
> diff --git a/tests/qemu-iotests/083.out b/tests/qemu-iotests/083.out
> index fb71b6f8ad..25dde519e3 100644
> --- a/tests/qemu-iotests/083.out
> +++ b/tests/qemu-iotests/083.out
> @@ -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
> +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 ===

Reviewed-by: Daniel P. Berrange <berrange@redhat.com>

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|

Re: [Qemu-devel] [PATCH 1/3] io: Yield rather than wait when already in coroutine

[Paolo Bonzini]

Il 06 set 2017 10:51 AM, "Daniel P. Berrange" <berrange@redhat.com> ha
scritto:

On Tue, Sep 05, 2017 at 02:11:12PM -0500, Eric Blake wrote:
> 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...

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


Great, this works for me on the pr-helper too.

Paolo


>
> Signed-off-by: Eric Blake <eblake@redhat.com>
> ---
>  io/channel.c | 12 ++++++++++--
>  1 file changed, 10 insertions(+), 2 deletions(-)
>
> diff --git a/io/channel.c b/io/channel.c
> index 5e8c2f0a91..9e62794cab 100644
> --- a/io/channel.c
> +++ b/io/channel.c
> @@ -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);
>          if (len == QIO_CHANNEL_ERR_BLOCK) {
> -            qio_channel_wait(ioc, G_IO_OUT);
> +            if (qemu_in_coroutine()) {
> +                qio_channel_yield(ioc, G_IO_OUT);
> +            } else {
> +                qio_channel_wait(ioc, G_IO_OUT);
> +            }
>              continue;
>          }
>          if (len < 0) {

Acked-by: Daniel P. Berrange <berrange@redhat.com>

Regards,
Daniel
--
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange
:|
|: https://libvirt.org         -o-            https://fstop138.berrange.com
:|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange
:|

Re: [Qemu-devel] [PATCH 0/3] nbd: Use common read/write-all qio functions

[Eric Blake]

[-- Attachment #1: Type: text/plain, Size: 908 bytes --]

On 09/05/2017 02:11 PM, Eric Blake wrote:
> 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

Thanks; queued on my NBD tree (after amending the commit message of 1/3,
per Dan's review):
git://repo.or.cz/qemu/ericb.git nbd

-- 
Eric Blake, Principal Software Engineer
Red Hat, Inc.           +1-919-301-3266
Virtualization:  qemu.org | libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 619 bytes --]

Re: [Qemu-devel] [PATCH 1/3] io: Yield rather than wait when already in coroutine

[Eric Blake]

[-- Attachment #1: Type: text/plain, Size: 1579 bytes --]

On 09/06/2017 03:50 AM, Daniel P. Berrange wrote:
> On Tue, Sep 05, 2017 at 02:11:12PM -0500, Eric Blake wrote:
>> 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...
> 
>> 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.

Okay, tweaking the commit message:

    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 nested event loop under the hood (so it is that secondary loop
    which 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 nested event loop.

> 
> Acked-by: Daniel P. Berrange <berrange@redhat.com>
> 
> Regards,
> Daniel
> 

-- 
Eric Blake, Principal Software Engineer
Red Hat, Inc.           +1-919-301-3266
Virtualization:  qemu.org | libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 619 bytes --]