[thirdparty/qemu.git] / block / replication.c

/*
 * Replication Block filter
 *
 * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
 * Copyright (c) 2016 Intel Corporation
 * Copyright (c) 2016 FUJITSU LIMITED
 *
 * Author:
 *   Wen Congyang <wency@cn.fujitsu.com>
 *
 * 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 "qemu/osdep.h"
#include "qemu-common.h"
#include "block/nbd.h"
#include "block/blockjob.h"
#include "block/block_int.h"
#include "block/block_backup.h"
#include "sysemu/block-backend.h"
#include "qapi/error.h"
#include "replication.h"

typedef enum {
    BLOCK_REPLICATION_NONE,             /* block replication is not started */
    BLOCK_REPLICATION_RUNNING,          /* block replication is running */
    BLOCK_REPLICATION_FAILOVER,         /* failover is running in background */
    BLOCK_REPLICATION_FAILOVER_FAILED,  /* failover failed */
    BLOCK_REPLICATION_DONE,             /* block replication is done */
} ReplicationStage;

typedef struct BDRVReplicationState {
    ReplicationMode mode;
    ReplicationStage stage;
    BdrvChild *active_disk;
    BdrvChild *hidden_disk;
    BdrvChild *secondary_disk;
    char *top_id;
    ReplicationState *rs;
    Error *blocker;
    int orig_hidden_flags;
    int orig_secondary_flags;
    int error;
} BDRVReplicationState;

static void replication_start(ReplicationState *rs, ReplicationMode mode,
                              Error **errp);
static void replication_do_checkpoint(ReplicationState *rs, Error **errp);
static void replication_get_error(ReplicationState *rs, Error **errp);
static void replication_stop(ReplicationState *rs, bool failover,
                             Error **errp);

#define REPLICATION_MODE        "mode"
#define REPLICATION_TOP_ID      "top-id"
static QemuOptsList replication_runtime_opts = {
    .name = "replication",
    .head = QTAILQ_HEAD_INITIALIZER(replication_runtime_opts.head),
    .desc = {
        {
            .name = REPLICATION_MODE,
            .type = QEMU_OPT_STRING,
        },
        {
            .name = REPLICATION_TOP_ID,
            .type = QEMU_OPT_STRING,
        },
        { /* end of list */ }
    },
};

static ReplicationOps replication_ops = {
    .start = replication_start,
    .checkpoint = replication_do_checkpoint,
    .get_error = replication_get_error,
    .stop = replication_stop,
};

static int replication_open(BlockDriverState *bs, QDict *options,
                            int flags, Error **errp)
{
    int ret;
    BDRVReplicationState *s = bs->opaque;
    Error *local_err = NULL;
    QemuOpts *opts = NULL;
    const char *mode;
    const char *top_id;

    bs->file = bdrv_open_child(NULL, options, "file", bs, &child_file,
                               false, errp);
    if (!bs->file) {
        return -EINVAL;
    }

    ret = -EINVAL;
    opts = qemu_opts_create(&replication_runtime_opts, NULL, 0, &error_abort);
    qemu_opts_absorb_qdict(opts, options, &local_err);
    if (local_err) {
        goto fail;
    }

    mode = qemu_opt_get(opts, REPLICATION_MODE);
    if (!mode) {
        error_setg(&local_err, "Missing the option mode");
        goto fail;
    }

    if (!strcmp(mode, "primary")) {
        s->mode = REPLICATION_MODE_PRIMARY;
        top_id = qemu_opt_get(opts, REPLICATION_TOP_ID);
        if (top_id) {
            error_setg(&local_err, "The primary side does not support option top-id");
            goto fail;
        }
    } else if (!strcmp(mode, "secondary")) {
        s->mode = REPLICATION_MODE_SECONDARY;
        top_id = qemu_opt_get(opts, REPLICATION_TOP_ID);
        s->top_id = g_strdup(top_id);
        if (!s->top_id) {
            error_setg(&local_err, "Missing the option top-id");
            goto fail;
        }
    } else {
        error_setg(&local_err,
                   "The option mode's value should be primary or secondary");
        goto fail;
    }

    s->rs = replication_new(bs, &replication_ops);

    ret = 0;

fail:
    qemu_opts_del(opts);
    error_propagate(errp, local_err);

    return ret;
}

static void replication_close(BlockDriverState *bs)
{
    BDRVReplicationState *s = bs->opaque;

    if (s->stage == BLOCK_REPLICATION_RUNNING) {
        replication_stop(s->rs, false, NULL);
    }
    if (s->stage == BLOCK_REPLICATION_FAILOVER) {
        block_job_cancel_sync(s->active_disk->bs->job);
    }

    if (s->mode == REPLICATION_MODE_SECONDARY) {
        g_free(s->top_id);
    }

    replication_remove(s->rs);
}

static void replication_child_perm(BlockDriverState *bs, BdrvChild *c,
                                   const BdrvChildRole *role,
                                   BlockReopenQueue *reopen_queue,
                                   uint64_t perm, uint64_t shared,
                                   uint64_t *nperm, uint64_t *nshared)
{
    *nperm = *nshared = BLK_PERM_CONSISTENT_READ \
                        | BLK_PERM_WRITE \
                        | BLK_PERM_WRITE_UNCHANGED;

    return;
}

static int64_t replication_getlength(BlockDriverState *bs)
{
    return bdrv_getlength(bs->file->bs);
}

static int replication_get_io_status(BDRVReplicationState *s)
{
    switch (s->stage) {
    case BLOCK_REPLICATION_NONE:
        return -EIO;
    case BLOCK_REPLICATION_RUNNING:
        return 0;
    case BLOCK_REPLICATION_FAILOVER:
        return s->mode == REPLICATION_MODE_PRIMARY ? -EIO : 0;
    case BLOCK_REPLICATION_FAILOVER_FAILED:
        return s->mode == REPLICATION_MODE_PRIMARY ? -EIO : 1;
    case BLOCK_REPLICATION_DONE:
        /*
         * active commit job completes, and active disk and secondary_disk
         * is swapped, so we can operate bs->file directly
         */
        return s->mode == REPLICATION_MODE_PRIMARY ? -EIO : 0;
    default:
        abort();
    }
}

