[Qemu-devel] [PATCH v5 00/15] block: generic image streaming

[Qemu-devel] [PATCH v5 00/15] block: generic image streaming

[Stefan Hajnoczi]

Note: This is a resend of v5 because the emails I sent earlier today
disappeared.

This series adds the 'block_stream' command which copies the contents of a
backing file into the image file while the VM is running.  The series builds on
the zero detection features which I sent out before Christmas. I suggest
grabbing my git tree to try it out without merging this dependency:

https://github.com/stefanha/qemu/tree/image-streaming-api

The image streaming HMP/QMP commands are documented in the patch and also
described here:

http://wiki.qemu.org/Features/LiveBlockMigration/ImageStreamingAPI

The basic idea is to execute 'block_stream virtio0' while the guest is running.
Progress can be monitored using 'info block-jobs'.  When the streaming
operation completes it raises a QMP event.

Note: The last patch includes includes a Python test script called
test-stream.py, I do not propose to merge it.  When run in a QEMU source tree
it performs basic image streaming QMP tests.

v5:
 * Handle block_job_create() failure [Luiz]
 * Mark BLOCK_JOB_COMPLETED error field optional [Luiz]
 * Mark block_stream base parameter optional [Luiz]
 * Check bdrv_getlength() failure and don't call twice [Kevin]
 * Rename "id" to "backing_file" in bdrv_find_backing_image() [Kevin]
 * Rename BaseIdNotFound qerror to BaseNotFound [Kevin]
 * Document BaseNotFound qerror from block_stream
 * Document that qemu_co_sleep_ns() needs main loop [Kevin]
 * Make bdrv_co_is_allocated_base() private to block/stream.c [Kevin]
 * Clean up commit messages

v4:
 * Drop SQMP/EQMP docs from qmp-commands.hx [Luiz]
 * Follow QAPI doc conventions [Luiz]
 * Document QMP events in QMP/qmp-events.txt [Luiz]
 * Protect against hotplug, resize, eject, etc [Kevin]
 * Move block job functions from header to block.c [Kevin]
 * Return error from bdrg_change_backing_file() [Kevin]
 * Merge Marcelo's block_stream base partial streaming series [Marcelo]

Marcelo Tosatti (4):
  block: add bdrv_find_backing_image
  add QERR_BASE_NOT_FOUND
  block: add support for partial streaming
  docs: describe live block operations

Stefan Hajnoczi (11):
  coroutine: add co_sleep_ns() coroutine sleep function
  block: check bdrv_in_use() before blockdev operations
  block: add BlockJob interface for long-running operations
  block: add image streaming block job
  block: rate-limit streaming operations
  qmp: add block_stream command
  qmp: add block_job_set_speed command
  qmp: add block_job_cancel command
  qmp: add query-block-jobs
  blockdev: make image streaming safe across hotplug
  test: add image streaming test cases

 Makefile.objs           |    2 +
 QMP/qmp-events.txt      |   53 ++++++++++
 block.c                 |   70 +++++++++++++
 block.h                 |    2 +
 block/stream.c          |  260 +++++++++++++++++++++++++++++++++++++++++++++++
 block_int.h             |   44 ++++++++
 blockdev.c              |  199 +++++++++++++++++++++++++++++++++++-
 docs/live-block-ops.txt |   58 +++++++++++
 hmp-commands.hx         |   41 ++++++++
 hmp.c                   |   68 ++++++++++++
 hmp.h                   |    4 +
 monitor.c               |   13 +++
 monitor.h               |    2 +
 qapi-schema.json        |  116 +++++++++++++++++++++
 qemu-coroutine-sleep.c  |   38 +++++++
 qemu-coroutine.h        |    9 ++
 qerror.c                |    8 ++
 qerror.h                |    6 +
 qmp-commands.hx         |   24 +++++
 test-stream.py          |  208 +++++++++++++++++++++++++++++++++++++
 trace-events            |    9 ++
 21 files changed, 1233 insertions(+), 1 deletions(-)
 create mode 100644 block/stream.c
 create mode 100644 docs/live-block-ops.txt
 create mode 100644 qemu-coroutine-sleep.c
 create mode 100644 test-stream.py

-- 
1.7.7.3

[Qemu-devel] [PATCH v5 01/15] coroutine: add co_sleep_ns() coroutine sleep function

[Stefan Hajnoczi]

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 Makefile.objs          |    1 +
 qemu-coroutine-sleep.c |   38 ++++++++++++++++++++++++++++++++++++++
 qemu-coroutine.h       |    9 +++++++++
 3 files changed, 48 insertions(+), 0 deletions(-)
 create mode 100644 qemu-coroutine-sleep.c

diff --git a/Makefile.objs b/Makefile.objs
index 4f6d26c..f4f52e0 100644
--- a/Makefile.objs
+++ b/Makefile.objs
@@ -13,6 +13,7 @@ oslib-obj-$(CONFIG_POSIX) += oslib-posix.o qemu-thread-posix.o
 #######################################################################
 # coroutines
 coroutine-obj-y = qemu-coroutine.o qemu-coroutine-lock.o qemu-coroutine-io.o
+coroutine-obj-y += qemu-coroutine-sleep.o
 ifeq ($(CONFIG_UCONTEXT_COROUTINE),y)
 coroutine-obj-$(CONFIG_POSIX) += coroutine-ucontext.o
 else
diff --git a/qemu-coroutine-sleep.c b/qemu-coroutine-sleep.c
new file mode 100644
index 0000000..fd65274
--- /dev/null
+++ b/qemu-coroutine-sleep.c
@@ -0,0 +1,38 @@
+/*
+ * QEMU coroutine sleep
+ *
+ * Copyright IBM, Corp. 2011
+ *
+ * Authors:
+ *  Stefan Hajnoczi    <stefanha@linux.vnet.ibm.com>
+ *
+ * This work is licensed under the terms of the GNU LGPL, version 2 or later.
+ * See the COPYING.LIB file in the top-level directory.
+ *
+ */
+
+#include "qemu-coroutine.h"
+#include "qemu-timer.h"
+
+typedef struct CoSleepCB {
+    QEMUTimer *ts;
+    Coroutine *co;
+} CoSleepCB;
+
+static void co_sleep_cb(void *opaque)
+{
+    CoSleepCB *sleep_cb = opaque;
+
+    qemu_free_timer(sleep_cb->ts);
+    qemu_coroutine_enter(sleep_cb->co, NULL);
+}
+
+void coroutine_fn co_sleep_ns(QEMUClock *clock, int64_t ns)
+{
+    CoSleepCB sleep_cb = {
+        .co = qemu_coroutine_self(),
+    };
+    sleep_cb.ts = qemu_new_timer(clock, SCALE_NS, co_sleep_cb, &sleep_cb);
+    qemu_mod_timer(sleep_cb.ts, qemu_get_clock_ns(clock) + ns);
+    qemu_coroutine_yield();
+}
diff --git a/qemu-coroutine.h b/qemu-coroutine.h
index 8a55fe1..34c15d4 100644
--- a/qemu-coroutine.h
+++ b/qemu-coroutine.h
@@ -17,6 +17,7 @@
 
 #include <stdbool.h>
 #include "qemu-queue.h"
+#include "qemu-timer.h"
 
 /**
  * Coroutines are a mechanism for stack switching and can be used for
@@ -199,4 +200,12 @@ void qemu_co_rwlock_wrlock(CoRwlock *lock);
  */
 void qemu_co_rwlock_unlock(CoRwlock *lock);
 
+/**
+ * Yield the coroutine for a given duration
+ *
+ * Note this function uses timers and hence only works when a main loop is in
+ * use.  See main-loop.h and do not use from qemu-tool programs.
+ */
+void coroutine_fn co_sleep_ns(QEMUClock *clock, int64_t ns);
+
 #endif /* QEMU_COROUTINE_H */
-- 
1.7.7.3

Re: [Qemu-devel] [PATCH v5 01/15] coroutine: add co_sleep_ns() coroutine sleep function

