/*
 * block_copy API
 *
 * Copyright (C) 2013 Proxmox Server Solutions
 * Copyright (c) 2019 Virtuozzo International GmbH.
 *
 * Authors:
 *  Dietmar Maurer (dietmar@proxmox.com)
 *  Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.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 "trace.h"
#include "qapi/error.h"
#include "block/block-copy.h"
#include "block/block_int-io.h"
#include "block/dirty-bitmap.h"
#include "block/reqlist.h"
#include "sysemu/block-backend.h"
#include "qemu/units.h"
#include "qemu/co-shared-resource.h"
#include "qemu/coroutine.h"
#include "qemu/ratelimit.h"
#include "block/aio_task.h"
#include "qemu/error-report.h"
#include "qemu/memalign.h"

#define BLOCK_COPY_MAX_COPY_RANGE (16 * MiB)
#define BLOCK_COPY_MAX_BUFFER (1 * MiB)
#define BLOCK_COPY_MAX_MEM (128 * MiB)
#define BLOCK_COPY_MAX_WORKERS 64
#define BLOCK_COPY_SLICE_TIME 100000000ULL /* ns */
#define BLOCK_COPY_CLUSTER_SIZE_DEFAULT (1 << 16)

typedef enum {
    COPY_READ_WRITE_CLUSTER,
    COPY_READ_WRITE,
    COPY_WRITE_ZEROES,
    COPY_RANGE_SMALL,
    COPY_RANGE_FULL
} BlockCopyMethod;

static coroutine_fn int block_copy_task_entry(AioTask *task);

typedef struct BlockCopyCallState {
    /* Fields initialized in block_copy_async() and never changed. */
    BlockCopyState *s;
    int64_t offset;
    int64_t bytes;
    int max_workers;
    int64_t max_chunk;
    bool ignore_ratelimit;
    BlockCopyAsyncCallbackFunc cb;
    void *cb_opaque;
    /* Coroutine where async block-copy is running */
    Coroutine *co;

    /* Fields whose state changes throughout the execution */
    bool finished; /* atomic */
    QemuCoSleep sleep; /* TODO: protect API with a lock */
    bool cancelled; /* atomic */
    /* To reference all call states from BlockCopyState */
    QLIST_ENTRY(BlockCopyCallState) list;

    /*
     * Fields that report information about return values and errors.
     * Protected by lock in BlockCopyState.
     */
    bool error_is_read;
    /*
     * @ret is set concurrently by tasks under mutex. Only set once by first
     * failed task (and untouched if no task failed).
     * After finishing (call_state->finished is true), it is not modified
     * anymore and may be safely read without mutex.
     */
    int ret;
} BlockCopyCallState;

typedef struct BlockCopyTask {
    AioTask task;

    /*
     * Fields initialized in block_copy_task_create()
     * and never changed.
     */
    BlockCopyState *s;
    BlockCopyCallState *call_state;
    /*
     * @method can also be set again in the while loop of
     * block_copy_dirty_clusters(), but it is never accessed concurrently
     * because the only other function that reads it is
     * block_copy_task_entry() and it is invoked afterwards in the same
     * iteration.
     */
    BlockCopyMethod method;

    /*
     * Generally, req is protected by lock in BlockCopyState, Still req.offset
     * is only set on task creation, so may be read concurrently after creation.
     * req.bytes is changed at most once, and need only protecting the case of
     * parallel read while updating @bytes value in block_copy_task_shrink().
     */
    BlockReq req;
} BlockCopyTask;

static int64_t task_end(BlockCopyTask *task)
{
    return task->req.offset + task->req.bytes;
}

typedef struct BlockCopyState {
    /*
     * BdrvChild objects are not owned or managed by block-copy. They are
     * provided by block-copy user and user is responsible for appropriate
     * permissions on these children.
     */
    BdrvChild *source;
    BdrvChild *target;

    /*
     * Fields initialized in block_copy_state_new()
     * and never changed.
     */
    int64_t cluster_size;
    int64_t max_transfer;
    uint64_t len;
    BdrvRequestFlags write_flags;

    /*
     * Fields whose state changes throughout the execution
     * Protected by lock.
     */
    CoMutex lock;
    int64_t in_flight_bytes;
    BlockCopyMethod method;
    bool discard_source;
    BlockReqList reqs;
    QLIST_HEAD(, BlockCopyCallState) calls;
    /*
     * skip_unallocated:
     *
     * Used by sync=top jobs, which first scan the source node for unallocated
     * areas and clear them in the copy_bitmap.  During this process, the bitmap
     * is thus not fully initialized: It may still have bits set for areas that
     * are unallocated and should actually not be copied.
     *
     * This is indicated by skip_unallocated.
     *
     * In this case, block_copy() will query the source’s allocation status,
     * skip unallocated regions, clear them in the copy_bitmap, and invoke
     * block_copy_reset_unallocated() every time it does.
     */
    bool skip_unallocated; /* atomic */
    /* State fields that use a thread-safe API */
    BdrvDirtyBitmap *copy_bitmap;
    ProgressMeter *progress;
    SharedResource *mem;
    RateLimit rate_limit;
} BlockCopyState;

