[Qemu-devel] [RFC PATCH 0/9] scsi, block: Asynchronous request cancellation

[Qemu-devel] [RFC PATCH 0/9] scsi, block: Asynchronous request cancellation

[Fam Zheng]

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.

Because of this, 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.

The first patches introduces the interface and a (default) degradad
implementation, which is emulated with bdrv_aio_cancel.

Then some testing on it is added in patch 2.

Following patches are the implementation of .cancel_async in several AIOCBInfo
structures. Note that blkldebug's and iscsi's implementations are not tested,
just for the purpose to see what it takes to implement it. In the long term, we
may consider completely dropping bdrv_aio_cancel, if it turns out the
asynchronous version works better. The callers are not many, and it feels
counterintuitive to have a synchronous bdrv_aio_* function.

In the end, scsi emulation code is shifted to the async cancelling.

One major benefit is 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 no longer need to block the whole
vm with a busy wait polling loop, which is how bdrv_aio_cancel is implemented
now.

To test this series, you can throttle a scsi-disk to a very low limit, for
example 50 bps, then stress the guest block device with dd or fio.

Before, 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.

After, the guest is not blocked, so the responsiveness of the VM is much
better.

The same applies for a disappeared/hung (nfs, iscsi, etc.) storage target.

Last but not least, the remaing users of bdrv_aio_cancel, after this series,
are only 4:

    block/quorum.c
    hw/block/nvme.c
    hw/ide/ahci.c
    hw/ide/core.c

I said too many things that you already knew. Please review and make any
comments! Thank you very much!

Fam


Fam Zheng (9):
  block: Add bdrv_aio_cancel_async
  tests: Add testing code for bdrv_aio_cancel_async
  iscsi: Implement .cancel_async in acb info
  linux-aio: Implement .cancel_async
  thread-pool: Implement .cancel_async
  blkdebug: Implement .cancel_async
  dma: Implement .cancel_async
  block: Implement stub bdrv_em_co_aiocb_info.cancel_async
  scsi: Cancel request asynchronously

 block.c                  | 40 +++++++++++++++++++++++++++++++++
 block/blkdebug.c         | 20 +++++++++++------
 block/iscsi.c            | 18 ++++++++++++---
 block/linux-aio.c        | 20 +++++++++++++++--
 dma-helpers.c            | 28 +++++++++++++++++++++++
 hw/scsi/scsi-bus.c       |  6 ++---
 hw/scsi/scsi-disk.c      | 41 +++++++++-------------------------
 hw/scsi/scsi-generic.c   | 21 +++++-------------
 include/block/aio.h      |  2 ++
 include/block/block.h    |  1 +
 tests/test-thread-pool.c | 39 +++++++++++++++++++++++++-------
 thread-pool.c            | 58 +++++++++++++++++++++++++++++++++++++++---------
 12 files changed, 213 insertions(+), 81 deletions(-)

-- 
2.0.3

[Qemu-devel] [RFC PATCH 1/9] 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               | 35 +++++++++++++++++++++++++++++++++++
 include/block/aio.h   |  2 ++
 include/block/block.h |  1 +
 3 files changed, 38 insertions(+)

diff --git a/block.c b/block.c
index 6fa0201..1cc8926 100644
--- a/block.c
+++ b/block.c
@@ -4607,6 +4607,41 @@ void bdrv_aio_cancel(BlockDriverAIOCB *acb)
     acb->aiocb_info->cancel(acb);
 }
 
+static void bdrv_aio_cancel_async_cb(void *opaque, int ret)
+{
+    BlockDriverAIOCB *acb = opaque;
+
+    acb->cb     = acb->cancel_acb_save->cb;
+    acb->opaque = acb->cancel_acb_save->opaque;
+    g_free(acb->cancel_acb_save);
+    acb->cancel_acb_save = NULL;
+    acb->cb(acb->opaque, -ECANCELED);
+}
+
+/* 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 {
+        BlockDriverAIOCB *save = g_new(BlockDriverAIOCB, 1);
+        save->cb               = acb->cb;
+        save->opaque           = acb->opaque;
+        acb->cb                = bdrv_aio_cancel_async_cb;
+        acb->opaque            = acb;
+        acb->cancel_acb_save   = save;
+
+        /* Use the synchronous version and hope our cb is called. */
+        acb->aiocb_info->cancel(acb);
+        if (acb->cancel_acb_save) {
+            /* cb is not called yet, let's call it */
+            bdrv_aio_cancel_async_cb(acb->opaque, -ECANCELED);
+        }
+    }
+}
+
 /**************************************************************/
 /* async block device emulation */
 
diff --git a/include/block/aio.h b/include/block/aio.h
index c23de3c..fcc5c87 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;
 
