vdi: Handle failure for potentially large allocations

[qemu.git] / thread-pool.c

diff --git a/thread-pool.c b/thread-pool.c
index d1e4570829fbbc0fd5d944fd29437533826f9f62..23888dcfc4d74d9c9343462271dd15cb6eaa926d 100644 (file)
--- a/thread-pool.c
+++ b/thread-pool.c
@@ -21,8 +21,8 @@
 #include "block/coroutine.h"
 #include "trace.h"
 #include "block/block_int.h"
-#include "qemu/event_notifier.h"
 #include "block/thread-pool.h"
+#include "qemu/main-loop.h"
 
 static void do_spawn_thread(ThreadPool *pool);
 
@@ -56,8 +56,8 @@ struct ThreadPoolElement {
 };
 
 struct ThreadPool {
-    EventNotifier notifier;
     AioContext *ctx;
+    QEMUBH *completion_bh;
     QemuMutex lock;
     QemuCond check_cancel;
     QemuCond worker_stopped;
@@ -78,9 +78,6 @@ struct ThreadPool {
     bool stopping;
 };
 
-/* Currently there is only one thread pool instance. */
-static ThreadPool global_pool;
-
 static void *worker_thread(void *opaque)
 {
     ThreadPool *pool = opaque;
@@ -121,7 +118,7 @@ static void *worker_thread(void *opaque)
             qemu_cond_broadcast(&pool->check_cancel);
         }
 
-        event_notifier_set(&pool->notifier);
+        qemu_bh_schedule(pool->completion_bh);
     }
 
     pool->cur_threads--;
@@ -142,7 +139,7 @@ static void do_spawn_thread(ThreadPool *pool)
     pool->new_threads--;
     pool->pending_threads++;
 
-    qemu_thread_create(&t, worker_thread, pool, QEMU_THREAD_DETACHED);
+    qemu_thread_create(&t, "worker", worker_thread, pool, QEMU_THREAD_DETACHED);
 }
 
 static void spawn_thread_bh_fn(void *opaque)
@@ -170,12 +167,11 @@ static void spawn_thread(ThreadPool *pool)
     }
 }
 
-static void event_notifier_ready(EventNotifier *notifier)
+static void thread_pool_completion_bh(void *opaque)
 {
-    ThreadPool *pool = container_of(notifier, ThreadPool, notifier);
+    ThreadPool *pool = opaque;
     ThreadPoolElement *elem, *next;
 
-    event_notifier_test_and_clear(notifier);
 restart:
     QLIST_FOREACH_SAFE(elem, &pool->head, all, next) {
         if (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
@@ -189,6 +185,12 @@ restart:
             QLIST_REMOVE(elem, all);
             /* Read state before ret.  */
             smp_rmb();
+
+            /* Schedule ourselves in case elem->common.cb() calls aio_poll() to
+             * wait for another request that completed at the same time.
+             */
+            qemu_bh_schedule(pool->completion_bh);
+
             elem->common.cb(elem->common.opaque, elem->ret);
             qemu_aio_release(elem);
             goto restart;
@@ -200,12 +202,6 @@ restart:
     }
 }
 
-static int thread_pool_active(EventNotifier *notifier)
-{
-    ThreadPool *pool = container_of(notifier, ThreadPool, notifier);
-    return !QLIST_EMPTY(&pool->head);
-}
-
 static void thread_pool_cancel(BlockDriverAIOCB *acb)
 {
     ThreadPoolElement *elem = (ThreadPoolElement *)acb;
@@ -223,7 +219,7 @@ static void thread_pool_cancel(BlockDriverAIOCB *acb)
         qemu_sem_timedwait(&pool->sem, 0) == 0) {
         QTAILQ_REMOVE(&pool->request_list, elem, reqs);
         elem->state = THREAD_CANCELED;
-        event_notifier_set(&pool->notifier);
+        qemu_bh_schedule(pool->completion_bh);
     } else {
         pool->pending_cancellations++;
         while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
@@ -232,6 +228,7 @@ static void thread_pool_cancel(BlockDriverAIOCB *acb)
         pool->pending_cancellations--;
     }
     qemu_mutex_unlock(&pool->lock);
+    thread_pool_completion_bh(pool);
 }
 
 static const AIOCBInfo thread_pool_aiocb_info = {
@@ -239,10 +236,10 @@ static const AIOCBInfo thread_pool_aiocb_info = {
     .cancel             = thread_pool_cancel,
 };
 
-BlockDriverAIOCB *thread_pool_submit_aio(ThreadPoolFunc *func, void *arg,
+BlockDriverAIOCB *thread_pool_submit_aio(ThreadPool *pool,
+        ThreadPoolFunc *func, void *arg,
         BlockDriverCompletionFunc *cb, void *opaque)
 {
-    ThreadPool *pool = &global_pool;
     ThreadPoolElement *req;
 
     req = qemu_aio_get(&thread_pool_aiocb_info, NULL, cb, opaque);
@@ -278,18 +275,19 @@ static void thread_pool_co_cb(void *opaque, int ret)
     qemu_coroutine_enter(co->co, NULL);
 }
 
-int coroutine_fn thread_pool_submit_co(ThreadPoolFunc *func, void *arg)
+int coroutine_fn thread_pool_submit_co(ThreadPool *pool, ThreadPoolFunc *func,
+                                       void *arg)
 {
     ThreadPoolCo tpc = { .co = qemu_coroutine_self(), .ret = -EINPROGRESS };
     assert(qemu_in_coroutine());
-    thread_pool_submit_aio(func, arg, thread_pool_co_cb, &tpc);
+    thread_pool_submit_aio(pool, func, arg, thread_pool_co_cb, &tpc);
     qemu_coroutine_yield();
     return tpc.ret;
 }
 
-void thread_pool_submit(ThreadPoolFunc *func, void *arg)
+void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg)
 {
-    thread_pool_submit_aio(func, arg, NULL, NULL);
+    thread_pool_submit_aio(pool, func, arg, NULL, NULL);
 }
 
 static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
@@ -299,8 +297,8 @@ static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
     }
 
     memset(pool, 0, sizeof(*pool));
-    event_notifier_init(&pool->notifier, false);
     pool->ctx = ctx;
+    pool->completion_bh = aio_bh_new(ctx, thread_pool_completion_bh, pool);
     qemu_mutex_init(&pool->lock);
     qemu_cond_init(&pool->check_cancel);
     qemu_cond_init(&pool->worker_stopped);
@@ -310,9 +308,6 @@ static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
 
     QLIST_INIT(&pool->head);
     QTAILQ_INIT(&pool->request_list);
-
-    aio_set_event_notifier(ctx, &pool->notifier, event_notifier_ready,
-                           thread_pool_active);
 }
 
 ThreadPool *thread_pool_new(AioContext *ctx)
@@ -346,18 +341,10 @@ void thread_pool_free(ThreadPool *pool)
 
     qemu_mutex_unlock(&pool->lock);
 
-    aio_set_event_notifier(pool->ctx, &pool->notifier, NULL, NULL);
+    qemu_bh_delete(pool->completion_bh);
     qemu_sem_destroy(&pool->sem);
     qemu_cond_destroy(&pool->check_cancel);
     qemu_cond_destroy(&pool->worker_stopped);
     qemu_mutex_destroy(&pool->lock);
-    event_notifier_cleanup(&pool->notifier);
     g_free(pool);
 }
-
-static void thread_pool_init(void)
-{
-    thread_pool_init_one(&global_pool, NULL);
-}
-
-block_init(thread_pool_init)