/* Called with lock held */
static int64_t block_copy_chunk_size(BlockCopyState *s)
{
    switch (s->method) {
    case COPY_READ_WRITE_CLUSTER:
        return s->cluster_size;
    case COPY_READ_WRITE:
    case COPY_RANGE_SMALL:
        return MIN(MAX(s->cluster_size, BLOCK_COPY_MAX_BUFFER),
                   s->max_transfer);
    case COPY_RANGE_FULL:
        return MIN(MAX(s->cluster_size, BLOCK_COPY_MAX_COPY_RANGE),
                   s->max_transfer);
    default:
        /* Cannot have COPY_WRITE_ZEROES here.  */
        abort();
    }
}

/*
 * Search for the first dirty area in offset/bytes range and create task at
 * the beginning of it.
 */
static coroutine_fn BlockCopyTask *
block_copy_task_create(BlockCopyState *s, BlockCopyCallState *call_state,
                       int64_t offset, int64_t bytes)
{
    BlockCopyTask *task;
    int64_t max_chunk;

    QEMU_LOCK_GUARD(&s->lock);
    max_chunk = MIN_NON_ZERO(block_copy_chunk_size(s), call_state->max_chunk);
    if (!bdrv_dirty_bitmap_next_dirty_area(s->copy_bitmap,
                                           offset, offset + bytes,
                                           max_chunk, &offset, &bytes))
    {
        return NULL;
    }

    assert(QEMU_IS_ALIGNED(offset, s->cluster_size));
    bytes = QEMU_ALIGN_UP(bytes, s->cluster_size);

    /* region is dirty, so no existent tasks possible in it */
    assert(!reqlist_find_conflict(&s->reqs, offset, bytes));

    bdrv_reset_dirty_bitmap(s->copy_bitmap, offset, bytes);
    s->in_flight_bytes += bytes;

    task = g_new(BlockCopyTask, 1);
    *task = (BlockCopyTask) {
        .task.func = block_copy_task_entry,
        .s = s,
        .call_state = call_state,
        .method = s->method,
    };
    reqlist_init_req(&s->reqs, &task->req, offset, bytes);

    return task;
}

/*
 * block_copy_task_shrink
 *
 * Drop the tail of the task to be handled later. Set dirty bits back and
 * wake up all tasks waiting for us (may be some of them are not intersecting
 * with shrunk task)
 */
static void coroutine_fn block_copy_task_shrink(BlockCopyTask *task,
                                                int64_t new_bytes)
{
    QEMU_LOCK_GUARD(&task->s->lock);
    if (new_bytes == task->req.bytes) {
        return;
    }

    assert(new_bytes > 0 && new_bytes < task->req.bytes);

    task->s->in_flight_bytes -= task->req.bytes - new_bytes;
    bdrv_set_dirty_bitmap(task->s->copy_bitmap,
                          task->req.offset + new_bytes,
                          task->req.bytes - new_bytes);

    reqlist_shrink_req(&task->req, new_bytes);
}

static void coroutine_fn block_copy_task_end(BlockCopyTask *task, int ret)
{
    QEMU_LOCK_GUARD(&task->s->lock);
    task->s->in_flight_bytes -= task->req.bytes;
    if (ret < 0) {
        bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->req.offset,
                              task->req.bytes);
    }
    if (task->s->progress) {
        progress_set_remaining(task->s->progress,
                               bdrv_get_dirty_count(task->s->copy_bitmap) +
                               task->s->in_flight_bytes);
    }
    reqlist_remove_req(&task->req);
}

