[2/7] Add blkmirror block driver

Mirrored writes are used by live block copy.

Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>

On Tue, May 24, 2011 at 12:31 AM, Marcelo Tosatti <mtosatti@redhat.com> wrote:
> Mirrored writes are used by live block copy.
>
> Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
>
> Index: qemu-block-copy/block/blkmirror.c
> ===================================================================
> --- /dev/null
> +++ qemu-block-copy/block/blkmirror.c
> @@ -0,0 +1,239 @@
> +/*
> + * Block driver for mirrored writes.
> + *
> + * Copyright (C) 2011 Red Hat, Inc.
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or later.
> + * See the COPYING file in the top-level directory.
> + */
> +
> +#include <stdarg.h>
> +#include "block_int.h"
> +
> +typedef struct {
> +    BlockDriverState *bs[2];
> +} BdrvMirrorState;
> +
> +/* Valid blkmirror filenames look like blkmirror:path/to/image1:path/to/image2 */
> +static int blkmirror_open(BlockDriverState *bs, const char *filename, int flags)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    int ret;
> +    char *raw, *c;
> +
> +    /* Parse the blkmirror: prefix */
> +    if (strncmp(filename, "blkmirror:", strlen("blkmirror:"))) {
> +        return -EINVAL;
> +    }
> +    filename += strlen("blkmirror:");
> +
> +    /* Parse the raw image filename */
> +    c = strchr(filename, ':');
> +    if (c == NULL) {
> +        return -EINVAL;
> +    }
> +
> +    raw = strdup(filename);
> +    raw[c - filename] = '\0';
> +    ret = bdrv_file_open(&m->bs[0], raw, flags);
> +    free(raw);
> +    if (ret < 0) {
> +        return ret;
> +    }
> +    filename = c + 1;
> +
> +    ret = bdrv_file_open(&m->bs[1], filename, flags);
> +    if (ret < 0) {
> +        bdrv_delete(m->bs[0]);
> +        return ret;
> +    }
> +
> +    return 0;
> +}
> +
> +static void blkmirror_close(BlockDriverState *bs)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    int i;
> +
> +    for (i=0; i<2; i++) {
> +        bdrv_delete(m->bs[i]);
> +        m->bs[i] = NULL;
> +    }
> +}
> +
> +static int blkmirror_flush(BlockDriverState *bs)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +
> +    bdrv_flush(m->bs[0]);
> +    bdrv_flush(m->bs[1]);
> +
> +    return 0;
> +}
> +
> +static int64_t blkmirror_getlength(BlockDriverState *bs)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +
> +    return bdrv_getlength(m->bs[0]);
> +}
> +
> +static BlockDriverAIOCB *blkmirror_aio_readv(BlockDriverState *bs,
> +                                             int64_t sector_num,
> +                                             QEMUIOVector *qiov,
> +                                             int nb_sectors,
> +                                             BlockDriverCompletionFunc *cb,
> +                                             void *opaque)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    return bdrv_aio_readv(m->bs[0], sector_num, qiov, nb_sectors, cb, opaque);
> +}
> +
> +typedef struct DupAIOCB {
> +    BlockDriverAIOCB common;
> +    int count;
> +
> +    BlockDriverCompletionFunc *cb;
> +    void *opaque;
> +
> +    BlockDriverAIOCB *src_aiocb;
> +    BlockDriverAIOCB *dest_aiocb;
> +
> +    int ret;
> +} DupAIOCB;

Usually all types are defined at the top of the file.

