Re: [Qemu-devel] [patch 2/7] Add blkmirror block driver

[Kevin Wolf <kwolf@redhat.com>]

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