[Kevin Wolf]

Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
> ---
>  Makefile.objs          |    1 +
>  qemu-coroutine-sleep.c |   38 ++++++++++++++++++++++++++++++++++++++
>  qemu-coroutine.h       |    9 +++++++++
>  3 files changed, 48 insertions(+), 0 deletions(-)
>  create mode 100644 qemu-coroutine-sleep.c
> 
> diff --git a/Makefile.objs b/Makefile.objs
> index 4f6d26c..f4f52e0 100644
> --- a/Makefile.objs
> +++ b/Makefile.objs
> @@ -13,6 +13,7 @@ oslib-obj-$(CONFIG_POSIX) += oslib-posix.o qemu-thread-posix.o
>  #######################################################################
>  # coroutines
>  coroutine-obj-y = qemu-coroutine.o qemu-coroutine-lock.o qemu-coroutine-io.o
> +coroutine-obj-y += qemu-coroutine-sleep.o
>  ifeq ($(CONFIG_UCONTEXT_COROUTINE),y)
>  coroutine-obj-$(CONFIG_POSIX) += coroutine-ucontext.o
>  else
> diff --git a/qemu-coroutine-sleep.c b/qemu-coroutine-sleep.c
> new file mode 100644
> index 0000000..fd65274
> --- /dev/null
> +++ b/qemu-coroutine-sleep.c
> @@ -0,0 +1,38 @@
> +/*
> + * QEMU coroutine sleep
> + *
> + * Copyright IBM, Corp. 2011
> + *
> + * Authors:
> + *  Stefan Hajnoczi    <stefanha@linux.vnet.ibm.com>
> + *
> + * This work is licensed under the terms of the GNU LGPL, version 2 or later.
> + * See the COPYING.LIB file in the top-level directory.
> + *
> + */
> +
> +#include "qemu-coroutine.h"
> +#include "qemu-timer.h"
> +
> +typedef struct CoSleepCB {
> +    QEMUTimer *ts;
> +    Coroutine *co;
> +} CoSleepCB;
> +
> +static void co_sleep_cb(void *opaque)
> +{
> +    CoSleepCB *sleep_cb = opaque;
> +
> +    qemu_free_timer(sleep_cb->ts);

I think you need to call qemu_del_timer() first.

Kevin

Re: [Qemu-devel] [PATCH v5 01/15] coroutine: add co_sleep_ns() coroutine sleep function

[Stefan Hajnoczi]

On Tue, Jan 17, 2012 at 12:54 PM, Kevin Wolf <kwolf@redhat.com> wrote:
> Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
>> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
>> ---
>>  Makefile.objs          |    1 +
>>  qemu-coroutine-sleep.c |   38 ++++++++++++++++++++++++++++++++++++++
>>  qemu-coroutine.h       |    9 +++++++++
>>  3 files changed, 48 insertions(+), 0 deletions(-)
>>  create mode 100644 qemu-coroutine-sleep.c
>>
>> diff --git a/Makefile.objs b/Makefile.objs
>> index 4f6d26c..f4f52e0 100644
>> --- a/Makefile.objs
>> +++ b/Makefile.objs
>> @@ -13,6 +13,7 @@ oslib-obj-$(CONFIG_POSIX) += oslib-posix.o qemu-thread-posix.o
>>  #######################################################################
>>  # coroutines
>>  coroutine-obj-y = qemu-coroutine.o qemu-coroutine-lock.o qemu-coroutine-io.o
>> +coroutine-obj-y += qemu-coroutine-sleep.o
>>  ifeq ($(CONFIG_UCONTEXT_COROUTINE),y)
>>  coroutine-obj-$(CONFIG_POSIX) += coroutine-ucontext.o
>>  else
>> diff --git a/qemu-coroutine-sleep.c b/qemu-coroutine-sleep.c
>> new file mode 100644
>> index 0000000..fd65274
>> --- /dev/null
>> +++ b/qemu-coroutine-sleep.c
>> @@ -0,0 +1,38 @@
>> +/*
>> + * QEMU coroutine sleep
>> + *
>> + * Copyright IBM, Corp. 2011
>> + *
>> + * Authors:
>> + *  Stefan Hajnoczi    <stefanha@linux.vnet.ibm.com>
>> + *
>> + * This work is licensed under the terms of the GNU LGPL, version 2 or later.
>> + * See the COPYING.LIB file in the top-level directory.
>> + *
>> + */
>> +
>> +#include "qemu-coroutine.h"
>> +#include "qemu-timer.h"
>> +
>> +typedef struct CoSleepCB {
>> +    QEMUTimer *ts;
>> +    Coroutine *co;
>> +} CoSleepCB;
>> +
>> +static void co_sleep_cb(void *opaque)
>> +{
>> +    CoSleepCB *sleep_cb = opaque;
>> +
>> +    qemu_free_timer(sleep_cb->ts);
>
> I think you need to call qemu_del_timer() first.

I think we're okay with just qemu_free_timer().  qemu_run_timers()
removes the timer from the active_timers list before invoking its
callback.  qemu_del_timer() is not needed because it just searches the
active_timers list and removes the timer, if found.  We're no longer
on the active_timers list at the point when co_sleep_cb() is called,
so there's no need.

Stefan

Re: [Qemu-devel] [PATCH v5 01/15] coroutine: add co_sleep_ns() coroutine sleep function

[Kevin Wolf]

Am 17.01.2012 14:31, schrieb Stefan Hajnoczi:
> On Tue, Jan 17, 2012 at 12:54 PM, Kevin Wolf <kwolf@redhat.com> wrote:
>> Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
>>> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
>>> ---
>>>  Makefile.objs          |    1 +
>>>  qemu-coroutine-sleep.c |   38 ++++++++++++++++++++++++++++++++++++++
>>>  qemu-coroutine.h       |    9 +++++++++
>>>  3 files changed, 48 insertions(+), 0 deletions(-)
>>>  create mode 100644 qemu-coroutine-sleep.c
>>>
>>> diff --git a/Makefile.objs b/Makefile.objs
>>> index 4f6d26c..f4f52e0 100644
>>> --- a/Makefile.objs
>>> +++ b/Makefile.objs
>>> @@ -13,6 +13,7 @@ oslib-obj-$(CONFIG_POSIX) += oslib-posix.o qemu-thread-posix.o
>>>  #######################################################################
>>>  # coroutines
>>>  coroutine-obj-y = qemu-coroutine.o qemu-coroutine-lock.o qemu-coroutine-io.o
>>> +coroutine-obj-y += qemu-coroutine-sleep.o
>>>  ifeq ($(CONFIG_UCONTEXT_COROUTINE),y)
>>>  coroutine-obj-$(CONFIG_POSIX) += coroutine-ucontext.o
>>>  else
>>> diff --git a/qemu-coroutine-sleep.c b/qemu-coroutine-sleep.c
>>> new file mode 100644
>>> index 0000000..fd65274
>>> --- /dev/null
>>> +++ b/qemu-coroutine-sleep.c
>>> @@ -0,0 +1,38 @@
>>> +/*
>>> + * QEMU coroutine sleep
>>> + *
>>> + * Copyright IBM, Corp. 2011
>>> + *
>>> + * Authors:
>>> + *  Stefan Hajnoczi    <stefanha@linux.vnet.ibm.com>
>>> + *
>>> + * This work is licensed under the terms of the GNU LGPL, version 2 or later.
>>> + * See the COPYING.LIB file in the top-level directory.
>>> + *
>>> + */
>>> +
>>> +#include "qemu-coroutine.h"
>>> +#include "qemu-timer.h"
>>> +
>>> +typedef struct CoSleepCB {
>>> +    QEMUTimer *ts;
>>> +    Coroutine *co;
>>> +} CoSleepCB;
>>> +
>>> +static void co_sleep_cb(void *opaque)
>>> +{
>>> +    CoSleepCB *sleep_cb = opaque;
>>> +
>>> +    qemu_free_timer(sleep_cb->ts);
>>
>> I think you need to call qemu_del_timer() first.
> 
> I think we're okay with just qemu_free_timer().  qemu_run_timers()
> removes the timer from the active_timers list before invoking its
> callback.  qemu_del_timer() is not needed because it just searches the
> active_timers list and removes the timer, if found.  We're no longer
> on the active_timers list at the point when co_sleep_cb() is called,
> so there's no need.

Yes, seems you're right. qemu_del_timer() isn't a good name...

Kevin

[Qemu-devel] [PATCH v5 02/15] block: check bdrv_in_use() before blockdev operations

[Stefan Hajnoczi]

Long-running block operations like block migration and image streaming
must have continual access to their block device.  It is not safe to
perform operations like hotplug, eject, change, resize, commit, or
external snapshot while a long-running operation is in progress.

This patch adds the missing bdrv_in_use() checks so that block migration
and image streaming never have the rug pulled out from underneath them.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 block.c    |    4 ++++
 blockdev.c |   16 +++++++++++++++-
 2 files changed, 19 insertions(+), 1 deletions(-)

diff --git a/block.c b/block.c
index 967a583..daf92c2 100644
--- a/block.c
+++ b/block.c
@@ -1020,6 +1020,10 @@ int bdrv_commit(BlockDriverState *bs)
         return -EACCES;
     }
 
+    if (bdrv_in_use(bs) || bdrv_in_use(bs->backing_hd)) {
+        return -EBUSY;
+    }
+
     backing_drv = bs->backing_hd->drv;
     ro = bs->backing_hd->read_only;
     strncpy(filename, bs->backing_hd->filename, sizeof(filename));
diff --git a/blockdev.c b/blockdev.c
index c832782..6d78b36 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -592,12 +592,18 @@ void do_commit(Monitor *mon, const QDict *qdict)
     if (!strcmp(device, "all")) {
         bdrv_commit_all();
     } else {
+        int ret;
+
         bs = bdrv_find(device);
         if (!bs) {
             qerror_report(QERR_DEVICE_NOT_FOUND, device);
             return;
         }
-        bdrv_commit(bs);
+        ret = bdrv_commit(bs);
+        if (ret == -EBUSY) {
+            qerror_report(QERR_DEVICE_IN_USE, device);
+            return;
+        }
     }
 }
 
@@ -616,6 +622,10 @@ void qmp_blockdev_snapshot_sync(const char *device, const char *snapshot_file,
         error_set(errp, QERR_DEVICE_NOT_FOUND, device);
         return;
     }
+    if (bdrv_in_use(bs)) {
+        error_set(errp, QERR_DEVICE_IN_USE, device);
+        return;
+    }
 
     pstrcpy(old_filename, sizeof(old_filename), bs->filename);
 
@@ -667,6 +677,10 @@ void qmp_blockdev_snapshot_sync(const char *device, const char *snapshot_file,
 
 static int eject_device(Monitor *mon, BlockDriverState *bs, int force)
 {
+    if (bdrv_in_use(bs)) {
+        qerror_report(QERR_DEVICE_IN_USE, bdrv_get_device_name(bs));
+        return -1;
+    }
     if (!bdrv_dev_has_removable_media(bs)) {
         qerror_report(QERR_DEVICE_NOT_REMOVABLE, bdrv_get_device_name(bs));
         return -1;
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 03/15] block: add BlockJob interface for long-running operations

[Stefan Hajnoczi]

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 block.c     |   48 ++++++++++++++++++++++++++++++++++++++++++++++++
 block_int.h |   40 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 88 insertions(+), 0 deletions(-)

diff --git a/block.c b/block.c
index daf92c2..d588ee8 100644
--- a/block.c
+++ b/block.c
@@ -3877,3 +3877,51 @@ out:
 
     return ret;
 }
+
+void *block_job_create(const BlockJobType *job_type, BlockDriverState *bs,
+                       BlockDriverCompletionFunc *cb, void *opaque)
+{
+    BlockJob *job;
+
+    if (bs->job || bdrv_in_use(bs)) {
+        return NULL;
+    }
+    bdrv_set_in_use(bs, 1);
+
+    job = g_malloc0(job_type->instance_size);
+    job->job_type      = job_type;
+    job->bs            = bs;
+    job->cb            = cb;
+    job->opaque        = opaque;
+    bs->job = job;
+    return job;
+}
+
+void block_job_complete(BlockJob *job, int ret)
+{
+    BlockDriverState *bs = job->bs;
+
+    assert(bs->job == job);
+    job->cb(job->opaque, ret);
+    bs->job = NULL;
+    g_free(job);
+    bdrv_set_in_use(bs, 0);
+}
+
+int block_job_set_speed(BlockJob *job, int64_t value)
+{
+    if (!job->job_type->set_speed) {
+        return -ENOTSUP;
+    }
+    return job->job_type->set_speed(job, value);
+}
+
+void block_job_cancel(BlockJob *job)
+{
+    job->cancelled = true;
+}
+
+bool block_job_is_cancelled(BlockJob *job)
+{
+    return job->cancelled;
+}
diff --git a/block_int.h b/block_int.h
index 5362180..316443e 100644
--- a/block_int.h
+++ b/block_int.h
@@ -69,6 +69,36 @@ typedef struct BlockIOBaseValue {
     uint64_t ios[2];
 } BlockIOBaseValue;
 
+typedef void BlockJobCancelFunc(void *opaque);
+typedef struct BlockJob BlockJob;
+typedef struct BlockJobType {
+    /** Derived BlockJob struct size */
+    size_t instance_size;
+
+    /** String describing the operation, part of query-block-jobs QMP API */
+    const char *job_type;
+
+    /** Optional callback for job types that support setting a speed limit */
+    int (*set_speed)(BlockJob *job, int64_t value);
+} BlockJobType;
+
+/**
+ * Long-running operation on a BlockDriverState
+ */
+struct BlockJob {
+    const BlockJobType *job_type;
+    BlockDriverState *bs;
+    bool cancelled;
+
+    /* These fields are published by the query-block-jobs QMP API */
+    int64_t offset;
+    int64_t len;
+    int64_t speed;
+
+    BlockDriverCompletionFunc *cb;
+    void *opaque;
+};
+
 struct BlockDriver {
     const char *format_name;
     int instance_size;
@@ -269,6 +299,9 @@ struct BlockDriverState {
     void *private;
 
     QLIST_HEAD(, BdrvTrackedRequest) tracked_requests;
+
+    /* long-running background operation */
+    BlockJob *job;
 };
 
 struct BlockDriverAIOCB {
@@ -292,4 +325,11 @@ void bdrv_set_io_limits(BlockDriverState *bs,
 int is_windows_drive(const char *filename);
 #endif
 
+void *block_job_create(const BlockJobType *job_type, BlockDriverState *bs,
+                       BlockDriverCompletionFunc *cb, void *opaque);
+void block_job_complete(BlockJob *job, int ret);
+int block_job_set_speed(BlockJob *job, int64_t value);
+void block_job_cancel(BlockJob *job);
+bool block_job_is_cancelled(BlockJob *job);
+
 #endif /* BLOCK_INT_H */
-- 
1.7.7.3

Re: [Qemu-devel] [PATCH v5 03/15] block: add BlockJob interface for long-running operations

[Kevin Wolf]

Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
> ---
>  block.c     |   48 ++++++++++++++++++++++++++++++++++++++++++++++++
>  block_int.h |   40 ++++++++++++++++++++++++++++++++++++++++
>  2 files changed, 88 insertions(+), 0 deletions(-)
> 
> diff --git a/block.c b/block.c
> index daf92c2..d588ee8 100644
> --- a/block.c
> +++ b/block.c
> @@ -3877,3 +3877,51 @@ out:
>  
>      return ret;
>  }
> +
> +void *block_job_create(const BlockJobType *job_type, BlockDriverState *bs,
> +                       BlockDriverCompletionFunc *cb, void *opaque)
> +{
> +    BlockJob *job;
> +
> +    if (bs->job || bdrv_in_use(bs)) {
> +        return NULL;
> +    }
> +    bdrv_set_in_use(bs, 1);
> +
> +    job = g_malloc0(job_type->instance_size);
> +    job->job_type      = job_type;
> +    job->bs            = bs;
> +    job->cb            = cb;
> +    job->opaque        = opaque;
> +    bs->job = job;
> +    return job;
> +}
> +
> +void block_job_complete(BlockJob *job, int ret)
> +{
> +    BlockDriverState *bs = job->bs;
> +
> +    assert(bs->job == job);
> +    job->cb(job->opaque, ret);
> +    bs->job = NULL;
> +    g_free(job);
> +    bdrv_set_in_use(bs, 0);
> +}
> +
> +int block_job_set_speed(BlockJob *job, int64_t value)
> +{
> +    if (!job->job_type->set_speed) {
> +        return -ENOTSUP;
> +    }
> +    return job->job_type->set_speed(job, value);
> +}
> +
> +void block_job_cancel(BlockJob *job)
> +{
> +    job->cancelled = true;
> +}
> +
> +bool block_job_is_cancelled(BlockJob *job)
> +{
> +    return job->cancelled;
> +}
> diff --git a/block_int.h b/block_int.h
> index 5362180..316443e 100644
> --- a/block_int.h
> +++ b/block_int.h
> @@ -69,6 +69,36 @@ typedef struct BlockIOBaseValue {
>      uint64_t ios[2];
>  } BlockIOBaseValue;
>  
> +typedef void BlockJobCancelFunc(void *opaque);
> +typedef struct BlockJob BlockJob;
> +typedef struct BlockJobType {
> +    /** Derived BlockJob struct size */
> +    size_t instance_size;
> +
> +    /** String describing the operation, part of query-block-jobs QMP API */
> +    const char *job_type;
> +
> +    /** Optional callback for job types that support setting a speed limit */
> +    int (*set_speed)(BlockJob *job, int64_t value);

Would be worth mentioning what the unit of value is.

Kevin

Re: [Qemu-devel] [PATCH v5 03/15] block: add BlockJob interface for long-running operations

[Stefan Hajnoczi]

On Tue, Jan 17, 2012 at 1:00 PM, Kevin Wolf <kwolf@redhat.com> wrote:
> Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
>> +typedef struct BlockJobType {
>> +    /** Derived BlockJob struct size */
>> +    size_t instance_size;
>> +
>> +    /** String describing the operation, part of query-block-jobs QMP API */
>> +    const char *job_type;
>> +
>> +    /** Optional callback for job types that support setting a speed limit */
>> +    int (*set_speed)(BlockJob *job, int64_t value);
>
> Would be worth mentioning what the unit of value is.

I left this open on purpose so future block jobs could support
block_job_set_speed with whatever unit makes sense for them.  At the
interface level it's an arbitrary int64_t.  Each block job type can
decide how to interpret the values.

I could add "The meaning of value and its units depend on the block
job type".  Or do you think it's problematic to allow different
meanings?

Stefan

Re: [Qemu-devel] [PATCH v5 03/15] block: add BlockJob interface for long-running operations

[Kevin Wolf]

Am 17.01.2012 14:33, schrieb Stefan Hajnoczi:
> On Tue, Jan 17, 2012 at 1:00 PM, Kevin Wolf <kwolf@redhat.com> wrote:
>> Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
>>> +typedef struct BlockJobType {
>>> +    /** Derived BlockJob struct size */
>>> +    size_t instance_size;
>>> +
>>> +    /** String describing the operation, part of query-block-jobs QMP API */
>>> +    const char *job_type;
>>> +
>>> +    /** Optional callback for job types that support setting a speed limit */
>>> +    int (*set_speed)(BlockJob *job, int64_t value);
>>
>> Would be worth mentioning what the unit of value is.
> 
> I left this open on purpose so future block jobs could support
> block_job_set_speed with whatever unit makes sense for them.  At the
> interface level it's an arbitrary int64_t.  Each block job type can
> decide how to interpret the values.

I see.

> I could add "The meaning of value and its units depend on the block
> job type".  Or do you think it's problematic to allow different
> meanings?

Might be confusing to have different meanings. But we can leave it open
for now and commit the comment as it is.

Kevin

[Qemu-devel] [PATCH v5 04/15] block: add image streaming block job

[Stefan Hajnoczi]

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 Makefile.objs  |    1 +
 block/stream.c |  124 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 block_int.h    |    3 +
 trace-events   |    4 ++
 4 files changed, 132 insertions(+), 0 deletions(-)
 create mode 100644 block/stream.c

diff --git a/Makefile.objs b/Makefile.objs
index f4f52e0..949308d 100644
--- a/Makefile.objs
+++ b/Makefile.objs
@@ -35,6 +35,7 @@ block-nested-y += qcow2.o qcow2-refcount.o qcow2-cluster.o qcow2-snapshot.o qcow
 block-nested-y += qed.o qed-gencb.o qed-l2-cache.o qed-table.o qed-cluster.o
 block-nested-y += qed-check.o
 block-nested-y += parallels.o nbd.o blkdebug.o sheepdog.o blkverify.o
+block-nested-y += stream.o
 block-nested-$(CONFIG_WIN32) += raw-win32.o
 block-nested-$(CONFIG_POSIX) += raw-posix.o
 block-nested-$(CONFIG_LIBISCSI) += iscsi.o
diff --git a/block/stream.c b/block/stream.c
new file mode 100644
index 0000000..5255a61
--- /dev/null
+++ b/block/stream.c
@@ -0,0 +1,124 @@
+/*
+ * Image streaming
+ *
+ * Copyright IBM, Corp. 2011
+ *
+ * Authors:
+ *  Stefan Hajnoczi   <stefanha@linux.vnet.ibm.com>
+ *
+ * This work is licensed under the terms of the GNU LGPL, version 2 or later.
+ * See the COPYING.LIB file in the top-level directory.
+ *
+ */
+
+#include "trace.h"
+#include "block_int.h"
+
+enum {
+    /*
+     * Size of data buffer for populating the image file.  This should be large
+     * enough to process multiple clusters in a single call, so that populating
+     * contiguous regions of the image is efficient.
+     */
+    STREAM_BUFFER_SIZE = 512 * 1024, /* in bytes */
+};
+
+typedef struct StreamBlockJob {
+    BlockJob common;
+    BlockDriverState *base;
+} StreamBlockJob;
+
+static int coroutine_fn stream_populate(BlockDriverState *bs,
+                                        int64_t sector_num, int nb_sectors,
+                                        void *buf)
+{
+    struct iovec iov = {
+        .iov_base = buf,
+        .iov_len  = nb_sectors * BDRV_SECTOR_SIZE,
+    };
+    QEMUIOVector qiov;
+
+    qemu_iovec_init_external(&qiov, &iov, 1);
+
+    /* Copy-on-read the unallocated clusters */
+    return bdrv_co_readv(bs, sector_num, nb_sectors, &qiov);
+}
+
+static void coroutine_fn stream_run(void *opaque)
+{
+    StreamBlockJob *s = opaque;
+    BlockDriverState *bs = s->common.bs;
+    int64_t sector_num, end;
+    int ret = 0;
+    int n;
+    void *buf;
+
+    s->common.len = bdrv_getlength(bs);
+    if (s->common.len < 0) {
+        block_job_complete(&s->common, s->common.len);
+        return;
+    }
+
+    end = s->common.len >> BDRV_SECTOR_BITS;
+    buf = qemu_blockalign(bs, STREAM_BUFFER_SIZE);
+
+    bdrv_enable_copy_on_read(bs);
+
+    for (sector_num = 0; sector_num < end; sector_num += n) {
+        if (block_job_is_cancelled(&s->common)) {
+            break;
+        }
+
+        /* TODO rate-limit */
+        /* Note that even when no rate limit is applied we need to yield with
+         * no pending I/O here so that qemu_aio_flush() is able to return.
+         */
+        co_sleep_ns(rt_clock, 0);
+
+        ret = bdrv_co_is_allocated(bs, sector_num,
+                                   STREAM_BUFFER_SIZE / BDRV_SECTOR_SIZE, &n);
+        trace_stream_one_iteration(s, sector_num, n, ret);
+        if (ret == 0) {
+            ret = stream_populate(bs, sector_num, n, buf);
+        }
+        if (ret < 0) {
+            break;
+        }
+
+        /* Publish progress */
+        s->common.offset += n * BDRV_SECTOR_SIZE;
+    }
+
+    bdrv_disable_copy_on_read(bs);
+
+    if (sector_num == end && ret == 0) {
+        ret = bdrv_change_backing_file(bs, NULL, NULL);
+    }
+
+    qemu_vfree(buf);
+    block_job_complete(&s->common, ret);
+}
+
+static BlockJobType stream_job_type = {
+    .instance_size = sizeof(StreamBlockJob),
+    .job_type      = "stream",
+};
+
+int stream_start(BlockDriverState *bs, BlockDriverState *base,
+                 BlockDriverCompletionFunc *cb, void *opaque)
+{
+    StreamBlockJob *s;
+    Coroutine *co;
+
+    s = block_job_create(&stream_job_type, bs, cb, opaque);
+    if (!s) {
+        return -EBUSY; /* bs must already be in use */
+    }
+
+    s->base = base;
+
+    co = qemu_coroutine_create(stream_run);
+    trace_stream_start(bs, base, s, co, opaque);
+    qemu_coroutine_enter(co, s);
+    return 0;
+}
diff --git a/block_int.h b/block_int.h
index 316443e..c7c9178 100644
--- a/block_int.h
+++ b/block_int.h
@@ -332,4 +332,7 @@ int block_job_set_speed(BlockJob *job, int64_t value);
 void block_job_cancel(BlockJob *job);
 bool block_job_is_cancelled(BlockJob *job);
 
+int stream_start(BlockDriverState *bs, BlockDriverState *base,
+                 BlockDriverCompletionFunc *cb, void *opaque);
+
 #endif /* BLOCK_INT_H */
diff --git a/trace-events b/trace-events
index 360f039..c5368fa 100644
--- a/trace-events
+++ b/trace-events
@@ -70,6 +70,10 @@ bdrv_co_write_zeroes(void *bs, int64_t sector_num, int nb_sector) "bs %p sector_
 bdrv_co_io_em(void *bs, int64_t sector_num, int nb_sectors, int is_write, void *acb) "bs %p sector_num %"PRId64" nb_sectors %d is_write %d acb %p"
 bdrv_co_copy_on_readv(void *bs, int64_t sector_num, int nb_sectors, int64_t cluster_sector_num, int cluster_nb_sectors) "bs %p sector_num %"PRId64" nb_sectors %d cluster_sector_num %"PRId64" cluster_nb_sectors %d"
 
+# block/stream.c
+stream_one_iteration(void *s, int64_t sector_num, int nb_sectors, int is_allocated) "s %p sector_num %"PRId64" nb_sectors %d is_allocated %d"
+stream_start(void *bs, void *base, void *s, void *co, void *opaque) "bs %p base %p s %p co %p opaque %p"
+
 # hw/virtio-blk.c
 virtio_blk_req_complete(void *req, int status) "req %p status %d"
 virtio_blk_rw_complete(void *req, int ret) "req %p ret %d"
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 05/15] block: rate-limit streaming operations

[Stefan Hajnoczi]

This patch implements rate-limiting for image streaming.  If we've
exceeded the bandwidth quota for a 100 ms time slice we sleep the
coroutine until the next slice begins.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 block/stream.c |   65 ++++++++++++++++++++++++++++++++++++++++++++++++++-----
 1 files changed, 59 insertions(+), 6 deletions(-)

diff --git a/block/stream.c b/block/stream.c
index 5255a61..93f0305 100644
--- a/block/stream.c
+++ b/block/stream.c
@@ -23,8 +23,39 @@ enum {
     STREAM_BUFFER_SIZE = 512 * 1024, /* in bytes */
 };
 
+#define SLICE_TIME 100000000ULL /* ns */
+
+typedef struct {
+    int64_t next_slice_time;
+    uint64_t slice_quota;
+    uint64_t dispatched;
+} RateLimit;
+
+static int64_t ratelimit_calculate_delay(RateLimit *limit, uint64_t n)
+{
+    int64_t delay_ns = 0;
+    int64_t now = qemu_get_clock_ns(rt_clock);
+
+    if (limit->next_slice_time < now) {
+        limit->next_slice_time = now + SLICE_TIME;
+        limit->dispatched = 0;
+    }
+    if (limit->dispatched + n > limit->slice_quota) {
+        delay_ns = limit->next_slice_time - now;
+    } else {
+        limit->dispatched += n;
+    }
+    return delay_ns;
+}
+
+static void ratelimit_set_speed(RateLimit *limit, uint64_t speed)
+{
+    limit->slice_quota = speed / (1000000000ULL / SLICE_TIME);
+}
+
 typedef struct StreamBlockJob {
     BlockJob common;
+    RateLimit limit;
     BlockDriverState *base;
 } StreamBlockJob;
 
@@ -65,20 +96,24 @@ static void coroutine_fn stream_run(void *opaque)
     bdrv_enable_copy_on_read(bs);
 
     for (sector_num = 0; sector_num < end; sector_num += n) {
+retry:
         if (block_job_is_cancelled(&s->common)) {
             break;
         }
 
-        /* TODO rate-limit */
-        /* Note that even when no rate limit is applied we need to yield with
-         * no pending I/O here so that qemu_aio_flush() is able to return.
-         */
-        co_sleep_ns(rt_clock, 0);
-
         ret = bdrv_co_is_allocated(bs, sector_num,
                                    STREAM_BUFFER_SIZE / BDRV_SECTOR_SIZE, &n);
         trace_stream_one_iteration(s, sector_num, n, ret);
         if (ret == 0) {
+            if (s->common.speed) {
+                uint64_t delay_ns = ratelimit_calculate_delay(&s->limit, n);
+                if (delay_ns > 0) {
+                    co_sleep_ns(rt_clock, delay_ns);
+
+                    /* Recheck cancellation and that sectors are unallocated */
+                    goto retry;
+                }
+            }
             ret = stream_populate(bs, sector_num, n, buf);
         }
         if (ret < 0) {
@@ -87,6 +122,11 @@ static void coroutine_fn stream_run(void *opaque)
 
         /* Publish progress */
         s->common.offset += n * BDRV_SECTOR_SIZE;
+
+        /* Note that even when no rate limit is applied we need to yield
+         * with no pending I/O here so that qemu_aio_flush() returns.
+         */
+        co_sleep_ns(rt_clock, 0);
     }
 
     bdrv_disable_copy_on_read(bs);
@@ -99,9 +139,22 @@ static void coroutine_fn stream_run(void *opaque)
     block_job_complete(&s->common, ret);
 }
 
+static int stream_set_speed(BlockJob *job, int64_t value)
+{
+    StreamBlockJob *s = container_of(job, StreamBlockJob, common);
+
+    if (value < 0) {
+        return -EINVAL;
+    }
+    job->speed = value;
+    ratelimit_set_speed(&s->limit, value / BDRV_SECTOR_SIZE);
+    return 0;
+}
+
 static BlockJobType stream_job_type = {
     .instance_size = sizeof(StreamBlockJob),
     .job_type      = "stream",
+    .set_speed     = stream_set_speed,
 };
 
 int stream_start(BlockDriverState *bs, BlockDriverState *base,
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 06/15] qmp: add block_stream command

[Stefan Hajnoczi]

Add the block_stream command, which starts copy backing file contents
into the image file.  Also add the BLOCK_JOB_COMPLETED QMP event which
is emitted when image streaming completes.  Later patches add control
over the background copy speed, cancelation, and querying running
streaming operations.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 QMP/qmp-events.txt |   29 ++++++++++++++++++++++
 blockdev.c         |   67 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 hmp-commands.hx    |   13 ++++++++++
 hmp.c              |   11 ++++++++
 hmp.h              |    1 +
 monitor.c          |    3 ++
 monitor.h          |    1 +
 qapi-schema.json   |   32 ++++++++++++++++++++++++
 qerror.c           |    4 +++
 qerror.h           |    3 ++
 qmp-commands.hx    |    6 ++++
 trace-events       |    4 +++
 12 files changed, 174 insertions(+), 0 deletions(-)

diff --git a/QMP/qmp-events.txt b/QMP/qmp-events.txt
index af586ec..0cd2275 100644
--- a/QMP/qmp-events.txt
+++ b/QMP/qmp-events.txt
@@ -264,3 +264,32 @@ Example:
 
 Note: If action is "reset", "shutdown", or "pause" the WATCHDOG event is
 followed respectively by the RESET, SHUTDOWN, or STOP events.
+
+
+BLOCK_JOB_COMPLETED
+-------------------
+
+Emitted when a block job has completed.
+
+Data:
+
+- "type":     Job type ("stream" for image streaming, json-string)
+- "device":   Device name (json-string)
+- "len":      Maximum progress value (json-int)
+- "offset":   Current progress value (json-int)
+              On success this is equal to len.
+              On failure this is less than len.
+- "speed":    Rate limit, bytes per second (json-int)
+- "error":    Error message (json-string, optional)
+              Only present on failure.  This field contains a human-readable
+              error message.  There are no semantics other than that streaming
+              has failed and clients should not try to interpret the error
+              string.
+
+Example:
+
+{ "event": "BLOCK_JOB_COMPLETED",
+     "data": { "type": "stream", "device": "virtio-disk0",
+               "len": 10737418240, "offset": 10737418240,
+               "speed": 0 },
+     "timestamp": { "seconds": 1267061043, "microseconds": 959568 } }
diff --git a/blockdev.c b/blockdev.c
index 6d78b36..ba973b0 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -13,9 +13,11 @@
 #include "qerror.h"
 #include "qemu-option.h"
 #include "qemu-config.h"
+#include "qemu-objects.h"
 #include "sysemu.h"
 #include "block_int.h"
 #include "qmp-commands.h"
+#include "trace.h"
 
 static QTAILQ_HEAD(drivelist, DriveInfo) drives = QTAILQ_HEAD_INITIALIZER(drives);
 
@@ -880,3 +882,68 @@ void qmp_block_resize(const char *device, int64_t size, Error **errp)
         return;
     }
 }
+
+static QObject *qobject_from_block_job(BlockJob *job)
+{
+    return qobject_from_jsonf("{ 'type': %s,"
+                              "'device': %s,"
+                              "'len': %" PRId64 ","
+                              "'offset': %" PRId64 ","
+                              "'speed': %" PRId64 " }",
+                              job->job_type->job_type,
+                              bdrv_get_device_name(job->bs),
+                              job->len,
+                              job->offset,
+                              job->speed);
+}
+
+static void block_stream_cb(void *opaque, int ret)
+{
+    BlockDriverState *bs = opaque;
+    QObject *obj;
+
+    trace_block_stream_cb(bs, bs->job, ret);
+
+    assert(bs->job);
+    obj = qobject_from_block_job(bs->job);
+    if (ret < 0) {
+        QDict *dict = qobject_to_qdict(obj);
+        qdict_put(dict, "error", qstring_from_str(strerror(-ret)));
+    }
+
+    monitor_protocol_event(QEVENT_BLOCK_JOB_COMPLETED, obj);
+    qobject_decref(obj);
+}
+
+void qmp_block_stream(const char *device, bool has_base,
+                      const char *base, Error **errp)
+{
+    BlockDriverState *bs;
+    int ret;
+
+    bs = bdrv_find(device);
+    if (!bs) {
+        error_set(errp, QERR_DEVICE_NOT_FOUND, device);
+        return;
+    }
+
+    /* Base device not supported */
+    if (base) {
+        error_set(errp, QERR_NOT_SUPPORTED);
+        return;
+    }
+
+    ret = stream_start(bs, NULL, block_stream_cb, bs);
+    if (ret < 0) {
+        switch (ret) {
+        case -EBUSY:
+            error_set(errp, QERR_DEVICE_IN_USE, device);
+            return;
+        default:
+            error_set(errp, QERR_NOT_SUPPORTED);
+            return;
+        }
+    }
+
+    trace_qmp_block_stream(bs, bs->job);
+}
diff --git a/hmp-commands.hx b/hmp-commands.hx
index a586498..dc6c8c3 100644
--- a/hmp-commands.hx
+++ b/hmp-commands.hx
@@ -69,6 +69,19 @@ but should be used with extreme caution.  Note that this command only
 resizes image files, it can not resize block devices like LVM volumes.
 ETEXI
 
+    {
+        .name       = "block_stream",
+        .args_type  = "device:B,base:s?",
+        .params     = "device [base]",
+        .help       = "copy data from a backing file into a block device",
+        .mhandler.cmd = hmp_block_stream,
+    },
+
+STEXI
+@item block_stream
+@findex block_stream
+Copy data from a backing file into a block device.
+ETEXI
 
     {
         .name       = "eject",
diff --git a/hmp.c b/hmp.c
index e7659d5..b6e5913 100644
--- a/hmp.c
+++ b/hmp.c
@@ -679,3 +679,14 @@ void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict)
     int64_t value = qdict_get_int(qdict, "value");
     qmp_migrate_set_speed(value, NULL);
 }
+
+void hmp_block_stream(Monitor *mon, const QDict *qdict)
+{
+    Error *error = NULL;
+    const char *device = qdict_get_str(qdict, "device");
+    const char *base = qdict_get_try_str(qdict, "base");
+
+    qmp_block_stream(device, base != NULL, base, &error);
+
+    hmp_handle_error(mon, &error);
+}
diff --git a/hmp.h b/hmp.h
index 093242d..b55c295 100644
--- a/hmp.h
+++ b/hmp.h
@@ -49,5 +49,6 @@ void hmp_snapshot_blkdev(Monitor *mon, const QDict *qdict);
 void hmp_migrate_cancel(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_downtime(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict);
+void hmp_block_stream(Monitor *mon, const QDict *qdict);
 
 #endif
diff --git a/monitor.c b/monitor.c
index 7334401..bb42580 100644
--- a/monitor.c
+++ b/monitor.c
@@ -479,6 +479,9 @@ void monitor_protocol_event(MonitorEvent event, QObject *data)
         case QEVENT_SPICE_DISCONNECTED:
             event_name = "SPICE_DISCONNECTED";
             break;
+        case QEVENT_BLOCK_JOB_COMPLETED:
+            event_name = "BLOCK_JOB_COMPLETED";
+            break;
         default:
             abort();
             break;
diff --git a/monitor.h b/monitor.h
index cfa2f67..7324236 100644
--- a/monitor.h
+++ b/monitor.h
@@ -35,6 +35,7 @@ typedef enum MonitorEvent {
     QEVENT_SPICE_CONNECTED,
     QEVENT_SPICE_INITIALIZED,
     QEVENT_SPICE_DISCONNECTED,
+    QEVENT_BLOCK_JOB_COMPLETED,
     QEVENT_MAX,
 } MonitorEvent;
 
diff --git a/qapi-schema.json b/qapi-schema.json
index 44cf764..3821982 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -1275,3 +1275,35 @@
 { 'command': 'qom-set',
   'data': { 'path': 'str', 'property': 'str', 'value': 'visitor' },
   'gen': 'no' }
+
+##
+# @block_stream:
+#
+# Copy data from a backing file into a block device.
+#
+# The block streaming operation is performed in the background until the entire
+# backing file has been copied.  This command returns immediately once streaming
+# has started.  The status of ongoing block streaming operations can be checked
+# with query-block-jobs.  The operation can be stopped before it has completed
+# using the block_job_cancel command.
+#
+# If a base file is specified then sectors are not copied from that base file and
+# its backing chain.  When streaming completes the image file will have the base
+# file as its backing file.  This can be used to stream a subset of the backing
+# file chain instead of flattening the entire image.
+#
+# On successful completion the image file is updated to drop the backing file
+# and the BLOCK_JOB_COMPLETED event is emitted.
+#
+# @device: the device name
+#
+# @base:   #optional the common backing file name
+#
+# Returns: Nothing on success
+#          If streaming is already active on this device, DeviceInUse
+#          If @device does not exist, DeviceNotFound
+#          If image streaming is not supported by this device, NotSupported
+#
+# Since: 1.1
+##
+{ 'command': 'block_stream', 'data': { 'device': 'str', '*base': 'str' } }
diff --git a/qerror.c b/qerror.c
index 9a75d06..feb3d35 100644
--- a/qerror.c
+++ b/qerror.c
@@ -182,6 +182,10 @@ static const QErrorStringTable qerror_table[] = {
         .desc      = "No '%(bus)' bus found for device '%(device)'",
     },
     {
+        .error_fmt = QERR_NOT_SUPPORTED,
+        .desc      = "Not supported",
+    },
+    {
         .error_fmt = QERR_OPEN_FILE_FAILED,
         .desc      = "Could not open '%(filename)'",
     },
diff --git a/qerror.h b/qerror.h
index efda232..095ba9d 100644
--- a/qerror.h
+++ b/qerror.h
@@ -153,6 +153,9 @@ QError *qobject_to_qerror(const QObject *obj);
 #define QERR_NO_BUS_FOR_DEVICE \
     "{ 'class': 'NoBusForDevice', 'data': { 'device': %s, 'bus': %s } }"
 
+#define QERR_NOT_SUPPORTED \
+    "{ 'class': 'NotSupported', 'data': {} }"
+
 #define QERR_OPEN_FILE_FAILED \
     "{ 'class': 'OpenFileFailed', 'data': { 'filename': %s } }"
 
diff --git a/qmp-commands.hx b/qmp-commands.hx
index 7e3f4b9..b9ebb76 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -655,6 +655,12 @@ Example:
 EQMP
 
     {
+        .name       = "block_stream",
+        .args_type  = "device:B,base:s?",
+        .mhandler.cmd_new = qmp_marshal_input_block_stream,
+    },
+
+    {
         .name       = "blockdev-snapshot-sync",
         .args_type  = "device:B,snapshot-file:s,format:s?",
         .mhandler.cmd_new = qmp_marshal_input_blockdev_snapshot_sync,
diff --git a/trace-events b/trace-events
index c5368fa..6ff0d43 100644
--- a/trace-events
+++ b/trace-events
@@ -74,6 +74,10 @@ bdrv_co_copy_on_readv(void *bs, int64_t sector_num, int nb_sectors, int64_t clus
 stream_one_iteration(void *s, int64_t sector_num, int nb_sectors, int is_allocated) "s %p sector_num %"PRId64" nb_sectors %d is_allocated %d"
 stream_start(void *bs, void *base, void *s, void *co, void *opaque) "bs %p base %p s %p co %p opaque %p"
 
+# blockdev.c
+block_stream_cb(void *bs, void *job, int ret) "bs %p job %p ret %d"
+qmp_block_stream(void *bs, void *job) "bs %p job %p"
+
 # hw/virtio-blk.c
 virtio_blk_req_complete(void *req, int status) "req %p status %d"
 virtio_blk_rw_complete(void *req, int ret) "req %p ret %d"
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 07/15] qmp: add block_job_set_speed command

[Stefan Hajnoczi]

Add block_job_set_speed, which sets the maximum speed for a background
block operation.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 blockdev.c       |   25 +++++++++++++++++++++++++
 hmp-commands.hx  |   14 ++++++++++++++
 hmp.c            |   11 +++++++++++
 hmp.h            |    1 +
 qapi-schema.json |   22 ++++++++++++++++++++++
 qmp-commands.hx  |    6 ++++++
 6 files changed, 79 insertions(+), 0 deletions(-)

diff --git a/blockdev.c b/blockdev.c
index ba973b0..2dfca40 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -947,3 +947,28 @@ void qmp_block_stream(const char *device, bool has_base,
 
     trace_qmp_block_stream(bs, bs->job);
 }
+
+static BlockJob *find_block_job(const char *device)
+{
+    BlockDriverState *bs;
+
+    bs = bdrv_find(device);
+    if (!bs || !bs->job) {
+        return NULL;
+    }
+    return bs->job;
+}
+
+void qmp_block_job_set_speed(const char *device, int64_t value, Error **errp)
+{
+    BlockJob *job = find_block_job(device);
+
+    if (!job) {
+        error_set(errp, QERR_DEVICE_NOT_ACTIVE, device);
+        return;
+    }
+
+    if (block_job_set_speed(job, value) < 0) {
+        error_set(errp, QERR_NOT_SUPPORTED);
+    }
+}
diff --git a/hmp-commands.hx b/hmp-commands.hx
index dc6c8c3..12b8433 100644
--- a/hmp-commands.hx
+++ b/hmp-commands.hx
@@ -84,6 +84,20 @@ Copy data from a backing file into a block device.
 ETEXI
 
     {
+        .name       = "block_job_set_speed",
+        .args_type  = "device:B,value:o",
+        .params     = "device value",
+        .help       = "set maximum speed for a background block operation",
+        .mhandler.cmd = hmp_block_job_set_speed,
+    },
+
+STEXI
+@item block_job_set_stream
+@findex block_job_set_stream
+Set maximum speed for a background block operation.
+ETEXI
+
+    {
         .name       = "eject",
         .args_type  = "force:-f,device:B",
         .params     = "[-f] device",
diff --git a/hmp.c b/hmp.c
index b6e5913..1144d53 100644
--- a/hmp.c
+++ b/hmp.c
@@ -690,3 +690,14 @@ void hmp_block_stream(Monitor *mon, const QDict *qdict)
 
     hmp_handle_error(mon, &error);
 }
+
+void hmp_block_job_set_speed(Monitor *mon, const QDict *qdict)
+{
+    Error *error = NULL;
+    const char *device = qdict_get_str(qdict, "device");
+    int64_t value = qdict_get_int(qdict, "value");
+
+    qmp_block_job_set_speed(device, value, &error);
+
+    hmp_handle_error(mon, &error);
+}
diff --git a/hmp.h b/hmp.h
index b55c295..2c871ea 100644
--- a/hmp.h
+++ b/hmp.h
@@ -50,5 +50,6 @@ void hmp_migrate_cancel(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_downtime(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict);
 void hmp_block_stream(Monitor *mon, const QDict *qdict);
+void hmp_block_job_set_speed(Monitor *mon, const QDict *qdict);
 
 #endif
diff --git a/qapi-schema.json b/qapi-schema.json
index 3821982..5872096 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -1307,3 +1307,25 @@
 # Since: 1.1
 ##
 { 'command': 'block_stream', 'data': { 'device': 'str', '*base': 'str' } }
+
+##
+# @block_job_set_speed:
+#
+# Set maximum speed for a background block operation.
+#
+# This command can only be issued when there is an active block job.
+#
+# Throttling can be disabled by setting the speed to 0.
+#
+# @device: the device name
+#
+# @value:  the maximum speed, in bytes per second
+#
+# Returns: Nothing on success
+#          If the job type does not support throttling, NotSupported
+#          If streaming is not active on this device, DeviceNotActive
+#
+# Since: 1.1
+##
+{ 'command': 'block_job_set_speed',
+  'data': { 'device': 'str', 'value': 'int' } }
diff --git a/qmp-commands.hx b/qmp-commands.hx
index b9ebb76..dc6bc2e 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -661,6 +661,12 @@ EQMP
     },
 
     {
+        .name       = "block_job_set_speed",
+        .args_type  = "device:B,value:o",
+        .mhandler.cmd_new = qmp_marshal_input_block_job_set_speed,
+    },
+
+    {
         .name       = "blockdev-snapshot-sync",
         .args_type  = "device:B,snapshot-file:s,format:s?",
         .mhandler.cmd_new = qmp_marshal_input_blockdev_snapshot_sync,
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 08/15] qmp: add block_job_cancel command

[Stefan Hajnoczi]

Add block_job_cancel, which stops an active block streaming operation.
When the operation has been cancelled the new BLOCK_JOB_CANCELLED event
is emitted.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 QMP/qmp-events.txt |   24 ++++++++++++++++++++++++
 blockdev.c         |   19 ++++++++++++++++++-
 hmp-commands.hx    |   14 ++++++++++++++
 hmp.c              |   10 ++++++++++
 hmp.h              |    1 +
 monitor.c          |    3 +++
 monitor.h          |    1 +
 qapi-schema.json   |   29 +++++++++++++++++++++++++++++
 qmp-commands.hx    |    6 ++++++
 trace-events       |    1 +
 10 files changed, 107 insertions(+), 1 deletions(-)

diff --git a/QMP/qmp-events.txt b/QMP/qmp-events.txt
index 0cd2275..06cb404 100644
--- a/QMP/qmp-events.txt
+++ b/QMP/qmp-events.txt
@@ -293,3 +293,27 @@ Example:
                "len": 10737418240, "offset": 10737418240,
                "speed": 0 },
      "timestamp": { "seconds": 1267061043, "microseconds": 959568 } }
+
+
+BLOCK_JOB_CANCELLED
+-------------------
+
+Emitted when a block job has been cancelled.
+
+Data:
+
+- "type":     Job type ("stream" for image streaming, json-string)
+- "device":   Device name (json-string)
+- "len":      Maximum progress value (json-int)
+- "offset":   Current progress value (json-int)
+              On success this is equal to len.
+              On failure this is less than len.
+- "speed":    Rate limit, bytes per second (json-int)
+
+Example:
+
+{ "event": "BLOCK_JOB_CANCELLED",
+     "data": { "type": "stream", "device": "virtio-disk0",
+               "len": 10737418240, "offset": 134217728,
+               "speed": 0 },
+     "timestamp": { "seconds": 1267061043, "microseconds": 959568 } }
diff --git a/blockdev.c b/blockdev.c
index 2dfca40..35de3bc 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -911,7 +911,11 @@ static void block_stream_cb(void *opaque, int ret)
         qdict_put(dict, "error", qstring_from_str(strerror(-ret)));
     }
 
-    monitor_protocol_event(QEVENT_BLOCK_JOB_COMPLETED, obj);
+    if (block_job_is_cancelled(bs->job)) {
+        monitor_protocol_event(QEVENT_BLOCK_JOB_CANCELLED, obj);
+    } else {
+        monitor_protocol_event(QEVENT_BLOCK_JOB_COMPLETED, obj);
+    }
     qobject_decref(obj);
 }
 
@@ -972,3 +976,16 @@ void qmp_block_job_set_speed(const char *device, int64_t value, Error **errp)
         error_set(errp, QERR_NOT_SUPPORTED);
     }
 }
