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

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

[Fam Zheng]

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: Interface tweak of nbd_client_new
  nbd: Coroutine based nbd_send_negotiate

 blockdev-nbd.c      |  5 ++-
 include/block/nbd.h |  6 ++--
 nbd.c               | 87 ++++++++++++++++++++++++++++++++++++++++-------------
 qemu-nbd.c          | 16 +++++-----
 4 files changed, 81 insertions(+), 33 deletions(-)

-- 
2.4.3

[Qemu-devel] [PATCH 1/2] nbd: Interface tweak of nbd_client_new

[Fam Zheng]

In preparation for an async implementation, introduce a callback and
move the shutdown/close to the function.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 blockdev-nbd.c      |  5 ++---
 include/block/nbd.h |  6 ++++--
 nbd.c               | 20 +++++++++++++++-----
 qemu-nbd.c          | 16 +++++++++-------
 4 files changed, 30 insertions(+), 17 deletions(-)

diff --git a/blockdev-nbd.c b/blockdev-nbd.c
index bcdd18b..f708e0f 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, NULL);
     }
 }
 
diff --git a/include/block/nbd.h b/include/block/nbd.h
index 65f409d..11194e0 100644
--- a/include/block/nbd.h
+++ b/include/block/nbd.h
@@ -98,8 +98,10 @@ 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 *));
+typedef void (*NBDClientNewCB)(NBDExport *exp, int csock, int ret);
+void nbd_client_new(NBDExport *exp, int csock,
+                    void (*close_fn)(NBDClient *),
+                    NBDClientNewCB cb);
 void nbd_client_get(NBDClient *client);
 void nbd_client_put(NBDClient *client);
 
diff --git a/nbd.c b/nbd.c
index b3d9654..bcb79d4 100644
--- a/nbd.c
+++ b/nbd.c
@@ -1475,9 +1475,13 @@ static void nbd_update_can_read(NBDClient *client)
     }
 }
 
-NBDClient *nbd_client_new(NBDExport *exp, int csock,
-                          void (*close)(NBDClient *))
+/* Create and initialize a new client. If it fails, @csock is closed.
+ * cb will be called with the status code when done. */
+void nbd_client_new(NBDExport *exp, int csock,
+                    void (*close_fn)(NBDClient *),
+                    NBDClientNewCB cb)
 {
+    int ret = 0;
     NBDClient *client;
     client = g_malloc0(sizeof(NBDClient));
     client->refcount = 1;
@@ -1485,10 +1489,13 @@ NBDClient *nbd_client_new(NBDExport *exp, int csock,
     client->sock = csock;
     client->can_read = true;
     if (nbd_send_negotiate(client)) {
+        shutdown(csock, 2);
+        close(csock);
         g_free(client);
-        return NULL;
+        ret = -1;
+        goto out;
     }
-    client->close = close;
+    client->close = close_fn;
     qemu_co_mutex_init(&client->send_lock);
     nbd_set_handlers(client);
 
@@ -1496,5 +1503,8 @@ NBDClient *nbd_client_new(NBDExport *exp, int csock,
         QTAILQ_INSERT_TAIL(&exp->clients, client, next);
         nbd_export_get(exp);
     }
-    return client;
+out:
+    if (cb) {
+        cb(exp, csock, ret);
+    }
 }
diff --git a/qemu-nbd.c b/qemu-nbd.c
index 65dc30c..bdec228 100644
--- a/qemu-nbd.c
+++ b/qemu-nbd.c
@@ -319,6 +319,14 @@ static void nbd_client_closed(NBDClient *client)
     nbd_client_put(client);
 }
 
+static void nbd_client_new_cb(NBDExport *exp, int fd, int ret)
+{
+    if (ret == 0) {
+        nb_fds++;
+        nbd_update_server_fd_handler(server_fd);
+    }
+}
+
 static void nbd_accept(void *opaque)
 {
     struct sockaddr_in addr;
@@ -335,13 +343,7 @@ 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);
-    }
+    nbd_client_new(exp, fd, nbd_client_closed, nbd_client_new_cb);
 }
 
 static void nbd_update_server_fd_handler(int fd)
