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

[Stefan Hajnoczi]

On Thu, Jul 28, 2011 at 4:53 PM, Marcelo Tosatti <address@hidden> wrote:
> 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?

I think Kevin has said this will not be merged into qemu.git.  But I
am sharing it as a reference implementation against which the libvirt
API works.

Next I will send out a stub implementation of the QMP/HMP block_stream
APIs without any of the block driver-specific functionality.  It may
make sense to merge this into QEMU just to nail down the QMP/HMP
interface, even if it does not do anything yet (we have errors for
ENOTSUP).

>> +
>> +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().

Yes thanks.

Stefan