void block_copy_state_free(BlockCopyState *s)
{
    if (!s) {
        return;
    }

    ratelimit_destroy(&s->rate_limit);
    bdrv_release_dirty_bitmap(s->copy_bitmap);
    shres_destroy(s->mem);
    g_free(s);
}

static uint32_t block_copy_max_transfer(BdrvChild *source, BdrvChild *target)
{
    return MIN_NON_ZERO(INT_MAX,
                        MIN_NON_ZERO(source->bs->bl.max_transfer,
                                     target->bs->bl.max_transfer));
}

void block_copy_set_copy_opts(BlockCopyState *s, bool use_copy_range,
                              bool compress)
{
    /* Keep BDRV_REQ_SERIALISING set (or not set) in block_copy_state_new() */
    s->write_flags = (s->write_flags & BDRV_REQ_SERIALISING) |
        (compress ? BDRV_REQ_WRITE_COMPRESSED : 0);

    if (s->max_transfer < s->cluster_size) {
        /*
         * copy_range does not respect max_transfer. We don't want to bother
         * with requests smaller than block-copy cluster size, so fallback to
         * buffered copying (read and write respect max_transfer on their
         * behalf).
         */
        s->method = COPY_READ_WRITE_CLUSTER;
    } else if (compress) {
        /* Compression supports only cluster-size writes and no copy-range. */
        s->method = COPY_READ_WRITE_CLUSTER;
    } else {
        /*
         * If copy range enabled, start with COPY_RANGE_SMALL, until first
         * successful copy_range (look at block_copy_do_copy).
         */
        s->method = use_copy_range ? COPY_RANGE_SMALL : COPY_READ_WRITE;
    }
}

static int64_t block_copy_calculate_cluster_size(BlockDriverState *target,
                                                 int64_t min_cluster_size,
                                                 Error **errp)
{
    int ret;
    BlockDriverInfo bdi;
    bool target_does_cow;

    GLOBAL_STATE_CODE();
    GRAPH_RDLOCK_GUARD_MAINLOOP();

    min_cluster_size = MAX(min_cluster_size,
                           (int64_t)BLOCK_COPY_CLUSTER_SIZE_DEFAULT);

    target_does_cow = bdrv_backing_chain_next(target);

    /*
     * If there is no backing file on the target, we cannot rely on COW if our
     * backup cluster size is smaller than the target cluster size. Even for
     * targets with a backing file, try to avoid COW if possible.
     */
    ret = bdrv_get_info(target, &bdi);
    if (ret == -ENOTSUP && !target_does_cow) {
        /* Cluster size is not defined */
        warn_report("The target block device doesn't provide information about "
                    "the block size and it doesn't have a backing file. The "
                    "(default) block size of %" PRIi64 " bytes is used. If the "
                    "actual block size of the target exceeds this value, the "
                    "backup may be unusable",
                    min_cluster_size);
        return min_cluster_size;
    } else if (ret < 0 && !target_does_cow) {
        error_setg_errno(errp, -ret,
            "Couldn't determine the cluster size of the target image, "
            "which has no backing file");
        error_append_hint(errp,
            "Aborting, since this may create an unusable destination image\n");
        return ret;
    } else if (ret < 0 && target_does_cow) {
        /* Not fatal; just trudge on ahead. */
        return min_cluster_size;
    }

    return MAX(min_cluster_size, bdi.cluster_size);
}

