[Qemu-devel] [RFC PATCH v2 0/8] block: Asynchronous request cancellation

[Qemu-devel] [RFC PATCH v2 0/8] block: Asynchronous request cancellation

[Fam Zheng]

v2: Drop the unfinished scsi part, which was broken in v1. (Paolo)
    Add refcnt in BlockDriverAIOCB to maintain invariant of acb availability in
    bdrv_aio_cancel_async. (Paolo)
    Drop blkdebug change. (Stefan)

This series adds a new block API:

  void bdrv_aio_cancel_async(BlockDriverAIOCB *acb);

which is similar to existing bdrv_aio_cancel in that it cancels an AIO request,
but different that it doesn't block until the request is completely cancelled
or done.

More importantly, the completion callback, BlockDriverAIOCB.cb, is guaranteed
to be called, so that the cb can take care of resource releasing and status
reporting to guest, etc.

In the following work, scsi emulation code will be shifted to use the async
cancelling.

One major benefit would be that when guest tries to cancel a request, where the
request cannot be cancelled easily, (due to throttled BlockDriverState, a lost
connection, or a large request queue), we don't need to block the whole vm with
a busy loop, which is how bdrv_aio_cancel is implemented now.

A test case that is easy to reproduce is, throttle a scsi-disk to a very low
limit, for example 50 bps, then stress the guest block device with dd or fio.

Currently, the vm will quickly hang when it loses patience and sends a tmf
command to cancel the request, at which point we will spin in bdrv_aio_cancel,
until the request is slowly spit out from throttled_reqs.

We should change scsi device code to make this asynchronous on top of
bdrv_aio_cancel_async.

Fam


Fam Zheng (8):
  block: Add refcnt in BlockDriverAIOCB
  block: Add bdrv_aio_cancel_async
  tests: Add testing code for bdrv_aio_cancel_async
  linux-aio: Implement .cancel_async
  thread-pool: Implement .cancel_async
  dma: Implement .cancel_async
  block: Implement bdrv_em_co_aiocb_info.cancel_async
  iscsi: Implement .cancel_async in acb info

 block.c                  | 45 ++++++++++++++++++++++++++++++++++++-
 block/iscsi.c            | 18 ++++++++++++---
 block/linux-aio.c        | 20 +++++++++++++++--
 dma-helpers.c            | 28 +++++++++++++++++++++++
 include/block/aio.h      |  3 +++
 include/block/block.h    |  1 +
 tests/test-thread-pool.c | 39 +++++++++++++++++++++++++-------
 thread-pool.c            | 58 +++++++++++++++++++++++++++++++++++++++---------
 8 files changed, 187 insertions(+), 25 deletions(-)

-- 
2.1.0

[Qemu-devel] [RFC PATCH v2 1/8] block: Add refcnt in BlockDriverAIOCB

[Fam Zheng]

This will be useful in situations like asynchronous cancel emulation.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 block.c             | 12 +++++++++++-
 include/block/aio.h |  2 ++
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/block.c b/block.c
index e9380f6..f8e342f 100644
--- a/block.c
+++ b/block.c
@@ -4860,13 +4860,23 @@ void *qemu_aio_get(const AIOCBInfo *aiocb_info, BlockDriverState *bs,
     acb->bs = bs;
     acb->cb = cb;
     acb->opaque = opaque;
+    acb->refcnt = 1;
     return acb;
 }
 
+void qemu_aio_ref(void *p)
+{
+    BlockDriverAIOCB *acb = p;
+    acb->refcnt++;
+}
+
 void qemu_aio_release(void *p)
 {
     BlockDriverAIOCB *acb = p;
-    g_slice_free1(acb->aiocb_info->aiocb_size, acb);
+    assert(acb->refcnt > 0);
+    if (--acb->refcnt == 0) {
+        g_slice_free1(acb->aiocb_info->aiocb_size, acb);
+    }
 }
 
 /**************************************************************/
