thread-pool.c

   1 /*
   2  * QEMU block layer thread pool
   3  *
   4  * Copyright IBM, Corp. 2008
   5  * Copyright Red Hat, Inc. 2012
   6  *
   7  * Authors:
   8  *  Anthony Liguori   <aliguori@us.ibm.com>
   9  *  Paolo Bonzini     <pbonzini@redhat.com>
  10  *
  11  * This work is licensed under the terms of the GNU GPL, version 2.  See
  12  * the COPYING file in the top-level directory.
  13  *
  14  * Contributions after 2012-01-13 are licensed under the terms of the
  15  * GNU GPL, version 2 or (at your option) any later version.
  16  */
  17 #include "qemu-common.h"
  18 #include "qemu-queue.h"
  19 #include "qemu-thread.h"
  20 #include "osdep.h"
  21 #include "qemu-coroutine.h"
  22 #include "trace.h"
  23 #include "block_int.h"
  24 #include "event_notifier.h"
  25 #include "thread-pool.h"
  26
  27 static void do_spawn_thread(void);
  28
  29 typedef struct ThreadPoolElement ThreadPoolElement;
  30
  31 enum ThreadState {
  32     THREAD_QUEUED,
  33     THREAD_ACTIVE,
  34     THREAD_DONE,
  35     THREAD_CANCELED,
  36 };
  37
  38 struct ThreadPoolElement {
  39     BlockDriverAIOCB common;
  40     ThreadPoolFunc *func;
  41     void *arg;
  42     enum ThreadState state;
  43     int ret;
  44
  45     /* Access to this list is protected by lock.  */
  46     QTAILQ_ENTRY(ThreadPoolElement) reqs;
  47
  48     /* Access to this list is protected by the global mutex.  */
  49     QLIST_ENTRY(ThreadPoolElement) all;
  50 };
  51
  52 static EventNotifier notifier;
  53 static QemuMutex lock;
  54 static QemuCond check_cancel;
  55 static QemuSemaphore sem;
  56 static int max_threads = 64;
  57 static QEMUBH *new_thread_bh;
  58
  59 /* The following variables are protected by the global mutex.  */
  60 static QLIST_HEAD(, ThreadPoolElement) head;
  61
  62 /* The following variables are protected by lock.  */
  63 static QTAILQ_HEAD(, ThreadPoolElement) request_list;
  64 static int cur_threads;
  65 static int idle_threads;
  66 static int new_threads;     /* backlog of threads we need to create */
  67 static int pending_threads; /* threads created but not running yet */
  68 static int pending_cancellations; /* whether we need a cond_broadcast */
  69
  70 static void *worker_thread(void *unused)
  71 {
  72     qemu_mutex_lock(&lock);
  73     pending_threads--;
  74     do_spawn_thread();
  75
  76     while (1) {
  77         ThreadPoolElement *req;
  78         int ret;
  79
  80         do {
  81             idle_threads++;
  82             qemu_mutex_unlock(&lock);
  83             ret = qemu_sem_timedwait(&sem, 10000);
  84             qemu_mutex_lock(&lock);
  85             idle_threads--;
  86         } while (ret == -1 && !QTAILQ_EMPTY(&request_list));
  87         if (ret == -1) {
  88             break;
  89         }
  90
  91         req = QTAILQ_FIRST(&request_list);
  92         QTAILQ_REMOVE(&request_list, req, reqs);
  93         req->state = THREAD_ACTIVE;
  94         qemu_mutex_unlock(&lock);
  95
  96         ret = req->func(req->arg);
  97
  98         qemu_mutex_lock(&lock);
  99         req->state = THREAD_DONE;
 100         req->ret = ret;
 101         if (pending_cancellations) {
 102             qemu_cond_broadcast(&check_cancel);
 103         }
 104
 105         event_notifier_set(&notifier);
 106     }
 107
 108     cur_threads--;
 109     qemu_mutex_unlock(&lock);
 110     return NULL;
 111 }
 112
 113 static void do_spawn_thread(void)
 114 {
 115     QemuThread t;
 116
 117     /* Runs with lock taken.  */
 118     if (!new_threads) {
 119         return;
 120     }
 121
 122     new_threads--;
 123     pending_threads++;
 124
 125     qemu_thread_create(&t, worker_thread, NULL, QEMU_THREAD_DETACHED);
 126 }
 127
 128 static void spawn_thread_bh_fn(void *opaque)
 129 {
 130     qemu_mutex_lock(&lock);
 131     do_spawn_thread();
 132     qemu_mutex_unlock(&lock);
 133 }
 134
 135 static void spawn_thread(void)
 136 {
 137     cur_threads++;
 138     new_threads++;
 139     /* If there are threads being created, they will spawn new workers, so
 140      * we don't spend time creating many threads in a loop holding a mutex or
 141      * starving the current vcpu.
 142      *
 143      * If there are no idle threads, ask the main thread to create one, so we
 144      * inherit the correct affinity instead of the vcpu affinity.
 145      */
 146     if (!pending_threads) {
 147         qemu_bh_schedule(new_thread_bh);
 148     }
 149 }
 150
 151 static void event_notifier_ready(EventNotifier *notifier)
 152 {
 153     ThreadPoolElement *elem, *next;
 154
 155     event_notifier_test_and_clear(notifier);
 156 restart:
 157     QLIST_FOREACH_SAFE(elem, &head, all, next) {
 158         if (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
 159             continue;
 160         }
 161         if (elem->state == THREAD_DONE) {
 162             trace_thread_pool_complete(elem, elem->common.opaque, elem->ret);
 163         }
 164         if (elem->state == THREAD_DONE && elem->common.cb) {
 165             qemu_mutex_lock(&lock);
 166             int ret = elem->ret;
 167             qemu_mutex_unlock(&lock);
 168             QLIST_REMOVE(elem, all);
 169             elem->common.cb(elem->common.opaque, ret);
 170             qemu_aio_release(elem);
 171             goto restart;
 172         } else {
 173             /* remove the request */
 174             QLIST_REMOVE(elem, all);
 175             qemu_aio_release(elem);
 176         }
 177     }
 178 }
 179
 180 static int thread_pool_active(EventNotifier *notifier)
 181 {
 182     return !QLIST_EMPTY(&head);
 183 }
 184
 185 static void thread_pool_cancel(BlockDriverAIOCB *acb)
 186 {
 187     ThreadPoolElement *elem = (ThreadPoolElement *)acb;
 188
 189     trace_thread_pool_cancel(elem, elem->common.opaque);
 190
 191     qemu_mutex_lock(&lock);
 192     if (elem->state == THREAD_QUEUED &&
 193         /* No thread has yet started working on elem. we can try to "steal"
 194          * the item from the worker if we can get a signal from the
 195          * semaphore.  Because this is non-blocking, we can do it with
 196          * the lock taken and ensure that elem will remain THREAD_QUEUED.
 197          */
 198         qemu_sem_timedwait(&sem, 0) == 0) {
 199         QTAILQ_REMOVE(&request_list, elem, reqs);
 200         elem->state = THREAD_CANCELED;
 201         event_notifier_set(&notifier);
 202     } else {
 203         pending_cancellations++;
 204         while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
 205             qemu_cond_wait(&check_cancel, &lock);
 206         }
 207         pending_cancellations--;
 208     }
 209     qemu_mutex_unlock(&lock);
 210 }
 211
 212 static AIOPool thread_pool_cb_pool = {
 213     .aiocb_size         = sizeof(ThreadPoolElement),
 214     .cancel             = thread_pool_cancel,
 215 };
 216
 217 BlockDriverAIOCB *thread_pool_submit_aio(ThreadPoolFunc *func, void *arg,
 218         BlockDriverCompletionFunc *cb, void *opaque)
 219 {
 220     ThreadPoolElement *req;
 221
 222     req = qemu_aio_get(&thread_pool_cb_pool, NULL, cb, opaque);
 223     req->func = func;
 224     req->arg = arg;
 225     req->state = THREAD_QUEUED;
 226
 227     QLIST_INSERT_HEAD(&head, req, all);
 228
 229     trace_thread_pool_submit(req, arg);
 230
 231     qemu_mutex_lock(&lock);
 232     if (idle_threads == 0 && cur_threads < max_threads) {
 233         spawn_thread();
 234     }
 235     QTAILQ_INSERT_TAIL(&request_list, req, reqs);
 236     qemu_mutex_unlock(&lock);
 237     qemu_sem_post(&sem);
 238     return &req->common;
 239 }
 240
 241 typedef struct ThreadPoolCo {
 242     Coroutine *co;
 243     int ret;
 244 } ThreadPoolCo;
 245
 246 static void thread_pool_co_cb(void *opaque, int ret)
 247 {
 248     ThreadPoolCo *co = opaque;
 249
 250     co->ret = ret;
 251     qemu_coroutine_enter(co->co, NULL);
 252 }
 253
 254 int coroutine_fn thread_pool_submit_co(ThreadPoolFunc *func, void *arg)
 255 {
 256     ThreadPoolCo tpc = { .co = qemu_coroutine_self(), .ret = -EINPROGRESS };
 257     assert(qemu_in_coroutine());
 258     thread_pool_submit_aio(func, arg, thread_pool_co_cb, &tpc);
 259     qemu_coroutine_yield();
 260     return tpc.ret;
 261 }
 262
 263 void thread_pool_submit(ThreadPoolFunc *func, void *arg)
 264 {
 265     thread_pool_submit_aio(func, arg, NULL, NULL);
 266 }
 267
 268 static void thread_pool_init(void)
 269 {
 270     QLIST_INIT(&head);
 271     event_notifier_init(&notifier, false);
 272     qemu_mutex_init(&lock);
 273     qemu_cond_init(&check_cancel);
 274     qemu_sem_init(&sem, 0);
 275     qemu_aio_set_event_notifier(&notifier, event_notifier_ready,
 276                                 thread_pool_active);
 277
 278     QTAILQ_INIT(&request_list);
 279     new_thread_bh = qemu_bh_new(spawn_thread_bh_fn, NULL);
 280 }
 281
 282 block_init(thread_pool_init)