[Qemu-devel] [RFC][PATCH 11/12] qcow2: Convert qcow2 to use coroutines for async I/O

[Stefan Hajnoczi]

Converting qcow2 to use coroutines is fairly simple since most of qcow2
is synchronous.  The synchronous I/O functions likes bdrv_pread() now
transparently work when called from a coroutine, so all the synchronous
code just works.

The explicitly asynchronous code is adjusted to repeatedly call
qcow2_aio_read_cb() or qcow2_aio_write_cb() until the request completes.
At that point the coroutine will return from its entry function and its
resources are freed.

The bdrv_aio_readv() and bdrv_aio_writev() user callback is now invoked
from a BH.  This is necessary since the user callback code does not
expect to be executed from a coroutine.

This conversion is not completely correct because the safety the
synchronous code does not carry over to the coroutine version.
Previously, a synchronous code path could assume that it will never be
interleaved with another request executing.  This is no longer true
because bdrv_pread() and bdrv_pwrite() cause the coroutine to yield and
other requests can be processed during that time.

The solution is to carefully introduce checks so that pending requests
do not step on each other's toes.  That is left for a future patch...

Signed-off-by: Stefan Hajnoczi <address@hidden>
---
 block/qcow2.c |  160 ++++++++++++++++++++++++++++++---------------------------
 1 files changed, 85 insertions(+), 75 deletions(-)

diff --git a/block/qcow2.c b/block/qcow2.c
index b6b094c..4b33ef3 100644
--- a/block/qcow2.c
+++ b/block/qcow2.c
@@ -361,19 +361,20 @@ typedef struct QCowAIOCB {
     uint64_t bytes_done;
     uint64_t cluster_offset;
     uint8_t *cluster_data;
-    BlockDriverAIOCB *hd_aiocb;
     QEMUIOVector hd_qiov;
     QEMUBH *bh;
     QCowL2Meta l2meta;
     QLIST_ENTRY(QCowAIOCB) next_depend;
+    Coroutine *coroutine;
+    int ret; /* return code for user callback */
 } QCowAIOCB;
 
 static void qcow2_aio_cancel(BlockDriverAIOCB *blockacb)
 {
     QCowAIOCB *acb = container_of(blockacb, QCowAIOCB, common);
-    if (acb->hd_aiocb)
-        bdrv_aio_cancel(acb->hd_aiocb);
     qemu_aio_release(acb);
+    /* XXX This function looks broken, we could be in the middle of a request
+     * and releasing the acb is not a good idea */
 }
 
 static AIOPool qcow2_aio_pool = {
@@ -381,13 +382,14 @@ static AIOPool qcow2_aio_pool = {
     .cancel             = qcow2_aio_cancel,
 };
 
-static void qcow2_aio_read_cb(void *opaque, int ret);
-static void qcow2_aio_read_bh(void *opaque)
+static void qcow2_aio_bh(void *opaque)
 {
     QCowAIOCB *acb = opaque;
     qemu_bh_delete(acb->bh);
     acb->bh = NULL;
-    qcow2_aio_read_cb(opaque, 0);
+    acb->common.cb(acb->common.opaque, acb->ret);
+    qemu_iovec_destroy(&acb->hd_qiov);
+    qemu_aio_release(acb);
 }
 
 static int qcow2_schedule_bh(QEMUBHFunc *cb, QCowAIOCB *acb)
@@ -404,14 +406,13 @@ static int qcow2_schedule_bh(QEMUBHFunc *cb, QCowAIOCB 
*acb)
     return 0;
 }
 