+
+void qmp_block_job_cancel(const char *device, Error **errp)
+{
+    BlockJob *job = find_block_job(device);
+
+    if (!job) {
+        error_set(errp, QERR_DEVICE_NOT_ACTIVE, device);
+        return;
+    }
+
+    trace_qmp_block_job_cancel(job);
+    block_job_cancel(job);
+}
diff --git a/hmp-commands.hx b/hmp-commands.hx
index 12b8433..b991ee0 100644
--- a/hmp-commands.hx
+++ b/hmp-commands.hx
@@ -98,6 +98,20 @@ Set maximum speed for a background block operation.
 ETEXI
 
     {
+        .name       = "block_job_cancel",
+        .args_type  = "device:B",
+        .params     = "device",
+        .help       = "stop an active block streaming operation",
+        .mhandler.cmd = hmp_block_job_cancel,
+    },
+
+STEXI
+@item block_job_cancel
+@findex block_job_cancel
+Stop an active block streaming operation.
+ETEXI
+
+    {
         .name       = "eject",
         .args_type  = "force:-f,device:B",
         .params     = "[-f] device",
diff --git a/hmp.c b/hmp.c
index 1144d53..851885b 100644
--- a/hmp.c
+++ b/hmp.c
@@ -701,3 +701,13 @@ void hmp_block_job_set_speed(Monitor *mon, const QDict *qdict)
 
     hmp_handle_error(mon, &error);
 }
