@samitouri / QOSamiQemu
Code Commits Issues Pulls Sessions CI/CD
master
master
c 497 lines 13.5 KB
Raw
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/osdep.h"
18 #include "qemu/defer-call.h"
19 #include "qemu/queue.h"
20 #include "qemu/thread.h"
21 #include "qemu/atomic.h"
22 #include "qemu/coroutine.h"
23 #include "trace.h"
24 #include "block/thread-pool.h"
25 #include "qemu/main-loop.h"
26
27 static void do_spawn_thread(ThreadPoolAio *pool);
28
29 typedef struct ThreadPoolElementAio ThreadPoolElementAio;
30
31 enum ThreadState {
32 THREAD_QUEUED,
33 THREAD_ACTIVE,
34 THREAD_DONE,
35 };
36
37 struct ThreadPoolElementAio {
38 BlockAIOCB common;
39 ThreadPoolAio *pool;
40 ThreadPoolFunc *func;
41 void *arg;
42
43 /*
44 * Accessed with atomics. Moving state out of THREAD_QUEUED is
45 * protected by pool->lock and only the worker thread can move
46 * the state from THREAD_ACTIVE to THREAD_DONE.
47 *
48 * When state is THREAD_DONE, ret must have been written already.
49 * Use acquire/release ordering when reading/writing ret as well.
50 */
51 enum ThreadState state;
52 int ret;
53
54 /* Access to this list is protected by lock. */
55 QTAILQ_ENTRY(ThreadPoolElementAio) reqs;
56
57 /* This list is only written by the thread pool's mother thread. */
58 QLIST_ENTRY(ThreadPoolElementAio) all;
59 };
60
61 struct ThreadPoolAio {
62 AioContext *ctx;
63 QEMUBH *completion_bh;
64 QemuMutex lock;
65 QemuCond worker_stopped;
66 QemuCond request_cond;
67 QEMUBH *new_thread_bh;
68
69 /* The following variables are only accessed from one AioContext. */
70 QLIST_HEAD(, ThreadPoolElementAio) head;
71
72 /* The following variables are protected by lock. */
73 QTAILQ_HEAD(, ThreadPoolElementAio) 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 min_threads;
79 int max_threads;
80 };
81
82 static void *worker_thread(void *opaque)
83 {
84 ThreadPoolAio *pool = opaque;
85
86 qemu_mutex_lock(&pool->lock);
87 pool->pending_threads--;
88 do_spawn_thread(pool);
89
90 while (pool->cur_threads <= pool->max_threads) {
91 ThreadPoolElementAio *req;
92 int ret;
93
94 if (QTAILQ_EMPTY(&pool->request_list)) {
95 pool->idle_threads++;
96 ret = qemu_cond_timedwait(&pool->request_cond, &pool->lock, 10000);
97 pool->idle_threads--;
98 if (ret == 0 &&
99 QTAILQ_EMPTY(&pool->request_list) &&
100 pool->cur_threads > pool->min_threads) {
101 /* Timed out + no work to do + no need for warm threads = exit. */
102 break;
103 }
104 /*
105 * Even if there was some work to do, check if there aren't
106 * too many worker threads before picking it up.
107 */
108 continue;
109 }
110
111 req = QTAILQ_FIRST(&pool->request_list);
112 QTAILQ_REMOVE(&pool->request_list, req, reqs);
113 qatomic_set(&req->state, THREAD_ACTIVE);
114 qemu_mutex_unlock(&pool->lock);
115
116 ret = req->func(req->arg);
117
118 qatomic_set(&req->ret, ret);
119 /* _release to write ret before state. */
120 qatomic_store_release(&req->state, THREAD_DONE);
121
122 qemu_bh_schedule(pool->completion_bh);
123 qemu_mutex_lock(&pool->lock);
124 }
125
126 pool->cur_threads--;
127 qemu_cond_signal(&pool->worker_stopped);
128
129 /*
130 * Wake up another thread, in case we got a wakeup but decided
131 * to exit due to pool->cur_threads > pool->max_threads.
132 */
133 qemu_cond_signal(&pool->request_cond);
134 qemu_mutex_unlock(&pool->lock);
135 return NULL;
136 }
137
138 static void do_spawn_thread(ThreadPoolAio *pool)
139 {
140 QemuThread t;
141
142 /* Runs with lock taken. */
143 if (!pool->new_threads) {
144 return;
145 }
146
147 pool->new_threads--;
148 pool->pending_threads++;
149
150 qemu_thread_create(&t, "worker", worker_thread, pool, QEMU_THREAD_DETACHED);
151 }
152
153 static void spawn_thread_bh_fn(void *opaque)
154 {
155 ThreadPoolAio *pool = opaque;
156
157 qemu_mutex_lock(&pool->lock);
158 do_spawn_thread(pool);
159 qemu_mutex_unlock(&pool->lock);
160 }
161
162 static void spawn_thread(ThreadPoolAio *pool)
163 {
164 pool->cur_threads++;
165 pool->new_threads++;
166 /* If there are threads being created, they will spawn new workers, so
167 * we don't spend time creating many threads in a loop holding a mutex or
168 * starving the current vcpu.
169 *
170 * If there are no idle threads, ask the main thread to create one, so we
171 * inherit the correct affinity instead of the vcpu affinity.
172 */
173 if (!pool->pending_threads) {
174 qemu_bh_schedule(pool->new_thread_bh);
175 }
176 }
177
178 static void thread_pool_completion_bh(void *opaque)
179 {
180 ThreadPoolAio *pool = opaque;
181 ThreadPoolElementAio *elem, *next;
182
183 defer_call_begin(); /* cb() may use defer_call() to coalesce work */
184
185 restart:
186 QLIST_FOREACH_SAFE(elem, &pool->head, all, next) {
187 /* _acquire to read state before ret. */
188 if (qatomic_load_acquire(&elem->state) != THREAD_DONE) {
189 continue;
190 }
191
192 trace_thread_pool_complete_aio(pool, elem, elem->common.opaque,
193 elem->ret);
194 QLIST_REMOVE(elem, all);
195
196 if (elem->common.cb) {
197 /* Schedule ourselves in case elem->common.cb() calls aio_poll() to
198 * wait for another request that completed at the same time.
199 */
200 qemu_bh_schedule(pool->completion_bh);
201
202 elem->common.cb(elem->common.opaque, elem->ret);
203
204 /* We can safely cancel the completion_bh here regardless of someone
205 * else having scheduled it meanwhile because we reenter the
206 * completion function anyway (goto restart).
207 */
208 qemu_bh_cancel(pool->completion_bh);
209
210 qemu_aio_unref(elem);
211 goto restart;
212 } else {
213 qemu_aio_unref(elem);
214 }
215 }
216
217 defer_call_end();
218 }
219
220 static void thread_pool_cancel(BlockAIOCB *acb)
221 {
222 ThreadPoolElementAio *elem = (ThreadPoolElementAio *)acb;
223 ThreadPoolAio *pool = elem->pool;
224
225 trace_thread_pool_cancel_aio(elem, elem->common.opaque);
226
227 QEMU_LOCK_GUARD(&pool->lock);
228 if (qatomic_read(&elem->state) == THREAD_QUEUED) {
229 QTAILQ_REMOVE(&pool->request_list, elem, reqs);
230 qemu_bh_schedule(pool->completion_bh);
231
232 qatomic_set(&elem->ret, -ECANCELED);
233 qatomic_store_release(&elem->state, THREAD_DONE);
234 }
235
236 }
237
238 static const AIOCBInfo thread_pool_aiocb_info = {
239 .aiocb_size = sizeof(ThreadPoolElementAio),
240 .cancel_async = thread_pool_cancel,
241 };
242
243 BlockAIOCB *thread_pool_submit_aio(ThreadPoolFunc *func, void *arg,
244 BlockCompletionFunc *cb, void *opaque)
245 {
246 ThreadPoolElementAio *req;
247 AioContext *ctx = qemu_get_current_aio_context();
248 ThreadPoolAio *pool = aio_get_thread_pool(ctx);
249
250 /* Assert that the thread submitting work is the same running the pool */
251 assert(pool->ctx == qemu_get_current_aio_context());
252
253 req = qemu_aio_get(&thread_pool_aiocb_info, NULL, cb, opaque);
254 req->func = func;
255 req->arg = arg;
256 req->state = THREAD_QUEUED;
257 req->pool = pool;
258
259 QLIST_INSERT_HEAD(&pool->head, req, all);
260
261 trace_thread_pool_submit_aio(pool, req, arg);
262
263 qemu_mutex_lock(&pool->lock);
264 if (pool->idle_threads == 0 && pool->cur_threads < pool->max_threads) {
265 spawn_thread(pool);
266 }
267 QTAILQ_INSERT_TAIL(&pool->request_list, req, reqs);
268 qemu_mutex_unlock(&pool->lock);
269 qemu_cond_signal(&pool->request_cond);
270 return &req->common;
271 }
272
273 typedef struct ThreadPoolCo {
274 Coroutine *co;
275 int ret;
276 } ThreadPoolCo;
277
278 static void thread_pool_co_cb(void *opaque, int ret)
279 {
280 ThreadPoolCo *co = opaque;
281
282 co->ret = ret;
283 aio_co_wake(co->co);
284 }
285
286 int coroutine_fn thread_pool_submit_co(ThreadPoolFunc *func, void *arg)
287 {
288 ThreadPoolCo tpc = { .co = qemu_coroutine_self(), .ret = -EINPROGRESS };
289 assert(qemu_in_coroutine());
290 thread_pool_submit_aio(func, arg, thread_pool_co_cb, &tpc);
291 qemu_coroutine_yield();
292 return tpc.ret;
293 }
294
295 void thread_pool_update_params(ThreadPoolAio *pool, AioContext *ctx)
296 {
297 qemu_mutex_lock(&pool->lock);
298
299 pool->min_threads = ctx->thread_pool_min;
300 pool->max_threads = ctx->thread_pool_max;
301
302 /*
303 * We either have to:
304 * - Increase the number available of threads until over the min_threads
305 * threshold.
306 * - Bump the worker threads so that they exit, until under the max_threads
307 * threshold.
308 * - Do nothing. The current number of threads fall in between the min and
309 * max thresholds. We'll let the pool manage itself.
310 */
311 for (int i = pool->cur_threads; i < pool->min_threads; i++) {
312 spawn_thread(pool);
313 }
314
315 for (int i = pool->cur_threads; i > pool->max_threads; i--) {
316 qemu_cond_signal(&pool->request_cond);
317 }
318
319 qemu_mutex_unlock(&pool->lock);
320 }
321
322 static void thread_pool_init_one(ThreadPoolAio *pool, AioContext *ctx)
323 {
324 if (!ctx) {
325 ctx = qemu_get_aio_context();
326 }
327
328 memset(pool, 0, sizeof(*pool));
329 pool->ctx = ctx;
330 pool->completion_bh = aio_bh_new(ctx, thread_pool_completion_bh, pool);
331 qemu_mutex_init(&pool->lock);
332 qemu_cond_init(&pool->worker_stopped);
333 qemu_cond_init(&pool->request_cond);
334 pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);
335
336 QLIST_INIT(&pool->head);
337 QTAILQ_INIT(&pool->request_list);
338
339 thread_pool_update_params(pool, ctx);
340 }
341
342 ThreadPoolAio *thread_pool_new_aio(AioContext *ctx)
343 {
344 ThreadPoolAio *pool = g_new(ThreadPoolAio, 1);
345 thread_pool_init_one(pool, ctx);
346 return pool;
347 }
348
349 void thread_pool_free_aio(ThreadPoolAio *pool)
350 {
351 if (!pool) {
352 return;
353 }
354
355 assert(QLIST_EMPTY(&pool->head));
356
357 qemu_mutex_lock(&pool->lock);
358
359 /* Stop new threads from spawning */
360 qemu_bh_delete(pool->new_thread_bh);
361 pool->cur_threads -= pool->new_threads;
362 pool->new_threads = 0;
363
364 /* Wait for worker threads to terminate */
365 pool->max_threads = 0;
366 qemu_cond_broadcast(&pool->request_cond);
367 while (pool->cur_threads > 0) {
368 qemu_cond_wait(&pool->worker_stopped, &pool->lock);
369 }
370
371 qemu_mutex_unlock(&pool->lock);
372
373 qemu_bh_delete(pool->completion_bh);
374 qemu_cond_destroy(&pool->request_cond);
375 qemu_cond_destroy(&pool->worker_stopped);
376 qemu_mutex_destroy(&pool->lock);
377 g_free(pool);
378 }
379
380 struct ThreadPool {
381 GThreadPool *t;
382 size_t cur_work;
383 QemuMutex cur_work_lock;
384 QemuCond all_finished_cond;
385 };
386
387 typedef struct {
388 ThreadPoolFunc *func;
389 void *opaque;
390 GDestroyNotify opaque_destroy;
391 } ThreadPoolElement;
392
393 static void thread_pool_func(gpointer data, gpointer user_data)
394 {
395 ThreadPool *pool = user_data;
396 g_autofree ThreadPoolElement *el = data;
397
398 el->func(el->opaque);
399
400 if (el->opaque_destroy) {
401 el->opaque_destroy(el->opaque);
402 }
403
404 QEMU_LOCK_GUARD(&pool->cur_work_lock);
405
406 assert(pool->cur_work > 0);
407 pool->cur_work--;
408
409 if (pool->cur_work == 0) {
410 qemu_cond_signal(&pool->all_finished_cond);
411 }
412 }
413
414 ThreadPool *thread_pool_new(void)
415 {
416 ThreadPool *pool = g_new(ThreadPool, 1);
417
418 pool->cur_work = 0;
419 qemu_mutex_init(&pool->cur_work_lock);
420 qemu_cond_init(&pool->all_finished_cond);
421
422 pool->t = g_thread_pool_new(thread_pool_func, pool, 0, TRUE, NULL);
423 /*
424 * g_thread_pool_new() can only return errors if initial thread(s)
425 * creation fails but we ask for 0 initial threads above.
426 */
427 assert(pool->t);
428
429 return pool;
430 }
431
432 void thread_pool_free(ThreadPool *pool)
433 {
434 /*
435 * With _wait = TRUE this effectively waits for all
436 * previously submitted work to complete first.
437 */
438 g_thread_pool_free(pool->t, FALSE, TRUE);
439
440 qemu_cond_destroy(&pool->all_finished_cond);
441 qemu_mutex_destroy(&pool->cur_work_lock);
442
443 g_free(pool);
444 }
445
446 void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func,
447 void *opaque, GDestroyNotify opaque_destroy)
448 {
449 ThreadPoolElement *el = g_new(ThreadPoolElement, 1);
450
451 el->func = func;
452 el->opaque = opaque;
453 el->opaque_destroy = opaque_destroy;
454
455 WITH_QEMU_LOCK_GUARD(&pool->cur_work_lock) {
456 pool->cur_work++;
457 }
458
459 /*
460 * Ignore the return value since this function can only return errors
461 * if creation of an additional thread fails but even in this case the
462 * provided work is still getting queued (just for the existing threads).
463 */
464 g_thread_pool_push(pool->t, el, NULL);
465 }
466
467 void thread_pool_submit_immediate(ThreadPool *pool, ThreadPoolFunc *func,
468 void *opaque, GDestroyNotify opaque_destroy)
469 {
470 thread_pool_submit(pool, func, opaque, opaque_destroy);
471 thread_pool_adjust_max_threads_to_work(pool);
472 }
473
474 void thread_pool_wait(ThreadPool *pool)
475 {
476 QEMU_LOCK_GUARD(&pool->cur_work_lock);
477
478 while (pool->cur_work > 0) {
479 qemu_cond_wait(&pool->all_finished_cond,
480 &pool->cur_work_lock);
481 }
482 }
483
484 bool thread_pool_set_max_threads(ThreadPool *pool,
485 int max_threads)
486 {
487 assert(max_threads > 0);
488
489 return g_thread_pool_set_max_threads(pool->t, max_threads, NULL);
490 }
491
492 bool thread_pool_adjust_max_threads_to_work(ThreadPool *pool)
493 {
494 QEMU_LOCK_GUARD(&pool->cur_work_lock);
495
496 return thread_pool_set_max_threads(pool, MAX(pool->cur_work, 1));
497 }