BlockCopyState *block_copy_state_new(BdrvChild *source, BdrvChild *target,
                                     BlockDriverState *copy_bitmap_bs,
                                     const BdrvDirtyBitmap *bitmap,
                                     bool discard_source,
                                     uint64_t min_cluster_size,
                                     Error **errp)
{
    ERRP_GUARD();
    BlockCopyState *s;
    int64_t cluster_size;
    BdrvDirtyBitmap *copy_bitmap;
    bool is_fleecing;

    GLOBAL_STATE_CODE();

    if (min_cluster_size > INT64_MAX) {
        error_setg(errp, "min-cluster-size too large: %" PRIu64 " > %" PRIi64,
                   min_cluster_size, INT64_MAX);
        return NULL;
    } else if (min_cluster_size && !is_power_of_2(min_cluster_size)) {
        error_setg(errp, "min-cluster-size needs to be a power of 2");
        return NULL;
    }

    cluster_size = block_copy_calculate_cluster_size(target->bs,
                                                     (int64_t)min_cluster_size,
                                                     errp);
    if (cluster_size < 0) {
        return NULL;
    }

    copy_bitmap = bdrv_create_dirty_bitmap(copy_bitmap_bs, cluster_size, NULL,
                                           errp);
    if (!copy_bitmap) {
        return NULL;
    }
    bdrv_disable_dirty_bitmap(copy_bitmap);
    if (bitmap) {
        if (!bdrv_merge_dirty_bitmap(copy_bitmap, bitmap, NULL, errp)) {
            error_prepend(errp, "Failed to merge bitmap '%s' to internal "
                          "copy-bitmap: ", bdrv_dirty_bitmap_name(bitmap));
            bdrv_release_dirty_bitmap(copy_bitmap);
            return NULL;
        }
    } else {
        bdrv_set_dirty_bitmap(copy_bitmap, 0,
                              bdrv_dirty_bitmap_size(copy_bitmap));
    }

    /*
     * If source is in backing chain of target assume that target is going to be
     * used for "image fleecing", i.e. it should represent a kind of snapshot of
     * source at backup-start point in time. And target is going to be read by
     * somebody (for example, used as NBD export) during backup job.
     *
     * In this case, we need to add BDRV_REQ_SERIALISING write flag to avoid
     * intersection of backup writes and third party reads from target,
     * otherwise reading from target we may occasionally read already updated by
     * guest data.
     *
     * For more information see commit f8d59dfb40bb and test
     * tests/qemu-iotests/222
     */
    bdrv_graph_rdlock_main_loop();
    is_fleecing = bdrv_chain_contains(target->bs, source->bs);
    bdrv_graph_rdunlock_main_loop();

    s = g_new(BlockCopyState, 1);
    *s = (BlockCopyState) {
        .source = source,
        .target = target,
        .copy_bitmap = copy_bitmap,
        .cluster_size = cluster_size,
        .len = bdrv_dirty_bitmap_size(copy_bitmap),
        .write_flags = (is_fleecing ? BDRV_REQ_SERIALISING : 0),
        .mem = shres_create(BLOCK_COPY_MAX_MEM),
        .max_transfer = QEMU_ALIGN_DOWN(
                                    block_copy_max_transfer(source, target),
                                    cluster_size),
    };

    s->discard_source = discard_source;
    block_copy_set_copy_opts(s, false, false);

    ratelimit_init(&s->rate_limit);
    qemu_co_mutex_init(&s->lock);
    QLIST_INIT(&s->reqs);
    QLIST_INIT(&s->calls);

    return s;
}

/* Only set before running the job, no need for locking. */
void block_copy_set_progress_meter(BlockCopyState *s, ProgressMeter *pm)
{
    s->progress = pm;
}

/*
 * Takes ownership of @task
 *
 * If pool is NULL directly run the task, otherwise schedule it into the pool.
 *
 * Returns: task.func return code if pool is NULL
 *          otherwise -ECANCELED if pool status is bad
 *          otherwise 0 (successfully scheduled)
 */
static coroutine_fn int block_copy_task_run(AioTaskPool *pool,
                                            BlockCopyTask *task)
{
    if (!pool) {
        int ret = task->task.func(&task->task);

        g_free(task);
        return ret;
    }

    aio_task_pool_wait_slot(pool);
    if (aio_task_pool_status(pool) < 0) {
        co_put_to_shres(task->s->mem, task->req.bytes);
        block_copy_task_end(task, -ECANCELED);
        g_free(task);
        return -ECANCELED;
    }

    aio_task_pool_start_task(pool, &task->task);

    return 0;
}

/*
 * block_copy_do_copy
 *
 * Do copy of cluster-aligned chunk. Requested region is allowed to exceed
 * s->len only to cover last cluster when s->len is not aligned to clusters.
 *
 * No sync here: neither bitmap nor intersecting requests handling, only copy.
 *
 * @method is an in-out argument, so that copy_range can be either extended to
 * a full-size buffer or disabled if the copy_range attempt fails.  The output
 * value of @method should be used for subsequent tasks.
 * Returns 0 on success.
 */