+
+void hmp_block_job_cancel(Monitor *mon, const QDict *qdict)
+{
+    Error *error = NULL;
+    const char *device = qdict_get_str(qdict, "device");
+
+    qmp_block_job_cancel(device, &error);
+
+    hmp_handle_error(mon, &error);
+}
diff --git a/hmp.h b/hmp.h
index 2c871ea..0ad2004 100644
--- a/hmp.h
+++ b/hmp.h
@@ -51,5 +51,6 @@ void hmp_migrate_set_downtime(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict);
 void hmp_block_stream(Monitor *mon, const QDict *qdict);
 void hmp_block_job_set_speed(Monitor *mon, const QDict *qdict);
+void hmp_block_job_cancel(Monitor *mon, const QDict *qdict);
 
 #endif
diff --git a/monitor.c b/monitor.c
index bb42580..01850ca 100644
--- a/monitor.c
+++ b/monitor.c
@@ -482,6 +482,9 @@ void monitor_protocol_event(MonitorEvent event, QObject *data)
         case QEVENT_BLOCK_JOB_COMPLETED:
             event_name = "BLOCK_JOB_COMPLETED";
             break;
+        case QEVENT_BLOCK_JOB_CANCELLED:
+            event_name = "BLOCK_JOB_CANCELLED";
+            break;
         default:
             abort();
             break;
