[PATCH] util/async: make bh_aio_poll() O(1)

[PATCH] util/async: make bh_aio_poll() O(1)

[Stefan Hajnoczi]

The ctx->first_bh list contains all created BHs, including those that
are not scheduled.  The list is iterated by the event loop and therefore
has O(n) time complexity with respected to the number of created BHs.

Rewrite BHs so that only scheduled or deleted BHs are enqueued.
Only BHs that actually require action will be iterated.

One semantic change is required: qemu_bh_delete() enqueues the BH and
therefore invokes aio_notify().  The
tests/test-aio.c:test_source_bh_delete_from_cb() test case assumed that
g_main_context_iteration(NULL, false) returns false after
qemu_bh_delete() but it now returns true for one iteration.  Fix up the
test case.

This patch makes aio_compute_timeout() and aio_bh_poll() drop from a CPU
profile reported by perf-top(1).  Previously they combined to 9% CPU
utilization when AioContext polling is commented out and the guest has 2
virtio-blk,num-queues=1 and 99 virtio-blk,num-queues=32 devices.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/block/aio.h |  17 +++-
 tests/test-aio.c    |   3 +-
 util/async.c        | 242 ++++++++++++++++++++++++++------------------
 3 files changed, 160 insertions(+), 102 deletions(-)

diff --git a/include/block/aio.h b/include/block/aio.h
index 7ba9bd7874..51f4aa09cb 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -51,6 +51,18 @@ struct ThreadPool;
 struct LinuxAioState;
 struct LuringState;
 
+/*
+ * Each aio_bh_poll() call carves off a slice of the BH list.  This way newly
+ * scheduled BHs are not processed until the next aio_bh_poll() call.  This
+ * concept extends to nested aio_bh_poll() calls because slices are chained
+ * together.
+ */
+typedef struct BHListSlice BHListSlice;
+struct BHListSlice {
+    QEMUBH *first_bh;
+    BHListSlice *next;
+};
+
 struct AioContext {
     GSource source;
 
@@ -91,8 +103,9 @@ struct AioContext {
      */
     QemuLockCnt list_lock;
 
-    /* Anchor of the list of Bottom Halves belonging to the context */
-    struct QEMUBH *first_bh;
+    /* Bottom Halves pending aio_bh_poll() processing */
+    BHListSlice bh_list;
+    BHListSlice **bh_list_tail;
 
     /* Used by aio_notify.
      *
diff --git a/tests/test-aio.c b/tests/test-aio.c
index 86fb73b3d5..8a46078463 100644
--- a/tests/test-aio.c
+++ b/tests/test-aio.c
@@ -615,7 +615,8 @@ static void test_source_bh_delete_from_cb(void)
     g_assert_cmpint(data1.n, ==, data1.max);
     g_assert(data1.bh == NULL);
 
-    g_assert(!g_main_context_iteration(NULL, false));
+    assert(g_main_context_iteration(NULL, false));
+    assert(!g_main_context_iteration(NULL, false));
 }
 
 static void test_source_bh_delete_from_cb_many(void)
diff --git a/util/async.c b/util/async.c
index c192a24a61..a7d334efc0 100644
--- a/util/async.c
+++ b/util/async.c
@@ -36,16 +36,83 @@
 /***********************************************************/
 /* bottom halves (can be seen as timers which expire ASAP) */
 
+/* QEMUBH::flags values */
+enum {
+    /* Already enqueued and waiting for aio_bh_poll() */
+    BH_PENDING   = (1 << 0),
+
+    /* Invoke the callback */
+    BH_SCHEDULED = (1 << 1),
+
+    /* Delete without invoking callback */
+    BH_DELETED   = (1 << 2),
+
+    /* Delete after invoking callback */
+    BH_ONESHOT   = (1 << 3),
+
+    /* Schedule periodically when the event loop is idle */
+    BH_IDLE      = (1 << 4),
+};
+
 struct QEMUBH {
     AioContext *ctx;
     QEMUBHFunc *cb;
     void *opaque;
     QEMUBH *next;
-    bool scheduled;
-    bool idle;
-    bool deleted;
+    unsigned flags;
 };
 
+/* Called concurrently from any thread */
+static void aio_bh_enqueue(QEMUBH *bh, unsigned new_flags)
+{
+    AioContext *ctx = bh->ctx;
+    unsigned old_flags;
+
+    /*
+     * The memory barrier implicit in atomic_fetch_or makes sure that:
+     * 1. idle & any writes needed by the callback are done before the
+     *    locations are read in the aio_bh_poll.
+     * 2. ctx is loaded before the callback has a chance to execute and bh
+     *    could be freed.
+     */
+    old_flags = atomic_fetch_or(&bh->flags, BH_PENDING | new_flags);
+    if (!(old_flags & BH_PENDING)) {
+        QEMUBH *old;
+
+        do {
+            old = ctx->bh_list.first_bh;
+            bh->next = old;
+        } while (atomic_cmpxchg(&ctx->bh_list.first_bh, old, bh) != old);
+    }
+
+    aio_notify(ctx);
+}
+
+/* Only called from aio_bh_poll() and aio_ctx_finalize() */
+static QEMUBH *aio_bh_dequeue(QEMUBH **first_bh, unsigned *flags)
+{
+    QEMUBH *bh;
+
+    bh = *first_bh;
+    if (!bh) {
+        return NULL;
+    }
+
+    *first_bh = bh->next;
+    bh->next = NULL;
+
+    /*
+     * The atomic_and is paired with aio_bh_enqueue().  The implicit memory
+     * barrier ensures that the callback sees all writes done by the scheduling
+     * thread.  It also ensures that the scheduling thread sees the cleared
+     * flag before bh->cb has run, and thus will call aio_notify again if
+     * necessary.
+     */
+    *flags = atomic_fetch_and(&bh->flags,
+                              ~(BH_PENDING | BH_SCHEDULED | BH_IDLE));
+    return bh;
+}
+
 void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
 {
     QEMUBH *bh;
@@ -55,15 +122,7 @@ void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
         .cb = cb,
         .opaque = opaque,
     };
-    qemu_lockcnt_lock(&ctx->list_lock);
-    bh->next = ctx->first_bh;
-    bh->scheduled = 1;
-    bh->deleted = 1;
-    /* Make sure that the members are ready before putting bh into list */
-    smp_wmb();
-    ctx->first_bh = bh;
-    qemu_lockcnt_unlock(&ctx->list_lock);
-    aio_notify(ctx);
+    aio_bh_enqueue(bh, BH_SCHEDULED | BH_ONESHOT);
 }
 
 QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
@@ -75,12 +134,6 @@ QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
         .cb = cb,
         .opaque = opaque,
     };
