From mboxrd@z Thu Jan 1 00:00:00 1970 Received: from eggs.gnu.org ([2001:4830:134:3::10]:35743) by lists.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1XKQzH-0001HL-LD for qemu-devel@nongnu.org; Thu, 21 Aug 2014 07:57:09 -0400 Received: from Debian-exim by eggs.gnu.org with spam-scanned (Exim 4.71) (envelope-from ) id 1XKQzA-0006Bj-5J for qemu-devel@nongnu.org; Thu, 21 Aug 2014 07:57:03 -0400 Received: from mx1.redhat.com ([209.132.183.28]:63073) by eggs.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1XKQz8-0006BP-RJ for qemu-devel@nongnu.org; Thu, 21 Aug 2014 07:56:56 -0400 Received: from int-mx14.intmail.prod.int.phx2.redhat.com (int-mx14.intmail.prod.int.phx2.redhat.com [10.5.11.27]) by mx1.redhat.com (8.14.4/8.14.4) with ESMTP id s7LBurK8006477 (version=TLSv1/SSLv3 cipher=DHE-RSA-AES256-GCM-SHA384 bits=256 verify=OK) for ; Thu, 21 Aug 2014 07:56:54 -0400 From: Fam Zheng Date: Thu, 21 Aug 2014 19:56:52 +0800 Message-Id: <1408622216-9578-6-git-send-email-famz@redhat.com> In-Reply-To: <1408622216-9578-1-git-send-email-famz@redhat.com> References: <1408622216-9578-1-git-send-email-famz@redhat.com> Subject: [Qemu-devel] [RFC PATCH 5/9] thread-pool: Implement .cancel_async List-Id: List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: qemu-devel@nongnu.org Cc: Kevin Wolf , Paolo Bonzini , Stefan Hajnoczi The .cancel_async reuses the first half of .cancel: try to steal the request if not submitted yet. In this case set the elem to a special status THREAD_CANCELED_ASYNC, which means thread_pool_completion_bh should call the cb with -ECANCELED. If the request is already submitted, do nothing, as we know the normal completion will happen in the future. Signed-off-by: Fam Zheng --- thread-pool.c | 58 +++++++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 47 insertions(+), 11 deletions(-) diff --git a/thread-pool.c b/thread-pool.c index 23888dc..1ccc993 100644 --- a/thread-pool.c +++ b/thread-pool.c @@ -33,6 +33,10 @@ enum ThreadState { THREAD_ACTIVE, THREAD_DONE, THREAD_CANCELED, + /* + * Request is cancelled before submission by thread_pool_cancel_async. + * */ + THREAD_CANCELED_ASYNC, }; struct ThreadPoolElement { @@ -174,15 +178,22 @@ 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_CANCELED_ASYNC + && elem->state != THREAD_CANCELED + && elem->state != THREAD_DONE) { continue; } if (elem->state == THREAD_DONE) { trace_thread_pool_complete(pool, elem, elem->common.opaque, elem->ret); } - if (elem->state == THREAD_DONE && elem->common.cb) { + if ((elem->state == THREAD_DONE || + elem->state == THREAD_CANCELED_ASYNC) + && elem->common.cb) { QLIST_REMOVE(elem, all); + if (elem->state == THREAD_CANCELED_ASYNC) { + elem->ret = -ECANCELED; + } /* Read state before ret. */ smp_rmb(); @@ -202,6 +213,38 @@ restart: } } +/* With elem->pool->lock held */ +static bool thread_pool_cancel_from_queue(ThreadPoolElement *elem) +{ + if (elem->state == THREAD_QUEUED && + /* No thread has yet started working on elem. we can try to "steal" + * the item from the worker if we can get a signal from the + * semaphore. Because this is non-blocking, we can do it with + * the lock taken and ensure that elem will remain THREAD_QUEUED. + */ + qemu_sem_timedwait(&elem->pool->sem, 0) == 0) { + QTAILQ_REMOVE(&elem->pool->request_list, elem, reqs); + qemu_bh_schedule(elem->pool->completion_bh); + return true; + } + return false; +} + +static void thread_pool_cancel_async(BlockDriverAIOCB *acb) +{ + ThreadPoolElement *elem = (ThreadPoolElement *)acb; + ThreadPool *pool = elem->pool; + + trace_thread_pool_cancel(elem, elem->common.opaque); + + qemu_mutex_lock(&pool->lock); + if (thread_pool_cancel_from_queue(elem)) { + elem->state = THREAD_CANCELED_ASYNC; + } + + qemu_mutex_unlock(&pool->lock); +} + static void thread_pool_cancel(BlockDriverAIOCB *acb) { ThreadPoolElement *elem = (ThreadPoolElement *)acb; @@ -210,16 +253,8 @@ static void thread_pool_cancel(BlockDriverAIOCB *acb) trace_thread_pool_cancel(elem, elem->common.opaque); qemu_mutex_lock(&pool->lock); - if (elem->state == THREAD_QUEUED && - /* No thread has yet started working on elem. we can try to "steal" - * the item from the worker if we can get a signal from the - * semaphore. Because this is non-blocking, we can do it with - * the lock taken and ensure that elem will remain THREAD_QUEUED. - */ - qemu_sem_timedwait(&pool->sem, 0) == 0) { - QTAILQ_REMOVE(&pool->request_list, elem, reqs); + if (thread_pool_cancel_from_queue(elem)) { elem->state = THREAD_CANCELED; - qemu_bh_schedule(pool->completion_bh); } else { pool->pending_cancellations++; while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) { @@ -234,6 +269,7 @@ static void thread_pool_cancel(BlockDriverAIOCB *acb) static const AIOCBInfo thread_pool_aiocb_info = { .aiocb_size = sizeof(ThreadPoolElement), .cancel = thread_pool_cancel, + .cancel_async = thread_pool_cancel_async, }; BlockDriverAIOCB *thread_pool_submit_aio(ThreadPool *pool, -- 2.0.3