@@ -35,6 +36,7 @@ struct BlockDriverAIOCB {
     BlockDriverState *bs;
     BlockDriverCompletionFunc *cb;
     void *opaque;
+    BlockDriverAIOCB *cancel_acb_save;
 };
 
 void *qemu_aio_get(const AIOCBInfo *aiocb_info, BlockDriverState *bs,
diff --git a/include/block/block.h b/include/block/block.h
index e94b701..ed2a3a8 100644
--- a/include/block/block.h
+++ b/include/block/block.h
@@ -335,6 +335,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.0.3

[Qemu-devel] [RFC PATCH 2/9] 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.0.3

[Qemu-devel] [RFC PATCH 3/9] 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 2c9cfc1..8b27a5c 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.0.3

[Qemu-devel] [RFC PATCH 4/9] 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.0.3

[Qemu-devel] [RFC PATCH 5/9] 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.0.3

[Qemu-devel] [RFC PATCH 6/9] blkdebug: Implement .cancel_async

[Fam Zheng]

Very similar to .cancel, except that cb is called before releasing the
aio.

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

diff --git a/block/blkdebug.c b/block/blkdebug.c
index 1586ed9..d269956 100644
--- a/block/blkdebug.c
+++ b/block/blkdebug.c
@@ -48,13 +48,6 @@ typedef struct BlkdebugSuspendedReq {
     QLIST_ENTRY(BlkdebugSuspendedReq) next;
 } BlkdebugSuspendedReq;
 
-static void blkdebug_aio_cancel(BlockDriverAIOCB *blockacb);
-
-static const AIOCBInfo blkdebug_aiocb_info = {
-    .aiocb_size = sizeof(BlkdebugAIOCB),
-    .cancel     = blkdebug_aio_cancel,
-};
-
 enum {
     ACTION_INJECT_ERROR,
     ACTION_SET_STATE,
@@ -446,12 +439,25 @@ static void error_callback_bh(void *opaque)
     qemu_aio_release(acb);
 }
 
+static void blkdebug_aio_cancel_async(BlockDriverAIOCB *blockacb)
+{
+    BlkdebugAIOCB *acb = container_of(blockacb, BlkdebugAIOCB, common);
+    blockacb->cb(blockacb->opaque, -ECANCELED);
+    qemu_aio_release(acb);
+}
+
 static void blkdebug_aio_cancel(BlockDriverAIOCB *blockacb)
 {
     BlkdebugAIOCB *acb = container_of(blockacb, BlkdebugAIOCB, common);
     qemu_aio_release(acb);
 }
 
+static const AIOCBInfo blkdebug_aiocb_info = {
+    .aiocb_size   = sizeof(BlkdebugAIOCB),
+    .cancel       = blkdebug_aio_cancel,
+    .cancel_async = blkdebug_aio_cancel_async,
+};
+
 static BlockDriverAIOCB *inject_error(BlockDriverState *bs,
     BlockDriverCompletionFunc *cb, void *opaque, BlkdebugRule *rule)
 {
-- 
2.0.3

[Qemu-devel] [RFC PATCH 7/9] 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.0.3

[Qemu-devel] [RFC PATCH 8/9] block: Implement stub bdrv_em_co_aiocb_info.cancel_async

[Fam Zheng]

Nothing to do here.

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

diff --git a/block.c b/block.c
index 1cc8926..1860e4b 100644
--- a/block.c
+++ b/block.c
@@ -4750,9 +4750,14 @@ static void bdrv_aio_co_cancel_em(BlockDriverAIOCB *blockacb)
     }
 }
 
+static void bdrv_aio_co_cancel_em_async(BlockDriverAIOCB *blockacb)
+{
+}
+
 static const AIOCBInfo bdrv_em_co_aiocb_info = {
     .aiocb_size         = sizeof(BlockDriverAIOCBCoroutine),
     .cancel             = bdrv_aio_co_cancel_em,
+    .cancel_async       = bdrv_aio_co_cancel_em_async,
 };
 
 static void bdrv_co_em_bh(void *opaque)
-- 
2.0.3

[Qemu-devel] [RFC PATCH 9/9] scsi: Cancel request asynchronously

[Fam Zheng]

We are blocking the whole VM, which means that an irresponsive storage
backend will hang the whole guest. Let's switch to bdrv_aio_cancel_async
to improve this.

Signed-off-by: Fam Zheng <famz@redhat.com>
---
 hw/scsi/scsi-bus.c     |  6 ++----
 hw/scsi/scsi-disk.c    | 41 ++++++++++-------------------------------
 hw/scsi/scsi-generic.c | 21 ++++++---------------
 3 files changed, 18 insertions(+), 50 deletions(-)

