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

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

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

Thanks,

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

[Ming Lei]

In the enqueue path, we can't complete request, 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 by
	schedule an BH in following completion cb
	- for part of completion, also update the io queue
	- for other failure, return the failure if in enqueue path,
	otherwise, abort all queued I/O

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index d92513b..f66e8ad 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;
+
+    /* handle -EAGAIN and partial completion */
+    QEMUBH *retry;
 } LaioQueue;
 
 struct qemu_laio_state {
@@ -137,6 +145,12 @@ static void qemu_laio_completion_bh(void *opaque)
     }
 }
 
+static void qemu_laio_start_retry(struct qemu_laio_state *s)
+{
+    if (s->io_q.idx)
+        qemu_bh_schedule(s->io_q.retry);
+}
+
 static void qemu_laio_completion_cb(EventNotifier *e)
 {
     struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e);
@@ -144,6 +158,7 @@ static void qemu_laio_completion_cb(EventNotifier *e)
     if (event_notifier_test_and_clear(&s->e)) {
         qemu_bh_schedule(s->completion_bh);
     }
+    qemu_laio_start_retry(s);
 }
 
 static void laio_cancel(BlockAIOCB *blockacb)
@@ -163,6 +178,9 @@ static void laio_cancel(BlockAIOCB *blockacb)
     }
 
     laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
+
+    /* check if there are requests in io queue */
+    qemu_laio_start_retry(laiocb->ctx);
 }
 
 static const AIOCBInfo laio_aiocb_info = {
@@ -177,45 +195,80 @@ static void ioq_init(LaioQueue *io_q)
     io_q->plugged = 0;
 }
 
-static int ioq_submit(struct qemu_laio_state *s)
+static void abort_queue(struct qemu_laio_state *s)
+{
+    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);
+    }
+}
+
+static int ioq_submit(struct qemu_laio_state *s, bool enqueue)
 {
     int ret, i = 0;
     int len = s->io_q.idx;
+    int j = 0;
 
-    do {
-        ret = io_submit(s->ctx, len, s->io_q.iocbs);
-    } while (i++ < 3 && ret == -EAGAIN);
+    if (!len) {
+        return 0;
+    }
 
-    /* empty io queue */
-    s->io_q.idx = 0;
+    ret = io_submit(s->ctx, len, s->io_q.iocbs);
+    if (ret == -EAGAIN) { /* retry in following completion cb */
+        return 0;
+    } else if (ret < 0) {
+        if (enqueue) {
+            return ret;
+        }
 
-    if (ret < 0) {
-        i = 0;
-    } else {
-        i = ret;
+        /* in non-queue path, all IOs have to be completed */
+        abort_queue(s);
+        ret = len;
+    } else if (ret == 0) {
+        goto out;
     }
 
-    for (; i < len; i++) {
-        struct qemu_laiocb *laiocb =
-            container_of(s->io_q.iocbs[i], struct qemu_laiocb, iocb);
-
-        laiocb->ret = (ret < 0) ? ret : -EIO;
-        qemu_laio_process_completion(s, laiocb);
+    for (i = ret; i < len; i++) {
+        s->io_q.iocbs[j++] = s->io_q.iocbs[i];
     }
+
+ out:
+    /*
+     * update io queue, for partial completion, retry will be
+     * started automatically in following completion cb.
+     */
+    s->io_q.idx -= ret;
+
     return ret;
 }
 
-static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
+static void ioq_submit_retry(void *opaque)
+{
+    struct qemu_laio_state *s = opaque;
+    ioq_submit(s, false);
+}
+
+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)) {
+        return -1;
+    }
+
     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, true);
     }
+
+    return 0;
 }
 
 void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
@@ -237,7 +290,7 @@ int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
     }
 
     if (s->io_q.idx > 0) {
-        ret = ioq_submit(s);
+        ret = ioq_submit(s, false);
     }
 
     return ret;