diff --git a/monitor.h b/monitor.h
index 7324236..86c997d 100644
--- a/monitor.h
+++ b/monitor.h
@@ -36,6 +36,7 @@ typedef enum MonitorEvent {
     QEVENT_SPICE_INITIALIZED,
     QEVENT_SPICE_DISCONNECTED,
     QEVENT_BLOCK_JOB_COMPLETED,
+    QEVENT_BLOCK_JOB_CANCELLED,
     QEVENT_MAX,
 } MonitorEvent;
 
diff --git a/qapi-schema.json b/qapi-schema.json
index 5872096..3d23ce2 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -1329,3 +1329,32 @@
 ##
 { 'command': 'block_job_set_speed',
   'data': { 'device': 'str', 'value': 'int' } }
+
+##
+# @block_job_cancel:
+#
+# Stop an active block streaming operation.
+#
+# This command returns immediately after marking the active block streaming
+# operation for cancellation.  It is an error to call this command if no
+# operation is in progress.
+#
+# The operation will cancel as soon as possible and then emit the
+# BLOCK_JOB_CANCELLED event.  Before that happens the job is still visible when
+# enumerated using query-block-jobs.
+#
+# The image file retains its backing file unless the streaming operation happens
+# to complete just as it is being cancelled.
+#
+# A new block streaming operation can be started at a later time to finish
+# copying all data from the backing file.
+#
+# @device: the device name
+#
+# Returns: Nothing on success
+#          If streaming is not active on this device, DeviceNotActive
+#          If cancellation already in progress, DeviceInUse
+#
+# Since: 1.1
+##
+{ 'command': 'block_job_cancel', 'data': { 'device': 'str' } }
diff --git a/qmp-commands.hx b/qmp-commands.hx
index dc6bc2e..0a0335f 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -667,6 +667,12 @@ EQMP
     },
 
     {
+        .name       = "block_job_cancel",
+        .args_type  = "device:B",
+        .mhandler.cmd_new = qmp_marshal_input_block_job_cancel,
+    },
+
+    {
         .name       = "blockdev-snapshot-sync",
         .args_type  = "device:B,snapshot-file:s,format:s?",
         .mhandler.cmd_new = qmp_marshal_input_blockdev_snapshot_sync,
diff --git a/trace-events b/trace-events
index 6ff0d43..196a872 100644
--- a/trace-events
+++ b/trace-events
@@ -75,6 +75,7 @@ stream_one_iteration(void *s, int64_t sector_num, int nb_sectors, int is_allocat
 stream_start(void *bs, void *base, void *s, void *co, void *opaque) "bs %p base %p s %p co %p opaque %p"
 
 # blockdev.c
+qmp_block_job_cancel(void *job) "job %p"
 block_stream_cb(void *bs, void *job, int ret) "bs %p job %p ret %d"
 qmp_block_stream(void *bs, void *job) "bs %p job %p"
 
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 09/15] qmp: add query-block-jobs

[Stefan Hajnoczi]

Add query-block-jobs, which shows the progress of ongoing block device
operations.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 blockdev.c       |   33 +++++++++++++++++++++++++++++++++
 hmp.c            |   36 ++++++++++++++++++++++++++++++++++++
 hmp.h            |    1 +
 monitor.c        |    7 +++++++
 qapi-schema.json |   32 ++++++++++++++++++++++++++++++++
 qmp-commands.hx  |    6 ++++++
 6 files changed, 115 insertions(+), 0 deletions(-)

diff --git a/blockdev.c b/blockdev.c
index 35de3bc..4549c9e 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -989,3 +989,36 @@ void qmp_block_job_cancel(const char *device, Error **errp)
     trace_qmp_block_job_cancel(job);
     block_job_cancel(job);
 }
+
+static void do_qmp_query_block_jobs_one(void *opaque, BlockDriverState *bs)
+{
+    BlockJobInfoList **prev = opaque;
+    BlockJob *job = bs->job;
+
+    if (job) {
+        BlockJobInfoList *elem;
+        BlockJobInfo *info = g_new(BlockJobInfo, 1);
+        *info = (BlockJobInfo){
+            .type   = g_strdup(job->job_type->job_type),
+            .device = g_strdup(bdrv_get_device_name(bs)),
+            .len    = job->len,
+            .offset = job->offset,
+            .speed  = job->speed,
+        };
+
+        elem = g_new0(BlockJobInfoList, 1);
+        elem->value = info;
+
+        (*prev)->next = elem;
+        *prev = elem;
+    }
+}
+
+BlockJobInfoList *qmp_query_block_jobs(Error **errp)
+{
+    /* Dummy is a fake list element for holding the head pointer */
+    BlockJobInfoList dummy = {};
+    BlockJobInfoList *prev = &dummy;
+    bdrv_iterate(do_qmp_query_block_jobs_one, &prev);
+    return dummy.next;
+}
diff --git a/hmp.c b/hmp.c
index 851885b..76e89f8 100644
--- a/hmp.c
+++ b/hmp.c
@@ -507,6 +507,42 @@ void hmp_info_pci(Monitor *mon)
     qapi_free_PciInfoList(info);
 }
 
+void hmp_info_block_jobs(Monitor *mon)
+{
+    BlockJobInfoList *list;
+    Error *err = NULL;
+
+    list = qmp_query_block_jobs(&err);
+    assert(!err);
+
+    if (!list) {
+        monitor_printf(mon, "No active jobs\n");
+        return;
+    }
+
+    while (list) {
+        if (strcmp(list->value->type, "stream") == 0) {
+            monitor_printf(mon, "Streaming device %s: Completed %" PRId64
+                           " of %" PRId64 " bytes, speed limit %" PRId64
+                           " bytes/s\n",
+                           list->value->device,
+                           list->value->offset,
+                           list->value->len,
+                           list->value->speed);
+        } else {
+            monitor_printf(mon, "Type %s, device %s: Completed %" PRId64
+                           " of %" PRId64 " bytes, speed limit %" PRId64
+                           " bytes/s\n",
+                           list->value->type,
+                           list->value->device,
+                           list->value->offset,
+                           list->value->len,
+                           list->value->speed);
+        }
+        list = list->next;
+    }
+}
+
 void hmp_quit(Monitor *mon, const QDict *qdict)
 {
     monitor_suspend(mon);
diff --git a/hmp.h b/hmp.h
index 0ad2004..23bfca2 100644
--- a/hmp.h
+++ b/hmp.h
@@ -32,6 +32,7 @@ void hmp_info_vnc(Monitor *mon);
 void hmp_info_spice(Monitor *mon);
 void hmp_info_balloon(Monitor *mon);
 void hmp_info_pci(Monitor *mon);
+void hmp_info_block_jobs(Monitor *mon);
 void hmp_quit(Monitor *mon, const QDict *qdict);
 void hmp_stop(Monitor *mon, const QDict *qdict);
 void hmp_system_reset(Monitor *mon, const QDict *qdict);
diff --git a/monitor.c b/monitor.c
index 01850ca..f96a296 100644
--- a/monitor.c
+++ b/monitor.c
@@ -2483,6 +2483,13 @@ static mon_cmd_t info_cmds[] = {
         .mhandler.info = hmp_info_blockstats,
     },
     {
+        .name       = "block-jobs",
+        .args_type  = "",
+        .params     = "",
+        .help       = "show progress of ongoing block device operations",
+        .mhandler.info = hmp_info_block_jobs,
+    },
+    {
         .name       = "registers",
         .args_type  = "",
         .params     = "",
diff --git a/qapi-schema.json b/qapi-schema.json
index 3d23ce2..b4f6b15 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -845,6 +845,38 @@
 { 'command': 'query-pci', 'returns': ['PciInfo'] }
 
 ##
+# @BlockJobInfo:
+#
+# Information about a long-running block device operation.
+#
+# @type: the job type ('stream' for image streaming)
+#
+# @device: the block device name
+#
+# @len: the maximum progress value
+#
+# @offset: the current progress value
+#
+# @speed: the rate limit, bytes per second
+#
+# Since: 1.1
+##
+{ 'type': 'BlockJobInfo',
+  'data': {'type': 'str', 'device': 'str', 'len': 'int',
+           'offset': 'int', 'speed': 'int'} }
+
+##
+# @query-block-jobs:
+#
+# Return information about long-running block device operations.
+#
+# Returns: a list of @BlockJobInfo for each active block job
+#
+# Since: 1.1
+##
+{ 'command': 'query-block-jobs', 'returns': ['BlockJobInfo'] }
+
+##
 # @quit:
 #
 # This command will cause the QEMU process to exit gracefully.  While every
diff --git a/qmp-commands.hx b/qmp-commands.hx
index 0a0335f..4be6632 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -2029,6 +2029,12 @@ EQMP
     },
 
     {
+        .name       = "query-block-jobs",
+        .args_type  = "",
+        .mhandler.cmd_new = qmp_marshal_input_query_block_jobs,
+    },
+
+    {
         .name       = "qom-list",
         .args_type  = "path:s",
         .mhandler.cmd_new = qmp_marshal_input_qom_list,
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 10/15] blockdev: make image streaming safe across hotplug

[Stefan Hajnoczi]

Unplugging a storage interface like virtio-blk causes the host block
device to be deleted too.  Long-running operations like block migration
must take a DriveInfo reference to prevent the BlockDriverState from
being freed.  For image streaming we can do the same thing.

Note that it is not possible to acquire/release the drive reference in
block.c where the block job functions live because
drive_get_ref()/drive_put_ref() are blockdev.c functions.  Calling them
from block.c would be a layering violation - tools like qemu-img don't
even link against blockdev.c.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 blockdev.c |   38 ++++++++++++++++++++++++++++++++++++++
 1 files changed, 38 insertions(+), 0 deletions(-)

diff --git a/blockdev.c b/blockdev.c
index 4549c9e..45a6ba6 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -202,6 +202,37 @@ void drive_get_ref(DriveInfo *dinfo)
     dinfo->refcount++;
 }
 
+typedef struct {
+    QEMUBH *bh;
+    DriveInfo *dinfo;
+} DrivePutRefBH;
+
+static void drive_put_ref_bh(void *opaque)
+{
+    DrivePutRefBH *s = opaque;
+
+    drive_put_ref(s->dinfo);
+    qemu_bh_delete(s->bh);
+    g_free(s);
+}
+
+/*
+ * Release a drive reference in a BH
+ *
+ * It is not possible to use drive_put_ref() from a callback function when the
+ * callers still need the drive.  In such cases we schedule a BH to release the
+ * reference.
+ */
+static void drive_put_ref_bh_schedule(DriveInfo *dinfo)
+{
+    DrivePutRefBH *s;
+
+    s = g_new(DrivePutRefBH, 1);
+    s->bh = qemu_bh_new(drive_put_ref_bh, s);
+    s->dinfo = dinfo;
+    qemu_bh_schedule(s->bh);
+}
+
 static int parse_block_error_action(const char *buf, int is_read)
 {
     if (!strcmp(buf, "ignore")) {
@@ -917,6 +948,8 @@ static void block_stream_cb(void *opaque, int ret)
         monitor_protocol_event(QEVENT_BLOCK_JOB_COMPLETED, obj);
     }
     qobject_decref(obj);
+
+    drive_put_ref_bh_schedule(drive_get_by_blockdev(bs));
 }
 
 void qmp_block_stream(const char *device, bool has_base,
@@ -949,6 +982,11 @@ void qmp_block_stream(const char *device, bool has_base,
         }
     }
 
+    /* Grab a reference so hotplug does not delete the BlockDriverState from
+     * underneath us.
+     */
+    drive_get_ref(drive_get_by_blockdev(bs));
+
     trace_qmp_block_stream(bs, bs->job);
 }
 
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 11/15] block: add bdrv_find_backing_image

[Stefan Hajnoczi]

From: Marcelo Tosatti <mtosatti@redhat.com>

Add bdrv_find_backing_image: given a BlockDriverState pointer, and an id,
traverse the backing image chain to locate the id.

Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 block.c |   18 ++++++++++++++++++
 block.h |    2 ++
 2 files changed, 20 insertions(+), 0 deletions(-)

diff --git a/block.c b/block.c
index d588ee8..d8ae716 100644
--- a/block.c
+++ b/block.c
@@ -2614,6 +2614,24 @@ int bdrv_snapshot_load_tmp(BlockDriverState *bs,
     return -ENOTSUP;
 }
 
+BlockDriverState *bdrv_find_backing_image(BlockDriverState *bs,
+        const char *backing_file)
+{
+    if (!bs->drv) {
+        return NULL;
+    }
+
+    if (bs->backing_hd) {
+        if (strcmp(bs->backing_file, backing_file) == 0) {
+            return bs->backing_hd;
+        } else {
+            return bdrv_find_backing_image(bs->backing_hd, backing_file);
+        }
+    }
+
+    return NULL;
+}
+
 #define NB_SUFFIXES 4
 
 char *get_human_readable_size(char *buf, int buf_size, int64_t size)