static int replication_return_value(BDRVReplicationState *s, int ret)
{
    if (s->mode == REPLICATION_MODE_SECONDARY) {
        return ret;
    }

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

    return ret;
}

static coroutine_fn int replication_co_readv(BlockDriverState *bs,
                                             int64_t sector_num,
                                             int remaining_sectors,
                                             QEMUIOVector *qiov)
{
    BDRVReplicationState *s = bs->opaque;
    BdrvChild *child = s->secondary_disk;
    BlockJob *job = NULL;
    CowRequest req;
    int ret;

    if (s->mode == REPLICATION_MODE_PRIMARY) {
        /* We only use it to forward primary write requests */
        return -EIO;
    }

    ret = replication_get_io_status(s);
    if (ret < 0) {
        return ret;
    }

    if (child && child->bs) {
        job = child->bs->job;
    }

    if (job) {
        uint64_t remaining_bytes = remaining_sectors * BDRV_SECTOR_SIZE;

        backup_wait_for_overlapping_requests(child->bs->job,
                                             sector_num * BDRV_SECTOR_SIZE,
                                             remaining_bytes);
        backup_cow_request_begin(&req, child->bs->job,
                                 sector_num * BDRV_SECTOR_SIZE,
                                 remaining_bytes);
        ret = bdrv_co_readv(bs->file, sector_num, remaining_sectors,
                            qiov);
        backup_cow_request_end(&req);
        goto out;
    }

    ret = bdrv_co_readv(bs->file, sector_num, remaining_sectors, qiov);
out:
    return replication_return_value(s, ret);
}

static coroutine_fn int replication_co_writev(BlockDriverState *bs,
                                              int64_t sector_num,
                                              int remaining_sectors,
                                              QEMUIOVector *qiov)
{
    BDRVReplicationState *s = bs->opaque;
    QEMUIOVector hd_qiov;
    uint64_t bytes_done = 0;
    BdrvChild *top = bs->file;
    BdrvChild *base = s->secondary_disk;
    BdrvChild *target;
    int ret;
    int64_t n;

    ret = replication_get_io_status(s);
    if (ret < 0) {
        goto out;
    }

    if (ret == 0) {
        ret = bdrv_co_writev(top, sector_num,
                             remaining_sectors, qiov);
        return replication_return_value(s, ret);
    }

    /*
     * Failover failed, only write to active disk if the sectors
     * have already been allocated in active disk/hidden disk.
     */
    qemu_iovec_init(&hd_qiov, qiov->niov);
    while (remaining_sectors > 0) {
        int64_t count;

        ret = bdrv_is_allocated_above(top->bs, base->bs,
                                      sector_num * BDRV_SECTOR_SIZE,
                                      remaining_sectors * BDRV_SECTOR_SIZE,
                                      &count);
        if (ret < 0) {
            goto out1;
        }

        assert(QEMU_IS_ALIGNED(count, BDRV_SECTOR_SIZE));
        n = count >> BDRV_SECTOR_BITS;
        qemu_iovec_reset(&hd_qiov);
        qemu_iovec_concat(&hd_qiov, qiov, bytes_done, count);

        target = ret ? top : base;
        ret = bdrv_co_writev(target, sector_num, n, &hd_qiov);
        if (ret < 0) {
            goto out1;
        }

        remaining_sectors -= n;
        sector_num += n;
        bytes_done += count;
    }

out1:
    qemu_iovec_destroy(&hd_qiov);
out:
    return ret;
}

static bool replication_recurse_is_first_non_filter(BlockDriverState *bs,
                                                    BlockDriverState *candidate)
{
    return bdrv_recurse_is_first_non_filter(bs->file->bs, candidate);
}

static void secondary_do_checkpoint(BDRVReplicationState *s, Error **errp)
{
    Error *local_err = NULL;
    int ret;

    if (!s->secondary_disk->bs->job) {
        error_setg(errp, "Backup job was cancelled unexpectedly");
        return;
    }

    backup_do_checkpoint(s->secondary_disk->bs->job, &local_err);
    if (local_err) {
        error_propagate(errp, local_err);
        return;
    }

    ret = s->active_disk->bs->drv->bdrv_make_empty(s->active_disk->bs);
    if (ret < 0) {
        error_setg(errp, "Cannot make active disk empty");
        return;
    }

    ret = s->hidden_disk->bs->drv->bdrv_make_empty(s->hidden_disk->bs);
    if (ret < 0) {
        error_setg(errp, "Cannot make hidden disk empty");
        return;
    }
}

static void reopen_backing_file(BlockDriverState *bs, bool writable,
                                Error **errp)
{
    BDRVReplicationState *s = bs->opaque;
    BlockReopenQueue *reopen_queue = NULL;
    int orig_hidden_flags, orig_secondary_flags;
    int new_hidden_flags, new_secondary_flags;
    Error *local_err = NULL;

    if (writable) {
        orig_hidden_flags = s->orig_hidden_flags =
                                bdrv_get_flags(s->hidden_disk->bs);
        new_hidden_flags = (orig_hidden_flags | BDRV_O_RDWR) &
                                                    ~BDRV_O_INACTIVE;
        orig_secondary_flags = s->orig_secondary_flags =
                                bdrv_get_flags(s->secondary_disk->bs);
        new_secondary_flags = (orig_secondary_flags | BDRV_O_RDWR) &
                                                     ~BDRV_O_INACTIVE;
    } else {
        orig_hidden_flags = (s->orig_hidden_flags | BDRV_O_RDWR) &
                                                    ~BDRV_O_INACTIVE;
        new_hidden_flags = s->orig_hidden_flags;
        orig_secondary_flags = (s->orig_secondary_flags | BDRV_O_RDWR) &
                                                    ~BDRV_O_INACTIVE;
        new_secondary_flags = s->orig_secondary_flags;
    }

    if (orig_hidden_flags != new_hidden_flags) {
        reopen_queue = bdrv_reopen_queue(reopen_queue, s->hidden_disk->bs, NULL,
                                         new_hidden_flags);
    }

    if (!(orig_secondary_flags & BDRV_O_RDWR)) {
        reopen_queue = bdrv_reopen_queue(reopen_queue, s->secondary_disk->bs,
                                         NULL, new_secondary_flags);
    }

    if (reopen_queue) {
        bdrv_reopen_multiple(bdrv_get_aio_context(bs),
                             reopen_queue, &local_err);
        error_propagate(errp, local_err);
    }
}

