1 /* 2 * Image mirroring 3 * 4 * Copyright Red Hat, Inc. 2012 5 * 6 * Authors: 7 * Paolo Bonzini <pbonzini@redhat.com> 8 * 9 * This work is licensed under the terms of the GNU LGPL, version 2 or later. 10 * See the COPYING.LIB file in the top-level directory. 11 * 12 */ 13 14 #include "qemu/osdep.h" 15 #include "qemu/cutils.h" 16 #include "qemu/coroutine.h" 17 #include "qemu/range.h" 18 #include "trace.h" 19 #include "block/blockjob_int.h" 20 #include "block/block_int.h" 21 #include "sysemu/block-backend.h" 22 #include "qapi/error.h" 23 #include "qapi/qmp/qerror.h" 24 #include "qemu/ratelimit.h" 25 #include "qemu/bitmap.h" 26 27 #define MAX_IN_FLIGHT 16 28 #define MAX_IO_BYTES (1 << 20) /* 1 Mb */ 29 #define DEFAULT_MIRROR_BUF_SIZE (MAX_IN_FLIGHT * MAX_IO_BYTES) 30 31 /* The mirroring buffer is a list of granularity-sized chunks. 32 * Free chunks are organized in a list. 33 */ 34 typedef struct MirrorBuffer { 35 QSIMPLEQ_ENTRY(MirrorBuffer) next; 36 } MirrorBuffer; 37 38 typedef struct MirrorOp MirrorOp; 39 40 typedef struct MirrorBlockJob { 41 BlockJob common; 42 BlockBackend *target; 43 BlockDriverState *mirror_top_bs; 44 BlockDriverState *base; 45 46 /* The name of the graph node to replace */ 47 char *replaces; 48 /* The BDS to replace */ 49 BlockDriverState *to_replace; 50 /* Used to block operations on the drive-mirror-replace target */ 51 Error *replace_blocker; 52 bool is_none_mode; 53 BlockMirrorBackingMode backing_mode; 54 /* Whether the target image requires explicit zero-initialization */ 55 bool zero_target; 56 MirrorCopyMode copy_mode; 57 BlockdevOnError on_source_error, on_target_error; 58 bool synced; 59 /* Set when the target is synced (dirty bitmap is clean, nothing 60 * in flight) and the job is running in active mode */ 61 bool actively_synced; 62 bool should_complete; 63 int64_t granularity; 64 size_t buf_size; 65 int64_t bdev_length; 66 unsigned long *cow_bitmap; 67 BdrvDirtyBitmap *dirty_bitmap; 68 BdrvDirtyBitmapIter *dbi; 69 uint8_t *buf; 70 QSIMPLEQ_HEAD(, MirrorBuffer) buf_free; 71 int buf_free_count; 72 73 uint64_t last_pause_ns; 74 unsigned long *in_flight_bitmap; 75 int in_flight; 76 int64_t bytes_in_flight; 77 QTAILQ_HEAD(, MirrorOp) ops_in_flight; 78 int ret; 79 bool unmap; 80 int target_cluster_size; 81 int max_iov; 82 bool initial_zeroing_ongoing; 83 int in_active_write_counter; 84 bool prepared; 85 bool in_drain; 86 } MirrorBlockJob; 87 88 typedef struct MirrorBDSOpaque { 89 MirrorBlockJob *job; 90 bool stop; 91 } MirrorBDSOpaque; 92 93 struct MirrorOp { 94 MirrorBlockJob *s; 95 QEMUIOVector qiov; 96 int64_t offset; 97 uint64_t bytes; 98 99 /* The pointee is set by mirror_co_read(), mirror_co_zero(), and 100 * mirror_co_discard() before yielding for the first time */ 101 int64_t *bytes_handled; 102 103 bool is_pseudo_op; 104 bool is_active_write; 105 CoQueue waiting_requests; 106 Coroutine *co; 107 108 QTAILQ_ENTRY(MirrorOp) next; 109 }; 110 111 typedef enum MirrorMethod { 112 MIRROR_METHOD_COPY, 113 MIRROR_METHOD_ZERO, 114 MIRROR_METHOD_DISCARD, 115 } MirrorMethod; 116 117 static BlockErrorAction mirror_error_action(MirrorBlockJob *s, bool read, 118 int error) 119 { 120 s->synced = false; 121 s->actively_synced = false; 122 if (read) { 123 return block_job_error_action(&s->common, s->on_source_error, 124 true, error); 125 } else { 126 return block_job_error_action(&s->common, s->on_target_error, 127 false, error); 128 } 129 } 130 131 static void coroutine_fn mirror_wait_on_conflicts(MirrorOp *self, 132 MirrorBlockJob *s, 133 uint64_t offset, 134 uint64_t bytes) 135 { 136 uint64_t self_start_chunk = offset / s->granularity; 137 uint64_t self_end_chunk = DIV_ROUND_UP(offset + bytes, s->granularity); 138 uint64_t self_nb_chunks = self_end_chunk - self_start_chunk; 139 140 while (find_next_bit(s->in_flight_bitmap, self_end_chunk, 141 self_start_chunk) < self_end_chunk && 142 s->ret >= 0) 143 { 144 MirrorOp *op; 145 146 QTAILQ_FOREACH(op, &s->ops_in_flight, next) { 147 uint64_t op_start_chunk = op->offset / s->granularity; 148 uint64_t op_nb_chunks = DIV_ROUND_UP(op->offset + op->bytes, 149 s->granularity) - 150 op_start_chunk; 151 152 if (op == self) { 153 continue; 154 } 155 156 if (ranges_overlap(self_start_chunk, self_nb_chunks, 157 op_start_chunk, op_nb_chunks)) 158 { 159 qemu_co_queue_wait(&op->waiting_requests, NULL); 160 break; 161 } 162 } 163 } 164 } 165 166 static void coroutine_fn mirror_iteration_done(MirrorOp *op, int ret) 167 { 168 MirrorBlockJob *s = op->s; 169 struct iovec *iov; 170 int64_t chunk_num; 171 int i, nb_chunks; 172 173 trace_mirror_iteration_done(s, op->offset, op->bytes, ret); 174 175 s->in_flight--; 176 s->bytes_in_flight -= op->bytes; 177 iov = op->qiov.iov; 178 for (i = 0; i < op->qiov.niov; i++) { 179 MirrorBuffer *buf = (MirrorBuffer *) iov[i].iov_base; 180 QSIMPLEQ_INSERT_TAIL(&s->buf_free, buf, next); 181 s->buf_free_count++; 182 } 183 184 chunk_num = op->offset / s->granularity; 185 nb_chunks = DIV_ROUND_UP(op->bytes, s->granularity); 186 187 bitmap_clear(s->in_flight_bitmap, chunk_num, nb_chunks); 188 QTAILQ_REMOVE(&s->ops_in_flight, op, next); 189 if (ret >= 0) { 190 if (s->cow_bitmap) { 191 bitmap_set(s->cow_bitmap, chunk_num, nb_chunks); 192 } 193 if (!s->initial_zeroing_ongoing) { 194 job_progress_update(&s->common.job, op->bytes); 195 } 196 } 197 qemu_iovec_destroy(&op->qiov); 198 199 qemu_co_queue_restart_all(&op->waiting_requests); 200 g_free(op); 201 } 202 203 static void coroutine_fn mirror_write_complete(MirrorOp *op, int ret) 204 { 205 MirrorBlockJob *s = op->s; 206 207 if (ret < 0) { 208 BlockErrorAction action; 209 210 bdrv_set_dirty_bitmap(s->dirty_bitmap, op->offset, op->bytes); 211 action = mirror_error_action(s, false, -ret); 212 if (action == BLOCK_ERROR_ACTION_REPORT && s->ret >= 0) { 213 s->ret = ret; 214 } 215 } 216 217 mirror_iteration_done(op, ret); 218 } 219 220 static void coroutine_fn mirror_read_complete(MirrorOp *op, int ret) 221 { 222 MirrorBlockJob *s = op->s; 223 224 if (ret < 0) { 225 BlockErrorAction action; 226 227 bdrv_set_dirty_bitmap(s->dirty_bitmap, op->offset, op->bytes); 228 action = mirror_error_action(s, true, -ret); 229 if (action == BLOCK_ERROR_ACTION_REPORT && s->ret >= 0) { 230 s->ret = ret; 231 } 232 233 mirror_iteration_done(op, ret); 234 return; 235 } 236 237 ret = blk_co_pwritev(s->target, op->offset, op->qiov.size, &op->qiov, 0); 238 mirror_write_complete(op, ret); 239 } 240 241 /* Clip bytes relative to offset to not exceed end-of-file */ 242 static inline int64_t mirror_clip_bytes(MirrorBlockJob *s, 243 int64_t offset, 244 int64_t bytes) 245 { 246 return MIN(bytes, s->bdev_length - offset); 247 } 248 249 /* Round offset and/or bytes to target cluster if COW is needed, and 250 * return the offset of the adjusted tail against original. */ 251 static int mirror_cow_align(MirrorBlockJob *s, int64_t *offset, 252 uint64_t *bytes) 253 { 254 bool need_cow; 255 int ret = 0; 256 int64_t align_offset = *offset; 257 int64_t align_bytes = *bytes; 258 int max_bytes = s->granularity * s->max_iov; 259 260 need_cow = !test_bit(*offset / s->granularity, s->cow_bitmap); 261 need_cow |= !test_bit((*offset + *bytes - 1) / s->granularity, 262 s->cow_bitmap); 263 if (need_cow) { 264 bdrv_round_to_clusters(blk_bs(s->target), *offset, *bytes, 265 &align_offset, &align_bytes); 266 } 267 268 if (align_bytes > max_bytes) { 269 align_bytes = max_bytes; 270 if (need_cow) { 271 align_bytes = QEMU_ALIGN_DOWN(align_bytes, s->target_cluster_size); 272 } 273 } 274 /* Clipping may result in align_bytes unaligned to chunk boundary, but 275 * that doesn't matter because it's already the end of source image. */ 276 align_bytes = mirror_clip_bytes(s, align_offset, align_bytes); 277 278 ret = align_offset + align_bytes - (*offset + *bytes); 279 *offset = align_offset; 280 *bytes = align_bytes; 281 assert(ret >= 0); 282 return ret; 283 } 284 285 static inline void coroutine_fn 286 mirror_wait_for_any_operation(MirrorBlockJob *s, MirrorOp *self, bool active) 287 { 288 MirrorOp *op; 289 290 QTAILQ_FOREACH(op, &s->ops_in_flight, next) { 291 if (self == op) { 292 continue; 293 } 294 /* Do not wait on pseudo ops, because it may in turn wait on 295 * some other operation to start, which may in fact be the 296 * caller of this function. Since there is only one pseudo op 297 * at any given time, we will always find some real operation 298 * to wait on. */ 299 if (!op->is_pseudo_op && op->is_active_write == active) { 300 qemu_co_queue_wait(&op->waiting_requests, NULL); 301 return; 302 } 303 } 304 abort(); 305 } 306 307 static inline void coroutine_fn 308 mirror_wait_for_free_in_flight_slot(MirrorBlockJob *s, MirrorOp *self) 309 { 310 /* Only non-active operations use up in-flight slots */ 311 mirror_wait_for_any_operation(s, self, false); 312 } 313 314 /* Perform a mirror copy operation. 315 * 316 * *op->bytes_handled is set to the number of bytes copied after and 317 * including offset, excluding any bytes copied prior to offset due 318 * to alignment. This will be op->bytes if no alignment is necessary, 319 * or (new_end - op->offset) if the tail is rounded up or down due to 320 * alignment or buffer limit. 321 */ 322 static void coroutine_fn mirror_co_read(void *opaque) 323 { 324 MirrorOp *op = opaque; 325 MirrorBlockJob *s = op->s; 326 int nb_chunks; 327 uint64_t ret; 328 uint64_t max_bytes; 329 330 max_bytes = s->granularity * s->max_iov; 331 332 /* We can only handle as much as buf_size at a time. */ 333 op->bytes = MIN(s->buf_size, MIN(max_bytes, op->bytes)); 334 assert(op->bytes); 335 assert(op->bytes < BDRV_REQUEST_MAX_BYTES); 336 *op->bytes_handled = op->bytes; 337 338 if (s->cow_bitmap) { 339 *op->bytes_handled += mirror_cow_align(s, &op->offset, &op->bytes); 340 } 341 /* Cannot exceed BDRV_REQUEST_MAX_BYTES + INT_MAX */ 342 assert(*op->bytes_handled <= UINT_MAX); 343 assert(op->bytes <= s->buf_size); 344 /* The offset is granularity-aligned because: 345 * 1) Caller passes in aligned values; 346 * 2) mirror_cow_align is used only when target cluster is larger. */ 347 assert(QEMU_IS_ALIGNED(op->offset, s->granularity)); 348 /* The range is sector-aligned, since bdrv_getlength() rounds up. */ 349 assert(QEMU_IS_ALIGNED(op->bytes, BDRV_SECTOR_SIZE)); 350 nb_chunks = DIV_ROUND_UP(op->bytes, s->granularity); 351 352 while (s->buf_free_count < nb_chunks) { 353 trace_mirror_yield_in_flight(s, op->offset, s->in_flight); 354 mirror_wait_for_free_in_flight_slot(s, op); 355 } 356 357 /* Now make a QEMUIOVector taking enough granularity-sized chunks 358 * from s->buf_free. 359 */ 360 qemu_iovec_init(&op->qiov, nb_chunks); 361 while (nb_chunks-- > 0) { 362 MirrorBuffer *buf = QSIMPLEQ_FIRST(&s->buf_free); 363 size_t remaining = op->bytes - op->qiov.size; 364 365 QSIMPLEQ_REMOVE_HEAD(&s->buf_free, next); 366 s->buf_free_count--; 367 qemu_iovec_add(&op->qiov, buf, MIN(s->granularity, remaining)); 368 } 369 370 /* Copy the dirty cluster. */ 371 s->in_flight++; 372 s->bytes_in_flight += op->bytes; 373 trace_mirror_one_iteration(s, op->offset, op->bytes); 374 375 ret = bdrv_co_preadv(s->mirror_top_bs->backing, op->offset, op->bytes, 376 &op->qiov, 0); 377 mirror_read_complete(op, ret); 378 } 379 380 static void coroutine_fn mirror_co_zero(void *opaque) 381 { 382 MirrorOp *op = opaque; 383 int ret; 384 385 op->s->in_flight++; 386 op->s->bytes_in_flight += op->bytes; 387 *op->bytes_handled = op->bytes; 388 389 ret = blk_co_pwrite_zeroes(op->s->target, op->offset, op->bytes, 390 op->s->unmap ? BDRV_REQ_MAY_UNMAP : 0); 391 mirror_write_complete(op, ret); 392 } 393 394 static void coroutine_fn mirror_co_discard(void *opaque) 395 { 396 MirrorOp *op = opaque; 397 int ret; 398 399 op->s->in_flight++; 400 op->s->bytes_in_flight += op->bytes; 401 *op->bytes_handled = op->bytes; 402 403 ret = blk_co_pdiscard(op->s->target, op->offset, op->bytes); 404 mirror_write_complete(op, ret); 405 } 406 407 static unsigned mirror_perform(MirrorBlockJob *s, int64_t offset, 408 unsigned bytes, MirrorMethod mirror_method) 409 { 410 MirrorOp *op; 411 Coroutine *co; 412 int64_t bytes_handled = -1; 413 414 op = g_new(MirrorOp, 1); 415 *op = (MirrorOp){ 416 .s = s, 417 .offset = offset, 418 .bytes = bytes, 419 .bytes_handled = &bytes_handled, 420 }; 421 qemu_co_queue_init(&op->waiting_requests); 422 423 switch (mirror_method) { 424 case MIRROR_METHOD_COPY: 425 co = qemu_coroutine_create(mirror_co_read, op); 426 break; 427 case MIRROR_METHOD_ZERO: 428 co = qemu_coroutine_create(mirror_co_zero, op); 429 break; 430 case MIRROR_METHOD_DISCARD: 431 co = qemu_coroutine_create(mirror_co_discard, op); 432 break; 433 default: 434 abort(); 435 } 436 op->co = co; 437 438 QTAILQ_INSERT_TAIL(&s->ops_in_flight, op, next); 439 qemu_coroutine_enter(co); 440 /* At this point, ownership of op has been moved to the coroutine 441 * and the object may already be freed */ 442 443 /* Assert that this value has been set */ 444 assert(bytes_handled >= 0); 445 446 /* Same assertion as in mirror_co_read() (and for mirror_co_read() 447 * and mirror_co_discard(), bytes_handled == op->bytes, which 448 * is the @bytes parameter given to this function) */ 449 assert(bytes_handled <= UINT_MAX); 450 return bytes_handled; 451 } 452 453 static uint64_t coroutine_fn mirror_iteration(MirrorBlockJob *s) 454 { 455 BlockDriverState *source = s->mirror_top_bs->backing->bs; 456 MirrorOp *pseudo_op; 457 int64_t offset; 458 uint64_t delay_ns = 0, ret = 0; 459 /* At least the first dirty chunk is mirrored in one iteration. */ 460 int nb_chunks = 1; 461 bool write_zeroes_ok = bdrv_can_write_zeroes_with_unmap(blk_bs(s->target)); 462 int max_io_bytes = MAX(s->buf_size / MAX_IN_FLIGHT, MAX_IO_BYTES); 463 464 bdrv_dirty_bitmap_lock(s->dirty_bitmap); 465 offset = bdrv_dirty_iter_next(s->dbi); 466 if (offset < 0) { 467 bdrv_set_dirty_iter(s->dbi, 0); 468 offset = bdrv_dirty_iter_next(s->dbi); 469 trace_mirror_restart_iter(s, bdrv_get_dirty_count(s->dirty_bitmap)); 470 assert(offset >= 0); 471 } 472 bdrv_dirty_bitmap_unlock(s->dirty_bitmap); 473 474 mirror_wait_on_conflicts(NULL, s, offset, 1); 475 476 job_pause_point(&s->common.job); 477 478 /* Find the number of consective dirty chunks following the first dirty 479 * one, and wait for in flight requests in them. */ 480 bdrv_dirty_bitmap_lock(s->dirty_bitmap); 481 while (nb_chunks * s->granularity < s->buf_size) { 482 int64_t next_dirty; 483 int64_t next_offset = offset + nb_chunks * s->granularity; 484 int64_t next_chunk = next_offset / s->granularity; 485 if (next_offset >= s->bdev_length || 486 !bdrv_dirty_bitmap_get_locked(s->dirty_bitmap, next_offset)) { 487 break; 488 } 489 if (test_bit(next_chunk, s->in_flight_bitmap)) { 490 break; 491 } 492 493 next_dirty = bdrv_dirty_iter_next(s->dbi); 494 if (next_dirty > next_offset || next_dirty < 0) { 495 /* The bitmap iterator's cache is stale, refresh it */ 496 bdrv_set_dirty_iter(s->dbi, next_offset); 497 next_dirty = bdrv_dirty_iter_next(s->dbi); 498 } 499 assert(next_dirty == next_offset); 500 nb_chunks++; 501 } 502 503 /* Clear dirty bits before querying the block status, because 504 * calling bdrv_block_status_above could yield - if some blocks are 505 * marked dirty in this window, we need to know. 506 */ 507 bdrv_reset_dirty_bitmap_locked(s->dirty_bitmap, offset, 508 nb_chunks * s->granularity); 509 bdrv_dirty_bitmap_unlock(s->dirty_bitmap); 510 511 /* Before claiming an area in the in-flight bitmap, we have to 512 * create a MirrorOp for it so that conflicting requests can wait 513 * for it. mirror_perform() will create the real MirrorOps later, 514 * for now we just create a pseudo operation that will wake up all 515 * conflicting requests once all real operations have been 516 * launched. */ 517 pseudo_op = g_new(MirrorOp, 1); 518 *pseudo_op = (MirrorOp){ 519 .offset = offset, 520 .bytes = nb_chunks * s->granularity, 521 .is_pseudo_op = true, 522 }; 523 qemu_co_queue_init(&pseudo_op->waiting_requests); 524 QTAILQ_INSERT_TAIL(&s->ops_in_flight, pseudo_op, next); 525 526 bitmap_set(s->in_flight_bitmap, offset / s->granularity, nb_chunks); 527 while (nb_chunks > 0 && offset < s->bdev_length) { 528 int ret; 529 int64_t io_bytes; 530 int64_t io_bytes_acct; 531 MirrorMethod mirror_method = MIRROR_METHOD_COPY; 532 533 assert(!(offset % s->granularity)); 534 ret = bdrv_block_status_above(source, NULL, offset, 535 nb_chunks * s->granularity, 536 &io_bytes, NULL, NULL); 537 if (ret < 0) { 538 io_bytes = MIN(nb_chunks * s->granularity, max_io_bytes); 539 } else if (ret & BDRV_BLOCK_DATA) { 540 io_bytes = MIN(io_bytes, max_io_bytes); 541 } 542 543 io_bytes -= io_bytes % s->granularity; 544 if (io_bytes < s->granularity) { 545 io_bytes = s->granularity; 546 } else if (ret >= 0 && !(ret & BDRV_BLOCK_DATA)) { 547 int64_t target_offset; 548 int64_t target_bytes; 549 bdrv_round_to_clusters(blk_bs(s->target), offset, io_bytes, 550 &target_offset, &target_bytes); 551 if (target_offset == offset && 552 target_bytes == io_bytes) { 553 mirror_method = ret & BDRV_BLOCK_ZERO ? 554 MIRROR_METHOD_ZERO : 555 MIRROR_METHOD_DISCARD; 556 } 557 } 558 559 while (s->in_flight >= MAX_IN_FLIGHT) { 560 trace_mirror_yield_in_flight(s, offset, s->in_flight); 561 mirror_wait_for_free_in_flight_slot(s, pseudo_op); 562 } 563 564 if (s->ret < 0) { 565 ret = 0; 566 goto fail; 567 } 568 569 io_bytes = mirror_clip_bytes(s, offset, io_bytes); 570 io_bytes = mirror_perform(s, offset, io_bytes, mirror_method); 571 if (mirror_method != MIRROR_METHOD_COPY && write_zeroes_ok) { 572 io_bytes_acct = 0; 573 } else { 574 io_bytes_acct = io_bytes; 575 } 576 assert(io_bytes); 577 offset += io_bytes; 578 nb_chunks -= DIV_ROUND_UP(io_bytes, s->granularity); 579 delay_ns = block_job_ratelimit_get_delay(&s->common, io_bytes_acct); 580 } 581 582 ret = delay_ns; 583 fail: 584 QTAILQ_REMOVE(&s->ops_in_flight, pseudo_op, next); 585 qemu_co_queue_restart_all(&pseudo_op->waiting_requests); 586 g_free(pseudo_op); 587 588 return ret; 589 } 590 591 static void mirror_free_init(MirrorBlockJob *s) 592 { 593 int granularity = s->granularity; 594 size_t buf_size = s->buf_size; 595 uint8_t *buf = s->buf; 596 597 assert(s->buf_free_count == 0); 598 QSIMPLEQ_INIT(&s->buf_free); 599 while (buf_size != 0) { 600 MirrorBuffer *cur = (MirrorBuffer *)buf; 601 QSIMPLEQ_INSERT_TAIL(&s->buf_free, cur, next); 602 s->buf_free_count++; 603 buf_size -= granularity; 604 buf += granularity; 605 } 606 } 607 608 /* This is also used for the .pause callback. There is no matching 609 * mirror_resume() because mirror_run() will begin iterating again 610 * when the job is resumed. 611 */ 612 static void coroutine_fn mirror_wait_for_all_io(MirrorBlockJob *s) 613 { 614 while (s->in_flight > 0) { 615 mirror_wait_for_free_in_flight_slot(s, NULL); 616 } 617 } 618 619 /** 620 * mirror_exit_common: handle both abort() and prepare() cases. 621 * for .prepare, returns 0 on success and -errno on failure. 622 * for .abort cases, denoted by abort = true, MUST return 0. 623 */ 624 static int mirror_exit_common(Job *job) 625 { 626 MirrorBlockJob *s = container_of(job, MirrorBlockJob, common.job); 627 BlockJob *bjob = &s->common; 628 MirrorBDSOpaque *bs_opaque; 629 AioContext *replace_aio_context = NULL; 630 BlockDriverState *src; 631 BlockDriverState *target_bs; 632 BlockDriverState *mirror_top_bs; 633 Error *local_err = NULL; 634 bool abort = job->ret < 0; 635 int ret = 0; 636 637 if (s->prepared) { 638 return 0; 639 } 640 s->prepared = true; 641 642 mirror_top_bs = s->mirror_top_bs; 643 bs_opaque = mirror_top_bs->opaque; 644 src = mirror_top_bs->backing->bs; 645 target_bs = blk_bs(s->target); 646 647 if (bdrv_chain_contains(src, target_bs)) { 648 bdrv_unfreeze_backing_chain(mirror_top_bs, target_bs); 649 } 650 651 bdrv_release_dirty_bitmap(s->dirty_bitmap); 652 653 /* Make sure that the source BDS doesn't go away during bdrv_replace_node, 654 * before we can call bdrv_drained_end */ 655 bdrv_ref(src); 656 bdrv_ref(mirror_top_bs); 657 bdrv_ref(target_bs); 658 659 /* 660 * Remove target parent that still uses BLK_PERM_WRITE/RESIZE before 661 * inserting target_bs at s->to_replace, where we might not be able to get 662 * these permissions. 663 */ 664 blk_unref(s->target); 665 s->target = NULL; 666 667 /* We don't access the source any more. Dropping any WRITE/RESIZE is 668 * required before it could become a backing file of target_bs. Not having 669 * these permissions any more means that we can't allow any new requests on 670 * mirror_top_bs from now on, so keep it drained. */ 671 bdrv_drained_begin(mirror_top_bs); 672 bs_opaque->stop = true; 673 bdrv_child_refresh_perms(mirror_top_bs, mirror_top_bs->backing, 674 &error_abort); 675 if (!abort && s->backing_mode == MIRROR_SOURCE_BACKING_CHAIN) { 676 BlockDriverState *backing = s->is_none_mode ? src : s->base; 677 if (backing_bs(target_bs) != backing) { 678 bdrv_set_backing_hd(target_bs, backing, &local_err); 679 if (local_err) { 680 error_report_err(local_err); 681 ret = -EPERM; 682 } 683 } 684 } 685 686 if (s->to_replace) { 687 replace_aio_context = bdrv_get_aio_context(s->to_replace); 688 aio_context_acquire(replace_aio_context); 689 } 690 691 if (s->should_complete && !abort) { 692 BlockDriverState *to_replace = s->to_replace ?: src; 693 bool ro = bdrv_is_read_only(to_replace); 694 695 if (ro != bdrv_is_read_only(target_bs)) { 696 bdrv_reopen_set_read_only(target_bs, ro, NULL); 697 } 698 699 /* The mirror job has no requests in flight any more, but we need to 700 * drain potential other users of the BDS before changing the graph. */ 701 assert(s->in_drain); 702 bdrv_drained_begin(target_bs); 703 /* 704 * Cannot use check_to_replace_node() here, because that would 705 * check for an op blocker on @to_replace, and we have our own 706 * there. 707 */ 708 if (bdrv_recurse_can_replace(src, to_replace)) { 709 bdrv_replace_node(to_replace, target_bs, &local_err); 710 } else { 711 error_setg(&local_err, "Can no longer replace '%s' by '%s', " 712 "because it can no longer be guaranteed that doing so " 713 "would not lead to an abrupt change of visible data", 714 to_replace->node_name, target_bs->node_name); 715 } 716 bdrv_drained_end(target_bs); 717 if (local_err) { 718 error_report_err(local_err); 719 ret = -EPERM; 720 } 721 } 722 if (s->to_replace) { 723 bdrv_op_unblock_all(s->to_replace, s->replace_blocker); 724 error_free(s->replace_blocker); 725 bdrv_unref(s->to_replace); 726 } 727 if (replace_aio_context) { 728 aio_context_release(replace_aio_context); 729 } 730 g_free(s->replaces); 731 bdrv_unref(target_bs); 732 733 /* 734 * Remove the mirror filter driver from the graph. Before this, get rid of 735 * the blockers on the intermediate nodes so that the resulting state is 736 * valid. 737 */ 738 block_job_remove_all_bdrv(bjob); 739 bdrv_replace_node(mirror_top_bs, backing_bs(mirror_top_bs), &error_abort); 740 741 /* We just changed the BDS the job BB refers to (with either or both of the 742 * bdrv_replace_node() calls), so switch the BB back so the cleanup does 743 * the right thing. We don't need any permissions any more now. */ 744 blk_remove_bs(bjob->blk); 745 blk_set_perm(bjob->blk, 0, BLK_PERM_ALL, &error_abort); 746 blk_insert_bs(bjob->blk, mirror_top_bs, &error_abort); 747 748 bs_opaque->job = NULL; 749 750 bdrv_drained_end(src); 751 bdrv_drained_end(mirror_top_bs); 752 s->in_drain = false; 753 bdrv_unref(mirror_top_bs); 754 bdrv_unref(src); 755 756 return ret; 757 } 758 759 static int mirror_prepare(Job *job) 760 { 761 return mirror_exit_common(job); 762 } 763 764 static void mirror_abort(Job *job) 765 { 766 int ret = mirror_exit_common(job); 767 assert(ret == 0); 768 } 769 770 static void coroutine_fn mirror_throttle(MirrorBlockJob *s) 771 { 772 int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME); 773 774 if (now - s->last_pause_ns > BLOCK_JOB_SLICE_TIME) { 775 s->last_pause_ns = now; 776 job_sleep_ns(&s->common.job, 0); 777 } else { 778 job_pause_point(&s->common.job); 779 } 780 } 781 782 static int coroutine_fn mirror_dirty_init(MirrorBlockJob *s) 783 { 784 int64_t offset; 785 BlockDriverState *base = s->base; 786 BlockDriverState *bs = s->mirror_top_bs->backing->bs; 787 BlockDriverState *target_bs = blk_bs(s->target); 788 int ret; 789 int64_t count; 790 791 if (s->zero_target) { 792 if (!bdrv_can_write_zeroes_with_unmap(target_bs)) { 793 bdrv_set_dirty_bitmap(s->dirty_bitmap, 0, s->bdev_length); 794 return 0; 795 } 796 797 s->initial_zeroing_ongoing = true; 798 for (offset = 0; offset < s->bdev_length; ) { 799 int bytes = MIN(s->bdev_length - offset, 800 QEMU_ALIGN_DOWN(INT_MAX, s->granularity)); 801 802 mirror_throttle(s); 803 804 if (job_is_cancelled(&s->common.job)) { 805 s->initial_zeroing_ongoing = false; 806 return 0; 807 } 808 809 if (s->in_flight >= MAX_IN_FLIGHT) { 810 trace_mirror_yield(s, UINT64_MAX, s->buf_free_count, 811 s->in_flight); 812 mirror_wait_for_free_in_flight_slot(s, NULL); 813 continue; 814 } 815 816 mirror_perform(s, offset, bytes, MIRROR_METHOD_ZERO); 817 offset += bytes; 818 } 819 820 mirror_wait_for_all_io(s); 821 s->initial_zeroing_ongoing = false; 822 } 823 824 /* First part, loop on the sectors and initialize the dirty bitmap. */ 825 for (offset = 0; offset < s->bdev_length; ) { 826 /* Just to make sure we are not exceeding int limit. */ 827 int bytes = MIN(s->bdev_length - offset, 828 QEMU_ALIGN_DOWN(INT_MAX, s->granularity)); 829 830 mirror_throttle(s); 831 832 if (job_is_cancelled(&s->common.job)) { 833 return 0; 834 } 835 836 ret = bdrv_is_allocated_above(bs, base, false, offset, bytes, &count); 837 if (ret < 0) { 838 return ret; 839 } 840 841 assert(count); 842 if (ret == 1) { 843 bdrv_set_dirty_bitmap(s->dirty_bitmap, offset, count); 844 } 845 offset += count; 846 } 847 return 0; 848 } 849 850 /* Called when going out of the streaming phase to flush the bulk of the 851 * data to the medium, or just before completing. 852 */ 853 static int mirror_flush(MirrorBlockJob *s) 854 { 855 int ret = blk_flush(s->target); 856 if (ret < 0) { 857 if (mirror_error_action(s, false, -ret) == BLOCK_ERROR_ACTION_REPORT) { 858 s->ret = ret; 859 } 860 } 861 return ret; 862 } 863 864 static int coroutine_fn mirror_run(Job *job, Error **errp) 865 { 866 MirrorBlockJob *s = container_of(job, MirrorBlockJob, common.job); 867 BlockDriverState *bs = s->mirror_top_bs->backing->bs; 868 BlockDriverState *target_bs = blk_bs(s->target); 869 bool need_drain = true; 870 int64_t length; 871 BlockDriverInfo bdi; 872 char backing_filename[2]; /* we only need 2 characters because we are only 873 checking for a NULL string */ 874 int ret = 0; 875 876 if (job_is_cancelled(&s->common.job)) { 877 goto immediate_exit; 878 } 879 880 s->bdev_length = bdrv_getlength(bs); 881 if (s->bdev_length < 0) { 882 ret = s->bdev_length; 883 goto immediate_exit; 884 } 885 886 /* Active commit must resize the base image if its size differs from the 887 * active layer. */ 888 if (s->base == blk_bs(s->target)) { 889 int64_t base_length; 890 891 base_length = blk_getlength(s->target); 892 if (base_length < 0) { 893 ret = base_length; 894 goto immediate_exit; 895 } 896 897 if (s->bdev_length > base_length) { 898 ret = blk_truncate(s->target, s->bdev_length, false, 899 PREALLOC_MODE_OFF, NULL); 900 if (ret < 0) { 901 goto immediate_exit; 902 } 903 } 904 } 905 906 if (s->bdev_length == 0) { 907 /* Transition to the READY state and wait for complete. */ 908 job_transition_to_ready(&s->common.job); 909 s->synced = true; 910 s->actively_synced = true; 911 while (!job_is_cancelled(&s->common.job) && !s->should_complete) { 912 job_yield(&s->common.job); 913 } 914 s->common.job.cancelled = false; 915 goto immediate_exit; 916 } 917 918 length = DIV_ROUND_UP(s->bdev_length, s->granularity); 919 s->in_flight_bitmap = bitmap_new(length); 920 921 /* If we have no backing file yet in the destination, we cannot let 922 * the destination do COW. Instead, we copy sectors around the 923 * dirty data if needed. We need a bitmap to do that. 924 */ 925 bdrv_get_backing_filename(target_bs, backing_filename, 926 sizeof(backing_filename)); 927 if (!bdrv_get_info(target_bs, &bdi) && bdi.cluster_size) { 928 s->target_cluster_size = bdi.cluster_size; 929 } else { 930 s->target_cluster_size = BDRV_SECTOR_SIZE; 931 } 932 if (backing_filename[0] && !target_bs->backing && 933 s->granularity < s->target_cluster_size) { 934 s->buf_size = MAX(s->buf_size, s->target_cluster_size); 935 s->cow_bitmap = bitmap_new(length); 936 } 937 s->max_iov = MIN(bs->bl.max_iov, target_bs->bl.max_iov); 938 939 s->buf = qemu_try_blockalign(bs, s->buf_size); 940 if (s->buf == NULL) { 941 ret = -ENOMEM; 942 goto immediate_exit; 943 } 944 945 mirror_free_init(s); 946 947 s->last_pause_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME); 948 if (!s->is_none_mode) { 949 ret = mirror_dirty_init(s); 950 if (ret < 0 || job_is_cancelled(&s->common.job)) { 951 goto immediate_exit; 952 } 953 } 954 955 assert(!s->dbi); 956 s->dbi = bdrv_dirty_iter_new(s->dirty_bitmap); 957 for (;;) { 958 uint64_t delay_ns = 0; 959 int64_t cnt, delta; 960 bool should_complete; 961 962 /* Do not start passive operations while there are active 963 * writes in progress */ 964 while (s->in_active_write_counter) { 965 mirror_wait_for_any_operation(s, NULL, true); 966 } 967 968 if (s->ret < 0) { 969 ret = s->ret; 970 goto immediate_exit; 971 } 972 973 job_pause_point(&s->common.job); 974 975 cnt = bdrv_get_dirty_count(s->dirty_bitmap); 976 /* cnt is the number of dirty bytes remaining and s->bytes_in_flight is 977 * the number of bytes currently being processed; together those are 978 * the current remaining operation length */ 979 job_progress_set_remaining(&s->common.job, s->bytes_in_flight + cnt); 980 981 /* Note that even when no rate limit is applied we need to yield 982 * periodically with no pending I/O so that bdrv_drain_all() returns. 983 * We do so every BLKOCK_JOB_SLICE_TIME nanoseconds, or when there is 984 * an error, or when the source is clean, whichever comes first. */ 985 delta = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - s->last_pause_ns; 986 if (delta < BLOCK_JOB_SLICE_TIME && 987 s->common.iostatus == BLOCK_DEVICE_IO_STATUS_OK) { 988 if (s->in_flight >= MAX_IN_FLIGHT || s->buf_free_count == 0 || 989 (cnt == 0 && s->in_flight > 0)) { 990 trace_mirror_yield(s, cnt, s->buf_free_count, s->in_flight); 991 mirror_wait_for_free_in_flight_slot(s, NULL); 992 continue; 993 } else if (cnt != 0) { 994 delay_ns = mirror_iteration(s); 995 } 996 } 997 998 should_complete = false; 999 if (s->in_flight == 0 && cnt == 0) { 1000 trace_mirror_before_flush(s); 1001 if (!s->synced) { 1002 if (mirror_flush(s) < 0) { 1003 /* Go check s->ret. */ 1004 continue; 1005 } 1006 /* We're out of the streaming phase. From now on, if the job 1007 * is cancelled we will actually complete all pending I/O and 1008 * report completion. This way, block-job-cancel will leave 1009 * the target in a consistent state. 1010 */ 1011 job_transition_to_ready(&s->common.job); 1012 s->synced = true; 1013 if (s->copy_mode != MIRROR_COPY_MODE_BACKGROUND) { 1014 s->actively_synced = true; 1015 } 1016 } 1017 1018 should_complete = s->should_complete || 1019 job_is_cancelled(&s->common.job); 1020 cnt = bdrv_get_dirty_count(s->dirty_bitmap); 1021 } 1022 1023 if (cnt == 0 && should_complete) { 1024 /* The dirty bitmap is not updated while operations are pending. 1025 * If we're about to exit, wait for pending operations before 1026 * calling bdrv_get_dirty_count(bs), or we may exit while the 1027 * source has dirty data to copy! 1028 * 1029 * Note that I/O can be submitted by the guest while 1030 * mirror_populate runs, so pause it now. Before deciding 1031 * whether to switch to target check one last time if I/O has 1032 * come in the meanwhile, and if not flush the data to disk. 1033 */ 1034 trace_mirror_before_drain(s, cnt); 1035 1036 s->in_drain = true; 1037 bdrv_drained_begin(bs); 1038 cnt = bdrv_get_dirty_count(s->dirty_bitmap); 1039 if (cnt > 0 || mirror_flush(s) < 0) { 1040 bdrv_drained_end(bs); 1041 s->in_drain = false; 1042 continue; 1043 } 1044 1045 /* The two disks are in sync. Exit and report successful 1046 * completion. 1047 */ 1048 assert(QLIST_EMPTY(&bs->tracked_requests)); 1049 s->common.job.cancelled = false; 1050 need_drain = false; 1051 break; 1052 } 1053 1054 ret = 0; 1055 1056 if (s->synced && !should_complete) { 1057 delay_ns = (s->in_flight == 0 && 1058 cnt == 0 ? BLOCK_JOB_SLICE_TIME : 0); 1059 } 1060 trace_mirror_before_sleep(s, cnt, s->synced, delay_ns); 1061 job_sleep_ns(&s->common.job, delay_ns); 1062 if (job_is_cancelled(&s->common.job) && 1063 (!s->synced || s->common.job.force_cancel)) 1064 { 1065 break; 1066 } 1067 s->last_pause_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME); 1068 } 1069 1070 immediate_exit: 1071 if (s->in_flight > 0) { 1072 /* We get here only if something went wrong. Either the job failed, 1073 * or it was cancelled prematurely so that we do not guarantee that 1074 * the target is a copy of the source. 1075 */ 1076 assert(ret < 0 || ((s->common.job.force_cancel || !s->synced) && 1077 job_is_cancelled(&s->common.job))); 1078 assert(need_drain); 1079 mirror_wait_for_all_io(s); 1080 } 1081 1082 assert(s->in_flight == 0); 1083 qemu_vfree(s->buf); 1084 g_free(s->cow_bitmap); 1085 g_free(s->in_flight_bitmap); 1086 bdrv_dirty_iter_free(s->dbi); 1087 1088 if (need_drain) { 1089 s->in_drain = true; 1090 bdrv_drained_begin(bs); 1091 } 1092 1093 return ret; 1094 } 1095 1096 static void mirror_complete(Job *job, Error **errp) 1097 { 1098 MirrorBlockJob *s = container_of(job, MirrorBlockJob, common.job); 1099 BlockDriverState *target; 1100 1101 target = blk_bs(s->target); 1102 1103 if (!s->synced) { 1104 error_setg(errp, "The active block job '%s' cannot be completed", 1105 job->id); 1106 return; 1107 } 1108 1109 if (s->backing_mode == MIRROR_OPEN_BACKING_CHAIN) { 1110 int ret; 1111 1112 assert(!target->backing); 1113 ret = bdrv_open_backing_file(target, NULL, "backing", errp); 1114 if (ret < 0) { 1115 return; 1116 } 1117 } 1118 1119 /* block all operations on to_replace bs */ 1120 if (s->replaces) { 1121 AioContext *replace_aio_context; 1122 1123 s->to_replace = bdrv_find_node(s->replaces); 1124 if (!s->to_replace) { 1125 error_setg(errp, "Node name '%s' not found", s->replaces); 1126 return; 1127 } 1128 1129 replace_aio_context = bdrv_get_aio_context(s->to_replace); 1130 aio_context_acquire(replace_aio_context); 1131 1132 /* TODO Translate this into permission system. Current definition of 1133 * GRAPH_MOD would require to request it for the parents; they might 1134 * not even be BlockDriverStates, however, so a BdrvChild can't address 1135 * them. May need redefinition of GRAPH_MOD. */ 1136 error_setg(&s->replace_blocker, 1137 "block device is in use by block-job-complete"); 1138 bdrv_op_block_all(s->to_replace, s->replace_blocker); 1139 bdrv_ref(s->to_replace); 1140 1141 aio_context_release(replace_aio_context); 1142 } 1143 1144 s->should_complete = true; 1145 job_enter(job); 1146 } 1147 1148 static void coroutine_fn mirror_pause(Job *job) 1149 { 1150 MirrorBlockJob *s = container_of(job, MirrorBlockJob, common.job); 1151 1152 mirror_wait_for_all_io(s); 1153 } 1154 1155 static bool mirror_drained_poll(BlockJob *job) 1156 { 1157 MirrorBlockJob *s = container_of(job, MirrorBlockJob, common); 1158 1159 /* If the job isn't paused nor cancelled, we can't be sure that it won't 1160 * issue more requests. We make an exception if we've reached this point 1161 * from one of our own drain sections, to avoid a deadlock waiting for 1162 * ourselves. 1163 */ 1164 if (!s->common.job.paused && !s->common.job.cancelled && !s->in_drain) { 1165 return true; 1166 } 1167 1168 return !!s->in_flight; 1169 } 1170 1171 static const BlockJobDriver mirror_job_driver = { 1172 .job_driver = { 1173 .instance_size = sizeof(MirrorBlockJob), 1174 .job_type = JOB_TYPE_MIRROR, 1175 .free = block_job_free, 1176 .user_resume = block_job_user_resume, 1177 .run = mirror_run, 1178 .prepare = mirror_prepare, 1179 .abort = mirror_abort, 1180 .pause = mirror_pause, 1181 .complete = mirror_complete, 1182 }, 1183 .drained_poll = mirror_drained_poll, 1184 }; 1185 1186 static const BlockJobDriver commit_active_job_driver = { 1187 .job_driver = { 1188 .instance_size = sizeof(MirrorBlockJob), 1189 .job_type = JOB_TYPE_COMMIT, 1190 .free = block_job_free, 1191 .user_resume = block_job_user_resume, 1192 .run = mirror_run, 1193 .prepare = mirror_prepare, 1194 .abort = mirror_abort, 1195 .pause = mirror_pause, 1196 .complete = mirror_complete, 1197 }, 1198 .drained_poll = mirror_drained_poll, 1199 }; 1200 1201 static void coroutine_fn 1202 do_sync_target_write(MirrorBlockJob *job, MirrorMethod method, 1203 uint64_t offset, uint64_t bytes, 1204 QEMUIOVector *qiov, int flags) 1205 { 1206 int ret; 1207 size_t qiov_offset = 0; 1208 int64_t bitmap_offset, bitmap_end; 1209 1210 if (!QEMU_IS_ALIGNED(offset, job->granularity) && 1211 bdrv_dirty_bitmap_get(job->dirty_bitmap, offset)) 1212 { 1213 /* 1214 * Dirty unaligned padding: ignore it. 1215 * 1216 * Reasoning: 1217 * 1. If we copy it, we can't reset corresponding bit in 1218 * dirty_bitmap as there may be some "dirty" bytes still not 1219 * copied. 1220 * 2. It's already dirty, so skipping it we don't diverge mirror 1221 * progress. 1222 * 1223 * Note, that because of this, guest write may have no contribution 1224 * into mirror converge, but that's not bad, as we have background 1225 * process of mirroring. If under some bad circumstances (high guest 1226 * IO load) background process starve, we will not converge anyway, 1227 * even if each write will contribute, as guest is not guaranteed to 1228 * rewrite the whole disk. 1229 */ 1230 qiov_offset = QEMU_ALIGN_UP(offset, job->granularity) - offset; 1231 if (bytes <= qiov_offset) { 1232 /* nothing to do after shrink */ 1233 return; 1234 } 1235 offset += qiov_offset; 1236 bytes -= qiov_offset; 1237 } 1238 1239 if (!QEMU_IS_ALIGNED(offset + bytes, job->granularity) && 1240 bdrv_dirty_bitmap_get(job->dirty_bitmap, offset + bytes - 1)) 1241 { 1242 uint64_t tail = (offset + bytes) % job->granularity; 1243 1244 if (bytes <= tail) { 1245 /* nothing to do after shrink */ 1246 return; 1247 } 1248 bytes -= tail; 1249 } 1250 1251 /* 1252 * Tails are either clean or shrunk, so for bitmap resetting 1253 * we safely align the range down. 1254 */ 1255 bitmap_offset = QEMU_ALIGN_UP(offset, job->granularity); 1256 bitmap_end = QEMU_ALIGN_DOWN(offset + bytes, job->granularity); 1257 if (bitmap_offset < bitmap_end) { 1258 bdrv_reset_dirty_bitmap(job->dirty_bitmap, bitmap_offset, 1259 bitmap_end - bitmap_offset); 1260 } 1261 1262 job_progress_increase_remaining(&job->common.job, bytes); 1263 1264 switch (method) { 1265 case MIRROR_METHOD_COPY: 1266 ret = blk_co_pwritev_part(job->target, offset, bytes, 1267 qiov, qiov_offset, flags); 1268 break; 1269 1270 case MIRROR_METHOD_ZERO: 1271 assert(!qiov); 1272 ret = blk_co_pwrite_zeroes(job->target, offset, bytes, flags); 1273 break; 1274 1275 case MIRROR_METHOD_DISCARD: 1276 assert(!qiov); 1277 ret = blk_co_pdiscard(job->target, offset, bytes); 1278 break; 1279 1280 default: 1281 abort(); 1282 } 1283 1284 if (ret >= 0) { 1285 job_progress_update(&job->common.job, bytes); 1286 } else { 1287 BlockErrorAction action; 1288 1289 /* 1290 * We failed, so we should mark dirty the whole area, aligned up. 1291 * Note that we don't care about shrunk tails if any: they were dirty 1292 * at function start, and they must be still dirty, as we've locked 1293 * the region for in-flight op. 1294 */ 1295 bitmap_offset = QEMU_ALIGN_DOWN(offset, job->granularity); 1296 bitmap_end = QEMU_ALIGN_UP(offset + bytes, job->granularity); 1297 bdrv_set_dirty_bitmap(job->dirty_bitmap, bitmap_offset, 1298 bitmap_end - bitmap_offset); 1299 job->actively_synced = false; 1300 1301 action = mirror_error_action(job, false, -ret); 1302 if (action == BLOCK_ERROR_ACTION_REPORT) { 1303 if (!job->ret) { 1304 job->ret = ret; 1305 } 1306 } 1307 } 1308 } 1309 1310 static MirrorOp *coroutine_fn active_write_prepare(MirrorBlockJob *s, 1311 uint64_t offset, 1312 uint64_t bytes) 1313 { 1314 MirrorOp *op; 1315 uint64_t start_chunk = offset / s->granularity; 1316 uint64_t end_chunk = DIV_ROUND_UP(offset + bytes, s->granularity); 1317 1318 op = g_new(MirrorOp, 1); 1319 *op = (MirrorOp){ 1320 .s = s, 1321 .offset = offset, 1322 .bytes = bytes, 1323 .is_active_write = true, 1324 }; 1325 qemu_co_queue_init(&op->waiting_requests); 1326 QTAILQ_INSERT_TAIL(&s->ops_in_flight, op, next); 1327 1328 s->in_active_write_counter++; 1329 1330 mirror_wait_on_conflicts(op, s, offset, bytes); 1331 1332 bitmap_set(s->in_flight_bitmap, start_chunk, end_chunk - start_chunk); 1333 1334 return op; 1335 } 1336 1337 static void coroutine_fn active_write_settle(MirrorOp *op) 1338 { 1339 uint64_t start_chunk = op->offset / op->s->granularity; 1340 uint64_t end_chunk = DIV_ROUND_UP(op->offset + op->bytes, 1341 op->s->granularity); 1342 1343 if (!--op->s->in_active_write_counter && op->s->actively_synced) { 1344 BdrvChild *source = op->s->mirror_top_bs->backing; 1345 1346 if (QLIST_FIRST(&source->bs->parents) == source && 1347 QLIST_NEXT(source, next_parent) == NULL) 1348 { 1349 /* Assert that we are back in sync once all active write 1350 * operations are settled. 1351 * Note that we can only assert this if the mirror node 1352 * is the source node's only parent. */ 1353 assert(!bdrv_get_dirty_count(op->s->dirty_bitmap)); 1354 } 1355 } 1356 bitmap_clear(op->s->in_flight_bitmap, start_chunk, end_chunk - start_chunk); 1357 QTAILQ_REMOVE(&op->s->ops_in_flight, op, next); 1358 qemu_co_queue_restart_all(&op->waiting_requests); 1359 g_free(op); 1360 } 1361 1362 static int coroutine_fn bdrv_mirror_top_preadv(BlockDriverState *bs, 1363 uint64_t offset, uint64_t bytes, QEMUIOVector *qiov, int flags) 1364 { 1365 return bdrv_co_preadv(bs->backing, offset, bytes, qiov, flags); 1366 } 1367 1368 static int coroutine_fn bdrv_mirror_top_do_write(BlockDriverState *bs, 1369 MirrorMethod method, uint64_t offset, uint64_t bytes, QEMUIOVector *qiov, 1370 int flags) 1371 { 1372 MirrorOp *op = NULL; 1373 MirrorBDSOpaque *s = bs->opaque; 1374 int ret = 0; 1375 bool copy_to_target; 1376 1377 copy_to_target = s->job->ret >= 0 && 1378 s->job->copy_mode == MIRROR_COPY_MODE_WRITE_BLOCKING; 1379 1380 if (copy_to_target) { 1381 op = active_write_prepare(s->job, offset, bytes); 1382 } 1383 1384 switch (method) { 1385 case MIRROR_METHOD_COPY: 1386 ret = bdrv_co_pwritev(bs->backing, offset, bytes, qiov, flags); 1387 break; 1388 1389 case MIRROR_METHOD_ZERO: 1390 ret = bdrv_co_pwrite_zeroes(bs->backing, offset, bytes, flags); 1391 break; 1392 1393 case MIRROR_METHOD_DISCARD: 1394 ret = bdrv_co_pdiscard(bs->backing, offset, bytes); 1395 break; 1396 1397 default: 1398 abort(); 1399 } 1400 1401 if (ret < 0) { 1402 goto out; 1403 } 1404 1405 if (copy_to_target) { 1406 do_sync_target_write(s->job, method, offset, bytes, qiov, flags); 1407 } 1408 1409 out: 1410 if (copy_to_target) { 1411 active_write_settle(op); 1412 } 1413 return ret; 1414 } 1415 1416 static int coroutine_fn bdrv_mirror_top_pwritev(BlockDriverState *bs, 1417 uint64_t offset, uint64_t bytes, QEMUIOVector *qiov, int flags) 1418 { 1419 MirrorBDSOpaque *s = bs->opaque; 1420 QEMUIOVector bounce_qiov; 1421 void *bounce_buf; 1422 int ret = 0; 1423 bool copy_to_target; 1424 1425 copy_to_target = s->job->ret >= 0 && 1426 s->job->copy_mode == MIRROR_COPY_MODE_WRITE_BLOCKING; 1427 1428 if (copy_to_target) { 1429 /* The guest might concurrently modify the data to write; but 1430 * the data on source and destination must match, so we have 1431 * to use a bounce buffer if we are going to write to the 1432 * target now. */ 1433 bounce_buf = qemu_blockalign(bs, bytes); 1434 iov_to_buf_full(qiov->iov, qiov->niov, 0, bounce_buf, bytes); 1435 1436 qemu_iovec_init(&bounce_qiov, 1); 1437 qemu_iovec_add(&bounce_qiov, bounce_buf, bytes); 1438 qiov = &bounce_qiov; 1439 } 1440 1441 ret = bdrv_mirror_top_do_write(bs, MIRROR_METHOD_COPY, offset, bytes, qiov, 1442 flags); 1443 1444 if (copy_to_target) { 1445 qemu_iovec_destroy(&bounce_qiov); 1446 qemu_vfree(bounce_buf); 1447 } 1448 1449 return ret; 1450 } 1451 1452 static int coroutine_fn bdrv_mirror_top_flush(BlockDriverState *bs) 1453 { 1454 if (bs->backing == NULL) { 1455 /* we can be here after failed bdrv_append in mirror_start_job */ 1456 return 0; 1457 } 1458 return bdrv_co_flush(bs->backing->bs); 1459 } 1460 1461 static int coroutine_fn bdrv_mirror_top_pwrite_zeroes(BlockDriverState *bs, 1462 int64_t offset, int bytes, BdrvRequestFlags flags) 1463 { 1464 return bdrv_mirror_top_do_write(bs, MIRROR_METHOD_ZERO, offset, bytes, NULL, 1465 flags); 1466 } 1467 1468 static int coroutine_fn bdrv_mirror_top_pdiscard(BlockDriverState *bs, 1469 int64_t offset, int bytes) 1470 { 1471 return bdrv_mirror_top_do_write(bs, MIRROR_METHOD_DISCARD, offset, bytes, 1472 NULL, 0); 1473 } 1474 1475 static void bdrv_mirror_top_refresh_filename(BlockDriverState *bs) 1476 { 1477 if (bs->backing == NULL) { 1478 /* we can be here after failed bdrv_attach_child in 1479 * bdrv_set_backing_hd */ 1480 return; 1481 } 1482 pstrcpy(bs->exact_filename, sizeof(bs->exact_filename), 1483 bs->backing->bs->filename); 1484 } 1485 1486 static void bdrv_mirror_top_child_perm(BlockDriverState *bs, BdrvChild *c, 1487 const BdrvChildRole *role, 1488 BlockReopenQueue *reopen_queue, 1489 uint64_t perm, uint64_t shared, 1490 uint64_t *nperm, uint64_t *nshared) 1491 { 1492 MirrorBDSOpaque *s = bs->opaque; 1493 1494 if (s->stop) { 1495 /* 1496 * If the job is to be stopped, we do not need to forward 1497 * anything to the real image. 1498 */ 1499 *nperm = 0; 1500 *nshared = BLK_PERM_ALL; 1501 return; 1502 } 1503 1504 /* Must be able to forward guest writes to the real image */ 1505 *nperm = 0; 1506 if (perm & BLK_PERM_WRITE) { 1507 *nperm |= BLK_PERM_WRITE; 1508 } 1509 1510 *nshared = BLK_PERM_ALL; 1511 } 1512 1513 /* Dummy node that provides consistent read to its users without requiring it 1514 * from its backing file and that allows writes on the backing file chain. */ 1515 static BlockDriver bdrv_mirror_top = { 1516 .format_name = "mirror_top", 1517 .bdrv_co_preadv = bdrv_mirror_top_preadv, 1518 .bdrv_co_pwritev = bdrv_mirror_top_pwritev, 1519 .bdrv_co_pwrite_zeroes = bdrv_mirror_top_pwrite_zeroes, 1520 .bdrv_co_pdiscard = bdrv_mirror_top_pdiscard, 1521 .bdrv_co_flush = bdrv_mirror_top_flush, 1522 .bdrv_co_block_status = bdrv_co_block_status_from_backing, 1523 .bdrv_refresh_filename = bdrv_mirror_top_refresh_filename, 1524 .bdrv_child_perm = bdrv_mirror_top_child_perm, 1525 }; 1526 1527 static BlockJob *mirror_start_job( 1528 const char *job_id, BlockDriverState *bs, 1529 int creation_flags, BlockDriverState *target, 1530 const char *replaces, int64_t speed, 1531 uint32_t granularity, int64_t buf_size, 1532 BlockMirrorBackingMode backing_mode, 1533 bool zero_target, 1534 BlockdevOnError on_source_error, 1535 BlockdevOnError on_target_error, 1536 bool unmap, 1537 BlockCompletionFunc *cb, 1538 void *opaque, 1539 const BlockJobDriver *driver, 1540 bool is_none_mode, BlockDriverState *base, 1541 bool auto_complete, const char *filter_node_name, 1542 bool is_mirror, MirrorCopyMode copy_mode, 1543 Error **errp) 1544 { 1545 MirrorBlockJob *s; 1546 MirrorBDSOpaque *bs_opaque; 1547 BlockDriverState *mirror_top_bs; 1548 bool target_graph_mod; 1549 bool target_is_backing; 1550 Error *local_err = NULL; 1551 int ret; 1552 1553 if (granularity == 0) { 1554 granularity = bdrv_get_default_bitmap_granularity(target); 1555 } 1556 1557 assert(is_power_of_2(granularity)); 1558 1559 if (buf_size < 0) { 1560 error_setg(errp, "Invalid parameter 'buf-size'"); 1561 return NULL; 1562 } 1563 1564 if (buf_size == 0) { 1565 buf_size = DEFAULT_MIRROR_BUF_SIZE; 1566 } 1567 1568 if (bs == target) { 1569 error_setg(errp, "Can't mirror node into itself"); 1570 return NULL; 1571 } 1572 1573 /* In the case of active commit, add dummy driver to provide consistent 1574 * reads on the top, while disabling it in the intermediate nodes, and make 1575 * the backing chain writable. */ 1576 mirror_top_bs = bdrv_new_open_driver(&bdrv_mirror_top, filter_node_name, 1577 BDRV_O_RDWR, errp); 1578 if (mirror_top_bs == NULL) { 1579 return NULL; 1580 } 1581 if (!filter_node_name) { 1582 mirror_top_bs->implicit = true; 1583 } 1584 1585 /* So that we can always drop this node */ 1586 mirror_top_bs->never_freeze = true; 1587 1588 mirror_top_bs->total_sectors = bs->total_sectors; 1589 mirror_top_bs->supported_write_flags = BDRV_REQ_WRITE_UNCHANGED; 1590 mirror_top_bs->supported_zero_flags = BDRV_REQ_WRITE_UNCHANGED | 1591 BDRV_REQ_NO_FALLBACK; 1592 bs_opaque = g_new0(MirrorBDSOpaque, 1); 1593 mirror_top_bs->opaque = bs_opaque; 1594 1595 /* bdrv_append takes ownership of the mirror_top_bs reference, need to keep 1596 * it alive until block_job_create() succeeds even if bs has no parent. */ 1597 bdrv_ref(mirror_top_bs); 1598 bdrv_drained_begin(bs); 1599 bdrv_append(mirror_top_bs, bs, &local_err); 1600 bdrv_drained_end(bs); 1601 1602 if (local_err) { 1603 bdrv_unref(mirror_top_bs); 1604 error_propagate(errp, local_err); 1605 return NULL; 1606 } 1607 1608 /* Make sure that the source is not resized while the job is running */ 1609 s = block_job_create(job_id, driver, NULL, mirror_top_bs, 1610 BLK_PERM_CONSISTENT_READ, 1611 BLK_PERM_CONSISTENT_READ | BLK_PERM_WRITE_UNCHANGED | 1612 BLK_PERM_WRITE | BLK_PERM_GRAPH_MOD, speed, 1613 creation_flags, cb, opaque, errp); 1614 if (!s) { 1615 goto fail; 1616 } 1617 bs_opaque->job = s; 1618 1619 /* The block job now has a reference to this node */ 1620 bdrv_unref(mirror_top_bs); 1621 1622 s->mirror_top_bs = mirror_top_bs; 1623 1624 /* No resize for the target either; while the mirror is still running, a 1625 * consistent read isn't necessarily possible. We could possibly allow 1626 * writes and graph modifications, though it would likely defeat the 1627 * purpose of a mirror, so leave them blocked for now. 1628 * 1629 * In the case of active commit, things look a bit different, though, 1630 * because the target is an already populated backing file in active use. 1631 * We can allow anything except resize there.*/ 1632 target_is_backing = bdrv_chain_contains(bs, target); 1633 target_graph_mod = (backing_mode != MIRROR_LEAVE_BACKING_CHAIN); 1634 s->target = blk_new(s->common.job.aio_context, 1635 BLK_PERM_WRITE | BLK_PERM_RESIZE | 1636 (target_graph_mod ? BLK_PERM_GRAPH_MOD : 0), 1637 BLK_PERM_WRITE_UNCHANGED | 1638 (target_is_backing ? BLK_PERM_CONSISTENT_READ | 1639 BLK_PERM_WRITE | 1640 BLK_PERM_GRAPH_MOD : 0)); 1641 ret = blk_insert_bs(s->target, target, errp); 1642 if (ret < 0) { 1643 goto fail; 1644 } 1645 if (is_mirror) { 1646 /* XXX: Mirror target could be a NBD server of target QEMU in the case 1647 * of non-shared block migration. To allow migration completion, we 1648 * have to allow "inactivate" of the target BB. When that happens, we 1649 * know the job is drained, and the vcpus are stopped, so no write 1650 * operation will be performed. Block layer already has assertions to 1651 * ensure that. */ 1652 blk_set_force_allow_inactivate(s->target); 1653 } 1654 blk_set_allow_aio_context_change(s->target, true); 1655 blk_set_disable_request_queuing(s->target, true); 1656 1657 s->replaces = g_strdup(replaces); 1658 s->on_source_error = on_source_error; 1659 s->on_target_error = on_target_error; 1660 s->is_none_mode = is_none_mode; 1661 s->backing_mode = backing_mode; 1662 s->zero_target = zero_target; 1663 s->copy_mode = copy_mode; 1664 s->base = base; 1665 s->granularity = granularity; 1666 s->buf_size = ROUND_UP(buf_size, granularity); 1667 s->unmap = unmap; 1668 if (auto_complete) { 1669 s->should_complete = true; 1670 } 1671 1672 s->dirty_bitmap = bdrv_create_dirty_bitmap(bs, granularity, NULL, errp); 1673 if (!s->dirty_bitmap) { 1674 goto fail; 1675 } 1676 if (s->copy_mode == MIRROR_COPY_MODE_WRITE_BLOCKING) { 1677 bdrv_disable_dirty_bitmap(s->dirty_bitmap); 1678 } 1679 1680 ret = block_job_add_bdrv(&s->common, "source", bs, 0, 1681 BLK_PERM_WRITE_UNCHANGED | BLK_PERM_WRITE | 1682 BLK_PERM_CONSISTENT_READ, 1683 errp); 1684 if (ret < 0) { 1685 goto fail; 1686 } 1687 1688 /* Required permissions are already taken with blk_new() */ 1689 block_job_add_bdrv(&s->common, "target", target, 0, BLK_PERM_ALL, 1690 &error_abort); 1691 1692 /* In commit_active_start() all intermediate nodes disappear, so 1693 * any jobs in them must be blocked */ 1694 if (target_is_backing) { 1695 BlockDriverState *iter; 1696 for (iter = backing_bs(bs); iter != target; iter = backing_bs(iter)) { 1697 /* XXX BLK_PERM_WRITE needs to be allowed so we don't block 1698 * ourselves at s->base (if writes are blocked for a node, they are 1699 * also blocked for its backing file). The other options would be a 1700 * second filter driver above s->base (== target). */ 1701 ret = block_job_add_bdrv(&s->common, "intermediate node", iter, 0, 1702 BLK_PERM_WRITE_UNCHANGED | BLK_PERM_WRITE, 1703 errp); 1704 if (ret < 0) { 1705 goto fail; 1706 } 1707 } 1708 1709 if (bdrv_freeze_backing_chain(mirror_top_bs, target, errp) < 0) { 1710 goto fail; 1711 } 1712 } 1713 1714 QTAILQ_INIT(&s->ops_in_flight); 1715 1716 trace_mirror_start(bs, s, opaque); 1717 job_start(&s->common.job); 1718 1719 return &s->common; 1720 1721 fail: 1722 if (s) { 1723 /* Make sure this BDS does not go away until we have completed the graph 1724 * changes below */ 1725 bdrv_ref(mirror_top_bs); 1726 1727 g_free(s->replaces); 1728 blk_unref(s->target); 1729 bs_opaque->job = NULL; 1730 if (s->dirty_bitmap) { 1731 bdrv_release_dirty_bitmap(s->dirty_bitmap); 1732 } 1733 job_early_fail(&s->common.job); 1734 } 1735 1736 bs_opaque->stop = true; 1737 bdrv_child_refresh_perms(mirror_top_bs, mirror_top_bs->backing, 1738 &error_abort); 1739 bdrv_replace_node(mirror_top_bs, backing_bs(mirror_top_bs), &error_abort); 1740 1741 bdrv_unref(mirror_top_bs); 1742 1743 return NULL; 1744 } 1745 1746 void mirror_start(const char *job_id, BlockDriverState *bs, 1747 BlockDriverState *target, const char *replaces, 1748 int creation_flags, int64_t speed, 1749 uint32_t granularity, int64_t buf_size, 1750 MirrorSyncMode mode, BlockMirrorBackingMode backing_mode, 1751 bool zero_target, 1752 BlockdevOnError on_source_error, 1753 BlockdevOnError on_target_error, 1754 bool unmap, const char *filter_node_name, 1755 MirrorCopyMode copy_mode, Error **errp) 1756 { 1757 bool is_none_mode; 1758 BlockDriverState *base; 1759 1760 if ((mode == MIRROR_SYNC_MODE_INCREMENTAL) || 1761 (mode == MIRROR_SYNC_MODE_BITMAP)) { 1762 error_setg(errp, "Sync mode '%s' not supported", 1763 MirrorSyncMode_str(mode)); 1764 return; 1765 } 1766 is_none_mode = mode == MIRROR_SYNC_MODE_NONE; 1767 base = mode == MIRROR_SYNC_MODE_TOP ? backing_bs(bs) : NULL; 1768 mirror_start_job(job_id, bs, creation_flags, target, replaces, 1769 speed, granularity, buf_size, backing_mode, zero_target, 1770 on_source_error, on_target_error, unmap, NULL, NULL, 1771 &mirror_job_driver, is_none_mode, base, false, 1772 filter_node_name, true, copy_mode, errp); 1773 } 1774 1775 BlockJob *commit_active_start(const char *job_id, BlockDriverState *bs, 1776 BlockDriverState *base, int creation_flags, 1777 int64_t speed, BlockdevOnError on_error, 1778 const char *filter_node_name, 1779 BlockCompletionFunc *cb, void *opaque, 1780 bool auto_complete, Error **errp) 1781 { 1782 bool base_read_only; 1783 Error *local_err = NULL; 1784 BlockJob *ret; 1785 1786 base_read_only = bdrv_is_read_only(base); 1787 1788 if (base_read_only) { 1789 if (bdrv_reopen_set_read_only(base, false, errp) < 0) { 1790 return NULL; 1791 } 1792 } 1793 1794 ret = mirror_start_job( 1795 job_id, bs, creation_flags, base, NULL, speed, 0, 0, 1796 MIRROR_LEAVE_BACKING_CHAIN, false, 1797 on_error, on_error, true, cb, opaque, 1798 &commit_active_job_driver, false, base, auto_complete, 1799 filter_node_name, false, MIRROR_COPY_MODE_BACKGROUND, 1800 &local_err); 1801 if (local_err) { 1802 error_propagate(errp, local_err); 1803 goto error_restore_flags; 1804 } 1805 1806 return ret; 1807 1808 error_restore_flags: 1809 /* ignore error and errp for bdrv_reopen, because we want to propagate 1810 * the original error */ 1811 if (base_read_only) { 1812 bdrv_reopen_set_read_only(base, true, NULL); 1813 } 1814 return NULL; 1815 } 1816