[Qemu-devel] [PATCH v2 0/3] nbd: Adapt for dataplane

[Qemu-devel] [PATCH v2 0/3] nbd: Adapt for dataplane

[Max Reitz]

For the NBD server to work with dataplane, it needs to correctly access
the exported BDS. It makes the most sense to run both in the same
AioContext, therefore this series implements methods for tracking a
BDS's AioContext and makes NBD make use of this for keeping the clients
connected to that BDS in the same AioContext.


v2:
 - Patch 1: Drop NBDClient::restart_write; checking whether
   NBDClient::send_coroutine is not NULL suffices [Paolo]


git-backport-diff against v1:

Key:
[----] : patches are identical
[####] : number of functional differences between upstream/downstream patch
[down] : patch is downstream-only
The flags [FC] indicate (F)unctional and (C)ontextual differences, respectively

001/3:[0005] [FC] 'nbd: Drop nbd_can_read()'
002/3:[----] [--] 'block: Add AIO context notifiers'
003/3:[----] [--] 'nbd: Follow the BDS' AIO context'


Max Reitz (3):
  nbd: Drop nbd_can_read()
  block: Add AIO context notifiers
  nbd: Follow the BDS' AIO context

 block.c                   |  56 +++++++++++++++++++++++++
 include/block/block_int.h |  41 ++++++++++++++++++
 nbd.c                     | 105 +++++++++++++++++++++++++++++++++++++---------
 3 files changed, 183 insertions(+), 19 deletions(-)

-- 
2.0.0

[Qemu-devel] [PATCH v2 1/3] nbd: Drop nbd_can_read()

[Max Reitz]

There is no variant of aio_set_fd_handler() like qemu_set_fd_handler2(),
so we cannot give a can_read() callback function. Instead, unregister
the nbd_read() function whenever we cannot read and re-register it as
soon as we can read again.

All this is hidden behind the functions nbd_set_handlers() (which
registers all handlers for the AIO context and file descriptor belonging
to the given client), nbd_unset_handlers() (which unregisters them) and
nbd_update_can_read() (which checks whether NBD can read for the given
client and acts accordingly).

Signed-off-by: Max Reitz <mreitz@redhat.com>
---
 nbd.c | 74 ++++++++++++++++++++++++++++++++++++++++++++++++++-----------------
 1 file changed, 55 insertions(+), 19 deletions(-)

diff --git a/nbd.c b/nbd.c
index e0d032c..2715acc 100644
--- a/nbd.c
+++ b/nbd.c
@@ -18,6 +18,7 @@
 
 #include "block/nbd.h"
 #include "block/block.h"
+#include "block/block_int.h"
 
 #include "block/coroutine.h"
 
@@ -100,6 +101,8 @@ struct NBDExport {
     uint32_t nbdflags;
     QTAILQ_HEAD(, NBDClient) clients;
     QTAILQ_ENTRY(NBDExport) next;
+
+    AioContext *ctx;
 };
 
 static QTAILQ_HEAD(, NBDExport) exports = QTAILQ_HEAD_INITIALIZER(exports);
@@ -116,6 +119,8 @@ struct NBDClient {
     CoMutex send_lock;
     Coroutine *send_coroutine;
 
+    bool can_read;
+
     QTAILQ_ENTRY(NBDClient) next;
     int nb_requests;
     bool closing;
@@ -123,6 +128,10 @@ struct NBDClient {
 
 /* That's all folks */
 
+static void nbd_set_handlers(NBDClient *client);
+static void nbd_unset_handlers(NBDClient *client);
+static void nbd_update_can_read(NBDClient *client);
+
 ssize_t nbd_wr_sync(int fd, void *buffer, size_t size, bool do_read)
 {
     size_t offset = 0;
@@ -744,7 +753,7 @@ void nbd_client_put(NBDClient *client)
          */
         assert(client->closing);
 
-        qemu_set_fd_handler2(client->sock, NULL, NULL, NULL, NULL);
+        nbd_unset_handlers(client);
         close(client->sock);
         client->sock = -1;
         if (client->exp) {
@@ -780,6 +789,7 @@ static NBDRequest *nbd_request_get(NBDClient *client)
 
     assert(client->nb_requests <= MAX_NBD_REQUESTS - 1);
     client->nb_requests++;
+    nbd_update_can_read(client);
 
     req = g_slice_new0(NBDRequest);
     nbd_client_get(client);
@@ -796,9 +806,8 @@ static void nbd_request_put(NBDRequest *req)
     }
     g_slice_free(NBDRequest, req);
 
-    if (client->nb_requests-- == MAX_NBD_REQUESTS) {
-        qemu_notify_event();
-    }
+    client->nb_requests--;
+    nbd_update_can_read(client);
     nbd_client_put(client);
 }
 
@@ -814,6 +823,7 @@ NBDExport *nbd_export_new(BlockDriverState *bs, off_t dev_offset,
     exp->nbdflags = nbdflags;
     exp->size = size == -1 ? bdrv_getlength(bs) : size;
     exp->close = close;
+    exp->ctx = bdrv_get_aio_context(bs);
     bdrv_ref(bs);
     return exp;
 }
@@ -905,10 +915,6 @@ void nbd_export_close_all(void)
     }
 }
 
-static int nbd_can_read(void *opaque);
-static void nbd_read(void *opaque);
-static void nbd_restart_write(void *opaque);
-
 static ssize_t nbd_co_send_reply(NBDRequest *req, struct nbd_reply *reply,
                                  int len)
 {
@@ -917,9 +923,8 @@ static ssize_t nbd_co_send_reply(NBDRequest *req, struct nbd_reply *reply,
     ssize_t rc, ret;
 
     qemu_co_mutex_lock(&client->send_lock);
-    qemu_set_fd_handler2(csock, nbd_can_read, nbd_read,
-                         nbd_restart_write, client);
     client->send_coroutine = qemu_coroutine_self();
+    nbd_set_handlers(client);
 
     if (!len) {
         rc = nbd_send_reply(csock, reply);
@@ -936,7 +941,7 @@ static ssize_t nbd_co_send_reply(NBDRequest *req, struct nbd_reply *reply,
     }
 
     client->send_coroutine = NULL;
-    qemu_set_fd_handler2(csock, nbd_can_read, nbd_read, NULL, client);
+    nbd_set_handlers(client);
     qemu_co_mutex_unlock(&client->send_lock);
     return rc;
 }
@@ -949,6 +954,8 @@ static ssize_t nbd_co_receive_request(NBDRequest *req, struct nbd_request *reque
     ssize_t rc;
 
     client->recv_coroutine = qemu_coroutine_self();
+    nbd_update_can_read(client);
+
     rc = nbd_receive_request(csock, request);
     if (rc < 0) {
         if (rc != -EAGAIN) {
@@ -990,6 +997,8 @@ static ssize_t nbd_co_receive_request(NBDRequest *req, struct nbd_request *reque
 
 out:
     client->recv_coroutine = NULL;
+    nbd_update_can_read(client);
+
     return rc;
 }
 
@@ -1140,13 +1149,6 @@ out:
     nbd_client_close(client);
 }
 
-static int nbd_can_read(void *opaque)
-{
-    NBDClient *client = opaque;
-
-    return client->recv_coroutine || client->nb_requests < MAX_NBD_REQUESTS;
-}
-
 static void nbd_read(void *opaque)
 {
     NBDClient *client = opaque;
@@ -1165,6 +1167,39 @@ static void nbd_restart_write(void *opaque)
     qemu_coroutine_enter(client->send_coroutine, NULL);
 }
 
+static void nbd_set_handlers(NBDClient *client)
+{
+    if (client->exp && client->exp->ctx) {
+        aio_set_fd_handler(client->exp->ctx, client->sock,
+                           client->can_read ? nbd_read : NULL,
+                           client->send_coroutine ? nbd_restart_write : NULL,
+                           client);
+    }
+}
+
+static void nbd_unset_handlers(NBDClient *client)
+{
+    if (client->exp && client->exp->ctx) {
+        aio_set_fd_handler(client->exp->ctx, client->sock, NULL, NULL, NULL);
+    }
+}
+
+static void nbd_update_can_read(NBDClient *client)
+{
+    bool can_read = client->recv_coroutine ||
+                    client->nb_requests < MAX_NBD_REQUESTS;
+
+    if (can_read != client->can_read) {
+        client->can_read = can_read;
+        nbd_set_handlers(client);
+
+        /* If we got here, nb_requests had to be MAX_NBD_REQUESTS before */
+        if (client->nb_requests < MAX_NBD_REQUESTS) {
+            aio_notify(client->exp->ctx);
+        }
+    }
+}
+
 NBDClient *nbd_client_new(NBDExport *exp, int csock,
                           void (*close)(NBDClient *))
 {
@@ -1173,13 +1208,14 @@ NBDClient *nbd_client_new(NBDExport *exp, int csock,
     client->refcount = 1;
     client->exp = exp;
     client->sock = csock;
+    client->can_read = true;
     if (nbd_send_negotiate(client) < 0) {
         g_free(client);
         return NULL;
     }
     client->close = close;
     qemu_co_mutex_init(&client->send_lock);
-    qemu_set_fd_handler2(csock, nbd_can_read, nbd_read, NULL, client);
+    nbd_set_handlers(client);
 
     if (exp) {
         QTAILQ_INSERT_TAIL(&exp->clients, client, next);
-- 
2.0.0

[Qemu-devel] [PATCH v2 2/3] block: Add AIO context notifiers

[Max Reitz]

If a long-running operation on a BDS wants to always remain in the same
AIO context, it somehow needs to keep track of the BDS changing its
context. This adds a function for registering callbacks on a BDS which
are called whenever the BDS is attached or detached from an AIO context.

Signed-off-by: Max Reitz <mreitz@redhat.com>
---
 block.c                   | 56 +++++++++++++++++++++++++++++++++++++++++++++++
 include/block/block_int.h | 41 ++++++++++++++++++++++++++++++++++
 2 files changed, 97 insertions(+)

diff --git a/block.c b/block.c
index 43abe96..a3d09fa 100644
--- a/block.c
+++ b/block.c
@@ -1806,6 +1806,8 @@ void bdrv_reopen_abort(BDRVReopenState *reopen_state)
 
 void bdrv_close(BlockDriverState *bs)
 {
+    BdrvAioNotifier *ban, *ban_next;
+
     if (bs->job) {
         block_job_cancel_sync(bs->job);
     }
@@ -1848,6 +1850,11 @@ void bdrv_close(BlockDriverState *bs)
     if (bs->io_limits_enabled) {
         bdrv_io_limits_disable(bs);
     }
+
+    QLIST_FOREACH_SAFE(ban, &bs->aio_notifiers, list, ban_next) {
+        g_free(ban);
+    }
+    QLIST_INIT(&bs->aio_notifiers);
 }
 
 void bdrv_close_all(void)
@@ -5663,10 +5670,16 @@ AioContext *bdrv_get_aio_context(BlockDriverState *bs)
 
 void bdrv_detach_aio_context(BlockDriverState *bs)
 {
+    BdrvAioNotifier *baf;
+
     if (!bs->drv) {
         return;
     }
 
+    QLIST_FOREACH(baf, &bs->aio_notifiers, list) {
+        baf->detach_aio_context(baf->opaque);
+    }
+
     if (bs->io_limits_enabled) {
         throttle_detach_aio_context(&bs->throttle_state);
     }
@@ -5686,6 +5699,8 @@ void bdrv_detach_aio_context(BlockDriverState *bs)
 void bdrv_attach_aio_context(BlockDriverState *bs,
                              AioContext *new_context)
 {
+    BdrvAioNotifier *ban;
+
     if (!bs->drv) {
         return;
     }
@@ -5704,6 +5719,10 @@ void bdrv_attach_aio_context(BlockDriverState *bs,
     if (bs->io_limits_enabled) {
         throttle_attach_aio_context(&bs->throttle_state, new_context);
     }
+
+    QLIST_FOREACH(ban, &bs->aio_notifiers, list) {
+        ban->attached_aio_context(new_context, ban->opaque);
+    }
 }
 
 void bdrv_set_aio_context(BlockDriverState *bs, AioContext *new_context)
@@ -5720,6 +5739,43 @@ void bdrv_set_aio_context(BlockDriverState *bs, AioContext *new_context)
     aio_context_release(new_context);
 }
 
+void bdrv_add_aio_context_notifier(BlockDriverState *bs,
+        void (*attached_aio_context)(AioContext *new_context, void *opaque),
+        void (*detach_aio_context)(void *opaque), void *opaque)
+{
+    BdrvAioNotifier *ban = g_new(BdrvAioNotifier, 1);
+    *ban = (BdrvAioNotifier){
+        .attached_aio_context = attached_aio_context,
+        .detach_aio_context   = detach_aio_context,
+        .opaque               = opaque
+    };
+
+    QLIST_INSERT_HEAD(&bs->aio_notifiers, ban, list);
+}
+
+void bdrv_remove_aio_context_notifier(BlockDriverState *bs,
+                                      void (*attached_aio_context)(AioContext *,
+                                                                   void *),
+                                      void (*detach_aio_context)(void *),
+                                      void *opaque)
+{
+    BdrvAioNotifier *ban, *ban_next;
+
+    QLIST_FOREACH_SAFE(ban, &bs->aio_notifiers, list, ban_next) {
+        if (ban->attached_aio_context == attached_aio_context &&
+            ban->detach_aio_context   == detach_aio_context   &&
+            ban->opaque               == opaque)
+        {
+            QLIST_REMOVE(ban, list);
+            g_free(ban);
+
+            return;
+        }
+    }
+
+    abort();
+}
+
 void bdrv_add_before_write_notifier(BlockDriverState *bs,
                                     NotifierWithReturn *notifier)
 {
diff --git a/include/block/block_int.h b/include/block/block_int.h
index 7aa2213..a1885d3 100644
--- a/include/block/block_int.h
+++ b/include/block/block_int.h
@@ -282,6 +282,15 @@ typedef struct BlockLimits {
 
 typedef struct BdrvOpBlocker BdrvOpBlocker;
 
+typedef struct BdrvAioNotifier {
+    void (*attached_aio_context)(AioContext *new_context, void *opaque);
+    void (*detach_aio_context)(void *opaque);
+
+    void *opaque;
+
+    QLIST_ENTRY(BdrvAioNotifier) list;
+} BdrvAioNotifier;
+
 /*
  * Note: the function bdrv_append() copies and swaps contents of
  * BlockDriverStates, so if you add new fields to this struct, please
@@ -308,6 +317,10 @@ struct BlockDriverState {
     void *dev_opaque;
 
     AioContext *aio_context; /* event loop used for fd handlers, timers, etc */
+    /* long-running tasks intended to always use the same AioContext as this
+     * BDS may register themselves in this list to be notified of changes
+     * regarding this BDS's context */
+    QLIST_HEAD(, BdrvAioNotifier) aio_notifiers;
 
     char filename[1024];
     char backing_file[1024]; /* if non zero, the image is a diff of
@@ -422,6 +435,34 @@ void bdrv_detach_aio_context(BlockDriverState *bs);
 void bdrv_attach_aio_context(BlockDriverState *bs,
                              AioContext *new_context);
 
+/**
+ * bdrv_add_aio_context_notifier:
+ *
+ * If a long-running job intends to be always run in the same AioContext as a
+ * certain BDS, it may use this function to be notified of changes regarding the
+ * association of the BDS to an AioContext.
+ *
+ * attached_aio_context() is called after the target BDS has been attached to a
+ * new AioContext; detach_aio_context() is called before the target BDS is being
+ * detached from its old AioContext.
+ */
+void bdrv_add_aio_context_notifier(BlockDriverState *bs,
+        void (*attached_aio_context)(AioContext *new_context, void *opaque),
+        void (*detach_aio_context)(void *opaque), void *opaque);
+
+/**
+ * bdrv_remove_aio_context_notifier:
+ *
+ * Unsubscribe of change notifications regarding the BDS's AioContext. The
+ * parameters given here have to be the same as those given to
+ * bdrv_add_aio_context_notifier().
+ */
+void bdrv_remove_aio_context_notifier(BlockDriverState *bs,
+                                      void (*aio_context_attached)(AioContext *,
+                                                                   void *),
+                                      void (*aio_context_detached)(void *),
+                                      void *opaque);
+
 #ifdef _WIN32
 int is_windows_drive(const char *filename);
 #endif
-- 
2.0.0

[Qemu-devel] [PATCH v2 3/3] nbd: Follow the BDS' AIO context

[Max Reitz]

Keep the NBD server always in the same AIO context as the exported BDS
by calling bdrv_add_aio_context_notifier() and implementing the required
callbacks.

Signed-off-by: Max Reitz <mreitz@redhat.com>
---
 nbd.c | 31 +++++++++++++++++++++++++++++++
 1 file changed, 31 insertions(+)

diff --git a/nbd.c b/nbd.c
index 2715acc..cd92f88 100644
--- a/nbd.c
+++ b/nbd.c
@@ -811,6 +811,34 @@ static void nbd_request_put(NBDRequest *req)
     nbd_client_put(client);
 }
 
+static void bs_aio_attached(AioContext *ctx, void *opaque)
+{
+    NBDExport *exp = opaque;
+    NBDClient *client;
+
+    TRACE("Export %s: Attaching clients to AIO context %p\n", exp->name, ctx);
+
+    exp->ctx = ctx;
+
+    QTAILQ_FOREACH(client, &exp->clients, next) {
+        nbd_set_handlers(client);
+    }
+}
+
+static void bs_aio_detach(void *opaque)
+{
+    NBDExport *exp = opaque;
+    NBDClient *client;
+
+    TRACE("Export %s: Detaching clients from AIO context %p\n", exp->name, exp->ctx);
+
+    QTAILQ_FOREACH(client, &exp->clients, next) {
+        nbd_unset_handlers(client);
+    }
+
+    exp->ctx = NULL;
+}
+
 NBDExport *nbd_export_new(BlockDriverState *bs, off_t dev_offset,
                           off_t size, uint32_t nbdflags,
                           void (*close)(NBDExport *))
@@ -825,6 +853,7 @@ NBDExport *nbd_export_new(BlockDriverState *bs, off_t dev_offset,
     exp->close = close;
     exp->ctx = bdrv_get_aio_context(bs);
     bdrv_ref(bs);
+    bdrv_add_aio_context_notifier(bs, bs_aio_attached, bs_aio_detach, exp);
     return exp;
 }
 
@@ -872,6 +901,8 @@ void nbd_export_close(NBDExport *exp)
     nbd_export_set_name(exp, NULL);
     nbd_export_put(exp);
     if (exp->bs) {
+        bdrv_remove_aio_context_notifier(exp->bs, bs_aio_attached,
+                                         bs_aio_detach, exp);
         bdrv_unref(exp->bs);
         exp->bs = NULL;
     }
-- 
2.0.0

Re: [Qemu-devel] [PATCH v2 1/3] nbd: Drop nbd_can_read()

[Stefan Hajnoczi]

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

On Wed, Jun 18, 2014 at 09:06:41PM +0200, Max Reitz wrote:
> +static void nbd_update_can_read(NBDClient *client)
> +{
> +    bool can_read = client->recv_coroutine ||
> +                    client->nb_requests < MAX_NBD_REQUESTS;
> +
> +    if (can_read != client->can_read) {
> +        client->can_read = can_read;
> +        nbd_set_handlers(client);
> +
> +        /* If we got here, nb_requests had to be MAX_NBD_REQUESTS before */
> +        if (client->nb_requests < MAX_NBD_REQUESTS) {
> +            aio_notify(client->exp->ctx);
> +        }

nbd_set_handlers() indirectly invokes aio_notify(client->exp->ctx) via
aio_set_fd_handler().  This if statement is redundant.

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH v2 2/3] block: Add AIO context notifiers

[Stefan Hajnoczi]

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

On Wed, Jun 18, 2014 at 09:06:42PM +0200, Max Reitz wrote:
> If a long-running operation on a BDS wants to always remain in the same
> AIO context, it somehow needs to keep track of the BDS changing its
> context. This adds a function for registering callbacks on a BDS which
> are called whenever the BDS is attached or detached from an AIO context.
> 
> Signed-off-by: Max Reitz <mreitz@redhat.com>
> ---
>  block.c                   | 56 +++++++++++++++++++++++++++++++++++++++++++++++
>  include/block/block_int.h | 41 ++++++++++++++++++++++++++++++++++
>  2 files changed, 97 insertions(+)

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH v2 3/3] nbd: Follow the BDS' AIO context

[Stefan Hajnoczi]

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

On Wed, Jun 18, 2014 at 09:06:43PM +0200, Max Reitz wrote:
> Keep the NBD server always in the same AIO context as the exported BDS
> by calling bdrv_add_aio_context_notifier() and implementing the required
> callbacks.
> 
> Signed-off-by: Max Reitz <mreitz@redhat.com>
> ---
>  nbd.c | 31 +++++++++++++++++++++++++++++++
>  1 file changed, 31 insertions(+)

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH v2 1/3] nbd: Drop nbd_can_read()

[Paolo Bonzini]

Il 18/06/2014 21:06, Max Reitz ha scritto:
> +static void nbd_set_handlers(NBDClient *client)
> +{
> +    if (client->exp && client->exp->ctx) {
> +        aio_set_fd_handler(client->exp->ctx, client->sock,
> +                           client->can_read ? nbd_read : NULL,
> +                           client->send_coroutine ? nbd_restart_write : NULL,
> +                           client);
> +    }
> +}


Note that aio_set_fd_handler doesn't exist on Windows (yet), so this 
patch breaks compilation on Windows.  We can fix this for 2.2 and then 
apply your patches.

Paolo

Re: [Qemu-devel] [PATCH v2 1/3] nbd: Drop nbd_can_read()

[Max Reitz]

On 19.06.2014 10:44, Paolo Bonzini wrote:
> Il 18/06/2014 21:06, Max Reitz ha scritto:
>> +static void nbd_set_handlers(NBDClient *client)
>> +{
>> +    if (client->exp && client->exp->ctx) {
>> +        aio_set_fd_handler(client->exp->ctx, client->sock,
>> +                           client->can_read ? nbd_read : NULL,
>> +                           client->send_coroutine ? 
>> nbd_restart_write : NULL,
>> +                           client);
>> +    }
>> +}
>
>
> Note that aio_set_fd_handler doesn't exist on Windows (yet), so this 
> patch breaks compilation on Windows.  We can fix this for 2.2 and then 
> apply your patches.

Okay, I'll add "for 2.2" to the subject and a comment to the cover 
letter for v3.

Max

Re: [Qemu-devel] [PATCH v2 1/3] nbd: Drop nbd_can_read()

[Max Reitz]

On 19.06.2014 05:58, Stefan Hajnoczi wrote:
> On Wed, Jun 18, 2014 at 09:06:41PM +0200, Max Reitz wrote:
>> +static void nbd_update_can_read(NBDClient *client)
>> +{
>> +    bool can_read = client->recv_coroutine ||
>> +                    client->nb_requests < MAX_NBD_REQUESTS;
>> +
>> +    if (can_read != client->can_read) {
>> +        client->can_read = can_read;
>> +        nbd_set_handlers(client);
>> +
>> +        /* If we got here, nb_requests had to be MAX_NBD_REQUESTS before */
>> +        if (client->nb_requests < MAX_NBD_REQUESTS) {
>> +            aio_notify(client->exp->ctx);
>> +        }
> nbd_set_handlers() indirectly invokes aio_notify(client->exp->ctx) via
> aio_set_fd_handler().  This if statement is redundant.

Thanks for spotting this, I'll drop it in v3.

Max