static void backup_job_cleanup(BlockDriverState *bs)
{
    BDRVReplicationState *s = bs->opaque;
    BlockDriverState *top_bs;

    top_bs = bdrv_lookup_bs(s->top_id, s->top_id, NULL);
    if (!top_bs) {
        return;
    }
    bdrv_op_unblock_all(top_bs, s->blocker);
    error_free(s->blocker);
    reopen_backing_file(bs, false, NULL);
}

static void backup_job_completed(void *opaque, int ret)
{
    BlockDriverState *bs = opaque;
    BDRVReplicationState *s = bs->opaque;

    if (s->stage != BLOCK_REPLICATION_FAILOVER) {
        /* The backup job is cancelled unexpectedly */
        s->error = -EIO;
    }

    backup_job_cleanup(bs);
}

static bool check_top_bs(BlockDriverState *top_bs, BlockDriverState *bs)
{
    BdrvChild *child;

    /* The bs itself is the top_bs */
    if (top_bs == bs) {
        return true;
    }

    /* Iterate over top_bs's children */
    QLIST_FOREACH(child, &top_bs->children, next) {
        if (child->bs == bs || check_top_bs(child->bs, bs)) {
            return true;
        }
    }

    return false;
}

static void replication_start(ReplicationState *rs, ReplicationMode mode,
                              Error **errp)
{
    BlockDriverState *bs = rs->opaque;
    BDRVReplicationState *s;
    BlockDriverState *top_bs;
    int64_t active_length, hidden_length, disk_length;
    AioContext *aio_context;
    Error *local_err = NULL;
    BlockJob *job;

    aio_context = bdrv_get_aio_context(bs);
    aio_context_acquire(aio_context);
    s = bs->opaque;

    if (s->stage != BLOCK_REPLICATION_NONE) {
        error_setg(errp, "Block replication is running or done");
        aio_context_release(aio_context);
        return;
    }

    if (s->mode != mode) {
        error_setg(errp, "The parameter mode's value is invalid, needs %d,"
                   " but got %d", s->mode, mode);
        aio_context_release(aio_context);
        return;
    }

    switch (s->mode) {
    case REPLICATION_MODE_PRIMARY:
        break;
    case REPLICATION_MODE_SECONDARY:
        s->active_disk = bs->file;
        if (!s->active_disk || !s->active_disk->bs ||
                                    !s->active_disk->bs->backing) {
            error_setg(errp, "Active disk doesn't have backing file");
            aio_context_release(aio_context);
            return;
        }

        s->hidden_disk = s->active_disk->bs->backing;
        if (!s->hidden_disk->bs || !s->hidden_disk->bs->backing) {
            error_setg(errp, "Hidden disk doesn't have backing file");
            aio_context_release(aio_context);
            return;
        }

        s->secondary_disk = s->hidden_disk->bs->backing;
        if (!s->secondary_disk->bs || !bdrv_has_blk(s->secondary_disk->bs)) {
            error_setg(errp, "The secondary disk doesn't have block backend");
            aio_context_release(aio_context);
            return;
        }

        /* verify the length */
        active_length = bdrv_getlength(s->active_disk->bs);
        hidden_length = bdrv_getlength(s->hidden_disk->bs);
        disk_length = bdrv_getlength(s->secondary_disk->bs);
        if (active_length < 0 || hidden_length < 0 || disk_length < 0 ||
            active_length != hidden_length || hidden_length != disk_length) {
            error_setg(errp, "Active disk, hidden disk, secondary disk's length"
                       " are not the same");
            aio_context_release(aio_context);
            return;
        }

        if (!s->active_disk->bs->drv->bdrv_make_empty ||
            !s->hidden_disk->bs->drv->bdrv_make_empty) {
            error_setg(errp,
                       "Active disk or hidden disk doesn't support make_empty");
            aio_context_release(aio_context);
            return;
        }

        /* reopen the backing file in r/w mode */
        reopen_backing_file(bs, true, &local_err);
        if (local_err) {
            error_propagate(errp, local_err);
            aio_context_release(aio_context);
            return;
        }

        /* start backup job now */
        error_setg(&s->blocker,
                   "Block device is in use by internal backup job");

        top_bs = bdrv_lookup_bs(s->top_id, s->top_id, NULL);
        if (!top_bs || !bdrv_is_root_node(top_bs) ||
            !check_top_bs(top_bs, bs)) {
            error_setg(errp, "No top_bs or it is invalid");
            reopen_backing_file(bs, false, NULL);
            aio_context_release(aio_context);
            return;
        }
        bdrv_op_block_all(top_bs, s->blocker);
        bdrv_op_unblock(top_bs, BLOCK_OP_TYPE_DATAPLANE, s->blocker);

        job = backup_job_create(NULL, s->secondary_disk->bs, s->hidden_disk->bs,
                                0, MIRROR_SYNC_MODE_NONE, NULL, false,
                                BLOCKDEV_ON_ERROR_REPORT,
                                BLOCKDEV_ON_ERROR_REPORT, BLOCK_JOB_INTERNAL,
                                backup_job_completed, bs, NULL, &local_err);
        if (local_err) {
            error_propagate(errp, local_err);
            backup_job_cleanup(bs);
            aio_context_release(aio_context);
            return;
        }
        block_job_start(job);
        break;
    default:
        aio_context_release(aio_context);
        abort();
    }

    s->stage = BLOCK_REPLICATION_RUNNING;

    if (s->mode == REPLICATION_MODE_SECONDARY) {
        secondary_do_checkpoint(s, errp);
    }

    s->error = 0;
    aio_context_release(aio_context);
}

static void replication_do_checkpoint(ReplicationState *rs, Error **errp)
{
    BlockDriverState *bs = rs->opaque;
    BDRVReplicationState *s;
    AioContext *aio_context;

    aio_context = bdrv_get_aio_context(bs);
    aio_context_acquire(aio_context);
    s = bs->opaque;

    if (s->mode == REPLICATION_MODE_SECONDARY) {
        secondary_do_checkpoint(s, errp);
    }
    aio_context_release(aio_context);
}

