[Qemu-devel] [PATCH 0/4] linux-aio: rewrite and simplify queuing code

[Qemu-devel] [PATCH 0/4] linux-aio: rewrite and simplify queuing code

[Paolo Bonzini]

This series rewrites the io_submit code to avoid having to
synchronize the iocbs in two places (qemu_laiocb and LaioQueue).
Instead, the queue of pending I/O requests is kept in a list.

This fixes the problems with io_submit doing partial submissions
or failing with EAGAIN, without complicating the code.  The bug is
shown by "qemu-img bench -n -t none -s 65536 -d 1024 -c 100000"
pretty much instantly.

Paolo

Paolo Bonzini (4):
  linux-aio: queue requests that cannot be submitted
  linux-aio: track whether the queue is blocked
  linux-aio: rename LaioQueue idx field to "n"
  linux-aio: simplify removal of completed iocbs from the list

 block/linux-aio.c    | 86 +++++++++++++++++++++++++---------------------------
 include/qemu/queue.h | 11 +++++++
 2 files changed, 53 insertions(+), 44 deletions(-)

-- 
2.1.0

[Qemu-devel] [PATCH 1/4] linux-aio: queue requests that cannot be submitted

[Paolo Bonzini]

Keep a queue of requests that were not submitted; pass them to
the kernel when a completion is reported, unless the queue is
plugged.

The array of iocbs is rebuilt every time from scratch.  This
avoids keeping the iocbs array and list synchronized.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 block/linux-aio.c | 75 ++++++++++++++++++++++++-------------------------------
 1 file changed, 33 insertions(+), 42 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index d92513b..b6fbfd8 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -35,14 +35,13 @@ struct qemu_laiocb {
     size_t nbytes;
     QEMUIOVector *qiov;
     bool is_read;
-    QLIST_ENTRY(qemu_laiocb) node;
+    QSIMPLEQ_ENTRY(qemu_laiocb) next;
 };
 
 typedef struct {
-    struct iocb *iocbs[MAX_QUEUED_IO];
     int plugged;
-    unsigned int size;
     unsigned int idx;
+    QSIMPLEQ_HEAD(, qemu_laiocb) pending;
 } LaioQueue;
 
 struct qemu_laio_state {
@@ -59,6 +58,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);
@@ -135,6 +136,10 @@ static void qemu_laio_completion_bh(void *opaque)
 
         qemu_laio_process_completion(s, laiocb);
     }
+
+    if (!s->io_q.plugged && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
+        ioq_submit(s);
+    }
 }
 
 static void qemu_laio_completion_cb(EventNotifier *e)
@@ -172,52 +177,40 @@ static const AIOCBInfo laio_aiocb_info = {
 
 static void ioq_init(LaioQueue *io_q)
 {
-    io_q->size = MAX_QUEUED_IO;
-    io_q->idx = 0;
+    QSIMPLEQ_INIT(&io_q->pending);
     io_q->plugged = 0;
+    io_q->idx = 0;
 }
 
 static int ioq_submit(struct qemu_laio_state *s)
 {
-    int ret, i = 0;
-    int len = s->io_q.idx;
-
-    do {
-        ret = io_submit(s->ctx, len, s->io_q.iocbs);
-    } while (i++ < 3 && ret == -EAGAIN);
+    int ret, i;
+    int len = 0;
+    struct qemu_laiocb *aiocb;
+    struct iocb *iocbs[MAX_QUEUED_IO];
 
-    /* empty io queue */
-    s->io_q.idx = 0;
+    QSIMPLEQ_FOREACH(aiocb, &s->io_q.pending, next) {
+        iocbs[len++] = &aiocb->iocb;
+        if (len == MAX_QUEUED_IO) {
+            break;
+        }
+    }
 
+    ret = io_submit(s->ctx, len, iocbs);
+    if (ret == -EAGAIN) {
+        ret = 0;
+    }
     if (ret < 0) {
-        i = 0;
-    } else {
-        i = ret;
+        abort();
     }
 
-    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 = 0; i < ret; i++) {
+        s->io_q.idx--;
+        QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
     }
     return ret;
 }
 
