On 3.09.2024 00:07, Fabiano Rosas wrote:
"Maciej S. Szmigiero" <m...@maciej.szmigiero.name> writes:
From: "Maciej S. Szmigiero" <maciej.szmigi...@oracle.com>
Migration code wants to manage device data sending threads in one place.
QEMU has an existing thread pool implementation, however it was limited
to queuing AIO operations only and essentially had a 1:1 mapping between
the current AioContext and the ThreadPool in use.
Implement what is necessary to queue generic (non-AIO) work on a ThreadPool
too.
This brings a few new operations on a pool:
* thread_pool_set_minmax_threads() explicitly sets the minimum and maximum
thread count in the pool.
* thread_pool_join() operation waits until all the submitted work requests
have finished.
* thread_pool_poll() lets the new thread and / or thread completion bottom
halves run (if they are indeed scheduled to be run).
It is useful for thread pool users that need to launch or terminate new
threads without returning to the QEMU main loop.
Signed-off-by: Maciej S. Szmigiero <maciej.szmigi...@oracle.com>
---
include/block/thread-pool.h | 10 ++++-
tests/unit/test-thread-pool.c | 2 +-
util/thread-pool.c | 77 ++++++++++++++++++++++++++++++-----
3 files changed, 76 insertions(+), 13 deletions(-)
diff --git a/include/block/thread-pool.h b/include/block/thread-pool.h
index b484c4780ea6..1769496056cd 100644
--- a/include/block/thread-pool.h
+++ b/include/block/thread-pool.h
@@ -37,9 +37,15 @@ BlockAIOCB *thread_pool_submit_aio(ThreadPoolFunc *func,
void *arg, GDestroyNotify arg_destroy,
BlockCompletionFunc *cb, void *opaque);
int coroutine_fn thread_pool_submit_co(ThreadPoolFunc *func, void *arg);
-void thread_pool_submit(ThreadPoolFunc *func,
- void *arg, GDestroyNotify arg_destroy);
+BlockAIOCB *thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func,
+ void *arg, GDestroyNotify arg_destroy,
+ BlockCompletionFunc *cb, void *opaque);
These kinds of changes (create wrappers, change signatures, etc), could
be in their own patch as it's just code motion that should not have
functional impact. The "no_requests" stuff would be better discussed in
a separate patch.
These changes *all* should have no functional impact on existing callers.
But I get your overall point, will try to separate these really trivial
parts.
+void thread_pool_join(ThreadPool *pool);
+void thread_pool_poll(ThreadPool *pool);
+
+void thread_pool_set_minmax_threads(ThreadPool *pool,
+ int min_threads, int max_threads);
void thread_pool_update_params(ThreadPool *pool, struct AioContext *ctx);
#endif
diff --git a/tests/unit/test-thread-pool.c b/tests/unit/test-thread-pool.c
index e4afb9e36292..469c0f7057b6 100644
--- a/tests/unit/test-thread-pool.c
+++ b/tests/unit/test-thread-pool.c
@@ -46,7 +46,7 @@ static void done_cb(void *opaque, int ret)
static void test_submit(void)
{
WorkerTestData data = { .n = 0 };
- thread_pool_submit(worker_cb, &data, NULL);
+ thread_pool_submit(NULL, worker_cb, &data, NULL, NULL, NULL);
while (data.n == 0) {
aio_poll(ctx, true);
}
diff --git a/util/thread-pool.c b/util/thread-pool.c
index 69a87ee79252..2bf3be875a51 100644
--- a/util/thread-pool.c
+++ b/util/thread-pool.c
@@ -60,6 +60,7 @@ struct ThreadPool {
QemuMutex lock;
QemuCond worker_stopped;
QemuCond request_cond;
+ QemuCond no_requests_cond;
QEMUBH *new_thread_bh;
/* The following variables are only accessed from one AioContext. */
@@ -73,6 +74,7 @@ struct ThreadPool {
int pending_threads; /* threads created but not running yet */
int min_threads;
int max_threads;
+ size_t requests_executing;
What's with size_t? Should this be a uint32_t instead?
Sizes of objects are normally size_t, since otherwise bad
things happen if objects are bigger than 4 GiB.
Considering that the minimum object size is 1 byte the
max count of distinct objects also needs a size_t to not
risk an overflow.
I think that while 2^32 requests executing seems unlikely
saving 4 bytes seems not worth worrying that someone will
find a vulnerability triggered by overflowing a 32-bit
variable (not necessary in the migration code but in some
other thread pool user).
};
static void *worker_thread(void *opaque)
@@ -107,6 +109,10 @@ static void *worker_thread(void *opaque)
req = QTAILQ_FIRST(&pool->request_list);
QTAILQ_REMOVE(&pool->request_list, req, reqs);
req->state = THREAD_ACTIVE;
+
+ assert(pool->requests_executing < SIZE_MAX);
+ pool->requests_executing++;
+
qemu_mutex_unlock(&pool->lock);
ret = req->func(req->arg);
@@ -118,6 +124,14 @@ static void *worker_thread(void *opaque)
qemu_bh_schedule(pool->completion_bh);
qemu_mutex_lock(&pool->lock);
+
+ assert(pool->requests_executing > 0);
+ pool->requests_executing--;
+
+ if (pool->requests_executing == 0 &&
+ QTAILQ_EMPTY(&pool->request_list)) {
+ qemu_cond_signal(&pool->no_requests_cond);
+ }
An empty requests list and no request in flight means the worker will
now exit after the timeout, no? Can you just kick the worker out of the
wait and use pool->worker_stopped instead of the new condition variable?
First, all threads won't terminate if either min_threads or max_threads
isn't 0.
It might be in the migration thread pool case but we are adding a
generic thread pool so it should be as universal as possible.
thread_pool_free() can get away with overwriting these values since
it is destroying the pool anyway.
Also, the *_join() (or whatever its final name will be) operation is
about waiting for all requests / work items to finish, not about waiting
for threads to terminate.
It's essentially a synchronization point for a thread pool, not a cleanup.
}
pool->cur_threads--;
@@ -243,13 +257,16 @@ static const AIOCBInfo thread_pool_aiocb_info = {
.cancel_async = thread_pool_cancel,
};
-BlockAIOCB *thread_pool_submit_aio(ThreadPoolFunc *func,
- void *arg, GDestroyNotify arg_destroy,
- BlockCompletionFunc *cb, void *opaque)
+BlockAIOCB *thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func,
+ void *arg, GDestroyNotify arg_destroy,
+ BlockCompletionFunc *cb, void *opaque)
{
ThreadPoolElement *req;
AioContext *ctx = qemu_get_current_aio_context();
- ThreadPool *pool = aio_get_thread_pool(ctx);
+
+ if (!pool) {
+ pool = aio_get_thread_pool(ctx);
+ }
I'd go for a separate implementation to really drive the point that this
new usage is different. See the code snippet below.
I see your point there - will split these implementations.
It seems we're a short step away to being able to use this
implementation in a general way. Is there something that can be done
with the 'common' field in the ThreadPoolElement?
The non-AIO request flow still need the completion callback from BlockAIOCB
(and its argument pointer) so removing the "common" field from these requests
would need introducing two "flavors" of ThreadPoolElement.
Not sure memory saving here are worth the increase in code complexity.
========
static void thread_pool_submit_request(ThreadPool *pool, ThreadPoolElement *req)
{
req->state = THREAD_QUEUED;
req->pool = pool;
QLIST_INSERT_HEAD(&pool->head, req, all);
trace_thread_pool_submit(pool, req, 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_cond_signal(&pool->request_cond);
}
BlockAIOCB *thread_pool_submit_aio(ThreadPoolFunc *func, void *arg,
BlockCompletionFunc *cb, void *opaque)
{
ThreadPoolElement *req;
AioContext *ctx = qemu_get_current_aio_context();
ThreadPool *pool = aio_get_thread_pool(ctx);
/* Assert that the thread submitting work is the same running the pool */
assert(pool->ctx == qemu_get_current_aio_context());
req = qemu_aio_get(&thread_pool_aiocb_info, NULL, cb, opaque);
req->func = func;
req->arg = arg;
thread_pool_submit_request(pool, req);
return &req->common;
}
void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg)
{
ThreadPoolElement *req;
req = g_malloc(sizeof(ThreadPoolElement));
req->func = func;
req->arg = arg;
thread_pool_submit_request(pool, req);
}
=================
/* Assert that the thread submitting work is the same running the pool */
assert(pool->ctx == qemu_get_current_aio_context());
@@ -275,6 +292,18 @@ BlockAIOCB *thread_pool_submit_aio(ThreadPoolFunc *func,
return &req->common;
}
+BlockAIOCB *thread_pool_submit_aio(ThreadPoolFunc *func,
+ void *arg, GDestroyNotify arg_destroy,
+ BlockCompletionFunc *cb, void *opaque)
+{
+ return thread_pool_submit(NULL, func, arg, arg_destroy, cb, opaque);
+}
+
+void thread_pool_poll(ThreadPool *pool)
+{
+ aio_bh_poll(pool->ctx);
+}
+
typedef struct ThreadPoolCo {
Coroutine *co;
int ret;
@@ -297,18 +326,38 @@ int coroutine_fn thread_pool_submit_co(ThreadPoolFunc
*func, void *arg)
return tpc.ret;
}
-void thread_pool_submit(ThreadPoolFunc *func,
- void *arg, GDestroyNotify arg_destroy)
+void thread_pool_join(ThreadPool *pool)
This is misleading because it's about the requests, not the threads in
the pool. Compare with what thread_pool_free does:
/* Wait for worker threads to terminate */
pool->max_threads = 0;
qemu_cond_broadcast(&pool->request_cond);
while (pool->cur_threads > 0) {
qemu_cond_wait(&pool->worker_stopped, &pool->lock);
}
I'm open to thread_pool_join() better naming proposals.
Thanks,
Maciej