static void replication_get_error(ReplicationState *rs, Error **errp)
{
    BlockDriverState *bs = rs->opaque;
    BDRVReplicationState *s;
    AioContext *aio_context;

    aio_context = bdrv_get_aio_context(bs);
    aio_context_acquire(aio_context);
    s = bs->opaque;

    if (s->stage != BLOCK_REPLICATION_RUNNING) {
        error_setg(errp, "Block replication is not running");
        aio_context_release(aio_context);
        return;
    }

    if (s->error) {
        error_setg(errp, "I/O error occurred");
        aio_context_release(aio_context);
        return;
    }
    aio_context_release(aio_context);
}

static void replication_done(void *opaque, int ret)
{
    BlockDriverState *bs = opaque;
    BDRVReplicationState *s = bs->opaque;

    if (ret == 0) {
        s->stage = BLOCK_REPLICATION_DONE;

        /* refresh top bs's filename */
        bdrv_refresh_filename(bs);
        s->active_disk = NULL;
        s->secondary_disk = NULL;
        s->hidden_disk = NULL;
        s->error = 0;
    } else {
        s->stage = BLOCK_REPLICATION_FAILOVER_FAILED;
        s->error = -EIO;
    }
}

static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
{
    BlockDriverState *bs = rs->opaque;
    BDRVReplicationState *s;
    AioContext *aio_context;

    aio_context = bdrv_get_aio_context(bs);
    aio_context_acquire(aio_context);
    s = bs->opaque;

    if (s->stage != BLOCK_REPLICATION_RUNNING) {
        error_setg(errp, "Block replication is not running");
        aio_context_release(aio_context);
        return;
    }

    switch (s->mode) {
    case REPLICATION_MODE_PRIMARY:
        s->stage = BLOCK_REPLICATION_DONE;
        s->error = 0;
        break;
    case REPLICATION_MODE_SECONDARY:
        /*
         * This BDS will be closed, and the job should be completed
         * before the BDS is closed, because we will access hidden
         * disk, secondary disk in backup_job_completed().
         */
        if (s->secondary_disk->bs->job) {
            block_job_cancel_sync(s->secondary_disk->bs->job);
        }

        if (!failover) {
            secondary_do_checkpoint(s, errp);
            s->stage = BLOCK_REPLICATION_DONE;
            aio_context_release(aio_context);
            return;
        }

        s->stage = BLOCK_REPLICATION_FAILOVER;
        commit_active_start(NULL, s->active_disk->bs, s->secondary_disk->bs,
                            BLOCK_JOB_INTERNAL, 0, BLOCKDEV_ON_ERROR_REPORT,
                            NULL, replication_done, bs, true, errp);
        break;
    default:
        aio_context_release(aio_context);
        abort();
    }
    aio_context_release(aio_context);
}

BlockDriver bdrv_replication = {
    .format_name                = "replication",
    .protocol_name              = "replication",
    .instance_size              = sizeof(BDRVReplicationState),

    .bdrv_open                  = replication_open,
    .bdrv_close                 = replication_close,
    .bdrv_child_perm            = replication_child_perm,

    .bdrv_getlength             = replication_getlength,
    .bdrv_co_readv              = replication_co_readv,
    .bdrv_co_writev             = replication_co_writev,

    .is_filter                  = true,
    .bdrv_recurse_is_first_non_filter = replication_recurse_is_first_non_filter,

    .has_variable_length        = true,
};

static void bdrv_replication_init(void)
{
    bdrv_register(&bdrv_replication);
}

block_init(bdrv_replication_init);
Commit	Line	Data
29ff7890 WC	1	/*
	2	* Replication Block filter
	3	*
	4	* Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
	5	* Copyright (c) 2016 Intel Corporation
	6	* Copyright (c) 2016 FUJITSU LIMITED
	7	*
	8	* Author:
	9	* Wen Congyang <wency@cn.fujitsu.com>
	10	*
	11	* This work is licensed under the terms of the GNU GPL, version 2 or later.
	12	* See the COPYING file in the top-level directory.
	13	*/
	14
	15	#include "qemu/osdep.h"
	16	#include "qemu-common.h"
	17	#include "block/nbd.h"
	18	#include "block/blockjob.h"
	19	#include "block/block_int.h"
	20	#include "block/block_backup.h"
	21	#include "sysemu/block-backend.h"
	22	#include "qapi/error.h"
	23	#include "replication.h"
	24
3c76c606 FZ	25	typedef enum {
	26	BLOCK_REPLICATION_NONE, /* block replication is not started */
	27	BLOCK_REPLICATION_RUNNING, /* block replication is running */
	28	BLOCK_REPLICATION_FAILOVER, /* failover is running in background */
	29	BLOCK_REPLICATION_FAILOVER_FAILED, /* failover failed */
	30	BLOCK_REPLICATION_DONE, /* block replication is done */
	31	} ReplicationStage;
	32
29ff7890 WC	33	typedef struct BDRVReplicationState {
29ff7890 WC	34	ReplicationMode mode;
3c76c606	35	ReplicationStage stage;
29ff7890 WC	36	BdrvChild *active_disk;
	37	BdrvChild *hidden_disk;
	38	BdrvChild *secondary_disk;
	39	char *top_id;
	40	ReplicationState *rs;
	41	Error *blocker;
	42	int orig_hidden_flags;
	43	int orig_secondary_flags;
	44	int error;
	45	} BDRVReplicationState;
	46
