[Qemu-devel] [PATCH 1/3] block: Support streaming to an intermediate layer

[Alberto Garcia]

This adds a new 'top' parameter to stream_start(), that specifies the
block device where the data will be written. The image is changed to
read-write mode during the streaming and back to read-only afterwards.

This also unblocks the stream operation in backing files.

Signed-off-by: Alberto Garcia <address@hidden>
---
 block.c                   |  4 +++-
 block/stream.c            | 49 ++++++++++++++++++++++++++++++++++++-----------
 blockdev.c                |  2 +-
 include/block/block_int.h | 19 +++++++++---------
 4 files changed, 52 insertions(+), 22 deletions(-)

diff --git a/block.c b/block.c
index 210fd5f..6b68d90 100644
--- a/block.c
+++ b/block.c
@@ -1199,9 +1199,11 @@ void bdrv_set_backing_hd(BlockDriverState *bs, 
BlockDriverState *backing_hd)
             backing_hd->drv ? backing_hd->drv->format_name : "");
 
     bdrv_op_block_all(bs->backing_hd, bs->backing_blocker);
-    /* Otherwise we won't be able to commit due to check in bdrv_commit */
+    /* Otherwise we won't be able to commit or stream */
     bdrv_op_unblock(bs->backing_hd, BLOCK_OP_TYPE_COMMIT_TARGET,
                     bs->backing_blocker);
+    bdrv_op_unblock(bs->backing_hd, BLOCK_OP_TYPE_STREAM,
+                    bs->backing_blocker);
 out:
     bdrv_refresh_limits(bs, NULL);
 }
diff --git a/block/stream.c b/block/stream.c
index a628901..ecf5831 100644
--- a/block/stream.c
+++ b/block/stream.c
@@ -31,8 +31,10 @@ typedef struct StreamBlockJob {
     BlockJob common;
     RateLimit limit;
     BlockDriverState *base;
+    BlockDriverState *top;
     BlockdevOnError on_error;
     char *backing_file_str;
+    int top_flags;
 } StreamBlockJob;
 
 static int coroutine_fn stream_populate(BlockDriverState *bs,
@@ -99,8 +101,13 @@ static void stream_complete(BlockJob *job, void *opaque)
                 base_fmt = base->drv->format_name;
             }
         }
-        data->ret = bdrv_change_backing_file(job->bs, base_id, base_fmt);
-        close_unused_images(job->bs, base, base_id);
+        data->ret = bdrv_change_backing_file(s->top, base_id, base_fmt);
+        close_unused_images(s->top, base, base_id);
+    }
+
+    /* Reopen the image back in read-only mode if necessary */
+    if (s->top_flags != bdrv_get_flags(s->top)) {
+        bdrv_reopen(s->top, s->top_flags, NULL);
     }
 
     g_free(s->backing_file_str);