diff --git a/hw/scsi/scsi-bus.c b/hw/scsi/scsi-bus.c
index 6f4462b..59ec9f9 100644
--- a/hw/scsi/scsi-bus.c
+++ b/hw/scsi/scsi-bus.c
@@ -1615,7 +1615,6 @@ void scsi_req_continue(SCSIRequest *req)
 {
     if (req->io_canceled) {
         trace_scsi_req_continue_canceled(req->dev->id, req->lun, req->tag);
-        return;
     }
     trace_scsi_req_continue(req->dev->id, req->lun, req->tag);
     if (req->cmd.mode == SCSI_XFER_TO_DEV) {
@@ -1633,7 +1632,6 @@ void scsi_req_data(SCSIRequest *req, int len)
     uint8_t *buf;
     if (req->io_canceled) {
         trace_scsi_req_data_canceled(req->dev->id, req->lun, req->tag, len);
-        return;
     }
     trace_scsi_req_data(req->dev->id, req->lun, req->tag, len);
     assert(req->cmd.mode != SCSI_XFER_NONE);
@@ -1721,7 +1719,7 @@ void scsi_req_complete(SCSIRequest *req, int status)
 void scsi_req_cancel(SCSIRequest *req)
 {
     trace_scsi_req_cancel(req->dev->id, req->lun, req->tag);
-    if (!req->enqueued) {
+    if (req->io_canceled) {
         return;
     }
     scsi_req_ref(req);
@@ -1738,7 +1736,7 @@ void scsi_req_cancel(SCSIRequest *req)
 
 void scsi_req_abort(SCSIRequest *req, int status)
 {
-    if (!req->enqueued) {
+    if (req->io_canceled) {
         return;
     }
     scsi_req_ref(req);
diff --git a/hw/scsi/scsi-disk.c b/hw/scsi/scsi-disk.c
index d55521d..46b1e53 100644
--- a/hw/scsi/scsi-disk.c
+++ b/hw/scsi/scsi-disk.c
@@ -112,14 +112,9 @@ static void scsi_cancel_io(SCSIRequest *req)
 
     DPRINTF("Cancel tag=0x%x\n", req->tag);
     if (r->req.aiocb) {
-        bdrv_aio_cancel(r->req.aiocb);
-
-        /* This reference was left in by scsi_*_data.  We take ownership of
-         * it the moment scsi_req_cancel is called, independent of whether
-         * bdrv_aio_cancel completes the request or not.  */
-        scsi_req_unref(&r->req);
+        assert(req->io_canceled);
+        bdrv_aio_cancel_async(r->req.aiocb);
     }
-    r->req.aiocb = NULL;
 }
 
 static uint32_t scsi_init_iovec(SCSIDiskReq *r, size_t size)
@@ -197,9 +192,7 @@ static void scsi_aio_complete(void *opaque, int ret)
     scsi_req_complete(&r->req, GOOD);
 
 done:
-    if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
-    }
+    scsi_req_unref(&r->req);
 }
 
 static bool scsi_is_cmd_fua(SCSICommand *cmd)
@@ -245,9 +238,7 @@ static void scsi_write_do_fua(SCSIDiskReq *r)
     scsi_req_complete(&r->req, GOOD);
 
 done:
-    if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
-    }
+    scsi_req_unref(&r->req);
 }
 
 static void scsi_dma_complete_noio(void *opaque, int ret)
@@ -279,9 +270,7 @@ static void scsi_dma_complete_noio(void *opaque, int ret)
     }
 
 done:
-    if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
-    }
+    scsi_req_unref(&r->req);
 }
 
 static void scsi_dma_complete(void *opaque, int ret)
@@ -319,9 +308,7 @@ static void scsi_read_complete(void * opaque, int ret)
     scsi_req_data(&r->req, r->qiov.size);
 
 done:
-    if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
-    }
+    scsi_req_unref(&r->req);
 }
 
 /* Actually issue a read to the block device.  */
@@ -361,9 +348,7 @@ static void scsi_do_read(void *opaque, int ret)
     }
 
 done:
-    if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
-    }
+    scsi_req_unref(&r->req);
 }
 
 /* Read more data from scsi device into buffer.  */
@@ -478,9 +463,7 @@ static void scsi_write_complete(void * opaque, int ret)
     }
 
 done:
-    if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
-    }
+    scsi_req_unref(&r->req);
 }
 
 static void scsi_write_data(SCSIRequest *req)
@@ -1577,9 +1560,7 @@ static void scsi_unmap_complete(void *opaque, int ret)
     scsi_req_complete(&r->req, GOOD);
 
 done:
-    if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
-    }
+    scsi_req_unref(&r->req);
     g_free(data);
 }
 
@@ -1672,9 +1653,7 @@ static void scsi_write_same_complete(void *opaque, int ret)
     scsi_req_complete(&r->req, GOOD);
 
 done:
-    if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
-    }
+    scsi_req_unref(&r->req);
     qemu_vfree(data->iov.iov_base);
     g_free(data);
 }
diff --git a/hw/scsi/scsi-generic.c b/hw/scsi/scsi-generic.c
index 0b2ff90..1c29ecb 100644
--- a/hw/scsi/scsi-generic.c
+++ b/hw/scsi/scsi-generic.c
@@ -132,27 +132,20 @@ static void scsi_command_complete(void *opaque, int ret)
     DPRINTF("Command complete 0x%p tag=0x%x status=%d\n",
             r, r->req.tag, status);
 
