/* * QEMU block layer thread pool * * Copyright IBM, Corp. 2008 * Copyright Red Hat, Inc. 2012 * * Authors: * Anthony Liguori <aliguori@us.ibm.com> * Paolo Bonzini <pbonzini@redhat.com> * * This work is licensed under the terms of the GNU GPL, version 2. See * the COPYING file in the top-level directory. * * Contributions after 2012-01-13 are licensed under the terms of the * GNU GPL, version 2 or (at your option) any later version. */ #include "qemu-common.h" #include "qemu/queue.h" #include "qemu/thread.h" #include "qemu/osdep.h" #include "block/coroutine.h" #include "trace.h" #include "block/thread-pool.h" #include "qemu/main-loop.h" static void do_spawn_thread(ThreadPool *pool); typedef struct ThreadPoolElement ThreadPoolElement; enum ThreadState { THREAD_QUEUED, THREAD_ACTIVE, THREAD_DONE, }; struct ThreadPoolElement { BlockAIOCB common; ThreadPool *pool; ThreadPoolFunc *func; void *arg; /* Moving state out of THREAD_QUEUED is protected by lock. After * that, only the worker thread can write to it. Reads and writes * of state and ret are ordered with memory barriers. */ enum ThreadState state; int ret; /* Access to this list is protected by lock. */ QTAILQ_ENTRY(ThreadPoolElement) reqs; /* Access to this list is protected by the global mutex. */ QLIST_ENTRY(ThreadPoolElement) all; }; struct ThreadPool { AioContext *ctx; QEMUBH *completion_bh; QemuMutex lock; QemuCond worker_stopped; QemuSemaphore sem; int max_threads; QEMUBH *new_thread_bh; /* The following variables are only accessed from one AioContext. */ QLIST_HEAD(, ThreadPoolElement) head; /* The following variables are protected by lock. */ QTAILQ_HEAD(, ThreadPoolElement) request_list; int cur_threads; int idle_threads; int new_threads; /* backlog of threads we need to create */ int pending_threads; /* threads created but not running yet */ bool stopping; }; static void *worker_thread(void *opaque) { ThreadPool *pool = opaque; qemu_mutex_lock(&pool->lock); pool->pending_threads--; do_spawn_thread(pool); while (!pool->stopping) { ThreadPoolElement *req; int ret; do { pool->idle_threads++; qemu_mutex_unlock(&pool->lock); ret = qemu_sem_timedwait(&pool->sem, 10000); qemu_mutex_lock(&pool->lock); pool->idle_threads--; } while (ret == -1 && !QTAILQ_EMPTY(&pool->request_list)); if (ret == -1 || pool->stopping) { break; } req = QTAILQ_FIRST(&pool->request_list); QTAILQ_REMOVE(&pool->request_list, req, reqs); req->state = THREAD_ACTIVE; qemu_mutex_unlock(&pool->lock); ret = req->func(req->arg); req->ret = ret; /* Write ret before state. */ smp_wmb(); req->state = THREAD_DONE; qemu_mutex_lock(&pool->lock); qemu_bh_schedule(pool->completion_bh); } pool->cur_threads--; qemu_cond_signal(&pool->worker_stopped); qemu_mutex_unlock(&pool->lock); return NULL; } static void do_spawn_thread(ThreadPool *pool) { QemuThread t; /* Runs with lock taken. */ if (!pool->new_threads) { return; } pool->new_threads--; pool->pending_threads++; qemu_thread_create(&t, "worker", worker_thread, pool, QEMU_THREAD_DETACHED); } static void spawn_thread_bh_fn(void *opaque) { ThreadPool *pool = opaque; qemu_mutex_lock(&pool->lock); do_spawn_thread(pool); qemu_mutex_unlock(&pool->lock); } static void spawn_thread(ThreadPool *pool) { pool->cur_threads++; pool->new_threads++; /* If there are threads being created, they will spawn new workers, so * we don't spend time creating many threads in a loop holding a mutex or * starving the current vcpu. * * If there are no idle threads, ask the main thread to create one, so we * inherit the correct affinity instead of the vcpu affinity. */ if (!pool->pending_threads) { qemu_bh_schedule(pool->new_thread_bh); } } static void thread_pool_completion_bh(void *opaque) { ThreadPool *pool = opaque; ThreadPoolElement *elem, *next; restart: QLIST_FOREACH_SAFE(elem, &pool->head, all, next) { if (elem->state != THREAD_DONE) { continue; } if (elem->state == THREAD_DONE) { trace_thread_pool_complete(pool, elem, elem->common.opaque, elem->ret); } if (elem->state == THREAD_DONE && elem->common.cb) { 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_unref(elem); goto restart; } else { /* remove the request */ QLIST_REMOVE(elem, all); qemu_aio_unref(elem); } } } static void thread_pool_cancel(BlockAIOCB *acb) { ThreadPoolElement *elem = (ThreadPoolElement *)acb; ThreadPool *pool = elem->pool; trace_thread_pool_cancel(elem, elem->common.opaque); qemu_mutex_lock(&pool->lock); if (elem->state == THREAD_QUEUED && /* No thread has yet started working on elem. we can try to "steal" * the item from the worker if we can get a signal from the * semaphore. Because this is non-blocking, we can do it with * the lock taken and ensure that elem will remain THREAD_QUEUED. */ qemu_sem_timedwait(&pool->sem, 0) == 0) { QTAILQ_REMOVE(&pool->request_list, elem, reqs); qemu_bh_schedule(pool->completion_bh); elem->state = THREAD_DONE; elem->ret = -ECANCELED; } qemu_mutex_unlock(&pool->lock); } static AioContext *thread_pool_get_aio_context(BlockAIOCB *acb) { ThreadPoolElement *elem = (ThreadPoolElement *)acb; ThreadPool *pool = elem->pool; return pool->ctx; } static const AIOCBInfo thread_pool_aiocb_info = { .aiocb_size = sizeof(ThreadPoolElement), .cancel_async = thread_pool_cancel, .get_aio_context = thread_pool_get_aio_context, }; BlockAIOCB *thread_pool_submit_aio(ThreadPool *pool, ThreadPoolFunc *func, void *arg, BlockCompletionFunc *cb, void *opaque) { ThreadPoolElement *req; req = qemu_aio_get(&thread_pool_aiocb_info, NULL, cb, opaque); req->func = func; req->arg = arg; req->state = THREAD_QUEUED; req->pool = pool; QLIST_INSERT_HEAD(&pool->head, req, all); trace_thread_pool_submit(pool, req, arg); qemu_mutex_lock(&pool->lock); if (pool->idle_threads == 0 && pool->cur_threads < pool->max_threads) { spawn_thread(pool); } QTAILQ_INSERT_TAIL(&pool->request_list, req, reqs); qemu_mutex_unlock(&pool->lock); qemu_sem_post(&pool->sem); return &req->common; } typedef struct ThreadPoolCo { Coroutine *co; int ret; } ThreadPoolCo; static void thread_pool_co_cb(void *opaque, int ret) { ThreadPoolCo *co = opaque; co->ret = ret; qemu_coroutine_enter(co->co, NULL); } 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(pool, func, arg, thread_pool_co_cb, &tpc); qemu_coroutine_yield(); return tpc.ret; } void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg) { thread_pool_submit_aio(pool, func, arg, NULL, NULL); } static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx) { if (!ctx) { ctx = qemu_get_aio_context(); } memset(pool, 0, sizeof(*pool)); pool->ctx = ctx; pool->completion_bh = aio_bh_new(ctx, thread_pool_completion_bh, pool); qemu_mutex_init(&pool->lock); qemu_cond_init(&pool->worker_stopped); qemu_sem_init(&pool->sem, 0); pool->max_threads = 64; pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool); QLIST_INIT(&pool->head); QTAILQ_INIT(&pool->request_list); } ThreadPool *thread_pool_new(AioContext *ctx) { ThreadPool *pool = g_new(ThreadPool, 1); thread_pool_init_one(pool, ctx); return pool; } void thread_pool_free(ThreadPool *pool) { if (!pool) { return; } assert(QLIST_EMPTY(&pool->head)); qemu_mutex_lock(&pool->lock); /* Stop new threads from spawning */ qemu_bh_delete(pool->new_thread_bh); pool->cur_threads -= pool->new_threads; pool->new_threads = 0; /* Wait for worker threads to terminate */ pool->stopping = true; while (pool->cur_threads > 0) { qemu_sem_post(&pool->sem); qemu_cond_wait(&pool->worker_stopped, &pool->lock); } qemu_mutex_unlock(&pool->lock); qemu_bh_delete(pool->completion_bh); qemu_sem_destroy(&pool->sem); qemu_cond_destroy(&pool->worker_stopped); qemu_mutex_destroy(&pool->lock); g_free(pool); }