The following changes since commit 2ef2f16781af9dee6ba6517755e9073ba5799fa2:

Merge remote-tracking branch 'remotes/dgilbert/tags/pull-migration-20180615a' into staging (2018-06-15 18:13:35 +0100)

are available in the git repository at:

for you to fetch changes up to 4c790afe2503eab12874508acab5b388d7babfd2:

Merge remote-tracking branch 'mreitz/tags/pull-block-2018-06-18' into queue-block (2018-06-18 17:20:42 +0200)

- Active mirror (blockdev-mirror copy-mode=write-blocking)

- bdrv_drain_*() fixes and test cases

- Fix crash with scsi-hd and drive_del

Greg Kurz (1):

block: fix QEMU crash with scsi-hd and drive_del

Kevin Wolf (20):

test-bdrv-drain: bdrv_drain() works with cross-AioContext events

block: Use bdrv_do_drain_begin/end in bdrv_drain_all()

block: Remove 'recursive' parameter from bdrv_drain_invoke()

block: Don't manually poll in bdrv_drain_all()

tests/test-bdrv-drain: bdrv_drain_all() works in coroutines now

block: Avoid unnecessary aio_poll() in AIO_WAIT_WHILE()

block: Really pause block jobs on drain

block: Remove bdrv_drain_recurse()

block: Drain recursively with a single BDRV_POLL_WHILE()

test-bdrv-drain: Test node deletion in subtree recursion

block: Don't poll in parent drain callbacks

test-bdrv-drain: Graph change through parent callback

block: Defer .bdrv_drain_begin callback to polling phase

test-bdrv-drain: Test that bdrv_drain_invoke() doesn't poll

block: Allow AIO_WAIT_WHILE with NULL ctx

block: Move bdrv_drain_all_begin() out of coroutine context

block: ignore_bds_parents parameter for drain functions

block: Allow graph changes in bdrv_drain_all_begin/end sections

test-bdrv-drain: Test graph changes in drain_all section

Merge remote-tracking branch 'mreitz/tags/pull-block-2018-06-18' into queue-block

Max Reitz (15):

test-bdrv-drain: Add test for node deletion

block/mirror: Pull out mirror_perform()

block/mirror: Convert to coroutines

block/mirror: Use CoQueue to wait on in-flight ops

block/mirror: Wait for in-flight op conflicts

block/mirror: Use source as a BdrvChild

block: Generalize should_update_child() rule

hbitmap: Add @advance param to hbitmap_iter_next()

test-hbitmap: Add non-advancing iter_next tests

block/dirty-bitmap: Add bdrv_dirty_iter_next_area

block/mirror: Add MirrorBDSOpaque

job: Add job_progress_increase_remaining()

block/mirror: Add active mirroring

block/mirror: Add copy mode QAPI interface

iotests: Add test for active mirroring

qapi/block-core.json | 29 +-

include/block/aio-wait.h | 25 +-

include/block/block.h | 31 +-

include/block/block_int.h | 18 +-

include/block/blockjob_int.h | 8 +

include/block/dirty-bitmap.h | 2 +

include/qemu/hbitmap.h | 5 +-

include/qemu/job.h | 15 +

block.c | 96 +++++-

block/backup.c | 2 +-

block/block-backend.c | 5 +

block/dirty-bitmap.c | 57 +++-

block/io.c | 332 ++++++++++++--------

block/mirror.c | 613 +++++++++++++++++++++++++++++--------

block/vvfat.c | 1 +

blockdev.c | 9 +-

blockjob.c | 23 ++

job.c | 5 +

tests/test-bdrv-drain.c | 705 +++++++++++++++++++++++++++++++++++++++++--

tests/test-hbitmap.c | 38 ++-

util/hbitmap.c | 10 +-

tests/qemu-iotests/151 | 120 ++++++++

tests/qemu-iotests/151.out | 5 +

tests/qemu-iotests/group | 1 +

24 files changed, 1836 insertions(+), 319 deletions(-)

create mode 100755 tests/qemu-iotests/151

create mode 100644 tests/qemu-iotests/151.out

Anything can happen inside BDRV_POLL_WHILE(), including graph

changes that may interfere with its callers (e.g. child list iteration

in recursive callers of bdrv_do_drained_begin).

Switch to a single BDRV_POLL_WHILE() call for the whole subtree at the

end of bdrv_do_drained_begin() to avoid such effects. The recursion

happens now inside the loop condition. As the graph can only change

between bdrv_drain_poll() calls, but not inside of it, doing the

recursion here is safe.

include/block/block.h | 9 +++++---

block.c | 2 +-

block/io.c | 63 ++++++++++++++++++++++++++++++++++++---------------

3 files changed, 52 insertions(+), 22 deletions(-)

diff --git a/include/block/block.h b/include/block/block.h

--- a/include/block/block.h

+++ b/include/block/block.h

@@ -XXX,XX +XXX,XX @@ void bdrv_parent_drained_end(BlockDriverState *bs, BdrvChild *ignore);