diff --git a/block.h b/block.h
index 51b90c7..05c8c83 100644
--- a/block.h
+++ b/block.h
@@ -153,6 +153,8 @@ int coroutine_fn bdrv_co_write_zeroes(BlockDriverState *bs, int64_t sector_num,
     int nb_sectors);
 int coroutine_fn bdrv_co_is_allocated(BlockDriverState *bs, int64_t sector_num,
     int nb_sectors, int *pnum);
+BlockDriverState *bdrv_find_backing_image(BlockDriverState *bs,
+    const char *backing_file);
 int bdrv_truncate(BlockDriverState *bs, int64_t offset);
 int64_t bdrv_getlength(BlockDriverState *bs);
 int64_t bdrv_get_allocated_file_size(BlockDriverState *bs);
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 12/15] add QERR_BASE_NOT_FOUND

[Stefan Hajnoczi]

From: Marcelo Tosatti <mtosatti@redhat.com>

This qerror will be raised when a given streaming base (backing file)
cannot be found.

Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 qapi-schema.json |    1 +
 qerror.c         |    4 ++++
 qerror.h         |    3 +++
 3 files changed, 8 insertions(+), 0 deletions(-)

diff --git a/qapi-schema.json b/qapi-schema.json
index b4f6b15..b778639 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -1335,6 +1335,7 @@
 #          If streaming is already active on this device, DeviceInUse
 #          If @device does not exist, DeviceNotFound
 #          If image streaming is not supported by this device, NotSupported
+#          If @base does not exist, BaseNotFound
 #
 # Since: 1.1
 ##
diff --git a/qerror.c b/qerror.c
index feb3d35..272243491 100644
--- a/qerror.c
+++ b/qerror.c
@@ -49,6 +49,10 @@ static const QErrorStringTable qerror_table[] = {
         .desc      = "Device '%(device)' can't go on a %(bad_bus_type) bus",
     },
     {
+        .error_fmt = QERR_BASE_NOT_FOUND,
+        .desc      = "Base '%(base)' not found",
+    },
+    {
         .error_fmt = QERR_BLOCK_FORMAT_FEATURE_NOT_SUPPORTED,
         .desc      = "Block format '%(format)' used by device '%(name)' does not support feature '%(feature)'",
     },
diff --git a/qerror.h b/qerror.h
index 095ba9d..4351fe3 100644
--- a/qerror.h
+++ b/qerror.h
@@ -54,6 +54,9 @@ QError *qobject_to_qerror(const QObject *obj);
 #define QERR_BAD_BUS_FOR_DEVICE \
     "{ 'class': 'BadBusForDevice', 'data': { 'device': %s, 'bad_bus_type': %s } }"
 
+#define QERR_BASE_NOT_FOUND \
+    "{ 'class': 'BaseNotFound', 'data': { 'base': %s } }"
+
 #define QERR_BLOCK_FORMAT_FEATURE_NOT_SUPPORTED \
     "{ 'class': 'BlockFormatFeatureNotSupported', 'data': { 'format': %s, 'name': %s, 'feature': %s } }"
 
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 13/15] block: add support for partial streaming

[Stefan Hajnoczi]

From: Marcelo Tosatti <mtosatti@redhat.com>

Add support for streaming data from an intermediate section of the
image chain (see patch and documentation for details).

Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 block/stream.c |   91 +++++++++++++++++++++++++++++++++++++++++++++++++++++--
 block_int.h    |    3 +-
 blockdev.c     |   11 ++++--
 3 files changed, 96 insertions(+), 9 deletions(-)

