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 <f...@redhat.com> > --- > 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 > >