29ff7890 WC	47	static void replication_start(ReplicationState *rs, ReplicationMode mode,
	48	Error **errp);
	49	static void replication_do_checkpoint(ReplicationState rs, Error *errp);
	50	static void replication_get_error(ReplicationState rs, Error *errp);
	51	static void replication_stop(ReplicationState *rs, bool failover,
	52	Error **errp);
	53
	54	#define REPLICATION_MODE "mode"
	55	#define REPLICATION_TOP_ID "top-id"
	56	static QemuOptsList replication_runtime_opts = {
	57	.name = "replication",
	58	.head = QTAILQ_HEAD_INITIALIZER(replication_runtime_opts.head),
	59	.desc = {
	60	{
	61	.name = REPLICATION_MODE,
	62	.type = QEMU_OPT_STRING,
	63	},
	64	{
	65	.name = REPLICATION_TOP_ID,
	66	.type = QEMU_OPT_STRING,
	67	},
	68	{ /* end of list */ }
	69	},
	70	};
	71
	72	static ReplicationOps replication_ops = {
	73	.start = replication_start,
	74	.checkpoint = replication_do_checkpoint,
	75	.get_error = replication_get_error,
	76	.stop = replication_stop,
	77	};
	78
	79	static int replication_open(BlockDriverState bs, QDict options,
	80	int flags, Error **errp)
	81	{
	82	int ret;
	83	BDRVReplicationState *s = bs->opaque;
	84	Error *local_err = NULL;
	85	QemuOpts *opts = NULL;
	86	const char *mode;
	87	const char *top_id;
	88
4e4bf5c4 KW	89	bs->file = bdrv_open_child(NULL, options, "file", bs, &child_file,
	90	false, errp);
	91	if (!bs->file) {
	92	return -EINVAL;
	93	}
	94
29ff7890 WC	95	ret = -EINVAL;
	96	opts = qemu_opts_create(&replication_runtime_opts, NULL, 0, &error_abort);
	97	qemu_opts_absorb_qdict(opts, options, &local_err);
	98	if (local_err) {
	99	goto fail;
	100	}
	101
	102	mode = qemu_opt_get(opts, REPLICATION_MODE);
	103	if (!mode) {
	104	error_setg(&local_err, "Missing the option mode");
	105	goto fail;
	106	}
	107
	108	if (!strcmp(mode, "primary")) {
	109	s->mode = REPLICATION_MODE_PRIMARY;
f4f2539b CX	110	top_id = qemu_opt_get(opts, REPLICATION_TOP_ID);
	111	if (top_id) {
	112	error_setg(&local_err, "The primary side does not support option top-id");
	113	goto fail;
	114	}
29ff7890 WC	115	} else if (!strcmp(mode, "secondary")) {
	116	s->mode = REPLICATION_MODE_SECONDARY;
	117	top_id = qemu_opt_get(opts, REPLICATION_TOP_ID);
	118	s->top_id = g_strdup(top_id);
	119	if (!s->top_id) {
	120	error_setg(&local_err, "Missing the option top-id");
	121	goto fail;
	122	}
	123	} else {
	124	error_setg(&local_err,
	125	"The option mode's value should be primary or secondary");
	126	goto fail;
	127	}
	128
	129	s->rs = replication_new(bs, &replication_ops);
	130
	131	ret = 0;
	132
	133	fail:
	134	qemu_opts_del(opts);
	135	error_propagate(errp, local_err);
	136
	137	return ret;
	138	}
	139
	140	static void replication_close(BlockDriverState *bs)
	141	{
	142	BDRVReplicationState *s = bs->opaque;
	143
3c76c606	144	if (s->stage == BLOCK_REPLICATION_RUNNING) {
29ff7890 WC	145	replication_stop(s->rs, false, NULL);
29ff7890 WC	146	}
3c76c606	147	if (s->stage == BLOCK_REPLICATION_FAILOVER) {
50ab0e09 PB	148	block_job_cancel_sync(s->active_disk->bs->job);
50ab0e09 PB	149	}
29ff7890 WC	150
	151	if (s->mode == REPLICATION_MODE_SECONDARY) {
	152	g_free(s->top_id);
	153	}
	154
	155	replication_remove(s->rs);
	156	}
	157
37a9051c CX	158	static void replication_child_perm(BlockDriverState bs, BdrvChild c,
37a9051c CX	159	const BdrvChildRole *role,
e0995dc3	160	BlockReopenQueue *reopen_queue,
37a9051c CX	161	uint64_t perm, uint64_t shared,
	162	uint64_t nperm, uint64_t nshared)
	163	{
	164	nperm = nshared = BLK_PERM_CONSISTENT_READ \
	165	\| BLK_PERM_WRITE \
	166	\| BLK_PERM_WRITE_UNCHANGED;
	167
	168	return;
	169	}
	170
