[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
Re: [Qemu-devel] [PATCH v4 05/20] thread-pool: Convert thread_pool_aiocb
From: |
Benoît Canet |
Subject: |
Re: [Qemu-devel] [PATCH v4 05/20] thread-pool: Convert thread_pool_aiocb_info.cancel to cancel_async |
Date: |
Thu, 4 Sep 2014 17:29:13 +0200 |
User-agent: |
Mutt/1.5.23 (2014-03-12) |
The Wednesday 03 Sep 2014 à 19:23:40 (+0800), Fam Zheng wrote :
> The .cancel_async has the same the first half with .cancel: try to steal
"The .cancel_async share the same first half with .cancel" ?
> the request if not submitted yet. In this case set the elem to
> THREAD_DONE status and ret to -ECANCELED, which means
> thread_pool_completion_bh will call the cb with -ECANCELED.
>
> If the request is already submitted, do nothing, as we know the normal
> completion will happen in the future.
>
> Testing code update:
>
> Before, done_cb is only called if the request is already submitted by
> thread pool. Now done_cb is always called, even before it is submitted,
> because we emulate bdrv_aio_cancel with bdrv_aio_cancel_async. So also
> update the test criteria accordingly.
>
> Signed-off-by: Fam Zheng <address@hidden>
> ---
> tests/test-thread-pool.c | 34 ++++++++++++++++++++++++++--------
> thread-pool.c | 32 ++++++++++++++------------------
> 2 files changed, 40 insertions(+), 26 deletions(-)
>
> diff --git a/tests/test-thread-pool.c b/tests/test-thread-pool.c
> index f40b7fc..ed2b25b 100644
> --- a/tests/test-thread-pool.c
> +++ b/tests/test-thread-pool.c
> @@ -33,7 +33,7 @@ static int long_cb(void *opaque)
> static void done_cb(void *opaque, int ret)
> {
> WorkerTestData *data = opaque;
> - g_assert_cmpint(data->ret, ==, -EINPROGRESS);
> + g_assert(data->ret == -EINPROGRESS || data->ret == -ECANCELED);
> data->ret = ret;
> data->aiocb = NULL;
>
> @@ -132,7 +132,7 @@ static void test_submit_many(void)
> }
> }
>
> -static void test_cancel(void)
> +static void do_test_cancel(bool sync)
> {
> WorkerTestData data[100];
> int num_canceled;
> @@ -170,18 +170,25 @@ static void test_cancel(void)
> for (i = 0; i < 100; i++) {
> if (atomic_cmpxchg(&data[i].n, 0, 3) == 0) {
> data[i].ret = -ECANCELED;
> - bdrv_aio_cancel(data[i].aiocb);
> - active--;
> + if (sync) {
> + bdrv_aio_cancel(data[i].aiocb);
> + } else {
> + bdrv_aio_cancel_async(data[i].aiocb);
> + }
> num_canceled++;
> }
> }
> g_assert_cmpint(active, >, 0);
> g_assert_cmpint(num_canceled, <, 100);
>
> - /* Canceling the others will be a blocking operation. */
> for (i = 0; i < 100; i++) {
> if (data[i].aiocb && data[i].n != 3) {
> - bdrv_aio_cancel(data[i].aiocb);
> + if (sync) {
> + /* Canceling the others will be a blocking operation. */
> + bdrv_aio_cancel(data[i].aiocb);
> + } else {
> + bdrv_aio_cancel_async(data[i].aiocb);
> + }
> }
> }
>
> @@ -193,15 +200,25 @@ static void test_cancel(void)
> for (i = 0; i < 100; i++) {
> if (data[i].n == 3) {
> g_assert_cmpint(data[i].ret, ==, -ECANCELED);
> - g_assert(data[i].aiocb != NULL);
> + g_assert(data[i].aiocb == NULL);
> } else {
> g_assert_cmpint(data[i].n, ==, 2);
> - g_assert_cmpint(data[i].ret, ==, 0);
> + g_assert(data[i].ret == 0 || data[i].ret == -ECANCELED);
> g_assert(data[i].aiocb == NULL);
> }
> }
> }
>
> +static void test_cancel(void)
> +{
> + do_test_cancel(true);
> +}
> +
> +static void test_cancel_async(void)
> +{
> + do_test_cancel(false);
> +}
> +
> int main(int argc, char **argv)
> {
> int ret;
> @@ -217,6 +234,7 @@ int main(int argc, char **argv)
> g_test_add_func("/thread-pool/submit-co", test_submit_co);
> g_test_add_func("/thread-pool/submit-many", test_submit_many);
> g_test_add_func("/thread-pool/cancel", test_cancel);
> + g_test_add_func("/thread-pool/cancel-async", test_cancel_async);
>
> ret = g_test_run();
>
> diff --git a/thread-pool.c b/thread-pool.c
> index 23888dc..6afd343 100644
> --- a/thread-pool.c
> +++ b/thread-pool.c
> @@ -32,7 +32,6 @@ enum ThreadState {
> THREAD_QUEUED,
> THREAD_ACTIVE,
> THREAD_DONE,
> - THREAD_CANCELED,
> };
>
> struct ThreadPoolElement {
> @@ -59,7 +58,6 @@ struct ThreadPool {
> AioContext *ctx;
> QEMUBH *completion_bh;
> QemuMutex lock;
> - QemuCond check_cancel;
> QemuCond worker_stopped;
> QemuSemaphore sem;
> int max_threads;
> @@ -74,7 +72,6 @@ struct ThreadPool {
> int idle_threads;
> int new_threads; /* backlog of threads we need to create */
> int pending_threads; /* threads created but not running yet */
> - int pending_cancellations; /* whether we need a cond_broadcast */
> bool stopping;
> };
>
> @@ -114,9 +111,6 @@ static void *worker_thread(void *opaque)
> req->state = THREAD_DONE;
>
> qemu_mutex_lock(&pool->lock);
> - if (pool->pending_cancellations) {
> - qemu_cond_broadcast(&pool->check_cancel);
> - }
>
> qemu_bh_schedule(pool->completion_bh);
> }
> @@ -174,7 +168,7 @@ static void thread_pool_completion_bh(void *opaque)
>
> restart:
> QLIST_FOREACH_SAFE(elem, &pool->head, all, next) {
> - if (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
> + if (elem->state != THREAD_DONE) {
> continue;
> }
> if (elem->state == THREAD_DONE) {
> @@ -218,22 +212,26 @@ static void thread_pool_cancel(BlockDriverAIOCB *acb)
> */
> qemu_sem_timedwait(&pool->sem, 0) == 0) {
> QTAILQ_REMOVE(&pool->request_list, elem, reqs);
> - elem->state = THREAD_CANCELED;
> qemu_bh_schedule(pool->completion_bh);
> - } else {
> - pool->pending_cancellations++;
> - while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE)
> {
> - qemu_cond_wait(&pool->check_cancel, &pool->lock);
> - }
> - pool->pending_cancellations--;
> +
> + elem->state = THREAD_DONE;
> + elem->ret = -ECANCELED;
> }
> +
> qemu_mutex_unlock(&pool->lock);
> - thread_pool_completion_bh(pool);
> +}
> +
> +static AioContext *thread_pool_get_aio_context(BlockDriverAIOCB *acb)
> +{
> + ThreadPoolElement *elem = (ThreadPoolElement *)acb;
> + ThreadPool *pool = elem->pool;
> + return pool->ctx;
> }
>
> static const AIOCBInfo thread_pool_aiocb_info = {
> .aiocb_size = sizeof(ThreadPoolElement),
> - .cancel = thread_pool_cancel,
> + .cancel_async = thread_pool_cancel,
> + .get_aio_context = thread_pool_get_aio_context,
> };
>
> BlockDriverAIOCB *thread_pool_submit_aio(ThreadPool *pool,
> @@ -300,7 +298,6 @@ static void thread_pool_init_one(ThreadPool *pool,
> AioContext *ctx)
> pool->ctx = ctx;
> pool->completion_bh = aio_bh_new(ctx, thread_pool_completion_bh, pool);
> qemu_mutex_init(&pool->lock);
> - qemu_cond_init(&pool->check_cancel);
> qemu_cond_init(&pool->worker_stopped);
> qemu_sem_init(&pool->sem, 0);
> pool->max_threads = 64;
> @@ -343,7 +340,6 @@ void thread_pool_free(ThreadPool *pool)
>
> qemu_bh_delete(pool->completion_bh);
> qemu_sem_destroy(&pool->sem);
> - qemu_cond_destroy(&pool->check_cancel);
> qemu_cond_destroy(&pool->worker_stopped);
> qemu_mutex_destroy(&pool->lock);
> g_free(pool);
> --
> 2.1.0.27.g96db324
>
>
- [Qemu-devel] [PATCH v4 02/20] block: Add bdrv_aio_cancel_async, (continued)
[Qemu-devel] [PATCH v4 05/20] thread-pool: Convert thread_pool_aiocb_info.cancel to cancel_async, Fam Zheng, 2014/09/03
- Re: [Qemu-devel] [PATCH v4 05/20] thread-pool: Convert thread_pool_aiocb_info.cancel to cancel_async,
Benoît Canet <=
[Qemu-devel] [PATCH v4 06/20] linux-aio: Convert laio_aiocb_info.cancel to .cancel_async, Fam Zheng, 2014/09/03
[Qemu-devel] [PATCH v4 07/20] dma: Check iov pointer before unmap memory, Fam Zheng, 2014/09/03
[Qemu-devel] [PATCH v4 08/20] dma: Convert dma_aiocb_info.cancel to .cancel_async, Fam Zheng, 2014/09/03
[Qemu-devel] [PATCH v4 09/20] iscsi: Convert iscsi_aiocb_info.cancel to .cancel_async, Fam Zheng, 2014/09/03
[Qemu-devel] [PATCH v4 10/20] archipelago: Drop archipelago_aiocb_info.cancel, Fam Zheng, 2014/09/03
[Qemu-devel] [PATCH v4 11/20] blkdebug: Convert blkdebug_aiocb_info.cancel to .cancel_async, Fam Zheng, 2014/09/03
[Qemu-devel] [PATCH v4 12/20] blkverify: Drop blkverify_aiocb_info.cancel, Fam Zheng, 2014/09/03
[Qemu-devel] [PATCH v4 13/20] curl: Drop curl_aiocb_info.cancel, Fam Zheng, 2014/09/03