[Qemu-devel] [RFC PATCH v11 15/23] aio: replace stack of bottom halves with queue

[Pavel Dovgalyuk <Pavel.Dovgaluk@ispras.ru>]

Bottom halves in AIO context are stored and removes
in LIFO order. It makes their execution non-deterministic.
This patch replaces the stack with queue to preserve the
order of bottom halves processing.

Signed-off-by: Pavel Dovgalyuk <pavel.dovgaluk@ispras.ru>
---
 async.c              |   26 +++++++++++---------------
 include/block/aio.h  |    4 ++--
 include/qemu/queue.h |   18 ++++++++++++++++++
 3 files changed, 31 insertions(+), 17 deletions(-)

diff --git a/async.c b/async.c
index 2b51e87..bd975c9 100644
--- a/async.c
+++ b/async.c
@@ -35,7 +35,7 @@ struct QEMUBH {
     AioContext *ctx;
     QEMUBHFunc *cb;
     void *opaque;
-    QEMUBH *next;
+    QSIMPLEQ_ENTRY(QEMUBH) next;
     bool scheduled;
     bool idle;
     bool deleted;
@@ -51,10 +51,7 @@ QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
         .opaque = opaque,
     };
     qemu_mutex_lock(&ctx->bh_lock);
-    bh->next = ctx->first_bh;
-    /* Make sure that the members are ready before putting bh into list */
-    smp_wmb();
-    ctx->first_bh = bh;
+    QSIMPLEQ_INSERT_TAIL_RCU(&ctx->bh_queue, bh, next);
     qemu_mutex_unlock(&ctx->bh_lock);
     return bh;
 }
@@ -62,16 +59,15 @@ QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
 /* Multiple occurrences of aio_bh_poll cannot be called concurrently */
 int aio_bh_poll(AioContext *ctx)
 {
-    QEMUBH *bh, **bhp, *next;
+    QEMUBH *bh, *next, *prev;
     int ret;
 
     ctx->walking_bh++;
 
     ret = 0;
-    for (bh = ctx->first_bh; bh; bh = next) {
+    QSIMPLEQ_FOREACH(bh, &ctx->bh_queue, next) {
         /* Make sure that fetching bh happens before accessing its members */
         smp_read_barrier_depends();
-        next = bh->next;
         /* The atomic_xchg is paired with the one in qemu_bh_schedule.  The
          * implicit memory barrier ensures that the callback sees all writes
          * done by the scheduling thread.  It also ensures that the scheduling
@@ -91,14 +87,13 @@ 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;
+        prev = NULL;
+        QSIMPLEQ_FOREACH_SAFE(bh, &ctx->bh_queue, next, next) {
             if (bh->deleted) {
-                *bhp = bh->next;
+                QSIMPLEQ_REMOVE_AFTER(&ctx->bh_queue, prev, QEMUBH, next);
                 g_free(bh);
             } else {
-                bhp = &bh->next;
+                prev = bh;
             }
         }
         qemu_mutex_unlock(&ctx->bh_lock);
@@ -157,7 +152,7 @@ aio_compute_timeout(AioContext *ctx)
     int timeout = -1;
     QEMUBH *bh;
 
-    for (bh = ctx->first_bh; bh; bh = bh->next) {
+    QSIMPLEQ_FOREACH(bh, &ctx->bh_queue, next) {
         if (!bh->deleted && bh->scheduled) {
             if (bh->idle) {
                 /* idle bottom halves will be polled at least
@@ -200,7 +195,7 @@ aio_ctx_check(GSource *source)
     AioContext *ctx = (AioContext *) source;
     QEMUBH *bh;
 
-    for (bh = ctx->first_bh; bh; bh = bh->next) {
+    QSIMPLEQ_FOREACH(bh, &ctx->bh_queue, next) {
         if (!bh->deleted && bh->scheduled) {
             return true;
 	}
@@ -307,6 +302,7 @@ AioContext *aio_context_new(Error **errp)
     qemu_mutex_init(&ctx->bh_lock);
     rfifolock_init(&ctx->lock, aio_rfifolock_cb, ctx);
     timerlistgroup_init(&ctx->tlg, aio_timerlist_notify, ctx);
+    QSIMPLEQ_INIT(&ctx->bh_queue);
 
     return ctx;
 }
diff --git a/include/block/aio.h b/include/block/aio.h
index 7d1e26b..82cdf78 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -71,8 +71,8 @@ struct AioContext {
     /* 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;
+    /* List of Bottom Halves belonging to the context */
+    QSIMPLEQ_HEAD(, QEMUBH) bh_queue;
 
     /* A simple lock used to protect the first_bh list, and ensure that
      * no callbacks are removed while we're walking and dispatching callbacks.
diff --git a/include/qemu/queue.h b/include/qemu/queue.h
index f781aa2..99564bc 100644
--- a/include/qemu/queue.h
+++ b/include/qemu/queue.h
@@ -271,6 +271,13 @@ struct {                                                                \
     (head)->sqh_last = &(elm)->field.sqe_next;                          \
 } while (/*CONSTCOND*/0)
 
+#define QSIMPLEQ_INSERT_TAIL_RCU(head, elm, field) do {                 \
+    (elm)->field.sqe_next = NULL;                                       \
+    smp_wmb();                                                          \
+    atomic_rcu_set((head)->sqh_last, (elm));                            \
+    (head)->sqh_last = &(elm)->field.sqe_next;                          \
+} while (/*CONSTCOND*/0)
+
 #define QSIMPLEQ_INSERT_AFTER(head, listelm, elm, field) do {           \
     if (((elm)->field.sqe_next = (listelm)->field.sqe_next) == NULL)    \
         (head)->sqh_last = &(elm)->field.sqe_next;                      \
@@ -306,6 +313,17 @@ struct {                                                                \
     }                                                                   \
 } while (/*CONSTCOND*/0)
 
+#define QSIMPLEQ_REMOVE_AFTER(head, curelm, type, field) do {           \
+    if ((curelm) == NULL) {                                             \
+        QSIMPLEQ_REMOVE_HEAD((head), field);                            \
+    } else {                                                            \
+        if (((curelm)->field.sqe_next =                                 \
+            (curelm)->field.sqe_next->field.sqe_next) == NULL) {        \
+                (head)->sqh_last = &(curelm)->field.sqe_next;           \
+        }                                                               \
+    }                                                                   \
+} while (/*CONSTCOND*/0)
+
 #define QSIMPLEQ_FOREACH(var, head, field)                              \
     for ((var) = ((head)->sqh_first);                                   \
         (var);                                                          \