29ff7890 WC	171	static int64_t replication_getlength(BlockDriverState *bs)
	172	{
	173	return bdrv_getlength(bs->file->bs);
	174	}
	175
	176	static int replication_get_io_status(BDRVReplicationState *s)
	177	{
3c76c606	178	switch (s->stage) {
29ff7890 WC	179	case BLOCK_REPLICATION_NONE:
	180	return -EIO;
	181	case BLOCK_REPLICATION_RUNNING:
	182	return 0;
	183	case BLOCK_REPLICATION_FAILOVER:
	184	return s->mode == REPLICATION_MODE_PRIMARY ? -EIO : 0;
	185	case BLOCK_REPLICATION_FAILOVER_FAILED:
	186	return s->mode == REPLICATION_MODE_PRIMARY ? -EIO : 1;
	187	case BLOCK_REPLICATION_DONE:
	188	/*
	189	* active commit job completes, and active disk and secondary_disk
	190	* is swapped, so we can operate bs->file directly
	191	*/
	192	return s->mode == REPLICATION_MODE_PRIMARY ? -EIO : 0;
	193	default:
	194	abort();
	195	}
	196	}
	197
	198	static int replication_return_value(BDRVReplicationState *s, int ret)
	199	{
	200	if (s->mode == REPLICATION_MODE_SECONDARY) {
	201	return ret;
	202	}
	203
	204	if (ret < 0) {
	205	s->error = ret;
	206	ret = 0;
	207	}
	208
	209	return ret;
	210	}
	211
	212	static coroutine_fn int replication_co_readv(BlockDriverState *bs,
	213	int64_t sector_num,
	214	int remaining_sectors,
	215	QEMUIOVector *qiov)
	216	{
	217	BDRVReplicationState *s = bs->opaque;
	218	BdrvChild *child = s->secondary_disk;
	219	BlockJob *job = NULL;
	220	CowRequest req;
	221	int ret;
	222
	223	if (s->mode == REPLICATION_MODE_PRIMARY) {
	224	/* We only use it to forward primary write requests */
	225	return -EIO;
	226	}
	227
	228	ret = replication_get_io_status(s);
	229	if (ret < 0) {
	230	return ret;
	231	}
	232
	233	if (child && child->bs) {
	234	job = child->bs->job;
	235	}
	236
	237	if (job) {
f6ac2078 EB	238	uint64_t remaining_bytes = remaining_sectors * BDRV_SECTOR_SIZE;
	239
	240	backup_wait_for_overlapping_requests(child->bs->job,
	241	sector_num * BDRV_SECTOR_SIZE,
	242	remaining_bytes);
	243	backup_cow_request_begin(&req, child->bs->job,
	244	sector_num * BDRV_SECTOR_SIZE,
	245	remaining_bytes);
29ff7890 WC	246	ret = bdrv_co_readv(bs->file, sector_num, remaining_sectors,
	247	qiov);
	248	backup_cow_request_end(&req);
	249	goto out;
	250	}
	251
	252	ret = bdrv_co_readv(bs->file, sector_num, remaining_sectors, qiov);
	253	out:
	254	return replication_return_value(s, ret);
	255	}
	256
	257	static coroutine_fn int replication_co_writev(BlockDriverState *bs,
	258	int64_t sector_num,
	259	int remaining_sectors,
	260	QEMUIOVector *qiov)
	261	{
	262	BDRVReplicationState *s = bs->opaque;
	263	QEMUIOVector hd_qiov;
	264	uint64_t bytes_done = 0;
	265	BdrvChild *top = bs->file;
	266	BdrvChild *base = s->secondary_disk;
	267	BdrvChild *target;
51b0a488 EB	268	int ret;
51b0a488 EB	269	int64_t n;
29ff7890 WC	270
	271	ret = replication_get_io_status(s);
	272	if (ret < 0) {
	273	goto out;
	274	}
	275
	276	if (ret == 0) {
	277	ret = bdrv_co_writev(top, sector_num,
	278	remaining_sectors, qiov);
	279	return replication_return_value(s, ret);
	280	}
	281
	282	/*
	283	* Failover failed, only write to active disk if the sectors
	284	* have already been allocated in active disk/hidden disk.
	285	*/
	286	qemu_iovec_init(&hd_qiov, qiov->niov);
	287	while (remaining_sectors > 0) {
51b0a488 EB	288	int64_t count;
	289
	290	ret = bdrv_is_allocated_above(top->bs, base->bs,
	291	sector_num * BDRV_SECTOR_SIZE,
	292	remaining_sectors * BDRV_SECTOR_SIZE,
	293	&count);
29ff7890 WC	294	if (ret < 0) {
	295	goto out1;
	296	}
	297
51b0a488 EB	298	assert(QEMU_IS_ALIGNED(count, BDRV_SECTOR_SIZE));
51b0a488 EB	299	n = count >> BDRV_SECTOR_BITS;
29ff7890	300	qemu_iovec_reset(&hd_qiov);
51b0a488	301	qemu_iovec_concat(&hd_qiov, qiov, bytes_done, count);
29ff7890 WC	302
	303	target = ret ? top : base;
	304	ret = bdrv_co_writev(target, sector_num, n, &hd_qiov);
	305	if (ret < 0) {
	306	goto out1;
	307	}
	308
	309	remaining_sectors -= n;
	310	sector_num += n;
51b0a488	311	bytes_done += count;
29ff7890 WC	312	}
	313
	314	out1:
	315	qemu_iovec_destroy(&hd_qiov);
	316	out:
	317	return ret;
	318	}
	319
	320	static bool replication_recurse_is_first_non_filter(BlockDriverState *bs,
	321	BlockDriverState *candidate)
	322	{
	323	return bdrv_recurse_is_first_non_filter(bs->file->bs, candidate);
	324	}
	325
	326	static void secondary_do_checkpoint(BDRVReplicationState s, Error *errp)
	327	{
	328	Error *local_err = NULL;
	329	int ret;
	330
	331	if (!s->secondary_disk->bs->job) {
	332	error_setg(errp, "Backup job was cancelled unexpectedly");
	333	return;
	334	}
	335
	336	backup_do_checkpoint(s->secondary_disk->bs->job, &local_err);
	337	if (local_err) {
	338	error_propagate(errp, local_err);
	339	return;
	340	}
	341
	342	ret = s->active_disk->bs->drv->bdrv_make_empty(s->active_disk->bs);
	343	if (ret < 0) {
	344	error_setg(errp, "Cannot make active disk empty");
	345	return;
	346	}
	347
	348	ret = s->hidden_disk->bs->drv->bdrv_make_empty(s->hidden_disk->bs);
	349	if (ret < 0) {
	350	error_setg(errp, "Cannot make hidden disk empty");
	351	return;
	352	}
	353	}
	354
8dd9006e	355	static void reopen_backing_file(BlockDriverState *bs, bool writable,
29ff7890 WC	356	Error **errp)
29ff7890 WC	357	{
8dd9006e	358	BDRVReplicationState *s = bs->opaque;
29ff7890 WC	359	BlockReopenQueue *reopen_queue = NULL;
	360	int orig_hidden_flags, orig_secondary_flags;
	361	int new_hidden_flags, new_secondary_flags;
	362	Error *local_err = NULL;
	363
	364	if (writable) {
	365	orig_hidden_flags = s->orig_hidden_flags =
	366	bdrv_get_flags(s->hidden_disk->bs);
	367	new_hidden_flags = (orig_hidden_flags \| BDRV_O_RDWR) &
	368	~BDRV_O_INACTIVE;
	369	orig_secondary_flags = s->orig_secondary_flags =
	370	bdrv_get_flags(s->secondary_disk->bs);
	371	new_secondary_flags = (orig_secondary_flags \| BDRV_O_RDWR) &
	372	~BDRV_O_INACTIVE;
	373	} else {
	374	orig_hidden_flags = (s->orig_hidden_flags \| BDRV_O_RDWR) &
	375	~BDRV_O_INACTIVE;
	376	new_hidden_flags = s->orig_hidden_flags;
	377	orig_secondary_flags = (s->orig_secondary_flags \| BDRV_O_RDWR) &
	378	~BDRV_O_INACTIVE;
	379	new_secondary_flags = s->orig_secondary_flags;
	380	}
	381
	382	if (orig_hidden_flags != new_hidden_flags) {
	383	reopen_queue = bdrv_reopen_queue(reopen_queue, s->hidden_disk->bs, NULL,
	384	new_hidden_flags);
	385	}
	386
	387	if (!(orig_secondary_flags & BDRV_O_RDWR)) {
	388	reopen_queue = bdrv_reopen_queue(reopen_queue, s->secondary_disk->bs,
	389	NULL, new_secondary_flags);
	390	}
	391
	392	if (reopen_queue) {
720150f3 PB	393	bdrv_reopen_multiple(bdrv_get_aio_context(bs),
720150f3 PB	394	reopen_queue, &local_err);
29ff7890 WC	395	error_propagate(errp, local_err);
	396	}
	397	}
	398