-    scsi_req_complete(&r->req, status);
     if (!r->req.io_canceled) {
-        scsi_req_unref(&r->req);
+        scsi_req_complete(&r->req, status);
     }
+    scsi_req_unref(&r->req);
 }
 
 /* Cancel a pending data transfer.  */
 static void scsi_cancel_io(SCSIRequest *req)
 {
-    SCSIGenericReq *r = DO_UPCAST(SCSIGenericReq, req, req);
-
     DPRINTF("Cancel tag=0x%x\n", req->tag);
-    if (r->req.aiocb) {
-        bdrv_aio_cancel(r->req.aiocb);
-
-        /* This reference was left in by scsi_*_data.  We take ownership of
-         * it independent of whether bdrv_aio_cancel completes the request
-         * or not.  */
-        scsi_req_unref(&r->req);
+    if (req->aiocb) {
+        req->io_canceled = true;
+        bdrv_aio_cancel_async(req->aiocb);
     }
-    r->req.aiocb = NULL;
 }
 
 static int execute_command(BlockDriverState *bdrv,
@@ -211,9 +204,7 @@ static void scsi_read_complete(void * opaque, int ret)
         bdrv_set_guest_block_size(s->conf.bs, s->blocksize);
 
         scsi_req_data(&r->req, len);
-        if (!r->req.io_canceled) {
-            scsi_req_unref(&r->req);
-        }
+        scsi_req_unref(&r->req);
     }
 }
 
-- 
2.0.3

Re: [Qemu-devel] [RFC PATCH 1/9] block: Add bdrv_aio_cancel_async

[Paolo Bonzini]

Il 21/08/2014 13:56, Fam Zheng ha scritto:
> +        BlockDriverAIOCB *save = g_new(BlockDriverAIOCB, 1);
> +        save->cb               = acb->cb;
> +        save->opaque           = acb->opaque;
> +        acb->cb                = bdrv_aio_cancel_async_cb;
> +        acb->opaque            = acb;
> +        acb->cancel_acb_save   = save;

No need for this extra field.  You can make a struct with {acb->cb,
acb->opaque, acb} and pass it as the opaque.  You also do not need to
allocate it on the heap, since everything is synchronous.

> +
> +        /* Use the synchronous version and hope our cb is called. */
> +        acb->aiocb_info->cancel(acb);

Unfortunately, acb has been freed at this point, so you'll be accessing
a dangling pointer.  I think you need to modify all cancel
implementations.  For example:

- return whether they have called the callback

- only free the acb if they have called the callback

- otherwise, free the acb in bdrv_aio_cancel

Paolo

> +        if (acb->cancel_acb_save) {
> +            /* cb is not called yet, let's call it */
> +            bdrv_aio_cancel_async_cb(acb->opaque, -ECANCELED);
> +        }
> +    }
> +}
> +
>  /**************************************************************/
>  /* async block device emulation */
>  
> diff --git a/include/block/aio.h b/include/block/aio.h
> index c23de3c..fcc5c87 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;
>  
> @@ -35,6 +36,7 @@ struct BlockDriverAIOCB {
>      BlockDriverState *bs;
>      BlockDriverCompletionFunc *cb;
>      void *opaque;
> +    BlockDriverAIOCB *cancel_acb_save;
>  };

Re: [Qemu-devel] [RFC PATCH 9/9] scsi: Cancel request asynchronously

[Paolo Bonzini]

Il 21/08/2014 13:56, Fam Zheng ha scritto:
> We are blocking the whole VM, which means that an irresponsive storage
> backend will hang the whole guest. Let's switch to bdrv_aio_cancel_async
> to improve this.

Unforuntately, the TMF must only return after the request has been
canceled.  I think you need to add a scsi_cancel_io_async function, and
keep all the remaining machinery (also, you need a better commit message
that explains what you are removing and the new invariants).

Then in virtio-scsi you need to add a list of "dependent" (controlq)
VirtIOSCSIReq to the "main" (requestq) VirtIOSCSIReq, and complete them
all after signaling the completion of the main request.

Paolo

