[Qemu-devel] [PATCH v2 0/2] nbd: Async built-in server negotiation

[Qemu-devel] [PATCH v2 0/2] nbd: Async built-in server negotiation

[Fam Zheng]

v2: Adopt Daniel and Paolo's idea: always call close_fn.

During nbd_send_negotiate, if the client simply doesn't respond, the function
will not return, and the whole event loop is blocked.

Make the I/O effectively asynchronous by using coroutine read/write, so that a
malicious or disappeared client cannot make a hang.

Fam


Fam Zheng (2):
  nbd: Always call "close_fn" in nbd_client_new
  nbd: Coroutine based nbd_send_negotiate

 blockdev-nbd.c      |  5 ++--
 include/block/nbd.h |  3 +--
 nbd.c               | 73 ++++++++++++++++++++++++++++++++++++++---------------
 qemu-nbd.c          | 10 +++-----
 4 files changed, 58 insertions(+), 33 deletions(-)

-- 
2.4.3

[Qemu-devel] [PATCH v2 1/2] nbd: Always call "close_fn" in nbd_client_new

[Fam Zheng]

Rename the parameter "close" to "close_fn" to disambiguous with
close(2).

This unifies error handling paths of NBDClient allocation:
nbd_client_new will shutdown the socket and call the "close_fn" callback
if negotiation failed, so the caller don't need a different path than
the normal close.

The returned pointer is never used, make it void in preparation for the
next patch.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 blockdev-nbd.c      |  5 ++---
 include/block/nbd.h |  3 +--
 nbd.c               | 11 +++++------
 qemu-nbd.c          | 10 +++-------
 4 files changed, 11 insertions(+), 18 deletions(-)

diff --git a/blockdev-nbd.c b/blockdev-nbd.c
index bcdd18b..4a758ac 100644
--- a/blockdev-nbd.c
+++ b/blockdev-nbd.c
@@ -27,9 +27,8 @@ static void nbd_accept(void *opaque)
     socklen_t addr_len = sizeof(addr);
 
     int fd = accept(server_fd, (struct sockaddr *)&addr, &addr_len);
-    if (fd >= 0 && !nbd_client_new(NULL, fd, nbd_client_put)) {
-        shutdown(fd, 2);
-        close(fd);
+    if (fd >= 0) {
+        nbd_client_new(NULL, fd, nbd_client_put);
     }
 }
 
diff --git a/include/block/nbd.h b/include/block/nbd.h
index 65f409d..7eccb41 100644
--- a/include/block/nbd.h
+++ b/include/block/nbd.h
@@ -98,8 +98,7 @@ NBDExport *nbd_export_find(const char *name);
 void nbd_export_set_name(NBDExport *exp, const char *name);
 void nbd_export_close_all(void);
 
-NBDClient *nbd_client_new(NBDExport *exp, int csock,
-                          void (*close)(NBDClient *));
+void nbd_client_new(NBDExport *exp, int csock, void (*close_fn)(NBDClient *));
 void nbd_client_get(NBDClient *client);
 void nbd_client_put(NBDClient *client);
 
diff --git a/nbd.c b/nbd.c
index b3d9654..f8d0221 100644
--- a/nbd.c
+++ b/nbd.c
@@ -1475,8 +1475,7 @@ static void nbd_update_can_read(NBDClient *client)
     }
 }
 
-NBDClient *nbd_client_new(NBDExport *exp, int csock,
-                          void (*close)(NBDClient *))
+void nbd_client_new(NBDExport *exp, int csock, void (*close_fn)(NBDClient *))
 {
     NBDClient *client;
     client = g_malloc0(sizeof(NBDClient));
@@ -1485,10 +1484,11 @@ NBDClient *nbd_client_new(NBDExport *exp, int csock,
     client->sock = csock;
     client->can_read = true;
     if (nbd_send_negotiate(client)) {
-        g_free(client);
-        return NULL;
+        shutdown(client->sock, 2);
+        close_fn(client);
+        return;
     }
-    client->close = close;
+    client->close = close_fn;
     qemu_co_mutex_init(&client->send_lock);
     nbd_set_handlers(client);
 
@@ -1496,5 +1496,4 @@ NBDClient *nbd_client_new(NBDExport *exp, int csock,
         QTAILQ_INSERT_TAIL(&exp->clients, client, next);
         nbd_export_get(exp);
     }
-    return client;
 }
diff --git a/qemu-nbd.c b/qemu-nbd.c
index 65dc30c..31c58eb 100644
--- a/qemu-nbd.c
+++ b/qemu-nbd.c
@@ -335,13 +335,9 @@ static void nbd_accept(void *opaque)
         return;
     }
 
-    if (nbd_client_new(exp, fd, nbd_client_closed)) {
-        nb_fds++;
-        nbd_update_server_fd_handler(server_fd);
-    } else {
-        shutdown(fd, 2);
-        close(fd);
-    }
+    nb_fds++;
+    nbd_update_server_fd_handler(server_fd);
+    nbd_client_new(exp, fd, nbd_client_closed);
 }
 
 static void nbd_update_server_fd_handler(int fd)
-- 
2.4.3

[Qemu-devel] [PATCH v2 2/2] nbd: Coroutine based nbd_send_negotiate

[Fam Zheng]

Create a coroutine in nbd_client_new, so that nbd_send_negotiate doesn't
need qemu_set_block().

A handler is needed for csock fd in case the coroutine yields during
I/O.

With this, if the other end disappears in the middle of the negotiation,
we don't block the whole event loop.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 nbd.c | 68 +++++++++++++++++++++++++++++++++++++++++++++++++------------------
 1 file changed, 50 insertions(+), 18 deletions(-)

