Re: [Qemu-devel] [PATCH 08/15] qmp: add block_stream command

[Marcelo Tosatti]

On Wed, Jul 27, 2011 at 02:44:48PM +0100, Stefan Hajnoczi wrote:
> For leaf images with copy-on-read semantics, the stream command allows
> the user to populate the image file by copying data from the backing
> file while the guest is running.  Once all blocks have been streamed,
> the dependency on the original backing file is removed.  Therefore,
> stream commands can be used to implement post-copy live block migration
> and rapid deployment.
> 
> The command synopsis is:
> 
> 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.
> 
> Arguments:
> 
> - device: device name (json-string)
> - base:   common backing file (json-string, optional)
> 
> Errors:
> 
> DeviceInUse:    streaming is already active on this device
> DeviceNotFound: device name is invalid
> NotSupported:   image streaming is not supported by this device
> 
> Events:
> 
> On completion the BLOCK_JOB_COMPLETED event is raised with the following
> fields:
> 
> - type:     job type ("stream" for image streaming, json-string)
> - device:   device name (json-string)
> - end:      maximum progress value (json-int)
> - position: current progress value (json-int)
> - speed:    rate limit, bytes per second (json-int)
> - error:    error message (json-string, only on error)
> 
> The completion event is raised both on success and on failure.  On
> success position is equal to end.  On failure position and end can be
> used to indicate at which point the operation failed.
> 
> On failure the error 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.
> 
> Examples:
> 
> -> { "execute": "block_stream", "arguments": { "device": "virtio0" } }
> <- { "return":  {} }
> 
> Signed-off-by: Adam Litke <address@hidden>
> Signed-off-by: Stefan Hajnoczi <address@hidden>
> ---
>  blockdev.c      |  133 
> +++++++++++++++++++++++++++++++++++++++++++++++++++++++
>  blockdev.h      |    1 +
>  hmp-commands.hx |   14 ++++++
>  monitor.c       |    3 +
>  monitor.h       |    1 +
>  qerror.c        |    8 +++
>  qerror.h        |    6 +++
>  qmp-commands.hx |   64 ++++++++++++++++++++++++++
>  8 files changed, 230 insertions(+), 0 deletions(-)
> 
> diff --git a/blockdev.c b/blockdev.c
> index b337732..cd5e49c 100644
> --- a/blockdev.c
> +++ b/blockdev.c
> @@ -16,6 +16,7 @@
>  #include "sysemu.h"
>  #include "hw/qdev.h"
>  #include "block_int.h"
> +#include "qjson.h"
>  
>  static QTAILQ_HEAD(drivelist, DriveInfo) drives = 
> QTAILQ_HEAD_INITIALIZER(drives);
>  
> @@ -50,6 +51,131 @@ static const int if_max_devs[IF_COUNT] = {
>      [IF_SCSI] = 7,
>  };
>  
> +typedef struct StreamState {
> +    int64_t offset;             /* current position in block device */
> +    BlockDriverState *bs;
> +    QEMUTimer *timer;
> +    QLIST_ENTRY(StreamState) list;
> +} StreamState;
> +
> +static QLIST_HEAD(, StreamState) block_streams =
> +    QLIST_HEAD_INITIALIZER(block_streams);
> +
> +static QObject *stream_get_qobject(StreamState *s)
> +{
> +    const char *name = bdrv_get_device_name(s->bs);
> +    int64_t len = bdrv_getlength(s->bs);
> +
> +    return qobject_from_jsonf("{ 'device': %s, 'type': 'stream', "
> +                              "'offset': %" PRId64 ", 'len': %" PRId64 ", "
> +                              "'speed': %" PRId64 " }",
> +                              name, s->offset, len, (int64_t)0);
> +}
> +
> +static void stream_mon_event(StreamState *s, int ret)
> +{
> +    QObject *data = stream_get_qobject(s);
> +
> +    if (ret < 0) {
> +        QDict *qdict = qobject_to_qdict(data);
> +
> +        qdict_put(qdict, "error", qstring_from_str(strerror(-ret)));
> +    }
> +
> +    monitor_protocol_event(QEVENT_BLOCK_JOB_COMPLETED, data);
> +    qobject_decref(data);
> +}
> +
> +static void stream_free(StreamState *s)
> +{
> +    QLIST_REMOVE(s, list);
> +
> +    qemu_del_timer(s->timer);
> +    qemu_free_timer(s->timer);
> +    qemu_free(s);
> +}
> +
> +static void stream_complete(StreamState *s, int ret)
> +{
> +    stream_mon_event(s, ret);
> +    stream_free(s);
> +}
> +
> +static void stream_cb(void *opaque, int nb_sectors)
> +{
> +    StreamState *s = opaque;
> +
> +    if (nb_sectors < 0) {
> +        stream_complete(s, nb_sectors);
> +        return;
> +    }
> +
> +    s->offset += nb_sectors * BDRV_SECTOR_SIZE;
> +
> +    if (s->offset == bdrv_getlength(s->bs)) {
> +        bdrv_change_backing_file(s->bs, NULL, NULL);
> +        stream_complete(s, 0);
> +    } else {
> +        qemu_mod_timer(s->timer, qemu_get_clock_ns(rt_clock));
> +    }
> +}
> +
> +/* We can't call bdrv_aio_stream() directly from the callback because that
> + * makes qemu_aio_flush() not complete until the streaming is completed.
> + * By delaying with a timer, we give qemu_aio_flush() a chance to complete.
> + */
> +static void stream_next_iteration(void *opaque)
> +{
> +    StreamState *s = opaque;
> +
> +    bdrv_aio_copy_backing(s->bs, s->offset / BDRV_SECTOR_SIZE, stream_cb, s);
> +}

The plan is to replace format specific code with the generic
implementation in the future?

> +
> +static StreamState *stream_find(const char *device)
> +{
> +    StreamState *s;
> +
> +    QLIST_FOREACH(s, &block_streams, list) {
> +        if (strcmp(bdrv_get_device_name(s->bs), device) == 0) {
> +            return s;
> +        }
> +    }
> +    return NULL;
> +}
> +
> +static StreamState *stream_start(const char *device)
> +{
> +    StreamState *s;
> +    BlockDriverAIOCB *acb;
> +    BlockDriverState *bs;
> +
> +    s = stream_find(device);
> +    if (s) {
> +        qerror_report(QERR_DEVICE_IN_USE, device);
> +        return NULL;
> +    }
> +
> +    bs = bdrv_find(device);
> +    if (!bs) {
> +        qerror_report(QERR_DEVICE_NOT_FOUND, device);
> +        return NULL;
> +    }
> +
> +    s = qemu_mallocz(sizeof(*s));
> +    s->bs = bs;
> +    s->timer = qemu_new_timer_ns(rt_clock, stream_next_iteration, s);
> +    QLIST_INSERT_HEAD(&block_streams, s, list);

Should increase refcount with drive_get_ref().