[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Qemu-devel] [RFC PATCH 5/9] thread-pool: Implement .cancel_async
From: |
Fam Zheng |
Subject: |
[Qemu-devel] [RFC PATCH 5/9] thread-pool: Implement .cancel_async |
Date: |
Thu, 21 Aug 2014 19:56:52 +0800 |
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 <address@hidden>
---
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
- [Qemu-devel] [RFC PATCH 0/9] scsi, block: Asynchronous request cancellation, Fam Zheng, 2014/08/21
- [Qemu-devel] [RFC PATCH 1/9] block: Add bdrv_aio_cancel_async, Fam Zheng, 2014/08/21
- [Qemu-devel] [RFC PATCH 2/9] tests: Add testing code for bdrv_aio_cancel_async, Fam Zheng, 2014/08/21
- [Qemu-devel] [RFC PATCH 3/9] iscsi: Implement .cancel_async in acb info, Fam Zheng, 2014/08/21
- [Qemu-devel] [RFC PATCH 4/9] linux-aio: Implement .cancel_async, Fam Zheng, 2014/08/21
- [Qemu-devel] [RFC PATCH 5/9] thread-pool: Implement .cancel_async,
Fam Zheng <=
- [Qemu-devel] [RFC PATCH 7/9] dma: Implement .cancel_async, Fam Zheng, 2014/08/21
- [Qemu-devel] [RFC PATCH 8/9] block: Implement stub bdrv_em_co_aiocb_info.cancel_async, Fam Zheng, 2014/08/21
- [Qemu-devel] [RFC PATCH 6/9] blkdebug: Implement .cancel_async, Fam Zheng, 2014/08/21
- [Qemu-devel] [RFC PATCH 9/9] scsi: Cancel request asynchronously, Fam Zheng, 2014/08/21