Re: [PATCH v2 05/17] thread-pool: Implement non-AIO (generic) pool support

[Fabiano Rosas <farosas@suse.de>]

"Maciej S. Szmigiero" <mail@maciej.szmigiero.name> writes:

> On 3.09.2024 00:07, Fabiano Rosas wrote:
>> "Maciej S. Szmigiero" <mail@maciej.szmigiero.name> writes:
>> 
>>> From: "Maciej S. Szmigiero" <maciej.szmigiero@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.szmigiero@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.

Yeah, I guess I meant that one set of changes has a larger potential for
introducing a bug while the other is clearly harmless.

>
>>>   
>>> +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.

Ok, but requests_executing is not the size of an object. It's the number
of objects in a linked list that satisfy a certain predicate. There are
no address space size considerations here.

>
> 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'm not sure I get you, there's no overflow since you're bounds checking
with the assert. Or is this a more abstract line of thought about how
many ThreadPoolElements can be present in memory at a time and you'd
like a type that's certain to fit the theoretical amount of objects?

>
> 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.

Ah I overlooked the break condition, nevermind.

> 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.

Right, but the idea was to piggyback on the thread termination to infer
(the obvious) requests service termination. We cannot do that, as you've
explained, fine.

> 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.

I'm not asking that of you, but I think it should be done
eventually. The QEMU block layer is very particular and I wouldn't want
the use-cases for the thread-pool to get confused. But I can't see a way
out right now, so let's postpone this, see if anyone else has comments.

>
>> ========
>> 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.

thread_pool_wait() might be better.

>
> Thanks,
> Maciej