-    qemu_lockcnt_lock(&ctx->list_lock);
-    bh->next = ctx->first_bh;
-    /* Make sure that the members are ready before putting bh into list */
-    smp_wmb();
-    ctx->first_bh = bh;
-    qemu_lockcnt_unlock(&ctx->list_lock);
     return bh;
 }
 
@@ -89,91 +142,78 @@ void aio_bh_call(QEMUBH *bh)
     bh->cb(bh->opaque);
 }
 
+/* Returns old bh_list_tail to be restored with pop_bh_list_slice() */
+static BHListSlice **push_bh_list_slice(AioContext *ctx, BHListSlice *slice)
+{
+    BHListSlice **old_tail;
+
+    slice->first_bh = atomic_xchg(&ctx->bh_list.first_bh, NULL);
+    slice->next = NULL;
+    *ctx->bh_list_tail = slice;
+    old_tail = ctx->bh_list_tail;
+    ctx->bh_list_tail = &slice->next;
+    return old_tail;
+}
+
+static void pop_bh_list_slice(AioContext *ctx, BHListSlice **old_tail)
+{
+    /* All BHs must have been processed before a slice can be popped */
+    assert((*old_tail)->first_bh == NULL);
+
+    *old_tail = NULL;
+    ctx->bh_list_tail = old_tail;
+}
+
 /* Multiple occurrences of aio_bh_poll cannot be called concurrently.
  * The count in ctx->list_lock is incremented before the call, and is
  * not affected by the call.
  */
 int aio_bh_poll(AioContext *ctx)
 {
-    QEMUBH *bh, **bhp, *next;
+    BHListSlice slice;
+    BHListSlice **old_tail;
+    BHListSlice *s;
+    QEMUBH *bh;
+    unsigned flags;
     int ret;
-    bool deleted = false;
+
+    old_tail = push_bh_list_slice(ctx, &slice);
 
     ret = 0;
-    for (bh = atomic_rcu_read(&ctx->first_bh); bh; bh = next) {
-        next = atomic_rcu_read(&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
-         * thread sees the zero before bh->cb has run, and thus will call
-         * aio_notify again if necessary.
-         */
-        if (atomic_xchg(&bh->scheduled, 0)) {
-            /* Idle BHs don't count as progress */
-            if (!bh->idle) {
-                ret = 1;
+    for (s = ctx->bh_list.next; s; s = s->next) {
+        while ((bh = aio_bh_dequeue(&s->first_bh, &flags))) {
+            if ((flags & (BH_SCHEDULED | BH_DELETED)) == BH_SCHEDULED) {
+                /* Idle BHs don't count as progress */
+                if (!(flags & BH_IDLE)) {
+                    ret = 1;
+                }
+                aio_bh_call(bh);
             }
-            bh->idle = 0;
-            aio_bh_call(bh);
-        }
-        if (bh->deleted) {
-            deleted = true;
-        }
-    }
-
-    /* remove deleted bhs */
-    if (!deleted) {
-        return ret;
-    }
-
-    if (qemu_lockcnt_dec_if_lock(&ctx->list_lock)) {
-        bhp = &ctx->first_bh;
-        while (*bhp) {
-            bh = *bhp;
-            if (bh->deleted && !bh->scheduled) {
-                *bhp = bh->next;
+            if (flags & (BH_DELETED | BH_ONESHOT)) {
                 g_free(bh);
-            } else {
-                bhp = &bh->next;
             }
         }
-        qemu_lockcnt_inc_and_unlock(&ctx->list_lock);
     }
+
+    pop_bh_list_slice(ctx, old_tail);
     return ret;
 }
 
 void qemu_bh_schedule_idle(QEMUBH *bh)
 {
-    bh->idle = 1;
-    /* Make sure that idle & any writes needed by the callback are done
-     * before the locations are read in the aio_bh_poll.
-     */
-    atomic_mb_set(&bh->scheduled, 1);
+    aio_bh_enqueue(bh, BH_SCHEDULED | BH_IDLE);
 }
 
 void qemu_bh_schedule(QEMUBH *bh)
 {
-    AioContext *ctx;
-
-    ctx = bh->ctx;
-    bh->idle = 0;
-    /* The memory barrier implicit in atomic_xchg makes sure that:
-     * 1. idle & any writes needed by the callback are done before the
-     *    locations are read in the aio_bh_poll.
-     * 2. ctx is loaded before scheduled is set and the callback has a chance
-     *    to execute.
-     */
-    if (atomic_xchg(&bh->scheduled, 1) == 0) {
-        aio_notify(ctx);
-    }
+    aio_bh_enqueue(bh, BH_SCHEDULED);
 }
 
-
 /* This func is async.
  */
 void qemu_bh_cancel(QEMUBH *bh)
 {
-    atomic_mb_set(&bh->scheduled, 0);
+    atomic_and(&bh->flags, ~BH_SCHEDULED);
 }
 
 /* This func is async.The bottom half will do the delete action at the finial
@@ -181,8 +221,7 @@ void qemu_bh_cancel(QEMUBH *bh)
  */
 void qemu_bh_delete(QEMUBH *bh)
 {
-    bh->scheduled = 0;
-    bh->deleted = 1;
+    aio_bh_enqueue(bh, BH_DELETED);
 }
 
 int64_t
@@ -191,18 +230,20 @@ aio_compute_timeout(AioContext *ctx)
     int64_t deadline;
     int timeout = -1;
     QEMUBH *bh;
-
-    for (bh = atomic_rcu_read(&ctx->first_bh); bh;
-         bh = atomic_rcu_read(&bh->next)) {
-        if (bh->scheduled) {
-            if (bh->idle) {
-                /* idle bottom halves will be polled at least
-                 * every 10ms */
-                timeout = 10000000;
-            } else {
-                /* non-idle bottom halves will be executed
-                 * immediately */
-                return 0;
+    BHListSlice *s = &ctx->bh_list;
+
+    for (s = &ctx->bh_list; s; s = s->next) {
+        for (bh = atomic_rcu_read(&s->first_bh); bh; bh = bh->next) {
+            if ((bh->flags & (BH_SCHEDULED | BH_DELETED)) == BH_SCHEDULED) {
+                if (bh->flags & BH_IDLE) {
+                    /* idle bottom halves will be polled at least
+                     * every 10ms */
+                    timeout = 10000000;
+                } else {
+                    /* non-idle bottom halves will be executed
+                     * immediately */
+                    return 0;
+                }
             }
         }
     }
@@ -237,13 +278,16 @@ aio_ctx_check(GSource *source)
 {
     AioContext *ctx = (AioContext *) source;
     QEMUBH *bh;
+    BHListSlice *s;
 
     atomic_and(&ctx->notify_me, ~1);
     aio_notify_accept(ctx);
 
-    for (bh = ctx->first_bh; bh; bh = bh->next) {
-        if (bh->scheduled) {
-            return true;
+    for (s = &ctx->bh_list; s; s = s->next) {
+        for (bh = atomic_rcu_read(&s->first_bh); bh; bh = bh->next) {
+            if ((bh->flags & (BH_SCHEDULED | BH_DELETED)) == BH_SCHEDULED) {
+                return true;
+            }
         }
     }
     return aio_pending(ctx) || (timerlistgroup_deadline_ns(&ctx->tlg) == 0);
@@ -265,6 +309,8 @@ static void
 aio_ctx_finalize(GSource     *source)
 {
     AioContext *ctx = (AioContext *) source;
+    QEMUBH *bh;
+    unsigned flags;
 
     thread_pool_free(ctx->thread_pool);
 
@@ -287,18 +333,15 @@ aio_ctx_finalize(GSource     *source)
     assert(QSLIST_EMPTY(&ctx->scheduled_coroutines));
     qemu_bh_delete(ctx->co_schedule_bh);
 
-    qemu_lockcnt_lock(&ctx->list_lock);
-    assert(!qemu_lockcnt_count(&ctx->list_lock));
-    while (ctx->first_bh) {
-        QEMUBH *next = ctx->first_bh->next;
+    /* There must be no aio_bh_poll() calls going on */
+    assert(ctx->bh_list.next == NULL);
 
+    while ((bh = aio_bh_dequeue(&ctx->bh_list.first_bh, &flags))) {
         /* qemu_bh_delete() must have been called on BHs in this AioContext */
-        assert(ctx->first_bh->deleted);
+        assert(flags & BH_DELETED);
 
-        g_free(ctx->first_bh);
-        ctx->first_bh = next;
+        g_free(bh);
     }
-    qemu_lockcnt_unlock(&ctx->list_lock);
 
     aio_set_event_notifier(ctx, &ctx->notifier, false, NULL, NULL);
     event_notifier_cleanup(&ctx->notifier);
@@ -445,6 +488,7 @@ AioContext *aio_context_new(Error **errp)
     AioContext *ctx;
 
     ctx = (AioContext *) g_source_new(&aio_source_funcs, sizeof(AioContext));
+    ctx->bh_list_tail = &ctx->bh_list.next;
     aio_context_setup(ctx);
 
     ret = event_notifier_init(&ctx->notifier, false);
-- 
2.24.1

Re: [PATCH] util/async: make bh_aio_poll() O(1)

[Paolo Bonzini]

Really a great idea, though I have some remarks on the implementation below.

On 19/02/20 11:00, Stefan Hajnoczi wrote:
> + * Each aio_bh_poll() call carves off a slice of the BH list.  This way newly
> + * scheduled BHs are not processed until the next aio_bh_poll() call.  This
> + * concept extends to nested aio_bh_poll() calls because slices are chained
> + * together.

This is the tricky part so I would expand a bit on why it's needed:

/*
 * Each aio_bh_poll() call carves off a slice of the BH list, so that
 * newly scheduled BHs are not processed until the next aio_bh_poll()
 * call.  All active aio_bh_poll() calls chained their slices together
 * in a list, so that nested aio_bh_poll() calls process all scheduled
 * bottom halves.
 */

> +struct BHListSlice {
> +    QEMUBH *first_bh;
> +    BHListSlice *next;
> +};
> +

Using QLIST and QSLIST removes the need to create your own lists, since
you can use QSLIST_MOVE_ATOMIC and QSLIST_INSERT_HEAD_ATOMIC.  For example:

struct BHListSlice {
    QSLIST_HEAD(, QEMUBH) first_bh;
    QLIST_ENTRY(BHListSlice) next;
};

...

    QSLIST_HEAD(, QEMUBH) active_bh;
    QLIST_HEAD(, BHListSlice) bh_list;

Related to this, in the aio_bh_poll() loop:

    for (s = ctx->bh_list.next; s; s = s->next) {
    }

You can actually do the removal inside the loop.  This is slightly more
efficient since you can remove slices early from the nested
aio_bh_poll().  Not that it's relevant for performance, but I think the
FIFO order for slices is also more intuitive than LIFO.

Putting this idea together with the QLIST one, you would get:

    /*
     * If a bottom half causes a recursive call, this slice will be
     * removed by the nested aio_bh_poll().
     */
    QSLIST_MOVE_ATOMIC(&slice.first_bh, ctx->active_bh);
    QLIST_INSERT_TAIL(&ctx->bh_list, slice);
    while ((s = QLIST_FIRST(&ctx->bh_list)) {
        while ((bh = aio_bh_dequeue(&s, &flags))) {
        }
        QLIST_REMOVE_HEAD(s, next);
    }

>  /* Multiple occurrences of aio_bh_poll cannot be called concurrently.
>   * The count in ctx->list_lock is incremented before the call, and is
>   * not affected by the call.

The second sentence is now stale.

Paolo

>   */
>  int aio_bh_poll(AioContext *ctx)
>  {
> -    QEMUBH *bh, **bhp, *next;
> +    BHListSlice slice;
> +    BHListSlice **old_tail;
> +    BHListSlice *s;
> +    QEMUBH *bh;
> +    unsigned flags;
>      int ret;
> -    bool deleted = false;
> +
> +    old_tail = push_bh_list_slice(ctx, &slice);
>  

Re: [PATCH] util/async: make bh_aio_poll() O(1)

[Stefan Hajnoczi]

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

On Wed, Feb 19, 2020 at 12:09:48PM +0100, Paolo Bonzini wrote:
> Really a great idea, though I have some remarks on the implementation below.
> 
> On 19/02/20 11:00, Stefan Hajnoczi wrote:
> > + * Each aio_bh_poll() call carves off a slice of the BH list.  This way newly
> > + * scheduled BHs are not processed until the next aio_bh_poll() call.  This
> > + * concept extends to nested aio_bh_poll() calls because slices are chained
> > + * together.
> 
> This is the tricky part so I would expand a bit on why it's needed:
> 
> /*
>  * Each aio_bh_poll() call carves off a slice of the BH list, so that
>  * newly scheduled BHs are not processed until the next aio_bh_poll()
>  * call.  All active aio_bh_poll() calls chained their slices together
>  * in a list, so that nested aio_bh_poll() calls process all scheduled
>  * bottom halves.
>  */

Thanks, will fix in v2.

> > +struct BHListSlice {
> > +    QEMUBH *first_bh;
> > +    BHListSlice *next;
> > +};
> > +
> 
> Using QLIST and QSLIST removes the need to create your own lists, since
> you can use QSLIST_MOVE_ATOMIC and QSLIST_INSERT_HEAD_ATOMIC.  For example:
> 
> struct BHListSlice {
>     QSLIST_HEAD(, QEMUBH) first_bh;
>     QLIST_ENTRY(BHListSlice) next;
> };
> 
> ...
> 
>     QSLIST_HEAD(, QEMUBH) active_bh;
>     QLIST_HEAD(, BHListSlice) bh_list;

I thought about this but chose the explicit tail pointer approach
because it lets aio_compute_timeout() and aio_ctx_check() iterate over
both the active BH list and slices in a single for loop :).  But
thinking about it more, maybe it can still be done by replacing
active_bh with a permanently present first BHListSlice element.

> 
> Related to this, in the aio_bh_poll() loop:
> 
>     for (s = ctx->bh_list.next; s; s = s->next) {
>     }
> 
> You can actually do the removal inside the loop.  This is slightly more
> efficient since you can remove slices early from the nested
> aio_bh_poll().  Not that it's relevant for performance, but I think the
> FIFO order for slices is also more intuitive than LIFO.
> 
> Putting this idea together with the QLIST one, you would get:
> 
>     /*
>      * If a bottom half causes a recursive call, this slice will be
>      * removed by the nested aio_bh_poll().
>      */
>     QSLIST_MOVE_ATOMIC(&slice.first_bh, ctx->active_bh);
>     QLIST_INSERT_TAIL(&ctx->bh_list, slice);
>     while ((s = QLIST_FIRST(&ctx->bh_list)) {
>         while ((bh = aio_bh_dequeue(&s, &flags))) {
>         }
>         QLIST_REMOVE_HEAD(s, next);
>     }

Cool, reusing "queue.h" is nice.

> 
> >  /* Multiple occurrences of aio_bh_poll cannot be called concurrently.
> >   * The count in ctx->list_lock is incremented before the call, and is
> >   * not affected by the call.
> 
> The second sentence is now stale.

Thanks, will fix in v2.

Stefan

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH] util/async: make bh_aio_poll() O(1)

[Paolo Bonzini]

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

Il mer 19 feb 2020, 18:58 Stefan Hajnoczi <stefanha@redhat.com> ha scritto:

> On Wed, Feb 19, 2020 at 12:09:48PM +0100, Paolo Bonzini wrote:
> > Really a great idea, though I have some remarks on the implementation
> below.
> >
> > On 19/02/20 11:00, Stefan Hajnoczi wrote:
> > > + * Each aio_bh_poll() call carves off a slice of the BH list.  This
> way newly
> > > + * scheduled BHs are not processed until the next aio_bh_poll()
> call.  This
> > > + * concept extends to nested aio_bh_poll() calls because slices are
> chained
> > > + * together.
> >
> > This is the tricky part so I would expand a bit on why it's needed:
> >
> > /*
> >  * Each aio_bh_poll() call carves off a slice of the BH list, so that
> >  * newly scheduled BHs are not processed until the next aio_bh_poll()
> >  * call.  All active aio_bh_poll() calls chained their slices together
> >  * in a list, so that nested aio_bh_poll() calls process all scheduled
> >  * bottom halves.
> >  */
>
> Thanks, will fix in v2.
>
> > > +struct BHListSlice {
> > > +    QEMUBH *first_bh;
> > > +    BHListSlice *next;
> > > +};
> > > +
> >
> > Using QLIST and QSLIST removes the need to create your own lists, since
> > you can use QSLIST_MOVE_ATOMIC and QSLIST_INSERT_HEAD_ATOMIC.  For
> example:
> >
> > struct BHListSlice {
> >     QSLIST_HEAD(, QEMUBH) first_bh;
> >     QLIST_ENTRY(BHListSlice) next;
> > };
> >
> > ...
> >
> >     QSLIST_HEAD(, QEMUBH) active_bh;
> >     QLIST_HEAD(, BHListSlice) bh_list;
>
> I thought about this but chose the explicit tail pointer approach
> because it lets aio_compute_timeout() and aio_ctx_check() iterate over
> both the active BH list and slices in a single for loop :).  But
> thinking about it more, maybe it can still be done by replacing
> active_bh with a permanently present first BHListSlice element.
>

Probably not so easy since the idea was to empty the slices list entirely
via the FIFO order.

But since you are rewriting everything anyway, can you add a flag for
whether there are any non-idle bottom halves anywhere in the list? It need
not be computed perfectly, because any non-idle bh will cause the idle
bottom halves to be triggered as well; you can just set in qemu_bh_schedule
and clear it at the end of aio_bh_poll.

Then if there is any active bh or slice you consult the flag and use it to
return the timeout, which will be either 0 or 10ms depending on the flag.
That's truly O(1). (More precisely, this patch goes from O(#created-bh) to
O(#scheduled-bh), which of course is optimal for aio_bh_poll but not for
aio_compute_timeout; making timeout computation O(1) can lower latency a
bit by decreasing the constant factor).

Paolo


> >
> > Related to this, in the aio_bh_poll() loop:
> >
> >     for (s = ctx->bh_list.next; s; s = s->next) {
> >     }
> >
> > You can actually do the removal inside the loop.  This is slightly more
> > efficient since you can remove slices early from the nested
> > aio_bh_poll().  Not that it's relevant for performance, but I think the
> > FIFO order for slices is also more intuitive than LIFO.
> >
> > Putting this idea together with the QLIST one, you would get:
> >
> >     /*
> >      * If a bottom half causes a recursive call, this slice will be
> >      * removed by the nested aio_bh_poll().
> >      */
> >     QSLIST_MOVE_ATOMIC(&slice.first_bh, ctx->active_bh);
> >     QLIST_INSERT_TAIL(&ctx->bh_list, slice);
> >     while ((s = QLIST_FIRST(&ctx->bh_list)) {
> >         while ((bh = aio_bh_dequeue(&s, &flags))) {
> >         }
> >         QLIST_REMOVE_HEAD(s, next);
> >     }
>
> Cool, reusing "queue.h" is nice.
>
> >
> > >  /* Multiple occurrences of aio_bh_poll cannot be called concurrently.
> > >   * The count in ctx->list_lock is incremented before the call, and is
> > >   * not affected by the call.
> >
> > The second sentence is now stale.
>
> Thanks, will fix in v2.
>
> Stefan
>

[-- Attachment #2: Type: text/html, Size: 5256 bytes --]

Re: [PATCH] util/async: make bh_aio_poll() O(1)

[Stefan Hajnoczi]

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

On Wed, Feb 19, 2020 at 08:05:12PM +0100, Paolo Bonzini wrote:
> Il mer 19 feb 2020, 18:58 Stefan Hajnoczi <stefanha@redhat.com> ha scritto:
> 
> > On Wed, Feb 19, 2020 at 12:09:48PM +0100, Paolo Bonzini wrote:
> > > Really a great idea, though I have some remarks on the implementation
> > below.
> > >
> > > On 19/02/20 11:00, Stefan Hajnoczi wrote:
> > > > + * Each aio_bh_poll() call carves off a slice of the BH list.  This
> > way newly
> > > > + * scheduled BHs are not processed until the next aio_bh_poll()
> > call.  This
> > > > + * concept extends to nested aio_bh_poll() calls because slices are
> > chained
> > > > + * together.
> > >
> > > This is the tricky part so I would expand a bit on why it's needed:
> > >
> > > /*
> > >  * Each aio_bh_poll() call carves off a slice of the BH list, so that
> > >  * newly scheduled BHs are not processed until the next aio_bh_poll()
> > >  * call.  All active aio_bh_poll() calls chained their slices together
> > >  * in a list, so that nested aio_bh_poll() calls process all scheduled
> > >  * bottom halves.
> > >  */
> >
> > Thanks, will fix in v2.
> >
> > > > +struct BHListSlice {
> > > > +    QEMUBH *first_bh;
> > > > +    BHListSlice *next;
> > > > +};
> > > > +
> > >
> > > Using QLIST and QSLIST removes the need to create your own lists, since
> > > you can use QSLIST_MOVE_ATOMIC and QSLIST_INSERT_HEAD_ATOMIC.  For
> > example:
> > >
> > > struct BHListSlice {
> > >     QSLIST_HEAD(, QEMUBH) first_bh;
> > >     QLIST_ENTRY(BHListSlice) next;
> > > };
> > >
> > > ...
> > >
> > >     QSLIST_HEAD(, QEMUBH) active_bh;
> > >     QLIST_HEAD(, BHListSlice) bh_list;
> >
> > I thought about this but chose the explicit tail pointer approach
> > because it lets aio_compute_timeout() and aio_ctx_check() iterate over
> > both the active BH list and slices in a single for loop :).  But
> > thinking about it more, maybe it can still be done by replacing
> > active_bh with a permanently present first BHListSlice element.
> >
> 
> Probably not so easy since the idea was to empty the slices list entirely
> via the FIFO order.
> 
> But since you are rewriting everything anyway, can you add a flag for
> whether there are any non-idle bottom halves anywhere in the list? It need
> not be computed perfectly, because any non-idle bh will cause the idle
> bottom halves to be triggered as well; you can just set in qemu_bh_schedule
> and clear it at the end of aio_bh_poll.
> 
> Then if there is any active bh or slice you consult the flag and use it to
> return the timeout, which will be either 0 or 10ms depending on the flag.
> That's truly O(1). (More precisely, this patch goes from O(#created-bh) to
> O(#scheduled-bh), which of course is optimal for aio_bh_poll but not for
> aio_compute_timeout; making timeout computation O(1) can lower latency a
> bit by decreasing the constant factor).

Yes, good idea.  I'll send a follow-up patch.

Stefan

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]