[Qemu-devel] [PATCH v2 2/2] QEMUBH: make AioContext's bh re-entrant

[Liu Ping Fan <qemulist@gmail.com>]

BH will be used outside big lock, so introduce lock to protect
between the writers, ie, bh's adders and deleter.
Note that the lock only affects the writers and bh's callback does
not take this extra lock.

Signed-off-by: Liu Ping Fan <pingfanl@linux.vnet.ibm.com>
---
 async.c             | 21 +++++++++++++++++++++
 include/block/aio.h |  3 +++
 2 files changed, 24 insertions(+)

diff --git a/async.c b/async.c
index 90fe906..6a3269f 100644
--- a/async.c
+++ b/async.c
@@ -47,8 +47,12 @@ QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
     bh->ctx = ctx;
     bh->cb = cb;
     bh->opaque = opaque;
+    qemu_mutex_lock(&ctx->bh_lock);
     bh->next = ctx->first_bh;
+    /* Make sure the memebers ready before putting bh into list */
+    smp_wmb();
     ctx->first_bh = bh;
+    qemu_mutex_unlock(&ctx->bh_lock);
     return bh;
 }
 
@@ -61,12 +65,18 @@ int aio_bh_poll(AioContext *ctx)
 
     ret = 0;
     for (bh = ctx->first_bh; bh; bh = next) {
+        /* Make sure fetching bh before accessing its members */
+        smp_read_barrier_depends();
         next = bh->next;
         if (!bh->deleted && bh->scheduled) {
             bh->scheduled = 0;
             if (!bh->idle)
                 ret = 1;
             bh->idle = 0;
+            /* Paired with write barrier in bh schedule to ensure reading for
+             *  callbacks coming after bh's scheduling.
+             */
+            smp_rmb();
             bh->cb(bh->opaque);
         }
     }
@@ -75,6 +85,7 @@ int aio_bh_poll(AioContext *ctx)
 
     /* remove deleted bhs */
     if (!ctx->walking_bh) {
+        qemu_mutex_lock(&ctx->bh_lock);
         bhp = &ctx->first_bh;
         while (*bhp) {
             bh = *bhp;
@@ -85,6 +96,7 @@ int aio_bh_poll(AioContext *ctx)
                 bhp = &bh->next;
             }
         }
+        qemu_mutex_unlock(&ctx->bh_lock);
     }
 
     return ret;
@@ -94,6 +106,10 @@ void qemu_bh_schedule_idle(QEMUBH *bh)
 {
     if (bh->scheduled)
         return;
+    /* Make sure any writes that are needed by the callback are done
+     * before the locations are read in the aio_bh_poll.
+     */
+    smp_wmb();
     bh->scheduled = 1;
     bh->idle = 1;
 }
@@ -102,6 +118,10 @@ void qemu_bh_schedule(QEMUBH *bh)
 {
     if (bh->scheduled)
         return;
+    /* Make sure any writes that are needed by the callback are done
+     * before the locations are read in the aio_bh_poll.
+     */
+    smp_wmb();
     bh->scheduled = 1;
     bh->idle = 0;
     aio_notify(bh->ctx);
@@ -211,6 +231,7 @@ AioContext *aio_context_new(void)
     ctx = (AioContext *) g_source_new(&aio_source_funcs, sizeof(AioContext));
     ctx->pollfds = g_array_new(FALSE, FALSE, sizeof(GPollFD));
     ctx->thread_pool = NULL;
+    qemu_mutex_init(&ctx->bh_lock);
     event_notifier_init(&ctx->notifier, false);
     aio_set_event_notifier(ctx, &ctx->notifier, 
                            (EventNotifierHandler *)
diff --git a/include/block/aio.h b/include/block/aio.h
index 1836793..971fbef 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -17,6 +17,7 @@
 #include "qemu-common.h"
 #include "qemu/queue.h"
 #include "qemu/event_notifier.h"
+#include "qemu/thread.h"
 
 typedef struct BlockDriverAIOCB BlockDriverAIOCB;
 typedef void BlockDriverCompletionFunc(void *opaque, int ret);
@@ -53,6 +54,8 @@ typedef struct AioContext {
      */
     int walking_handlers;
 
+    /* lock to protect between bh's adders and deleter */
+    QemuMutex bh_lock;
     /* Anchor of the list of Bottom Halves belonging to the context */
     struct QEMUBH *first_bh;
 
-- 
1.8.1.4