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

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

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 |  108 +++++++++++++++++++++++++++++++++++++++++------------
 1 file changed, 84 insertions(+), 24 deletions(-)

Thanks
Ming Lei

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

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index d92513b..11fcedb 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -38,11 +38,19 @@ 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;
 } LaioQueue;
 
 struct qemu_laio_state {
@@ -59,6 +67,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 +101,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 +152,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)
@@ -177,45 +196,75 @@ static void ioq_init(LaioQueue *io_q)
     io_q->plugged = 0;
 }
 
+/* always return >= 0 */
 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 */
+        qemu_bh_schedule(s->io_q.abort_bh);
+        ret = len;
     }
 
-    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;
+    }
+
+    return ret;
+}
+
+static void ioq_abort_bh(void *opaque)
+{
+    struct qemu_laio_state *s = opaque;
+    int i;
 
-        laiocb->ret = (ret < 0) ? ret : -EIO;
+    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;
 }
 
-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;
 
+    if (unlikely(idx == s->io_q.size)) {
+        ioq_submit(s);
+        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 +330,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 +347,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 v4 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 |   20 ++++++++++++++------
 1 file changed, 14 insertions(+), 6 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index 11fcedb..974e4f9 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -259,6 +259,11 @@ static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
     s->io_q.iocbs[idx++] = iocb;
     s->io_q.idx = idx;
 
+    /* don't submit until next completion for -EAGAIN of non plug case */
+    if (unlikely(!s->io_q.plugged)) {
+        return 0;
+    }
+
     /* submit immediately if queue depth is above 2/3 */
     if (idx > s->io_q.size * 2 / 3) {
         return ioq_submit(s);
@@ -325,15 +330,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 v4 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 974e4f9..f36e96c 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 v4 2/3] linux-aio: handling -EAGAIN for !s->io_q.plugged case

[Paolo Bonzini]

On 24/11/2014 12:31, Ming Lei wrote:
> +    /* don't submit until next completion for -EAGAIN of non plug case */
> +    if (unlikely(!s->io_q.plugged)) {
> +        return 0;
> +    }
> +
>      /* submit immediately if queue depth is above 2/3 */
>      if (idx > s->io_q.size * 2 / 3) {
>          return ioq_submit(s);
      }

      return 0;

so I fail to see why my proposal:


    /* 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 (likely(idx <= s->io_q.size * 2 / 3) || unlikely(!s->io_q.plugged) {
        return 0;
    }

    return ioq_submit(s);

was wrong.  Can you explain?

Paolo

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

[Paolo Bonzini]

On 24/11/2014 12:31, Ming Lei wrote:
> +static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
>  {
>      unsigned int idx = s->io_q.idx;
>  
> +    if (unlikely(idx == s->io_q.size)) {
> +        ioq_submit(s);
> +        return -EAGAIN;

Only return -EAGAIN if ioq_submit(s) returns 0?  Otherwise reload idx
and go on.

Paolo

> +    }
> +

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

[Paolo Bonzini]

On 24/11/2014 12:31, 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.

Just a couple comments.  I think v5 will be okay.

Paolo

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

[Ming Lei]

On Mon, Nov 24, 2014 at 7:38 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
> On 24/11/2014 12:31, Ming Lei wrote:
>> +    /* don't submit until next completion for -EAGAIN of non plug case */
>> +    if (unlikely(!s->io_q.plugged)) {
>> +        return 0;
>> +    }
>> +
>>      /* submit immediately if queue depth is above 2/3 */
>>      if (idx > s->io_q.size * 2 / 3) {
>>          return ioq_submit(s);
>       }
>
>       return 0;
>
> so I fail to see why my proposal:
>
>
>     /* 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 (likely(idx <= s->io_q.size * 2 / 3) || unlikely(!s->io_q.plugged) {
>         return 0;
>     }
>
>     return ioq_submit(s);
>
> was wrong.  Can you explain?

I didn't say your proposal is wrong, and this patch is correct too
without fat comment.

The difference is only that this patch returns immediately in case
of !s->io_q.plugged after putting the req into io queue.

Thanks,

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

[Ming Lei]

On Mon, Nov 24, 2014 at 7:39 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
> On 24/11/2014 12:31, Ming Lei wrote:
>> +static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
>>  {
>>      unsigned int idx = s->io_q.idx;
>>
>> +    if (unlikely(idx == s->io_q.size)) {
>> +        ioq_submit(s);
>> +        return -EAGAIN;
>
> Only return -EAGAIN if ioq_submit(s) returns 0?  Otherwise reload idx
> and go on.

Good point.

Thanks,
Ming Lei

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

[Paolo Bonzini]

On 24/11/2014 12:49, Ming Lei wrote:
> On Mon, Nov 24, 2014 at 7:38 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>>
>>
>> On 24/11/2014 12:31, Ming Lei wrote:
>>> +    /* don't submit until next completion for -EAGAIN of non plug case */
>>> +    if (unlikely(!s->io_q.plugged)) {
>>> +        return 0;
>>> +    }
>>> +
>>>      /* submit immediately if queue depth is above 2/3 */
>>>      if (idx > s->io_q.size * 2 / 3) {
>>>          return ioq_submit(s);
>>       }
>>
>>       return 0;
>>
>> so I fail to see why my proposal:
>>
>>
>>     /* 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 (likely(idx <= s->io_q.size * 2 / 3) || unlikely(!s->io_q.plugged) {
>>         return 0;
>>     }
>>
>>     return ioq_submit(s);
>>
>> was wrong.  Can you explain?
> 
> I didn't say your proposal is wrong, and this patch is correct too
> without fat comment.
> 
> The difference is only that this patch returns immediately in case
> of !s->io_q.plugged after putting the req into io queue.

There is no difference in the behavior of the code, right?

The maintainers can decide if they want a v5 of this patch.

Thanks,

Paolo