> Signed-off-by: Fam Zheng <famz@redhat.com>
> ---
>  hw/scsi/scsi-bus.c     |  6 ++----
>  hw/scsi/scsi-disk.c    | 41 ++++++++++-------------------------------
>  hw/scsi/scsi-generic.c | 21 ++++++---------------
>  3 files changed, 18 insertions(+), 50 deletions(-)
> 
> diff --git a/hw/scsi/scsi-bus.c b/hw/scsi/scsi-bus.c
> index 6f4462b..59ec9f9 100644
> --- a/hw/scsi/scsi-bus.c
> +++ b/hw/scsi/scsi-bus.c
> @@ -1615,7 +1615,6 @@ void scsi_req_continue(SCSIRequest *req)
>  {
>      if (req->io_canceled) {
>          trace_scsi_req_continue_canceled(req->dev->id, req->lun, req->tag);
> -        return;
>      }
>      trace_scsi_req_continue(req->dev->id, req->lun, req->tag);
>      if (req->cmd.mode == SCSI_XFER_TO_DEV) {
> @@ -1633,7 +1632,6 @@ void scsi_req_data(SCSIRequest *req, int len)
>      uint8_t *buf;
>      if (req->io_canceled) {
>          trace_scsi_req_data_canceled(req->dev->id, req->lun, req->tag, len);
> -        return;
>      }
>      trace_scsi_req_data(req->dev->id, req->lun, req->tag, len);
>      assert(req->cmd.mode != SCSI_XFER_NONE);
> @@ -1721,7 +1719,7 @@ void scsi_req_complete(SCSIRequest *req, int status)
>  void scsi_req_cancel(SCSIRequest *req)
>  {
>      trace_scsi_req_cancel(req->dev->id, req->lun, req->tag);
> -    if (!req->enqueued) {
> +    if (req->io_canceled) {
>          return;
>      }
>      scsi_req_ref(req);
> @@ -1738,7 +1736,7 @@ void scsi_req_cancel(SCSIRequest *req)
>  
>  void scsi_req_abort(SCSIRequest *req, int status)
>  {
> -    if (!req->enqueued) {
> +    if (req->io_canceled) {
>          return;
>      }
>      scsi_req_ref(req);
> diff --git a/hw/scsi/scsi-disk.c b/hw/scsi/scsi-disk.c
> index d55521d..46b1e53 100644
> --- a/hw/scsi/scsi-disk.c
> +++ b/hw/scsi/scsi-disk.c
> @@ -112,14 +112,9 @@ static void scsi_cancel_io(SCSIRequest *req)
>  
>      DPRINTF("Cancel tag=0x%x\n", req->tag);
>      if (r->req.aiocb) {
> -        bdrv_aio_cancel(r->req.aiocb);
> -
> -        /* This reference was left in by scsi_*_data.  We take ownership of
> -         * it the moment scsi_req_cancel is called, independent of whether
> -         * bdrv_aio_cancel completes the request or not.  */
> -        scsi_req_unref(&r->req);
> +        assert(req->io_canceled);
> +        bdrv_aio_cancel_async(r->req.aiocb);
>      }
> -    r->req.aiocb = NULL;
>  }
>  
>  static uint32_t scsi_init_iovec(SCSIDiskReq *r, size_t size)
> @@ -197,9 +192,7 @@ static void scsi_aio_complete(void *opaque, int ret)
>      scsi_req_complete(&r->req, GOOD);
>  
>  done:
> -    if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> -    }
> +    scsi_req_unref(&r->req);
>  }
>  
>  static bool scsi_is_cmd_fua(SCSICommand *cmd)
> @@ -245,9 +238,7 @@ static void scsi_write_do_fua(SCSIDiskReq *r)
>      scsi_req_complete(&r->req, GOOD);
>  
>  done:
> -    if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> -    }
> +    scsi_req_unref(&r->req);
>  }
>  
>  static void scsi_dma_complete_noio(void *opaque, int ret)
> @@ -279,9 +270,7 @@ static void scsi_dma_complete_noio(void *opaque, int ret)
>      }
>  
>  done:
> -    if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> -    }
> +    scsi_req_unref(&r->req);
>  }
>  
>  static void scsi_dma_complete(void *opaque, int ret)
> @@ -319,9 +308,7 @@ static void scsi_read_complete(void * opaque, int ret)
>      scsi_req_data(&r->req, r->qiov.size);
>  
>  done:
> -    if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> -    }
> +    scsi_req_unref(&r->req);
>  }
>  
>  /* Actually issue a read to the block device.  */
> @@ -361,9 +348,7 @@ static void scsi_do_read(void *opaque, int ret)
>      }
>  
>  done:
> -    if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> -    }
> +    scsi_req_unref(&r->req);
>  }
>  
>  /* Read more data from scsi device into buffer.  */
> @@ -478,9 +463,7 @@ static void scsi_write_complete(void * opaque, int ret)
>      }
>  
>  done:
> -    if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> -    }
> +    scsi_req_unref(&r->req);
>  }
>  
>  static void scsi_write_data(SCSIRequest *req)
> @@ -1577,9 +1560,7 @@ static void scsi_unmap_complete(void *opaque, int ret)
>      scsi_req_complete(&r->req, GOOD);
>  
>  done:
> -    if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> -    }
> +    scsi_req_unref(&r->req);
>      g_free(data);
>  }
>  
> @@ -1672,9 +1653,7 @@ static void scsi_write_same_complete(void *opaque, int ret)
>      scsi_req_complete(&r->req, GOOD);
>  
>  done:
> -    if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> -    }
> +    scsi_req_unref(&r->req);
>      qemu_vfree(data->iov.iov_base);
>      g_free(data);
>  }
> diff --git a/hw/scsi/scsi-generic.c b/hw/scsi/scsi-generic.c
> index 0b2ff90..1c29ecb 100644
> --- a/hw/scsi/scsi-generic.c
> +++ b/hw/scsi/scsi-generic.c
> @@ -132,27 +132,20 @@ static void scsi_command_complete(void *opaque, int ret)
>      DPRINTF("Command complete 0x%p tag=0x%x status=%d\n",
>              r, r->req.tag, status);
>  
> -    scsi_req_complete(&r->req, status);
>      if (!r->req.io_canceled) {
> -        scsi_req_unref(&r->req);
> +        scsi_req_complete(&r->req, status);
>      }
> +    scsi_req_unref(&r->req);
>  }
>  
>  /* Cancel a pending data transfer.  */
>  static void scsi_cancel_io(SCSIRequest *req)
>  {
> -    SCSIGenericReq *r = DO_UPCAST(SCSIGenericReq, req, req);
> -
>      DPRINTF("Cancel tag=0x%x\n", req->tag);
> -    if (r->req.aiocb) {
> -        bdrv_aio_cancel(r->req.aiocb);
> -
> -        /* This reference was left in by scsi_*_data.  We take ownership of
> -         * it independent of whether bdrv_aio_cancel completes the request
> -         * or not.  */
> -        scsi_req_unref(&r->req);
> +    if (req->aiocb) {
> +        req->io_canceled = true;
> +        bdrv_aio_cancel_async(req->aiocb);
>      }
> -    r->req.aiocb = NULL;
>  }
>  
>  static int execute_command(BlockDriverState *bdrv,
> @@ -211,9 +204,7 @@ static void scsi_read_complete(void * opaque, int ret)
>          bdrv_set_guest_block_size(s->conf.bs, s->blocksize);
>  
>          scsi_req_data(&r->req, len);
> -        if (!r->req.io_canceled) {
> -            scsi_req_unref(&r->req);
> -        }
> +        scsi_req_unref(&r->req);
>      }
>  }
>  
> 

