[Qemu-devel] [RFC PATCH v2 4/6] linux-aio: Support partial io_submits

[Kevin Wolf <kwolf@redhat.com>]

io_submit() can submit less requests than we passed it. In this case, we
used to fail the remaining requests, which isn't very nice. Instead,
let's just keep the remaining requests around in the request queue and
resubmit them after the next request completion when resources should be
available for new requests.

Signed-off-by: Kevin Wolf <kwolf@redhat.com>
---
 block/linux-aio.c | 50 +++++++++++++++++++++++++++++++-------------------
 1 file changed, 31 insertions(+), 19 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index fd8f0e4..1406a71 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -60,6 +60,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);
@@ -136,6 +138,12 @@ static void qemu_laio_completion_bh(void *opaque)
 
         qemu_laio_process_completion(s, laiocb);
     }
+
+    /* If we couldn't submit all queued requests before, now that we have
+     * completed some, we might be able to submit more of them. */
+    if (s->io_q.idx) {
+        ioq_submit(s);
+    }
 }
 
 static void qemu_laio_completion_cb(EventNotifier *e)
@@ -163,29 +171,33 @@ static int ioq_submit(struct qemu_laio_state *s)
         ret = io_submit(s->ctx, len, s->io_q.iocbs);
     } while (i++ < 3 && ret == -EAGAIN);
 
-    /* empty io queue */
-    s->io_q.idx = 0;
-
+    /* On io_submit() failure, fail all requests in the queue */
     if (ret < 0) {
-        i = 0;
-    } else {
-        i = ret;
-    }
+        s->io_q.idx = 0;
 
-    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;
-        if (laiocb->co != qemu_coroutine_self()) {
-            qemu_coroutine_enter(laiocb->co, NULL);
-        } else {
-            /* The return value is used for the currently active coroutine.
-             * We're always in ioq_enqueue() here, ioq_submit() never runs from
-             * a request's coroutine.*/
-            ret = laiocb->ret;
+        for (i = 0; i < len; i++) {
+            struct qemu_laiocb *laiocb =
+                container_of(s->io_q.iocbs[i], struct qemu_laiocb, iocb);
+
+            laiocb->ret = ret;
+            if (laiocb->co != qemu_coroutine_self()) {
+                qemu_coroutine_enter(laiocb->co, NULL);
+            } else {
+                /* The return value is used for the currently active coroutine.
+                 * We're always in ioq_enqueue() here, ioq_submit() never runs
+                 * from a queued request's coroutine.*/
+            }
         }
+        return ret;
     }
+
+    /* Remove submitted requests from the queue */
+    s->io_q.idx = len - ret;
+    if (s->io_q.idx) {
+        memmove(s->io_q.iocbs, s->io_q.iocbs + ret,
+                s->io_q.idx * sizeof(s->io_q.iocbs[0]));
+    }
+
     return ret;
 }
 
-- 
1.8.3.1