-- 
2.4.3

Re: [Qemu-devel] [PATCH 1/2] nbd: Interface tweak of nbd_client_new

[Daniel P. Berrange]

On Wed, Dec 30, 2015 at 01:49:25PM +0800, Fam Zheng wrote:
> In preparation for an async implementation, introduce a callback and
> move the shutdown/close to the function.
> 
> Signed-off-by: Fam Zheng <famz@redhat.com>
> ---
>  blockdev-nbd.c      |  5 ++---
>  include/block/nbd.h |  6 ++++--
>  nbd.c               | 20 +++++++++++++++-----
>  qemu-nbd.c          | 16 +++++++++-------
>  4 files changed, 30 insertions(+), 17 deletions(-)
> 
> diff --git a/blockdev-nbd.c b/blockdev-nbd.c
> index bcdd18b..f708e0f 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, NULL);
>      }
>  }
>  
> diff --git a/include/block/nbd.h b/include/block/nbd.h
> index 65f409d..11194e0 100644
> --- a/include/block/nbd.h
> +++ b/include/block/nbd.h
> @@ -98,8 +98,10 @@ 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 *));
> +typedef void (*NBDClientNewCB)(NBDExport *exp, int csock, int ret);
> +void nbd_client_new(NBDExport *exp, int csock,
> +                    void (*close_fn)(NBDClient *),
> +                    NBDClientNewCB cb);
>  void nbd_client_get(NBDClient *client);
>  void nbd_client_put(NBDClient *client);
>  
> diff --git a/nbd.c b/nbd.c
> index b3d9654..bcb79d4 100644
> --- a/nbd.c
> +++ b/nbd.c
> @@ -1475,9 +1475,13 @@ static void nbd_update_can_read(NBDClient *client)
>      }
>  }
>  
> -NBDClient *nbd_client_new(NBDExport *exp, int csock,
> -                          void (*close)(NBDClient *))
> +/* Create and initialize a new client. If it fails, @csock is closed.
> + * cb will be called with the status code when done. */
> +void nbd_client_new(NBDExport *exp, int csock,
> +                    void (*close_fn)(NBDClient *),
> +                    NBDClientNewCB cb)
>  {
> +    int ret = 0;
>      NBDClient *client;
>      client = g_malloc0(sizeof(NBDClient));
>      client->refcount = 1;
> @@ -1485,10 +1489,13 @@ NBDClient *nbd_client_new(NBDExport *exp, int csock,
>      client->sock = csock;
>      client->can_read = true;
>      if (nbd_send_negotiate(client)) {
> +        shutdown(csock, 2);
> +        close(csock);
>          g_free(client);
> -        return NULL;
> +        ret = -1;
> +        goto out;

If you simply make this failure code branch call close_fn() then I
think you can adding needing the new NBDClientNewCB entirely if....

> diff --git a/qemu-nbd.c b/qemu-nbd.c
> index 65dc30c..bdec228 100644
> --- a/qemu-nbd.c
> +++ b/qemu-nbd.c
> @@ -319,6 +319,14 @@ static void nbd_client_closed(NBDClient *client)
>      nbd_client_put(client);
>  }
>  
> +static void nbd_client_new_cb(NBDExport *exp, int fd, int ret)
> +{
> +    if (ret == 0) {
> +        nb_fds++;
> +        nbd_update_server_fd_handler(server_fd);
> +    }
> +}
> +
>  static void nbd_accept(void *opaque)
>  {
>      struct sockaddr_in addr;
> @@ -335,13 +343,7 @@ 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);
> -    }
> +    nbd_client_new(exp, fd, nbd_client_closed, nbd_client_new_cb);

...you make this do

    nb_fds++;
    nbd_update_server_fd_handler(server_fd);
    nbd_client_new(exp, fd, nbd_client_closed, nbd_client_new_cb);

ie, you once guarantee that *every* invocation of nbd_client_new()
will eventually lead to a call to 'nbd_client_closed', you can
unconditionally increment nb_fds before calling nbd_client_new.


This has the added benefit in that the 'nb_fds' count now takes
account of client connections that are in the negotiate phase,
whereas your approach allows for an unlimited number of clients
to be in the negotiate phase, only limiting them post-negotiate

Regards,
Daniel
-- 
|: http://berrange.com      -o-    http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org              -o-             http://virt-manager.org :|
|: http://autobuild.org       -o-         http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org       -o-       http://live.gnome.org/gtk-vnc :|

Re: [Qemu-devel] [PATCH 1/2] nbd: Interface tweak of nbd_client_new

[Paolo Bonzini]

On 08/01/2016 17:24, Daniel P. Berrange wrote:
>> >      if (nbd_send_negotiate(client)) {
>> > +        shutdown(csock, 2);
>> > +        close(csock);
>> >          g_free(client);
>> > -        return NULL;
>> > +        ret = -1;
>> > +        goto out;
> If you simply make this failure code branch call close_fn() then I
> think you can adding needing the new NBDClientNewCB entirely if....

Good idea, but note that close_fn will call nbd_client_put, so the
close/g_free must be removed.  It's probably cleanest to change csock to
client->sock in the shutdown call, too.

Paolo

Re: [Qemu-devel] [PATCH 1/2] nbd: Interface tweak of nbd_client_new

[Fam Zheng]

On Fri, 01/08 19:24, Paolo Bonzini wrote:
> 
> 
> On 08/01/2016 17:24, Daniel P. Berrange wrote:
> >> >      if (nbd_send_negotiate(client)) {
> >> > +        shutdown(csock, 2);
> >> > +        close(csock);
> >> >          g_free(client);
> >> > -        return NULL;
> >> > +        ret = -1;
> >> > +        goto out;
> > If you simply make this failure code branch call close_fn() then I
> > think you can adding needing the new NBDClientNewCB entirely if....
> 
> Good idea, but note that close_fn will call nbd_client_put, so the
> close/g_free must be removed.  It's probably cleanest to change csock to
> client->sock in the shutdown call, too.

Good suggestions, will update and resend. Thanks!

Fam

[Qemu-devel] [PATCH 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 | 85 +++++++++++++++++++++++++++++++++++++++++++++++--------------------
 1 file changed, 60 insertions(+), 25 deletions(-)

diff --git a/nbd.c b/nbd.c
index bcb79d4..4a75d2d 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,26 @@ static int nbd_receive_options(NBDClient *client)
     }
 }
 
-static int nbd_send_negotiate(NBDClient *client)
+typedef struct {
+    NBDClient *client;
+    NBDClientNewCB cb;
+    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 +543,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 +589,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,36 +1490,56 @@ static void nbd_update_can_read(NBDClient *client)
     }
 }
 
+static coroutine_fn void nbd_co_client_start(void *opaque)
+{
+    int ret = 0;
+    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);
+        close(client->sock);
+        g_free(client);
+        ret = -1;
+        nbd_export_put(exp);
+        goto out;
+    }
+    qemu_co_mutex_init(&client->send_lock);
+    nbd_set_handlers(client);
+
+    if (exp) {
+        QTAILQ_INSERT_TAIL(&exp->clients, client, next);
+    }
+out:
+    if (data->cb) {
+        data->cb(exp, client->sock, ret);
+    }
+    g_free(data);
+}
+
 /* Create and initialize a new client. If it fails, @csock is closed.
  * cb will be called with the status code when done. */
 void nbd_client_new(NBDExport *exp, int csock,
                     void (*close_fn)(NBDClient *),
                     NBDClientNewCB cb)
 {
-    int ret = 0;
     NBDClient *client;
+    Coroutine *co;
+    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(csock, 2);
-        close(csock);
-        g_free(client);
-        ret = -1;
-        goto out;
-    }
     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);
-    }
-out:
-    if (cb) {
-        cb(exp, csock, ret);
-    }
+    data->client = client;
+    data->cb = cb;
+    co = qemu_coroutine_create(nbd_co_client_start);
+    qemu_coroutine_enter(co, data);
 }
-- 
2.4.3