@@ -281,7 +334,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 +351,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.retry);
 }
 
 void laio_attach_aio_context(void *s_, AioContext *new_context)
 {
     struct qemu_laio_state *s = s_;
 
+    s->io_q.retry = aio_bh_new(new_context, ioq_submit_retry, 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 v3 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 |   22 +++++++++++++++++++++-
 1 file changed, 21 insertions(+), 1 deletion(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index f66e8ad..f5ca41d 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -263,6 +263,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, true);
@@ -330,10 +335,25 @@ 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) {
+        int ret;
+
+        if (!s->io_q.idx) {
+            ret = io_submit(s->ctx, 1, &iocbs);
+        } else {
+            ret = -EAGAIN;
+        }
+        /*
+         * Switch to queue mode until -EAGAIN is handled, we suppose
+         * there is always uncompleted I/O, so try to enqueue it first,
+         * and will be submitted again in following aio completion cb.
+         */
+        if (ret == -EAGAIN) {
+            goto enqueue;
+        } else if (ret < 0) {
             goto out_free_aiocb;
         }
     } else {
+ enqueue:
         if (ioq_enqueue(s, iocbs) < 0) {
             goto out_free_aiocb;
         }
-- 
1.7.9.5

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

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 f5ca41d..b12da25 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 v3 2/3] linux-aio: handling -EAGAIN for !s->io_q.plugged case

[Paolo Bonzini]

On 06/11/2014 16:10, Ming Lei wrote:
> +    /* don't submit until next completion for -EAGAIN of non plug case */
> +    if (unlikely(!s->io_q.plugged)) {
> +        return 0;
> +    }
> +

Is this an optimization or a fix for something?

> +        /*
> +         * Switch to queue mode until -EAGAIN is handled, we suppose
> +         * there is always uncompleted I/O, so try to enqueue it first,
> +         * and will be submitted again in following aio completion cb.
> +         */
> +        if (ret == -EAGAIN) {
> +            goto enqueue;
> +        } else if (ret < 0) {
>              goto out_free_aiocb;
>          }

Better:

     if (!s->io_q.plugged && !s->io_q.idx) {
        ret = io_submit(s->ctx, 1, &iocbs);
        if (ret >= 0) {
            return &laiocb->common;
        }
        if (ret != -EAGAIN) {
            goto out_free_aiocb;
        }
    }

    /* code for queue mode.  */

Paolo

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

[Paolo Bonzini]

On 06/11/2014 16:10, Ming Lei wrote:
> 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.
> 
> 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 f5ca41d..b12da25 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;
>  };
>  
>  /*
> 

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

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

[Paolo Bonzini]

> @@ -137,6 +145,12 @@ static void qemu_laio_completion_bh(void *opaque)
>      }
>  }
>  
> +static void qemu_laio_start_retry(struct qemu_laio_state *s)
> +{
> +    if (s->io_q.idx)
> +        qemu_bh_schedule(s->io_q.retry);
> +}
> +
>  static void qemu_laio_completion_cb(EventNotifier *e)
>  {
>      struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e);
> @@ -144,6 +158,7 @@ static void qemu_laio_completion_cb(EventNotifier *e)
>      if (event_notifier_test_and_clear(&s->e)) {
>          qemu_bh_schedule(s->completion_bh);
>      }
> +    qemu_laio_start_retry(s);

I think you do not even need two bottom halves.  Just call ioq_submit
from completion_bh instead, after the call to io_getevents.

>  }
>  
>  static void laio_cancel(BlockAIOCB *blockacb)
> @@ -163,6 +178,9 @@ static void laio_cancel(BlockAIOCB *blockacb)
>      }
>  
>      laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
> +
> +    /* check if there are requests in io queue */
> +    qemu_laio_start_retry(laiocb->ctx);
>  }
>  
>  static const AIOCBInfo laio_aiocb_info = {
> @@ -177,45 +195,80 @@ static void ioq_init(LaioQueue *io_q)
>      io_q->plugged = 0;
>  }
>  
> -static int ioq_submit(struct qemu_laio_state *s)
> +static void abort_queue(struct qemu_laio_state *s)
> +{
> +    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);
> +    }
> +}
> +
> +static int ioq_submit(struct qemu_laio_state *s, bool enqueue)
>  {
>      int ret, i = 0;
>      int len = s->io_q.idx;
> +    int j = 0;
>  
> -    do {
> -        ret = io_submit(s->ctx, len, s->io_q.iocbs);
> -    } while (i++ < 3 && ret == -EAGAIN);
> +    if (!len) {
> +        return 0;
> +    }
>  
> -    /* empty io queue */
> -    s->io_q.idx = 0;
> +    ret = io_submit(s->ctx, len, s->io_q.iocbs);
> +    if (ret == -EAGAIN) { /* retry in following completion cb */
> +        return 0;
> +    } else if (ret < 0) {
> +        if (enqueue) {
> +            return ret;
> +        }
>  
> -    if (ret < 0) {
> -        i = 0;
> -    } else {
> -        i = ret;
> +        /* in non-queue path, all IOs have to be completed */
> +        abort_queue(s);
> +        ret = len;
> +    } else if (ret == 0) {
> +        goto out;

No need for goto; just move the "for" loop inside this conditional.  Or
better, just use memmove.  That is:

    if (ret < 0) {
        if (ret == -EAGAIN) {
            return 0;
        }
        if (enqueue) {
            return ret;
        }

        abort_queue(s);
        ret = len;
    }

    if (ret > 0) {
        memmove(...)
        s->io_q.idx -= ret;
    }
    return ret;

> +     * update io queue, for partial completion, retry will be
> +     * started automatically in following completion cb.
> +     */
> +    s->io_q.idx -= ret;
> +
>      return ret;
>  }
>  
> -static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
> +static void ioq_submit_retry(void *opaque)
> +{
> +    struct qemu_laio_state *s = opaque;
> +    ioq_submit(s, false);
> +}
> +
> +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)) {
> +        return -1;

return -EAGAIN?

Thanks,

Paolo

> +    }
> +
>      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, true);
>      }
> +
> +    return 0;
>  }
>  
>  void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
> @@ -237,7 +290,7 @@ int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
>      }
>  
>      if (s->io_q.idx > 0) {
> -        ret = ioq_submit(s);
> +        ret = ioq_submit(s, false);
>      }
>  
>      return ret;
> @@ -281,7 +334,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 +351,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.retry);
>  }
>  
>  void laio_attach_aio_context(void *s_, AioContext *new_context)
>  {
>      struct qemu_laio_state *s = s_;
>  
> +    s->io_q.retry = aio_bh_new(new_context, ioq_submit_retry, 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);
>  }
> 

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