> +
> +static void dup_aio_cancel(BlockDriverAIOCB *blockacb)
> +{
> +    DupAIOCB *acb = container_of(blockacb, DupAIOCB, common);
> +
> +    bdrv_aio_cancel(acb->src_aiocb);
> +    bdrv_aio_cancel(acb->dest_aiocb);
> +    qemu_aio_release(acb);
> +}
> +
> +static AIOPool dup_aio_pool = {
> +    .aiocb_size         = sizeof(DupAIOCB),
> +    .cancel             = dup_aio_cancel,
> +};
> +
> +static void blkmirror_aio_cb(void *opaque, int ret)
> +{
> +    DupAIOCB *dcb = opaque;
> +
> +    dcb->count--;
> +    assert(dcb->count >= 0);
> +    if (dcb->count == 0) {
> +        if (dcb->ret < 0) {
> +            ret = dcb->ret;
> +        }
> +        dcb->common.cb(dcb->opaque, ret);
> +        qemu_aio_release(dcb);
> +    }
> +    dcb->ret = ret;
> +}
> +
> +static DupAIOCB *dup_aio_get(BlockDriverState *bs,
> +                             BlockDriverCompletionFunc *cb,
> +                             void *opaque)
> +{
> +    DupAIOCB *dcb;
> +
> +    dcb = qemu_aio_get(&dup_aio_pool, bs, cb, opaque);
> +    if (!dcb)
> +        return NULL;

Please use QEMU's scripts/checkpatch.pl.

> +    dcb->count = 2;
> +    dcb->ret = 0;
> +    dcb->opaque = opaque;
> +
> +    return dcb;
> +}
> +
> +static BlockDriverAIOCB *blkmirror_aio_writev(BlockDriverState *bs,
> +                                              int64_t sector_num,
> +                                              QEMUIOVector *qiov,
> +                                              int nb_sectors,
> +                                              BlockDriverCompletionFunc *cb,
> +                                              void *opaque)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    DupAIOCB *dcb = dup_aio_get(bs, cb, opaque);
> +
> +    dcb->src_aiocb = bdrv_aio_writev(m->bs[0], sector_num, qiov, nb_sectors,
> +                                     &blkmirror_aio_cb, dcb);
> +    if (!dcb->src_aiocb) {
> +        qemu_aio_release(dcb);
> +        return NULL;
> +    }
> +
> +    dcb->dest_aiocb = bdrv_aio_writev(m->bs[1], sector_num, qiov, nb_sectors,
> +                                      &blkmirror_aio_cb, dcb);
> +    if (!dcb->dest_aiocb) {
> +        bdrv_aio_cancel(dcb->src_aiocb);
> +        qemu_aio_release(dcb);
> +        return NULL;
> +    }
> +
> +    return &dcb->common;
> +}
> +
> +static BlockDriverAIOCB *blkmirror_aio_flush(BlockDriverState *bs,
> +                                             BlockDriverCompletionFunc *cb,
> +                                             void *opaque)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    DupAIOCB *dcb = dup_aio_get(bs, cb, opaque);
> +
> +    dcb->src_aiocb = bdrv_aio_flush(m->bs[0], &blkmirror_aio_cb, dcb);
> +    if (!dcb->src_aiocb) {
> +        qemu_aio_release(dcb);
> +        return NULL;
> +    }
> +    dcb->dest_aiocb = bdrv_aio_flush(m->bs[1], &blkmirror_aio_cb, dcb);
> +    if (!dcb->dest_aiocb) {
> +        bdrv_aio_cancel(dcb->src_aiocb);
> +        qemu_aio_release(dcb);
> +        return NULL;
> +    }
> +
> +    return &dcb->common;
> +}
> +
> +static int blkmirror_discard(BlockDriverState *bs, int64_t sector_num,
> +                             int nb_sectors)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    int ret;
> +
> +    ret = bdrv_discard(m->bs[0], sector_num, nb_sectors);
> +    if (ret < 0) {
> +        return ret;
> +    }
> +
> +    return bdrv_discard(m->bs[1], sector_num, nb_sectors);
> +}
> +
> +
> +static BlockDriver bdrv_blkmirror = {
> +    .format_name        = "blkmirror",
> +    .protocol_name      = "blkmirror",
> +    .instance_size      = sizeof(BdrvMirrorState),
> +
> +    .bdrv_getlength     = blkmirror_getlength,
> +
> +    .bdrv_file_open     = blkmirror_open,
> +    .bdrv_close         = blkmirror_close,
> +    .bdrv_flush         = blkmirror_flush,
> +    .bdrv_discard       = blkmirror_discard,
> +
> +    .bdrv_aio_readv     = blkmirror_aio_readv,
> +    .bdrv_aio_writev    = blkmirror_aio_writev,
> +    .bdrv_aio_flush     = blkmirror_aio_flush,
> +};
> +
> +static void bdrv_blkmirror_init(void)
> +{
> +    bdrv_register(&bdrv_blkmirror);
> +}
> +
> +block_init(bdrv_blkmirror_init);
> Index: qemu-block-copy/Makefile.objs
> ===================================================================
> --- qemu-block-copy.orig/Makefile.objs
> +++ qemu-block-copy/Makefile.objs
> @@ -22,7 +22,7 @@ block-nested-y += raw.o cow.o qcow.o vdi
>  block-nested-y += qcow2.o qcow2-refcount.o qcow2-cluster.o qcow2-snapshot.o qcow2-cache.o
>  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 += parallels.o nbd.o blkdebug.o sheepdog.o blkverify.o blkmirror.o
>  block-nested-$(CONFIG_WIN32) += raw-win32.o
>  block-nested-$(CONFIG_POSIX) += raw-posix.o
>  block-nested-$(CONFIG_CURL) += curl.o
>
>
>
>

