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