/* * QEMU block layer thread pool * * Copyright IBM, Corp. 2008 * Copyright Red Hat, Inc. 2012 * * Authors: * Anthony Liguori * Paolo Bonzini * * 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/osdep.h" #include "qemu-common.h" #include "qemu/queue.h" #include "qemu/thread.h" #include "qemu/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; } trace_thread_pool_complete(pool, elem, elem->common.opaque, elem->ret); QLIST_REMOVE(elem, all); if (elem->common.cb) { /* 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 { qemu_aio_unref(elem); } } } static void thread_pool_cancel(BlockAI