static int coroutine_fn GRAPH_RDLOCK
block_copy_do_copy(BlockCopyState *s, int64_t offset, int64_t bytes,
                   BlockCopyMethod *method, bool *error_is_read)
{
    int ret;
    int64_t nbytes = MIN(offset + bytes, s->len) - offset;
    void *bounce_buffer = NULL;

    assert(offset >= 0 && bytes > 0 && INT64_MAX - offset >= bytes);
    assert(QEMU_IS_ALIGNED(offset, s->cluster_size));
    assert(QEMU_IS_ALIGNED(bytes, s->cluster_size));
    assert(offset < s->len);
    assert(offset + bytes <= s->len ||
           offset + bytes == QEMU_ALIGN_UP(s->len, s->cluster_size));
    assert(nbytes < INT_MAX);

    switch (*method) {
    case COPY_WRITE_ZEROES:
        ret = bdrv_co_pwrite_zeroes(s->target, offset, nbytes, s->write_flags &
                                    ~BDRV_REQ_WRITE_COMPRESSED);
        if (ret < 0) {
            trace_block_copy_write_zeroes_fail(s, offset, ret);
            *error_is_read = false;
        }
        return ret;

    case COPY_RANGE_SMALL:
    case COPY_RANGE_FULL:
        ret = bdrv_co_copy_range(s->source, offset, s->target, offset, nbytes,
                                 0, s->write_flags);
        if (ret >= 0) {
            /* Successful copy-range, increase chunk size.  */
            *method = COPY_RANGE_FULL;
            return 0;
        }

        trace_block_copy_copy_range_fail(s, offset, ret);
        *method = COPY_READ_WRITE;
        /* Fall through to read+write with allocated buffer */

    case COPY_READ_WRITE_CLUSTER:
    case COPY_READ_WRITE:
        /*
         * In case of failed copy_range request above, we may proceed with
         * buffered request larger than BLOCK_COPY_MAX_BUFFER.
         * Still, further requests will be properly limited, so don't care too
         * much. Moreover the most likely case (copy_range is unsupported for
         * the configuration, so the very first copy_range request fails)
         * is handled by setting large copy_size only after first successful
         * copy_range.
         */

        bounce_buffer = qemu_blockalign(s->source->bs, nbytes);

        ret = bdrv_co_pread(s->source, offset, nbytes, bounce_buffer, 0);
        if (ret < 0) {
            trace_block_copy_read_fail(s, offset, ret);
            *error_is_read = true;
            goto out;
        }

        ret = bdrv_co_pwrite(s->target, offset, nbytes, bounce_buffer,
                             s->write_flags);
        if (ret < 0) {
            trace_block_copy_write_fail(s, offset, ret);
            *error_is_read = false;
            goto out;
        }

    out:
        qemu_vfree(bounce_buffer);
        break;

    default:
        abort();
    }

    return ret;
}

static coroutine_fn int block_copy_task_entry(AioTask *task)
{
    BlockCopyTask *t = container_of(task, BlockCopyTask, task);
    BlockCopyState *s = t->s;
    bool error_is_read = false;
    BlockCopyMethod method = t->method;
    int ret;

    WITH_GRAPH_RDLOCK_GUARD() {
        ret = block_copy_do_copy(s, t->req.offset, t->req.bytes, &method,
                                 &error_is_read);
    }

    WITH_QEMU_LOCK_GUARD(&s->lock) {
        if (s->method == t->method) {
            s->method = method;
        }

        if (ret < 0) {
            if (!t->call_state->ret) {
                t->call_state->ret = ret;
                t->call_state->error_is_read = error_is_read;
            }
        } else if (s->progress) {
            progress_work_done(s->progress, t->req.bytes);
        }
    }
    co_put_to_shres(s->mem, t->req.bytes);
    block_copy_task_end(t, ret);

    if (s->discard_source && ret == 0) {
        int64_t nbytes =
            MIN(t->req.offset + t->req.bytes, s->len) - t->req.offset;
        WITH_GRAPH_RDLOCK_GUARD() {
            bdrv_co_pdiscard(s->source, t->req.offset, nbytes);
        }
    }

    return ret;
}

static coroutine_fn GRAPH_RDLOCK
int block_copy_block_status(BlockCopyState *s, int64_t offset, int64_t bytes,
                            int64_t *pnum)
{
    int64_t num;
    BlockDriverState *base;
    int ret;

    if (qatomic_read(&s->skip_unallocated)) {
        base = bdrv_backing_chain_next(s->source->bs);
    } else {
        base = NULL;
    }

    ret = bdrv_co_block_status_above(s->source->bs, base, offset, bytes, &num,
                                     NULL, NULL);
    if (ret < 0 || num < s->cluster_size) {
        /*
         * On error or if failed to obtain large enough chunk just fallback to
         * copy one cluster.
         */
        num = s->cluster_size;
        ret = BDRV_BLOCK_ALLOCATED | BDRV_BLOCK_DATA;
    } else if (offset + num == s->len) {
        num = QEMU_ALIGN_UP(num, s->cluster_size);
    } else {
        num = QEMU_ALIGN_DOWN(num, s->cluster_size);
    }

    *pnum = num;
    return ret;
}