diff --git a/include/block/aio.h b/include/block/aio.h
index c23de3c..8c216f6 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -35,11 +35,13 @@ struct BlockDriverAIOCB {
     BlockDriverState *bs;
     BlockDriverCompletionFunc *cb;
     void *opaque;
+    int refcnt;
 };
 
 void *qemu_aio_get(const AIOCBInfo *aiocb_info, BlockDriverState *bs,
                    BlockDriverCompletionFunc *cb, void *opaque);
 void qemu_aio_release(void *p);
+void qemu_aio_ref(void *p);
 
 typedef struct AioHandler AioHandler;
 typedef void QEMUBHFunc(void *opaque);
-- 
2.1.0

[Qemu-devel] [RFC PATCH v2 2/8] block: Add bdrv_aio_cancel_async

[Fam Zheng]

This is the async version of bdrv_aio_cancel, which doesn't block the
caller. It guarantees that the cb is called either before returning or
some time later.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 block.c               | 26 ++++++++++++++++++++++++++
 include/block/aio.h   |  1 +
 include/block/block.h |  1 +
 3 files changed, 28 insertions(+)

diff --git a/block.c b/block.c
index f8e342f..f4c77ec 100644
--- a/block.c
+++ b/block.c
@@ -4612,6 +4612,32 @@ void bdrv_aio_cancel(BlockDriverAIOCB *acb)
     acb->aiocb_info->cancel(acb);
 }
 
+static void bdrv_aio_cancel_cb_nop(void *opaque, int ret)
+{
+    /* nop */
+}
+
+/* Async version of aio cancel. The caller is not blocked if the acb implements
+ * cancel_async, otherwise fall back to bdrv_aio_cancel. In both cases, acb->cb
+ * is guarenteed to be called, before or after function returns. */
+void bdrv_aio_cancel_async(BlockDriverAIOCB *acb)
+{
+    if (acb->aiocb_info->cancel_async) {
+        acb->aiocb_info->cancel_async(acb);
+    } else {
+        /* Mask the cb and cancel, we'll call it manually once the synchronous
+         * cancel is done. */
+        BlockDriverCompletionFunc *cb = acb->cb;
+        void *opaque = acb->opaque;
+        acb->cb = bdrv_aio_cancel_cb_nop;
+        acb->opaque = NULL;
+        qemu_aio_ref(acb);
+        acb->aiocb_info->cancel(acb);
+        cb(opaque, -ECANCELED);
+        qemu_aio_release(acb);
+    }
+}
+
 /**************************************************************/
 /* async block device emulation */
 
diff --git a/include/block/aio.h b/include/block/aio.h
index 8c216f6..c434466 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -27,6 +27,7 @@ typedef void BlockDriverCompletionFunc(void *opaque, int ret);
 
 typedef struct AIOCBInfo {
     void (*cancel)(BlockDriverAIOCB *acb);
+    void (*cancel_async)(BlockDriverAIOCB *acb);
     size_t aiocb_size;
 } AIOCBInfo;
 
diff --git a/include/block/block.h b/include/block/block.h
index 8f4ad16..35a2448 100644
--- a/include/block/block.h
+++ b/include/block/block.h
@@ -336,6 +336,7 @@ BlockDriverAIOCB *bdrv_aio_discard(BlockDriverState *bs,
                                    int64_t sector_num, int nb_sectors,
                                    BlockDriverCompletionFunc *cb, void *opaque);
 void bdrv_aio_cancel(BlockDriverAIOCB *acb);