/**

* bdrv_drain_poll:

*

- * Poll for pending requests in @bs and its parents (except for

- * @ignore_parent). This is part of bdrv_drained_begin.

+ * Poll for pending requests in @bs, its parents (except for @ignore_parent),

+ * and if @recursive is true its children as well.

+ * This is part of bdrv_drained_begin.

*/

-bool bdrv_drain_poll(BlockDriverState *bs, BdrvChild *ignore_parent);

+bool bdrv_drain_poll(BlockDriverState *bs, bool recursive,

+ BdrvChild *ignore_parent);

/**

* bdrv_drained_begin:

diff --git a/block.c b/block.c

index XXXXXXX..XXXXXXX 100644

--- a/block.c

+++ b/block.c

@@ -XXX,XX +XXX,XX @@ static void bdrv_child_cb_drained_begin(BdrvChild *child)

static bool bdrv_child_cb_drained_poll(BdrvChild *child)

BlockDriverState *bs = child->opaque;

- return bdrv_drain_poll(bs, NULL);

+ return bdrv_drain_poll(bs, false, NULL);

}

static void bdrv_child_cb_drained_end(BdrvChild *child)

diff --git a/block/io.c b/block/io.c

index XXXXXXX..XXXXXXX 100644

--- a/block/io.c

+++ b/block/io.c

@@ -XXX,XX +XXX,XX @@ typedef struct {

bool done;

bool begin;

bool recursive;

+ bool poll;

BdrvChild *parent;

} BdrvCoDrainData;

@@ -XXX,XX +XXX,XX @@ static void bdrv_drain_invoke(BlockDriverState *bs, bool begin)

}

/* Returns true if BDRV_POLL_WHILE() should go into a blocking aio_poll() */

-bool bdrv_drain_poll(BlockDriverState *bs, BdrvChild *ignore_parent)

+bool bdrv_drain_poll(BlockDriverState *bs, bool recursive,

+ BdrvChild *ignore_parent)