@@ -112,23 +119,23 @@ static void coroutine_fn stream_run(void *opaque)
 {
     StreamBlockJob *s = opaque;
     StreamCompleteData *data;
-    BlockDriverState *bs = s->common.bs;
+    BlockDriverState *bs = s->top;
     BlockDriverState *base = s->base;
-    int64_t sector_num, end;
+    int64_t sector_num = 0;
+    int64_t end = -1;
     int error = 0;
     int ret = 0;
     int n = 0;
     void *buf;
 
     if (!bs->backing_hd) {
-        block_job_completed(&s->common, 0);
-        return;
+        goto out;
     }
 
     s->common.len = bdrv_getlength(bs);
     if (s->common.len < 0) {
-        block_job_completed(&s->common, s->common.len);
-        return;
+        ret = s->common.len;
+        goto out;
     }
 
     end = s->common.len >> BDRV_SECTOR_BITS;
@@ -215,6 +222,7 @@ wait:
 
     qemu_vfree(buf);
 
+out:
     /* Modify backing chain and close BDSes in main loop */
     data = g_malloc(sizeof(*data));
     data->ret = ret;
@@ -239,13 +247,18 @@ static const BlockJobDriver stream_job_driver = {
     .set_speed     = stream_set_speed,
 };
 
-void stream_start(BlockDriverState *bs, BlockDriverState *base,
-                  const char *backing_file_str, int64_t speed,
-                  BlockdevOnError on_error,
+void stream_start(BlockDriverState *bs, BlockDriverState *top,
+                  BlockDriverState *base, const char *backing_file_str,
+                  int64_t speed, BlockdevOnError on_error,
                   BlockCompletionFunc *cb,
                   void *opaque, Error **errp)
 {
     StreamBlockJob *s;
+    int orig_top_flags;
+
+    if (!top) {
+        top = bs;
+    }
 
     if ((on_error == BLOCKDEV_ON_ERROR_STOP ||
          on_error == BLOCKDEV_ON_ERROR_ENOSPC) &&
@@ -254,14 +267,28 @@ void stream_start(BlockDriverState *bs, BlockDriverState 
*base,
         return;
     }
 
+    /* Make sure that the top image in opened in read-write mode */
+    orig_top_flags = bdrv_get_flags(top);
+    if (!(orig_top_flags & BDRV_O_RDWR)) {
+        Error *local_err = NULL;
+        bdrv_reopen(top, orig_top_flags | BDRV_O_RDWR, &local_err);
+        if (local_err != NULL) {
+            error_propagate(errp, local_err);
+            return;
+        }
+    }
+
     s = block_job_create(&stream_job_driver, bs, speed, cb, opaque, errp);
     if (!s) {
         return;
     }
 
+    s->top = top;
     s->base = base;
     s->backing_file_str = g_strdup(backing_file_str);
 
+    s->top_flags = orig_top_flags;
+
     s->on_error = on_error;
     s->common.co = qemu_coroutine_create(stream_run);
     trace_stream_start(bs, base, s, s->common.co, opaque);
diff --git a/blockdev.c b/blockdev.c
index 7d34960..06628ca 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -2148,7 +2148,7 @@ void qmp_block_stream(const char *device,
     /* backing_file string overrides base bs filename */
     base_name = has_backing_file ? backing_file : base_name;
 
-    stream_start(bs, base_bs, base_name, has_speed ? speed : 0,
+    stream_start(bs, NULL, base_bs, base_name, has_speed ? speed : 0,
                  on_error, block_job_cb, bs, &local_err);
     if (local_err) {
         error_propagate(errp, local_err);
diff --git a/include/block/block_int.h b/include/block/block_int.h
index 7ad1950..959d19c 100644
--- a/include/block/block_int.h
+++ b/include/block/block_int.h
@@ -498,10 +498,11 @@ int is_windows_drive(const char *filename);
 
 /**
  * stream_start:
- * @bs: Block device to operate on.
+ * @bs: Active block device.
+ * @top: Block device to copy the data to, defaults to @bs if %NULL.
  * @base: Block device that will become the new base, or %NULL to
- * flatten the whole backing file chain onto @bs.
- * @base_id: The file name that will be written to @bs as the new
+ * flatten the whole backing file chain onto @top.
+ * @base_id: The file name that will be written to @top as the new
  * backing file if the job completes.  Ignored if @base is %NULL.
  * @speed: The maximum speed, in bytes per second, or 0 for unlimited.
  * @on_error: The action to take upon error.
@@ -510,14 +511,14 @@ int is_windows_drive(const char *filename);
  * @errp: Error object.
  *
  * Start a streaming operation on @bs.  Clusters that are unallocated
- * in @bs, but allocated in any image between @base and @bs (both
- * exclusive) will be written to @bs.  At the end of a successful
- * streaming job, the backing file of @bs will be changed to
+ * in @top, but allocated in any image between @base and @top (both
+ * exclusive) will be written to @top.  At the end of a successful
+ * streaming job, the backing file of @top will be changed to
  * @base_id in the written image and to @base in the live BlockDriverState.
  */
-void stream_start(BlockDriverState *bs, BlockDriverState *base,
-                  const char *base_id, int64_t speed, BlockdevOnError on_error,
-                  BlockCompletionFunc *cb,
+void stream_start(BlockDriverState *bs, BlockDriverState *top,
+                  BlockDriverState *base, const char *base_id, int64_t speed,
+                  BlockdevOnError on_error, BlockCompletionFunc *cb,
                   void *opaque, Error **errp);
 
 /**
-- 
2.1.4