[Qemu-devel] [RFC PATCH v2 5/8] thread-pool: Implement .cancel_async

[Fam Zheng <famz@redhat.com>]

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