Re: [Qemu-devel] [RFC PATCH 1/9] block: Add bdrv_aio_cancel_async

[Stefan Hajnoczi]

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

On Thu, Aug 21, 2014 at 07:56:48PM +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 {
> +        BlockDriverAIOCB *save = g_new(BlockDriverAIOCB, 1);

Please don't create a dummy BlockDriverAIOCB.  It makes the code
confusing because all other BlockDriverAIOCBs in QEMU are allocated with
qemu_aio_get() and behave in a certain way.

This is not really a BlockDriverAIOCB, it's just a struct to stash the
old cb/opaque in.

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

Re: [Qemu-devel] [RFC PATCH 4/9] linux-aio: Implement .cancel_async

[Stefan Hajnoczi]

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

On Thu, Aug 21, 2014 at 07:56:51PM +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;
> +    }

No callback will be invoked if io_cancel(2) every cancels the request
immediately.

The current kernel implementation always returns -EINPROGRESS or some of
other error value.  But some day it might return 0 and this would leak
the request!

> +
> +    laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
> +}

It would be cleaner to reuse laio_cancel_async() from laio_cancel() to
avoid code duplication.  For example, there is a useful comment in
laio_cancel() explaining that io_cancel(2) doesn't cancel I/O in
practice on 2.6.31 era kernels.

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

Re: [Qemu-devel] [RFC PATCH 6/9] blkdebug: Implement .cancel_async

[Stefan Hajnoczi]

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

On Thu, Aug 21, 2014 at 07:56:53PM +0800, Fam Zheng wrote:
> @@ -446,12 +439,25 @@ static void error_callback_bh(void *opaque)
>      qemu_aio_release(acb);
>  }
>  
> +static void blkdebug_aio_cancel_async(BlockDriverAIOCB *blockacb)
> +{
> +    BlkdebugAIOCB *acb = container_of(blockacb, BlkdebugAIOCB, common);
> +    blockacb->cb(blockacb->opaque, -ECANCELED);
> +    qemu_aio_release(acb);
> +}
> +
>  static void blkdebug_aio_cancel(BlockDriverAIOCB *blockacb)
>  {
>      BlkdebugAIOCB *acb = container_of(blockacb, BlkdebugAIOCB, common);
>      qemu_aio_release(acb);
>  }

Both blkdebug_aio_cancel() and blkdebug_aio_cancel_async() look
incorrect.  It is not safe to release acb because the
error_callback_bh() BH may still be scheduled.

I guess we don't hit this problem because the error injection happens
within the same event loop iteration.  In practice no one ever calls
blkdebug_aio_cancel()?

Stefan

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

Re: [Qemu-devel] [RFC PATCH 8/9] block: Implement stub bdrv_em_co_aiocb_info.cancel_async