Am 23.05.2011 23:31, schrieb Marcelo Tosatti:
> Mirrored writes are used by live block copy.
> 
> Signed-off-by: Marcelo Tosatti <mtosatti@redhat.com>
> 
> Index: qemu-block-copy/block/blkmirror.c
> ===================================================================
> --- /dev/null
> +++ qemu-block-copy/block/blkmirror.c
> @@ -0,0 +1,239 @@
> +/*
> + * Block driver for mirrored writes.
> + *
> + * Copyright (C) 2011 Red Hat, Inc.
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or later.
> + * See the COPYING file in the top-level directory.
> + */
> +
> +#include <stdarg.h>
> +#include "block_int.h"
> +
> +typedef struct {
> +    BlockDriverState *bs[2];
> +} BdrvMirrorState;
> +
> +/* Valid blkmirror filenames look like blkmirror:path/to/image1:path/to/image2 */

We'll probably need a method to escape colons in the file name. It
didn't matter much for blkdebug and blkverify because both are only for
debugging, but for block migration we need to be able to handle
everything that works locally.

> +static int blkmirror_open(BlockDriverState *bs, const char *filename, int flags)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    int ret;
> +    char *raw, *c;
> +
> +    /* Parse the blkmirror: prefix */
> +    if (strncmp(filename, "blkmirror:", strlen("blkmirror:"))) {
> +        return -EINVAL;
> +    }
> +    filename += strlen("blkmirror:");
> +
> +    /* Parse the raw image filename */
> +    c = strchr(filename, ':');
> +    if (c == NULL) {
> +        return -EINVAL;
> +    }
> +
> +    raw = strdup(filename);
> +    raw[c - filename] = '\0';
> +    ret = bdrv_file_open(&m->bs[0], raw, flags);
> +    free(raw);
> +    if (ret < 0) {
> +        return ret;
> +    }
> +    filename = c + 1;
> +
> +    ret = bdrv_file_open(&m->bs[1], filename, flags);
> +    if (ret < 0) {
> +        bdrv_delete(m->bs[0]);
> +        return ret;
> +    }

Use bdrv_open instead of bdrv_file_open, otherwise it only works with
raw images.

> +
> +    return 0;
> +}
> +
> +static void blkmirror_close(BlockDriverState *bs)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    int i;
> +
> +    for (i=0; i<2; i++) {
> +        bdrv_delete(m->bs[i]);
> +        m->bs[i] = NULL;
> +    }
> +}
> +
> +static int blkmirror_flush(BlockDriverState *bs)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +
> +    bdrv_flush(m->bs[0]);
> +    bdrv_flush(m->bs[1]);
> +
> +    return 0;
> +}
> +
> +static int64_t blkmirror_getlength(BlockDriverState *bs)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +
> +    return bdrv_getlength(m->bs[0]);
> +}
> +
> +static BlockDriverAIOCB *blkmirror_aio_readv(BlockDriverState *bs,
> +                                             int64_t sector_num,
> +                                             QEMUIOVector *qiov,
> +                                             int nb_sectors,
> +                                             BlockDriverCompletionFunc *cb,
> +                                             void *opaque)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    return bdrv_aio_readv(m->bs[0], sector_num, qiov, nb_sectors, cb, opaque);
> +}
> +
> +typedef struct DupAIOCB {
> +    BlockDriverAIOCB common;
> +    int count;
> +
> +    BlockDriverCompletionFunc *cb;
> +    void *opaque;
> +
> +    BlockDriverAIOCB *src_aiocb;
> +    BlockDriverAIOCB *dest_aiocb;
> +
> +    int ret;
> +} DupAIOCB;
> +
> +static void dup_aio_cancel(BlockDriverAIOCB *blockacb)
> +{
> +    DupAIOCB *acb = container_of(blockacb, DupAIOCB, common);
> +
> +    bdrv_aio_cancel(acb->src_aiocb);
> +    bdrv_aio_cancel(acb->dest_aiocb);
> +    qemu_aio_release(acb);
> +}

You must not cancel a request that has already completed. It could
happen that only the first request has completed yet and the
bdrv_aio_cancel happens before the whole blkmirror request has
completed. Even worse, the first bdrv_aio_cancel may cause the second
request to complete.

