block/block-copy.c

   1 /*
   2  * block_copy API
   3  *
   4  * Copyright (C) 2013 Proxmox Server Solutions
   5  * Copyright (c) 2019 Virtuozzo International GmbH.
   6  *
   7  * Authors:
   8  *  Dietmar Maurer (dietmar@proxmox.com)
   9  *  Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
  10  *
  11  * This work is licensed under the terms of the GNU GPL, version 2 or later.
  12  * See the COPYING file in the top-level directory.
  13  */
  14
  15 #include "qemu/osdep.h"
  16
  17 #include "trace.h"
  18 #include "qapi/error.h"
  19 #include "block/block-copy.h"
  20 #include "sysemu/block-backend.h"
  21 #include "qemu/units.h"
  22 #include "qemu/coroutine.h"
  23 #include "block/aio_task.h"
  24
  25 #define BLOCK_COPY_MAX_COPY_RANGE (16 * MiB)
  26 #define BLOCK_COPY_MAX_BUFFER (1 * MiB)
  27 #define BLOCK_COPY_MAX_MEM (128 * MiB)
  28 #define BLOCK_COPY_MAX_WORKERS 64
  29 #define BLOCK_COPY_SLICE_TIME 100000000ULL /* ns */
  30
  31 static coroutine_fn int block_copy_task_entry(AioTask *task);
  32
  33 typedef struct BlockCopyCallState {
  34     /* IN parameters. Initialized in block_copy_async() and never changed. */
  35     BlockCopyState *s;
  36     int64_t offset;
  37     int64_t bytes;
  38     int max_workers;
  39     int64_t max_chunk;
  40     bool ignore_ratelimit;
  41     BlockCopyAsyncCallbackFunc cb;
  42     void *cb_opaque;
  43
  44     /* Coroutine where async block-copy is running */
  45     Coroutine *co;
  46
  47     /* To reference all call states from BlockCopyState */
  48     QLIST_ENTRY(BlockCopyCallState) list;
  49
  50     /* State */
  51     int ret;
  52     bool finished;
  53     QemuCoSleepState *sleep_state;
  54     bool cancelled;
  55
  56     /* OUT parameters */
  57     bool error_is_read;
  58 } BlockCopyCallState;
  59
  60 typedef struct BlockCopyTask {
  61     AioTask task;
  62
  63     BlockCopyState *s;
  64     BlockCopyCallState *call_state;
  65     int64_t offset;
  66     int64_t bytes;
  67     bool zeroes;
  68     QLIST_ENTRY(BlockCopyTask) list;
  69     CoQueue wait_queue; /* coroutines blocked on this task */
  70 } BlockCopyTask;
  71
  72 static int64_t task_end(BlockCopyTask *task)
  73 {
  74     return task->offset + task->bytes;
  75 }
  76
  77 typedef struct BlockCopyState {
  78     /*
  79      * BdrvChild objects are not owned or managed by block-copy. They are
  80      * provided by block-copy user and user is responsible for appropriate
  81      * permissions on these children.
  82      */
  83     BdrvChild *source;
  84     BdrvChild *target;
  85     BdrvDirtyBitmap *copy_bitmap;
  86     int64_t in_flight_bytes;
  87     int64_t cluster_size;
  88     bool use_copy_range;
  89     int64_t copy_size;
  90     uint64_t len;
  91     QLIST_HEAD(, BlockCopyTask) tasks; /* All tasks from all block-copy calls */
  92     QLIST_HEAD(, BlockCopyCallState) calls;
  93
  94     BdrvRequestFlags write_flags;
  95
  96     /*
  97      * skip_unallocated:
  98      *
  99      * Used by sync=top jobs, which first scan the source node for unallocated
 100      * areas and clear them in the copy_bitmap.  During this process, the bitmap
 101      * is thus not fully initialized: It may still have bits set for areas that
 102      * are unallocated and should actually not be copied.
 103      *
 104      * This is indicated by skip_unallocated.
 105      *
 106      * In this case, block_copy() will query the source’s allocation status,
 107      * skip unallocated regions, clear them in the copy_bitmap, and invoke
 108      * block_copy_reset_unallocated() every time it does.
 109      */
 110     bool skip_unallocated;
 111
 112     ProgressMeter *progress;
 113     /* progress_bytes_callback: called when some copying progress is done. */
 114     ProgressBytesCallbackFunc progress_bytes_callback;
 115     void *progress_opaque;
 116
 117     SharedResource *mem;
 118
 119     uint64_t speed;
 120     RateLimit rate_limit;
 121 } BlockCopyState;
 122
 123 static BlockCopyTask *find_conflicting_task(BlockCopyState *s,
 124                                             int64_t offset, int64_t bytes)
 125 {
 126     BlockCopyTask *t;
 127
 128     QLIST_FOREACH(t, &s->tasks, list) {
 129         if (offset + bytes > t->offset && offset < t->offset + t->bytes) {
 130             return t;
 131         }
 132     }
 133
 134     return NULL;
 135 }
 136
 137 /*
 138  * If there are no intersecting tasks return false. Otherwise, wait for the
 139  * first found intersecting tasks to finish and return true.
 140  */
 141 static bool coroutine_fn block_copy_wait_one(BlockCopyState *s, int64_t offset,
 142                                              int64_t bytes)
 143 {
 144     BlockCopyTask *task = find_conflicting_task(s, offset, bytes);
 145
 146     if (!task) {
 147         return false;
 148     }
 149
 150     qemu_co_queue_wait(&task->wait_queue, NULL);
 151
 152     return true;
 153 }
 154
 155 /*
 156  * Search for the first dirty area in offset/bytes range and create task at
 157  * the beginning of it.
 158  */
 159 static BlockCopyTask *block_copy_task_create(BlockCopyState *s,
 160                                              BlockCopyCallState *call_state,
 161                                              int64_t offset, int64_t bytes)
 162 {
 163     BlockCopyTask *task;
 164     int64_t max_chunk = MIN_NON_ZERO(s->copy_size, call_state->max_chunk);
 165
 166     if (!bdrv_dirty_bitmap_next_dirty_area(s->copy_bitmap,
 167                                            offset, offset + bytes,
 168                                            max_chunk, &offset, &bytes))
 169     {
 170         return NULL;
 171     }
 172
 173     assert(QEMU_IS_ALIGNED(offset, s->cluster_size));
 174     bytes = QEMU_ALIGN_UP(bytes, s->cluster_size);
 175
 176     /* region is dirty, so no existent tasks possible in it */
 177     assert(!find_conflicting_task(s, offset, bytes));
 178
 179     bdrv_reset_dirty_bitmap(s->copy_bitmap, offset, bytes);
 180     s->in_flight_bytes += bytes;
 181
 182     task = g_new(BlockCopyTask, 1);
 183     *task = (BlockCopyTask) {
 184         .task.func = block_copy_task_entry,
 185         .s = s,
 186         .call_state = call_state,
 187         .offset = offset,
 188         .bytes = bytes,
 189     };
 190     qemu_co_queue_init(&task->wait_queue);
 191     QLIST_INSERT_HEAD(&s->tasks, task, list);
 192
 193     return task;
 194 }
 195
 196 /*
 197  * block_copy_task_shrink
 198  *
 199  * Drop the tail of the task to be handled later. Set dirty bits back and
 200  * wake up all tasks waiting for us (may be some of them are not intersecting
 201  * with shrunk task)
 202  */
 203 static void coroutine_fn block_copy_task_shrink(BlockCopyTask *task,
 204                                                 int64_t new_bytes)
 205 {
 206     if (new_bytes == task->bytes) {
 207         return;
 208     }
 209
 210     assert(new_bytes > 0 && new_bytes < task->bytes);
 211
 212     task->s->in_flight_bytes -= task->bytes - new_bytes;
 213     bdrv_set_dirty_bitmap(task->s->copy_bitmap,
 214                           task->offset + new_bytes, task->bytes - new_bytes);
 215
 216     task->bytes = new_bytes;
 217     qemu_co_queue_restart_all(&task->wait_queue);
 218 }
 219
 220 static void coroutine_fn block_copy_task_end(BlockCopyTask *task, int ret)
 221 {
 222     task->s->in_flight_bytes -= task->bytes;
 223     if (ret < 0) {
 224         bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->offset, task->bytes);
 225     }
 226     QLIST_REMOVE(task, list);
 227     qemu_co_queue_restart_all(&task->wait_queue);
 228 }
 229
 230 void block_copy_state_free(BlockCopyState *s)
 231 {
 232     if (!s) {
 233         return;
 234     }
 235
 236     bdrv_release_dirty_bitmap(s->copy_bitmap);
 237     shres_destroy(s->mem);
 238     g_free(s);
 239 }
 240
 241 static uint32_t block_copy_max_transfer(BdrvChild *source, BdrvChild *target)
 242 {
 243     return MIN_NON_ZERO(INT_MAX,
 244                         MIN_NON_ZERO(source->bs->bl.max_transfer,
 245                                      target->bs->bl.max_transfer));
 246 }
 247
 248 BlockCopyState *block_copy_state_new(BdrvChild *source, BdrvChild *target,
 249                                      int64_t cluster_size, bool use_copy_range,
 250                                      BdrvRequestFlags write_flags, Error **errp)
 251 {
 252     BlockCopyState *s;
 253     BdrvDirtyBitmap *copy_bitmap;
 254
 255     copy_bitmap = bdrv_create_dirty_bitmap(source->bs, cluster_size, NULL,
 256                                            errp);
 257     if (!copy_bitmap) {
 258         return NULL;
 259     }
 260     bdrv_disable_dirty_bitmap(copy_bitmap);
 261
 262     s = g_new(BlockCopyState, 1);
 263     *s = (BlockCopyState) {
 264         .source = source,
 265         .target = target,
 266         .copy_bitmap = copy_bitmap,
 267         .cluster_size = cluster_size,
 268         .len = bdrv_dirty_bitmap_size(copy_bitmap),
 269         .write_flags = write_flags,
 270         .mem = shres_create(BLOCK_COPY_MAX_MEM),
 271     };
 272
 273     if (block_copy_max_transfer(source, target) < cluster_size) {
 274         /*
 275          * copy_range does not respect max_transfer. We don't want to bother
 276          * with requests smaller than block-copy cluster size, so fallback to
 277          * buffered copying (read and write respect max_transfer on their
 278          * behalf).
 279          */
 280         s->use_copy_range = false;
 281         s->copy_size = cluster_size;
 282     } else if (write_flags & BDRV_REQ_WRITE_COMPRESSED) {
 283         /* Compression supports only cluster-size writes and no copy-range. */
 284         s->use_copy_range = false;
 285         s->copy_size = cluster_size;
 286     } else {
 287         /*
 288          * We enable copy-range, but keep small copy_size, until first
 289          * successful copy_range (look at block_copy_do_copy).
 290          */
 291         s->use_copy_range = use_copy_range;
 292         s->copy_size = MAX(s->cluster_size, BLOCK_COPY_MAX_BUFFER);
 293     }
 294
 295     QLIST_INIT(&s->tasks);
 296     QLIST_INIT(&s->calls);
 297
 298     return s;
 299 }
 300
 301 void block_copy_set_progress_callback(
 302         BlockCopyState *s,
 303         ProgressBytesCallbackFunc progress_bytes_callback,
 304         void *progress_opaque)
 305 {
 306     s->progress_bytes_callback = progress_bytes_callback;
 307     s->progress_opaque = progress_opaque;
 308 }
 309
 310 void block_copy_set_progress_meter(BlockCopyState *s, ProgressMeter *pm)
 311 {
 312     s->progress = pm;
 313 }
 314
 315 /*
 316  * Takes ownership of @task
 317  *
 318  * If pool is NULL directly run the task, otherwise schedule it into the pool.
 319  *
 320  * Returns: task.func return code if pool is NULL
 321  *          otherwise -ECANCELED if pool status is bad
 322  *          otherwise 0 (successfully scheduled)
 323  */
 324 static coroutine_fn int block_copy_task_run(AioTaskPool *pool,
 325                                             BlockCopyTask *task)
 326 {
 327     if (!pool) {
 328         int ret = task->task.func(&task->task);
 329
 330         g_free(task);
 331         return ret;
 332     }
 333
 334     aio_task_pool_wait_slot(pool);
 335     if (aio_task_pool_status(pool) < 0) {
 336         co_put_to_shres(task->s->mem, task->bytes);
 337         block_copy_task_end(task, -ECANCELED);
 338         g_free(task);
 339         return -ECANCELED;
 340     }
 341
 342     aio_task_pool_start_task(pool, &task->task);
 343
 344     return 0;
 345 }
 346
 347 /*
 348  * block_copy_do_copy
 349  *
 350  * Do copy of cluster-aligned chunk. Requested region is allowed to exceed
 351  * s->len only to cover last cluster when s->len is not aligned to clusters.
 352  *
 353  * No sync here: nor bitmap neighter intersecting requests handling, only copy.
 354  *
 355  * Returns 0 on success.
 356  */
 357 static int coroutine_fn block_copy_do_copy(BlockCopyState *s,
 358                                            int64_t offset, int64_t bytes,
 359                                            bool zeroes, bool *error_is_read)
 360 {
 361     int ret;
 362     int64_t nbytes = MIN(offset + bytes, s->len) - offset;
 363     void *bounce_buffer = NULL;
 364
 365     assert(offset >= 0 && bytes > 0 && INT64_MAX - offset >= bytes);
 366     assert(QEMU_IS_ALIGNED(offset, s->cluster_size));
 367     assert(QEMU_IS_ALIGNED(bytes, s->cluster_size));
 368     assert(offset < s->len);
 369     assert(offset + bytes <= s->len ||
 370            offset + bytes == QEMU_ALIGN_UP(s->len, s->cluster_size));
 371     assert(nbytes < INT_MAX);
 372
 373     if (zeroes) {
 374         ret = bdrv_co_pwrite_zeroes(s->target, offset, nbytes, s->write_flags &
 375                                     ~BDRV_REQ_WRITE_COMPRESSED);
 376         if (ret < 0) {
 377             trace_block_copy_write_zeroes_fail(s, offset, ret);
 378             *error_is_read = false;
 379         }
 380         return ret;
 381     }
 382
 383     if (s->use_copy_range) {
 384         ret = bdrv_co_copy_range(s->source, offset, s->target, offset, nbytes,
 385                                  0, s->write_flags);
 386         if (ret < 0) {
 387             trace_block_copy_copy_range_fail(s, offset, ret);
 388             s->use_copy_range = false;
 389             s->copy_size = MAX(s->cluster_size, BLOCK_COPY_MAX_BUFFER);
 390             /* Fallback to read+write with allocated buffer */
 391         } else {
 392             if (s->use_copy_range) {
 393                 /*
 394                  * Successful copy-range. Now increase copy_size.  copy_range
 395                  * does not respect max_transfer (it's a TODO), so we factor
 396                  * that in here.
 397                  *
 398                  * Note: we double-check s->use_copy_range for the case when
 399                  * parallel block-copy request unsets it during previous
 400                  * bdrv_co_copy_range call.
 401                  */
 402                 s->copy_size =
 403                         MIN(MAX(s->cluster_size, BLOCK_COPY_MAX_COPY_RANGE),
 404                             QEMU_ALIGN_DOWN(block_copy_max_transfer(s->source,
 405                                                                     s->target),
 406                                             s->cluster_size));
 407             }
 408             goto out;
 409         }
 410     }
 411
 412     /*
 413      * In case of failed copy_range request above, we may proceed with buffered
 414      * request larger than BLOCK_COPY_MAX_BUFFER. Still, further requests will
 415      * be properly limited, so don't care too much. Moreover the most likely
 416      * case (copy_range is unsupported for the configuration, so the very first
 417      * copy_range request fails) is handled by setting large copy_size only
 418      * after first successful copy_range.
 419      */
 420
 421     bounce_buffer = qemu_blockalign(s->source->bs, nbytes);
 422
 423     ret = bdrv_co_pread(s->source, offset, nbytes, bounce_buffer, 0);
 424     if (ret < 0) {
 425         trace_block_copy_read_fail(s, offset, ret);
 426         *error_is_read = true;
 427         goto out;
 428     }
 429
 430     ret = bdrv_co_pwrite(s->target, offset, nbytes, bounce_buffer,
 431                          s->write_flags);
 432     if (ret < 0) {
 433         trace_block_copy_write_fail(s, offset, ret);
 434         *error_is_read = false;
 435         goto out;
 436     }
 437
 438 out:
 439     qemu_vfree(bounce_buffer);
 440
 441     return ret;
 442 }
 443
 444 static coroutine_fn int block_copy_task_entry(AioTask *task)
 445 {
 446     BlockCopyTask *t = container_of(task, BlockCopyTask, task);
 447     bool error_is_read = false;
 448     int ret;
 449
 450     ret = block_copy_do_copy(t->s, t->offset, t->bytes, t->zeroes,
 451                              &error_is_read);
 452     if (ret < 0 && !t->call_state->ret) {
 453         t->call_state->ret = ret;
 454         t->call_state->error_is_read = error_is_read;
 455     } else {
 456         progress_work_done(t->s->progress, t->bytes);
 457         t->s->progress_bytes_callback(t->bytes, t->s->progress_opaque);
 458     }
 459     co_put_to_shres(t->s->mem, t->bytes);
 460     block_copy_task_end(t, ret);
 461
 462     return ret;
 463 }
 464
 465 static int block_copy_block_status(BlockCopyState *s, int64_t offset,
 466                                    int64_t bytes, int64_t *pnum)
 467 {
 468     int64_t num;
 469     BlockDriverState *base;
 470     int ret;
 471
 472     if (s->skip_unallocated) {
 473         base = bdrv_backing_chain_next(s->source->bs);
 474     } else {
 475         base = NULL;
 476     }
 477
 478     ret = bdrv_block_status_above(s->source->bs, base, offset, bytes, &num,
 479                                   NULL, NULL);
 480     if (ret < 0 || num < s->cluster_size) {
 481         /*
 482          * On error or if failed to obtain large enough chunk just fallback to
 483          * copy one cluster.
 484          */
 485         num = s->cluster_size;
 486         ret = BDRV_BLOCK_ALLOCATED | BDRV_BLOCK_DATA;
 487     } else if (offset + num == s->len) {
 488         num = QEMU_ALIGN_UP(num, s->cluster_size);
 489     } else {
 490         num = QEMU_ALIGN_DOWN(num, s->cluster_size);
 491     }
 492
 493     *pnum = num;
 494     return ret;
 495 }
 496
 497 /*
 498  * Check if the cluster starting at offset is allocated or not.
 499  * return via pnum the number of contiguous clusters sharing this allocation.
 500  */
 501 static int block_copy_is_cluster_allocated(BlockCopyState *s, int64_t offset,
 502                                            int64_t *pnum)
 503 {
 504     BlockDriverState *bs = s->source->bs;
 505     int64_t count, total_count = 0;
 506     int64_t bytes = s->len - offset;
 507     int ret;
 508
 509     assert(QEMU_IS_ALIGNED(offset, s->cluster_size));
 510
 511     while (true) {
 512         ret = bdrv_is_allocated(bs, offset, bytes, &count);
 513         if (ret < 0) {
 514             return ret;
 515         }
 516
 517         total_count += count;
 518
 519         if (ret || count == 0) {
 520             /*
 521              * ret: partial segment(s) are considered allocated.
 522              * otherwise: unallocated tail is treated as an entire segment.
 523              */
 524             *pnum = DIV_ROUND_UP(total_count, s->cluster_size);
 525             return ret;
 526         }
 527
 528         /* Unallocated segment(s) with uncertain following segment(s) */
 529         if (total_count >= s->cluster_size) {
 530             *pnum = total_count / s->cluster_size;
 531             return 0;
 532         }
 533
 534         offset += count;
 535         bytes -= count;
 536     }
 537 }
 538
 539 /*
 540  * Reset bits in copy_bitmap starting at offset if they represent unallocated
 541  * data in the image. May reset subsequent contiguous bits.
 542  * @return 0 when the cluster at @offset was unallocated,
 543  *         1 otherwise, and -ret on error.
 544  */
 545 int64_t block_copy_reset_unallocated(BlockCopyState *s,
 546                                      int64_t offset, int64_t *count)
 547 {
 548     int ret;
 549     int64_t clusters, bytes;
 550
 551     ret = block_copy_is_cluster_allocated(s, offset, &clusters);
 552     if (ret < 0) {
 553         return ret;
 554     }
 555
 556     bytes = clusters * s->cluster_size;
 557
 558     if (!ret) {
 559         bdrv_reset_dirty_bitmap(s->copy_bitmap, offset, bytes);
 560         progress_set_remaining(s->progress,
 561                                bdrv_get_dirty_count(s->copy_bitmap) +
 562                                s->in_flight_bytes);
 563     }
 564
 565     *count = bytes;
 566     return ret;
 567 }
 568
 569 /*
 570  * block_copy_dirty_clusters
 571  *
 572  * Copy dirty clusters in @offset/@bytes range.
 573  * Returns 1 if dirty clusters found and successfully copied, 0 if no dirty
 574  * clusters found and -errno on failure.
 575  */
 576 static int coroutine_fn
 577 block_copy_dirty_clusters(BlockCopyCallState *call_state)
 578 {
 579     BlockCopyState *s = call_state->s;
 580     int64_t offset = call_state->offset;
 581     int64_t bytes = call_state->bytes;
 582
 583     int ret = 0;
 584     bool found_dirty = false;
 585     int64_t end = offset + bytes;
 586     AioTaskPool *aio = NULL;
 587
 588     /*
 589      * block_copy() user is responsible for keeping source and target in same
 590      * aio context
 591      */
 592     assert(bdrv_get_aio_context(s->source->bs) ==
 593            bdrv_get_aio_context(s->target->bs));
 594
 595     assert(QEMU_IS_ALIGNED(offset, s->cluster_size));
 596     assert(QEMU_IS_ALIGNED(bytes, s->cluster_size));
 597
 598     while (bytes && aio_task_pool_status(aio) == 0 && !call_state->cancelled) {
 599         BlockCopyTask *task;
 600         int64_t status_bytes;
 601
 602         task = block_copy_task_create(s, call_state, offset, bytes);
 603         if (!task) {
 604             /* No more dirty bits in the bitmap */
 605             trace_block_copy_skip_range(s, offset, bytes);
 606             break;
 607         }
 608         if (task->offset > offset) {
 609             trace_block_copy_skip_range(s, offset, task->offset - offset);
 610         }
 611
 612         found_dirty = true;
 613
 614         ret = block_copy_block_status(s, task->offset, task->bytes,
 615                                       &status_bytes);
 616         assert(ret >= 0); /* never fail */
 617         if (status_bytes < task->bytes) {
 618             block_copy_task_shrink(task, status_bytes);
 619         }
 620         if (s->skip_unallocated && !(ret & BDRV_BLOCK_ALLOCATED)) {
 621             block_copy_task_end(task, 0);
 622             progress_set_remaining(s->progress,
 623                                    bdrv_get_dirty_count(s->copy_bitmap) +
 624                                    s->in_flight_bytes);
 625             trace_block_copy_skip_range(s, task->offset, task->bytes);
 626             offset = task_end(task);
 627             bytes = end - offset;
 628             g_free(task);
 629             continue;
 630         }
 631         task->zeroes = ret & BDRV_BLOCK_ZERO;
 632
 633         if (s->speed) {
 634             if (!call_state->ignore_ratelimit) {
 635                 uint64_t ns = ratelimit_calculate_delay(&s->rate_limit, 0);
 636                 if (ns > 0) {
 637                     block_copy_task_end(task, -EAGAIN);
 638                     g_free(task);
 639                     qemu_co_sleep_ns_wakeable(QEMU_CLOCK_REALTIME, ns,
 640                                               &call_state->sleep_state);
 641                     continue;
 642                 }
 643             }
 644
 645             ratelimit_calculate_delay(&s->rate_limit, task->bytes);
 646         }
 647
 648         trace_block_copy_process(s, task->offset);
 649
 650         co_get_from_shres(s->mem, task->bytes);
 651
 652         offset = task_end(task);
 653         bytes = end - offset;
 654
 655         if (!aio && bytes) {
 656             aio = aio_task_pool_new(call_state->max_workers);
 657         }
 658
 659         ret = block_copy_task_run(aio, task);
 660         if (ret < 0) {
 661             goto out;
 662         }
 663     }
 664
 665 out:
 666     if (aio) {
 667         aio_task_pool_wait_all(aio);
 668
 669         /*
 670          * We are not really interested in -ECANCELED returned from
 671          * block_copy_task_run. If it fails, it means some task already failed
 672          * for real reason, let's return first failure.
 673          * Still, assert that we don't rewrite failure by success.
 674          *
 675          * Note: ret may be positive here because of block-status result.
 676          */
 677         assert(ret >= 0 || aio_task_pool_status(aio) < 0);
 678         ret = aio_task_pool_status(aio);
 679
 680         aio_task_pool_free(aio);
 681     }
 682
 683     return ret < 0 ? ret : found_dirty;
 684 }
 685
 686 void block_copy_kick(BlockCopyCallState *call_state)
 687 {
 688     if (call_state->sleep_state) {
 689         qemu_co_sleep_wake(call_state->sleep_state);
 690     }
 691 }
 692
 693 /*
 694  * block_copy_common
 695  *
 696  * Copy requested region, accordingly to dirty bitmap.
 697  * Collaborate with parallel block_copy requests: if they succeed it will help
 698  * us. If they fail, we will retry not-copied regions. So, if we return error,
 699  * it means that some I/O operation failed in context of _this_ block_copy call,
 700  * not some parallel operation.
 701  */
 702 static int coroutine_fn block_copy_common(BlockCopyCallState *call_state)
 703 {
 704     int ret;
 705
 706     QLIST_INSERT_HEAD(&call_state->s->calls, call_state, list);
 707
 708     do {
 709         ret = block_copy_dirty_clusters(call_state);
 710
 711         if (ret == 0 && !call_state->cancelled) {
 712             ret = block_copy_wait_one(call_state->s, call_state->offset,
 713                                       call_state->bytes);
 714         }
 715
 716         /*
 717          * We retry in two cases:
 718          * 1. Some progress done
 719          *    Something was copied, which means that there were yield points
 720          *    and some new dirty bits may have appeared (due to failed parallel
 721          *    block-copy requests).
 722          * 2. We have waited for some intersecting block-copy request
 723          *    It may have failed and produced new dirty bits.
 724          */
 725     } while (ret > 0 && !call_state->cancelled);
 726
 727     call_state->finished = true;
 728
 729     if (call_state->cb) {
 730         call_state->cb(call_state->cb_opaque);
 731     }
 732
 733     QLIST_REMOVE(call_state, list);
 734
 735     return ret;
 736 }
 737
 738 int coroutine_fn block_copy(BlockCopyState *s, int64_t start, int64_t bytes,
 739                             bool ignore_ratelimit, bool *error_is_read)
 740 {
 741     BlockCopyCallState call_state = {
 742         .s = s,
 743         .offset = start,
 744         .bytes = bytes,
 745         .ignore_ratelimit = ignore_ratelimit,
 746         .max_workers = BLOCK_COPY_MAX_WORKERS,
 747     };
 748
 749     int ret = block_copy_common(&call_state);
 750
 751     if (error_is_read && ret < 0) {
 752         *error_is_read = call_state.error_is_read;
 753     }
 754
 755     return ret;
 756 }
 757
 758 static void coroutine_fn block_copy_async_co_entry(void *opaque)
 759 {
 760     block_copy_common(opaque);
 761 }
 762
 763 BlockCopyCallState *block_copy_async(BlockCopyState *s,
 764                                      int64_t offset, int64_t bytes,
 765                                      int max_workers, int64_t max_chunk,
 766                                      BlockCopyAsyncCallbackFunc cb,
 767                                      void *cb_opaque)
 768 {
 769     BlockCopyCallState *call_state = g_new(BlockCopyCallState, 1);
 770
 771     *call_state = (BlockCopyCallState) {
 772         .s = s,
 773         .offset = offset,
 774         .bytes = bytes,
 775         .max_workers = max_workers,
 776         .max_chunk = max_chunk,
 777         .cb = cb,
 778         .cb_opaque = cb_opaque,
 779
 780         .co = qemu_coroutine_create(block_copy_async_co_entry, call_state),
 781     };
 782
 783     qemu_coroutine_enter(call_state->co);
 784
 785     return call_state;
 786 }
 787
 788 void block_copy_call_free(BlockCopyCallState *call_state)
 789 {
 790     if (!call_state) {
 791         return;
 792     }
 793
 794     assert(call_state->finished);
 795     g_free(call_state);
 796 }
 797
 798 bool block_copy_call_finished(BlockCopyCallState *call_state)
 799 {
 800     return call_state->finished;
 801 }
 802
 803 bool block_copy_call_succeeded(BlockCopyCallState *call_state)
 804 {
 805     return call_state->finished && !call_state->cancelled &&
 806         call_state->ret == 0;
 807 }
 808
 809 bool block_copy_call_failed(BlockCopyCallState *call_state)
 810 {
 811     return call_state->finished && !call_state->cancelled &&
 812         call_state->ret < 0;
 813 }
 814
 815 bool block_copy_call_cancelled(BlockCopyCallState *call_state)
 816 {
 817     return call_state->cancelled;
 818 }
 819
 820 int block_copy_call_status(BlockCopyCallState *call_state, bool *error_is_read)
 821 {
 822     assert(call_state->finished);
 823     if (error_is_read) {
 824         *error_is_read = call_state->error_is_read;
 825     }
 826     return call_state->ret;
 827 }
 828
 829 void block_copy_call_cancel(BlockCopyCallState *call_state)
 830 {
 831     call_state->cancelled = true;
 832     block_copy_kick(call_state);
 833 }
 834
 835 BdrvDirtyBitmap *block_copy_dirty_bitmap(BlockCopyState *s)
 836 {
 837     return s->copy_bitmap;
 838 }
 839
 840 void block_copy_set_skip_unallocated(BlockCopyState *s, bool skip)
 841 {
 842     s->skip_unallocated = skip;
 843 }
 844
 845 void block_copy_set_speed(BlockCopyState *s, uint64_t speed)
 846 {
 847     s->speed = speed;
 848     if (speed > 0) {
 849         ratelimit_set_speed(&s->rate_limit, speed, BLOCK_COPY_SLICE_TIME);
 850     }
 851
 852     /*
 853      * Note: it's good to kick all call states from here, but it should be done
 854      * only from a coroutine, to not crash if s->calls list changed while
 855      * entering one call. So for now, the only user of this function kicks its
 856      * only one call_state by hand.
 857      */
 858 }