+void bdrv_aio_cancel_async(BlockDriverAIOCB *acb);
 
 typedef struct BlockRequest {
     /* Fields to be filled by multiwrite caller */
-- 
2.1.0

[Qemu-devel] [RFC PATCH v2 3/8] tests: Add testing code for bdrv_aio_cancel_async

[Fam Zheng]

Before, done_cb is only called if the request is submitted by thread
pool. Now in the async test, done_cb is always called. So update the
test criteria accordingly.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 tests/test-thread-pool.c | 39 +++++++++++++++++++++++++++++++--------
 1 file changed, 31 insertions(+), 8 deletions(-)

diff --git a/tests/test-thread-pool.c b/tests/test-thread-pool.c
index f40b7fc..5c08aa3 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_true(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,26 @@ 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);
+                active--;
+            } 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 +201,29 @@ 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);
+            if (sync) {
+                g_assert(data[i].aiocb != NULL);
+            } else {
+                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 +239,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();
 
-- 
2.1.0

[Qemu-devel] [RFC PATCH v2 4/8] linux-aio: Implement .cancel_async

[Fam Zheng]

Just call io_cancel (2), if it fails, it means the request is not
canceled, so the event loop will eventually call
qemu_laio_process_completion.

In qemu_laio_process_completion, change to call the cb unconditionally.
It is required by .cancel_async, and also acceptable by .cancel.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 block/linux-aio.c | 20 ++++++++++++++++++--
 1 file changed, 18 insertions(+), 2 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index 7ac7e8c..adf3b2e 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -79,9 +79,8 @@ static void qemu_laio_process_completion(struct qemu_laio_state *s,
                 ret = -EINVAL;
             }
         }
-
-        laiocb->common.cb(laiocb->common.opaque, ret);
     }
+    laiocb->common.cb(laiocb->common.opaque, ret);
 
     qemu_aio_release(laiocb);
 }
@@ -110,6 +109,22 @@ static void qemu_laio_completion_cb(EventNotifier *e)
     }
 }
 
