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

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

[Fam Zheng]

v3: Drop "RFC".
    Improvements according to Paolo's comments:
    05: Just use THREAD_DONE and ret = -ECANCELED in thread-pool.c
    06: Don't check dbs->cancelled for twice.
        Don't set dbs->acb to NULL.

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 layer 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 send a tmf
command to cancel the request, at which point we will busy wait in
bdrv_aio_cancel, until the request is slowly spit out from throttled_reqs.

Later, we will 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            | 24 ++++++++++++++++++++++++
 include/block/aio.h      |  3 +++
 include/block/block.h    |  1 +
 tests/test-thread-pool.c | 39 +++++++++++++++++++++++++++++++--------
 thread-pool.c            | 44 +++++++++++++++++++++++++++++++++++---------
 8 files changed, 171 insertions(+), 23 deletions(-)

-- 
2.1.0

[Qemu-devel] [PATCH v3 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] [PATCH v3 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

Re: [Qemu-devel] [PATCH v3 2/8] block: Add bdrv_aio_cancel_async

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1155 bytes --]

On Wed, Aug 27, 2014 at 10:49:10AM +0800, Fam Zheng wrote:
> +/* 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);
> +    }
> +}

It is not totally obvious why we hijack the callback.  If you respin,
please rephrase the comment along the lines of:

/* bdrv_aio_cancel() does not guarantee to invoke cb() so mask it during
 * bdrv_aio_cancel() and always invoke it ourselves.
 */

Stefan

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

[Qemu-devel] [PATCH v3 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] [PATCH v3 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

Re: [Qemu-devel] [PATCH v3 4/8] linux-aio: Implement .cancel_async

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 730 bytes --]

On Wed, Aug 27, 2014 at 10:49:12AM +0800, Fam Zheng wrote:
> @@ -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;
> +    }

This comment doesn't make sense, io_cancel() returns 0 on success so it
has been cancelled.  We need to call the completion callback!

Stefan

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

[Qemu-devel] [PATCH v3 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 | 44 +++++++++++++++++++++++++++++++++++---------
 1 file changed, 35 insertions(+), 9 deletions(-)

diff --git a/thread-pool.c b/thread-pool.c
index 23888dc..9cb7a1d 100644
--- a/thread-pool.c
+++ b/thread-pool.c
@@ -202,6 +202,39 @@ 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_DONE;
+        elem->ret = -ECANCELED;
+    }
+
+    qemu_mutex_unlock(&pool->lock);
+}
+
 static void thread_pool_cancel(BlockDriverAIOCB *acb)
 {
     ThreadPoolElement *elem = (ThreadPoolElement *)acb;
@@ -210,16 +243,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 +259,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

[Qemu-devel] [PATCH v3 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 | 24 ++++++++++++++++++++++++
 1 file changed, 24 insertions(+)

diff --git a/dma-helpers.c b/dma-helpers.c
index 499b52b..f9c1a79 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);
@@ -141,6 +145,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 +192,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 +204,24 @@ 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;
+        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 +241,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

[Qemu-devel] [PATCH v3 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] [PATCH v3 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

Re: [Qemu-devel] [PATCH v3 0/8] block: Asynchronous request cancellation

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 2533 bytes --]

On Wed, Aug 27, 2014 at 10:49:08AM +0800, Fam Zheng wrote:
> v3: Drop "RFC".
>     Improvements according to Paolo's comments:
>     05: Just use THREAD_DONE and ret = -ECANCELED in thread-pool.c
>     06: Don't check dbs->cancelled for twice.
>         Don't set dbs->acb to NULL.
> 
> 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 layer 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 send a tmf
> command to cancel the request, at which point we will busy wait in
> bdrv_aio_cancel, until the request is slowly spit out from throttled_reqs.
> 
> Later, we will change scsi device code to make this asynchronous, on top of
> bdrv_aio_cancel_async.

We need to get rid of .bdrv_aio_cancel().  Keeping both
.bdrv_aio_cancel() and .bdrv_aio_cancel_async() around is problematic
because they have slightly different semantics.

This patch series makes block driver cancellation more complex because
we support both approaches :(.

Could we do something like:

void bdrv_aio_cancel(BdrvAIOCB *acb)
{
    bdrv_aiocb_ref(acb);
    bdrv_aio_cancel_async(acb);
    while (acb->ref > 1) {
        aio_poll(bdrv_get_aio_context(acb->bs), true);
    }
    bdrv_aiocb_release(acb);
}

(pseudo-code)

Then .bdrv_aio_cancel() should be deleted.

Stefan

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH v3 0/8] block: Asynchronous request cancellation

[Fam Zheng]

On Tue, 09/02 16:09, Stefan Hajnoczi wrote:
> On Wed, Aug 27, 2014 at 10:49:08AM +0800, Fam Zheng wrote:
> > v3: Drop "RFC".
> >     Improvements according to Paolo's comments:
> >     05: Just use THREAD_DONE and ret = -ECANCELED in thread-pool.c
> >     06: Don't check dbs->cancelled for twice.
> >         Don't set dbs->acb to NULL.
> > 
> > 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 layer 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 send a tmf
> > command to cancel the request, at which point we will busy wait in
> > bdrv_aio_cancel, until the request is slowly spit out from throttled_reqs.
> > 
> > Later, we will change scsi device code to make this asynchronous, on top of
> > bdrv_aio_cancel_async.
> 
> We need to get rid of .bdrv_aio_cancel().  Keeping both
> .bdrv_aio_cancel() and .bdrv_aio_cancel_async() around is problematic
> because they have slightly different semantics.
> 
> This patch series makes block driver cancellation more complex because
> we support both approaches :(.
> 
> Could we do something like:
> 
> void bdrv_aio_cancel(BdrvAIOCB *acb)
> {
>     bdrv_aiocb_ref(acb);
>     bdrv_aio_cancel_async(acb);
>     while (acb->ref > 1) {
>         aio_poll(bdrv_get_aio_context(acb->bs), true);
>     }
>     bdrv_aiocb_release(acb);
> }
> 
> (pseudo-code)
> 
> Then .bdrv_aio_cancel() should be deleted.
> 

OK, I will drop .io_cancel field from AIOCBInfo, and convert all AIO
implementations to only support .io_cancel_async. That way we can emulate
bdrv_aio_cancel with bdrv_aio_cancel_async.

Thanks for reviewing!

Fam