{

+ BdrvChild *child, *next;

if (bdrv_parent_drained_poll(bs, ignore_parent)) {

return true;

}

- return atomic_read(&bs->in_flight);

+ if (atomic_read(&bs->in_flight)) {

+ return true;

+ }

+

+ if (recursive) {

+ QLIST_FOREACH_SAFE(child, &bs->children, next, next) {

+ if (bdrv_drain_poll(child->bs, recursive, child)) {

+ return true;

+ }

+ return false;

-static bool bdrv_drain_poll_top_level(BlockDriverState *bs,

+static bool bdrv_drain_poll_top_level(BlockDriverState *bs, bool recursive,

BdrvChild *ignore_parent)

/* Execute pending BHs first and check everything else only after the BHs

* have executed. */

while (aio_poll(bs->aio_context, false));

- return bdrv_drain_poll(bs, ignore_parent);

+ return bdrv_drain_poll(bs, recursive, ignore_parent);

static void bdrv_do_drained_begin(BlockDriverState *bs, bool recursive,

- BdrvChild *parent);

+ BdrvChild *parent, bool poll);

static void bdrv_do_drained_end(BlockDriverState *bs, bool recursive,

BdrvChild *parent);

@@ -XXX,XX +XXX,XX @@ static void bdrv_co_drain_bh_cb(void *opaque)

bdrv_dec_in_flight(bs);

if (data->begin) {

- bdrv_do_drained_begin(bs, data->recursive, data->parent);

+ bdrv_do_drained_begin(bs, data->recursive, data->parent, data->poll);

} else {

bdrv_do_drained_end(bs, data->recursive, data->parent);

}

@@ -XXX,XX +XXX,XX @@ static void bdrv_co_drain_bh_cb(void *opaque)

static void coroutine_fn bdrv_co_yield_to_drain(BlockDriverState *bs,

bool begin, bool recursive,

- BdrvChild *parent)

+ BdrvChild *parent, bool poll)

{

BdrvCoDrainData data;

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn bdrv_co_yield_to_drain(BlockDriverState *bs,

.begin = begin,

.recursive = recursive,

.parent = parent,

+ .poll = poll,

};

bdrv_inc_in_flight(bs);

aio_bh_schedule_oneshot(bdrv_get_aio_context(bs),

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn bdrv_co_yield_to_drain(BlockDriverState *bs,

}

void bdrv_do_drained_begin(BlockDriverState *bs, bool recursive,

- BdrvChild *parent)

+ BdrvChild *parent, bool poll)

{

BdrvChild *child, *next;

if (qemu_in_coroutine()) {

- bdrv_co_yield_to_drain(bs, true, recursive, parent);

+ bdrv_co_yield_to_drain(bs, true, recursive, parent, poll);

return;

}

@@ -XXX,XX +XXX,XX @@ void bdrv_do_drained_begin(BlockDriverState *bs, bool recursive,

bdrv_parent_drained_begin(bs, parent);

bdrv_drain_invoke(bs, true);

- /* Wait for drained requests to finish */

- BDRV_POLL_WHILE(bs, bdrv_drain_poll_top_level(bs, parent));

-

if (recursive) {

bs->recursive_quiesce_counter++;

QLIST_FOREACH_SAFE(child, &bs->children, next, next) {

- bdrv_do_drained_begin(child->bs, true, child);

+ bdrv_do_drained_begin(child->bs, true, child, false);

}

}

+

+ /*

+ * Wait for drained requests to finish.

+ *

+ * Calling BDRV_POLL_WHILE() only once for the top-level node is okay: The

+ * call is needed so things in this AioContext can make progress even

+ * though we don't return to the main AioContext loop - this automatically

+ * includes other nodes in the same AioContext and therefore all child

+ * nodes.

+ */

+ if (poll) {

+ BDRV_POLL_WHILE(bs, bdrv_drain_poll_top_level(bs, recursive, parent));

+ }

}

void bdrv_drained_begin(BlockDriverState *bs)

{

- bdrv_do_drained_begin(bs, false, NULL);

+ bdrv_do_drained_begin(bs, false, NULL, true);

}

void bdrv_subtree_drained_begin(BlockDriverState *bs)

{

- bdrv_do_drained_begin(bs, true, NULL);

+ bdrv_do_drained_begin(bs, true, NULL, true);

}

void bdrv_do_drained_end(BlockDriverState *bs, bool recursive,

@@ -XXX,XX +XXX,XX @@ void bdrv_do_drained_end(BlockDriverState *bs, bool recursive,

int old_quiesce_counter;

if (qemu_in_coroutine()) {

- bdrv_co_yield_to_drain(bs, false, recursive, parent);

+ bdrv_co_yield_to_drain(bs, false, recursive, parent, false);

return;

}

assert(bs->quiesce_counter > 0);

@@ -XXX,XX +XXX,XX @@ void bdrv_apply_subtree_drain(BdrvChild *child, BlockDriverState *new_parent)

int i;

for (i = 0; i < new_parent->recursive_quiesce_counter; i++) {

- bdrv_do_drained_begin(child->bs, true, child);

+ bdrv_do_drained_begin(child->bs, true, child, true);

@@ -XXX,XX +XXX,XX @@ void bdrv_drain_all_begin(void)

AioContext *aio_context = bdrv_get_aio_context(bs);

aio_context_acquire(aio_context);

- bdrv_do_drained_begin(bs, true, NULL);

+ bdrv_do_drained_begin(bs, true, NULL, true);

aio_context_release(aio_context);

2.13.6

From: Max Reitz <mreitz@redhat.com>

This patch implements active synchronous mirroring. In active mode, the

passive mechanism will still be in place and is used to copy all

initially dirty clusters off the source disk; but every write request

will write data both to the source and the target disk, so the source

cannot be dirtied faster than data is mirrored to the target. Also,

once the block job has converged (BLOCK_JOB_READY sent), source and

target are guaranteed to stay in sync (unless an error occurs).

Active mode is completely optional and currently disabled at runtime. A

later patch will add a way for users to enable it.

Signed-off-by: Max Reitz <mreitz@redhat.com>

Reviewed-by: Fam Zheng <famz@redhat.com>

Message-id: 20180613181823.13618-13-mreitz@redhat.com

Signed-off-by: Max Reitz <mreitz@redhat.com>

---

qapi/block-core.json | 18 ++++

block/mirror.c | 252 ++++++++++++++++++++++++++++++++++++++++++++++++++-

2 files changed, 265 insertions(+), 5 deletions(-)

diff --git a/qapi/block-core.json b/qapi/block-core.json

--- a/qapi/block-core.json

+++ b/qapi/block-core.json

'data': ['top', 'full', 'none', 'incremental'] }

##

+# @MirrorCopyMode:

+#

+# An enumeration whose values tell the mirror block job when to

+# trigger writes to the target.

+#

+# @background: copy data in background only.

+#

+# @write-blocking: when data is written to the source, write it

+# (synchronously) to the target as well. In

+# addition, data is copied in background just like in

+# @background mode.

+#

+# Since: 3.0

+##

+{ 'enum': 'MirrorCopyMode',

+ 'data': ['background', 'write-blocking'] }

+##

# @BlockJobInfo:

#

# Information about a long-running block device operation.

diff --git a/block/mirror.c b/block/mirror.c

index XXXXXXX..XXXXXXX 100644

--- a/block/mirror.c

+++ b/block/mirror.c

@@ -XXX,XX +XXX,XX @@ typedef struct MirrorBlockJob {

Error *replace_blocker;

bool is_none_mode;

BlockMirrorBackingMode backing_mode;

+ MirrorCopyMode copy_mode;

BlockdevOnError on_source_error, on_target_error;

bool synced;

+ /* Set when the target is synced (dirty bitmap is clean, nothing

+ * in flight) and the job is running in active mode */

+ bool actively_synced;

bool should_complete;

int64_t granularity;

size_t buf_size;

@@ -XXX,XX +XXX,XX @@ typedef struct MirrorBlockJob {

int target_cluster_size;

int max_iov;

bool initial_zeroing_ongoing;

+ int in_active_write_counter;

} MirrorBlockJob;

typedef struct MirrorBDSOpaque {

@@ -XXX,XX +XXX,XX @@ struct MirrorOp {

int64_t *bytes_handled;

bool is_pseudo_op;

+ bool is_active_write;

CoQueue waiting_requests;

QTAILQ_ENTRY(MirrorOp) next;

@@ -XXX,XX +XXX,XX @@ static BlockErrorAction mirror_error_action(MirrorBlockJob *s, bool read,

int error)

{

s->synced = false;

+ s->actively_synced = false;

if (read) {

return block_job_error_action(&s->common, s->on_source_error,

true, error);

@@ -XXX,XX +XXX,XX @@ static int mirror_cow_align(MirrorBlockJob *s, int64_t *offset,

return ret;

}

-static inline void mirror_wait_for_free_in_flight_slot(MirrorBlockJob *s)

+static inline void mirror_wait_for_any_operation(MirrorBlockJob *s, bool active)

{

MirrorOp *op;

@@ -XXX,XX +XXX,XX @@ static inline void mirror_wait_for_free_in_flight_slot(MirrorBlockJob *s)

* caller of this function. Since there is only one pseudo op

* at any given time, we will always find some real operation

* to wait on. */

- if (!op->is_pseudo_op) {

+ if (!op->is_pseudo_op && op->is_active_write == active) {

qemu_co_queue_wait(&op->waiting_requests, NULL);

return;

}

@@ -XXX,XX +XXX,XX @@ static inline void mirror_wait_for_free_in_flight_slot(MirrorBlockJob *s)

abort();

}

+static inline void mirror_wait_for_free_in_flight_slot(MirrorBlockJob *s)

+ /* Only non-active operations use up in-flight slots */

+ mirror_wait_for_any_operation(s, false);

+}

/* Perform a mirror copy operation.

*

* *op->bytes_handled is set to the number of bytes copied after and

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn mirror_run(void *opaque)

/* Transition to the READY state and wait for complete. */

job_transition_to_ready(&s->common.job);

s->synced = true;

+ s->actively_synced = true;

while (!job_is_cancelled(&s->common.job) && !s->should_complete) {

job_yield(&s->common.job);

}

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn mirror_run(void *opaque)

int64_t cnt, delta;

bool should_complete;

+ /* Do not start passive operations while there are active

+ * writes in progress */

+ while (s->in_active_write_counter) {

+ mirror_wait_for_any_operation(s, true);

+ }

if (s->ret < 0) {

ret = s->ret;

goto immediate_exit;

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn mirror_run(void *opaque)

*/

job_transition_to_ready(&s->common.job);

s->synced = true;

+ if (s->copy_mode != MIRROR_COPY_MODE_BACKGROUND) {

+ s->actively_synced = true;

+ }

}

should_complete = s->should_complete ||

@@ -XXX,XX +XXX,XX @@ static const BlockJobDriver commit_active_job_driver = {

.drain = mirror_drain,

};

+static void do_sync_target_write(MirrorBlockJob *job, MirrorMethod method,

+ uint64_t offset, uint64_t bytes,

+ QEMUIOVector *qiov, int flags)

+{

+ BdrvDirtyBitmapIter *iter;

+ QEMUIOVector target_qiov;

+ uint64_t dirty_offset;

+ int dirty_bytes;

+

+ if (qiov) {

+ qemu_iovec_init(&target_qiov, qiov->niov);

+ }

+

+ iter = bdrv_dirty_iter_new(job->dirty_bitmap);

+ bdrv_set_dirty_iter(iter, offset);

+

+ while (true) {

+ bool valid_area;

+ int ret;

+

+ bdrv_dirty_bitmap_lock(job->dirty_bitmap);

+ valid_area = bdrv_dirty_iter_next_area(iter, offset + bytes,

+ &dirty_offset, &dirty_bytes);

+ if (!valid_area) {

+ bdrv_dirty_bitmap_unlock(job->dirty_bitmap);

+ bdrv_reset_dirty_bitmap_locked(job->dirty_bitmap,

+ dirty_offset, dirty_bytes);

+ bdrv_dirty_bitmap_unlock(job->dirty_bitmap);

+

+ job_progress_increase_remaining(&job->common.job, dirty_bytes);

+

+ assert(dirty_offset - offset <= SIZE_MAX);

+ if (qiov) {

+ qemu_iovec_reset(&target_qiov);

+ qemu_iovec_concat(&target_qiov, qiov,

+ dirty_offset - offset, dirty_bytes);

+ }

+

+ switch (method) {

+ case MIRROR_METHOD_COPY:

+ ret = blk_co_pwritev(job->target, dirty_offset, dirty_bytes,

+ qiov ? &target_qiov : NULL, flags);

+ break;

+

+ case MIRROR_METHOD_ZERO:

+ assert(!qiov);

+ ret = blk_co_pwrite_zeroes(job->target, dirty_offset, dirty_bytes,

+ flags);

+ break;

+

+ case MIRROR_METHOD_DISCARD:

+ assert(!qiov);

+ ret = blk_co_pdiscard(job->target, dirty_offset, dirty_bytes);

+ break;

+

+ default:

+ abort();

+ }

+

+ if (ret >= 0) {

+ job_progress_update(&job->common.job, dirty_bytes);

+ } else {

+ BlockErrorAction action;

+

+ bdrv_set_dirty_bitmap(job->dirty_bitmap, dirty_offset, dirty_bytes);

+ job->actively_synced = false;

+

+ action = mirror_error_action(job, false, -ret);

+ if (action == BLOCK_ERROR_ACTION_REPORT) {

+ if (!job->ret) {

+ job->ret = ret;

+ }

+ break;

+ }

+ }

+ bdrv_dirty_iter_free(iter);

+ if (qiov) {

+ qemu_iovec_destroy(&target_qiov);

+ }

+static MirrorOp *coroutine_fn active_write_prepare(MirrorBlockJob *s,

+ uint64_t offset,

+ uint64_t bytes)

+ MirrorOp *op;

+ uint64_t start_chunk = offset / s->granularity;

+ uint64_t end_chunk = DIV_ROUND_UP(offset + bytes, s->granularity);

+ op = g_new(MirrorOp, 1);

+ *op = (MirrorOp){

+ .s = s,

+ .offset = offset,

+ .bytes = bytes,

+ .is_active_write = true,

+ };

+ qemu_co_queue_init(&op->waiting_requests);

+ QTAILQ_INSERT_TAIL(&s->ops_in_flight, op, next);

+

+ s->in_active_write_counter++;

+

+ mirror_wait_on_conflicts(op, s, offset, bytes);

+

+ bitmap_set(s->in_flight_bitmap, start_chunk, end_chunk - start_chunk);

+

+ return op;

+

+static void coroutine_fn active_write_settle(MirrorOp *op)

+{

+ uint64_t start_chunk = op->offset / op->s->granularity;

+ uint64_t end_chunk = DIV_ROUND_UP(op->offset + op->bytes,

+ op->s->granularity);

+

+ if (!--op->s->in_active_write_counter && op->s->actively_synced) {

+ BdrvChild *source = op->s->mirror_top_bs->backing;

+

+ if (QLIST_FIRST(&source->bs->parents) == source &&

+ QLIST_NEXT(source, next_parent) == NULL)

+ {

+ /* Assert that we are back in sync once all active write

+ * operations are settled.

+ * Note that we can only assert this if the mirror node

+ * is the source node's only parent. */

+ assert(!bdrv_get_dirty_count(op->s->dirty_bitmap));

+ }

+ }

+ bitmap_clear(op->s->in_flight_bitmap, start_chunk, end_chunk - start_chunk);

+ QTAILQ_REMOVE(&op->s->ops_in_flight, op, next);

+ qemu_co_queue_restart_all(&op->waiting_requests);

+ g_free(op);

+}

+

static int coroutine_fn bdrv_mirror_top_preadv(BlockDriverState *bs,

uint64_t offset, uint64_t bytes, QEMUIOVector *qiov, int flags)

{

return bdrv_co_preadv(bs->backing, offset, bytes, qiov, flags);

}

+static int coroutine_fn bdrv_mirror_top_do_write(BlockDriverState *bs,

+ MirrorMethod method, uint64_t offset, uint64_t bytes, QEMUIOVector *qiov,

+ int flags)

+{

+ MirrorOp *op = NULL;

+ MirrorBDSOpaque *s = bs->opaque;

+ int ret = 0;

+ bool copy_to_target;

+

+ copy_to_target = s->job->ret >= 0 &&

+ s->job->copy_mode == MIRROR_COPY_MODE_WRITE_BLOCKING;

+

+ if (copy_to_target) {

+ op = active_write_prepare(s->job, offset, bytes);

+ }

+

+ switch (method) {

+ case MIRROR_METHOD_COPY:

+ ret = bdrv_co_pwritev(bs->backing, offset, bytes, qiov, flags);

+ break;

+

+ case MIRROR_METHOD_ZERO:

+ ret = bdrv_co_pwrite_zeroes(bs->backing, offset, bytes, flags);

+ break;

+

+ case MIRROR_METHOD_DISCARD:

+ ret = bdrv_co_pdiscard(bs->backing->bs, offset, bytes);

+ break;

+

+ default:

+ abort();

+ }

+

+ if (ret < 0) {

+ goto out;

+ }

+

+ if (copy_to_target) {

+ do_sync_target_write(s->job, method, offset, bytes, qiov, flags);

+ }

+

+out:

+ if (copy_to_target) {

+ active_write_settle(op);

+ }

+ return ret;

+}

+

static int coroutine_fn bdrv_mirror_top_pwritev(BlockDriverState *bs,

uint64_t offset, uint64_t bytes, QEMUIOVector *qiov, int flags)

{

- return bdrv_co_pwritev(bs->backing, offset, bytes, qiov, flags);

+ MirrorBDSOpaque *s = bs->opaque;

+ QEMUIOVector bounce_qiov;

+ void *bounce_buf;

+ int ret = 0;

+ bool copy_to_target;

+

+ copy_to_target = s->job->ret >= 0 &&

+ s->job->copy_mode == MIRROR_COPY_MODE_WRITE_BLOCKING;

+

+ if (copy_to_target) {

+ /* The guest might concurrently modify the data to write; but

+ * the data on source and destination must match, so we have

+ * to use a bounce buffer if we are going to write to the

+ * target now. */

+ bounce_buf = qemu_blockalign(bs, bytes);

+ iov_to_buf_full(qiov->iov, qiov->niov, 0, bounce_buf, bytes);

+

+ qemu_iovec_init(&bounce_qiov, 1);

+ qemu_iovec_add(&bounce_qiov, bounce_buf, bytes);

+ qiov = &bounce_qiov;

+ }

+

+ ret = bdrv_mirror_top_do_write(bs, MIRROR_METHOD_COPY, offset, bytes, qiov,

+ flags);

+

+ if (copy_to_target) {

+ qemu_iovec_destroy(&bounce_qiov);

+ qemu_vfree(bounce_buf);

+ }

+

+ return ret;

}

static int coroutine_fn bdrv_mirror_top_flush(BlockDriverState *bs)

@@ -XXX,XX +XXX,XX @@ static int coroutine_fn bdrv_mirror_top_flush(BlockDriverState *bs)

static int coroutine_fn bdrv_mirror_top_pwrite_zeroes(BlockDriverState *bs,

int64_t offset, int bytes, BdrvRequestFlags flags)

{

- return bdrv_co_pwrite_zeroes(bs->backing, offset, bytes, flags);

+ return bdrv_mirror_top_do_write(bs, MIRROR_METHOD_ZERO, offset, bytes, NULL,

+ flags);

}

static int coroutine_fn bdrv_mirror_top_pdiscard(BlockDriverState *bs,

int64_t offset, int bytes)

{

- return bdrv_co_pdiscard(bs->backing->bs, offset, bytes);

+ return bdrv_mirror_top_do_write(bs, MIRROR_METHOD_DISCARD, offset, bytes,

+ NULL, 0);

}

static void bdrv_mirror_top_refresh_filename(BlockDriverState *bs, QDict *opts)

@@ -XXX,XX +XXX,XX @@ static void mirror_start_job(const char *job_id, BlockDriverState *bs,

s->on_target_error = on_target_error;

s->is_none_mode = is_none_mode;

s->backing_mode = backing_mode;

+ s->copy_mode = MIRROR_COPY_MODE_BACKGROUND;

s->base = base;

s->granularity = granularity;

s->buf_size = ROUND_UP(buf_size, granularity);

2.13.6

We cannot allow aio_poll() in bdrv_drain_invoke(begin=true) until we're

done with propagating the drain through the graph and are doing the

single final BDRV_POLL_WHILE().

Just schedule the coroutine with the callback and increase bs->in_flight

to make sure that the polling phase will wait for it.

block/io.c | 28 +++++++++++++++++++++++-----

1 file changed, 23 insertions(+), 5 deletions(-)

diff --git a/block/io.c b/block/io.c

--- a/block/io.c

+++ b/block/io.c

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn bdrv_drain_invoke_entry(void *opaque)

/* Set data->done before reading bs->wakeup. */

atomic_mb_set(&data->done, true);

- bdrv_wakeup(bs);

+ bdrv_dec_in_flight(bs);

+

+ if (data->begin) {

+ g_free(data);

+ }

}

/* Recursively call BlockDriver.bdrv_co_drain_begin/end callbacks */

static void bdrv_drain_invoke(BlockDriverState *bs, bool begin)

{

- BdrvCoDrainData data = { .bs = bs, .done = false, .begin = begin};

+ BdrvCoDrainData *data;

if (!bs->drv || (begin && !bs->drv->bdrv_co_drain_begin) ||

(!begin && !bs->drv->bdrv_co_drain_end)) {

- data.co = qemu_coroutine_create(bdrv_drain_invoke_entry, &data);

- bdrv_coroutine_enter(bs, data.co);

- BDRV_POLL_WHILE(bs, !data.done);

+ data = g_new(BdrvCoDrainData, 1);

+ *data = (BdrvCoDrainData) {

+ .bs = bs,

+ .done = false,

+ .begin = begin

+ };

+

+ /* Make sure the driver callback completes during the polling phase for

+ * drain_begin. */

+ bdrv_inc_in_flight(bs);

+ data->co = qemu_coroutine_create(bdrv_drain_invoke_entry, data);

+ aio_co_schedule(bdrv_get_aio_context(bs), data->co);

+

+ if (!begin) {

+ BDRV_POLL_WHILE(bs, !data->done);

+ g_free(data);

+ }

/* Returns true if BDRV_POLL_WHILE() should go into a blocking aio_poll() */

2.13.6

As long as nobody keeps the other I/O thread from working, there is no

reason why bdrv_drain() wouldn't work with cross-AioContext events. The

key is that the root request we're waiting for is in the AioContext

we're polling (which it always is for bdrv_drain()) so that aio_poll()

is woken up in the end.

Add a test case that shows that it works. Remove the comment in

bdrv_drain() that claims otherwise.

block/io.c | 4 --

tests/test-bdrv-drain.c | 187 +++++++++++++++++++++++++++++++++++++++++++++++-

2 files changed, 186 insertions(+), 5 deletions(-)

diff --git a/block/io.c b/block/io.c

--- a/block/io.c

+++ b/block/io.c

@@ -XXX,XX +XXX,XX @@ void bdrv_unapply_subtree_drain(BdrvChild *child, BlockDriverState *old_parent)

*

* Note that unlike bdrv_drain_all(), the caller must hold the BlockDriverState

* AioContext.

- *

- * Only this BlockDriverState's AioContext is run, so in-flight requests must

- * not depend on events in other AioContexts. In that case, use

- * bdrv_drain_all() instead.

*/

void coroutine_fn bdrv_co_drain(BlockDriverState *bs)

{

diff --git a/tests/test-bdrv-drain.c b/tests/test-bdrv-drain.c

index XXXXXXX..XXXXXXX 100644

--- a/tests/test-bdrv-drain.c

+++ b/tests/test-bdrv-drain.c

@@ -XXX,XX +XXX,XX @@

#include "block/blockjob_int.h"

#include "sysemu/block-backend.h"

#include "qapi/error.h"

+#include "iothread.h"

+

+static QemuEvent done_event;

typedef struct BDRVTestState {

int drain_count;

+ AioContext *bh_indirection_ctx;

} BDRVTestState;

static void coroutine_fn bdrv_test_co_drain_begin(BlockDriverState *bs)

@@ -XXX,XX +XXX,XX @@ static void bdrv_test_close(BlockDriverState *bs)

g_assert_cmpint(s->drain_count, >, 0);

}

+static void co_reenter_bh(void *opaque)

+{

+ aio_co_wake(opaque);

+}

+

static int coroutine_fn bdrv_test_co_preadv(BlockDriverState *bs,

uint64_t offset, uint64_t bytes,

QEMUIOVector *qiov, int flags)

{

+ BDRVTestState *s = bs->opaque;

+

/* We want this request to stay until the polling loop in drain waits for

* it to complete. We need to sleep a while as bdrv_drain_invoke() comes

* first and polls its result, too, but it shouldn't accidentally complete

* this request yet. */

qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 100000);

+ if (s->bh_indirection_ctx) {

+ aio_bh_schedule_oneshot(s->bh_indirection_ctx, co_reenter_bh,

+ qemu_coroutine_self());

+ qemu_coroutine_yield();

+ }

+

return 0;

}

@@ -XXX,XX +XXX,XX @@ static void test_graph_change(void)

blk_unref(blk_b);

}

+struct test_iothread_data {

+ BlockDriverState *bs;

+ enum drain_type drain_type;

+ int *aio_ret;

+};

+

+static void test_iothread_drain_entry(void *opaque)

+{

+ struct test_iothread_data *data = opaque;

+

+ aio_context_acquire(bdrv_get_aio_context(data->bs));

+ do_drain_begin(data->drain_type, data->bs);

+ g_assert_cmpint(*data->aio_ret, ==, 0);

+ do_drain_end(data->drain_type, data->bs);

+ aio_context_release(bdrv_get_aio_context(data->bs));

+

+ qemu_event_set(&done_event);

+}

+

+static void test_iothread_aio_cb(void *opaque, int ret)

+{

+ int *aio_ret = opaque;

+ *aio_ret = ret;

+ qemu_event_set(&done_event);

+}

+

+/*

+ * Starts an AIO request on a BDS that runs in the AioContext of iothread 1.

+ * The request involves a BH on iothread 2 before it can complete.

+ *

+ * @drain_thread = 0 means that do_drain_begin/end are called from the main

+ * thread, @drain_thread = 1 means that they are called from iothread 1. Drain

+ * for this BDS cannot be called from iothread 2 because only the main thread

+ * may do cross-AioContext polling.

+ */

+static void test_iothread_common(enum drain_type drain_type, int drain_thread)

+{

+ BlockBackend *blk;

+ BlockDriverState *bs;

+ BDRVTestState *s;

+ BlockAIOCB *acb;

+ int aio_ret;

+ struct test_iothread_data data;

+

+ IOThread *a = iothread_new();

+ IOThread *b = iothread_new();

+ AioContext *ctx_a = iothread_get_aio_context(a);

+ AioContext *ctx_b = iothread_get_aio_context(b);

+

+ QEMUIOVector qiov;

+ struct iovec iov = {

+ .iov_base = NULL,

+ .iov_len = 0,

+ };

+ qemu_iovec_init_external(&qiov, &iov, 1);

+

+ /* bdrv_drain_all() may only be called from the main loop thread */

+ if (drain_type == BDRV_DRAIN_ALL && drain_thread != 0) {

+ goto out;

+ }

+

+ blk = blk_new(BLK_PERM_ALL, BLK_PERM_ALL);

+ bs = bdrv_new_open_driver(&bdrv_test, "test-node", BDRV_O_RDWR,

+ &error_abort);

+ s = bs->opaque;

+ blk_insert_bs(blk, bs, &error_abort);

+

+ blk_set_aio_context(blk, ctx_a);

+ aio_context_acquire(ctx_a);

+

+ s->bh_indirection_ctx = ctx_b;

+

+ aio_ret = -EINPROGRESS;

+ if (drain_thread == 0) {

+ acb = blk_aio_preadv(blk, 0, &qiov, 0, test_iothread_aio_cb, &aio_ret);

+ } else {

+ acb = blk_aio_preadv(blk, 0, &qiov, 0, aio_ret_cb, &aio_ret);

+ }

+ g_assert(acb != NULL);

+ g_assert_cmpint(aio_ret, ==, -EINPROGRESS);

+

+ aio_context_release(ctx_a);

+

+ data = (struct test_iothread_data) {

+ .bs = bs,

+ .drain_type = drain_type,

+ .aio_ret = &aio_ret,

+ };

+

+ switch (drain_thread) {

+ case 0:

+ if (drain_type != BDRV_DRAIN_ALL) {

+ aio_context_acquire(ctx_a);

+ /* The request is running on the IOThread a. Draining its block device

+ * will make sure that it has completed as far as the BDS is concerned,

+ * but the drain in this thread can continue immediately after

+ * bdrv_dec_in_flight() and aio_ret might be assigned only slightly

+ * later. */

+ qemu_event_reset(&done_event);

+ do_drain_begin(drain_type, bs);

+ g_assert_cmpint(bs->in_flight, ==, 0);

+

+ if (drain_type != BDRV_DRAIN_ALL) {

+ aio_context_release(ctx_a);

+ }

+ qemu_event_wait(&done_event);

+ if (drain_type != BDRV_DRAIN_ALL) {

+ aio_context_acquire(ctx_a);

+ }

+

+ g_assert_cmpint(aio_ret, ==, 0);

+ do_drain_end(drain_type, bs);

+

+ if (drain_type != BDRV_DRAIN_ALL) {

+ aio_context_release(ctx_a);

+ }

+ break;

+ case 1:

+ qemu_event_reset(&done_event);

+ aio_bh_schedule_oneshot(ctx_a, test_iothread_drain_entry, &data);

+ qemu_event_wait(&done_event);

+ break;

+ default:

+ g_assert_not_reached();

+ }

+

+ aio_context_acquire(ctx_a);

+ blk_set_aio_context(blk, qemu_get_aio_context());

+ aio_context_release(ctx_a);

+

+ bdrv_unref(bs);

+ blk_unref(blk);

+

+out:

+ iothread_join(a);

+ iothread_join(b);

+}

+

+static void test_iothread_drain_all(void)

+{

+ test_iothread_common(BDRV_DRAIN_ALL, 0);

+ test_iothread_common(BDRV_DRAIN_ALL, 1);

+}

+

+static void test_iothread_drain(void)

+{

+ test_iothread_common(BDRV_DRAIN, 0);

+ test_iothread_common(BDRV_DRAIN, 1);

+}

+

+static void test_iothread_drain_subtree(void)

+{

+ test_iothread_common(BDRV_SUBTREE_DRAIN, 0);

+ test_iothread_common(BDRV_SUBTREE_DRAIN, 1);

+}

+

typedef struct TestBlockJob {

BlockJob common;

@@ -XXX,XX +XXX,XX @@ static void test_blockjob_drain_subtree(void)

int main(int argc, char **argv)

{

+ int ret;

+

bdrv_init();

qemu_init_main_loop(&error_abort);

g_test_init(&argc, &argv, NULL);

+ qemu_event_init(&done_event, false);

g_test_add_func("/bdrv-drain/driver-cb/drain_all", test_drv_cb_drain_all);

g_test_add_func("/bdrv-drain/driver-cb/drain", test_drv_cb_drain);

@@ -XXX,XX +XXX,XX @@ int main(int argc, char **argv)

g_test_add_func("/bdrv-drain/multiparent", test_multiparent);

g_test_add_func("/bdrv-drain/graph-change", test_graph_change);

+ g_test_add_func("/bdrv-drain/iothread/drain_all", test_iothread_drain_all);

+ g_test_add_func("/bdrv-drain/iothread/drain", test_iothread_drain);

+ g_test_add_func("/bdrv-drain/iothread/drain_subtree",

+ test_iothread_drain_subtree);

+

g_test_add_func("/bdrv-drain/blockjob/drain_all", test_blockjob_drain_all);

g_test_add_func("/bdrv-drain/blockjob/drain", test_blockjob_drain);

g_test_add_func("/bdrv-drain/blockjob/drain_subtree",

test_blockjob_drain_subtree);

- return g_test_run();

+ ret = g_test_run();

+ qemu_event_destroy(&done_event);

+ return ret;

}

2.13.6