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 "block/thread-pool.h"
  25 #include "qemu/main-loop.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     AioContext *ctx;
  60     QEMUBH *completion_bh;
  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         qemu_bh_schedule(pool->completion_bh);
 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", 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 thread_pool_completion_bh(void *opaque)
 171 {
 172     ThreadPool *pool = opaque;
 173     ThreadPoolElement *elem, *next;
 174
 175 restart:
 176     QLIST_FOREACH_SAFE(elem, &pool->head, all, next) {
 177         if (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
 178             continue;
 179         }
 180         if (elem->state == THREAD_DONE) {
 181             trace_thread_pool_complete(pool, elem, elem->common.opaque,
 182                                        elem->ret);
 183         }
 184         if (elem->state == THREAD_DONE && elem->common.cb) {
 185             QLIST_REMOVE(elem, all);
 186             /* Read state before ret.  */
 187             smp_rmb();
 188
 189             /* Schedule ourselves in case elem->common.cb() calls aio_poll() to
 190              * wait for another request that completed at the same time.
 191              */
 192             qemu_bh_schedule(pool->completion_bh);
 193
 194             elem->common.cb(elem->common.opaque, elem->ret);
 195             qemu_aio_release(elem);
 196             goto restart;
 197         } else {
 198             /* remove the request */
 199             QLIST_REMOVE(elem, all);
 200             qemu_aio_release(elem);
 201         }
 202     }
 203 }
 204
 205 static void thread_pool_cancel(BlockDriverAIOCB *acb)
 206 {
 207     ThreadPoolElement *elem = (ThreadPoolElement *)acb;
 208     ThreadPool *pool = elem->pool;
 209
 210     trace_thread_pool_cancel(elem, elem->common.opaque);
 211
 212     qemu_mutex_lock(&pool->lock);
 213     if (elem->state == THREAD_QUEUED &&
 214         /* No thread has yet started working on elem. we can try to "steal"
 215          * the item from the worker if we can get a signal from the
 216          * semaphore.  Because this is non-blocking, we can do it with
 217          * the lock taken and ensure that elem will remain THREAD_QUEUED.
 218          */
 219         qemu_sem_timedwait(&pool->sem, 0) == 0) {
 220         QTAILQ_REMOVE(&pool->request_list, elem, reqs);
 221         elem->state = THREAD_CANCELED;
 222         qemu_bh_schedule(pool->completion_bh);
 223     } else {
 224         pool->pending_cancellations++;
 225         while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
 226             qemu_cond_wait(&pool->check_cancel, &pool->lock);
 227         }
 228         pool->pending_cancellations--;
 229     }
 230     qemu_mutex_unlock(&pool->lock);
 231     thread_pool_completion_bh(pool);
 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     pool->ctx = ctx;
 301     pool->completion_bh = aio_bh_new(ctx, thread_pool_completion_bh, pool);
 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
 313 ThreadPool *thread_pool_new(AioContext *ctx)
 314 {
 315     ThreadPool *pool = g_new(ThreadPool, 1);
 316     thread_pool_init_one(pool, ctx);
 317     return pool;
 318 }
 319
 320 void thread_pool_free(ThreadPool *pool)
 321 {
 322     if (!pool) {
 323         return;
 324     }
 325
 326     assert(QLIST_EMPTY(&pool->head));
 327
 328     qemu_mutex_lock(&pool->lock);
 329
 330     /* Stop new threads from spawning */
 331     qemu_bh_delete(pool->new_thread_bh);
 332     pool->cur_threads -= pool->new_threads;
 333     pool->new_threads = 0;
 334
 335     /* Wait for worker threads to terminate */
 336     pool->stopping = true;
 337     while (pool->cur_threads > 0) {
 338         qemu_sem_post(&pool->sem);
 339         qemu_cond_wait(&pool->worker_stopped, &pool->lock);
 340     }
 341
 342     qemu_mutex_unlock(&pool->lock);
 343
 344     qemu_bh_delete(pool->completion_bh);
 345     qemu_sem_destroy(&pool->sem);
 346     qemu_cond_destroy(&pool->check_cancel);
 347     qemu_cond_destroy(&pool->worker_stopped);
 348     qemu_mutex_destroy(&pool->lock);
 349     g_free(pool);
 350 }