> +
> +static AIOPool dup_aio_pool = {
> +    .aiocb_size         = sizeof(DupAIOCB),
> +    .cancel             = dup_aio_cancel,
> +};
> +
> +static void blkmirror_aio_cb(void *opaque, int ret)
> +{
> +    DupAIOCB *dcb = opaque;
> +
> +    dcb->count--;
> +    assert(dcb->count >= 0);
> +    if (dcb->count == 0) {
> +        if (dcb->ret < 0) {
> +            ret = dcb->ret;
> +        }
> +        dcb->common.cb(dcb->opaque, ret);
> +        qemu_aio_release(dcb);
> +    }
> +    dcb->ret = ret;
> +}
> +
> +static DupAIOCB *dup_aio_get(BlockDriverState *bs,
> +                             BlockDriverCompletionFunc *cb,
> +                             void *opaque)
> +{
> +    DupAIOCB *dcb;
> +
> +    dcb = qemu_aio_get(&dup_aio_pool, bs, cb, opaque);
> +    if (!dcb)
> +        return NULL;
> +    dcb->count = 2;
> +    dcb->ret = 0;
> +    dcb->opaque = opaque;

Why do we need dcb->opaque when there's dcb->common.opaque?

Kevin

On 05/24/2011 12:31 AM, Marcelo Tosatti wrote:
> Mirrored writes are used by live block copy.
>
>
> +
> +typedef struct {
> +    BlockDriverState *bs[2];
> +} BdrvMirrorState;

Can be generalized one day to N.

> +
> +/* Valid blkmirror filenames look like blkmirror:path/to/image1:path/to/image2 */
> +static int blkmirror_open(BlockDriverState *bs, const char *filename, int flags)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    int ret;
> +    char *raw, *c;
> +
> +    /* Parse the blkmirror: prefix */
> +    if (strncmp(filename, "blkmirror:", strlen("blkmirror:"))) {

Actual format should be documented somewhere.

> +        return -EINVAL;
> +    }
> +    filename += strlen("blkmirror:");
> +
> +    /* Parse the raw image filename */
> +    c = strchr(filename, ':');
> +    if (c == NULL) {
> +        return -EINVAL;
> +    }
> +
> +    raw = strdup(filename);
> +    raw[c - filename] = '\0';
> +    ret = bdrv_file_open(&m->bs[0], raw, flags);
> +    free(raw);
> +    if (ret<  0) {
> +        return ret;
> +    }
> +    filename = c + 1;
> +
> +    ret = bdrv_file_open(&m->bs[1], filename, flags);
> +    if (ret<  0) {
> +        bdrv_delete(m->bs[0]);
> +        return ret;
> +    }
> +
> +    return 0;
> +}
> +
>
> +
> +static BlockDriverAIOCB *blkmirror_aio_readv(BlockDriverState *bs,
> +                                             int64_t sector_num,
> +                                             QEMUIOVector *qiov,
> +                                             int nb_sectors,
> +                                             BlockDriverCompletionFunc *cb,
> +                                             void *opaque)
> +{
> +    BdrvMirrorState *m = bs->opaque;
> +    return bdrv_aio_readv(m->bs[0], sector_num, qiov, nb_sectors, cb, opaque);

We could one day be clever and round-robin once we're synced.

> +}
> +
> +typedef struct DupAIOCB {
> +    BlockDriverAIOCB common;
> +    int count;
> +
> +    BlockDriverCompletionFunc *cb;
> +    void *opaque;
> +
> +    BlockDriverAIOCB *src_aiocb;
> +    BlockDriverAIOCB *dest_aiocb;
> +
> +    int ret;
> +} DupAIOCB;
> +
> +static void dup_aio_cancel(BlockDriverAIOCB *blockacb)
> +{
> +    DupAIOCB *acb = container_of(blockacb, DupAIOCB, common);
> +
> +    bdrv_aio_cancel(acb->src_aiocb);
> +    bdrv_aio_cancel(acb->dest_aiocb);

Shouldn't we cancel just the ones that haven't completed yet?

> +    qemu_aio_release(acb);
> +}
> +
> +static AIOPool dup_aio_pool = {
> +    .aiocb_size         = sizeof(DupAIOCB),
> +    .cancel             = dup_aio_cancel,
> +};
> +
> +static void blkmirror_aio_cb(void *opaque, int ret)
> +{
> +    DupAIOCB *dcb = opaque;
> +
> +    dcb->count--;
> +    assert(dcb->count>= 0);
> +    if (dcb->count == 0) {
> +        if (dcb->ret<  0) {
> +            ret = dcb->ret;
> +        }
> +        dcb->common.cb(dcb->opaque, ret);
> +        qemu_aio_release(dcb);
> +    }
> +    dcb->ret = ret;
> +}

It would be nicer to do

     if (ret < 0) {
         dcb->ret = ret;
     }

unconditionally.  The current code works, but only for N=2, and is a 
little obfuscated.

I see you aren't keeping sync/unsync state here.  Will read on.