/*
 * Check if the cluster starting at offset is allocated or not.
 * return via pnum the number of contiguous clusters sharing this allocation.
 */
static int coroutine_fn GRAPH_RDLOCK
block_copy_is_cluster_allocated(BlockCopyState *s, int64_t offset,
                                int64_t *pnum)
{
    BlockDriverState *bs = s->source->bs;
    int64_t count, total_count = 0;
    int64_t bytes = s->len - offset;
    int ret;

    assert(QEMU_IS_ALIGNED(offset, s->cluster_size));

    while (true) {
        /* protected in backup_run() */
        ret = bdrv_co_is_allocated(bs, offset, bytes, &count);
        if (ret < 0) {
            return ret;
        }

        total_count += count;

        if (ret || count == 0) {
            /*
             * ret: partial segment(s) are considered allocated.
             * otherwise: unallocated tail is treated as an entire segment.
             */
            *pnum = DIV_ROUND_UP(total_count, s->cluster_size);
            return ret;
        }

        /* Unallocated segment(s) with uncertain following segment(s) */
        if (total_count >= s->cluster_size) {
            *pnum = total_count / s->cluster_size;
            return 0;
        }

        offset += count;
        bytes -= count;
    }
}

void block_copy_reset(BlockCopyState *s, int64_t offset, int64_t bytes)
{
    QEMU_LOCK_GUARD(&s->lock);

    bdrv_reset_dirty_bitmap(s->copy_bitmap, offset, bytes);
    if (s->progress) {
        progress_set_remaining(s->progress,
                               bdrv_get_dirty_count(s->copy_bitmap) +
                               s->in_flight_bytes);
    }
}

/*
 * Reset bits in copy_bitmap starting at offset if they represent unallocated
 * data in the image. May reset subsequent contiguous bits.
 * @return 0 when the cluster at @offset was unallocated,
 *         1 otherwise, and -ret on error.
 */
int64_t coroutine_fn block_copy_reset_unallocated(BlockCopyState *s,
                                                  int64_t offset,
                                                  int64_t *count)
{
    int ret;
    int64_t clusters, bytes;

    ret = block_copy_is_cluster_allocated(s, offset, &clusters);
    if (ret < 0) {
        return ret;
    }

    bytes = clusters * s->cluster_size;

    if (!ret) {
        block_copy_reset(s, offset, bytes);
    }

    *count = bytes;
    return ret;
}

/*
 * block_copy_dirty_clusters
 *
 * Copy dirty clusters in @offset/@bytes range.
 * Returns 1 if dirty clusters found and successfully copied, 0 if no dirty
 * clusters found and -errno on failure.
 */