8dd9006e	399	static void backup_job_cleanup(BlockDriverState *bs)
29ff7890	400	{
8dd9006e	401	BDRVReplicationState *s = bs->opaque;
29ff7890 WC	402	BlockDriverState *top_bs;
	403
	404	top_bs = bdrv_lookup_bs(s->top_id, s->top_id, NULL);
	405	if (!top_bs) {
	406	return;
	407	}
	408	bdrv_op_unblock_all(top_bs, s->blocker);
	409	error_free(s->blocker);
8dd9006e	410	reopen_backing_file(bs, false, NULL);
29ff7890 WC	411	}
	412
	413	static void backup_job_completed(void *opaque, int ret)
	414	{
8dd9006e PB	415	BlockDriverState *bs = opaque;
8dd9006e PB	416	BDRVReplicationState *s = bs->opaque;
29ff7890	417
3c76c606	418	if (s->stage != BLOCK_REPLICATION_FAILOVER) {
29ff7890 WC	419	/* The backup job is cancelled unexpectedly */
	420	s->error = -EIO;
	421	}
	422
8dd9006e	423	backup_job_cleanup(bs);
29ff7890 WC	424	}
	425
	426	static bool check_top_bs(BlockDriverState top_bs, BlockDriverState bs)
	427	{
	428	BdrvChild *child;
	429
	430	/* The bs itself is the top_bs */
	431	if (top_bs == bs) {
	432	return true;
	433	}
	434
	435	/* Iterate over top_bs's children */
	436	QLIST_FOREACH(child, &top_bs->children, next) {
	437	if (child->bs == bs \|\| check_top_bs(child->bs, bs)) {
	438	return true;
	439	}
	440	}
	441
	442	return false;
	443	}
	444
	445	static void replication_start(ReplicationState *rs, ReplicationMode mode,
	446	Error **errp)
	447	{
	448	BlockDriverState *bs = rs->opaque;
	449	BDRVReplicationState *s;
	450	BlockDriverState *top_bs;
	451	int64_t active_length, hidden_length, disk_length;
	452	AioContext *aio_context;
	453	Error *local_err = NULL;
111049a4	454	BlockJob *job;
29ff7890 WC	455
	456	aio_context = bdrv_get_aio_context(bs);
	457	aio_context_acquire(aio_context);
	458	s = bs->opaque;
	459
3c76c606	460	if (s->stage != BLOCK_REPLICATION_NONE) {
29ff7890 WC	461	error_setg(errp, "Block replication is running or done");
	462	aio_context_release(aio_context);
	463	return;
	464	}
	465
	466	if (s->mode != mode) {
	467	error_setg(errp, "The parameter mode's value is invalid, needs %d,"
	468	" but got %d", s->mode, mode);
	469	aio_context_release(aio_context);
	470	return;
	471	}
	472
	473	switch (s->mode) {
	474	case REPLICATION_MODE_PRIMARY:
	475	break;
	476	case REPLICATION_MODE_SECONDARY:
	477	s->active_disk = bs->file;
	478	if (!s->active_disk \|\| !s->active_disk->bs \|\|
	479	!s->active_disk->bs->backing) {
	480	error_setg(errp, "Active disk doesn't have backing file");
	481	aio_context_release(aio_context);
	482	return;
	483	}
	484
	485	s->hidden_disk = s->active_disk->bs->backing;
	486	if (!s->hidden_disk->bs \|\| !s->hidden_disk->bs->backing) {
	487	error_setg(errp, "Hidden disk doesn't have backing file");
	488	aio_context_release(aio_context);
	489	return;
	490	}
	491
	492	s->secondary_disk = s->hidden_disk->bs->backing;
	493	if (!s->secondary_disk->bs \|\| !bdrv_has_blk(s->secondary_disk->bs)) {
	494	error_setg(errp, "The secondary disk doesn't have block backend");
	495	aio_context_release(aio_context);
	496	return;
	497	}
	498
	499	/* verify the length */
	500	active_length = bdrv_getlength(s->active_disk->bs);
	501	hidden_length = bdrv_getlength(s->hidden_disk->bs);
	502	disk_length = bdrv_getlength(s->secondary_disk->bs);
	503	if (active_length < 0 \|\| hidden_length < 0 \|\| disk_length < 0 \|\|
	504	active_length != hidden_length \|\| hidden_length != disk_length) {
	505	error_setg(errp, "Active disk, hidden disk, secondary disk's length"
	506	" are not the same");
	507	aio_context_release(aio_context);
	508	return;
	509	}
	510
	511	if (!s->active_disk->bs->drv->bdrv_make_empty \|\|
	512	!s->hidden_disk->bs->drv->bdrv_make_empty) {
	513	error_setg(errp,
	514	"Active disk or hidden disk doesn't support make_empty");
	515	aio_context_release(aio_context);
	516	return;
	517	}
	518
	519	/* reopen the backing file in r/w mode */
8dd9006e	520	reopen_backing_file(bs, true, &local_err);
29ff7890 WC	521	if (local_err) {
	522	error_propagate(errp, local_err);
	523	aio_context_release(aio_context);
	524	return;
	525	}
	526
	527	/* start backup job now */
	528	error_setg(&s->blocker,
	529	"Block device is in use by internal backup job");
	530
	531	top_bs = bdrv_lookup_bs(s->top_id, s->top_id, NULL);
	532	if (!top_bs \|\| !bdrv_is_root_node(top_bs) \|\|
	533	!check_top_bs(top_bs, bs)) {
	534	error_setg(errp, "No top_bs or it is invalid");
8dd9006e	535	reopen_backing_file(bs, false, NULL);
29ff7890 WC	536	aio_context_release(aio_context);
	537	return;
	538	}
	539	bdrv_op_block_all(top_bs, s->blocker);
	540	bdrv_op_unblock(top_bs, BLOCK_OP_TYPE_DATAPLANE, s->blocker);
	541
111049a4 JS	542	job = backup_job_create(NULL, s->secondary_disk->bs, s->hidden_disk->bs,
	543	0, MIRROR_SYNC_MODE_NONE, NULL, false,
	544	BLOCKDEV_ON_ERROR_REPORT,
	545	BLOCKDEV_ON_ERROR_REPORT, BLOCK_JOB_INTERNAL,
	546	backup_job_completed, bs, NULL, &local_err);
29ff7890 WC	547	if (local_err) {
29ff7890 WC	548	error_propagate(errp, local_err);
8dd9006e	549	backup_job_cleanup(bs);
29ff7890 WC	550	aio_context_release(aio_context);
	551	return;
	552	}
111049a4	553	block_job_start(job);
29ff7890 WC	554	break;
	555	default:
	556	aio_context_release(aio_context);
	557	abort();
	558	}
	559
3c76c606	560	s->stage = BLOCK_REPLICATION_RUNNING;
29ff7890 WC	561
	562	if (s->mode == REPLICATION_MODE_SECONDARY) {
	563	secondary_do_checkpoint(s, errp);
	564	}
	565
	566	s->error = 0;
	567	aio_context_release(aio_context);
	568	}
	569
	570	static void replication_do_checkpoint(ReplicationState rs, Error *errp)
	571	{
	572	BlockDriverState *bs = rs->opaque;
	573	BDRVReplicationState *s;
	574	AioContext *aio_context;
	575
	576	aio_context = bdrv_get_aio_context(bs);
	577	aio_context_acquire(aio_context);
	578	s = bs->opaque;
	579
	580	if (s->mode == REPLICATION_MODE_SECONDARY) {
	581	secondary_do_checkpoint(s, errp);
	582	}
	583	aio_context_release(aio_context);
	584	}
	585
	586	static void replication_get_error(ReplicationState rs, Error *errp)
	587	{
	588	BlockDriverState *bs = rs->opaque;
	589	BDRVReplicationState *s;
	590	AioContext *aio_context;
	591
	592	aio_context = bdrv_get_aio_context(bs);
	593	aio_context_acquire(aio_context);
	594	s = bs->opaque;
	595
3c76c606	596	if (s->stage != BLOCK_REPLICATION_RUNNING) {
29ff7890 WC	597	error_setg(errp, "Block replication is not running");
	598	aio_context_release(aio_context);
	599	return;
	600	}
	601
	602	if (s->error) {
	603	error_setg(errp, "I/O error occurred");
	604	aio_context_release(aio_context);
	605	return;
	606	}
	607	aio_context_release(aio_context);
	608	}
	609
	610	static void replication_done(void *opaque, int ret)
	611	{
	612	BlockDriverState *bs = opaque;
	613	BDRVReplicationState *s = bs->opaque;
	614
	615	if (ret == 0) {
3c76c606	616	s->stage = BLOCK_REPLICATION_DONE;
29ff7890 WC	617
	618	/* refresh top bs's filename */
	619	bdrv_refresh_filename(bs);
	620	s->active_disk = NULL;
	621	s->secondary_disk = NULL;
	622	s->hidden_disk = NULL;
	623	s->error = 0;
	624	} else {
3c76c606	625	s->stage = BLOCK_REPLICATION_FAILOVER_FAILED;
29ff7890 WC	626	s->error = -EIO;
	627	}
	628	}
	629
	630	static void replication_stop(ReplicationState rs, bool failover, Error *errp)
	631	{
	632	BlockDriverState *bs = rs->opaque;
	633	BDRVReplicationState *s;
	634	AioContext *aio_context;
	635
	636	aio_context = bdrv_get_aio_context(bs);
	637	aio_context_acquire(aio_context);
	638	s = bs->opaque;
	639
3c76c606	640	if (s->stage != BLOCK_REPLICATION_RUNNING) {
29ff7890 WC	641	error_setg(errp, "Block replication is not running");
	642	aio_context_release(aio_context);
	643	return;
	644	}
	645
	646	switch (s->mode) {
	647	case REPLICATION_MODE_PRIMARY:
3c76c606	648	s->stage = BLOCK_REPLICATION_DONE;
29ff7890 WC	649	s->error = 0;
	650	break;
	651	case REPLICATION_MODE_SECONDARY:
	652	/*
	653	* This BDS will be closed, and the job should be completed
	654	* before the BDS is closed, because we will access hidden
	655	* disk, secondary disk in backup_job_completed().
	656	*/
	657	if (s->secondary_disk->bs->job) {
	658	block_job_cancel_sync(s->secondary_disk->bs->job);
	659	}
	660
	661	if (!failover) {
	662	secondary_do_checkpoint(s, errp);
3c76c606	663	s->stage = BLOCK_REPLICATION_DONE;
29ff7890 WC	664	aio_context_release(aio_context);
	665	return;
	666	}
	667
3c76c606	668	s->stage = BLOCK_REPLICATION_FAILOVER;
47970dfb JS	669	commit_active_start(NULL, s->active_disk->bs, s->secondary_disk->bs,
47970dfb JS	670	BLOCK_JOB_INTERNAL, 0, BLOCKDEV_ON_ERROR_REPORT,
78bbd910	671	NULL, replication_done, bs, true, errp);
29ff7890 WC	672	break;
	673	default:
	674	aio_context_release(aio_context);
	675	abort();
	676	}
	677	aio_context_release(aio_context);
	678	}
	679
	680	BlockDriver bdrv_replication = {
	681	.format_name = "replication",
	682	.protocol_name = "replication",
	683	.instance_size = sizeof(BDRVReplicationState),
	684
	685	.bdrv_open = replication_open,
	686	.bdrv_close = replication_close,
37a9051c	687	.bdrv_child_perm = replication_child_perm,
29ff7890 WC	688
	689	.bdrv_getlength = replication_getlength,
	690	.bdrv_co_readv = replication_co_readv,
	691	.bdrv_co_writev = replication_co_writev,
	692
	693	.is_filter = true,
	694	.bdrv_recurse_is_first_non_filter = replication_recurse_is_first_non_filter,
	695
	696	.has_variable_length = true,
	697	};
	698
	699	static void bdrv_replication_init(void)
	700	{
	701	bdrv_register(&bdrv_replication);
	702	}
	703
	704	block_init(bdrv_replication_init);