diff --git a/block/stream.c b/block/stream.c
index 93f0305..7532f5e 100644
--- a/block/stream.c
+++ b/block/stream.c
@@ -57,6 +57,7 @@ typedef struct StreamBlockJob {
     BlockJob common;
     RateLimit limit;
     BlockDriverState *base;
+    char backing_file_id[1024];
 } StreamBlockJob;
 
 static int coroutine_fn stream_populate(BlockDriverState *bs,
@@ -75,10 +76,76 @@ static int coroutine_fn stream_populate(BlockDriverState *bs,
     return bdrv_co_readv(bs, sector_num, nb_sectors, &qiov);
 }
 
+/*
+ * Given an image chain: [BASE] -> [INTER1] -> [INTER2] -> [TOP]
+ *
+ * Return true if the given sector is allocated in top.
+ * Return false if the given sector is allocated in intermediate images.
+ * Return true otherwise.
+ *
+ * 'pnum' is set to the number of sectors (including and immediately following
+ *  the specified sector) that are known to be in the same
+ *  allocated/unallocated state.
+ *
+ */
+static int coroutine_fn is_allocated_base(BlockDriverState *top,
+                                          BlockDriverState *base,
+                                          int64_t sector_num,
+                                          int nb_sectors, int *pnum)
+{
+    BlockDriverState *intermediate;
+    int ret, n;
+
+    ret = bdrv_co_is_allocated(top, sector_num, nb_sectors, &n);
+    if (ret) {
+        *pnum = n;
+        return ret;
+    }
+
+    /*
+     * Is the unallocated chunk [sector_num, n] also
+     * unallocated between base and top?
+     */
+    intermediate = top->backing_hd;
+
+    while (intermediate) {
+        int pnum_inter;
+
+        /* reached base */
+        if (intermediate == base) {
+            *pnum = n;
+            return 1;
+        }
+        ret = bdrv_co_is_allocated(intermediate, sector_num, nb_sectors,
+                                   &pnum_inter);
+        if (ret < 0) {
+            return ret;
+        } else if (ret) {
+            *pnum = pnum_inter;
+            return 0;
+        }
+
+        /*
+         * [sector_num, nb_sectors] is unallocated on top but intermediate
+         * might have
+         *
+         * [sector_num+x, nr_sectors] allocated.
+         */
+        if (n > pnum_inter) {
+            n = pnum_inter;
+        }
+
+        intermediate = intermediate->backing_hd;
+    }
+
+    return 1;
+}
+
 static void coroutine_fn stream_run(void *opaque)
 {
     StreamBlockJob *s = opaque;
     BlockDriverState *bs = s->common.bs;
+    BlockDriverState *base = s->base;
     int64_t sector_num, end;
     int ret = 0;
     int n;
@@ -101,8 +168,15 @@ retry:
             break;
         }
 
-        ret = bdrv_co_is_allocated(bs, sector_num,
-                                   STREAM_BUFFER_SIZE / BDRV_SECTOR_SIZE, &n);
+
+        if (base) {
+            ret = is_allocated_base(bs, base, sector_num,
+                                    STREAM_BUFFER_SIZE / BDRV_SECTOR_SIZE, &n);
+        } else {
+            ret = bdrv_co_is_allocated(bs, sector_num,
+                                       STREAM_BUFFER_SIZE / BDRV_SECTOR_SIZE,
+                                       &n);
+        }
         trace_stream_one_iteration(s, sector_num, n, ret);
         if (ret == 0) {
             if (s->common.speed) {
@@ -119,6 +193,7 @@ retry:
         if (ret < 0) {
             break;
         }
+        ret = 0;
 
         /* Publish progress */
         s->common.offset += n * BDRV_SECTOR_SIZE;
@@ -132,7 +207,11 @@ retry:
     bdrv_disable_copy_on_read(bs);
 
     if (sector_num == end && ret == 0) {
-        ret = bdrv_change_backing_file(bs, NULL, NULL);
+        const char *base_id = NULL;
+        if (base) {
+            base_id = s->backing_file_id;
+        }
+        ret = bdrv_change_backing_file(bs, base_id, NULL);
     }
 
     qemu_vfree(buf);
@@ -158,7 +237,8 @@ static BlockJobType stream_job_type = {
 };
 
 int stream_start(BlockDriverState *bs, BlockDriverState *base,
-                 BlockDriverCompletionFunc *cb, void *opaque)
+                 const char *base_id, BlockDriverCompletionFunc *cb,
+                 void *opaque)
 {
     StreamBlockJob *s;
     Coroutine *co;
@@ -169,6 +249,9 @@ int stream_start(BlockDriverState *bs, BlockDriverState *base,
     }
 
     s->base = base;
+    if (base_id) {
+        pstrcpy(s->backing_file_id, sizeof(s->backing_file_id), base_id);
+    }
 
     co = qemu_coroutine_create(stream_run);
     trace_stream_start(bs, base, s, co, opaque);
diff --git a/block_int.h b/block_int.h
index c7c9178..ed92884 100644
--- a/block_int.h
+++ b/block_int.h
@@ -333,6 +333,7 @@ void block_job_cancel(BlockJob *job);
 bool block_job_is_cancelled(BlockJob *job);
 
 int stream_start(BlockDriverState *bs, BlockDriverState *base,
-                 BlockDriverCompletionFunc *cb, void *opaque);
+                 const char *base_id, BlockDriverCompletionFunc *cb,
+                 void *opaque);
 
 #endif /* BLOCK_INT_H */
diff --git a/blockdev.c b/blockdev.c
index 45a6ba6..ed3002d 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -956,6 +956,7 @@ void qmp_block_stream(const char *device, bool has_base,
                       const char *base, Error **errp)
 {
     BlockDriverState *bs;
+    BlockDriverState *base_bs = NULL;
     int ret;
 
     bs = bdrv_find(device);
@@ -964,13 +965,15 @@ void qmp_block_stream(const char *device, bool has_base,
         return;
     }
 
-    /* Base device not supported */
     if (base) {
-        error_set(errp, QERR_NOT_SUPPORTED);
-        return;
+        base_bs = bdrv_find_backing_image(bs, base);
+        if (base_bs == NULL) {
+            error_set(errp, QERR_BASE_NOT_FOUND, base);
+            return;
+        }
     }
 
-    ret = stream_start(bs, NULL, block_stream_cb, bs);
+    ret = stream_start(bs, base_bs, base, block_stream_cb, bs);
     if (ret < 0) {
         switch (ret) {
         case -EBUSY:
-- 
1.7.7.3

Re: [Qemu-devel] [PATCH v5 13/15] block: add support for partial streaming

[Kevin Wolf]

Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
> From: Marcelo Tosatti <mtosatti@redhat.com>
> 
> Add support for streaming data from an intermediate section of the
> image chain (see patch and documentation for details).
> 
> Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>

I'm afraid that in the review for the previous version I couldn't see
the wood for the trees... This does limit the COR requests issued by
image streaming, but not those issued by the guest. Am I missing
something? This is not what we want, is it?

Kevin

Re: [Qemu-devel] [PATCH v5 13/15] block: add support for partial streaming

[Marcelo Tosatti]

On Tue, Jan 17, 2012 at 02:27:04PM +0100, Kevin Wolf wrote:
> Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
> > From: Marcelo Tosatti <mtosatti@redhat.com>
> > 
> > Add support for streaming data from an intermediate section of the
> > image chain (see patch and documentation for details).
> > 
> > Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
> > Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
> 
> I'm afraid that in the review for the previous version I couldn't see
> the wood for the trees... This does limit the COR requests issued by
> image streaming, but not those issued by the guest. Am I missing
> something? This is not what we want, is it?

What you mean "limit the COR requests"? 

bdrv_co_is_allocated_base (or its new name) relies on
bbdrv_co_is_allocated being synchronous.

Sectors are only allocated in the top image, and in that case the
situation regaring synchronicity is the same as without shared base
option, that is, the serialization in bdrv_aio_read/bdrv_aio_write level
is responsible for correctness.

Re: [Qemu-devel] [PATCH v5 13/15] block: add support for partial streaming

[Kevin Wolf]

Am 17.01.2012 14:50, schrieb Marcelo Tosatti:
> On Tue, Jan 17, 2012 at 02:27:04PM +0100, Kevin Wolf wrote:
>> Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
>>> From: Marcelo Tosatti <mtosatti@redhat.com>
>>>
>>> Add support for streaming data from an intermediate section of the
>>> image chain (see patch and documentation for details).
>>>
>>> Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
>>> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
>>
>> I'm afraid that in the review for the previous version I couldn't see
>> the wood for the trees... This does limit the COR requests issued by
>> image streaming, but not those issued by the guest. Am I missing
>> something? This is not what we want, is it?
> 
> What you mean "limit the COR requests"? 

base -> sn1 -> sn2

You only want to copy the content of sn1 into sn2 and keep base. The
streaming coroutine is doing the right thing because it checks
is_allocated_base. However, if it is the guest that reads some data from
base, COR copies it into sn2 even though it's in the common base file.

Maybe streaming shouldn't enable normal COR on images, but instead of
calling bdrv_co_read it could directly call bdrv_co_copy_on_readv().

Kevin

Re: [Qemu-devel] [PATCH v5 13/15] block: add support for partial streaming

[Marcelo Tosatti]

On Tue, Jan 17, 2012 at 03:05:29PM +0100, Kevin Wolf wrote:
> Am 17.01.2012 14:50, schrieb Marcelo Tosatti:
> > On Tue, Jan 17, 2012 at 02:27:04PM +0100, Kevin Wolf wrote:
> >> Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
> >>> From: Marcelo Tosatti <mtosatti@redhat.com>
> >>>
> >>> Add support for streaming data from an intermediate section of the
> >>> image chain (see patch and documentation for details).
> >>>
> >>> Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
> >>> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
> >>
> >> I'm afraid that in the review for the previous version I couldn't see
> >> the wood for the trees... This does limit the COR requests issued by
> >> image streaming, but not those issued by the guest. Am I missing
> >> something? This is not what we want, is it?
> > 
> > What you mean "limit the COR requests"? 
> 
> base -> sn1 -> sn2
> 
> You only want to copy the content of sn1 into sn2 and keep base. The
> streaming coroutine is doing the right thing because it checks
> is_allocated_base. However, if it is the guest that reads some data from
> base, COR copies it into sn2 even though it's in the common base file.

Ah, yes.

> Maybe streaming shouldn't enable normal COR on images, but instead of
> calling bdrv_co_read it could directly call bdrv_co_copy_on_readv().

That would work.

Re: [Qemu-devel] [PATCH v5 13/15] block: add support for partial streaming

[Stefan Hajnoczi]

On Tue, Jan 17, 2012 at 3:47 PM, Marcelo Tosatti <mtosatti@redhat.com> wrote:
> On Tue, Jan 17, 2012 at 03:05:29PM +0100, Kevin Wolf wrote:
>> Am 17.01.2012 14:50, schrieb Marcelo Tosatti:
>> > On Tue, Jan 17, 2012 at 02:27:04PM +0100, Kevin Wolf wrote:
>> >> Am 13.01.2012 14:14, schrieb Stefan Hajnoczi:
>> >>> From: Marcelo Tosatti <mtosatti@redhat.com>
>> >>>
>> >>> Add support for streaming data from an intermediate section of the
>> >>> image chain (see patch and documentation for details).
>> >>>
>> >>> Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
>> >>> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
>> >>
>> >> I'm afraid that in the review for the previous version I couldn't see
>> >> the wood for the trees... This does limit the COR requests issued by
>> >> image streaming, but not those issued by the guest. Am I missing
>> >> something? This is not what we want, is it?
>> >
>> > What you mean "limit the COR requests"?
>>
>> base -> sn1 -> sn2
>>
>> You only want to copy the content of sn1 into sn2 and keep base. The
>> streaming coroutine is doing the right thing because it checks
>> is_allocated_base. However, if it is the guest that reads some data from
>> base, COR copies it into sn2 even though it's in the common base file.
>
> Ah, yes.
>
>> Maybe streaming shouldn't enable normal COR on images, but instead of
>> calling bdrv_co_read it could directly call bdrv_co_copy_on_readv().
>
> That would work.

Sounds like a good suggestion.  It will prevent the case where a guest
is doing heavy read I/O during image streaming with a 'base' and we
bloat the destination image file.

I'll resend the series with this fix.

Stefan

[Qemu-devel] [PATCH v5 14/15] docs: describe live block operations

[Stefan Hajnoczi]

From: Marcelo Tosatti <mtosatti@redhat.com>

Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 docs/live-block-ops.txt |   58 +++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 58 insertions(+), 0 deletions(-)
 create mode 100644 docs/live-block-ops.txt

diff --git a/docs/live-block-ops.txt b/docs/live-block-ops.txt
new file mode 100644
index 0000000..a257087
--- /dev/null
+++ b/docs/live-block-ops.txt
@@ -0,0 +1,58 @@
+LIVE BLOCK OPERATIONS
+=====================
+
+High level description of live block operations. Note these are not
+supported for use with the raw format at the moment.
+
+Snapshot live merge
+===================
+
+Given a snapshot chain, described in this document in the following
+format:
+
+[A] -> [B] -> [C] -> [D]
+
+Where the rightmost object ([D] in the example) described is the current
+image which the guest OS has write access to. To the left of it is its base
+image, and so on accordingly until the leftmost image, which has no
+base.
+
+The snapshot live merge operation transforms such a chain into a
+smaller one with fewer elements, such as this transformation relative
+to the first example:
+
+[A] -> [D]
+
+Currently only forward merge with target being the active image is
+supported, that is, data copy is performed in the right direction with
+destination being the rightmost image.
+
+The operation is implemented in QEMU through image streaming facilities.
+
+The basic idea is to execute 'block_stream virtio0' while the guest is
+running. Progress can be monitored using 'info block-jobs'. When the
+streaming operation completes it raises a QMP event. 'block_stream'
+copies data from the backing file(s) into the active image. When finished,
+it adjusts the backing file pointer.
+
+The 'base' parameter specifies an image which data need not be streamed from.
+This image will be used as the backing file for the active image when the
+operation is finished.
+
+In the example above, the command would be:
+
+(qemu) block_stream virtio0 A
+
+
+Live block copy
+===============
+
+To copy an in use image to another destination in the filesystem, one
+should create a live snapshot in the desired destination, then stream
+into that image. Example:
+
+(qemu) snapshot_blkdev ide0-hd0 /new-path/disk.img qcow2
+
+(qemu) block_stream ide0-hd0
+
+
-- 
1.7.7.3

[Qemu-devel] [PATCH v5 15/15] test: add image streaming test cases

[Stefan Hajnoczi]

python test-stream.py

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 test-stream.py |  208 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 208 insertions(+), 0 deletions(-)
 create mode 100644 test-stream.py

diff --git a/test-stream.py b/test-stream.py
new file mode 100644
index 0000000..16cbe5d
--- /dev/null
+++ b/test-stream.py
@@ -0,0 +1,208 @@
+#!/usr/bin/env python
+import unittest
+import subprocess
+import re
+import os
+import sys; sys.path.append('QMP/')
+import qmp
+
+def qemu_img(*args):
+    devnull = open('/dev/null', 'r+')
+    return subprocess.call(['./qemu-img'] + list(args), stdin=devnull, stdout=devnull)
+
+def qemu_io(*args):
+    args = ['./qemu-io'] + list(args)
+    return subprocess.Popen(args, stdout=subprocess.PIPE).communicate()[0]
+
+class VM(object):
+    def __init__(self):
+        self._monitor_path = '/tmp/qemu-mon.%d' % os.getpid()
+        self._qemu_log_path = '/tmp/qemu-log.%d' % os.getpid()
+        self._args = ['x86_64-softmmu/qemu-system-x86_64',
+                      '-chardev', 'socket,id=mon,path=' + self._monitor_path,
+                      '-mon', 'chardev=mon,mode=control', '-nographic']
+        self._num_drives = 0
+
+    def add_drive(self, path, opts=''):
+        options = ['if=virtio',
+                   'cache=none',
+                   'file=%s' % path,
+                   'id=drive%d' % self._num_drives]
+        if opts:
+            options.append(opts)
+
+        self._args.append('-drive')
+        self._args.append(','.join(options))
+        self._num_drives += 1
+        return self
+
+    def launch(self):
+        devnull = open('/dev/null', 'rb')
+        qemulog = open(self._qemu_log_path, 'wb')
+        self._qmp = qmp.QEMUMonitorProtocol(self._monitor_path, server=True)
+        self._popen = subprocess.Popen(self._args, stdin=devnull, stdout=qemulog,
+                                       stderr=subprocess.STDOUT)
+        self._qmp.accept()
+
+    def shutdown(self):
+        self._qmp.cmd('quit')
+        self._popen.wait()
+        os.remove(self._monitor_path)
+        os.remove(self._qemu_log_path)
+
+    def qmp(self, cmd, **args):
+        return self._qmp.cmd(cmd, args=args)
+
+    def get_qmp_events(self, wait=False):
+        events = self._qmp.get_events(wait=wait)
+        self._qmp.clear_events()
+        return events
+
+index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
+
+class QMPTestCase(unittest.TestCase):
+    def dictpath(self, d, path):
+        """Traverse a path in a nested dict"""
+        for component in path.split('/'):
+            m = index_re.match(component)
+            if m:
+                component, idx = m.groups()
+                idx = int(idx)
+
+            if not isinstance(d, dict) or component not in d:
+                self.fail('failed path traversal for "%s" in "%s"' % (path, str(d)))
+            d = d[component]
+
+            if m:
+                if not isinstance(d, list):
+                    self.fail('path component "%s" in "%s" is not a list in "%s"' % (component, path, str(d)))
+                try:
+                    d = d[idx]
+                except IndexError:
+                    self.fail('invalid index "%s" in path "%s" in "%s"' % (idx, path, str(d)))
+        return d
+
+    def assert_qmp(self, d, path, value):
+        result = self.dictpath(d, path)
+        self.assertEqual(result, value, 'values not equal "%s" and "%s"' % (str(result), str(value)))
+
+    def assert_no_active_streams(self):
+        result = self.vm.qmp('query-block-jobs')
+        self.assert_qmp(result, 'return', [])
+
+class TestSingleDrive(QMPTestCase):
+    image_len = 1 * 1024 * 1024 # MB
+
+    def setUp(self):
+        qemu_img('create', 'backing.img', str(TestSingleDrive.image_len))
+        qemu_img('create', '-f', 'qed', '-o', 'backing_file=backing.img', 'test.qed')
+        self.vm = VM().add_drive('test.qed')
+        self.vm.launch()
+
+    def tearDown(self):
+        self.vm.shutdown()
+        os.remove('test.qed')
+        os.remove('backing.img')
+
+    def test_stream(self):
+        self.assert_no_active_streams()
+
+        result = self.vm.qmp('block_stream', device='drive0')
+        self.assert_qmp(result, 'return', {})
+
+        completed = False
+        while not completed:
+            for event in self.vm.get_qmp_events(wait=True):
+                if event['event'] == 'BLOCK_JOB_COMPLETED':
+                    self.assert_qmp(event, 'data/type', 'stream')
+                    self.assert_qmp(event, 'data/device', 'drive0')
+                    self.assert_qmp(event, 'data/offset', self.image_len)
+                    self.assert_qmp(event, 'data/len', self.image_len)
+                    completed = True
+
+        self.assert_no_active_streams()
+
+        self.assertFalse('sectors not allocated' in qemu_io('-c', 'map', 'test.qed'),
+                         'image file not fully populated after streaming')
+
+    def test_device_not_found(self):
+        result = self.vm.qmp('block_stream', device='nonexistent')
+        self.assert_qmp(result, 'error/class', 'DeviceNotFound')
+
+class TestStreamStop(QMPTestCase):
+    image_len = 8 * 1024 * 1024 * 1024 # GB
+
+    def setUp(self):
+        qemu_img('create', 'backing.img', str(TestStreamStop.image_len))
+        qemu_img('create', '-f', 'qed', '-o', 'backing_file=backing.img', 'test.qed')
+        self.vm = VM().add_drive('test.qed')
+        self.vm.launch()
+
+    def tearDown(self):
+        self.vm.shutdown()
+        os.remove('test.qed')
+        os.remove('backing.img')
+
+    def test_stream_stop(self):
+        import time
+
+        self.assert_no_active_streams()
+
+        result = self.vm.qmp('block_stream', device='drive0')
+        self.assert_qmp(result, 'return', {})
+
+        time.sleep(1)
+        events = self.vm.get_qmp_events(wait=False)
+        self.assertEqual(events, [], 'unexpected QMP event: %s' % events)
+
+        self.vm.qmp('block_job_cancel', device='drive0')
+        self.assert_qmp(result, 'return', {})
+
+        cancelled = False
+        while not cancelled:
+            for event in self.vm.get_qmp_events(wait=True):
+                if event['event'] == 'BLOCK_JOB_CANCELLED':
+                    self.assert_qmp(event, 'data/type', 'stream')
+                    self.assert_qmp(event, 'data/device', 'drive0')
+                    cancelled = True
+
+        self.assert_no_active_streams()
+
+class TestSetSpeed(QMPTestCase):
+    image_len = 80 * 1024 * 1024 # MB
+
+    def setUp(self):
+        qemu_img('create', 'backing.img', str(TestSetSpeed.image_len))
+        qemu_img('create', '-f', 'qed', '-o', 'backing_file=backing.img', 'test.qed')
+        self.vm = VM().add_drive('test.qed')
+        self.vm.launch()
+
+    def tearDown(self):
+        self.vm.shutdown()
+        os.remove('test.qed')
+        os.remove('backing.img')
+
+    # This doesn't print or verify anything, only use it via "test-stream.py TestSetSpeed"
+    def test_stream_set_speed(self):
+        self.assert_no_active_streams()
+
+        result = self.vm.qmp('block_stream', device='drive0')
+        self.assert_qmp(result, 'return', {})
+
+        result = self.vm.qmp('block_job_set_speed', device='drive0', value=8 * 1024 * 1024)
+        self.assert_qmp(result, 'return', {})
+
+        completed = False
+        while not completed:
+            for event in self.vm.get_qmp_events(wait=True):
+                if event['event'] == 'BLOCK_JOB_COMPLETED':
+                    self.assert_qmp(event, 'data/type', 'stream')
+                    self.assert_qmp(event, 'data/device', 'drive0')
+                    self.assert_qmp(event, 'data/offset', self.image_len)
+                    self.assert_qmp(event, 'data/len', self.image_len)
+                    completed = True
+
+        self.assert_no_active_streams()
+
+if __name__ == '__main__':
+    unittest.main()
-- 
1.7.7.3

Re: [Qemu-devel] [PATCH v5 15/15] test: add image streaming test cases

[Stefan Hajnoczi]

Hi Lucas,
The Python script below verifies the image streaming feature.  It's
built on the standard library "unittest" module, as well as QEMU's
qmp.py module.  It spawns a qemu process and creates necessary disk
image files.  The tests themselves issue QMP commands and check their
result or wait for QMP events to be raised.

I think this sort of test could be done with kvm-autotest but I don't
see much usage of cmd_qmp() in client/tests/kvm/tests/.  I'm curious
how you would approach this.  The high-level steps are:

1. Create a backing file.
2. Create a test QED image file using the backing file.
3. Issue "block_stream device=drive0" to the running VM.  This should
return no value.
4. Wait for the BLOCK_JOB_COMPLETED QMP event and check its fields -
they must contain expected values.
5. Ensure "query-block-job" does not show any active jobs anymore.
6. Use qemu-io's map command to verify that the image stays compact
and isn't bloated with actual zero bytes (this is kind of unrelated to
the rest of the test).

The other test cases share much of the same building blocks as
TestSingleDrive, so they are less interesting.

Would it be possible to look at TestSingleDrive below and give a
kvm-autotest equivalent?

Thanks,
Stefan

On Fri, Jan 13, 2012 at 1:14 PM, Stefan Hajnoczi
<stefanha@linux.vnet.ibm.com> wrote:
> python test-stream.py
>
> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
> ---
>  test-stream.py |  208 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>  1 files changed, 208 insertions(+), 0 deletions(-)
>  create mode 100644 test-stream.py
>
> diff --git a/test-stream.py b/test-stream.py
> new file mode 100644
> index 0000000..16cbe5d
> --- /dev/null
> +++ b/test-stream.py
> @@ -0,0 +1,208 @@
> +#!/usr/bin/env python
> +import unittest
> +import subprocess
> +import re
> +import os
> +import sys; sys.path.append('QMP/')
> +import qmp
> +
> +def qemu_img(*args):
> +    devnull = open('/dev/null', 'r+')
> +    return subprocess.call(['./qemu-img'] + list(args), stdin=devnull, stdout=devnull)
> +
> +def qemu_io(*args):
> +    args = ['./qemu-io'] + list(args)
> +    return subprocess.Popen(args, stdout=subprocess.PIPE).communicate()[0]
> +
> +class VM(object):
> +    def __init__(self):
> +        self._monitor_path = '/tmp/qemu-mon.%d' % os.getpid()
> +        self._qemu_log_path = '/tmp/qemu-log.%d' % os.getpid()
> +        self._args = ['x86_64-softmmu/qemu-system-x86_64',
> +                      '-chardev', 'socket,id=mon,path=' + self._monitor_path,
> +                      '-mon', 'chardev=mon,mode=control', '-nographic']
> +        self._num_drives = 0
> +
> +    def add_drive(self, path, opts=''):
> +        options = ['if=virtio',
> +                   'cache=none',
> +                   'file=%s' % path,
> +                   'id=drive%d' % self._num_drives]
> +        if opts:
> +            options.append(opts)
> +
> +        self._args.append('-drive')
> +        self._args.append(','.join(options))
> +        self._num_drives += 1
> +        return self
> +
> +    def launch(self):
> +        devnull = open('/dev/null', 'rb')
> +        qemulog = open(self._qemu_log_path, 'wb')
> +        self._qmp = qmp.QEMUMonitorProtocol(self._monitor_path, server=True)
> +        self._popen = subprocess.Popen(self._args, stdin=devnull, stdout=qemulog,
> +                                       stderr=subprocess.STDOUT)
> +        self._qmp.accept()
> +
> +    def shutdown(self):
> +        self._qmp.cmd('quit')
> +        self._popen.wait()
> +        os.remove(self._monitor_path)
> +        os.remove(self._qemu_log_path)
> +
> +    def qmp(self, cmd, **args):
> +        return self._qmp.cmd(cmd, args=args)
> +
> +    def get_qmp_events(self, wait=False):
> +        events = self._qmp.get_events(wait=wait)
> +        self._qmp.clear_events()
> +        return events
> +
> +index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
> +
> +class QMPTestCase(unittest.TestCase):
> +    def dictpath(self, d, path):
> +        """Traverse a path in a nested dict"""
> +        for component in path.split('/'):
> +            m = index_re.match(component)
> +            if m:
> +                component, idx = m.groups()
> +                idx = int(idx)
> +
> +            if not isinstance(d, dict) or component not in d:
> +                self.fail('failed path traversal for "%s" in "%s"' % (path, str(d)))
> +            d = d[component]
> +
> +            if m:
> +                if not isinstance(d, list):
> +                    self.fail('path component "%s" in "%s" is not a list in "%s"' % (component, path, str(d)))
> +                try:
> +                    d = d[idx]
> +                except IndexError:
> +                    self.fail('invalid index "%s" in path "%s" in "%s"' % (idx, path, str(d)))
> +        return d
> +
> +    def assert_qmp(self, d, path, value):
> +        result = self.dictpath(d, path)
> +        self.assertEqual(result, value, 'values not equal "%s" and "%s"' % (str(result), str(value)))
> +
> +    def assert_no_active_streams(self):
> +        result = self.vm.qmp('query-block-jobs')
> +        self.assert_qmp(result, 'return', [])
> +
> +class TestSingleDrive(QMPTestCase):
> +    image_len = 1 * 1024 * 1024 # MB
> +
> +    def setUp(self):
> +        qemu_img('create', 'backing.img', str(TestSingleDrive.image_len))
> +        qemu_img('create', '-f', 'qed', '-o', 'backing_file=backing.img', 'test.qed')
> +        self.vm = VM().add_drive('test.qed')
> +        self.vm.launch()
> +
> +    def tearDown(self):
> +        self.vm.shutdown()
> +        os.remove('test.qed')
> +        os.remove('backing.img')
> +
> +    def test_stream(self):
> +        self.assert_no_active_streams()
> +
> +        result = self.vm.qmp('block_stream', device='drive0')
> +        self.assert_qmp(result, 'return', {})
> +
> +        completed = False
> +        while not completed:
> +            for event in self.vm.get_qmp_events(wait=True):
> +                if event['event'] == 'BLOCK_JOB_COMPLETED':
> +                    self.assert_qmp(event, 'data/type', 'stream')
> +                    self.assert_qmp(event, 'data/device', 'drive0')
> +                    self.assert_qmp(event, 'data/offset', self.image_len)
> +                    self.assert_qmp(event, 'data/len', self.image_len)
> +                    completed = True
> +
> +        self.assert_no_active_streams()
> +
> +        self.assertFalse('sectors not allocated' in qemu_io('-c', 'map', 'test.qed'),
> +                         'image file not fully populated after streaming')
> +
> +    def test_device_not_found(self):
> +        result = self.vm.qmp('block_stream', device='nonexistent')
> +        self.assert_qmp(result, 'error/class', 'DeviceNotFound')
> +
> +class TestStreamStop(QMPTestCase):
> +    image_len = 8 * 1024 * 1024 * 1024 # GB
> +
> +    def setUp(self):
> +        qemu_img('create', 'backing.img', str(TestStreamStop.image_len))
> +        qemu_img('create', '-f', 'qed', '-o', 'backing_file=backing.img', 'test.qed')
> +        self.vm = VM().add_drive('test.qed')
> +        self.vm.launch()
> +
> +    def tearDown(self):
> +        self.vm.shutdown()
> +        os.remove('test.qed')
> +        os.remove('backing.img')
> +
> +    def test_stream_stop(self):
> +        import time
> +
> +        self.assert_no_active_streams()
> +
> +        result = self.vm.qmp('block_stream', device='drive0')
> +        self.assert_qmp(result, 'return', {})
> +
> +        time.sleep(1)
> +        events = self.vm.get_qmp_events(wait=False)
> +        self.assertEqual(events, [], 'unexpected QMP event: %s' % events)
> +
> +        self.vm.qmp('block_job_cancel', device='drive0')
> +        self.assert_qmp(result, 'return', {})
> +
> +        cancelled = False
> +        while not cancelled:
> +            for event in self.vm.get_qmp_events(wait=True):
> +                if event['event'] == 'BLOCK_JOB_CANCELLED':
> +                    self.assert_qmp(event, 'data/type', 'stream')
> +                    self.assert_qmp(event, 'data/device', 'drive0')
> +                    cancelled = True
> +
> +        self.assert_no_active_streams()
> +
> +class TestSetSpeed(QMPTestCase):
> +    image_len = 80 * 1024 * 1024 # MB
> +
> +    def setUp(self):
> +        qemu_img('create', 'backing.img', str(TestSetSpeed.image_len))
> +        qemu_img('create', '-f', 'qed', '-o', 'backing_file=backing.img', 'test.qed')
> +        self.vm = VM().add_drive('test.qed')
> +        self.vm.launch()
> +
> +    def tearDown(self):
> +        self.vm.shutdown()
> +        os.remove('test.qed')
> +        os.remove('backing.img')
> +
> +    # This doesn't print or verify anything, only use it via "test-stream.py TestSetSpeed"
> +    def test_stream_set_speed(self):
> +        self.assert_no_active_streams()
> +
> +        result = self.vm.qmp('block_stream', device='drive0')
> +        self.assert_qmp(result, 'return', {})
> +
> +        result = self.vm.qmp('block_job_set_speed', device='drive0', value=8 * 1024 * 1024)
> +        self.assert_qmp(result, 'return', {})
> +
> +        completed = False
> +        while not completed:
> +            for event in self.vm.get_qmp_events(wait=True):
> +                if event['event'] == 'BLOCK_JOB_COMPLETED':
> +                    self.assert_qmp(event, 'data/type', 'stream')
> +                    self.assert_qmp(event, 'data/device', 'drive0')
> +                    self.assert_qmp(event, 'data/offset', self.image_len)
> +                    self.assert_qmp(event, 'data/len', self.image_len)
> +                    completed = True
> +
> +        self.assert_no_active_streams()
> +
> +if __name__ == '__main__':
> +    unittest.main()
> --
> 1.7.7.3
>
>

Re: [Qemu-devel] [PATCH v5 15/15] test: add image streaming test cases

[Lucas Meneghel Rodrigues]

On 01/13/2012 02:49 PM, Stefan Hajnoczi wrote:
> Hi Lucas,
> The Python script below verifies the image streaming feature.  It's
> built on the standard library "unittest" module, as well as QEMU's
> qmp.py module.  It spawns a qemu process and creates necessary disk
> image files.  The tests themselves issue QMP commands and check their
> result or wait for QMP events to be raised.
>
> I think this sort of test could be done with kvm-autotest but I don't
> see much usage of cmd_qmp() in client/tests/kvm/tests/.  I'm curious
> how you would approach this.  The high-level steps are:
>
> 1. Create a backing file.
> 2. Create a test QED image file using the backing file.
> 3. Issue "block_stream device=drive0" to the running VM.  This should
> return no value.
> 4. Wait for the BLOCK_JOB_COMPLETED QMP event and check its fields -
> they must contain expected values.
> 5. Ensure "query-block-job" does not show any active jobs anymore.
> 6. Use qemu-io's map command to verify that the image stays compact
> and isn't bloated with actual zero bytes (this is kind of unrelated to
> the rest of the test).
>
> The other test cases share much of the same building blocks as
> TestSingleDrive, so they are less interesting.
>
> Would it be possible to look at TestSingleDrive below and give a
> kvm-autotest equivalent?

Yes Stefan, sorry for the late reply. I was in FUDCon, therefore taking 
care of some Fedora related autotest stuff, but I'm putting on my todo 
list to create a KVM autotest equivalent of it.

Cheers,

Lucas

Re: [Qemu-devel] [PATCH v5 15/15] test: add image streaming test cases

[Stefan Hajnoczi]

On Tue, Jan 17, 2012 at 7:07 PM, Lucas Meneghel Rodrigues
<lmr@redhat.com> wrote:
> On 01/13/2012 02:49 PM, Stefan Hajnoczi wrote:
>>
>> Hi Lucas,
>> The Python script below verifies the image streaming feature.  It's
>> built on the standard library "unittest" module, as well as QEMU's
>> qmp.py module.  It spawns a qemu process and creates necessary disk
>> image files.  The tests themselves issue QMP commands and check their
>> result or wait for QMP events to be raised.
>>
>> I think this sort of test could be done with kvm-autotest but I don't
>> see much usage of cmd_qmp() in client/tests/kvm/tests/.  I'm curious
>> how you would approach this.  The high-level steps are:
>>
>> 1. Create a backing file.
>> 2. Create a test QED image file using the backing file.
>> 3. Issue "block_stream device=drive0" to the running VM.  This should
>> return no value.
>> 4. Wait for the BLOCK_JOB_COMPLETED QMP event and check its fields -
>> they must contain expected values.
>> 5. Ensure "query-block-job" does not show any active jobs anymore.
>> 6. Use qemu-io's map command to verify that the image stays compact
>> and isn't bloated with actual zero bytes (this is kind of unrelated to
>> the rest of the test).
>>
>> The other test cases share much of the same building blocks as
>> TestSingleDrive, so they are less interesting.
>>
>> Would it be possible to look at TestSingleDrive below and give a
>> kvm-autotest equivalent?
>
>
> Yes Stefan, sorry for the late reply. I was in FUDCon, therefore taking care
> of some Fedora related autotest stuff, but I'm putting on my todo list to
> create a KVM autotest equivalent of it.

Great, thank you!

Stefan

Re: [Qemu-devel] [PATCH v5 00/15] block: generic image streaming

[Luiz Capitulino]

On Fri, 13 Jan 2012 13:14:02 +0000
Stefan Hajnoczi <stefanha@linux.vnet.ibm.com> wrote:

> Note: This is a resend of v5 because the emails I sent earlier today
> disappeared.
> 
> This series adds the 'block_stream' command which copies the contents of a
> backing file into the image file while the VM is running.  The series builds on
> the zero detection features which I sent out before Christmas. I suggest
> grabbing my git tree to try it out without merging this dependency:
> 
> https://github.com/stefanha/qemu/tree/image-streaming-api
> 
> The image streaming HMP/QMP commands are documented in the patch and also
> described here:
> 
> http://wiki.qemu.org/Features/LiveBlockMigration/ImageStreamingAPI

As far as QMP is concerned:

Acked-by: Luiz Capitulino <lcapitulino@redhat.com>

> 
> The basic idea is to execute 'block_stream virtio0' while the guest is running.
> Progress can be monitored using 'info block-jobs'.  When the streaming
> operation completes it raises a QMP event.
> 
> Note: The last patch includes includes a Python test script called
> test-stream.py, I do not propose to merge it.  When run in a QEMU source tree
> it performs basic image streaming QMP tests.
> 
> v5:
>  * Handle block_job_create() failure [Luiz]
>  * Mark BLOCK_JOB_COMPLETED error field optional [Luiz]
>  * Mark block_stream base parameter optional [Luiz]
>  * Check bdrv_getlength() failure and don't call twice [Kevin]
>  * Rename "id" to "backing_file" in bdrv_find_backing_image() [Kevin]
>  * Rename BaseIdNotFound qerror to BaseNotFound [Kevin]
>  * Document BaseNotFound qerror from block_stream
>  * Document that qemu_co_sleep_ns() needs main loop [Kevin]
>  * Make bdrv_co_is_allocated_base() private to block/stream.c [Kevin]
>  * Clean up commit messages
> 
> v4:
>  * Drop SQMP/EQMP docs from qmp-commands.hx [Luiz]
>  * Follow QAPI doc conventions [Luiz]
>  * Document QMP events in QMP/qmp-events.txt [Luiz]
>  * Protect against hotplug, resize, eject, etc [Kevin]
>  * Move block job functions from header to block.c [Kevin]
>  * Return error from bdrg_change_backing_file() [Kevin]
>  * Merge Marcelo's block_stream base partial streaming series [Marcelo]
> 
> Marcelo Tosatti (4):
>   block: add bdrv_find_backing_image
>   add QERR_BASE_NOT_FOUND
>   block: add support for partial streaming
>   docs: describe live block operations
> 
> Stefan Hajnoczi (11):
>   coroutine: add co_sleep_ns() coroutine sleep function
>   block: check bdrv_in_use() before blockdev operations
>   block: add BlockJob interface for long-running operations
>   block: add image streaming block job
>   block: rate-limit streaming operations
>   qmp: add block_stream command
>   qmp: add block_job_set_speed command
>   qmp: add block_job_cancel command
>   qmp: add query-block-jobs
>   blockdev: make image streaming safe across hotplug
>   test: add image streaming test cases
> 
>  Makefile.objs           |    2 +
>  QMP/qmp-events.txt      |   53 ++++++++++
>  block.c                 |   70 +++++++++++++
>  block.h                 |    2 +
>  block/stream.c          |  260 +++++++++++++++++++++++++++++++++++++++++++++++
>  block_int.h             |   44 ++++++++
>  blockdev.c              |  199 +++++++++++++++++++++++++++++++++++-
>  docs/live-block-ops.txt |   58 +++++++++++
>  hmp-commands.hx         |   41 ++++++++
>  hmp.c                   |   68 ++++++++++++
>  hmp.h                   |    4 +
>  monitor.c               |   13 +++
>  monitor.h               |    2 +
>  qapi-schema.json        |  116 +++++++++++++++++++++
>  qemu-coroutine-sleep.c  |   38 +++++++
>  qemu-coroutine.h        |    9 ++
>  qerror.c                |    8 ++
>  qerror.h                |    6 +
>  qmp-commands.hx         |   24 +++++
>  test-stream.py          |  208 +++++++++++++++++++++++++++++++++++++
>  trace-events            |    9 ++
>  21 files changed, 1233 insertions(+), 1 deletions(-)
>  create mode 100644 block/stream.c
>  create mode 100644 docs/live-block-ops.txt
>  create mode 100644 qemu-coroutine-sleep.c
>  create mode 100644 test-stream.py
>