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 "qemu/osdep.h"
  21 #include "block/coroutine.h"
  22 #include "trace.h"
  23 #include "block/block_int.h"
  24 #include "qemu/event_notifier.h"
  25 #include "block/thread-pool.h"
  26
  27 static void do_spawn_thread(ThreadPool *pool);
  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     ThreadPool *pool;
  41     ThreadPoolFunc *func;
  42     void *arg;
  43
  44     /* Moving state out of THREAD_QUEUED is protected by lock.  After
  45      * that, only the worker thread can write to it.  Reads and writes
  46      * of state and ret are ordered with memory barriers.
  47      */
  48     enum ThreadState state;
  49     int ret;
  50
  51     /* Access to this list is protected by lock.  */
  52     QTAILQ_ENTRY(ThreadPoolElement) reqs;
  53
  54     /* Access to this list is protected by the global mutex.  */
  55     QLIST_ENTRY(ThreadPoolElement) all;
  56 };
  57
  58 struct ThreadPool {
  59     EventNotifier notifier;
  60     AioContext *ctx;
  61     QemuMutex lock;
  62     QemuCond check_cancel;
  63     QemuCond worker_stopped;
  64     QemuSemaphore sem;
  65     int max_threads;
  66     QEMUBH *new_thread_bh;
  67
  68     /* The following variables are only accessed from one AioContext. */
  69     QLIST_HEAD(, ThreadPoolElement) head;
  70
  71     /* The following variables are protected by lock.  */
  72     QTAILQ_HEAD(, ThreadPoolElement) request_list;
  73     int cur_threads;
  74     int idle_threads;
  75     int new_threads;     /* backlog of threads we need to create */
  76     int pending_threads; /* threads created but not running yet */
  77     int pending_cancellations; /* whether we need a cond_broadcast */
  78     bool stopping;
  79 };
  80
  81 static void *worker_thread(void *opaque)
  82 {
  83     ThreadPool *pool = opaque;
  84
  85     qemu_mutex_lock(&pool->lock);
  86     pool->pending_threads--;
  87     do_spawn_thread(pool);
  88
  89     while (!pool->stopping) {
  90         ThreadPoolElement *req;
  91         int ret;
  92
  93         do {
  94             pool->idle_threads++;
  95             qemu_mutex_unlock(&pool->lock);
  96             ret = qemu_sem_timedwait(&pool->sem, 10000);
  97             qemu_mutex_lock(&pool->lock);
  98             pool->idle_threads--;
  99         } while (ret == -1 && !QTAILQ_EMPTY(&pool->request_list));
 100         if (ret == -1 || pool->stopping) {
 101             break;
 102         }
 103
 104         req = QTAILQ_FIRST(&pool->request_list);
 105         QTAILQ_REMOVE(&pool->request_list, req, reqs);
 106         req->state = THREAD_ACTIVE;
 107         qemu_mutex_unlock(&pool->lock);
 108
 109         ret = req->func(req->arg);
 110
 111         req->ret = ret;
 112         /* Write ret before state.  */
 113         smp_wmb();
 114         req->state = THREAD_DONE;
 115
 116         qemu_mutex_lock(&pool->lock);
 117         if (pool->pending_cancellations) {
 118             qemu_cond_broadcast(&pool->check_cancel);
 119         }
 120
 121         event_notifier_set(&pool->notifier);
 122     }
 123
 124     pool->cur_threads--;
 125     qemu_cond_signal(&pool->worker_stopped);
 126     qemu_mutex_unlock(&pool->lock);
 127     return NULL;
 128 }
 129
 130 static void do_spawn_thread(ThreadPool *pool)
 131 {
 132     QemuThread t;
 133
 134     /* Runs with lock taken.  */
 135     if (!pool->new_threads) {
 136         return;
 137     }
 138
 139     pool->new_threads--;
 140     pool->pending_threads++;
 141
 142     qemu_thread_create(&t, worker_thread, pool, QEMU_THREAD_DETACHED);
 143 }
 144
 145 static void spawn_thread_bh_fn(void *opaque)
 146 {
 147     ThreadPool *pool = opaque;
 148
 149     qemu_mutex_lock(&pool->lock);
 150     do_spawn_thread(pool);
 151     qemu_mutex_unlock(&pool->lock);
 152 }
 153
 154 static void spawn_thread(ThreadPool *pool)
 155 {
 156     pool->cur_threads++;
 157     pool->new_threads++;
 158     /* If there are threads being created, they will spawn new workers, so
 159      * we don't spend time creating many threads in a loop holding a mutex or
 160      * starving the current vcpu.
 161      *
 162      * If there are no idle threads, ask the main thread to create one, so we
 163      * inherit the correct affinity instead of the vcpu affinity.
 164      */
 165     if (!pool->pending_threads) {
 166         qemu_bh_schedule(pool->new_thread_bh);
 167     }
 168 }
 169
 170 static void event_notifier_ready(EventNotifier *notifier)
 171 {
 172     ThreadPool *pool = container_of(notifier, ThreadPool, notifier);
 173     ThreadPoolElement *elem, *next;
 174
 175     event_notifier_test_and_clear(notifier);
 176 restart:
 177     QLIST_FOREACH_SAFE(elem, &pool->head, all, next) {
 178         if (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
 179             continue;
 180         }
 181         if (elem->state == THREAD_DONE) {
 182             trace_thread_pool_complete(pool, elem, elem->common.opaque,
 183                                        elem->ret);
 184         }
 185         if (elem->state == THREAD_DONE && elem->common.cb) {
 186             QLIST_REMOVE(elem, all);
 187             /* Read state before ret.  */
 188             smp_rmb();
 189             elem->common.cb(elem->common.opaque, elem->ret);
 190             qemu_aio_release(elem);
 191             goto restart;
 192         } else {
 193             /* remove the request */
 194             QLIST_REMOVE(elem, all);
 195             qemu_aio_release(elem);
 196         }
 197     }
 198 }
 199
 200 static void thread_pool_cancel(BlockDriverAIOCB *acb)
 201 {
 202     ThreadPoolElement *elem = (ThreadPoolElement *)acb;
 203     ThreadPool *pool = elem->pool;
 204
 205     trace_thread_pool_cancel(elem, elem->common.opaque);
 206
 207     qemu_mutex_lock(&pool->lock);
 208     if (elem->state == THREAD_QUEUED &&
 209         /* No thread has yet started working on elem. we can try to "steal"
 210          * the item from the worker if we can get a signal from the
 211          * semaphore.  Because this is non-blocking, we can do it with
 212          * the lock taken and ensure that elem will remain THREAD_QUEUED.
 213          */
 214         qemu_sem_timedwait(&pool->sem, 0) == 0) {
 215         QTAILQ_REMOVE(&pool->request_list, elem, reqs);
 216         elem->state = THREAD_CANCELED;
 217         event_notifier_set(&pool->notifier);
 218     } else {
 219         pool->pending_cancellations++;
 220         while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
 221             qemu_cond_wait(&pool->check_cancel, &pool->lock);
 222         }
 223         pool->pending_cancellations--;
 224     }
 225     qemu_mutex_unlock(&pool->lock);
 226 }
 227
 228 static const AIOCBInfo thread_pool_aiocb_info = {
 229     .aiocb_size         = sizeof(ThreadPoolElement),
 230     .cancel             = thread_pool_cancel,
 231 };
 232
 233 BlockDriverAIOCB *thread_pool_submit_aio(ThreadPool *pool,
 234         ThreadPoolFunc *func, void *arg,
 235         BlockDriverCompletionFunc *cb, void *opaque)
 236 {
 237     ThreadPoolElement *req;
 238
 239     req = qemu_aio_get(&thread_pool_aiocb_info, NULL, cb, opaque);
 240     req->func = func;
 241     req->arg = arg;
 242     req->state = THREAD_QUEUED;
 243     req->pool = pool;
 244
 245     QLIST_INSERT_HEAD(&pool->head, req, all);
 246
 247     trace_thread_pool_submit(pool, req, arg);
 248
 249     qemu_mutex_lock(&pool->lock);
 250     if (pool->idle_threads == 0 && pool->cur_threads < pool->max_threads) {
 251         spawn_thread(pool);
 252     }
 253     QTAILQ_INSERT_TAIL(&pool->request_list, req, reqs);
 254     qemu_mutex_unlock(&pool->lock);
 255     qemu_sem_post(&pool->sem);
 256     return &req->common;
 257 }
 258
 259 typedef struct ThreadPoolCo {
 260     Coroutine *co;
 261     int ret;
 262 } ThreadPoolCo;
 263
 264 static void thread_pool_co_cb(void *opaque, int ret)
 265 {
 266     ThreadPoolCo *co = opaque;
 267
 268     co->ret = ret;
 269     qemu_coroutine_enter(co->co, NULL);
 270 }
 271
 272 int coroutine_fn thread_pool_submit_co(ThreadPool *pool, ThreadPoolFunc *func,
 273                                        void *arg)
 274 {
 275     ThreadPoolCo tpc = { .co = qemu_coroutine_self(), .ret = -EINPROGRESS };
 276     assert(qemu_in_coroutine());
 277     thread_pool_submit_aio(pool, func, arg, thread_pool_co_cb, &tpc);
 278     qemu_coroutine_yield();
 279     return tpc.ret;
 280 }
 281
 282 void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg)
 283 {
 284     thread_pool_submit_aio(pool, func, arg, NULL, NULL);
 285 }
 286
 287 static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
 288 {
 289     if (!ctx) {
 290         ctx = qemu_get_aio_context();
 291     }
 292
 293     memset(pool, 0, sizeof(*pool));
 294     event_notifier_init(&pool->notifier, false);
 295     pool->ctx = ctx;
 296     qemu_mutex_init(&pool->lock);
 297     qemu_cond_init(&pool->check_cancel);
 298     qemu_cond_init(&pool->worker_stopped);
 299     qemu_sem_init(&pool->sem, 0);
 300     pool->max_threads = 64;
 301     pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);
 302
 303     QLIST_INIT(&pool->head);
 304     QTAILQ_INIT(&pool->request_list);
 305
 306     aio_set_event_notifier(ctx, &pool->notifier, event_notifier_ready);
 307 }
 308
 309 ThreadPool *thread_pool_new(AioContext *ctx)
 310 {
 311     ThreadPool *pool = g_new(ThreadPool, 1);
 312     thread_pool_init_one(pool, ctx);
 313     return pool;
 314 }
 315
 316 void thread_pool_free(ThreadPool *pool)
 317 {
 318     if (!pool) {
 319         return;
 320     }
 321
 322     assert(QLIST_EMPTY(&pool->head));
 323
 324     qemu_mutex_lock(&pool->lock);
 325
 326     /* Stop new threads from spawning */
 327     qemu_bh_delete(pool->new_thread_bh);
 328     pool->cur_threads -= pool->new_threads;
 329     pool->new_threads = 0;
 330
 331     /* Wait for worker threads to terminate */
 332     pool->stopping = true;
 333     while (pool->cur_threads > 0) {
 334         qemu_sem_post(&pool->sem);
 335         qemu_cond_wait(&pool->worker_stopped, &pool->lock);
 336     }
 337
 338     qemu_mutex_unlock(&pool->lock);
 339
 340     aio_set_event_notifier(pool->ctx, &pool->notifier, NULL);
 341     qemu_sem_destroy(&pool->sem);
 342     qemu_cond_destroy(&pool->check_cancel);
 343     qemu_cond_destroy(&pool->worker_stopped);
 344     qemu_mutex_destroy(&pool->lock);
 345     event_notifier_cleanup(&pool->notifier);
 346     g_free(pool);
 347 }