[Qemu-devel] [PATCH v5 0/3] linux-aio: fix batch submission

[Qemu-devel] [PATCH v5 0/3] linux-aio: fix batch submission

[Ming Lei]

The 1st patch fixes batch submission.

The 2nd one fixes -EAGAIN for non-batch case.

The 3rd one is a cleanup.

This patchset is splitted from previous patchset(dataplane: optimization
and multi virtqueue support), as suggested by Stefan.

V5:
    - in case of submission failure, return -EIO for new coming requests
      until aborting is handled
	- in patch2, follow Paolo's suggestion about ioq_enqueue() changes

V4:
	- abort reuqests in BH to abvoid potential "Co-routine re-entered recursively"
	- remove 'enqueue' parameter to ioq_submit() to simpify change
	- beautify code as suggested by Paolo

v3:
    - rebase on QEMU master
v2:
    - code style fix and commit log fix as suggested by Benoît Canet
v1:
    - rebase on latest QEMU master

 block/linux-aio.c |  131 +++++++++++++++++++++++++++++++++++++++++++----------
 1 file changed, 107 insertions(+), 24 deletions(-)

Thanks,
Ming Lei

[Qemu-devel] [PATCH v5 1/3] linux-aio: fix submit aio as a batch

[Ming Lei]

In the submit path, we can't complete request directly,
otherwise "Co-routine re-entered recursively" may be caused,
so this patch fixes the issue with below ideas:

	- for -EAGAIN or partial completion, retry the submision
	in following completion cb which is run in BH context
	- for part of completion, update the io queue too
	- for case of io queue full, submit queued requests
	immediatelly and return failure to caller
	- for other failure, abort all queued requests in BH
    context, and requests won't be allow to submit until
    aborting is handled

Signed-off-by: Ming Lei <ming.lei@canonical.com>
---
 block/linux-aio.c |  116 ++++++++++++++++++++++++++++++++++++++++++++---------
 1 file changed, 96 insertions(+), 20 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index d92513b..70312a4 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -38,11 +38,20 @@ struct qemu_laiocb {
     QLIST_ENTRY(qemu_laiocb) node;
 };
 