-static void qcow2_aio_read_cb(void *opaque, int ret)
+static int coroutine_fn qcow2_aio_read_cb(void *opaque, int ret)
 {
     QCowAIOCB *acb = opaque;
     BlockDriverState *bs = acb->common.bs;
     BDRVQcowState *s = bs->opaque;
     int index_in_cluster, n1;
 
-    acb->hd_aiocb = NULL;
     if (ret < 0)
         goto done;
 
@@ -469,22 +470,13 @@ static void qcow2_aio_read_cb(void *opaque, int ret)
                 acb->sector_num, acb->cur_nr_sectors);
             if (n1 > 0) {
                 BLKDBG_EVENT(bs->file, BLKDBG_READ_BACKING_AIO);
-                acb->hd_aiocb = bdrv_aio_readv(bs->backing_hd, acb->sector_num,
-                                    &acb->hd_qiov, acb->cur_nr_sectors,
-                                   qcow2_aio_read_cb, acb);
-                if (acb->hd_aiocb == NULL)
-                    goto done;
-            } else {
-                ret = qcow2_schedule_bh(qcow2_aio_read_bh, acb);
-                if (ret < 0)
+                ret = bdrv_co_readv(bs->backing_hd, acb->sector_num, 
&acb->hd_qiov, acb->cur_nr_sectors);
+                if (ret < 0) {
                     goto done;
+                }
             }
         } else {
-            /* Note: in this case, no need to wait */
             qemu_iovec_memset(&acb->hd_qiov, 0, 512 * acb->cur_nr_sectors);
-            ret = qcow2_schedule_bh(qcow2_aio_read_bh, acb);
-            if (ret < 0)
-                goto done;
         }
     } else if (acb->cluster_offset & QCOW_OFLAG_COMPRESSED) {
         /* add AIO support for compressed blocks ? */
@@ -494,10 +486,6 @@ static void qcow2_aio_read_cb(void *opaque, int ret)
         qemu_iovec_from_buffer(&acb->hd_qiov,
             s->cluster_cache + index_in_cluster * 512,
             512 * acb->cur_nr_sectors);
-
-        ret = qcow2_schedule_bh(qcow2_aio_read_bh, acb);
-        if (ret < 0)
-            goto done;
     } else {
         if ((acb->cluster_offset & 511) != 0) {
             ret = -EIO;
@@ -522,34 +510,50 @@ static void qcow2_aio_read_cb(void *opaque, int ret)
         }
 
         BLKDBG_EVENT(bs->file, BLKDBG_READ_AIO);
-        acb->hd_aiocb = bdrv_aio_readv(bs->file,
+        ret = bdrv_co_readv(bs->file,
                             (acb->cluster_offset >> 9) + index_in_cluster,
-                            &acb->hd_qiov, acb->cur_nr_sectors,
-                            qcow2_aio_read_cb, acb);
-        if (acb->hd_aiocb == NULL) {
-            ret = -EIO;
+                            &acb->hd_qiov, acb->cur_nr_sectors);
+        if (ret < 0) {
             goto done;
         }
     }
 
-    return;
+    return 1;
 done:
-    acb->common.cb(acb->common.opaque, ret);
-    qemu_iovec_destroy(&acb->hd_qiov);
-    qemu_aio_release(acb);
+    acb->ret = ret;
+    qcow2_schedule_bh(qcow2_aio_bh, acb);
+    return 0;
+}
+
+static void * coroutine_fn qcow2_co_read(void *opaque)
+{
+    QCowAIOCB *acb = opaque;
+
+    while (qcow2_aio_read_cb(acb, 0)) {
+    }
+    return NULL;
+}
+
+static int coroutine_fn qcow2_aio_write_cb(void *opaque, int ret);
+static void * coroutine_fn qcow2_co_write(void *opaque)
+{
+    QCowAIOCB *acb = opaque;
+
+    while (qcow2_aio_write_cb(acb, 0)) {
+    }
+    return NULL;
 }
 
