[Qemu-devel] [RFC v9 09/27] virtio-blk: Add Linux AIO queue

[Stefan Hajnoczi]

Requests read from the vring will be placed in a queue where they can be
merged as necessary.  Once all requests have been read from the vring,
the queue can be submitted.

Signed-off-by: Stefan Hajnoczi <address@hidden>
---
 hw/dataplane/ioq.h |  104 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 hw/virtio-blk.c    |   33 ++++++++---------
 2 files changed, 120 insertions(+), 17 deletions(-)
 create mode 100644 hw/dataplane/ioq.h

diff --git a/hw/dataplane/ioq.h b/hw/dataplane/ioq.h
new file mode 100644
index 0000000..26ca307
--- /dev/null
+++ b/hw/dataplane/ioq.h
@@ -0,0 +1,104 @@
+#ifndef IO_QUEUE_H
+#define IO_QUEUE_H
+
+typedef struct {
+    int fd;                         /* file descriptor */
+    unsigned int maxreqs;           /* max length of freelist and queue */
+
+    io_context_t io_ctx;            /* Linux AIO context */
+    EventNotifier notifier;         /* Linux AIO eventfd */
+
+    /* Requests can complete in any order so a free list is necessary to manage
+     * available iocbs.
+     */
+    struct iocb **freelist;         /* free iocbs */
+    unsigned int freelist_idx;
+
+    /* Multiple requests are queued up before submitting them all in one go */
+    struct iocb **queue;            /* queued iocbs */
+    unsigned int queue_idx;
+} IOQueue;
+
+static void ioq_init(IOQueue *ioq, int fd, unsigned int maxreqs)
+{
+    ioq->fd = fd;
+    ioq->maxreqs = maxreqs;
+
+    if (io_setup(maxreqs, &ioq->io_ctx) != 0) {
+        fprintf(stderr, "ioq io_setup failed\n");
+        exit(1);
+    }
+
+    if (event_notifier_init(&ioq->notifier, 0) != 0) {
+        fprintf(stderr, "ioq io event notifier creation failed\n");
+        exit(1);
+    }
+
+    ioq->freelist = g_malloc0(sizeof ioq->freelist[0] * maxreqs);
+    ioq->freelist_idx = 0;
+
+    ioq->queue = g_malloc0(sizeof ioq->queue[0] * maxreqs);
+    ioq->queue_idx = 0;
+}
+
+static void ioq_cleanup(IOQueue *ioq)
+{
+    g_free(ioq->freelist);
+    g_free(ioq->queue);
+
+    event_notifier_cleanup(&ioq->notifier);
+    io_destroy(ioq->io_ctx);
+}
+
+static EventNotifier *ioq_get_notifier(IOQueue *ioq)
+{
+    return &ioq->notifier;
+}
+
+static struct iocb *ioq_get_iocb(IOQueue *ioq)
+{
+    if (unlikely(ioq->freelist_idx == 0)) {
+        fprintf(stderr, "ioq underflow\n");
+        exit(1);
+    }
+    struct iocb *iocb = ioq->freelist[--ioq->freelist_idx];
+    ioq->queue[ioq->queue_idx++] = iocb;
+}
+
+static __attribute__((unused)) void ioq_put_iocb(IOQueue *ioq, struct iocb 
*iocb)
+{
+    if (unlikely(ioq->freelist_idx == ioq->maxreqs)) {
+        fprintf(stderr, "ioq overflow\n");
+        exit(1);
+    }
+    ioq->freelist[ioq->freelist_idx++] = iocb;
+}
+
+static __attribute__((unused)) void ioq_rdwr(IOQueue *ioq, bool read, struct 
iovec *iov, unsigned int count, long long offset)
+{
+    struct iocb *iocb = ioq_get_iocb(ioq);
+
+    if (read) {
+        io_prep_preadv(iocb, ioq->fd, iov, count, offset);
+    } else {
+        io_prep_pwritev(iocb, ioq->fd, iov, count, offset);
+    }
+    io_set_eventfd(iocb, event_notifier_get_fd(&ioq->notifier));
+}
+
+static __attribute__((unused)) void ioq_fdsync(IOQueue *ioq)
+{
+    struct iocb *iocb = ioq_get_iocb(ioq);
+
+    io_prep_fdsync(iocb, ioq->fd);
+    io_set_eventfd(iocb, event_notifier_get_fd(&ioq->notifier));
+}
+
+static __attribute__((unused)) int ioq_submit(IOQueue *ioq)
+{
+    int rc = io_submit(ioq->io_ctx, ioq->queue_idx, ioq->queue);
+    ioq->queue_idx = 0; /* reset */
+    return rc;
+}
+
+#endif /* IO_QUEUE_H */
diff --git a/hw/virtio-blk.c b/hw/virtio-blk.c
index 91f1bab..5e1ed79 100644
--- a/hw/virtio-blk.c
+++ b/hw/virtio-blk.c
@@ -13,12 +13,14 @@
 
 #include <libaio.h>
 #include "qemu-common.h"
