[Qemu-devel] [PATCH v1 16/17] virtio-blk: dataplane: support multi virtqueue

[Ming Lei <ming.lei@canonical.com>]

This patch supports to handle host notify from multi
virt queues, but still process/submit io in the
one iothread.

One BH is introduced to process I/O from all virtqueues,
so that they can be sumitted to kernel in batch as far
as possible.

Signed-off-by: Ming Lei <ming.lei@canonical.com>
---
 hw/block/dataplane/virtio-blk.c |  211 ++++++++++++++++++++++++++++-----------
 1 file changed, 153 insertions(+), 58 deletions(-)

diff --git a/hw/block/dataplane/virtio-blk.c b/hw/block/dataplane/virtio-blk.c
index a0732e3..d61a920 100644
--- a/hw/block/dataplane/virtio-blk.c
+++ b/hw/block/dataplane/virtio-blk.c
@@ -26,6 +26,11 @@
 
 #define REQ_POOL_SZ 128
 
+typedef struct {
+    EventNotifier notifier;
+    VirtIOBlockDataPlane *s;
+} VirtIOBlockNotifier;
+
 struct VirtIOBlockDataPlane {
     bool started;
     bool starting;
@@ -35,9 +40,10 @@ struct VirtIOBlockDataPlane {
     VirtIOBlkConf *blk;
 
     VirtIODevice *vdev;
-    Vring vring;                    /* virtqueue vring */
-    EventNotifier *guest_notifier;  /* irq */
-    QEMUBH *bh;                     /* bh for guest notification */
+    Vring *vring;                    /* virtqueue vring */
+    EventNotifier **guest_notifier;  /* irq */
+    uint64_t   pending_guest_notifier;  /* pending guest notifer for vq */
+    QEMUBH *bh;                      /* bh for guest notification */
 
     /* Note that these EventNotifiers are assigned by value.  This is
      * fine as long as you do not call event_notifier_cleanup on them
@@ -47,7 +53,9 @@ struct VirtIOBlockDataPlane {
     IOThread *iothread;
     IOThread internal_iothread_obj;
     AioContext *ctx;
-    EventNotifier host_notifier;    /* doorbell */
+    VirtIOBlockNotifier *host_notifier; /* doorbell */
+    uint64_t   pending_host_notifier;   /* pending host notifer for vq */
+    QEMUBH *host_notifier_bh;           /* for handle host notifier */
 
     /* Operation blocker on BDS */
     Error *blocker;
@@ -60,20 +68,26 @@ struct VirtIOBlockDataPlane {
 };
 
 /* Raise an interrupt to signal guest, if necessary */
-static void notify_guest(VirtIOBlockDataPlane *s)
+static void notify_guest(VirtIOBlockDataPlane *s, unsigned int qid)
 {
-    if (!vring_should_notify(s->vdev, &s->vring)) {
-        return;
+    if (vring_should_notify(s->vdev, &s->vring[qid])) {
+        event_notifier_set(s->guest_notifier[qid]);
     }
-
-    event_notifier_set(s->guest_notifier);
 }
 
 static void notify_guest_bh(void *opaque)
 {
     VirtIOBlockDataPlane *s = opaque;
+    unsigned int qid;
+    uint64_t pending = s->pending_guest_notifier;
+
+    s->pending_guest_notifier = 0;
 
-    notify_guest(s);
+    while ((qid = ffsl(pending))) {
+        qid--;
+        notify_guest(s, qid);
+        pending &= ~(1 << qid);
+    }
 }
 
 static void complete_request_vring(VirtIOBlockReq *req, unsigned char status)
@@ -81,7 +95,7 @@ static void complete_request_vring(VirtIOBlockReq *req, unsigned char status)
     VirtIOBlockDataPlane *s = req->dev->dataplane;
     stb_p(&req->in->status, status);
 
-    vring_push(&req->dev->dataplane->vring, &req->elem,
+    vring_push(&s->vring[req->qid], &req->elem,
                req->qiov.size + sizeof(*req->in));
 
     /* Suppress notification to guest by BH and its scheduled
@@ -90,17 +104,15 @@ static void complete_request_vring(VirtIOBlockReq *req, unsigned char status)
      * executed in dataplane aio context even after it is
      * stopped, so needn't worry about notification loss with BH.
      */
+    assert(req->qid < 64);
+    s->pending_guest_notifier |= (1 << req->qid);
     qemu_bh_schedule(s->bh);
 }
 
-static void handle_notify(EventNotifier *e)
+static void process_vq_notify(VirtIOBlockDataPlane *s, unsigned short qid)
 {
-    VirtIOBlockDataPlane *s = container_of(e, VirtIOBlockDataPlane,
-                                           host_notifier);
     VirtIOBlock *vblk = VIRTIO_BLK(s->vdev);
 
-    event_notifier_test_and_clear(&s->host_notifier);
-    bdrv_io_plug(s->blk->conf.bs);
     for (;;) {
         MultiReqBuffer mrb = {
             .num_writes = 0,
@@ -108,12 +120,13 @@ static void handle_notify(EventNotifier *e)
         int ret;
 
         /* Disable guest->host notifies to avoid unnecessary vmexits */
-        vring_disable_notification(s->vdev, &s->vring);
+        vring_disable_notification(s->vdev, &s->vring[qid]);
 
         for (;;) {
             VirtIOBlockReq *req = virtio_blk_alloc_request(vblk);
 
-            ret = vring_pop(s->vdev, &s->vring, &req->elem);
+            req->qid = qid;
+            ret = vring_pop(s->vdev, &s->vring[qid], &req->elem);
             if (ret < 0) {
                 virtio_blk_free_request(req);
                 break; /* no more requests */
@@ -132,16 +145,48 @@ static void handle_notify(EventNotifier *e)
             /* Re-enable guest->host notifies and stop processing the vring.
              * But if the guest has snuck in more descriptors, keep processing.
              */
-            if (vring_enable_notification(s->vdev, &s->vring)) {
+            if (vring_enable_notification(s->vdev, &s->vring[qid])) {
                 break;
             }
         } else { /* fatal error */
             break;
         }
     }
+}
+
+static void process_notify(void *opaque)
+{
+    VirtIOBlockDataPlane *s = opaque;
+    unsigned int qid;
+    uint64_t pending = s->pending_host_notifier;
+
+    s->pending_host_notifier = 0;
+
+    bdrv_io_plug(s->blk->conf.bs);
+    while ((qid = ffsl(pending))) {
+        qid--;
+        process_vq_notify(s, qid);
+        pending &= ~(1 << qid);
+    }
     bdrv_io_unplug(s->blk->conf.bs);
 }
 
+/* TODO: handle requests from other vqs together */
+static void handle_notify(EventNotifier *e)
+{
+    VirtIOBlockNotifier *n = container_of(e, VirtIOBlockNotifier,
+		                          notifier);
+    VirtIOBlockDataPlane *s = n->s;
+    unsigned int qid = n - &s->host_notifier[0];
+
+    assert(qid < 64);
+
+    event_notifier_test_and_clear(e);
+
+    s->pending_host_notifier |= (1 << qid);
+    qemu_bh_schedule(s->host_notifier_bh);
+}
+
 /* Context: QEMU global mutex held */
 void virtio_blk_data_plane_create(VirtIODevice *vdev, VirtIOBlkConf *blk,
                                   VirtIOBlockDataPlane **dataplane,
@@ -197,6 +242,11 @@ void virtio_blk_data_plane_create(VirtIODevice *vdev, VirtIOBlkConf *blk,
     s->ctx = iothread_get_aio_context(s->iothread);
     s->bh = aio_bh_new(s->ctx, notify_guest_bh, s);
 
+    s->vring = g_new0(Vring, blk->num_queues);
+    s->guest_notifier = g_new(EventNotifier *, blk->num_queues);
+    s->host_notifier = g_new(VirtIOBlockNotifier, blk->num_queues);
+    s->host_notifier_bh = aio_bh_new(s->ctx, process_notify, s);
+
     error_setg(&s->blocker, "block device is in use by data plane");
     bdrv_op_block_all(blk->conf.bs, s->blocker);
 
@@ -217,16 +267,83 @@ void virtio_blk_data_plane_destroy(VirtIOBlockDataPlane *s)
     error_free(s->blocker);
     object_unref(OBJECT(s->iothread));
     qemu_bh_delete(s->bh);
+    qemu_bh_delete(s->host_notifier_bh);
+    g_free(s->vring);
+    g_free(s->guest_notifier);
+    g_free(s->host_notifier);
     g_free(s);
 }
 
+static int pre_start_vq(VirtIOBlockDataPlane *s, BusState *qbus,
+                        VirtioBusClass *k)
+{
+    int i;
+    int num = s->blk->num_queues;
+    VirtQueue *vq[num];
+
+    for (i = 0; i < num; i++) {
+        vq[i] = virtio_get_queue(s->vdev, i);
+        if (!vring_setup(&s->vring[i], s->vdev, i)) {
+            return -1;
+        }
+    }
+
+    /* Set up guest notifier (irq) */
+    if (k->set_guest_notifiers(qbus->parent, num, true) != 0) {
+        fprintf(stderr, "virtio-blk failed to set guest notifier, "
+                "ensure -enable-kvm is set\n");
+        exit(1);
+    }
+
+    for (i = 0; i < num; i++)
+        s->guest_notifier[i] = virtio_queue_get_guest_notifier(vq[i]);
+    s->pending_guest_notifier = 0;
+
+    /* Set up virtqueue notify */
+    for (i = 0; i < num; i++) {
+        if (k->set_host_notifier(qbus->parent, i, true) != 0) {
+            fprintf(stderr, "virtio-blk failed to set host notifier\n");
+            exit(1);
+        }
+        s->host_notifier[i].notifier = *virtio_queue_get_host_notifier(vq[i]);
+        s->host_notifier[i].s = s;
+    }
+    s->pending_host_notifier = 0;
+
+    return 0;
+}
+
+static void post_start_vq(VirtIOBlockDataPlane *s)
+{
+    int i;
+    int num = s->blk->num_queues;
+
+    for (i = 0; i < num; i++) {
+        VirtQueue *vq;
+        vq = virtio_get_queue(s->vdev, i);
+
+        /* Kick right away to begin processing requests already in vring */
+        event_notifier_set(virtio_queue_get_host_notifier(vq));
+    }
+
+    if (s->raw_format) {
+        bdrv_set_bypass_co(s->blk->conf.bs, true);
+    }
+
+    /* Get this show started by hooking up our callbacks */
+    aio_context_acquire(s->ctx);
+    for (i = 0; i < num; i++)
+        aio_set_event_notifier(s->ctx, &s->host_notifier[i].notifier,
+                               handle_notify);
+    aio_context_release(s->ctx);
+}
+
 /* Context: QEMU global mutex held */
 void virtio_blk_data_plane_start(VirtIOBlockDataPlane *s)
 {
     BusState *qbus = BUS(qdev_get_parent_bus(DEVICE(s->vdev)));
     VirtioBusClass *k = VIRTIO_BUS_GET_CLASS(qbus);
     VirtIOBlock *vblk = VIRTIO_BLK(s->vdev);
-    VirtQueue *vq;
 
     if (s->started) {
         return;
@@ -238,51 +355,24 @@ void virtio_blk_data_plane_start(VirtIOBlockDataPlane *s)
 
     s->starting = true;
 
-    vq = virtio_get_queue(s->vdev, 0);
-    if (!vring_setup(&s->vring, s->vdev, 0)) {
-        s->starting = false;
-        return;
-    }
-
     vblk->obj_pool = &s->req_pool;
     obj_pool_init(vblk->obj_pool, s->reqs, s->free_reqs,
                   sizeof(VirtIOBlockReq), REQ_POOL_SZ);
 
-    /* Set up guest notifier (irq) */
-    if (k->set_guest_notifiers(qbus->parent, 1, true) != 0) {
-        fprintf(stderr, "virtio-blk failed to set guest notifier, "
-                "ensure -enable-kvm is set\n");
-        exit(1);
-    }
-    s->guest_notifier = virtio_queue_get_guest_notifier(vq);
-
-    /* Set up virtqueue notify */
-    if (k->set_host_notifier(qbus->parent, 0, true) != 0) {
-        fprintf(stderr, "virtio-blk failed to set host notifier\n");
-        exit(1);
-    }
-    s->host_notifier = *virtio_queue_get_host_notifier(vq);
-
     s->saved_complete_request = vblk->complete_request;
     vblk->complete_request = complete_request_vring;
 
+    if (pre_start_vq(s, qbus, k)) {
+       s->starting = false;
+       return;
+     }
+
     s->starting = false;
     s->started = true;
     trace_virtio_blk_data_plane_start(s);
 
     bdrv_set_aio_context(s->blk->conf.bs, s->ctx);
-
-    /* Kick right away to begin processing requests already in vring */
-    event_notifier_set(virtio_queue_get_host_notifier(vq));
-
-    if (s->raw_format) {
-        bdrv_set_bypass_co(s->ctx, true);
-    }
-
-    /* Get this show started by hooking up our callbacks */
-    aio_context_acquire(s->ctx);
-    aio_set_event_notifier(s->ctx, &s->host_notifier, handle_notify);
-    aio_context_release(s->ctx);
+    post_start_vq(s);
 }
 
 /* Context: QEMU global mutex held */
@@ -291,6 +381,8 @@ void virtio_blk_data_plane_stop(VirtIOBlockDataPlane *s)
     BusState *qbus = BUS(qdev_get_parent_bus(DEVICE(s->vdev)));
     VirtioBusClass *k = VIRTIO_BUS_GET_CLASS(qbus);
     VirtIOBlock *vblk = VIRTIO_BLK(s->vdev);
+    int i;
+    int num = s->blk->num_queues;
     if (!s->started || s->stopping) {
         return;
     }
@@ -301,7 +393,8 @@ void virtio_blk_data_plane_stop(VirtIOBlockDataPlane *s)
     aio_context_acquire(s->ctx);
 
     /* Stop notifications for new requests from guest */
-    aio_set_event_notifier(s->ctx, &s->host_notifier, NULL);
+    for (i = 0; i < num; i++)
+        aio_set_event_notifier(s->ctx, &s->host_notifier[i].notifier, NULL);
 
     /* Drain and switch bs back to the QEMU main loop */
     bdrv_set_aio_context(s->blk->conf.bs, qemu_get_aio_context());
@@ -311,17 +404,19 @@ void virtio_blk_data_plane_stop(VirtIOBlockDataPlane *s)
     vblk->obj_pool = NULL;
 
     if (s->raw_format) {
-        bdrv_set_bypass_co(s->ctx, false);
+        bdrv_set_bypass_co(s->blk->conf.bs, false);
     }
     /* Sync vring state back to virtqueue so that non-dataplane request
      * processing can continue when we disable the host notifier below.
      */
-    vring_teardown(&s->vring, s->vdev, 0);
+    for (i = 0; i < num; i++)
+        vring_teardown(&s->vring[i], s->vdev, 0);
 
-    k->set_host_notifier(qbus->parent, 0, false);
+    for (i = 0; i < num; i++)
+        k->set_host_notifier(qbus->parent, i, false);
 
     /* Clean up guest notifier (irq) */
-    k->set_guest_notifiers(qbus->parent, 1, false);
+    k->set_guest_notifiers(qbus->parent, num, false);
 
     s->started = false;
     s->stopping = false;
-- 
1.7.9.5