[Stefan Hajnoczi]

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

On Thu, Aug 21, 2014 at 07:56:55PM +0800, Fam Zheng wrote:
> diff --git a/block.c b/block.c
> index 1cc8926..1860e4b 100644
> --- a/block.c
> +++ b/block.c
> @@ -4750,9 +4750,14 @@ static void bdrv_aio_co_cancel_em(BlockDriverAIOCB *blockacb)
>      }
>  }
>  
> +static void bdrv_aio_co_cancel_em_async(BlockDriverAIOCB *blockacb)
> +{
> +}

Please include a comment to explain why this function body is empty:

/* Do nothing, let caller wait for the request to complete */

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

Re: [Qemu-devel] [RFC PATCH 1/9] block: Add bdrv_aio_cancel_async

[Fam Zheng]

On Thu, 08/21 14:14, Paolo Bonzini wrote:
> Il 21/08/2014 13:56, Fam Zheng ha scritto:
> > +        BlockDriverAIOCB *save = g_new(BlockDriverAIOCB, 1);
> > +        save->cb               = acb->cb;
> > +        save->opaque           = acb->opaque;
> > +        acb->cb                = bdrv_aio_cancel_async_cb;
> > +        acb->opaque            = acb;
> > +        acb->cancel_acb_save   = save;
> 
> No need for this extra field.  You can make a struct with {acb->cb,
> acb->opaque, acb} and pass it as the opaque.  You also do not need to
> allocate it on the heap, since everything is synchronous.
> 
> > +
> > +        /* Use the synchronous version and hope our cb is called. */
> > +        acb->aiocb_info->cancel(acb);
> 
> Unfortunately, acb has been freed at this point

Oops!

> , so you'll be accessing
> a dangling pointer.  I think you need to modify all cancel
> implementations.  For example:
> 
> - return whether they have called the callback
> 
> - only free the acb if they have called the callback
> 
> - otherwise, free the acb in bdrv_aio_cancel

What about we save cb and opaque locally, and set acb->cb to a nop. When cancel
is done we can call the original cb?

Fam

Re: [Qemu-devel] [RFC PATCH 8/9] block: Implement stub bdrv_em_co_aiocb_info.cancel_async

[Fam Zheng]

On Thu, 08/21 18:01, Stefan Hajnoczi wrote:
> On Thu, Aug 21, 2014 at 07:56:55PM +0800, Fam Zheng wrote:
> > diff --git a/block.c b/block.c
> > index 1cc8926..1860e4b 100644
> > --- a/block.c
> > +++ b/block.c
> > @@ -4750,9 +4750,14 @@ static void bdrv_aio_co_cancel_em(BlockDriverAIOCB *blockacb)
> >      }
> >  }
> >  
> > +static void bdrv_aio_co_cancel_em_async(BlockDriverAIOCB *blockacb)
> > +{
> > +}
> 
> Please include a comment to explain why this function body is empty:
> 
> /* Do nothing, let caller wait for the request to complete */

OK.

Fam

Re: [Qemu-devel] [RFC PATCH 6/9] blkdebug: Implement .cancel_async

[Fam Zheng]

On Thu, 08/21 17:52, Stefan Hajnoczi wrote:
> On Thu, Aug 21, 2014 at 07:56:53PM +0800, Fam Zheng wrote:
> > @@ -446,12 +439,25 @@ static void error_callback_bh(void *opaque)
> >      qemu_aio_release(acb);
> >  }
> >  
> > +static void blkdebug_aio_cancel_async(BlockDriverAIOCB *blockacb)
> > +{
> > +    BlkdebugAIOCB *acb = container_of(blockacb, BlkdebugAIOCB, common);
> > +    blockacb->cb(blockacb->opaque, -ECANCELED);
> > +    qemu_aio_release(acb);
> > +}
> > +
> >  static void blkdebug_aio_cancel(BlockDriverAIOCB *blockacb)
> >  {
> >      BlkdebugAIOCB *acb = container_of(blockacb, BlkdebugAIOCB, common);
> >      qemu_aio_release(acb);
> >  }
> 
> Both blkdebug_aio_cancel() and blkdebug_aio_cancel_async() look
> incorrect.  It is not safe to release acb because the
> error_callback_bh() BH may still be scheduled.
> 
> I guess we don't hit this problem because the error injection happens
> within the same event loop iteration.  In practice no one ever calls
> blkdebug_aio_cancel()?
> 

I'll drop this patch and send a separate fix for blkdebug_aio_cancel.

Fam

Re: [Qemu-devel] [RFC PATCH 4/9] linux-aio: Implement .cancel_async

[Fam Zheng]