+static void laio_cancel_async(BlockDriverAIOCB *blockacb)
+{
+    struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
+    struct io_event event;
+    int ret;
+
+    ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
+    laiocb->ret = -ECANCELED;
+    if (!ret) {
+        /* iocb is not cancelled, cb will be called by the event loop later */
+        return;
+    }
+
+    laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
+}
+
 static void laio_cancel(BlockDriverAIOCB *blockacb)
 {
     struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
@@ -145,6 +160,7 @@ static void laio_cancel(BlockDriverAIOCB *blockacb)
 static const AIOCBInfo laio_aiocb_info = {
     .aiocb_size         = sizeof(struct qemu_laiocb),
     .cancel             = laio_cancel,
+    .cancel_async       = laio_cancel_async,
 };
 
 static void ioq_init(LaioQueue *io_q)
-- 
2.1.0

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

[Fam Zheng]

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

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

[Paolo Bonzini]

Il 26/08/2014 08:08, Fam Zheng ha scritto:
> +    qemu_mutex_lock(&pool->lock);
> +    if (thread_pool_cancel_from_queue(elem)) {
> +        elem->state = THREAD_CANCELED_ASYNC;
> +    }

Can you simply set it to THREAD_DONE (and set elem->ret to -ECANCELED)?

Paolo

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

[Fam Zheng]

On Tue, 08/26 10:42, Paolo Bonzini wrote:
> Il 26/08/2014 08:08, Fam Zheng ha scritto:
> > +    qemu_mutex_lock(&pool->lock);
> > +    if (thread_pool_cancel_from_queue(elem)) {
> > +        elem->state = THREAD_CANCELED_ASYNC;
> > +    }
> 
> Can you simply set it to THREAD_DONE (and set elem->ret to -ECANCELED)?
> 

Yes, that should work.

Fam

[Qemu-devel] [RFC PATCH v2 6/8] dma: Implement .cancel_async

[Fam Zheng]

Just forward the request to bdrv_aio_cancel_async. Use a flag to fix the
ret value in completion code. Also check memory address before calling
dma_memory_unmap.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 dma-helpers.c | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/dma-helpers.c b/dma-helpers.c
index 499b52b..4a03e18 100644
--- a/dma-helpers.c
+++ b/dma-helpers.c
@@ -74,6 +74,7 @@ typedef struct {
     uint64_t sector_num;
     DMADirection dir;
     bool in_cancel;
+    bool cancelled;
     int sg_cur_index;
     dma_addr_t sg_cur_byte;
     QEMUIOVector iov;
@@ -105,6 +106,9 @@ static void dma_bdrv_unmap(DMAAIOCB *dbs)
     int i;
 
     for (i = 0; i < dbs->iov.niov; ++i) {
+        if (!(dbs->iov.iov[i].iov_base && dbs->iov.iov[i].iov_len)) {
+            break;
+        }
         dma_memory_unmap(dbs->sg->as, dbs->iov.iov[i].iov_base,
                          dbs->iov.iov[i].iov_len, dbs->dir,
                          dbs->iov.iov[i].iov_len);
@@ -116,6 +120,9 @@ static void dma_complete(DMAAIOCB *dbs, int ret)
 {
     trace_dma_complete(dbs, ret, dbs->common.cb);
 
+    if (dbs->cancelled) {
+        ret = -ECANCELED;
+    }
     dma_bdrv_unmap(dbs);
     if (dbs->common.cb) {
         dbs->common.cb(dbs->common.opaque, ret);
@@ -141,6 +148,9 @@ static void dma_bdrv_cb(void *opaque, int ret)
 
     trace_dma_bdrv_cb(dbs, ret);
 
+    if (dbs->cancelled) {
+        ret = -ECANCELED;
+    }
     dbs->acb = NULL;
     dbs->sector_num += dbs->iov.size / 512;
 
@@ -185,6 +195,7 @@ static void dma_aio_cancel(BlockDriverAIOCB *acb)
 
     trace_dma_aio_cancel(dbs);
 
+    dbs->cancelled = true;
     if (dbs->acb) {
         BlockDriverAIOCB *acb = dbs->acb;
         dbs->acb = NULL;
@@ -196,9 +207,25 @@ static void dma_aio_cancel(BlockDriverAIOCB *acb)
     dma_complete(dbs, 0);
 }
 
+static void dma_aio_cancel_async(BlockDriverAIOCB *acb)
+{
+    DMAAIOCB *dbs = container_of(acb, DMAAIOCB, common);
+
+    trace_dma_aio_cancel(dbs);
+
+    dbs->cancelled = true;
+    if (dbs->acb) {
+        acb = dbs->acb;
+        dbs->acb = NULL;
+        bdrv_aio_cancel_async(acb);
+    }
+}
+
+
 static const AIOCBInfo dma_aiocb_info = {
     .aiocb_size         = sizeof(DMAAIOCB),
     .cancel             = dma_aio_cancel,
+    .cancel_async       = dma_aio_cancel_async,
 };
 
 BlockDriverAIOCB *dma_bdrv_io(
@@ -218,6 +245,7 @@ BlockDriverAIOCB *dma_bdrv_io(
     dbs->sg_cur_byte = 0;
     dbs->dir = dir;
     dbs->in_cancel = false;
+    dbs->cancelled = false;
     dbs->io_func = io_func;
     dbs->bh = NULL;
     qemu_iovec_init(&dbs->iov, sg->nsg);
-- 
2.1.0

Re: [Qemu-devel] [RFC PATCH v2 6/8] dma: Implement .cancel_async

[Paolo Bonzini]

Il 26/08/2014 08:08, Fam Zheng ha scritto:
> +    if (dbs->cancelled) {
> +        ret = -ECANCELED;
> +    }

Why is dbs->cancelled necessary?

>      dma_bdrv_unmap(dbs);
>      if (dbs->common.cb) {
>          dbs->common.cb(dbs->common.opaque, ret);
> @@ -141,6 +148,9 @@ static void dma_bdrv_cb(void *opaque, int ret)
>  
>      trace_dma_bdrv_cb(dbs, ret);
>  
> +    if (dbs->cancelled) {
> +        ret = -ECANCELED;
> +    }
>      dbs->acb = NULL;
>      dbs->sector_num += dbs->iov.size / 512;
>  
> @@ -185,6 +195,7 @@ static void dma_aio_cancel(BlockDriverAIOCB *acb)
>  
>      trace_dma_aio_cancel(dbs);
>  
> +    dbs->cancelled = true;
>      if (dbs->acb) {
>          BlockDriverAIOCB *acb = dbs->acb;
>          dbs->acb = NULL;
> @@ -196,9 +207,25 @@ static void dma_aio_cancel(BlockDriverAIOCB *acb)
>      dma_complete(dbs, 0);
>  }
>  
> +static void dma_aio_cancel_async(BlockDriverAIOCB *acb)
> +{
> +    DMAAIOCB *dbs = container_of(acb, DMAAIOCB, common);
> +
> +    trace_dma_aio_cancel(dbs);
> +
> +    dbs->cancelled = true;
> +    if (dbs->acb) {
> +        acb = dbs->acb;
> +        dbs->acb = NULL;

Why do you need to set dbs->acb to NULL, since the callback is going to
be called?

Paolo

> +        bdrv_aio_cancel_async(acb);
> +    }
> +}

Re: [Qemu-devel] [RFC PATCH v2 6/8] dma: Implement .cancel_async

[Fam Zheng]

On Tue, 08/26 10:46, Paolo Bonzini wrote:
> Il 26/08/2014 08:08, Fam Zheng ha scritto:
> > +    if (dbs->cancelled) {
> > +        ret = -ECANCELED;
> > +    }
> 
> Why is dbs->cancelled necessary?

Request may complete after bdrv_aio_cancel_async with other status, this flag
is checked to fix the status to -ECANCELED.

> 
> >      dma_bdrv_unmap(dbs);
> >      if (dbs->common.cb) {
> >          dbs->common.cb(dbs->common.opaque, ret);
> > @@ -141,6 +148,9 @@ static void dma_bdrv_cb(void *opaque, int ret)
> >  
> >      trace_dma_bdrv_cb(dbs, ret);
> >  
> > +    if (dbs->cancelled) {
> > +        ret = -ECANCELED;
> > +    }
> >      dbs->acb = NULL;
> >      dbs->sector_num += dbs->iov.size / 512;
> >  
> > @@ -185,6 +195,7 @@ static void dma_aio_cancel(BlockDriverAIOCB *acb)
> >  
> >      trace_dma_aio_cancel(dbs);
> >  
> > +    dbs->cancelled = true;
> >      if (dbs->acb) {
> >          BlockDriverAIOCB *acb = dbs->acb;
> >          dbs->acb = NULL;
> > @@ -196,9 +207,25 @@ static void dma_aio_cancel(BlockDriverAIOCB *acb)
> >      dma_complete(dbs, 0);
> >  }
> >  
> > +static void dma_aio_cancel_async(BlockDriverAIOCB *acb)
> > +{
> > +    DMAAIOCB *dbs = container_of(acb, DMAAIOCB, common);
> > +
> > +    trace_dma_aio_cancel(dbs);
> > +
> > +    dbs->cancelled = true;
> > +    if (dbs->acb) {
> > +        acb = dbs->acb;
> > +        dbs->acb = NULL;
> 
> Why do you need to set dbs->acb to NULL, since the callback is going to
> be called?

Just copied from dma_aio_cancel, but seems not particularly useful. It reminds
me that the one in dma_aio_cancel also looks suspicious.

Fam

Re: [Qemu-devel] [RFC PATCH v2 6/8] dma: Implement .cancel_async

[Paolo Bonzini]

Il 26/08/2014 11:21, Fam Zheng ha scritto:
> On Tue, 08/26 10:46, Paolo Bonzini wrote:
>> Il 26/08/2014 08:08, Fam Zheng ha scritto:
>>> +    if (dbs->cancelled) {
>>> +        ret = -ECANCELED;
>>> +    }
>>
>> Why is dbs->cancelled necessary?
> 
> Request may complete after bdrv_aio_cancel_async with other status, this flag
> is checked to fix the status to -ECANCELED.

Ah, that's because the operation could be partly incomplete?

I think it would be better to call dma_complete with -ECANCELED, instead
of doing the same dbs->cancelled check twice.  For example, in dma_bdrv_cb:

    if (dbs->sg_cur_index == dbs->sg->nsg || ret < 0) {
        dma_complete(dbs, ret);
        return;
    }
    if (dbs->cancelled) {
        dma_complete(dbs, -ECANCELED);
        return;
    }

>>> +static void dma_aio_cancel_async(BlockDriverAIOCB *acb)
>>> +{
>>> +    DMAAIOCB *dbs = container_of(acb, DMAAIOCB, common);
>>> +
>>> +    trace_dma_aio_cancel(dbs);
>>> +
>>> +    dbs->cancelled = true;
>>> +    if (dbs->acb) {
>>> +        acb = dbs->acb;
>>> +        dbs->acb = NULL;
>>
>> Why do you need to set dbs->acb to NULL, since the callback is going to
>> be called?
> 
> Just copied from dma_aio_cancel, but seems not particularly useful. It reminds
> me that the one in dma_aio_cancel also looks suspicious.

Yeah, that one looks useless too...

Paolo

[Qemu-devel] [RFC PATCH v2 7/8] block: Implement bdrv_em_co_aiocb_info.cancel_async

[Fam Zheng]

Nothing to do here.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 block.c | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/block.c b/block.c
index f4c77ec..ac1cfb4 100644
--- a/block.c
+++ b/block.c
@@ -4746,9 +4746,16 @@ static void bdrv_aio_co_cancel_em(BlockDriverAIOCB *blockacb)
     }
 }
 
+static void bdrv_aio_cancel_em_async(BlockDriverAIOCB *blockacb)
+{
+    /* A nop async cancel will just work for us, because later the request will
+     * complete in caller's coroutine. */
+}
+
 static const AIOCBInfo bdrv_em_co_aiocb_info = {
     .aiocb_size         = sizeof(BlockDriverAIOCBCoroutine),
     .cancel             = bdrv_aio_co_cancel_em,
+    .cancel_async       = bdrv_aio_cancel_em_async,
 };
 
 static void bdrv_co_em_bh(void *opaque)
-- 
2.1.0

[Qemu-devel] [RFC PATCH v2 8/8] iscsi: Implement .cancel_async in acb info

[Fam Zheng]

Factor out the cancellation submission part in iscsi_aio_cancel, which
is all what is needed for .cancel_async.

The cb passed to iscsi_task_mgmt_abort_task_async, iscsi_abort_task_cb,
will be called later, so for iscsi_bh_cb. In iscsi_bh_cb, acb->canceled
is not set if canceled by .cancel_async, so the completion cb is also
called.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 block/iscsi.c | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/block/iscsi.c b/block/iscsi.c
index 3e19202..9ea515f 100644
--- a/block/iscsi.c
+++ b/block/iscsi.c
@@ -121,6 +121,8 @@ iscsi_bh_cb(void *p)
     acb->buf = NULL;
 
     if (acb->canceled == 0) {
+        /* Either normally completed or cancelled by iscsi_aio_cancel_async,
+         * let's call the cb. */
         acb->common.cb(acb->common.opaque, acb->status);
     }
 
@@ -231,7 +233,7 @@ iscsi_abort_task_cb(struct iscsi_context *iscsi, int status, void *command_data,
 }
 
 static void
-iscsi_aio_cancel(BlockDriverAIOCB *blockacb)
+iscsi_aio_cancel_async(BlockDriverAIOCB *blockacb)
 {
     IscsiAIOCB *acb = (IscsiAIOCB *)blockacb;
     IscsiLun *iscsilun = acb->iscsilun;
@@ -240,12 +242,21 @@ iscsi_aio_cancel(BlockDriverAIOCB *blockacb)
         return;
     }
 
-    acb->canceled = 1;
-
     /* send a task mgmt call to the target to cancel the task on the target */
     iscsi_task_mgmt_abort_task_async(iscsilun->iscsi, acb->task,
                                      iscsi_abort_task_cb, acb);
 
+}
+
+static void
+iscsi_aio_cancel(BlockDriverAIOCB *blockacb)
+{
+    IscsiAIOCB *acb = (IscsiAIOCB *)blockacb;
+    IscsiLun *iscsilun = acb->iscsilun;
+
+    iscsi_aio_cancel_async(blockacb);
+
+    acb->canceled = 1;
     while (acb->status == -EINPROGRESS) {
         aio_poll(iscsilun->aio_context, true);
     }
@@ -254,6 +265,7 @@ iscsi_aio_cancel(BlockDriverAIOCB *blockacb)
 static const AIOCBInfo iscsi_aiocb_info = {
     .aiocb_size         = sizeof(IscsiAIOCB),
     .cancel             = iscsi_aio_cancel,
+    .cancel_async       = iscsi_aio_cancel_async,
 };
 
 
-- 
2.1.0