static int coroutine_fn GRAPH_RDLOCK
block_copy_dirty_clusters(BlockCopyCallState *call_state)
{
    BlockCopyState *s = call_state->s;
    int64_t offset = call_state->offset;
    int64_t bytes = call_state->bytes;

    int ret = 0;
    bool found_dirty = false;
    int64_t end = offset + bytes;
    AioTaskPool *aio = NULL;

    /*
     * block_copy() user is responsible for keeping source and target in same
     * aio context
     */
    assert(bdrv_get_aio_context(s->source->bs) ==
           bdrv_get_aio_context(s->target->bs));

    assert(QEMU_IS_ALIGNED(offset, s->cluster_size));
    assert(QEMU_IS_ALIGNED(bytes, s->cluster_size));

    while (bytes && aio_task_pool_status(aio) == 0 &&
           !qatomic_read(&call_state->cancelled)) {
        BlockCopyTask *task;
        int64_t status_bytes;

        task = block_copy_task_create(s, call_state, offset, bytes);
        if (!task) {
            /* No more dirty bits in the bitmap */
            trace_block_copy_skip_range(s, offset, bytes);
            break;
        }
        if (task->req.offset > offset) {
            trace_block_copy_skip_range(s, offset, task->req.offset - offset);
        }

        found_dirty = true;

        ret = block_copy_block_status(s, task->req.offset, task->req.bytes,
                                      &status_bytes);
        assert(ret >= 0); /* never fail */
        if (status_bytes < task->req.bytes) {
            block_copy_task_shrink(task, status_bytes);
        }
        if (qatomic_read(&s->skip_unallocated) &&
            !(ret & BDRV_BLOCK_ALLOCATED)) {
            block_copy_task_end(task, 0);
            trace_block_copy_skip_range(s, task->req.offset, task->req.bytes);
            offset = task_end(task);
            bytes = end - offset;
            g_free(task);
            continue;
        }
        if (ret & BDRV_BLOCK_ZERO) {
            task->method = COPY_WRITE_ZEROES;
        }

        if (!call_state->ignore_ratelimit) {
            uint64_t ns = ratelimit_calculate_delay(&s->rate_limit, 0);
            if (ns > 0) {
                block_copy_task_end(task, -EAGAIN);
                g_free(task);
                qemu_co_sleep_ns_wakeable(&call_state->sleep,
                                          QEMU_CLOCK_REALTIME, ns);
                continue;
            }
        }

        ratelimit_calculate_delay(&s->rate_limit, task->req.bytes);

        trace_block_copy_process(s, task->req.offset);

        co_get_from_shres(s->mem, task->req.bytes);

        offset = task_end(task);
        bytes = end - offset;

        if (!aio && bytes) {
            aio = aio_task_pool_new(call_state->max_workers);
        }

        ret = block_copy_task_run(aio, task);
        if (ret < 0) {
            goto out;
        }
    }

out:
    if (aio) {
        aio_task_pool_wait_all(aio);

        /*
         * We are not really interested in -ECANCELED returned from
         * block_copy_task_run. If it fails, it means some task already failed
         * for real reason, let's return first failure.
         * Still, assert that we don't rewrite failure by success.
         *
         * Note: ret may be positive here because of block-status result.
         */
        assert(ret >= 0 || aio_task_pool_status(aio) < 0);
        ret = aio_task_pool_status(aio);

        aio_task_pool_free(aio);
    }

    return ret < 0 ? ret : found_dirty;
}

void block_copy_kick(BlockCopyCallState *call_state)
{
    qemu_co_sleep_wake(&call_state->sleep);
}

/*
 * block_copy_common
 *
 * Copy requested region, accordingly to dirty bitmap.
 * Collaborate with parallel block_copy requests: if they succeed it will help
 * us. If they fail, we will retry not-copied regions. So, if we return error,
 * it means that some I/O operation failed in context of _this_ block_copy call,
 * not some parallel operation.
 */
static int coroutine_fn GRAPH_RDLOCK
block_copy_common(BlockCopyCallState *call_state)
{
    int ret;
    BlockCopyState *s = call_state->s;

    qemu_co_mutex_lock(&s->lock);
    QLIST_INSERT_HEAD(&s->calls, call_state, list);
    qemu_co_mutex_unlock(&s->lock);

    do {
        ret = block_copy_dirty_clusters(call_state);

        if (ret == 0 && !qatomic_read(&call_state->cancelled)) {
            WITH_QEMU_LOCK_GUARD(&s->lock) {
                /*
                 * Check that there is no task we still need to
                 * wait to complete
                 */
                ret = reqlist_wait_one(&s->reqs, call_state->offset,
                                       call_state->bytes, &s->lock);
                if (ret == 0) {
                    /*
                     * No pending tasks, but check again the bitmap in this
                     * same critical section, since a task might have failed
                     * between this and the critical section in
                     * block_copy_dirty_clusters().
                     *
                     * reqlist_wait_one return value 0 also means that it
                     * didn't release the lock. So, we are still in the same
                     * critical section, not interrupted by any concurrent
                     * access to state.
                     */
                    ret = bdrv_dirty_bitmap_next_dirty(s->copy_bitmap,
                                                       call_state->offset,
                                                       call_state->bytes) >= 0;
                }
            }
        }

        /*
         * We retry in two cases:
         * 1. Some progress done
         *    Something was copied, which means that there were yield points
         *    and some new dirty bits may have appeared (due to failed parallel
         *    block-copy requests).
         * 2. We have waited for some intersecting block-copy request
         *    It may have failed and produced new dirty bits.
         */
    } while (ret > 0 && !qatomic_read(&call_state->cancelled));

    qatomic_store_release(&call_state->finished, true);

    if (call_state->cb) {
        call_state->cb(call_state->cb_opaque);
    }

    qemu_co_mutex_lock(&s->lock);
    QLIST_REMOVE(call_state, list);
    qemu_co_mutex_unlock(&s->lock);

    return ret;
}

