[Qemu-devel] [PATCH 00/13] linux-aio/virtio-scsi: support AioContext wide IO submission as batch

[Qemu-devel] [PATCH 00/13] linux-aio/virtio-scsi: support AioContext wide IO submission as batch

[Ming Lei]

This patch implements AioContext wide IO submission as batch, and
the idea behind is very simple:

	- linux native aio(io_submit) supports to enqueue read/write requests
	to different files

	- in one AioContext, I/O requests from VM can be submitted to different
	backend in host, one typical example is multi-lun scsi 

This patch changes 'struct qemu_laio_state' as per AioContext, and
multiple 'bs' can be associted with one single instance of
'struct qemu_laio_state', then AioContext wide IO submission as batch
becomes easy to implement.

One simple test in my laptop shows ~20% throughput improvement
on randread from VM(using AioContext wide IO batch vs. not using io batch)
with below config:

	-drive id=drive_scsi1-0-0-0,if=none,format=raw,cache=none,aio=native,file=/dev/nullb2 \
	-drive id=drive_scsi1-0-0-1,if=none,format=raw,cache=none,aio=native,file=/dev/nullb3 \
	-device virtio-scsi-pci,num_queues=4,id=scsi1,addr=07,iothread=iothread0 \
	-device scsi-disk,bus=scsi1.0,channel=0,scsi-id=1,lun=0,drive=drive_scsi1-0-0-0,id=scsi1-0-0-0 \
	-device scsi-disk,bus=scsi1.0,channel=0,scsi-id=1,lun=1,drive=drive_scsi1-0-0-1,id=scsi1-0-0-1 \

BTW, maybe more boost can be obtained since ~33K/sec write() system call
can be observed when this test case is running, and it might be a recent
regression(BH?).

This patchset can be found on below tree too:

	git://kernel.ubuntu.com/ming/qemu.git aio-io-batch.2

and these patches depend on "linux-aio: fix batch submission" patches
in below link:

	http://marc.info/?l=qemu-devel&m=141528663106557&w=2

Any comments and suggestions are welcome.

 async.c                         |    1 +
 block.c                         |   16 +++
 block/linux-aio.c               |  251 ++++++++++++++++++++++++++++++---------
 block/raw-aio.h                 |    6 +-
 block/raw-posix.c               |    4 +-
 hw/scsi/virtio-scsi-dataplane.c |    8 ++
 hw/scsi/virtio-scsi.c           |    2 -
 include/block/aio.h             |   27 +++++
 include/block/block.h           |    3 +
 9 files changed, 259 insertions(+), 59 deletions(-)

Thanks,
Ming Lei

[Qemu-devel] [PATCH 01/13] block/linux-aio: allocate io queue dynamically

[Ming Lei]

This patch allocates io queue dynamically so that we
can support aio_context wide io queue in the following
patch.

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index b12da25..b9db28e 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -57,7 +57,7 @@ struct qemu_laio_state {
     EventNotifier e;
 
     /* io queue for submit at batch */
-    LaioQueue io_q;
+    LaioQueue *io_q;
 
     /* I/O completion processing */
     QEMUBH *completion_bh;
@@ -146,8 +146,8 @@ 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);
+    if (s->io_q->idx)
+        qemu_bh_schedule(s->io_q->retry);
 }
 
 static void qemu_laio_completion_cb(EventNotifier *e)
@@ -197,8 +197,8 @@ static void ioq_init(LaioQueue *io_q)
 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],
+    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;
@@ -209,14 +209,14 @@ static void abort_queue(struct qemu_laio_state *s)
 static int ioq_submit(struct qemu_laio_state *s, bool enqueue)
 {
     int ret, i = 0;
-    int len = s->io_q.idx;
+    int len = s->io_q->idx;
     int j = 0;
 
     if (!len) {
         return 0;
     }
 
-    ret = io_submit(s->ctx, len, s->io_q.iocbs);
+    ret = io_submit(s->ctx, len, s->io_q->iocbs);
     if (ret == -EAGAIN) { /* retry in following completion cb */
         return 0;
     } else if (ret < 0) {
@@ -232,7 +232,7 @@ static int ioq_submit(struct qemu_laio_state *s, bool enqueue)
     }
 
     for (i = ret; i < len; i++) {
-        s->io_q.iocbs[j++] = s->io_q.iocbs[i];
+        s->io_q->iocbs[j++] = s->io_q->iocbs[i];
     }
 
  out:
@@ -240,7 +240,7 @@ static int ioq_submit(struct qemu_laio_state *s, bool enqueue)
      * update io queue, for partial completion, retry will be
      * started automatically in following completion cb.
      */
-    s->io_q.idx -= ret;
+    s->io_q->idx -= ret;
 
     return ret;
 }