-static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
-{
-    unsigned int idx = s->io_q.idx;
-
-    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);
-    }
-}
-
 void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
 {
     struct qemu_laio_state *s = aio_ctx;
@@ -236,7 +229,7 @@ int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
         return 0;
     }
 
-    if (s->io_q.idx > 0) {
+    if (!QSIMPLEQ_EMPTY(&s->io_q.pending)) {
         ret = ioq_submit(s);
     }
 
@@ -276,12 +269,10 @@ 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 {
-        ioq_enqueue(s, iocbs);
+    QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
+    s->io_q.idx++;
+    if (s->io_q.idx == (s->io_q.plugged ? MAX_QUEUED_IO : 1)) {
+        ioq_submit(s);
     }
     return &laiocb->common;
 
-- 
2.1.0

Re: [Qemu-devel] [PATCH 1/4] linux-aio: queue requests that cannot be submitted

[Kevin Wolf]

Am 10.12.2014 um 15:51 hat Paolo Bonzini geschrieben:
> Keep a queue of requests that were not submitted; pass them to
> the kernel when a completion is reported, unless the queue is
> plugged.
> 
> The array of iocbs is rebuilt every time from scratch.  This
> avoids keeping the iocbs array and list synchronized.
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  block/linux-aio.c | 75 ++++++++++++++++++++++++-------------------------------
>  1 file changed, 33 insertions(+), 42 deletions(-)
> 
> diff --git a/block/linux-aio.c b/block/linux-aio.c
> index d92513b..b6fbfd8 100644
> --- a/block/linux-aio.c
> +++ b/block/linux-aio.c
> @@ -35,14 +35,13 @@ struct qemu_laiocb {
>      size_t nbytes;
>      QEMUIOVector *qiov;
>      bool is_read;
> -    QLIST_ENTRY(qemu_laiocb) node;
> +    QSIMPLEQ_ENTRY(qemu_laiocb) next;
>  };
>  
>  typedef struct {
> -    struct iocb *iocbs[MAX_QUEUED_IO];
>      int plugged;
> -    unsigned int size;
>      unsigned int idx;
> +    QSIMPLEQ_HEAD(, qemu_laiocb) pending;
>  } LaioQueue;
>  
>  struct qemu_laio_state {
> @@ -59,6 +58,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);
> @@ -135,6 +136,10 @@ static void qemu_laio_completion_bh(void *opaque)
>  
>          qemu_laio_process_completion(s, laiocb);
>      }
> +
> +    if (!s->io_q.plugged && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
> +        ioq_submit(s);
> +    }
>  }
>  
>  static void qemu_laio_completion_cb(EventNotifier *e)
> @@ -172,52 +177,40 @@ static const AIOCBInfo laio_aiocb_info = {
>  
>  static void ioq_init(LaioQueue *io_q)
>  {
> -    io_q->size = MAX_QUEUED_IO;
> -    io_q->idx = 0;
> +    QSIMPLEQ_INIT(&io_q->pending);
>      io_q->plugged = 0;
> +    io_q->idx = 0;
>  }
>  
>  static int ioq_submit(struct qemu_laio_state *s)
>  {
> -    int ret, i = 0;
> -    int len = s->io_q.idx;
> -
> -    do {
> -        ret = io_submit(s->ctx, len, s->io_q.iocbs);
> -    } while (i++ < 3 && ret == -EAGAIN);
> +    int ret, i;
> +    int len = 0;
> +    struct qemu_laiocb *aiocb;
> +    struct iocb *iocbs[MAX_QUEUED_IO];
>  
> -    /* empty io queue */
> -    s->io_q.idx = 0;
> +    QSIMPLEQ_FOREACH(aiocb, &s->io_q.pending, next) {
> +        iocbs[len++] = &aiocb->iocb;
> +        if (len == MAX_QUEUED_IO) {
> +            break;
> +        }
> +    }
>  
> +    ret = io_submit(s->ctx, len, iocbs);
> +    if (ret == -EAGAIN) {
> +        ret = 0;
> +    }
>      if (ret < 0) {
> -        i = 0;
> -    } else {
> -        i = ret;
> +        abort();
>      }

abort() doesn't feel right here.

>  
> -    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 = 0; i < ret; i++) {
> +        s->io_q.idx--;
> +        QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
>      }
>      return ret;
>  }
>  
> -static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
> -{
> -    unsigned int idx = s->io_q.idx;
> -
> -    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);
> -    }
> -}
> -
>  void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
>  {
>      struct qemu_laio_state *s = aio_ctx;
> @@ -236,7 +229,7 @@ int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
>          return 0;
>      }
>  
> -    if (s->io_q.idx > 0) {
> +    if (!QSIMPLEQ_EMPTY(&s->io_q.pending)) {
>          ret = ioq_submit(s);
>      }
>  
> @@ -276,12 +269,10 @@ 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 {
> -        ioq_enqueue(s, iocbs);
> +    QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
> +    s->io_q.idx++;
> +    if (s->io_q.idx == (s->io_q.plugged ? MAX_QUEUED_IO : 1)) {

More naturally written and more obviously correct as (!s->io_q,plugged ||
s->io_q.idx >= MAX_QUEUED_IO). Which happens to be what the next patch
converts it to, so I won't spend much time thinking about whether this
version is actually right.

> +        ioq_submit(s);
>      }
>      return &laiocb->common;

Kevin

Re: [Qemu-devel] [PATCH 1/4] linux-aio: queue requests that cannot be submitted

[Paolo Bonzini]

On 11/12/2014 13:49, Kevin Wolf wrote:
> > -    } else {
> > -        i = ret;
> > +        abort();
> >      }
> 
> abort() doesn't feel right here.

man doesn't suggest any error that can actually happen.

> > +    QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
> > +    s->io_q.idx++;
> > +    if (s->io_q.idx == (s->io_q.plugged ? MAX_QUEUED_IO : 1)) {
> 
> More naturally written and more obviously correct as (!s->io_q,plugged ||
> s->io_q.idx >= MAX_QUEUED_IO). Which happens to be what the next patch
> converts it to, so I won't spend much time thinking about whether this
> version is actually right.

Sort of.  If the queue is blocked due to -EAGAIN, I don't want to
io_submit every time an operation is queued, hence the ==.  The next
patch adds !s->io_q.blocked, so it can use the more natural and indeed
more obvious expression.

Paolo

Re: [Qemu-devel] [PATCH 1/4] linux-aio: queue requests that cannot be submitted

[Kevin Wolf]

Am 11.12.2014 um 13:52 hat Paolo Bonzini geschrieben:
> 
> 
> On 11/12/2014 13:49, Kevin Wolf wrote:
> > > -    } else {
> > > -        i = ret;
> > > +        abort();
> > >      }
> > 
> > abort() doesn't feel right here.
> 
> man doesn't suggest any error that can actually happen.

Yes. I guess I just like to be on the safe side. I would be fine with
dropping requests on the floor and thereby breaking the block device in
this unlikely case if proper error handling is too hard, but killing the
qemu process generally makes me feel uncomfortable.

> > > +    QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
> > > +    s->io_q.idx++;
> > > +    if (s->io_q.idx == (s->io_q.plugged ? MAX_QUEUED_IO : 1)) {
> > 
> > More naturally written and more obviously correct as (!s->io_q,plugged ||
> > s->io_q.idx >= MAX_QUEUED_IO). Which happens to be what the next patch
> > converts it to, so I won't spend much time thinking about whether this
> > version is actually right.
> 
> Sort of.  If the queue is blocked due to -EAGAIN, I don't want to
> io_submit every time an operation is queued, hence the ==.  The next
> patch adds !s->io_q.blocked, so it can use the more natural and indeed
> more obvious expression.

Oh, I see. I didn't get that this was an optimisation.

Kevin

Re: [Qemu-devel] [PATCH 1/4] linux-aio: queue requests that cannot be submitted

[Paolo Bonzini]

On 11/12/2014 14:02, Kevin Wolf wrote:
>> > > > -    } else {
>> > > > -        i = ret;
>> > > > +        abort();
>> > > >      }
> > > 
> > > abort() doesn't feel right here.
> > 
> > man doesn't suggest any error that can actually happen.
> 
> Yes. I guess I just like to be on the safe side. I would be fine with
> dropping requests on the floor and thereby breaking the block device in
> this unlikely case if proper error handling is too hard, but killing the
> qemu process generally makes me feel uncomfortable.

Hmm, since it shouldn't happen I prefer for it to fail right away and
make it easier to spot what happened in the core dump.

Paolo

[Qemu-devel] [PATCH 2/4] linux-aio: track whether the queue is blocked

[Paolo Bonzini]

Do not try submitting requests when io_submit reported that it
couldn't accept more; dually, try more io_submit calls if it
could handle the whole set of requests that were passed, so that
the "blocked" flag is reset as soon as possible.

This also fixes a hack in the previous patch, where s->io_q.idx
was compared with "==" in order to avoid repeated submissions.
It makes it more sense to compare with ">=", so do that now that
we have the "blocked" flag.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 block/linux-aio.c | 47 +++++++++++++++++++++++++++--------------------
 1 file changed, 27 insertions(+), 20 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index b6fbfd8..b870942 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -41,6 +41,7 @@ struct qemu_laiocb {
 typedef struct {
     int plugged;
     unsigned int idx;
+    bool blocked;
     QSIMPLEQ_HEAD(, qemu_laiocb) pending;
 } LaioQueue;
 
@@ -180,34 +181,39 @@ static void ioq_init(LaioQueue *io_q)
     QSIMPLEQ_INIT(&io_q->pending);
     io_q->plugged = 0;
     io_q->idx = 0;
+    io_q->blocked = false;
 }
 
 static int ioq_submit(struct qemu_laio_state *s)
 {
-    int ret, i;
-    int len = 0;
+    int ret, i, len;
     struct qemu_laiocb *aiocb;
     struct iocb *iocbs[MAX_QUEUED_IO];
 
-    QSIMPLEQ_FOREACH(aiocb, &s->io_q.pending, next) {
-        iocbs[len++] = &aiocb->iocb;
-        if (len == MAX_QUEUED_IO) {
-            break;
+    do {
+        len = 0;
+        QSIMPLEQ_FOREACH(aiocb, &s->io_q.pending, next) {
+            iocbs[len++] = &aiocb->iocb;
+            if (len == MAX_QUEUED_IO) {
+                break;
+            }
         }
-    }
 
-    ret = io_submit(s->ctx, len, iocbs);
-    if (ret == -EAGAIN) {
-        ret = 0;
-    }
-    if (ret < 0) {
-        abort();
-    }
+        ret = io_submit(s->ctx, len, iocbs);
+        if (ret == -EAGAIN) {
+            ret = 0;
+        }
+        if (ret < 0) {
+            abort();
+        }
+
+        for (i = 0; i < ret; i++) {
+            s->io_q.idx--;
+            QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
+        }
+    } while (ret == len && !QSIMPLEQ_EMPTY(&s->io_q.pending));
+    s->io_q.blocked = (s->io_q.idx > 0);
 
-    for (i = 0; i < ret; i++) {
-        s->io_q.idx--;
-        QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
-    }
     return ret;
 }
 
@@ -229,7 +235,7 @@ int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
         return 0;
     }
 
-    if (!QSIMPLEQ_EMPTY(&s->io_q.pending)) {
+    if (!s->io_q.blocked && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
         ret = ioq_submit(s);
     }
 
@@ -271,7 +277,8 @@ BlockAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
 
     QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
     s->io_q.idx++;
-    if (s->io_q.idx == (s->io_q.plugged ? MAX_QUEUED_IO : 1)) {
+    if (!s->io_q.blocked &&
+        (!s->io_q.plugged || s->io_q.idx >= MAX_QUEUED_IO)) {
         ioq_submit(s);
     }
     return &laiocb->common;
-- 
2.1.0

[Qemu-devel] [PATCH 3/4] linux-aio: rename LaioQueue idx field to "n"

[Paolo Bonzini]

It does not identify an index in an array anymore.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 block/linux-aio.c | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index b870942..b223d9e 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -40,7 +40,7 @@ struct qemu_laiocb {
 
 typedef struct {
     int plugged;
-    unsigned int idx;
+    unsigned int n;
     bool blocked;
     QSIMPLEQ_HEAD(, qemu_laiocb) pending;
 } LaioQueue;
@@ -180,7 +180,7 @@ static void ioq_init(LaioQueue *io_q)
 {
     QSIMPLEQ_INIT(&io_q->pending);
     io_q->plugged = 0;
-    io_q->idx = 0;
+    io_q->n = 0;
     io_q->blocked = false;
 }
 
@@ -208,11 +208,11 @@ static int ioq_submit(struct qemu_laio_state *s)
         }
 
         for (i = 0; i < ret; i++) {
-            s->io_q.idx--;
+            s->io_q.n--;
             QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
         }
     } while (ret == len && !QSIMPLEQ_EMPTY(&s->io_q.pending));
-    s->io_q.blocked = (s->io_q.idx > 0);
+    s->io_q.blocked = (s->io_q.n > 0);
 
     return ret;
 }
@@ -276,9 +276,9 @@ BlockAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
     io_set_eventfd(&laiocb->iocb, event_notifier_get_fd(&s->e));
 
     QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
-    s->io_q.idx++;
+    s->io_q.n++;
     if (!s->io_q.blocked &&
-        (!s->io_q.plugged || s->io_q.idx >= MAX_QUEUED_IO)) {
+        (!s->io_q.plugged || s->io_q.n >= MAX_QUEUED_IO)) {
         ioq_submit(s);
     }
     return &laiocb->common;
-- 
2.1.0

[Qemu-devel] [PATCH 4/4] linux-aio: simplify removal of completed iocbs from the list

[Paolo Bonzini]

There is no need to do another O(n) pass on the list; the iocb to
splice the list at is already available in the array we passed to
io_submit.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 block/linux-aio.c    | 12 ++++++------
 include/qemu/queue.h | 11 +++++++++++
 2 files changed, 17 insertions(+), 6 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index b223d9e..6c98f72 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -186,9 +186,10 @@ static void ioq_init(LaioQueue *io_q)
 
 static int ioq_submit(struct qemu_laio_state *s)
 {
-    int ret, i, len;
+    int ret, len;
     struct qemu_laiocb *aiocb;
     struct iocb *iocbs[MAX_QUEUED_IO];
+    QSIMPLEQ_HEAD(, qemu_laiocb) completed;
 
     do {
         len = 0;
@@ -201,16 +202,15 @@ static int ioq_submit(struct qemu_laio_state *s)
 
         ret = io_submit(s->ctx, len, iocbs);
         if (ret == -EAGAIN) {
-            ret = 0;
+            break;
         }
         if (ret < 0) {
             abort();
         }
 
-        for (i = 0; i < ret; i++) {
-            s->io_q.n--;
-            QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
-        }
+        s->io_q.n -= ret;
+        aiocb = container_of(iocbs[ret - 1], struct qemu_laiocb, iocb);
+        QSIMPLEQ_SPLIT_AFTER(&completed, &s->io_q.pending, aiocb, next);
     } while (ret == len && !QSIMPLEQ_EMPTY(&s->io_q.pending));
     s->io_q.blocked = (s->io_q.n > 0);
 
diff --git a/include/qemu/queue.h b/include/qemu/queue.h
index 0dedd29..2c21d28 100644
--- a/include/qemu/queue.h
+++ b/include/qemu/queue.h
@@ -279,6 +279,17 @@ struct {                                                                \
         (head)->sqh_last = &(head)->sqh_first;                          \
 } while (/*CONSTCOND*/0)
 
+#define QSIMPLEQ_SPLIT_AFTER(head1, head2, elm, field) do {             \
+    if (((head1)->sqh_first = (head2)->sqh_first) == NULL) {            \
+        (head1)->sqh_last = &(head1)->sqh_first;                        \
+    } else {                                                            \
+        (head1)->sqh_last = &(elm)->field.sqe_next;                     \
+        if (((head2)->sqh_first = (elm)->field.sqe_next) == NULL) {     \
+            (head2)->sqh_last = &(head2)->sqh_first;                    \
+        }                                                               \
+    }                                                                   \
+} while (/*CONSTCOND*/0)
+
 #define QSIMPLEQ_REMOVE(head, elm, type, field) do {                    \
     if ((head)->sqh_first == (elm)) {                                   \
         QSIMPLEQ_REMOVE_HEAD((head), field);                            \
-- 
2.1.0

Re: [Qemu-devel] [PATCH 4/4] linux-aio: simplify removal of completed iocbs from the list

[Kevin Wolf]

Am 10.12.2014 um 15:52 hat Paolo Bonzini geschrieben:
> There is no need to do another O(n) pass on the list; the iocb to
> splice the list at is already available in the array we passed to
> io_submit.
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  block/linux-aio.c    | 12 ++++++------
>  include/qemu/queue.h | 11 +++++++++++
>  2 files changed, 17 insertions(+), 6 deletions(-)
> 
> diff --git a/block/linux-aio.c b/block/linux-aio.c
> index b223d9e..6c98f72 100644
> --- a/block/linux-aio.c
> +++ b/block/linux-aio.c
> @@ -186,9 +186,10 @@ static void ioq_init(LaioQueue *io_q)
>  
>  static int ioq_submit(struct qemu_laio_state *s)
>  {
> -    int ret, i, len;
> +    int ret, len;
>      struct qemu_laiocb *aiocb;
>      struct iocb *iocbs[MAX_QUEUED_IO];
> +    QSIMPLEQ_HEAD(, qemu_laiocb) completed;
>  
>      do {
>          len = 0;
> @@ -201,16 +202,15 @@ static int ioq_submit(struct qemu_laio_state *s)
>  
>          ret = io_submit(s->ctx, len, iocbs);
>          if (ret == -EAGAIN) {
> -            ret = 0;
> +            break;
>          }

ioq_submit() returns -EAGAIN now instead of 0. Almost all callers ignore
the return value, except laio_io_unplug(), which inherits this change.
Fortunately, raw_aio_unplug() completely ignores the return value.

So seems that this didn't cause a bug, but we have some cleanup to do
and make functions void if their return value isn't used anywhere.

>          if (ret < 0) {
>              abort();
>          }
>  
> -        for (i = 0; i < ret; i++) {
> -            s->io_q.n--;
> -            QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
> -        }
> +        s->io_q.n -= ret;
> +        aiocb = container_of(iocbs[ret - 1], struct qemu_laiocb, iocb);
> +        QSIMPLEQ_SPLIT_AFTER(&completed, &s->io_q.pending, aiocb, next);
>      } while (ret == len && !QSIMPLEQ_EMPTY(&s->io_q.pending));
>      s->io_q.blocked = (s->io_q.n > 0);
>  
> diff --git a/include/qemu/queue.h b/include/qemu/queue.h
> index 0dedd29..2c21d28 100644
> --- a/include/qemu/queue.h
> +++ b/include/qemu/queue.h
> @@ -279,6 +279,17 @@ struct {                                                                \
>          (head)->sqh_last = &(head)->sqh_first;                          \
>  } while (/*CONSTCOND*/0)
>  
> +#define QSIMPLEQ_SPLIT_AFTER(head1, head2, elm, field) do {             \
> +    if (((head1)->sqh_first = (head2)->sqh_first) == NULL) {            \
> +        (head1)->sqh_last = &(head1)->sqh_first;                        \
> +    } else {                                                            \
> +        (head1)->sqh_last = &(elm)->field.sqe_next;                     \
> +        if (((head2)->sqh_first = (elm)->field.sqe_next) == NULL) {     \
> +            (head2)->sqh_last = &(head2)->sqh_first;                    \
> +        }                                                               \
> +    }                                                                   \
> +} while (/*CONSTCOND*/0)

Wouldn't it be easier to write a macro that doesn't split a queue in
two, but simply removes everything up to a given element?

Anyway, if you really want to implement a split operation, I think you
also need to break the actual chain, i.e. (head1)->sqh_last.sqe_next =
NULL.

Kevin

Re: [Qemu-devel] [PATCH 4/4] linux-aio: simplify removal of completed iocbs from the list

[Paolo Bonzini]

On 11/12/2014 14:13, Kevin Wolf wrote:
>> > +#define QSIMPLEQ_SPLIT_AFTER(head1, head2, elm, field) do {             \
>> > +    if (((head1)->sqh_first = (head2)->sqh_first) == NULL) {            \
>> > +        (head1)->sqh_last = &(head1)->sqh_first;                        \
>> > +    } else {                                                            \
>> > +        (head1)->sqh_last = &(elm)->field.sqe_next;                     \
>> > +        if (((head2)->sqh_first = (elm)->field.sqe_next) == NULL) {     \
>> > +            (head2)->sqh_last = &(head2)->sqh_first;                    \
>> > +        }                                                               \
>> > +    }                                                                   \
>> > +} while (/*CONSTCOND*/0)
> Wouldn't it be easier to write a macro that doesn't split a queue in
> two, but simply removes everything up to a given element?

Yeah, though I figured that in the common case you'd have to free those
elements or otherwise process them.  But I left this patch last because
I wasn't sure about the API.  Feel free to ignore it, also given the
above comment about EAGAIN.

Paolo

> Anyway, if you really want to implement a split operation, I think you
> also need to break the actual chain, i.e. (head1)->sqh_last.sqe_next =
> NULL.

Re: [Qemu-devel] [PATCH 4/4] linux-aio: simplify removal of completed iocbs from the list

[Kevin Wolf]

Am 11.12.2014 um 14:15 hat Paolo Bonzini geschrieben:
> 
> 
> On 11/12/2014 14:13, Kevin Wolf wrote:
> >> > +#define QSIMPLEQ_SPLIT_AFTER(head1, head2, elm, field) do {             \
> >> > +    if (((head1)->sqh_first = (head2)->sqh_first) == NULL) {            \
> >> > +        (head1)->sqh_last = &(head1)->sqh_first;                        \
> >> > +    } else {                                                            \
> >> > +        (head1)->sqh_last = &(elm)->field.sqe_next;                     \
> >> > +        if (((head2)->sqh_first = (elm)->field.sqe_next) == NULL) {     \
> >> > +            (head2)->sqh_last = &(head2)->sqh_first;                    \
> >> > +        }                                                               \
> >> > +    }                                                                   \
> >> > +} while (/*CONSTCOND*/0)
> > Wouldn't it be easier to write a macro that doesn't split a queue in
> > two, but simply removes everything up to a given element?
> 
> Yeah, though I figured that in the common case you'd have to free those
> elements or otherwise process them.  But I left this patch last because
> I wasn't sure about the API.  Feel free to ignore it, also given the
> above comment about EAGAIN.

I actually like the idea of this patch.

> > Anyway, if you really want to implement a split operation, I think you
> > also need to break the actual chain, i.e. (head1)->sqh_last.sqe_next =
> > NULL.

Can you please fix this one and send a v2?

The EAGAIN thing doesn't need to be fixed because it's ignored anyway. A
cleanup is unrelated and can be done later. As for the abort() in patch
2, I'll leave the decision to you.

Overall it looks like a nice series.

Kevin

Re: [Qemu-devel] [PATCH 4/4] linux-aio: simplify removal of completed iocbs from the list

[Paolo Bonzini]

On 11/12/2014 14:22, Kevin Wolf wrote:
> Can you please fix this one and send a v2?

Done.

> The EAGAIN thing doesn't need to be fixed because it's ignored anyway. A
> cleanup is unrelated and can be done later. As for the abort() in patch
> 2, I'll leave the decision to you.

Executive summary: did the cleanup now, and left the abort() in.

Paolo