On Thu, 08/21 17:31, Stefan Hajnoczi wrote:
> On Thu, Aug 21, 2014 at 07:56:51PM +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;
> > +    }
> 
> No callback will be invoked if io_cancel(2) every cancels the request
> immediately.
> 
> The current kernel implementation always returns -EINPROGRESS or some of
> other error value.  But some day it might return 0 and this would leak
> the request!
> 
> > +
> > +    laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
> > +}
> 
> It would be cleaner to reuse laio_cancel_async() from laio_cancel() to
> avoid code duplication.  For example, there is a useful comment in
> laio_cancel() explaining that io_cancel(2) doesn't cancel I/O in
> practice on 2.6.31 era kernels.

I'll take a closer look at it.

Fam

Re: [Qemu-devel] [RFC PATCH 9/9] scsi: Cancel request asynchronously

[Fam Zheng]

On Thu, 08/21 14:19, Paolo Bonzini wrote:
> Il 21/08/2014 13:56, Fam Zheng ha scritto:
> > We are blocking the whole VM, which means that an irresponsive storage
> > backend will hang the whole guest. Let's switch to bdrv_aio_cancel_async
> > to improve this.
> 
> Unforuntately, the TMF must only return after the request has been
> canceled.  I think you need to add a scsi_cancel_io_async function, and
> keep all the remaining machinery (also, you need a better commit message
> that explains what you are removing and the new invariants).
> 
> Then in virtio-scsi you need to add a list of "dependent" (controlq)
> VirtIOSCSIReq to the "main" (requestq) VirtIOSCSIReq, and complete them
> all after signaling the completion of the main request.

OK, I didn't know that. I'll try again :)

Thanks,
Fam

Re: [Qemu-devel] [RFC PATCH 1/9] block: Add bdrv_aio_cancel_async

[Paolo Bonzini]

Il 22/08/2014 03:23, Fam Zheng ha scritto:
> What about we save cb and opaque locally, and set acb->cb to a nop. When cancel
> is done we can call the original cb?

That might work but needs some auditing.  Right now the AIOCB is still
valid when the callback is called, and this would be changed.

Also, having different semantics for cancellation vs. completion would
be painful.

Paolo

Re: [Qemu-devel] [RFC PATCH 1/9] block: Add bdrv_aio_cancel_async

[Fam Zheng]

On Fri, 08/22 10:14, Paolo Bonzini wrote:
> Il 22/08/2014 03:23, Fam Zheng ha scritto:
> > What about we save cb and opaque locally, and set acb->cb to a nop. When cancel
> > is done we can call the original cb?
> 
> That might work but needs some auditing.  Right now the AIOCB is still
> valid when the callback is called, and this would be changed.
> 
> Also, having different semantics for cancellation vs. completion would
> be painful.
> 

Exactly. I'd rather not change the contract then.

Alternatively, we may add a refcnt field to BlockDriverAioCB and grab one before
calling .cancel, so the qemu_aio_release will not free it.

Fam

Re: [Qemu-devel] [RFC PATCH 1/9] block: Add bdrv_aio_cancel_async

[Paolo Bonzini]

Il 22/08/2014 11:37, Fam Zheng ha scritto:
> Exactly. I'd rather not change the contract then.
> 
> Alternatively, we may add a refcnt field to BlockDriverAioCB and grab one before
> calling .cancel, so the qemu_aio_release will not free it.

Yes, and I don't exclude that sooner or later we'll have to add
reference counts to AIOCB anyway.  However, reference counting is not
_that_ cheap so for now I'd rather see other solutions explored.

The problem is implementing cancel_sync in terms of cancel.  The
simplest solution, for now, is to make bdrv_aio_cancel_async return
false if the callback is not implemented, and fall back to synchronous
cancellation.

Paolo

Re: [Qemu-devel] [RFC PATCH 1/9] block: Add bdrv_aio_cancel_async

[Fam Zheng]

On Fri, 08/22 12:10, Paolo Bonzini wrote:
> Il 22/08/2014 11:37, Fam Zheng ha scritto:
> > Exactly. I'd rather not change the contract then.
> > 
> > Alternatively, we may add a refcnt field to BlockDriverAioCB and grab one before
> > calling .cancel, so the qemu_aio_release will not free it.
> 
> Yes, and I don't exclude that sooner or later we'll have to add
> reference counts to AIOCB anyway.  However, reference counting is not
> _that_ cheap so for now I'd rather see other solutions explored.

Why doesn it have an performance effect? Just because of the would-be
"if (likely(--acb->refcnt == 0))" testing?

> 
> The problem is implementing cancel_sync in terms of cancel.  The
> simplest solution, for now, is to make bdrv_aio_cancel_async return
> false if the callback is not implemented, and fall back to synchronous
> cancellation.

This does keep the code change local, but not necessarily simple, since there
would be two cancelling code paths in scsi-bus. I already find the scsi req
ref/unref pairs a bit hard to track.

I prefer that we change the implementation to avoid complicating interface:
don't call qemu_aio_release in .cancel, but call it in
bdrv_aio_cancel{,_async} after calling .cancel(). Does that work?

Fam