@@ -253,22 +253,22 @@ static void ioq_submit_retry(void *opaque)
 
 static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
 {
-    unsigned int idx = s->io_q.idx;
+    unsigned int idx = s->io_q->idx;
 
-    if (unlikely(idx == s->io_q.size)) {
+    if (unlikely(idx == s->io_q->size)) {
         return -1;
     }
 
-    s->io_q.iocbs[idx++] = iocb;
-    s->io_q.idx = idx;
+    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)) {
+    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) {
+    if (idx > s->io_q->size * 2 / 3) {
         return ioq_submit(s, true);
     }
 
@@ -279,7 +279,7 @@ void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
 {
     struct qemu_laio_state *s = aio_ctx;
 
-    s->io_q.plugged++;
+    s->io_q->plugged++;
 }
 
 int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
@@ -287,13 +287,13 @@ int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
     struct qemu_laio_state *s = aio_ctx;
     int ret = 0;
 
-    assert(s->io_q.plugged > 0 || !unplug);
+    assert(s->io_q->plugged > 0 || !unplug);
 
-    if (unplug && --s->io_q.plugged > 0) {
+    if (unplug && --s->io_q->plugged > 0) {
         return 0;
     }
 
-    if (s->io_q.idx > 0) {
+    if (s->io_q->idx > 0) {
         ret = ioq_submit(s, false);
     }
 
@@ -333,10 +333,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 (!s->io_q->plugged) {
         int ret;
 
-        if (!s->io_q.idx) {
+        if (!s->io_q->idx) {
             ret = io_submit(s->ctx, 1, &iocbs);
         } else {
             ret = -EAGAIN;
@@ -364,20 +364,38 @@ out_free_aiocb:
     return NULL;
 }
 
+static LaioQueue *laio_alloc_ioq(AioContext *ctx, struct qemu_laio_state *s)
+{
+    LaioQueue *ioq = g_malloc0(sizeof(*ioq));
+
+    ioq_init(ioq);
+    ioq->retry = aio_bh_new(ctx, ioq_submit_retry, s);
+    return ioq;
+}
+
+static void laio_free_ioq(struct qemu_laio_state *s, LaioQueue *ioq)
+{
+    qemu_bh_delete(ioq->retry);
+    g_free(ioq);
+    s->io_q = NULL;
+}
+
 void laio_detach_aio_context(void *s_, AioContext *old_context)
 {
     struct qemu_laio_state *s = s_;
 
     aio_set_event_notifier(old_context, &s->e, NULL);
     qemu_bh_delete(s->completion_bh);
-    qemu_bh_delete(s->io_q.retry);
+
+    laio_free_ioq(s, s->io_q);
 }
 
 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->io_q = laio_alloc_ioq(new_context, 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);
 }
@@ -395,8 +413,6 @@ void *laio_init(void)
         goto out_close_efd;
     }
 
-    ioq_init(&s->io_q);
-
     return s;
 
 out_close_efd:
-- 
1.7.9.5

[Qemu-devel] [PATCH 02/13] block: linux-aio: rename 'ctx' of qemu_laiocb as 'laio_state'

[Ming Lei]

So that it can be distinguished from the 'ctx' in qemu_laio_state.

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index b9db28e..cf8691e 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -29,7 +29,7 @@
 
 struct qemu_laiocb {
     BlockAIOCB common;
-    struct qemu_laio_state *ctx;
+    struct qemu_laio_state *aio_state;
     struct iocb iocb;
     ssize_t ret;
     size_t nbytes;
@@ -169,7 +169,7 @@ static void laio_cancel(BlockAIOCB *blockacb)
     if (laiocb->ret != -EINPROGRESS) {
         return;
     }
-    ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
+    ret = io_cancel(laiocb->aio_state->ctx, &laiocb->iocb, &event);
     laiocb->ret = -ECANCELED;
     if (ret != 0) {
         /* iocb is not cancelled, cb will be called by the event loop later */
@@ -179,7 +179,7 @@ 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);
+    qemu_laio_start_retry(laiocb->aio_state);
 }
 
 static const AIOCBInfo laio_aiocb_info = {
@@ -311,7 +311,7 @@ BlockAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
 
     laiocb = qemu_aio_get(&laio_aiocb_info, bs, cb, opaque);
     laiocb->nbytes = nb_sectors * 512;
-    laiocb->ctx = s;
+    laiocb->aio_state = s;
     laiocb->ret = -EINPROGRESS;
     laiocb->is_read = (type == QEMU_AIO_READ);
     laiocb->qiov = qiov;
-- 
1.7.9.5

[Qemu-devel] [PATCH 03/13] block/linux-aio: allocate 'struct qemu_laio_state' dynamically

[Ming Lei]

This patch allocates 'struct qemu_laio_state' in aio attach,
and frees it in aio detach, so that in the following patch
we can share one same instance of the structure among multiple
linux-aio backend in same AioContext.

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index cf8691e..5fa3c1e 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -52,6 +52,7 @@ typedef struct {
     QEMUBH *retry;
 } LaioQueue;
 
+/* lifetime: between aio_attach and aio_detach */
 struct qemu_laio_state {
     io_context_t ctx;
     EventNotifier e;
@@ -66,6 +67,10 @@ struct qemu_laio_state {
     int event_max;
 };
 
+typedef struct {
+    struct qemu_laio_state *state;
+} QemuLaioState;
+
 static inline ssize_t io_event_ret(struct io_event *ev)
 {
     return (ssize_t)(((uint64_t)ev->res2 << 32) | ev->res);
@@ -277,14 +282,16 @@ static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
 
 void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
 {
-    struct qemu_laio_state *s = aio_ctx;
+    QemuLaioState *qs = aio_ctx;
+    struct qemu_laio_state *s = qs->state;
 
     s->io_q->plugged++;
 }
 
 int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
 {
-    struct qemu_laio_state *s = aio_ctx;
+    QemuLaioState *qs = aio_ctx;
+    struct qemu_laio_state *s = qs->state;
     int ret = 0;
 
     assert(s->io_q->plugged > 0 || !unplug);
@@ -304,7 +311,8 @@ BlockAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
         BlockCompletionFunc *cb, void *opaque, int type)
 {
-    struct qemu_laio_state *s = aio_ctx;
+    QemuLaioState *qs = aio_ctx;
+    struct qemu_laio_state *s = qs->state;
     struct qemu_laiocb *laiocb;
     struct iocb *iocbs;
     off_t offset = sector_num * 512;
@@ -380,27 +388,7 @@ static void laio_free_ioq(struct qemu_laio_state *s, LaioQueue *ioq)
     s->io_q = NULL;
 }
 
-void laio_detach_aio_context(void *s_, AioContext *old_context)
-{
-    struct qemu_laio_state *s = s_;
-
-    aio_set_event_notifier(old_context, &s->e, NULL);
-    qemu_bh_delete(s->completion_bh);
-
-    laio_free_ioq(s, s->io_q);
-}
-
-void laio_attach_aio_context(void *s_, AioContext *new_context)
-{
-    struct qemu_laio_state *s = s_;
-
-    s->io_q = laio_alloc_ioq(new_context, 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);
-}
-
-void *laio_init(void)
+static struct qemu_laio_state *laio_state_alloc(void)
 {
     struct qemu_laio_state *s;
 
@@ -422,10 +410,8 @@ out_free_state:
     return NULL;
 }
 
-void laio_cleanup(void *s_)
+static void laio_state_free(struct qemu_laio_state *s)
 {
-    struct qemu_laio_state *s = s_;
-
     event_notifier_cleanup(&s->e);
 
     if (io_destroy(s->ctx) != 0) {
@@ -434,3 +420,43 @@ void laio_cleanup(void *s_)
     }
     g_free(s);
 }
+
+void laio_detach_aio_context(void *s_, AioContext *old_context)
+{
+    QemuLaioState *qs = s_;
+    struct qemu_laio_state *s = qs->state;
+
+    aio_set_event_notifier(old_context, &s->e, NULL);
+    qemu_bh_delete(s->completion_bh);
+
+    laio_free_ioq(s, s->io_q);
+    laio_state_free(s);
+    qs->state = NULL;
+}
+
+void laio_attach_aio_context(void *s_, AioContext *new_context)
+{
+    QemuLaioState *qs = s_;
+    struct qemu_laio_state *s = laio_state_alloc();
+
+    s->io_q = laio_alloc_ioq(new_context, 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);
+
+    qs->state = s;
+}
+
+void *laio_init(void)
+{
+    QemuLaioState *s = g_malloc0(sizeof(*s));
+
+    return s;
+}
+
+void laio_cleanup(void *s_)
+{
+    QemuLaioState *s = s_;
+
+    g_free(s);
+}
-- 
1.7.9.5

[Qemu-devel] [PATCH 04/13] block/linux-aio: do more things in laio_state_alloc() and its pair

[Ming Lei]

Now lifetime of 'completion_bh', io queue and io context is same,
so move their allocation into laio_state_alloc() and their
releasing into laio_state_free().

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index 5fa3c1e..95cd0dc 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -388,7 +388,7 @@ static void laio_free_ioq(struct qemu_laio_state *s, LaioQueue *ioq)
     s->io_q = NULL;
 }
 
-static struct qemu_laio_state *laio_state_alloc(void)
+static struct qemu_laio_state *laio_state_alloc(AioContext *context)
 {
     struct qemu_laio_state *s;
 
@@ -401,6 +401,10 @@ static struct qemu_laio_state *laio_state_alloc(void)
         goto out_close_efd;
     }
 
+    s->io_q = laio_alloc_ioq(context, s);
+    s->completion_bh = aio_bh_new(context, qemu_laio_completion_bh, s);
+    aio_set_event_notifier(context, &s->e, qemu_laio_completion_cb);
+
     return s;
 
 out_close_efd:
@@ -410,8 +414,12 @@ out_free_state:
     return NULL;
 }
 
-static void laio_state_free(struct qemu_laio_state *s)
+static void laio_state_free(struct qemu_laio_state *s, AioContext *context)
 {
+    aio_set_event_notifier(context, &s->e, NULL);
+    qemu_bh_delete(s->completion_bh);
+
+    laio_free_ioq(s, s->io_q);
     event_notifier_cleanup(&s->e);
 
     if (io_destroy(s->ctx) != 0) {
@@ -424,25 +432,15 @@ static void laio_state_free(struct qemu_laio_state *s)
 void laio_detach_aio_context(void *s_, AioContext *old_context)
 {
     QemuLaioState *qs = s_;
-    struct qemu_laio_state *s = qs->state;
 
-    aio_set_event_notifier(old_context, &s->e, NULL);
-    qemu_bh_delete(s->completion_bh);
-
-    laio_free_ioq(s, s->io_q);
-    laio_state_free(s);
+    laio_state_free(qs->state, old_context);
     qs->state = NULL;
 }
 
 void laio_attach_aio_context(void *s_, AioContext *new_context)
 {
     QemuLaioState *qs = s_;
-    struct qemu_laio_state *s = laio_state_alloc();
-
-    s->io_q = laio_alloc_ioq(new_context, 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);
+    struct qemu_laio_state *s = laio_state_alloc(new_context);
 
     qs->state = s;
 }
-- 
1.7.9.5

[Qemu-devel] [PATCH 05/13] block/linux-aio: pass 'BlockDriverState' to laio_attach_aio_context and its pair

[Ming Lei]

This patch introduces parameter of 'BlockDriverState' to
laio_attach_aio_context() and its pair, so that it will be
easier to support IO submission as batch in AioContext wide.

Signed-off-by: Ming Lei <ming.lei@canonical.com>
---
 block/linux-aio.c |    6 ++++--
 block/raw-aio.h   |    6 ++++--
 block/raw-posix.c |    4 ++--
 3 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/block/linux-aio.c b/block/linux-aio.c
index 95cd0dc..17de2e3 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -429,7 +429,8 @@ static void laio_state_free(struct qemu_laio_state *s, AioContext *context)
     g_free(s);
 }
 
-void laio_detach_aio_context(void *s_, AioContext *old_context)
+void laio_detach_aio_context(void *s_, BlockDriverState *bs,
+        AioContext *old_context)
 {
     QemuLaioState *qs = s_;
 
@@ -437,7 +438,8 @@ void laio_detach_aio_context(void *s_, AioContext *old_context)
     qs->state = NULL;
 }
 
-void laio_attach_aio_context(void *s_, AioContext *new_context)
+void laio_attach_aio_context(void *s_, BlockDriverState *bs,
+        AioContext *new_context)
 {
     QemuLaioState *qs = s_;
     struct qemu_laio_state *s = laio_state_alloc(new_context);
diff --git a/block/raw-aio.h b/block/raw-aio.h
index 80681ce..61a5b96 100644
--- a/block/raw-aio.h
+++ b/block/raw-aio.h
@@ -38,8 +38,10 @@ void laio_cleanup(void *s);
 BlockAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
         BlockCompletionFunc *cb, void *opaque, int type);
-void laio_detach_aio_context(void *s, AioContext *old_context);
-void laio_attach_aio_context(void *s, AioContext *new_context);
+void laio_detach_aio_context(void *s, BlockDriverState *bs,
+        AioContext *old_context);
+void laio_attach_aio_context(void *s, BlockDriverState *bs,
+        AioContext *new_context);
 void laio_io_plug(BlockDriverState *bs, void *aio_ctx);
 int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug);
 #endif
diff --git a/block/raw-posix.c b/block/raw-posix.c
index e100ae2..5b7f20b 100644
--- a/block/raw-posix.c
+++ b/block/raw-posix.c
@@ -323,7 +323,7 @@ static void raw_detach_aio_context(BlockDriverState *bs)
     BDRVRawState *s = bs->opaque;
 
     if (s->use_aio) {
-        laio_detach_aio_context(s->aio_ctx, bdrv_get_aio_context(bs));
+        laio_detach_aio_context(s->aio_ctx, bs, bdrv_get_aio_context(bs));
     }
 #endif
 }
@@ -335,7 +335,7 @@ static void raw_attach_aio_context(BlockDriverState *bs,
     BDRVRawState *s = bs->opaque;
 
     if (s->use_aio) {
-        laio_attach_aio_context(s->aio_ctx, new_context);
+        laio_attach_aio_context(s->aio_ctx, bs, new_context);
     }
 #endif
 }
-- 
1.7.9.5

[Qemu-devel] [PATCH 06/13] AioContext: introduce aio_attach_aio_bs() and its pair

[Ming Lei]

This patch introduces aio_attach_aio_bs() and its pair
for supporting IO submission as batch in AioContext wide,
and one typical use case is multi-lun SCSI.

aio_attach_aio_bs() will attach one 'bs' which is capable
of IO submission as batch, then all I/O submission in
this AioContext will borrow the io queue from this 'bs',
so that we can maximum IO submission as batch.

aio_detach_aio_bs() will detach the 'bs' when all 'bs'
in the AioContext is detached.

Signed-off-by: Ming Lei <ming.lei@canonical.com>
---
 async.c             |    1 +
 include/block/aio.h |   27 +++++++++++++++++++++++++++
 2 files changed, 28 insertions(+)

diff --git a/async.c b/async.c
index 6e1b282..0f4b530 100644
--- a/async.c
+++ b/async.c
@@ -305,6 +305,7 @@ AioContext *aio_context_new(Error **errp)
                            event_notifier_test_and_clear);
     ctx->pollfds = g_array_new(FALSE, FALSE, sizeof(GPollFD));
     ctx->thread_pool = NULL;
+    ctx->bs_refcnt = 0;
     qemu_mutex_init(&ctx->bh_lock);
     rfifolock_init(&ctx->lock, aio_rfifolock_cb, ctx);
     timerlistgroup_init(&ctx->tlg, aio_timerlist_notify, ctx);
diff --git a/include/block/aio.h b/include/block/aio.h
index 6bf0e04..e34a21f 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -90,6 +90,11 @@ struct AioContext {
 
     /* TimerLists for calling timers - one per clock type */
     QEMUTimerListGroup tlg;
+
+    /* IO submit as batch in AioContext wide */
+    BlockDriverState *master_aio_bs;
+    int bs_refcnt;
+    void *opaque;
 };
 
 /* Used internally to synchronize aio_poll against qemu_bh_schedule.  */
@@ -325,4 +330,26 @@ static inline void aio_timer_init(AioContext *ctx,
  */
 int64_t aio_compute_timeout(AioContext *ctx);
 
+static inline bool aio_attach_aio_bs(AioContext *ctx,
+                                     BlockDriverState *bs)
+{
+    if (!ctx->bs_refcnt++) {
+        ctx->master_aio_bs = bs;
+        return true;
+    }
+
+    return false;
+}
+
+static inline bool aio_detach_aio_bs(AioContext *ctx,
+                                     BlockDriverState *bs)
+{
+    if (!--ctx->bs_refcnt) {
+        ctx->master_aio_bs = NULL;
+        return true;
+    }
+
+    return false;
+}
+
 #endif
-- 
1.7.9.5

[Qemu-devel] [PATCH 07/13] block/linux-aio: support IO submission as batch in AioContext wide

[Ming Lei]

This patch supports IO submission as batch in AioContext wide
by sharing 'struct qemu_laio_state' instance among all linux-aio
backend in same AioContext.

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index 17de2e3..c5c7944 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -52,6 +52,11 @@ typedef struct {
     QEMUBH *retry;
 } LaioQueue;
 
+typedef struct LaioTrackedBs {
+    BlockDriverState *bs;
+    QLIST_ENTRY(LaioTrackedBs) list;
+} LaioTrackedBs;
+
 /* lifetime: between aio_attach and aio_detach */
 struct qemu_laio_state {
     io_context_t ctx;
@@ -65,6 +70,9 @@ struct qemu_laio_state {
     struct io_event events[MAX_EVENTS];
     int event_idx;
     int event_max;
+
+    /* All BS in the list shared this 'qemu_laio_state' */
+    QLIST_HEAD(, LaioTrackedBs) tracked_bs;
 };
 
 typedef struct {
@@ -433,6 +441,23 @@ void laio_detach_aio_context(void *s_, BlockDriverState *bs,
         AioContext *old_context)
 {
     QemuLaioState *qs = s_;
+    LaioTrackedBs *tbs, *ntbs;
+
+    QLIST_FOREACH_SAFE(tbs, &qs->state->tracked_bs, list, ntbs) {
+        if (tbs->bs == bs) {
+            QLIST_REMOVE(tbs, list);
+            g_free(tbs);
+        }
+    }
+
+    if (!aio_detach_aio_bs(old_context, bs)) {
+        /* assign new master aio bs for the aio context */
+        if (old_context->master_aio_bs == bs) {
+            tbs = QLIST_FIRST(&qs->state->tracked_bs);
+            old_context->master_aio_bs = tbs->bs;
+        }
+        return;
+    }
 
     laio_state_free(qs->state, old_context);
     qs->state = NULL;
@@ -442,9 +467,16 @@ void laio_attach_aio_context(void *s_, BlockDriverState *bs,
         AioContext *new_context)
 {
     QemuLaioState *qs = s_;
-    struct qemu_laio_state *s = laio_state_alloc(new_context);
+    LaioTrackedBs *tbs = g_malloc0(sizeof(*tbs));
+
+    if (aio_attach_aio_bs(new_context, bs)) {
+        new_context->opaque = qs->state = laio_state_alloc(new_context);
+    } else {
+        qs->state = new_context->opaque;
+    }
 
-    qs->state = s;
+    tbs->bs = bs;
+    QLIST_INSERT_HEAD(&qs->state->tracked_bs, tbs, list);
 }
 
 void *laio_init(void)
-- 
1.7.9.5

[Qemu-devel] [PATCH 08/13] block/linux-aio.c: allocate events dynamically

[Ming Lei]

This patch allocates events array of 'struct qemu_laio_state'
dynamically so that in the following patch we can allocate
resource elasticly in case of AioContext wide IO submission
as batch.

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index c5c7944..5be8036 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -67,7 +67,7 @@ struct qemu_laio_state {
 
     /* I/O completion processing */
     QEMUBH *completion_bh;
-    struct io_event events[MAX_EVENTS];
+    struct io_event *events;
     int event_idx;
     int event_max;
 
@@ -409,6 +409,7 @@ static struct qemu_laio_state *laio_state_alloc(AioContext *context)
         goto out_close_efd;
     }
 
+    s->events = g_malloc(sizeof(*s->events) * MAX_EVENTS);
     s->io_q = laio_alloc_ioq(context, s);
     s->completion_bh = aio_bh_new(context, qemu_laio_completion_bh, s);
     aio_set_event_notifier(context, &s->e, qemu_laio_completion_cb);
@@ -429,6 +430,7 @@ static void laio_state_free(struct qemu_laio_state *s, AioContext *context)
 
     laio_free_ioq(s, s->io_q);
     event_notifier_cleanup(&s->e);
+    g_free(s->events);
 
     if (io_destroy(s->ctx) != 0) {
         fprintf(stderr, "%s: destroy AIO context %p failed\n",
-- 
1.7.9.5

[Qemu-devel] [PATCH 09/13] block/linux-aio.c: introduce laio_alloc_resource()

[Ming Lei]

This patch introduces laio_alloc_resource() for allocating
resources for linux aio, then in the following patchs we
can allocate IO resources just in demand.

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index 5be8036..e3e0532 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -73,6 +73,7 @@ struct qemu_laio_state {
 
     /* All BS in the list shared this 'qemu_laio_state' */
     QLIST_HEAD(, LaioTrackedBs) tracked_bs;
+    AioContext *aio_context;
 };
 
 typedef struct {
@@ -380,20 +381,45 @@ out_free_aiocb:
     return NULL;
 }
 
-static LaioQueue *laio_alloc_ioq(AioContext *ctx, struct qemu_laio_state *s)
+static int laio_alloc_resources(AioContext *ctx,
+        struct qemu_laio_state *s)
 {
-    LaioQueue *ioq = g_malloc0(sizeof(*ioq));
+    LaioQueue *ioq;
 
+    if (io_setup(MAX_EVENTS, &s->ctx) != 0) {
+        return -1;
+    }
+
+    ioq = g_malloc0(sizeof(*s->io_q));
     ioq_init(ioq);
     ioq->retry = aio_bh_new(ctx, ioq_submit_retry, s);
-    return ioq;
+
+    s->events = g_malloc(sizeof(*s->events) * MAX_EVENTS);
+    s->io_q = ioq;
+
+    s->completion_bh = aio_bh_new(ctx, qemu_laio_completion_bh, s);
+    aio_set_event_notifier(ctx, &s->e, qemu_laio_completion_cb);
+
+    return 0;
 }
 
-static void laio_free_ioq(struct qemu_laio_state *s, LaioQueue *ioq)
+static void laio_free_resources(struct qemu_laio_state *s)
 {
+    LaioQueue *ioq = s->io_q;
+
+    aio_set_event_notifier(s->aio_context, &s->e, NULL);
+    qemu_bh_delete(s->completion_bh);
+
+    g_free(s->events);
+
     qemu_bh_delete(ioq->retry);
     g_free(ioq);
     s->io_q = NULL;
+
+    if (io_destroy(s->ctx) != 0) {
+        fprintf(stderr, "%s: destroy AIO context %p failed\n",
+                        __func__, &s->ctx);
+    }
 }
 
 static struct qemu_laio_state *laio_state_alloc(AioContext *context)
@@ -405,15 +431,10 @@ static struct qemu_laio_state *laio_state_alloc(AioContext *context)
         goto out_free_state;
     }
 
-    if (io_setup(MAX_EVENTS, &s->ctx) != 0) {
+    if (laio_alloc_resources(context, s) != 0) {
         goto out_close_efd;
     }
 
-    s->events = g_malloc(sizeof(*s->events) * MAX_EVENTS);
-    s->io_q = laio_alloc_ioq(context, s);
-    s->completion_bh = aio_bh_new(context, qemu_laio_completion_bh, s);
-    aio_set_event_notifier(context, &s->e, qemu_laio_completion_cb);
-
     return s;
 
 out_close_efd:
@@ -425,17 +446,8 @@ out_free_state:
 
 static void laio_state_free(struct qemu_laio_state *s, AioContext *context)
 {
-    aio_set_event_notifier(context, &s->e, NULL);
-    qemu_bh_delete(s->completion_bh);
-
-    laio_free_ioq(s, s->io_q);
+    laio_free_resources(s);
     event_notifier_cleanup(&s->e);
-    g_free(s->events);
-
-    if (io_destroy(s->ctx) != 0) {
-        fprintf(stderr, "%s: destroy AIO context %p failed\n",
-                        __func__, &s->ctx);
-    }
     g_free(s);
 }
 
@@ -473,6 +485,9 @@ void laio_attach_aio_context(void *s_, BlockDriverState *bs,
 
     if (aio_attach_aio_bs(new_context, bs)) {
         new_context->opaque = qs->state = laio_state_alloc(new_context);
+
+        /* qemu_laio_state is per AioContext */
+        qs->state->aio_context = new_context;
     } else {
         qs->state = new_context->opaque;
     }
-- 
1.7.9.5

[Qemu-devel] [PATCH 10/13] block/linux-aio.c: prepare for elastical resource's allocation

[Ming Lei]

Because we will support AioContext wide IO submission as batch,
so IO resources should be allocated according to how many there
are backends in the same AioContext.

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index e3e0532..e219b80 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -43,7 +43,7 @@ struct qemu_laiocb {
  * so in future the IO queue should be per AIO context.
  */
 typedef struct {
-    struct iocb *iocbs[MAX_QUEUED_IO];
+    struct iocb **iocbs;
     int plugged;
     unsigned int size;
     unsigned int idx;
@@ -66,6 +66,7 @@ struct qemu_laio_state {
     LaioQueue *io_q;
 
     /* I/O completion processing */
+    int nr_evt;
     QEMUBH *completion_bh;
     struct io_event *events;
     int event_idx;
@@ -73,6 +74,7 @@ struct qemu_laio_state {
 
     /* All BS in the list shared this 'qemu_laio_state' */
     QLIST_HEAD(, LaioTrackedBs) tracked_bs;
+    int nr_bs;
     AioContext *aio_context;
 };
 
@@ -131,7 +133,7 @@ static void qemu_laio_completion_bh(void *opaque)
     if (s->event_idx == s->event_max) {
         do {
             struct timespec ts = { 0 };
-            s->event_max = io_getevents(s->ctx, MAX_EVENTS, MAX_EVENTS,
+            s->event_max = io_getevents(s->ctx, s->nr_evt, s->nr_evt,
                                         s->events, &ts);
         } while (s->event_max == -EINTR);
 
@@ -201,11 +203,13 @@ static const AIOCBInfo laio_aiocb_info = {
     .cancel_async       = laio_cancel,
 };
 
-static void ioq_init(LaioQueue *io_q)
+static void ioq_init(LaioQueue *io_q, unsigned size)
 {
-    io_q->size = MAX_QUEUED_IO;
+    io_q->size = size;
     io_q->idx = 0;
     io_q->plugged = 0;
+
+    io_q->iocbs = g_malloc(sizeof(*io_q->iocbs) * size);
 }
 
 static void abort_queue(struct qemu_laio_state *s)
@@ -385,16 +389,34 @@ static int laio_alloc_resources(AioContext *ctx,
         struct qemu_laio_state *s)
 {
     LaioQueue *ioq;
+    unsigned nr = s->nr_bs ? : 1;
+
+    /* return immediately if resources are allocated already */
+    if (likely(s->io_q)) {
+        return 0;
+    }
+
+    if (nr > 10) {
+        nr = 10;
+    }
+
+    while (nr > 0) {
+        if (io_setup(MAX_EVENTS * nr, &s->ctx) == 0) {
+            break;
+        }
+        nr--;
+    }
 
-    if (io_setup(MAX_EVENTS, &s->ctx) != 0) {
+    if (nr == 0) {
         return -1;
     }
 
+    s->nr_evt = nr * MAX_EVENTS;
     ioq = g_malloc0(sizeof(*s->io_q));
-    ioq_init(ioq);
+    ioq_init(ioq, MAX_QUEUED_IO * nr);
     ioq->retry = aio_bh_new(ctx, ioq_submit_retry, s);
 
-    s->events = g_malloc(sizeof(*s->events) * MAX_EVENTS);
+    s->events = g_malloc(sizeof(*s->events) * s->nr_evt);
     s->io_q = ioq;
 
     s->completion_bh = aio_bh_new(ctx, qemu_laio_completion_bh, s);
@@ -464,6 +486,7 @@ void laio_detach_aio_context(void *s_, BlockDriverState *bs,
         }
     }
 
+    qs->state->nr_bs--;
     if (!aio_detach_aio_bs(old_context, bs)) {
         /* assign new master aio bs for the aio context */
         if (old_context->master_aio_bs == bs) {
@@ -494,6 +517,7 @@ void laio_attach_aio_context(void *s_, BlockDriverState *bs,
 
     tbs->bs = bs;
     QLIST_INSERT_HEAD(&qs->state->tracked_bs, tbs, list);
+    qs->state->nr_bs++;
 }
 
 void *laio_init(void)
-- 
1.7.9.5

[Qemu-devel] [PATCH 11/13] block/linux-aio: reallocate I/O resources when aio attached

[Ming Lei]

First event notifier and qemu BH is associated with aio
context, so it should be reallocated for making these
handlers running in the current attached aio context.

Secondly when new 'bs' is attached, we need to allocate
more io resources for performance purpose.

This patch only does the reallocation if there aren't
any pending I/O.

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

diff --git a/block/linux-aio.c b/block/linux-aio.c
index e219b80..c5a88e8 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -59,6 +59,7 @@ typedef struct LaioTrackedBs {
 
 /* lifetime: between aio_attach and aio_detach */
 struct qemu_laio_state {
+    unsigned long io_pending;
     io_context_t ctx;
     EventNotifier e;
 
@@ -75,6 +76,7 @@ struct qemu_laio_state {
     /* All BS in the list shared this 'qemu_laio_state' */
     QLIST_HEAD(, LaioTrackedBs) tracked_bs;
     int nr_bs;
+    bool need_realloc;
     AioContext *aio_context;
 };
 
@@ -82,6 +84,8 @@ typedef struct {
     struct qemu_laio_state *state;
 } QemuLaioState;
 
+static void laio_realloc_resources(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);
@@ -110,6 +114,8 @@ static void qemu_laio_process_completion(struct qemu_laio_state *s,
         }
     }
     laiocb->common.cb(laiocb->common.opaque, ret);
+    s->io_pending--;
+    laio_realloc_resources(s);
 
     qemu_aio_unref(laiocb);
 }
@@ -193,6 +199,8 @@ static void laio_cancel(BlockAIOCB *blockacb)
     }
 
     laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
+    laiocb->aio_state->io_pending--;
+    laio_realloc_resources(laiocb->aio_state);
 
     /* check if there are requests in io queue */
     qemu_laio_start_retry(laiocb->aio_state);
@@ -378,6 +386,8 @@ BlockAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
             goto out_free_aiocb;
         }
     }
+
+    s->io_pending++;
     return &laiocb->common;
 
 out_free_aiocb:
@@ -429,6 +439,10 @@ static void laio_free_resources(struct qemu_laio_state *s)
 {
     LaioQueue *ioq = s->io_q;
 
+    if (!ioq) {
+        return;
+    }
+
     aio_set_event_notifier(s->aio_context, &s->e, NULL);
     qemu_bh_delete(s->completion_bh);
 
@@ -441,6 +455,8 @@ static void laio_free_resources(struct qemu_laio_state *s)
     if (io_destroy(s->ctx) != 0) {
         fprintf(stderr, "%s: destroy AIO context %p failed\n",
                         __func__, &s->ctx);
+    } else {
+        s->ctx = 0;
     }
 }
 
@@ -473,6 +489,17 @@ static void laio_state_free(struct qemu_laio_state *s, AioContext *context)
     g_free(s);
 }
 
+static void laio_realloc_resources(struct qemu_laio_state *s)
+{
+    if (unlikely(s->need_realloc) && unlikely(!s->io_pending) &&
+        !s->io_q->plugged) {
+        laio_free_resources(s);
+        laio_alloc_resources(s->aio_context, s);
+
+        s->need_realloc = false;
+    }
+}
+
 void laio_detach_aio_context(void *s_, BlockDriverState *bs,
         AioContext *old_context)
 {
@@ -493,6 +520,7 @@ void laio_detach_aio_context(void *s_, BlockDriverState *bs,
             tbs = QLIST_FIRST(&qs->state->tracked_bs);
             old_context->master_aio_bs = tbs->bs;
         }
+        qs->state->need_realloc = true;
         return;
     }
 
@@ -513,11 +541,13 @@ void laio_attach_aio_context(void *s_, BlockDriverState *bs,
         qs->state->aio_context = new_context;
     } else {
         qs->state = new_context->opaque;
+        qs->state->need_realloc = true;
     }
 
     tbs->bs = bs;
     QLIST_INSERT_HEAD(&qs->state->tracked_bs, tbs, list);
     qs->state->nr_bs++;
+    qs->state->aio_context = new_context;
 }
 
 void *laio_init(void)
-- 
1.7.9.5

[Qemu-devel] [PATCH 12/13] block: introduce bdrv_aio_io_plug() and its pair

[Ming Lei]

These two APIs are introduced for using AioContext wide
IO submission as batch.

Signed-off-by: Ming Lei <ming.lei@canonical.com>
---
 block.c               |   16 ++++++++++++++++
 include/block/block.h |    3 +++
 2 files changed, 19 insertions(+)

diff --git a/block.c b/block.c
index dacd881..0200af0 100644
--- a/block.c
+++ b/block.c
@@ -5901,6 +5901,22 @@ void bdrv_io_unplug(BlockDriverState *bs)
     }
 }
 
+bool bdrv_aio_io_plug(AioContext *aio_ctx)
+{
+    if (aio_ctx->master_aio_bs) {
+        bdrv_io_plug(aio_ctx->master_aio_bs);
+        return true;
+    }
+    return false;
+}
+
+void bdrv_aio_io_unplug(AioContext *aio_ctx)
+{
+    if (aio_ctx->master_aio_bs) {
+        bdrv_io_unplug(aio_ctx->master_aio_bs);
+    }
+}
+
 void bdrv_flush_io_queue(BlockDriverState *bs)
 {
     BlockDriver *drv = bs->drv;
diff --git a/include/block/block.h b/include/block/block.h
index 13e4537..aae6b66 100644
--- a/include/block/block.h
+++ b/include/block/block.h
@@ -541,6 +541,9 @@ void bdrv_io_plug(BlockDriverState *bs);
 void bdrv_io_unplug(BlockDriverState *bs);
 void bdrv_flush_io_queue(BlockDriverState *bs);
 
+bool bdrv_aio_io_plug(AioContext *aio_ctx);
+void bdrv_aio_io_unplug(AioContext *aio_ctx);
+
 BlockAcctStats *bdrv_get_stats(BlockDriverState *bs);
 
 #endif
-- 
1.7.9.5

[Qemu-devel] [PATCH 13/13] virtio-scsi-dataplane: support AioContext wide IO submission as batch

[Ming Lei]

Replace previous usage with AioContext Wide approach.

Signed-off-by: Ming Lei <ming.lei@canonical.com>
---
 hw/scsi/virtio-scsi-dataplane.c |    8 ++++++++
 hw/scsi/virtio-scsi.c           |    2 --
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/hw/scsi/virtio-scsi-dataplane.c b/hw/scsi/virtio-scsi-dataplane.c
index 9651e6f..7fab14d 100644
--- a/hw/scsi/virtio-scsi-dataplane.c
+++ b/hw/scsi/virtio-scsi-dataplane.c
@@ -135,8 +135,12 @@ static void virtio_scsi_iothread_handle_cmd(EventNotifier *notifier)
     VirtIOSCSI *s = (VirtIOSCSI *)vring->parent;
     VirtIOSCSIReq *req, *next;
     QTAILQ_HEAD(, VirtIOSCSIReq) reqs = QTAILQ_HEAD_INITIALIZER(reqs);
+    AioContext *ctx = s->ctx;
+    bool plugged;
 
     event_notifier_test_and_clear(notifier);
+
+    plugged = bdrv_aio_io_plug(ctx);
     while ((req = virtio_scsi_pop_req_vring(s, vring))) {
         if (virtio_scsi_handle_cmd_req_prepare(s, req)) {
             QTAILQ_INSERT_TAIL(&reqs, req, next);
@@ -146,6 +150,10 @@ static void virtio_scsi_iothread_handle_cmd(EventNotifier *notifier)
     QTAILQ_FOREACH_SAFE(req, &reqs, next, next) {
         virtio_scsi_handle_cmd_req_submit(s, req);
     }
+
+    if (plugged) {
+        bdrv_aio_io_unplug(ctx);
+    }
 }
 
 /* assumes s->ctx held */
diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
index fdcacfd..6e34a2c 100644
--- a/hw/scsi/virtio-scsi.c
+++ b/hw/scsi/virtio-scsi.c
@@ -540,7 +540,6 @@ bool virtio_scsi_handle_cmd_req_prepare(VirtIOSCSI *s, VirtIOSCSIReq *req)
         return false;
     }
     scsi_req_ref(req->sreq);
-    blk_io_plug(d->conf.blk);
     return true;
 }
 
@@ -550,7 +549,6 @@ void virtio_scsi_handle_cmd_req_submit(VirtIOSCSI *s, VirtIOSCSIReq *req)
     if (scsi_req_enqueue(sreq)) {
         scsi_req_continue(sreq);
     }
-    blk_io_unplug(sreq->dev->conf.blk);
     scsi_req_unref(sreq);
 }
 
-- 
1.7.9.5

Re: [Qemu-devel] [PATCH 00/13] linux-aio/virtio-scsi: support AioContext wide IO submission as batch

[Paolo Bonzini]

On 09/11/2014 08:42, Ming Lei wrote:
> This patch implements AioContext wide IO submission as batch, and
> the idea behind is very simple:
> 
> 	- linux native aio(io_submit) supports to enqueue read/write requests
> 	to different files
> 
> 	- in one AioContext, I/O requests from VM can be submitted to different
> 	backend in host, one typical example is multi-lun scsi 
> 
> This patch changes 'struct qemu_laio_state' as per AioContext, and
> multiple 'bs' can be associted with one single instance of
> 'struct qemu_laio_state', then AioContext wide IO submission as batch
> becomes easy to implement.
> 
> One simple test in my laptop shows ~20% throughput improvement
> on randread from VM(using AioContext wide IO batch vs. not using io batch)
> with below config:
> 
> 	-drive id=drive_scsi1-0-0-0,if=none,format=raw,cache=none,aio=native,file=/dev/nullb2 \
> 	-drive id=drive_scsi1-0-0-1,if=none,format=raw,cache=none,aio=native,file=/dev/nullb3 \
> 	-device virtio-scsi-pci,num_queues=4,id=scsi1,addr=07,iothread=iothread0 \
> 	-device scsi-disk,bus=scsi1.0,channel=0,scsi-id=1,lun=0,drive=drive_scsi1-0-0-0,id=scsi1-0-0-0 \
> 	-device scsi-disk,bus=scsi1.0,channel=0,scsi-id=1,lun=1,drive=drive_scsi1-0-0-1,id=scsi1-0-0-1 \
> 
> BTW, maybe more boost can be obtained since ~33K/sec write() system call
> can be observed when this test case is running, and it might be a recent
> regression(BH?).

Ming,

these patches are interesting.  I would like to compare them with the
opposite approach (and, I think, more similar to your old work) where
the qemu_laio_state API is moved entirely into AioContext, with lazy
allocation (reference-counted too, probably).

Most of the patches would be the same, but you would replace
aio_attach_aio_bs/aio_detach_aio_bs with something like
aio_native_get/aio_native_unref.  Ultimately block/{linux,win32}-aio.c
could be merged into block/aio-{posix,win32}.c, but you do not have to
do that now.

Could you try that?  This way we can see which API turns out to be nicer.

Thanks,

Paolo

> This patchset can be found on below tree too:
> 
> 	git://kernel.ubuntu.com/ming/qemu.git aio-io-batch.2
> 
> and these patches depend on "linux-aio: fix batch submission" patches
> in below link:
> 
> 	http://marc.info/?l=qemu-devel&m=141528663106557&w=2
> 
> Any comments and suggestions are welcome.
> 
>  async.c                         |    1 +
>  block.c                         |   16 +++
>  block/linux-aio.c               |  251 ++++++++++++++++++++++++++++++---------
>  block/raw-aio.h                 |    6 +-
>  block/raw-posix.c               |    4 +-
>  hw/scsi/virtio-scsi-dataplane.c |    8 ++
>  hw/scsi/virtio-scsi.c           |    2 -
>  include/block/aio.h             |   27 +++++
>  include/block/block.h           |    3 +
>  9 files changed, 259 insertions(+), 59 deletions(-)
> 
> Thanks,
> Ming Lei
> 

Re: [Qemu-devel] [PATCH 00/13] linux-aio/virtio-scsi: support AioContext wide IO submission as batch

[Ming Lei]

On Tue, Nov 18, 2014 at 9:57 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
> On 09/11/2014 08:42, Ming Lei wrote:
>> This patch implements AioContext wide IO submission as batch, and
>> the idea behind is very simple:
>>
>>       - linux native aio(io_submit) supports to enqueue read/write requests
>>       to different files
>>
>>       - in one AioContext, I/O requests from VM can be submitted to different
>>       backend in host, one typical example is multi-lun scsi
>>
>> This patch changes 'struct qemu_laio_state' as per AioContext, and
>> multiple 'bs' can be associted with one single instance of
>> 'struct qemu_laio_state', then AioContext wide IO submission as batch
>> becomes easy to implement.
>>
>> One simple test in my laptop shows ~20% throughput improvement
>> on randread from VM(using AioContext wide IO batch vs. not using io batch)
>> with below config:
>>
>>       -drive id=drive_scsi1-0-0-0,if=none,format=raw,cache=none,aio=native,file=/dev/nullb2 \
>>       -drive id=drive_scsi1-0-0-1,if=none,format=raw,cache=none,aio=native,file=/dev/nullb3 \
>>       -device virtio-scsi-pci,num_queues=4,id=scsi1,addr=07,iothread=iothread0 \
>>       -device scsi-disk,bus=scsi1.0,channel=0,scsi-id=1,lun=0,drive=drive_scsi1-0-0-0,id=scsi1-0-0-0 \
>>       -device scsi-disk,bus=scsi1.0,channel=0,scsi-id=1,lun=1,drive=drive_scsi1-0-0-1,id=scsi1-0-0-1 \
>>
>> BTW, maybe more boost can be obtained since ~33K/sec write() system call
>> can be observed when this test case is running, and it might be a recent
>> regression(BH?).
>
> Ming,
>
> these patches are interesting.  I would like to compare them with the
> opposite approach (and, I think, more similar to your old work) where
> the qemu_laio_state API is moved entirely into AioContext, with lazy
> allocation (reference-counted too, probably).

Yes, it can be done in that way, but the feature is linux native aio
specific, so it might not be good to put it into AioContext.

Basically most of the implementation should be same, and the
difference should be where the io queue is put.

>
> Most of the patches would be the same, but you would replace
> aio_attach_aio_bs/aio_detach_aio_bs with something like
> aio_native_get/aio_native_unref.  Ultimately block/{linux,win32}-aio.c
> could be merged into block/aio-{posix,win32}.c, but you do not have to
> do that now.
>
> Could you try that?  This way we can see which API turns out to be nicer.

I can try that, could you share what the APIs you prefer to?

IMO, the APIs can be defined flexiably in this patchset, and only
the AioContext parameter is enough.

Thanks,

Re: [Qemu-devel] [PATCH 00/13] linux-aio/virtio-scsi: support AioContext wide IO submission as batch

[Paolo Bonzini]

On 22/11/2014 13:33, Ming Lei wrote:
> > these patches are interesting.  I would like to compare them with the
> > opposite approach (and, I think, more similar to your old work) where
> > the qemu_laio_state API is moved entirely into AioContext, with lazy
> > allocation (reference-counted too, probably).
> 
> Yes, it can be done in that way, but the feature is linux native aio
> specific, so it might not be good to put it into AioContext.

I think it's not a problem as long as the eventfd and io queue is
created lazily.  My main issue with these series is that
aio_attach_aio_bs() (and detach) feels like a very ad hoc API.  Adding
io queue support directly in AioContext sounds better.

> Basically most of the implementation should be same, and the
> difference should be where the io queue is put.

Yes, the change is not big.

Paolo