static void coroutine_fn block_copy_async_co_entry(void *opaque)
{
    GRAPH_RDLOCK_GUARD();
    block_copy_common(opaque);
}

int coroutine_fn block_copy(BlockCopyState *s, int64_t start, int64_t bytes,
                            bool ignore_ratelimit, uint64_t timeout_ns,
                            BlockCopyAsyncCallbackFunc cb,
                            void *cb_opaque)
{
    int ret;
    BlockCopyCallState *call_state = g_new(BlockCopyCallState, 1);

    *call_state = (BlockCopyCallState) {
        .s = s,
        .offset = start,
        .bytes = bytes,
        .ignore_ratelimit = ignore_ratelimit,
        .max_workers = BLOCK_COPY_MAX_WORKERS,
        .cb = cb,
        .cb_opaque = cb_opaque,
    };

    ret = qemu_co_timeout(block_copy_async_co_entry, call_state, timeout_ns,
                          g_free);
    if (ret < 0) {
        assert(ret == -ETIMEDOUT);
        block_copy_call_cancel(call_state);
        /* call_state will be freed by running coroutine. */
        return ret;
    }

    ret = call_state->ret;
    g_free(call_state);

    return ret;
}

BlockCopyCallState *block_copy_async(BlockCopyState *s,
                                     int64_t offset, int64_t bytes,
                                     int max_workers, int64_t max_chunk,
                                     BlockCopyAsyncCallbackFunc cb,
                                     void *cb_opaque)
{
    BlockCopyCallState *call_state = g_new(BlockCopyCallState, 1);

    *call_state = (BlockCopyCallState) {
        .s = s,
        .offset = offset,
        .bytes = bytes,
        .max_workers = max_workers,
        .max_chunk = max_chunk,
        .cb = cb,
        .cb_opaque = cb_opaque,

        .co = qemu_coroutine_create(block_copy_async_co_entry, call_state),
    };

    qemu_coroutine_enter(call_state->co);

    return call_state;
}

void block_copy_call_free(BlockCopyCallState *call_state)
{
    if (!call_state) {
        return;
    }

    assert(qatomic_read(&call_state->finished));
    g_free(call_state);
}

bool block_copy_call_finished(BlockCopyCallState *call_state)
{
    return qatomic_read(&call_state->finished);
}

bool block_copy_call_succeeded(BlockCopyCallState *call_state)
{
    return qatomic_load_acquire(&call_state->finished) &&
           !qatomic_read(&call_state->cancelled) &&
           call_state->ret == 0;
}

bool block_copy_call_failed(BlockCopyCallState *call_state)
{
    return qatomic_load_acquire(&call_state->finished) &&
           !qatomic_read(&call_state->cancelled) &&
           call_state->ret < 0;
}

bool block_copy_call_cancelled(BlockCopyCallState *call_state)
{
    return qatomic_read(&call_state->cancelled);
}

int block_copy_call_status(BlockCopyCallState *call_state, bool *error_is_read)
{
    assert(qatomic_load_acquire(&call_state->finished));
    if (error_is_read) {
        *error_is_read = call_state->error_is_read;
    }
    return call_state->ret;
}

/*
 * Note that cancelling and finishing are racy.
 * User can cancel a block-copy that is already finished.
 */
void block_copy_call_cancel(BlockCopyCallState *call_state)
{
    qatomic_set(&call_state->cancelled, true);
    block_copy_kick(call_state);
}

BdrvDirtyBitmap *block_copy_dirty_bitmap(BlockCopyState *s)
{
    return s->copy_bitmap;
}

int64_t block_copy_cluster_size(BlockCopyState *s)
{
    return s->cluster_size;
}

void block_copy_set_skip_unallocated(BlockCopyState *s, bool skip)
{
    qatomic_set(&s->skip_unallocated, skip);
}

void block_copy_set_speed(BlockCopyState *s, uint64_t speed)
{
    ratelimit_set_speed(&s->rate_limit, speed, BLOCK_COPY_SLICE_TIME);

    /*
     * Note: it's good to kick all call states from here, but it should be done
     * only from a coroutine, to not crash if s->calls list changed while
     * entering one call. So for now, the only user of this function kicks its
     * only one call_state by hand.
     */
}