[Ming Lei]

On Tue, Nov 18, 2014 at 10:18 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
>> @@ -137,6 +145,12 @@ static void qemu_laio_completion_bh(void *opaque)
>>      }
>>  }
>>
>> +static void qemu_laio_start_retry(struct qemu_laio_state *s)
>> +{
>> +    if (s->io_q.idx)
>> +        qemu_bh_schedule(s->io_q.retry);
>> +}
>> +
>>  static void qemu_laio_completion_cb(EventNotifier *e)
>>  {
>>      struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e);
>> @@ -144,6 +158,7 @@ static void qemu_laio_completion_cb(EventNotifier *e)
>>      if (event_notifier_test_and_clear(&s->e)) {
>>          qemu_bh_schedule(s->completion_bh);
>>      }
>> +    qemu_laio_start_retry(s);
>
> I think you do not even need two bottom halves.  Just call ioq_submit
> from completion_bh instead, after the call to io_getevents.

Yes, that can save one BH, actually the patch was written when
there wasn't completion BH, :-)

>
>>  }
>>
>>  static void laio_cancel(BlockAIOCB *blockacb)
>> @@ -163,6 +178,9 @@ static void laio_cancel(BlockAIOCB *blockacb)
>>      }
>>
>>      laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
>> +
>> +    /* check if there are requests in io queue */
>> +    qemu_laio_start_retry(laiocb->ctx);
>>  }
>>
>>  static const AIOCBInfo laio_aiocb_info = {
>> @@ -177,45 +195,80 @@ static void ioq_init(LaioQueue *io_q)
>>      io_q->plugged = 0;
>>  }
>>
>> -static int ioq_submit(struct qemu_laio_state *s)
>> +static void abort_queue(struct qemu_laio_state *s)
>> +{
>> +    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);
>> +    }
>> +}
>> +
>> +static int ioq_submit(struct qemu_laio_state *s, bool enqueue)
>>  {
>>      int ret, i = 0;
>>      int len = s->io_q.idx;
>> +    int j = 0;
>>
>> -    do {
>> -        ret = io_submit(s->ctx, len, s->io_q.iocbs);
>> -    } while (i++ < 3 && ret == -EAGAIN);
>> +    if (!len) {
>> +        return 0;
>> +    }
>>
>> -    /* empty io queue */
>> -    s->io_q.idx = 0;
>> +    ret = io_submit(s->ctx, len, s->io_q.iocbs);
>> +    if (ret == -EAGAIN) { /* retry in following completion cb */
>> +        return 0;
>> +    } else if (ret < 0) {
>> +        if (enqueue) {
>> +            return ret;
>> +        }
>>
>> -    if (ret < 0) {
>> -        i = 0;
>> -    } else {
>> -        i = ret;
>> +        /* in non-queue path, all IOs have to be completed */
>> +        abort_queue(s);
>> +        ret = len;
>> +    } else if (ret == 0) {
>> +        goto out;
>
> No need for goto; just move the "for" loop inside this conditional.  Or
> better, just use memmove.  That is:
>
>     if (ret < 0) {
>         if (ret == -EAGAIN) {
>             return 0;
>         }
>         if (enqueue) {
>             return ret;
>         }
>
>         abort_queue(s);
>         ret = len;
>     }
>
>     if (ret > 0) {
>         memmove(...)
>         s->io_q.idx -= ret;
>     }
>     return ret;

The above is better.

>> +     * update io queue, for partial completion, retry will be
>> +     * started automatically in following completion cb.
>> +     */
>> +    s->io_q.idx -= ret;
>> +
>>      return ret;
>>  }
>>
>> -static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
>> +static void ioq_submit_retry(void *opaque)
>> +{
>> +    struct qemu_laio_state *s = opaque;
>> +    ioq_submit(s, false);
>> +}
>> +
>> +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)) {
>> +        return -1;
>
> return -EAGAIN?

It means the io queue is full, so the code has to fail the current
request.

Thanks,
Ming Lei

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

[Ming Lei]

On Tue, Nov 18, 2014 at 10:06 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
> On 06/11/2014 16:10, Ming Lei wrote:
>> +    /* don't submit until next completion for -EAGAIN of non plug case */
>> +    if (unlikely(!s->io_q.plugged)) {
>> +        return 0;
>> +    }
>> +
>
> Is this an optimization or a fix for something?

It is for avoiding unnecessary submission which will cause
another -EAGAIN.

>
>> +        /*
>> +         * Switch to queue mode until -EAGAIN is handled, we suppose
>> +         * there is always uncompleted I/O, so try to enqueue it first,
>> +         * and will be submitted again in following aio completion cb.
>> +         */
>> +        if (ret == -EAGAIN) {
>> +            goto enqueue;
>> +        } else if (ret < 0) {
>>              goto out_free_aiocb;
>>          }
>
> Better:
>
>      if (!s->io_q.plugged && !s->io_q.idx) {
>         ret = io_submit(s->ctx, 1, &iocbs);
>         if (ret >= 0) {
>             return &laiocb->common;
>         }
>         if (ret != -EAGAIN) {
>             goto out_free_aiocb;
>         }
>     }

Right.

Thanks,
Ming Lei

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

[Paolo Bonzini]

On 22/11/2014 13:16, 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)) {
> > > +        return -1;
> >
> > return -EAGAIN?
>
> It means the io queue is full, so the code has to fail the current
> request.

Right, but "-1" is "-EPERM" which doesn't make much sense.  I think the
right thing to do is to return EAGAIN, and let the owner figure it out.
 For example, a SCSI device might return a "BUSY" status code.

Paolo

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

[Paolo Bonzini]

On 18/11/2014 15:06, Paolo Bonzini wrote:
>> > +    /* don't submit until next completion for -EAGAIN of non plug case */
>> > +    if (unlikely(!s->io_q.plugged)) {
>> > +        return 0;
>> > +    }
>> > +
> Is this an optimization or a fix for something?

Ah, if !io_q.plugged we must be in the -EAGAIN case, so no need to 
"submit immediately if queue depth is above 2/3".  Can you rewrite the 
comment like this:

    /* 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;
    }

Paolo

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

[Ming Lei]

On Mon, Nov 24, 2014 at 5:47 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
> On 22/11/2014 13:16, 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)) {
>> > > +        return -1;
>> >
>> > return -EAGAIN?
>>
>> It means the io queue is full, so the code has to fail the current
>> request.
>
> Right, but "-1" is "-EPERM" which doesn't make much sense.  I think the
> right thing to do is to return EAGAIN, and let the owner figure it out.
>  For example, a SCSI device might return a "BUSY" status code.

We can do that, but laio_submit() doesn't return the code to callers.


Thanks,
Ming Lei

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

[Paolo Bonzini]

On 24/11/2014 11:02, Ming Lei wrote:
>> > Right, but "-1" is "-EPERM" which doesn't make much sense.  I think the
>> > right thing to do is to return EAGAIN, and let the owner figure it out.
>> >  For example, a SCSI device might return a "BUSY" status code.
> We can do that, but laio_submit() doesn't return the code to callers.

That can be fixed later.  But not returning a sensible errno only makes
things more complicated later on.

Paolo

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

[Ming Lei]

On Mon, Nov 24, 2014 at 5:52 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
> On 18/11/2014 15:06, Paolo Bonzini wrote:
>>> > +    /* don't submit until next completion for -EAGAIN of non plug case */
>>> > +    if (unlikely(!s->io_q.plugged)) {
>>> > +        return 0;
>>> > +    }
>>> > +
>> Is this an optimization or a fix for something?
>
> Ah, if !io_q.plugged we must be in the -EAGAIN case, so no need to
> "submit immediately if queue depth is above 2/3".  Can you rewrite the

The above comment is only for io_q.plugged case, and for
!io_q.plugged with -EAGAIN, the requests is submitted in
retry path.

> comment like this:
>
>     /* 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;
>     }

Yes, the above change need the corresponding comment, but this
patch needn't since the 1st case won't reach here, :-)

Thanks,
Ming Lei

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

[Paolo Bonzini]

On 24/11/2014 11:11, Ming Lei wrote:
> Yes, the above change need the corresponding comment, but this
> patch needn't since the 1st case won't reach here, :-)

I'm not sure I follow, but you can send v4 with a clearer comment I
guess. :)

Paolo