X-Git-Url: https://repo.jachan.dev/qemu.git/blobdiff_plain/eabc977973103527bbb8fed69c91cfaa6691f8ab..7df9381b7aa56c897e344f3bfe43bf5848bbd3e0:/async.c diff --git a/async.c b/async.c index a232192148..d4dd2cc799 100644 --- a/async.c +++ b/async.c @@ -22,6 +22,7 @@ * THE SOFTWARE. */ +#include "qemu/osdep.h" #include "qemu-common.h" #include "block/aio.h" #include "block/thread-pool.h" @@ -59,6 +60,11 @@ QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque) return bh; } +void aio_bh_call(QEMUBH *bh) +{ + bh->cb(bh->opaque); +} + /* Multiple occurrences of aio_bh_poll cannot be called concurrently */ int aio_bh_poll(AioContext *ctx) { @@ -79,10 +85,12 @@ int aio_bh_poll(AioContext *ctx) * aio_notify again if necessary. */ if (!bh->deleted && atomic_xchg(&bh->scheduled, 0)) { - if (!bh->idle) + /* Idle BHs and the notify BH don't count as progress */ + if (!bh->idle && bh != ctx->notify_dummy_bh) { ret = 1; + } bh->idle = 0; - bh->cb(bh->opaque); + aio_bh_call(bh); } } @@ -203,6 +211,8 @@ aio_ctx_check(GSource *source) QEMUBH *bh; atomic_and(&ctx->notify_me, ~1); + aio_notify_accept(ctx); + for (bh = ctx->first_bh; bh; bh = bh->next) { if (!bh->deleted && bh->scheduled) { return true; @@ -228,8 +238,22 @@ aio_ctx_finalize(GSource *source) { AioContext *ctx = (AioContext *) source; + qemu_bh_delete(ctx->notify_dummy_bh); thread_pool_free(ctx->thread_pool); - aio_set_event_notifier(ctx, &ctx->notifier, NULL); + + qemu_mutex_lock(&ctx->bh_lock); + while (ctx->first_bh) { + QEMUBH *next = ctx->first_bh->next; + + /* qemu_bh_delete() must have been called on BHs in this AioContext */ + assert(ctx->first_bh->deleted); + + g_free(ctx->first_bh); + ctx->first_bh = next; + } + qemu_mutex_unlock(&ctx->bh_lock); + + aio_set_event_notifier(ctx, &ctx->notifier, false, NULL); event_notifier_cleanup(&ctx->notifier); rfifolock_destroy(&ctx->lock); qemu_mutex_destroy(&ctx->bh_lock); @@ -265,6 +289,14 @@ void aio_notify(AioContext *ctx) smp_mb(); if (ctx->notify_me) { event_notifier_set(&ctx->notifier); + atomic_mb_set(&ctx->notified, true); + } +} + +void aio_notify_accept(AioContext *ctx) +{ + if (atomic_xchg(&ctx->notified, false)) { + event_notifier_test_and_clear(&ctx->notifier); } } @@ -275,31 +307,54 @@ static void aio_timerlist_notify(void *opaque) static void aio_rfifolock_cb(void *opaque) { + AioContext *ctx = opaque; + /* Kick owner thread in case they are blocked in aio_poll() */ - aio_notify(opaque); + qemu_bh_schedule(ctx->notify_dummy_bh); +} + +static void notify_dummy_bh(void *opaque) +{ + /* Do nothing, we were invoked just to force the event loop to iterate */ +} + +static void event_notifier_dummy_cb(EventNotifier *e) +{ } AioContext *aio_context_new(Error **errp) { int ret; AioContext *ctx; + Error *local_err = NULL; + ctx = (AioContext *) g_source_new(&aio_source_funcs, sizeof(AioContext)); + aio_context_setup(ctx, &local_err); + if (local_err) { + error_propagate(errp, local_err); + goto fail; + } ret = event_notifier_init(&ctx->notifier, false); if (ret < 0) { - g_source_destroy(&ctx->source); error_setg_errno(errp, -ret, "Failed to initialize event notifier"); - return NULL; + goto fail; } g_source_set_can_recurse(&ctx->source, true); aio_set_event_notifier(ctx, &ctx->notifier, + false, (EventNotifierHandler *) - event_notifier_test_and_clear); + event_notifier_dummy_cb); ctx->thread_pool = NULL; qemu_mutex_init(&ctx->bh_lock); rfifolock_init(&ctx->lock, aio_rfifolock_cb, ctx); timerlistgroup_init(&ctx->tlg, aio_timerlist_notify, ctx); + ctx->notify_dummy_bh = aio_bh_new(ctx, notify_dummy_bh, NULL); + return ctx; +fail: + g_source_destroy(&ctx->source); + return NULL; } void aio_context_ref(AioContext *ctx)