-static QCowAIOCB *qcow2_aio_setup(BlockDriverState *bs, int64_t sector_num,
-                                  QEMUIOVector *qiov, int nb_sectors,
-                                  BlockDriverCompletionFunc *cb,
-                                  void *opaque, int is_write)
+static BlockDriverAIOCB *qcow2_aio_setup(BlockDriverState *bs,
+        int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
+        BlockDriverCompletionFunc *cb, void *opaque, int is_write)
 {
     QCowAIOCB *acb;
+    Coroutine *coroutine;
 
     acb = qemu_aio_get(&qcow2_aio_pool, bs, cb, opaque);
     if (!acb)
         return NULL;
-    acb->hd_aiocb = NULL;
     acb->sector_num = sector_num;
     acb->qiov = qiov;
 
@@ -561,7 +565,12 @@ static QCowAIOCB *qcow2_aio_setup(BlockDriverState *bs, 
int64_t sector_num,
     acb->cluster_offset = 0;
     acb->l2meta.nb_clusters = 0;
     QLIST_INIT(&acb->l2meta.dependent_requests);
-    return acb;
+
+    coroutine = qemu_coroutine_create(is_write ? qcow2_co_write
+                                               : qcow2_co_read);
+    acb->coroutine = coroutine;
+    qemu_coroutine_enter(coroutine, acb);
+    return &acb->common;
 }
 
 static BlockDriverAIOCB *qcow2_aio_readv(BlockDriverState *bs,
@@ -570,38 +579,48 @@ static BlockDriverAIOCB *qcow2_aio_readv(BlockDriverState 
*bs,
                                          BlockDriverCompletionFunc *cb,
                                          void *opaque)
 {
-    QCowAIOCB *acb;
-
-    acb = qcow2_aio_setup(bs, sector_num, qiov, nb_sectors, cb, opaque, 0);
-    if (!acb)
-        return NULL;
-
-    qcow2_aio_read_cb(acb, 0);
-    return &acb->common;
+    return qcow2_aio_setup(bs, sector_num, qiov, nb_sectors, cb, opaque, 0);
 }
 
-static void qcow2_aio_write_cb(void *opaque, int ret);
-
-static void run_dependent_requests(QCowL2Meta *m)
+static void qcow2_co_run_dependent_requests(void *opaque)
 {
+    QCowAIOCB *acb = opaque;
     QCowAIOCB *req;
     QCowAIOCB *next;
 
+    qemu_bh_delete(acb->bh);
+    acb->bh = NULL;
+
+    /* Restart all dependent requests */
+    QLIST_FOREACH_SAFE(req, &acb->l2meta.dependent_requests, next_depend, 
next) {
+        qemu_coroutine_enter(req->coroutine, NULL);
+    }
+
+    /* Reenter the original request */
+    qemu_coroutine_enter(acb->coroutine, NULL);
+}
+
+static void run_dependent_requests(QCowL2Meta *m)
+{
     /* Take the request off the list of running requests */
     if (m->nb_clusters != 0) {
         QLIST_REMOVE(m, next_in_flight);
     }
 
-    /* Restart all dependent requests */
-    QLIST_FOREACH_SAFE(req, &m->dependent_requests, next_depend, next) {
-        qcow2_aio_write_cb(req, 0);
+    if (!QLIST_EMPTY(&m->dependent_requests)) {
+        /* TODO This is a hack to get at the acb, may not be correct if called
+         * with a QCowL2Meta that is not part of a QCowAIOCB.
+         */
+        QCowAIOCB *acb = container_of(m, QCowAIOCB, l2meta);
+        qcow2_schedule_bh(qcow2_co_run_dependent_requests, acb);
+        qemu_coroutine_yield(NULL);
     }
 
     /* Empty the list for the next part of the request */
     QLIST_INIT(&m->dependent_requests);
 }
 
-static void qcow2_aio_write_cb(void *opaque, int ret)
+static int coroutine_fn qcow2_aio_write_cb(void *opaque, int ret)
 {
     QCowAIOCB *acb = opaque;
     BlockDriverState *bs = acb->common.bs;
@@ -609,8 +628,6 @@ static void qcow2_aio_write_cb(void *opaque, int ret)
     int index_in_cluster;
     int n_end;
 
-    acb->hd_aiocb = NULL;
-
     if (ret >= 0) {
         ret = qcow2_alloc_cluster_link_l2(bs, &acb->l2meta);
     }
@@ -648,7 +665,8 @@ static void qcow2_aio_write_cb(void *opaque, int ret)
     if (acb->l2meta.nb_clusters == 0 && acb->l2meta.depends_on != NULL) {
         QLIST_INSERT_HEAD(&acb->l2meta.depends_on->dependent_requests,
             acb, next_depend);
-        return;
+        qemu_coroutine_yield(NULL);
+        return 1;
     }
 
     assert((acb->cluster_offset & 511) == 0);
@@ -675,25 +693,22 @@ static void qcow2_aio_write_cb(void *opaque, int ret)
     }
 
     BLKDBG_EVENT(bs->file, BLKDBG_WRITE_AIO);
-    acb->hd_aiocb = bdrv_aio_writev(bs->file,
-                                    (acb->cluster_offset >> 9) + 
index_in_cluster,
-                                    &acb->hd_qiov, acb->cur_nr_sectors,
-                                    qcow2_aio_write_cb, acb);
-    if (acb->hd_aiocb == NULL) {
-        ret = -EIO;
+    ret = bdrv_co_writev(bs->file,
+                         (acb->cluster_offset >> 9) + index_in_cluster,
+                         &acb->hd_qiov, acb->cur_nr_sectors);
+    if (ret < 0) {
         goto fail;
     }
-
-    return;
+    return 1;
 
 fail:
     if (acb->l2meta.nb_clusters != 0) {
         QLIST_REMOVE(&acb->l2meta, next_in_flight);
     }
 done:
-    acb->common.cb(acb->common.opaque, ret);
-    qemu_iovec_destroy(&acb->hd_qiov);
-    qemu_aio_release(acb);
+    acb->ret = ret;
+    qcow2_schedule_bh(qcow2_aio_bh, acb);
+    return 0;
 }
 
 static BlockDriverAIOCB *qcow2_aio_writev(BlockDriverState *bs,
@@ -703,16 +718,10 @@ static BlockDriverAIOCB 
*qcow2_aio_writev(BlockDriverState *bs,
                                           void *opaque)
 {
     BDRVQcowState *s = bs->opaque;
-    QCowAIOCB *acb;
 
     s->cluster_cache_offset = -1; /* disable compressed cache */
 
-    acb = qcow2_aio_setup(bs, sector_num, qiov, nb_sectors, cb, opaque, 1);
-    if (!acb)
-        return NULL;
-
-    qcow2_aio_write_cb(acb, 0);
-    return &acb->common;
+    return qcow2_aio_setup(bs, sector_num, qiov, nb_sectors, cb, opaque, 1);
 }
 
 static void qcow2_close(BlockDriverState *bs)
@@ -824,6 +833,7 @@ static int qcow2_change_backing_file(BlockDriverState *bs,
     return qcow2_update_ext_header(bs, backing_file, backing_fmt);
 }
 
+/* TODO did we break this for coroutines? */
 static int preallocate(BlockDriverState *bs)
 {
     uint64_t nb_sectors;
-- 
1.7.2.3