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 int thread_pool_active(EventNotifier *notifier)
 201 {
 202     ThreadPool *pool = container_of(notifier, ThreadPool, notifier);
 203     return !QLIST_EMPTY(&pool->head);
 204 }
 205
 206 static void thread_pool_cancel(BlockDriverAIOCB *acb)
 207 {
 208     ThreadPoolElement *elem = (ThreadPoolElement *)acb;
 209     ThreadPool *pool = elem->pool;
 210
 211     trace_thread_pool_cancel(elem, elem->common.opaque);
 212
 213     qemu_mutex_lock(&pool->lock);
 214     if (elem->state == THREAD_QUEUED &&
 215         /* No thread has yet started working on elem. we can try to "steal"
 216          * the item from the worker if we can get a signal from the
 217          * semaphore.  Because this is non-blocking, we can do it with
 218          * the lock taken and ensure that elem will remain THREAD_QUEUED.
 219          */
 220         qemu_sem_timedwait(&pool->sem, 0) == 0) {
 221         QTAILQ_REMOVE(&pool->request_list, elem, reqs);
 222         elem->state = THREAD_CANCELED;
 223         event_notifier_set(&pool->notifier);
 224     } else {
 225         pool->pending_cancellations++;
 226         while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
 227             qemu_cond_wait(&pool->check_cancel, &pool->lock);
 228         }
 229         pool->pending_cancellations--;
 230     }
 231     qemu_mutex_unlock(&pool->lock);
 232 }
 233
 234 static const AIOCBInfo thread_pool_aiocb_info = {
 235     .aiocb_size         = sizeof(ThreadPoolElement),
 236     .cancel             = thread_pool_cancel,
 237 };
 238
 239 BlockDriverAIOCB *thread_pool_submit_aio(ThreadPool *pool,
 240         ThreadPoolFunc *func, void *arg,
 241         BlockDriverCompletionFunc *cb, void *opaque)
 242 {
 243     ThreadPoolElement *req;
 244
 245     req = qemu_aio_get(&thread_pool_aiocb_info, NULL, cb, opaque);
 246     req->func = func;
 247     req->arg = arg;
 248     req->state = THREAD_QUEUED;
 249     req->pool = pool;
 250
 251     QLIST_INSERT_HEAD(&pool->head, req, all);
 252
 253     trace_thread_pool_submit(pool, req, arg);
 254
 255     qemu_mutex_lock(&pool->lock);
 256     if (pool->idle_threads == 0 && pool->cur_threads < pool->max_threads) {
 257         spawn_thread(pool);
 258     }
 259     QTAILQ_INSERT_TAIL(&pool->request_list, req, reqs);
 260     qemu_mutex_unlock(&pool->lock);
 261     qemu_sem_post(&pool->sem);
 262     return &req->common;
 263 }
 264
 265 typedef struct ThreadPoolCo {
 266     Coroutine *co;
 267     int ret;
 268 } ThreadPoolCo;
 269
 270 static void thread_pool_co_cb(void *opaque, int ret)
 271 {
 272     ThreadPoolCo *co = opaque;
 273
 274     co->ret = ret;
 275     qemu_coroutine_enter(co->co, NULL);
 276 }
 277
 278 int coroutine_fn thread_pool_submit_co(ThreadPool *pool, ThreadPoolFunc *func,
 279                                        void *arg)
 280 {
 281     ThreadPoolCo tpc = { .co = qemu_coroutine_self(), .ret = -EINPROGRESS };
 282     assert(qemu_in_coroutine());
 283     thread_pool_submit_aio(pool, func, arg, thread_pool_co_cb, &tpc);
 284     qemu_coroutine_yield();
 285     return tpc.ret;
 286 }
 287
 288 void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg)
 289 {
 290     thread_pool_submit_aio(pool, func, arg, NULL, NULL);
 291 }
 292
 293 static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
 294 {
 295     if (!ctx) {
 296         ctx = qemu_get_aio_context();
 297     }
 298
 299     memset(pool, 0, sizeof(*pool));
 300     event_notifier_init(&pool->notifier, false);
 301     pool->ctx = ctx;
 302     qemu_mutex_init(&pool->lock);
 303     qemu_cond_init(&pool->check_cancel);
 304     qemu_cond_init(&pool->worker_stopped);
 305     qemu_sem_init(&pool->sem, 0);
 306     pool->max_threads = 64;
 307     pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);
 308
 309     QLIST_INIT(&pool->head);
 310     QTAILQ_INIT(&pool->request_list);
 311
 312     aio_set_event_notifier(ctx, &pool->notifier, event_notifier_ready,
 313                            thread_pool_active);
 314 }
 315
 316 ThreadPool *thread_pool_new(AioContext *ctx)
 317 {
 318     ThreadPool *pool = g_new(ThreadPool, 1);
 319     thread_pool_init_one(pool, ctx);
 320     return pool;
 321 }
 322
 323 void thread_pool_free(ThreadPool *pool)
 324 {
 325     if (!pool) {
 326         return;
 327     }
 328
 329     assert(QLIST_EMPTY(&pool->head));
 330
 331     qemu_mutex_lock(&pool->lock);
 332
 333     /* Stop new threads from spawning */
 334     qemu_bh_delete(pool->new_thread_bh);
 335     pool->cur_threads -= pool->new_threads;
 336     pool->new_threads = 0;
 337
 338     /* Wait for worker threads to terminate */
 339     pool->stopping = true;
 340     while (pool->cur_threads > 0) {
 341         qemu_sem_post(&pool->sem);
 342         qemu_cond_wait(&pool->worker_stopped, &pool->lock);
 343     }
 344
 345     qemu_mutex_unlock(&pool->lock);
 346
 347     aio_set_event_notifier(pool->ctx, &pool->notifier, NULL, NULL);
 348     qemu_sem_destroy(&pool->sem);
 349     qemu_cond_destroy(&pool->check_cancel);
 350     qemu_cond_destroy(&pool->worker_stopped);
 351     qemu_mutex_destroy(&pool->lock);
 352     event_notifier_cleanup(&pool->notifier);
 353     g_free(pool);
 354 }