virtio-net: only delete bh that existed
[qemu/kevin.git] / thread-pool.c
blob3735fd34bc9ffc4a0a993897eb0632bf7b508f2e
1 /*
2 * QEMU block layer thread pool
4 * Copyright IBM, Corp. 2008
5 * Copyright Red Hat, Inc. 2012
7 * Authors:
8 * Anthony Liguori <aliguori@us.ibm.com>
9 * Paolo Bonzini <pbonzini@redhat.com>
11 * This work is licensed under the terms of the GNU GPL, version 2. See
12 * the COPYING file in the top-level directory.
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.
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 #include "qemu/main-loop.h"
28 static void do_spawn_thread(ThreadPool *pool);
30 typedef struct ThreadPoolElement ThreadPoolElement;
32 enum ThreadState {
33 THREAD_QUEUED,
34 THREAD_ACTIVE,
35 THREAD_DONE,
36 THREAD_CANCELED,
39 struct ThreadPoolElement {
40 BlockDriverAIOCB common;
41 ThreadPool *pool;
42 ThreadPoolFunc *func;
43 void *arg;
45 /* Moving state out of THREAD_QUEUED is protected by lock. After
46 * that, only the worker thread can write to it. Reads and writes
47 * of state and ret are ordered with memory barriers.
49 enum ThreadState state;
50 int ret;
52 /* Access to this list is protected by lock. */
53 QTAILQ_ENTRY(ThreadPoolElement) reqs;
55 /* Access to this list is protected by the global mutex. */
56 QLIST_ENTRY(ThreadPoolElement) all;
59 struct ThreadPool {
60 EventNotifier notifier;
61 AioContext *ctx;
62 QemuMutex lock;
63 QemuCond check_cancel;
64 QemuCond worker_stopped;
65 QemuSemaphore sem;
66 int max_threads;
67 QEMUBH *new_thread_bh;
69 /* The following variables are only accessed from one AioContext. */
70 QLIST_HEAD(, ThreadPoolElement) head;
72 /* The following variables are protected by lock. */
73 QTAILQ_HEAD(, ThreadPoolElement) request_list;
74 int cur_threads;
75 int idle_threads;
76 int new_threads; /* backlog of threads we need to create */
77 int pending_threads; /* threads created but not running yet */
78 int pending_cancellations; /* whether we need a cond_broadcast */
79 bool stopping;
82 static void *worker_thread(void *opaque)
84 ThreadPool *pool = opaque;
86 qemu_mutex_lock(&pool->lock);
87 pool->pending_threads--;
88 do_spawn_thread(pool);
90 while (!pool->stopping) {
91 ThreadPoolElement *req;
92 int ret;
94 do {
95 pool->idle_threads++;
96 qemu_mutex_unlock(&pool->lock);
97 ret = qemu_sem_timedwait(&pool->sem, 10000);
98 qemu_mutex_lock(&pool->lock);
99 pool->idle_threads--;
100 } while (ret == -1 && !QTAILQ_EMPTY(&pool->request_list));
101 if (ret == -1 || pool->stopping) {
102 break;
105 req = QTAILQ_FIRST(&pool->request_list);
106 QTAILQ_REMOVE(&pool->request_list, req, reqs);
107 req->state = THREAD_ACTIVE;
108 qemu_mutex_unlock(&pool->lock);
110 ret = req->func(req->arg);
112 req->ret = ret;
113 /* Write ret before state. */
114 smp_wmb();
115 req->state = THREAD_DONE;
117 qemu_mutex_lock(&pool->lock);
118 if (pool->pending_cancellations) {
119 qemu_cond_broadcast(&pool->check_cancel);
122 event_notifier_set(&pool->notifier);
125 pool->cur_threads--;
126 qemu_cond_signal(&pool->worker_stopped);
127 qemu_mutex_unlock(&pool->lock);
128 return NULL;
131 static void do_spawn_thread(ThreadPool *pool)
133 QemuThread t;
135 /* Runs with lock taken. */
136 if (!pool->new_threads) {
137 return;
140 pool->new_threads--;
141 pool->pending_threads++;
143 qemu_thread_create(&t, worker_thread, pool, QEMU_THREAD_DETACHED);
146 static void spawn_thread_bh_fn(void *opaque)
148 ThreadPool *pool = opaque;
150 qemu_mutex_lock(&pool->lock);
151 do_spawn_thread(pool);
152 qemu_mutex_unlock(&pool->lock);
155 static void spawn_thread(ThreadPool *pool)
157 pool->cur_threads++;
158 pool->new_threads++;
159 /* If there are threads being created, they will spawn new workers, so
160 * we don't spend time creating many threads in a loop holding a mutex or
161 * starving the current vcpu.
163 * If there are no idle threads, ask the main thread to create one, so we
164 * inherit the correct affinity instead of the vcpu affinity.
166 if (!pool->pending_threads) {
167 qemu_bh_schedule(pool->new_thread_bh);
171 static void event_notifier_ready(EventNotifier *notifier)
173 ThreadPool *pool = container_of(notifier, ThreadPool, notifier);
174 ThreadPoolElement *elem, *next;
176 event_notifier_test_and_clear(notifier);
177 restart:
178 QLIST_FOREACH_SAFE(elem, &pool->head, all, next) {
179 if (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
180 continue;
182 if (elem->state == THREAD_DONE) {
183 trace_thread_pool_complete(pool, elem, elem->common.opaque,
184 elem->ret);
186 if (elem->state == THREAD_DONE && elem->common.cb) {
187 QLIST_REMOVE(elem, all);
188 /* Read state before ret. */
189 smp_rmb();
190 elem->common.cb(elem->common.opaque, elem->ret);
191 qemu_aio_release(elem);
192 goto restart;
193 } else {
194 /* remove the request */
195 QLIST_REMOVE(elem, all);
196 qemu_aio_release(elem);
201 static void thread_pool_cancel(BlockDriverAIOCB *acb)
203 ThreadPoolElement *elem = (ThreadPoolElement *)acb;
204 ThreadPool *pool = elem->pool;
206 trace_thread_pool_cancel(elem, elem->common.opaque);
208 qemu_mutex_lock(&pool->lock);
209 if (elem->state == THREAD_QUEUED &&
210 /* No thread has yet started working on elem. we can try to "steal"
211 * the item from the worker if we can get a signal from the
212 * semaphore. Because this is non-blocking, we can do it with
213 * the lock taken and ensure that elem will remain THREAD_QUEUED.
215 qemu_sem_timedwait(&pool->sem, 0) == 0) {
216 QTAILQ_REMOVE(&pool->request_list, elem, reqs);
217 elem->state = THREAD_CANCELED;
218 event_notifier_set(&pool->notifier);
219 } else {
220 pool->pending_cancellations++;
221 while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
222 qemu_cond_wait(&pool->check_cancel, &pool->lock);
224 pool->pending_cancellations--;
226 qemu_mutex_unlock(&pool->lock);
229 static const AIOCBInfo thread_pool_aiocb_info = {
230 .aiocb_size = sizeof(ThreadPoolElement),
231 .cancel = thread_pool_cancel,
234 BlockDriverAIOCB *thread_pool_submit_aio(ThreadPool *pool,
235 ThreadPoolFunc *func, void *arg,
236 BlockDriverCompletionFunc *cb, void *opaque)
238 ThreadPoolElement *req;
240 req = qemu_aio_get(&thread_pool_aiocb_info, NULL, cb, opaque);
241 req->func = func;
242 req->arg = arg;
243 req->state = THREAD_QUEUED;
244 req->pool = pool;
246 QLIST_INSERT_HEAD(&pool->head, req, all);
248 trace_thread_pool_submit(pool, req, arg);
250 qemu_mutex_lock(&pool->lock);
251 if (pool->idle_threads == 0 && pool->cur_threads < pool->max_threads) {
252 spawn_thread(pool);
254 QTAILQ_INSERT_TAIL(&pool->request_list, req, reqs);
255 qemu_mutex_unlock(&pool->lock);
256 qemu_sem_post(&pool->sem);
257 return &req->common;
260 typedef struct ThreadPoolCo {
261 Coroutine *co;
262 int ret;
263 } ThreadPoolCo;
265 static void thread_pool_co_cb(void *opaque, int ret)
267 ThreadPoolCo *co = opaque;
269 co->ret = ret;
270 qemu_coroutine_enter(co->co, NULL);
273 int coroutine_fn thread_pool_submit_co(ThreadPool *pool, ThreadPoolFunc *func,
274 void *arg)
276 ThreadPoolCo tpc = { .co = qemu_coroutine_self(), .ret = -EINPROGRESS };
277 assert(qemu_in_coroutine());
278 thread_pool_submit_aio(pool, func, arg, thread_pool_co_cb, &tpc);
279 qemu_coroutine_yield();
280 return tpc.ret;
283 void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg)
285 thread_pool_submit_aio(pool, func, arg, NULL, NULL);
288 static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
290 if (!ctx) {
291 ctx = qemu_get_aio_context();
294 memset(pool, 0, sizeof(*pool));
295 event_notifier_init(&pool->notifier, false);
296 pool->ctx = ctx;
297 qemu_mutex_init(&pool->lock);
298 qemu_cond_init(&pool->check_cancel);
299 qemu_cond_init(&pool->worker_stopped);
300 qemu_sem_init(&pool->sem, 0);
301 pool->max_threads = 64;
302 pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);
304 QLIST_INIT(&pool->head);
305 QTAILQ_INIT(&pool->request_list);
307 aio_set_event_notifier(ctx, &pool->notifier, event_notifier_ready);
310 ThreadPool *thread_pool_new(AioContext *ctx)
312 ThreadPool *pool = g_new(ThreadPool, 1);
313 thread_pool_init_one(pool, ctx);
314 return pool;
317 void thread_pool_free(ThreadPool *pool)
319 if (!pool) {
320 return;
323 assert(QLIST_EMPTY(&pool->head));
325 qemu_mutex_lock(&pool->lock);
327 /* Stop new threads from spawning */
328 qemu_bh_delete(pool->new_thread_bh);
329 pool->cur_threads -= pool->new_threads;
330 pool->new_threads = 0;
332 /* Wait for worker threads to terminate */
333 pool->stopping = true;
334 while (pool->cur_threads > 0) {
335 qemu_sem_post(&pool->sem);
336 qemu_cond_wait(&pool->worker_stopped, &pool->lock);
339 qemu_mutex_unlock(&pool->lock);
341 aio_set_event_notifier(pool->ctx, &pool->notifier, NULL);
342 qemu_sem_destroy(&pool->sem);
343 qemu_cond_destroy(&pool->check_cancel);
344 qemu_cond_destroy(&pool->worker_stopped);
345 qemu_mutex_destroy(&pool->lock);
346 event_notifier_cleanup(&pool->notifier);
347 g_free(pool);