+static gboolean
+aio_ctx_prepare(GSource *source, gint *timeout)
+{
+ AioContext *ctx = (AioContext *) source;
+
+ /* We assume there is no timeout already supplied */
+ *timeout = qemu_timeout_ns_to_ms(aio_compute_timeout(ctx));
+
+ if (aio_prepare(ctx)) {
+ *timeout = 0;
+ }
+
+ return *timeout == 0;
+}
+
+static gboolean
+aio_ctx_check(GSource *source)
+{
+ AioContext *ctx = (AioContext *) source;
+ QEMUBH *bh;
+
+ for (bh = ctx->first_bh; bh; bh = bh->next) {
+ if (!bh->deleted && bh->scheduled) {
+ return true;
+ }
+ }
+ return aio_pending(ctx) || (timerlistgroup_deadline_ns(&ctx->tlg) == 0);
+}
+
+static gboolean
+aio_ctx_dispatch(GSource *source,
+ GSourceFunc callback,
+ gpointer user_data)
+{
+ AioContext *ctx = (AioContext *) source;
+
+ assert(callback == NULL);
+ aio_dispatch(ctx);
+ return true;
+}
+
+static void
+aio_ctx_finalize(GSource *source)
+{
+ AioContext *ctx = (AioContext *) source;
+
+ thread_pool_free(ctx->thread_pool);
+ aio_set_event_notifier(ctx, &ctx->notifier, NULL);
+ event_notifier_cleanup(&ctx->notifier);
+ rfifolock_destroy(&ctx->lock);
+ qemu_mutex_destroy(&ctx->bh_lock);
+ g_array_free(ctx->pollfds, TRUE);
+ timerlistgroup_deinit(&ctx->tlg);
+}
+
+static GSourceFuncs aio_source_funcs = {
+ aio_ctx_prepare,
+ aio_ctx_check,
+ aio_ctx_dispatch,
+ aio_ctx_finalize
+};
+
+GSource *aio_get_g_source(AioContext *ctx)
+{
+ g_source_ref(&ctx->source);
+ return &ctx->source;
+}
+
+ThreadPool *aio_get_thread_pool(AioContext *ctx)
+{
+ if (!ctx->thread_pool) {
+ ctx->thread_pool = thread_pool_new(ctx);
+ }
+ return ctx->thread_pool;
+}
+
+void aio_set_dispatching(AioContext *ctx, bool dispatching)
+{
+ ctx->dispatching = dispatching;
+ if (!dispatching) {
+ /* Write ctx->dispatching before reading e.g. bh->scheduled.
+ * Optimization: this is only needed when we're entering the "unsafe"
+ * phase where other threads must call event_notifier_set.
+ */
+ smp_mb();
+ }
+}
+
+void aio_notify(AioContext *ctx)
+{
+ /* Write e.g. bh->scheduled before reading ctx->dispatching. */
+ smp_mb();
+ if (!ctx->dispatching) {
+ event_notifier_set(&ctx->notifier);
+ }
+}
+
+static void aio_timerlist_notify(void *opaque)
+{
+ aio_notify(opaque);
+}
+
+static void aio_rfifolock_cb(void *opaque)
+{
+ /* Kick owner thread in case they are blocked in aio_poll() */
+ aio_notify(opaque);
+}
+
+AioContext *aio_context_new(Error **errp)
+{
+ int ret;
+ AioContext *ctx;
+ ctx = (AioContext *) g_source_new(&aio_source_funcs, sizeof(AioContext));
+ 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;
+ }
+ aio_set_event_notifier(ctx, &ctx->notifier,
+ (EventNotifierHandler *)
+ event_notifier_test_and_clear);
+ ctx->pollfds = g_array_new(FALSE, FALSE, sizeof(GPollFD));
+ 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);
+
+ return ctx;
+}
+
+void aio_context_ref(AioContext *ctx)
+{
+ g_source_ref(&ctx->source);
+}
+
+void aio_context_unref(AioContext *ctx)
+{
+ g_source_unref(&ctx->source);
+}
+
+void aio_context_acquire(AioContext *ctx)
+{
+ rfifolock_lock(&ctx->lock);
+}
+
+void aio_context_release(AioContext *ctx)
+{
+ rfifolock_unlock(&ctx->lock);
+}