+/*
+ * TODO: support to batch I/O from multiple bs in one same
+ * AIO context, one important use case is multi-lun scsi,
+ * so in future the IO queue should be per AIO context.
+ */
 typedef struct {
     struct iocb *iocbs[MAX_QUEUED_IO];
     int plugged;
     unsigned int size;
     unsigned int idx;
+
+    /* abort queued requests in BH context */
+    QEMUBH *abort_bh;
+    bool  aborting;
 } LaioQueue;
 
 struct qemu_laio_state {
@@ -59,6 +68,8 @@ struct qemu_laio_state {
     int event_max;
 };
 
+static int ioq_submit(struct qemu_laio_state *s);
+
 static inline ssize_t io_event_ret(struct io_event *ev)
 {
     return (ssize_t)(((uint64_t)ev->res2 << 32) | ev->res);
@@ -91,6 +102,13 @@ static void qemu_laio_process_completion(struct qemu_laio_state *s,
     qemu_aio_unref(laiocb);
 }
 
+static void qemu_laio_start_retry(struct qemu_laio_state *s)
+{
+    if (s->io_q.idx) {
+        ioq_submit(s);
+    }
+}
+
 /* The completion BH fetches completed I/O requests and invokes their
  * callbacks.
  *
@@ -135,6 +153,8 @@ static void qemu_laio_completion_bh(void *opaque)
 
         qemu_laio_process_completion(s, laiocb);
     }
+
+    qemu_laio_start_retry(s);
 }
 
 static void qemu_laio_completion_cb(EventNotifier *e)
@@ -175,47 +195,99 @@ static void ioq_init(LaioQueue *io_q)
     io_q->size = MAX_QUEUED_IO;
     io_q->idx = 0;
     io_q->plugged = 0;
+    io_q->aborting = false;
 }
 
+/* Always return >= 0 and it means how many requests are submitted */
 static int ioq_submit(struct qemu_laio_state *s)
 {
-    int ret, i = 0;
+    int ret;
     int len = s->io_q.idx;
 
-    do {
-        ret = io_submit(s->ctx, len, s->io_q.iocbs);
-    } while (i++ < 3 && ret == -EAGAIN);
-
-    /* empty io queue */
-    s->io_q.idx = 0;
+    if (!len) {
+        return 0;
+    }
 
+    ret = io_submit(s->ctx, len, s->io_q.iocbs);
     if (ret < 0) {
-        i = 0;
-    } else {
-        i = ret;
+        /* retry in following completion cb */
+        if (ret == -EAGAIN) {
+            return 0;
+        }
+
+        /*
+         * Abort in BH context for avoiding Co-routine re-entered,
+         * and update io queue at that time
+         */
+        qemu_bh_schedule(s->io_q.abort_bh);
+        s->io_q.aborting = true;
+        ret = 0;
     }
 
-    for (; i < len; i++) {
-        struct qemu_laiocb *laiocb =
-            container_of(s->io_q.iocbs[i], struct qemu_laiocb, iocb);
+    /*
+     * update io queue, and retry will be started automatically
+     * in following completion cb for the remainder
+     */
+    if (ret > 0) {
+        if (ret < len) {
+            memmove(&s->io_q.iocbs[0], &s->io_q.iocbs[ret],
+                    (len - ret) * sizeof(struct iocb *));
+        }
+        s->io_q.idx -= ret;
+    }
 
-        laiocb->ret = (ret < 0) ? ret : -EIO;
+    return ret;
+}
+
+static void ioq_abort_bh(void *opaque)
+{
+    struct qemu_laio_state *s = opaque;
+    int i;
+
+    for (i = 0; i < s->io_q.idx; i++) {
+        struct qemu_laiocb *laiocb = container_of(s->io_q.iocbs[i],
+                                                  struct qemu_laiocb,
+                                                  iocb);
+        laiocb->ret = -EIO;
         qemu_laio_process_completion(s, laiocb);
     }
-    return ret;
+
+    s->io_q.idx = 0;
+    io_q->aborting = false;
 }
 
-static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
+static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
 {
     unsigned int idx = s->io_q.idx;
 
+    /* Request can't be allowed to submit until aborting is handled */
+    if (unlikely(s->io_q.aborting)) {
+        return -EIO;
+    }
+
+    if (unlikely(idx == s->io_q.size)) {
+        ioq_submit(s);
+
+        if (unlikely(s->io_q.aborting)) {
+            return -EIO;
+        }
+        idx = s->io_q.idx;
+    }
+
+    /* It has to return now if queue is still full */
+    if (unlikely(idx == s->io_q.size)) {
+        return -EAGAIN;
+    }
+
     s->io_q.iocbs[idx++] = iocb;
     s->io_q.idx = idx;
 
-    /* submit immediately if queue is full */
-    if (idx == s->io_q.size) {
-        ioq_submit(s);
+    /* submit immediately if queue depth is above 2/3 */
+    if (idx > s->io_q.size * 2 / 3) {
+        return ioq_submit(s);
     }
+
+    return 0;
 }
 
 void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
@@ -281,7 +353,9 @@ BlockAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
             goto out_free_aiocb;
         }
     } else {
-        ioq_enqueue(s, iocbs);
+        if (ioq_enqueue(s, iocbs) < 0) {
+            goto out_free_aiocb;
+        }
     }
     return &laiocb->common;
 
@@ -296,12 +370,14 @@ void laio_detach_aio_context(void *s_, AioContext *old_context)
 
     aio_set_event_notifier(old_context, &s->e, NULL);
     qemu_bh_delete(s->completion_bh);
+    qemu_bh_delete(s->io_q.abort_bh);
 }
 
 void laio_attach_aio_context(void *s_, AioContext *new_context)
 {
     struct qemu_laio_state *s = s_;
 
+    s->io_q.abort_bh = aio_bh_new(new_context, ioq_abort_bh, s);
     s->completion_bh = aio_bh_new(new_context, qemu_laio_completion_bh, s);
     aio_set_event_notifier(new_context, &s->e, qemu_laio_completion_cb);
 }
-- 
1.7.9.5

[Qemu-devel] [PATCH v5 2/3] linux-aio: handling -EAGAIN for !s->io_q.plugged case

[Ming Lei]

Previously -EAGAIN is simply ignored for !s->io_q.plugged case,
and sometimes it is easy to cause -EIO to VM, such as NVME device.

This patch handles -EAGAIN by io queue for !s->io_q.plugged case,
and it will be retried in following aio completion cb.

Suggested-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Ming Lei <ming.lei@canonical.com>
---
 block/linux-aio.c |   28 ++++++++++++++++++----------
 1 file changed, 18 insertions(+), 10 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index 70312a4..0cb098d 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -282,12 +282,17 @@ static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
     s->io_q.iocbs[idx++] = iocb;
     s->io_q.idx = idx;
 
-    /* submit immediately if queue depth is above 2/3 */
-    if (idx > s->io_q.size * 2 / 3) {
-        return ioq_submit(s);
+    /*
+     * This is reached in two cases: queue not plugged but io_submit
+     * returned -EAGAIN, or queue plugged.  In the latter case, start
+     * submitting some I/O if the queue is getting too full.  In the
+     * former case, instead, wait until an I/O operation is completed.
+     */
+    if (!s->io_q.plugged || likely(idx < s->io_q.size * 2 / 3)) {
+        return 0;
     }
 
-    return 0;
+    return ioq_submit(s);
 }
 
 void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
@@ -348,15 +353,18 @@ BlockAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
     }
     io_set_eventfd(&laiocb->iocb, event_notifier_get_fd(&s->e));
 
