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