+#include "block_int.h"
 #include "qemu-thread.h"
 #include "qemu-error.h"
 #include "blockdev.h"
 #include "virtio-blk.h"
 #include "hw/dataplane/event-poll.h"
 #include "hw/dataplane/vring.h"
+#include "hw/dataplane/ioq.h"
 #include "kvm.h"
 
 enum {
@@ -42,9 +44,9 @@ typedef struct VirtIOBlock
 
     Vring vring;                    /* virtqueue vring */
 
+    IOQueue ioqueue;                /* Linux AIO queue (should really be per 
dataplane thread) */
+
     EventPoll event_poll;           /* event poller */
-    io_context_t io_ctx;            /* Linux AIO context */
-    EventNotifier io_notifier;      /* Linux AIO eventfd */
     EventHandler io_handler;        /* Linux AIO completion handler */
     EventHandler notify_handler;    /* virtqueue notify handler */
 } VirtIOBlock;
@@ -128,6 +130,14 @@ static void *data_plane_thread(void *opaque)
     return NULL;
 }
 
+/* Normally the block driver passes down the fd, there's no way to get it from
+ * above.
+ */
+static int get_raw_posix_fd_hack(VirtIOBlock *s)
+{
+    return *(int*)s->bs->file->opaque;
+}
+
 static void data_plane_start(VirtIOBlock *s)
 {
     vring_setup(&s->vring, &s->vdev, 0);
@@ -138,23 +148,13 @@ static void data_plane_start(VirtIOBlock *s)
         fprintf(stderr, "virtio-blk failed to set host notifier, ensure 
-enable-kvm is set\n");
         exit(1);
     }
-
     event_poll_add(&s->event_poll, &s->notify_handler,
                    virtio_queue_get_host_notifier(s->vq),
                    handle_notify);
 
-    /* Create aio context */
-    if (io_setup(SEG_MAX, &s->io_ctx) != 0) {
-        fprintf(stderr, "virtio-blk io_setup failed\n");
-        exit(1);
-    }
-
-    if (event_notifier_init(&s->io_notifier, 0) != 0) {
-        fprintf(stderr, "virtio-blk io event notifier creation failed\n");
-        exit(1);
-    }
-
-    event_poll_add(&s->event_poll, &s->io_handler, &s->io_notifier, handle_io);
+    ioq_init(&s->ioqueue, get_raw_posix_fd_hack(s), REQ_MAX);
+    /* TODO populate ioqueue freelist */
+    event_poll_add(&s->event_poll, &s->io_handler, 
ioq_get_notifier(&s->ioqueue), handle_io);
 
     qemu_thread_create(&s->data_plane_thread, data_plane_thread, s, 
QEMU_THREAD_JOINABLE);
 
@@ -167,8 +167,7 @@ static void data_plane_stop(VirtIOBlock *s)
 
     /* TODO stop data plane thread */
 
-    event_notifier_cleanup(&s->io_notifier);
-    io_destroy(s->io_ctx);
+    ioq_cleanup(&s->ioqueue);
 
     s->vdev.binding->set_host_notifier(s->vdev.binding_opaque, 0, false);
 
-- 
1.7.10.4