-    if (!s->io_q.plugged) {
-        if (io_submit(s->ctx, 1, &iocbs) < 0) {
-            goto out_free_aiocb;
-        }
-    } else {
-        if (ioq_enqueue(s, iocbs) < 0) {
+    /* Switch to queue mode until -EAGAIN is handled */
+    if (!s->io_q.plugged && !s->io_q.idx) {
+        int ret = io_submit(s->ctx, 1, &iocbs);
+        if (ret >= 0) {
+            return &laiocb->common;
+        } else if (ret != -EAGAIN) {
             goto out_free_aiocb;
         }
     }
+    if (ioq_enqueue(s, iocbs) < 0) {
+        goto out_free_aiocb;
+    }
     return &laiocb->common;
 
 out_free_aiocb:
-- 
1.7.9.5

[Qemu-devel] [PATCH v5 3/3] linux-aio: remove 'node' from 'struct qemu_laiocb'

[Ming Lei]

No one uses the 'node' field any more, so remove it
from 'struct qemu_laiocb', and this can save 16byte
for the struct on 64bit arch.

Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Ming Lei <ming.lei@canonical.com>
---
 block/linux-aio.c |    1 -
 1 file changed, 1 deletion(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index 0cb098d..f1b26f5 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -35,7 +35,6 @@ struct qemu_laiocb {
     size_t nbytes;
     QEMUIOVector *qiov;
     bool is_read;
-    QLIST_ENTRY(qemu_laiocb) node;
 };
 
 /*
-- 
1.7.9.5

Re: [Qemu-devel] [PATCH v5 0/3] linux-aio: fix batch submission

[Paolo Bonzini]

On 24/11/2014 15:29, Ming Lei wrote:
> The 1st patch fixes batch submission.
> 
> The 2nd one fixes -EAGAIN for non-batch case.
> 
> The 3rd one is a cleanup.
> 
> This patchset is splitted from previous patchset(dataplane: optimization
> and multi virtqueue support), as suggested by Stefan.
> 
> V5:
>     - in case of submission failure, return -EIO for new coming requests
>       until aborting is handled
> 	- in patch2, follow Paolo's suggestion about ioq_enqueue() changes
> 
> V4:
> 	- abort reuqests in BH to abvoid potential "Co-routine re-entered recursively"
> 	- remove 'enqueue' parameter to ioq_submit() to simpify change
> 	- beautify code as suggested by Paolo
> 
> v3:
>     - rebase on QEMU master
> v2:
>     - code style fix and commit log fix as suggested by Benoît Canet
> v1:
>     - rebase on latest QEMU master
> 
>  block/linux-aio.c |  131 +++++++++++++++++++++++++++++++++++++++++++----------
>  1 file changed, 107 insertions(+), 24 deletions(-)
> 
> Thanks,
> Ming Lei
> 
> 

Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>

Re: [Qemu-devel] [PATCH v5 0/3] linux-aio: fix batch submission

[Ming Lei]

On Mon, Nov 24, 2014 at 10:31 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
> On 24/11/2014 15:29, Ming Lei wrote:
>> The 1st patch fixes batch submission.
>>
>> The 2nd one fixes -EAGAIN for non-batch case.
>>
>> The 3rd one is a cleanup.
>>
>> This patchset is splitted from previous patchset(dataplane: optimization
>> and multi virtqueue support), as suggested by Stefan.
>>
>> V5:
>>     - in case of submission failure, return -EIO for new coming requests
>>       until aborting is handled
>>       - in patch2, follow Paolo's suggestion about ioq_enqueue() changes
>>
>> V4:
>>       - abort reuqests in BH to abvoid potential "Co-routine re-entered recursively"
>>       - remove 'enqueue' parameter to ioq_submit() to simpify change
>>       - beautify code as suggested by Paolo
>>
>> v3:
>>     - rebase on QEMU master
>> v2:
>>     - code style fix and commit log fix as suggested by Benoît Canet
>> v1:
>>     - rebase on latest QEMU master
>>
>>  block/linux-aio.c |  131 +++++++++++++++++++++++++++++++++++++++++++----------
>>  1 file changed, 107 insertions(+), 24 deletions(-)
>>
>> Thanks,
>> Ming Lei
>>
>>
>
> Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>

Great thanks!

Re: [Qemu-devel] [PATCH v5 1/3] linux-aio: fix submit aio as a batch

[Stefan Hajnoczi]

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

On Mon, Nov 24, 2014 at 10:29:37PM +0800, Ming Lei wrote:
> +    /* submit immediately if queue depth is above 2/3 */
> +    if (idx > s->io_q.size * 2 / 3) {
> +        return ioq_submit(s);

The return value of ioq_submit() is the number of requests submitted (0
or more).  This has no meaning to the caller and therefore shouldn't be
the return value of this function.

Please drop the return here.

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

Re: [Qemu-devel] [PATCH v5 2/3] linux-aio: handling -EAGAIN for !s->io_q.plugged case

[Stefan Hajnoczi]

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

On Mon, Nov 24, 2014 at 10:29:38PM +0800, Ming Lei wrote:
> Previously -EAGAIN is simply ignored for !s->io_q.plugged case,
> and sometimes it is easy to cause -EIO to VM, such as NVME device.
> 
> This patch handles -EAGAIN by io queue for !s->io_q.plugged case,
> and it will be retried in following aio completion cb.
> 
> Suggested-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Ming Lei <ming.lei@canonical.com>
> ---
>  block/linux-aio.c |   28 ++++++++++++++++++----------
>  1 file changed, 18 insertions(+), 10 deletions(-)
> 
> diff --git a/block/linux-aio.c b/block/linux-aio.c
> index 70312a4..0cb098d 100644
> --- a/block/linux-aio.c
> +++ b/block/linux-aio.c
> @@ -282,12 +282,17 @@ static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
>      s->io_q.iocbs[idx++] = iocb;
>      s->io_q.idx = idx;
>  
> -    /* submit immediately if queue depth is above 2/3 */
> -    if (idx > s->io_q.size * 2 / 3) {
> -        return ioq_submit(s);
> +    /*
> +     * This is reached in two cases: queue not plugged but io_submit
> +     * returned -EAGAIN, or queue plugged.  In the latter case, start
> +     * submitting some I/O if the queue is getting too full.  In the
> +     * former case, instead, wait until an I/O operation is completed.
> +     */
> +    if (!s->io_q.plugged || likely(idx < s->io_q.size * 2 / 3)) {
> +        return 0;
>      }
>  
> -    return 0;
> +    return ioq_submit(s);

Same as in the previous patch, the ioq_submit() return value is not
meaningful to ioq_enqueue() callers.  It should not be returned.

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

Re: [Qemu-devel] [PATCH v5 1/3] linux-aio: fix submit aio as a batch

[Stefan Hajnoczi]

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

On Mon, Nov 24, 2014 at 10:29:37PM +0800, Ming Lei wrote:
> +static void ioq_abort_bh(void *opaque)
> +{
> +    struct qemu_laio_state *s = opaque;
> +    int i;
> +
> +    for (i = 0; i < s->io_q.idx; i++) {
> +        struct qemu_laiocb *laiocb = container_of(s->io_q.iocbs[i],
> +                                                  struct qemu_laiocb,
> +                                                  iocb);
> +        laiocb->ret = -EIO;
>          qemu_laio_process_completion(s, laiocb);
>      }
> -    return ret;
> +
> +    s->io_q.idx = 0;
> +    io_q->aborting = false;
>  }

This patch does not compile:

  CC    block/linux-aio.o
  block/linux-aio.c: In function ‘ioq_abort_bh’:
  block/linux-aio.c:255:5: error: ‘io_q’ undeclared (first use in this function)
       io_q->aborting = false;
       ^

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