diff --git a/nbd.c b/nbd.c
index f8d0221..8f288a0 100644
--- a/nbd.c
+++ b/nbd.c
@@ -238,10 +238,9 @@ ssize_t nbd_wr_sync(int fd, void *buffer, size_t size, bool do_read)
 
 static ssize_t read_sync(int fd, void *buffer, size_t size)
 {
-    /* 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.
+    /* 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.
      */
     return nbd_wr_sync(fd, buffer, size, true);
 }
@@ -504,13 +503,25 @@ static int nbd_receive_options(NBDClient *client)
     }
 }
 
-static int nbd_send_negotiate(NBDClient *client)
+typedef struct {
+    NBDClient *client;
+    Coroutine *co;
+} NBDClientNewData;
+
+static void nbd_negotiate_continue(void *opaque)
 {
+    qemu_coroutine_enter(opaque, NULL);
+}
+
+static coroutine_fn int nbd_send_negotiate(NBDClientNewData *data)
+{
+    NBDClient *client = data->client;
     int csock = client->sock;
     char buf[8 + 8 + 8 + 128];
     int rc;
     const int myflags = (NBD_FLAG_HAS_FLAGS | NBD_FLAG_SEND_TRIM |
                          NBD_FLAG_SEND_FLUSH | NBD_FLAG_SEND_FUA);
+    AioContext *ctx = client->exp ? client->exp->ctx : qemu_get_aio_context();
 
     /* Negotiation header without options:
         [ 0 ..   7]   passwd       ("NBDMAGIC")
@@ -531,9 +542,11 @@ static int nbd_send_negotiate(NBDClient *client)
         [28 .. 151]   reserved     (0)
      */
 
-    qemu_set_block(csock);
     rc = -EINVAL;
 
+    aio_set_fd_handler(ctx, client->sock, true,
+                       nbd_negotiate_continue,
+                       nbd_negotiate_continue, data->co);
     TRACE("Beginning negotiation.");
     memset(buf, 0, sizeof(buf));
     memcpy(buf, "NBDMAGIC", 8);
@@ -575,7 +588,8 @@ static int nbd_send_negotiate(NBDClient *client)
     TRACE("Negotiation succeeded.");
     rc = 0;
 fail:
-    qemu_set_nonblock(csock);
+    aio_set_fd_handler(ctx, client->sock, true,
+                       NULL, NULL, NULL);
     return rc;
 }
 
@@ -1475,25 +1489,43 @@ static void nbd_update_can_read(NBDClient *client)
     }
 }
 
+static coroutine_fn void nbd_co_client_start(void *opaque)
+{
+    NBDClientNewData *data = opaque;
+    NBDClient *client = data->client;
+    NBDExport *exp = client->exp;
+
+    if (exp) {
+        nbd_export_get(exp);
+    }
+    if (nbd_send_negotiate(data)) {
+        shutdown(client->sock, 2);
+        client->close(client);
+        goto out;
+    }
+    qemu_co_mutex_init(&client->send_lock);
+    nbd_set_handlers(client);
+
+    if (exp) {
+        QTAILQ_INSERT_TAIL(&exp->clients, client, next);
+    }
+out:
+    g_free(data);
+}
+
 void nbd_client_new(NBDExport *exp, int csock, void (*close_fn)(NBDClient *))
 {
     NBDClient *client;
+    NBDClientNewData *data = g_new(NBDClientNewData, 1);
+
     client = g_malloc0(sizeof(NBDClient));
     client->refcount = 1;
     client->exp = exp;
     client->sock = csock;
     client->can_read = true;
-    if (nbd_send_negotiate(client)) {
-        shutdown(client->sock, 2);
-        close_fn(client);
-        return;
-    }
     client->close = close_fn;
-    qemu_co_mutex_init(&client->send_lock);
-    nbd_set_handlers(client);
 
-    if (exp) {
-        QTAILQ_INSERT_TAIL(&exp->clients, client, next);
-        nbd_export_get(exp);
-    }
+    data->client = client;
+    data->co = qemu_coroutine_create(nbd_co_client_start);
+    qemu_coroutine_enter(data->co, data);
 }
-- 
2.4.3

Re: [Qemu-devel] [PATCH v2 2/2] nbd: Coroutine based nbd_send_negotiate

[Paolo Bonzini]

On 11/01/2016 04:36, Fam Zheng wrote:
>  
> +    aio_set_fd_handler(ctx, client->sock, true,
> +                       nbd_negotiate_continue,
> +                       nbd_negotiate_continue, data->co);
>      TRACE("Beginning negotiation.");
>      memset(buf, 0, sizeof(buf));

This causes a busy loop if the socket is writable but the client does
not send data.  I think you need to set/clear the handler (using
qemu_coroutine_self() instead of data->co, probably) every time the
direction of negotiation switches.  That is, set only a read handler
before read_sync, and only a write handler before write_sync.

Paolo

Re: [Qemu-devel] [PATCH v2 2/2] nbd: Coroutine based nbd_send_negotiate

[Fam Zheng]

On Mon, 01/11 15:00, Paolo Bonzini wrote:
> 
> 
> On 11/01/2016 04:36, Fam Zheng wrote:
> >  
> > +    aio_set_fd_handler(ctx, client->sock, true,
> > +                       nbd_negotiate_continue,
> > +                       nbd_negotiate_continue, data->co);
> >      TRACE("Beginning negotiation.");
> >      memset(buf, 0, sizeof(buf));
> 
> This causes a busy loop if the socket is writable but the client does
> not send data.  I think you need to set/clear the handler (using
> qemu_coroutine_self() instead of data->co, probably) every time the
> direction of negotiation switches.  That is, set only a read handler
> before read_sync, and only a write handler before write_sync.
> 

You are right, I'll take a look and work on v3.

Fam