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