Generic background jobs

index 4de48166b2..b38ed7e265 100644
@@ -702,7 +702,7 @@ void block_job_complete(BlockJob *job, Error **errp)
 
 void block_job_finalize(BlockJob *job, Error **errp)
 {
-    assert(job && job->id && job->txn);
+    assert(job && job->id);
     if (block_job_apply_verb(job, BLOCK_JOB_VERB_FINALIZE, errp)) {
         return;
     }

index fc645dac68..a2cc52233b 100644
@@ -278,6 +278,25 @@ void block_job_finalize(BlockJob *job, Error **errp);
 void block_job_dismiss(BlockJob **job, Error **errp);
 
 /**
+ * block_job_progress_update:
+ * @job: The job that has made progress
+ * @done: How much progress the job made
+ *
+ * Updates the progress counter of the job.
+ */
+void block_job_progress_update(BlockJob *job, uint64_t done);
+
+/**
+ * block_job_progress_set_remaining:
+ * @job: The job whose expected progress end value is set
+ * @remaining: Expected end value of the progress counter of the job
+ *
+ * Sets the expected end value of the progress counter of a job so that a
+ * completion percentage can be calculated when the progress is updated.
+ */
+void block_job_progress_set_remaining(BlockJob *job, uint64_t remaining);
+
+/**
  * block_job_query:
  * @job: The job to get information about.
  *

index 453cd62c24..5d95805472 100644
@@ -39,6 +39,7 @@ typedef struct BackupBlockJob {
     BlockdevOnError on_source_error;
     BlockdevOnError on_target_error;
     CoRwlock flush_rwlock;
+    uint64_t len;
     uint64_t bytes_read;
     int64_t cluster_size;
     bool compress;

@@ -118,7 +119,7 @@ static int coroutine_fn backup_do_cow(BackupBlockJob *job,
 
         trace_backup_do_cow_process(job, start);
 
-        n = MIN(job->cluster_size, job->common.len - start);
+        n = MIN(job->cluster_size, job->len - start);
 
         if (!bounce_buffer) {
             bounce_buffer = blk_blockalign(blk, job->cluster_size);

@@ -159,7 +160,7 @@ static int coroutine_fn backup_do_cow(BackupBlockJob *job,
          * offset field is an opaque progress value, it is not a disk offset.
          */
         job->bytes_read += n;
-        job->common.offset += n;
+        block_job_progress_update(&job->common, n);
     }
 
 out:

@@ -261,7 +262,7 @@ void backup_do_checkpoint(BlockJob *job, Error **errp)
         return;
     }
 
-    len = DIV_ROUND_UP(backup_job->common.len, backup_job->cluster_size);
+    len = DIV_ROUND_UP(backup_job->len, backup_job->cluster_size);
     hbitmap_set(backup_job->copy_bitmap, 0, len);
 }
 

@@ -420,8 +421,9 @@ static void backup_incremental_init_copy_bitmap(BackupBlockJob *job)
         bdrv_set_dirty_iter(dbi, next_cluster * job->cluster_size);
     }
 
-    job->common.offset = job->common.len -
-                         hbitmap_count(job->copy_bitmap) * job->cluster_size;
+    /* TODO block_job_progress_set_remaining() would make more sense */
+    block_job_progress_update(&job->common,
+        job->len - hbitmap_count(job->copy_bitmap) * job->cluster_size);
 
     bdrv_dirty_iter_free(dbi);
 }

@@ -437,7 +439,9 @@ static void coroutine_fn backup_run(void *opaque)
     QLIST_INIT(&job->inflight_reqs);
     qemu_co_rwlock_init(&job->flush_rwlock);
 
-    nb_clusters = DIV_ROUND_UP(job->common.len, job->cluster_size);
+    nb_clusters = DIV_ROUND_UP(job->len, job->cluster_size);
+    block_job_progress_set_remaining(&job->common, job->len);
+
     job->copy_bitmap = hbitmap_alloc(nb_clusters, 0);
     if (job->sync_mode == MIRROR_SYNC_MODE_INCREMENTAL) {
         backup_incremental_init_copy_bitmap(job);

@@ -461,7 +465,7 @@ static void coroutine_fn backup_run(void *opaque)
         ret = backup_run_incremental(job);
     } else {
         /* Both FULL and TOP SYNC_MODE's require copying.. */
-        for (offset = 0; offset < job->common.len;
+        for (offset = 0; offset < job->len;
              offset += job->cluster_size) {
             bool error_is_read;
             int alloced = 0;

@@ -620,7 +624,7 @@ BlockJob *backup_job_create(const char *job_id, BlockDriverState *bs,
         goto error;
     }
 
-    /* job->common.len is fixed, so we can't allow resize */
+    /* job->len is fixed, so we can't allow resize */
     job = block_job_create(job_id, &backup_job_driver, txn, bs,
                            BLK_PERM_CONSISTENT_READ,
                            BLK_PERM_CONSISTENT_READ | BLK_PERM_WRITE |

@@ -676,7 +680,7 @@ BlockJob *backup_job_create(const char *job_id, BlockDriverState *bs,
     /* Required permissions are already taken with target's blk_new() */
     block_job_add_bdrv(&job->common, "target", target, 0, BLK_PERM_ALL,
                        &error_abort);
-    job->common.len = len;
+    job->len = len;
 
     return &job->common;
 

index 1432baeef4..50b191c980 100644
@@ -146,21 +146,21 @@ static void coroutine_fn commit_run(void *opaque)
     int64_t n = 0; /* bytes */
     void *buf = NULL;
     int bytes_written = 0;
-    int64_t base_len;
+    int64_t len, base_len;
 
-    ret = s->common.len = blk_getlength(s->top);
-
-    if (s->common.len < 0) {
+    ret = len = blk_getlength(s->top);
+    if (len < 0) {
         goto out;
     }
+    block_job_progress_set_remaining(&s->common, len);
 
     ret = base_len = blk_getlength(s->base);
     if (base_len < 0) {
         goto out;
     }
 
-    if (base_len < s->common.len) {
-        ret = blk_truncate(s->base, s->common.len, PREALLOC_MODE_OFF, NULL);
+    if (base_len < len) {
+        ret = blk_truncate(s->base, len, PREALLOC_MODE_OFF, NULL);
         if (ret) {
             goto out;
         }

@@ -168,7 +168,7 @@ static void coroutine_fn commit_run(void *opaque)
 
     buf = blk_blockalign(s->top, COMMIT_BUFFER_SIZE);
 
-    for (offset = 0; offset < s->common.len; offset += n) {
+    for (offset = 0; offset < len; offset += n) {
         bool copy;
 
         /* Note that even when no rate limit is applied we need to yield

@@ -198,7 +198,7 @@ static void coroutine_fn commit_run(void *opaque)
             }
         }
         /* Publish progress */
-        s->common.offset += n;
+        block_job_progress_update(&s->common, n);
 
         if (copy && s->common.speed) {
             delay_ns = ratelimit_calculate_delay(&s->limit, n);

index 99da9c0858..77ee9b1791 100644
@@ -121,7 +121,7 @@ static void mirror_iteration_done(MirrorOp *op, int ret)
             bitmap_set(s->cow_bitmap, chunk_num, nb_chunks);
         }
         if (!s->initial_zeroing_ongoing) {
-            s->common.offset += op->bytes;
+            block_job_progress_update(&s->common, op->bytes);
         }
     }
     qemu_iovec_destroy(&op->qiov);

@@ -792,11 +792,10 @@ static void coroutine_fn mirror_run(void *opaque)
         block_job_pause_point(&s->common);
 
         cnt = bdrv_get_dirty_count(s->dirty_bitmap);
-        /* s->common.offset contains the number of bytes already processed so
-         * far, cnt is the number of dirty bytes remaining and
-         * s->bytes_in_flight is the number of bytes currently being
-         * processed; together those are the current total operation length */
-        s->common.len = s->common.offset + s->bytes_in_flight + cnt;
+        /* cnt is the number of dirty bytes remaining and s->bytes_in_flight is
+         * the number of bytes currently being processed; together those are
+         * the current total operation length */
+        block_job_progress_set_remaining(&s->common, s->bytes_in_flight + cnt);
 
         /* Note that even when no rate limit is applied we need to yield
          * periodically with no pending I/O so that bdrv_drain_all() returns.

index 1a85708fcf..8369852bda 100644
@@ -107,6 +107,7 @@ static void coroutine_fn stream_run(void *opaque)
     BlockBackend *blk = s->common.blk;
     BlockDriverState *bs = blk_bs(blk);
     BlockDriverState *base = s->base;
+    int64_t len;
     int64_t offset = 0;
     uint64_t delay_ns = 0;
     int error = 0;

@@ -118,11 +119,12 @@ static void coroutine_fn stream_run(void *opaque)
         goto out;
     }
 
-    s->common.len = bdrv_getlength(bs);
-    if (s->common.len < 0) {
-        ret = s->common.len;
+    len = bdrv_getlength(bs);
+    if (len < 0) {
+        ret = len;
         goto out;
     }
+    block_job_progress_set_remaining(&s->common, len);
 
     buf = qemu_blockalign(bs, STREAM_BUFFER_SIZE);
 

@@ -135,7 +137,7 @@ static void coroutine_fn stream_run(void *opaque)
         bdrv_enable_copy_on_read(bs);
     }
 
-    for ( ; offset < s->common.len; offset += n) {
+    for ( ; offset < len; offset += n) {
         bool copy;
 
         /* Note that even when no rate limit is applied we need to yield

@@ -159,7 +161,7 @@ static void coroutine_fn stream_run(void *opaque)
 
             /* Finish early if end of backing file has been reached */
             if (ret == 0 && n == 0) {
-                n = s->common.len - offset;
+                n = len - offset;
             }
 
             copy = (ret == 1);

@@ -185,7 +187,7 @@ static void coroutine_fn stream_run(void *opaque)
         ret = 0;
 
         /* Publish progress */
-        s->common.offset += n;
+        block_job_progress_update(&s->common, n);
         if (copy && s->common.speed) {
             delay_ns = ratelimit_calculate_delay(&s->limit, n);
         } else {

index b38ed7e265..bdd34f3a76 100644
@@ -810,6 +810,16 @@ int block_job_complete_sync(BlockJob *job, Error **errp)
     return block_job_finish_sync(job, &block_job_complete, errp);
 }
 
+void block_job_progress_update(BlockJob *job, uint64_t done)
+{
+    job->offset += done;
+}
+
+void block_job_progress_set_remaining(BlockJob *job, uint64_t remaining)
+{
+    job->len = job->offset + remaining;
+}
+
 BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
 {
     BlockJobInfo *info;

index a2cc52233b..22bf418209 100644
@@ -27,6 +27,7 @@
 #define BLOCKJOB_H
 
 #include "block/block.h"
+#include "qemu/ratelimit.h"
 
 typedef struct BlockJobDriver BlockJobDriver;
 typedef struct BlockJobTxn BlockJobTxn;

@@ -118,6 +119,9 @@ typedef struct BlockJob {
     /** Speed that was set with @block_job_set_speed.  */
     int64_t speed;
 
+    /** Rate limiting data structure for implementing @speed. */
+    RateLimit limit;
+
     /** The completion function that will be called when the job completes.  */
     BlockCompletionFunc *cb;
 

index 5d95805472..7585c4391e 100644
@@ -35,7 +35,6 @@ typedef struct BackupBlockJob {
     /* bitmap for sync=incremental */
     BdrvDirtyBitmap *sync_bitmap;
     MirrorSyncMode sync_mode;
-    RateLimit limit;
     BlockdevOnError on_source_error;
     BlockdevOnError on_target_error;
     CoRwlock flush_rwlock;

@@ -199,7 +198,7 @@ static void backup_set_speed(BlockJob *job, int64_t speed, Error **errp)
         error_setg(errp, QERR_INVALID_PARAMETER, "speed");
         return;
     }
-    ratelimit_set_speed(&s->limit, speed, SLICE_TIME);
+    ratelimit_set_speed(&s->common.limit, speed, SLICE_TIME);
 }
 
 static void backup_cleanup_sync_bitmap(BackupBlockJob *job, int ret)

@@ -346,7 +345,7 @@ static bool coroutine_fn yield_and_check(BackupBlockJob *job)
      * (without, VM does not reboot)
      */
     if (job->common.speed) {
-        uint64_t delay_ns = ratelimit_calculate_delay(&job->limit,
+        uint64_t delay_ns = ratelimit_calculate_delay(&job->common.limit,
                                                       job->bytes_read);
         job->bytes_read = 0;
         block_job_sleep_ns(&job->common, delay_ns);

index 50b191c980..beec5d0ad6 100644
@@ -35,7 +35,6 @@ enum {
 
 typedef struct CommitBlockJob {
     BlockJob common;
-    RateLimit limit;
     BlockDriverState *commit_top_bs;
     BlockBackend *top;
     BlockBackend *base;

@@ -201,7 +200,7 @@ static void coroutine_fn commit_run(void *opaque)
         block_job_progress_update(&s->common, n);
 
         if (copy && s->common.speed) {
-            delay_ns = ratelimit_calculate_delay(&s->limit, n);
+            delay_ns = ratelimit_calculate_delay(&s->common.limit, n);
         } else {
             delay_ns = 0;
         }

@@ -225,7 +224,7 @@ static void commit_set_speed(BlockJob *job, int64_t speed, Error **errp)
         error_setg(errp, QERR_INVALID_PARAMETER, "speed");
         return;
     }
-    ratelimit_set_speed(&s->limit, speed, SLICE_TIME);
+    ratelimit_set_speed(&s->common.limit, speed, SLICE_TIME);
 }
 
 static const BlockJobDriver commit_job_driver = {

index 77ee9b1791..95da62544f 100644
@@ -36,7 +36,6 @@ typedef struct MirrorBuffer {
 
 typedef struct MirrorBlockJob {
     BlockJob common;
-    RateLimit limit;
     BlockBackend *target;
     BlockDriverState *mirror_top_bs;
     BlockDriverState *source;

@@ -450,7 +449,8 @@ static uint64_t coroutine_fn mirror_iteration(MirrorBlockJob *s)
         offset += io_bytes;
         nb_chunks -= DIV_ROUND_UP(io_bytes, s->granularity);
         if (s->common.speed) {
-            delay_ns = ratelimit_calculate_delay(&s->limit, io_bytes_acct);
+            delay_ns = ratelimit_calculate_delay(&s->common.limit,
+                                                 io_bytes_acct);
         }
     }
     return delay_ns;

@@ -916,7 +916,7 @@ static void mirror_set_speed(BlockJob *job, int64_t speed, Error **errp)
         error_setg(errp, QERR_INVALID_PARAMETER, "speed");
         return;
     }
-    ratelimit_set_speed(&s->limit, speed, SLICE_TIME);
+    ratelimit_set_speed(&s->common.limit, speed, SLICE_TIME);
 }
 
 static void mirror_complete(BlockJob *job, Error **errp)

index 8369852bda..a1d4768c2e 100644
@@ -33,7 +33,6 @@ enum {
 
 typedef struct StreamBlockJob {
     BlockJob common;
-    RateLimit limit;
     BlockDriverState *base;
     BlockdevOnError on_error;
     char *backing_file_str;

@@ -189,7 +188,7 @@ static void coroutine_fn stream_run(void *opaque)
         /* Publish progress */
         block_job_progress_update(&s->common, n);
         if (copy && s->common.speed) {
-            delay_ns = ratelimit_calculate_delay(&s->limit, n);
+            delay_ns = ratelimit_calculate_delay(&s->common.limit, n);
         } else {
             delay_ns = 0;
         }

@@ -219,7 +218,7 @@ static void stream_set_speed(BlockJob *job, int64_t speed, Error **errp)
         error_setg(errp, QERR_INVALID_PARAMETER, "speed");
         return;
     }
-    ratelimit_set_speed(&s->limit, speed, SLICE_TIME);
+    ratelimit_set_speed(&s->common.limit, speed, SLICE_TIME);
 }
 
 static const BlockJobDriver stream_job_driver = {

index 22bf418209..5aa8a6aaec 100644
@@ -29,6 +29,8 @@
 #include "block/block.h"
 #include "qemu/ratelimit.h"
 
+#define SLICE_TIME 100000000ULL /* ns */
+
 typedef struct BlockJobDriver BlockJobDriver;
 typedef struct BlockJobTxn BlockJobTxn;
 

index 642adce68b..870bd346a8 100644
@@ -41,9 +41,6 @@ struct BlockJobDriver {
     /** String describing the operation, part of query-block-jobs QMP API */
     BlockJobType job_type;
 
-    /** Optional callback for job types that support setting a speed limit */
-    void (*set_speed)(BlockJob *job, int64_t speed, Error **errp);
-
     /** Mandatory: Entrypoint for the Coroutine. */
     CoroutineEntry *start;
 

index 7585c4391e..8468fd9f94 100644
@@ -27,7 +27,6 @@
 #include "qemu/error-report.h"
 
 #define BACKUP_CLUSTER_SIZE_DEFAULT (1 << 16)
-#define SLICE_TIME 100000000ULL /* ns */
 
 typedef struct BackupBlockJob {
     BlockJob common;

@@ -190,17 +189,6 @@ static int coroutine_fn backup_before_write_notify(
     return backup_do_cow(job, req->offset, req->bytes, NULL, true);
 }
 
-static void backup_set_speed(BlockJob *job, int64_t speed, Error **errp)
-{
-    BackupBlockJob *s = container_of(job, BackupBlockJob, common);
-
-    if (speed < 0) {
-        error_setg(errp, QERR_INVALID_PARAMETER, "speed");
-        return;
-    }
-    ratelimit_set_speed(&s->common.limit, speed, SLICE_TIME);
-}
-
 static void backup_cleanup_sync_bitmap(BackupBlockJob *job, int ret)
 {
     BdrvDirtyBitmap *bm;

@@ -540,7 +528,6 @@ static const BlockJobDriver backup_job_driver = {
     .instance_size          = sizeof(BackupBlockJob),
     .job_type               = BLOCK_JOB_TYPE_BACKUP,
     .start                  = backup_run,
-    .set_speed              = backup_set_speed,
     .commit                 = backup_commit,
     .abort                  = backup_abort,
     .clean                  = backup_clean,

index beec5d0ad6..46cbeaec3e 100644
@@ -31,8 +31,6 @@ enum {
     COMMIT_BUFFER_SIZE = 512 * 1024, /* in bytes */
 };
 
-#define SLICE_TIME 100000000ULL /* ns */
-
 typedef struct CommitBlockJob {
     BlockJob common;
     BlockDriverState *commit_top_bs;

@@ -216,21 +214,9 @@ out:
     block_job_defer_to_main_loop(&s->common, commit_complete, data);
 }
 
-static void commit_set_speed(BlockJob *job, int64_t speed, Error **errp)
-{
-    CommitBlockJob *s = container_of(job, CommitBlockJob, common);
-
-    if (speed < 0) {
-        error_setg(errp, QERR_INVALID_PARAMETER, "speed");
-        return;
-    }
-    ratelimit_set_speed(&s->common.limit, speed, SLICE_TIME);
-}
-
 static const BlockJobDriver commit_job_driver = {
     .instance_size = sizeof(CommitBlockJob),
     .job_type      = BLOCK_JOB_TYPE_COMMIT,
-    .set_speed     = commit_set_speed,
     .start         = commit_run,
 };
 

index 95da62544f..21eb04f1d2 100644
@@ -22,7 +22,6 @@
 #include "qemu/ratelimit.h"
 #include "qemu/bitmap.h"
 
-#define SLICE_TIME    100000000ULL /* ns */
 #define MAX_IN_FLIGHT 16
 #define MAX_IO_BYTES (1 << 20) /* 1 Mb */
 #define DEFAULT_MIRROR_BUF_SIZE (MAX_IN_FLIGHT * MAX_IO_BYTES)

@@ -908,17 +907,6 @@ immediate_exit:
     block_job_defer_to_main_loop(&s->common, mirror_exit, data);
 }
 
-static void mirror_set_speed(BlockJob *job, int64_t speed, Error **errp)
-{
-    MirrorBlockJob *s = container_of(job, MirrorBlockJob, common);
-
-    if (speed < 0) {
-        error_setg(errp, QERR_INVALID_PARAMETER, "speed");
-        return;
-    }
-    ratelimit_set_speed(&s->common.limit, speed, SLICE_TIME);
-}
-
 static void mirror_complete(BlockJob *job, Error **errp)
 {
     MirrorBlockJob *s = container_of(job, MirrorBlockJob, common);

@@ -1003,7 +991,6 @@ static void mirror_drain(BlockJob *job)
 static const BlockJobDriver mirror_job_driver = {
     .instance_size          = sizeof(MirrorBlockJob),
     .job_type               = BLOCK_JOB_TYPE_MIRROR,
-    .set_speed              = mirror_set_speed,
     .start                  = mirror_run,
     .complete               = mirror_complete,
     .pause                  = mirror_pause,

@@ -1014,7 +1001,6 @@ static const BlockJobDriver mirror_job_driver = {
 static const BlockJobDriver commit_active_job_driver = {
     .instance_size          = sizeof(MirrorBlockJob),
     .job_type               = BLOCK_JOB_TYPE_COMMIT,
-    .set_speed              = mirror_set_speed,
     .start                  = mirror_run,
     .complete               = mirror_complete,
     .pause                  = mirror_pause,

index a1d4768c2e..797d7c4f21 100644
@@ -29,8 +29,6 @@ enum {
     STREAM_BUFFER_SIZE = 512 * 1024, /* in bytes */
 };
 
-#define SLICE_TIME 100000000ULL /* ns */
-
 typedef struct StreamBlockJob {
     BlockJob common;
     BlockDriverState *base;

@@ -210,21 +208,9 @@ out:
     block_job_defer_to_main_loop(&s->common, stream_complete, data);
 }
 
-static void stream_set_speed(BlockJob *job, int64_t speed, Error **errp)
-{
-    StreamBlockJob *s = container_of(job, StreamBlockJob, common);
-
-    if (speed < 0) {
-        error_setg(errp, QERR_INVALID_PARAMETER, "speed");
-        return;
-    }
-    ratelimit_set_speed(&s->common.limit, speed, SLICE_TIME);
-}
-
 static const BlockJobDriver stream_job_driver = {
     .instance_size = sizeof(StreamBlockJob),
     .job_type      = BLOCK_JOB_TYPE_STREAM,
-    .set_speed     = stream_set_speed,
     .start         = stream_run,
 };
 

index bdd34f3a76..5d050e65d3 100644
@@ -659,22 +659,18 @@ static bool block_job_timer_pending(BlockJob *job)
 
 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
 {
-    Error *local_err = NULL;
     int64_t old_speed = job->speed;
 
-    if (!job->driver->set_speed) {
-        error_setg(errp, QERR_UNSUPPORTED);
-        return;
-    }
     if (block_job_apply_verb(job, BLOCK_JOB_VERB_SET_SPEED, errp)) {
         return;
     }
-    job->driver->set_speed(job, speed, &local_err);
-    if (local_err) {
-        error_propagate(errp, local_err);
+    if (speed < 0) {
+        error_setg(errp, QERR_INVALID_PARAMETER, "speed");
         return;
     }
 
+    ratelimit_set_speed(&job->limit, speed, SLICE_TIME);
+
     job->speed = speed;
     if (speed && speed <= old_speed) {
         return;

index 870bd346a8..4d0439c51e 100644
@@ -166,6 +166,14 @@ void block_job_sleep_ns(BlockJob *job, int64_t ns);
 void block_job_yield(BlockJob *job);
 
 /**
+ * block_job_ratelimit_get_delay:
+ *
+ * Calculate and return delay for the next request in ns. See the documentation
+ * of ratelimit_calculate_delay() for details.
+ */
+int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n);
+
+/**
  * block_job_pause_all:
  *
  * Asynchronously pause all jobs.

index 8468fd9f94..cfdb6ecdf5 100644
@@ -325,21 +325,17 @@ static void backup_complete(BlockJob *job, void *opaque)
 
 static bool coroutine_fn yield_and_check(BackupBlockJob *job)
 {
+    uint64_t delay_ns;
+
     if (block_job_is_cancelled(&job->common)) {
         return true;
     }
 
-    /* we need to yield so that bdrv_drain_all() returns.
-     * (without, VM does not reboot)
-     */
-    if (job->common.speed) {
-        uint64_t delay_ns = ratelimit_calculate_delay(&job->common.limit,
-                                                      job->bytes_read);
-        job->bytes_read = 0;
-        block_job_sleep_ns(&job->common, delay_ns);
-    } else {
-        block_job_sleep_ns(&job->common, 0);
-    }
+    /* We need to yield even for delay_ns = 0 so that bdrv_drain_all() can
+     * return. Without a yield, the VM would not reboot. */
+    delay_ns = block_job_ratelimit_get_delay(&job->common, job->bytes_read);
+    job->bytes_read = 0;
+    block_job_sleep_ns(&job->common, delay_ns);
 
     if (block_job_is_cancelled(&job->common)) {
         return true;

index 46cbeaec3e..ba5df6aa0a 100644
@@ -197,8 +197,8 @@ static void coroutine_fn commit_run(void *opaque)
         /* Publish progress */
         block_job_progress_update(&s->common, n);
 
-        if (copy && s->common.speed) {
-            delay_ns = ratelimit_calculate_delay(&s->common.limit, n);
+        if (copy) {
+            delay_ns = block_job_ratelimit_get_delay(&s->common, n);
         } else {
             delay_ns = 0;
         }

index 21eb04f1d2..350f733d0a 100644
@@ -447,10 +447,7 @@ static uint64_t coroutine_fn mirror_iteration(MirrorBlockJob *s)
         assert(io_bytes);
         offset += io_bytes;
         nb_chunks -= DIV_ROUND_UP(io_bytes, s->granularity);
-        if (s->common.speed) {
-            delay_ns = ratelimit_calculate_delay(&s->common.limit,
-                                                 io_bytes_acct);
-        }
+        delay_ns = block_job_ratelimit_get_delay(&s->common, io_bytes_acct);
     }
     return delay_ns;
 }

index 797d7c4f21..df9660d2fc 100644
@@ -185,8 +185,8 @@ static void coroutine_fn stream_run(void *opaque)
 
         /* Publish progress */
         block_job_progress_update(&s->common, n);
-        if (copy && s->common.speed) {
-            delay_ns = ratelimit_calculate_delay(&s->common.limit, n);
+        if (copy) {
+            delay_ns = block_job_ratelimit_get_delay(&s->common, n);
         } else {
             delay_ns = 0;
         }

index 5d050e65d3..b6e27c7f80 100644
@@ -680,6 +680,15 @@ void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
     block_job_enter_cond(job, block_job_timer_pending);
 }
 
+int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
+{
+    if (!job->speed) {
+        return 0;
+    }
+
+    return ratelimit_calculate_delay(&job->limit, n);
+}
+
 void block_job_complete(BlockJob *job, Error **errp)
 {
     /* Should not be reachable via external interface for internal jobs */

index 5aa8a6aaec..0b57d53084 100644
@@ -452,4 +452,11 @@ void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job);
  */
 bool block_job_is_internal(BlockJob *job);
 
+/**
+ * block_job_driver:
+ *
+ * Returns the driver associated with a block job.
+ */
+const BlockJobDriver *block_job_driver(BlockJob *job);
+
 #endif

index cfdb6ecdf5..e14d99560d 100644
@@ -47,6 +47,8 @@ typedef struct BackupBlockJob {
     HBitmap *copy_bitmap;
 } BackupBlockJob;
 
+static const BlockJobDriver backup_job_driver;
+
 /* See if in-flight requests overlap and wait for them to complete */
 static void coroutine_fn wait_for_overlapping_requests(BackupBlockJob *job,
                                                        int64_t start,

@@ -241,7 +243,7 @@ void backup_do_checkpoint(BlockJob *job, Error **errp)
     BackupBlockJob *backup_job = container_of(job, BackupBlockJob, common);
     int64_t len;
 
-    assert(job->driver->job_type == BLOCK_JOB_TYPE_BACKUP);
+    assert(block_job_driver(job) == &backup_job_driver);
 
     if (backup_job->sync_mode != MIRROR_SYNC_MODE_NONE) {
         error_setg(errp, "The backup job only supports block checkpoint in"

@@ -259,7 +261,7 @@ void backup_wait_for_overlapping_requests(BlockJob *job, int64_t offset,
     BackupBlockJob *backup_job = container_of(job, BackupBlockJob, common);
     int64_t start, end;
 
-    assert(job->driver->job_type == BLOCK_JOB_TYPE_BACKUP);
+    assert(block_job_driver(job) == &backup_job_driver);
 
     start = QEMU_ALIGN_DOWN(offset, backup_job->cluster_size);
     end = QEMU_ALIGN_UP(offset + bytes, backup_job->cluster_size);

@@ -272,7 +274,7 @@ void backup_cow_request_begin(CowRequest *req, BlockJob *job,
     BackupBlockJob *backup_job = container_of(job, BackupBlockJob, common);
     int64_t start, end;
 
-    assert(job->driver->job_type == BLOCK_JOB_TYPE_BACKUP);
+    assert(block_job_driver(job) == &backup_job_driver);
 
     start = QEMU_ALIGN_DOWN(offset, backup_job->cluster_size);
     end = QEMU_ALIGN_UP(offset + bytes, backup_job->cluster_size);

index b6e27c7f80..985621e256 100644
@@ -359,6 +359,11 @@ static bool block_job_started(BlockJob *job)
     return job->co;
 }
 
+const BlockJobDriver *block_job_driver(BlockJob *job)
+{
+    return job->driver;
+}
+
 /**
  * All jobs must allow a pause point before entering their job proper. This
  * ensures that jobs can be paused prior to being started, then resumed later.

index 4d0439c51e..62ec964d09 100644
@@ -174,20 +174,6 @@ void block_job_yield(BlockJob *job);
 int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n);
 
 /**
- * block_job_pause_all:
- *
- * Asynchronously pause all jobs.
- */
-void block_job_pause_all(void);
-
-/**
- * block_job_resume_all:
- *
- * Resume all block jobs.  Must be paired with a preceding block_job_pause_all.
- */
-void block_job_resume_all(void);
-
-/**
  * block_job_early_fail:
  * @bs: The block device.
  *

index 985621e256..9943218a34 100644
@@ -1010,19 +1010,6 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
     return job;
 }
 
-void block_job_pause_all(void)
-{
-    BlockJob *job = NULL;
-    while ((job = block_job_next(job))) {
-        AioContext *aio_context = blk_get_aio_context(job->blk);
-
-        aio_context_acquire(aio_context);
-        block_job_ref(job);
-        block_job_pause(job);
-        aio_context_release(aio_context);
-    }
-}
-
 void block_job_early_fail(BlockJob *job)
 {
     assert(job->status == BLOCK_JOB_STATUS_CREATED);

@@ -1100,20 +1087,6 @@ void coroutine_fn block_job_pause_point(BlockJob *job)
     }
 }
 
-void block_job_resume_all(void)
-{
-    BlockJob *job, *next;
-
-    QLIST_FOREACH_SAFE(job, &block_jobs, job_list, next) {
-        AioContext *aio_context = blk_get_aio_context(job->blk);
-
-        aio_context_acquire(aio_context);
-        block_job_resume(job);
-        block_job_unref(job);
-        aio_context_release(aio_context);
-    }
-}
-
 /*
  * Conditionally enter a block_job pending a call to fn() while
  * under the block_job_lock critical section.

index 0b57d53084..8acc1a236a 100644
@@ -26,6 +26,7 @@
 #ifndef BLOCKJOB_H
 #define BLOCKJOB_H
 
+#include "qemu/job.h"
 #include "block/block.h"
 #include "qemu/ratelimit.h"
 

@@ -40,6 +41,9 @@ typedef struct BlockJobTxn BlockJobTxn;
  * Long-running operation on a BlockDriverState.
  */
 typedef struct BlockJob {
+    /** Data belonging to the generic Job infrastructure */
+    Job job;
+
     /** The job type, including the job vtable.  */
     const BlockJobDriver *driver;
 

@@ -47,11 +51,6 @@ typedef struct BlockJob {
     BlockBackend *blk;
 
     /**
-     * The ID of the block job. May be NULL for internal jobs.
-     */
-    char *id;
-
-    /**
      * The coroutine that executes the job.  If not NULL, it is
      * reentered when busy is false and the job is cancelled.
      */

index 62ec964d09..e8eca44747 100644
@@ -35,8 +35,8 @@
  * A class type for block job driver.
  */
 struct BlockJobDriver {
-    /** Derived BlockJob struct size */
-    size_t instance_size;
+    /** Generic JobDriver callbacks and settings */
+    JobDriver job_driver;
 
     /** String describing the operation, part of query-block-jobs QMP API */
     BlockJobType job_type;

60 lines not on this page · raw message

index e14d99560d..9e672bbd5e 100644
@@ -523,7 +523,9 @@ static void coroutine_fn backup_run(void *opaque)
 }
 
 static const BlockJobDriver backup_job_driver = {
-    .instance_size          = sizeof(BackupBlockJob),
+    .job_driver = {
+        .instance_size          = sizeof(BackupBlockJob),
+    },
     .job_type               = BLOCK_JOB_TYPE_BACKUP,
     .start                  = backup_run,
     .commit                 = backup_commit,

index ba5df6aa0a..18cbb2f9c4 100644
@@ -215,7 +215,9 @@ out:
 }
 
 static const BlockJobDriver commit_job_driver = {
-    .instance_size = sizeof(CommitBlockJob),
+    .job_driver = {
+        .instance_size = sizeof(CommitBlockJob),
+    },
     .job_type      = BLOCK_JOB_TYPE_COMMIT,
     .start         = commit_run,
 };

index 350f733d0a..c75c7100fb 100644
@@ -913,7 +913,7 @@ static void mirror_complete(BlockJob *job, Error **errp)
 
     if (!s->synced) {
         error_setg(errp, "The active block job '%s' cannot be completed",
-                   job->id);
+                   job->job.id);
         return;
     }
 

@@ -986,7 +986,9 @@ static void mirror_drain(BlockJob *job)
 }
 
 static const BlockJobDriver mirror_job_driver = {
-    .instance_size          = sizeof(MirrorBlockJob),
+    .job_driver = {
+        .instance_size          = sizeof(MirrorBlockJob),
+    },
     .job_type               = BLOCK_JOB_TYPE_MIRROR,
     .start                  = mirror_run,
     .complete               = mirror_complete,

@@ -996,7 +998,9 @@ static const BlockJobDriver mirror_job_driver = {
 };
 
 static const BlockJobDriver commit_active_job_driver = {
-    .instance_size          = sizeof(MirrorBlockJob),
+    .job_driver = {
+        .instance_size          = sizeof(MirrorBlockJob),
+    },
     .job_type               = BLOCK_JOB_TYPE_COMMIT,
     .start                  = mirror_run,
     .complete               = mirror_complete,

index df9660d2fc..f88fc75141 100644
@@ -209,7 +209,9 @@ out:
 }
 
 static const BlockJobDriver stream_job_driver = {
-    .instance_size = sizeof(StreamBlockJob),
+    .job_driver = {
+        .instance_size = sizeof(StreamBlockJob),
+    },
     .job_type      = BLOCK_JOB_TYPE_STREAM,
     .start         = stream_run,
 };

154 lines not on this page · raw message

48 lines not on this page · raw message

index 7673de1062..fe9f412b39 100644
@@ -520,7 +520,9 @@ static void test_job_complete(BlockJob *job, Error **errp)
 }
 
 BlockJobDriver test_job_driver = {
-    .instance_size  = sizeof(TestBlockJob),
+    .job_driver = {
+        .instance_size  = sizeof(TestBlockJob),
+    },
     .start          = test_job_start,
     .complete       = test_job_complete,
 };

index 5789893dda..48b12d1744 100644
@@ -74,7 +74,9 @@ static void test_block_job_cb(void *opaque, int ret)
 }
 
 static const BlockJobDriver test_block_job_driver = {
-    .instance_size = sizeof(TestBlockJob),
+    .job_driver = {
+        .instance_size = sizeof(TestBlockJob),
+    },
     .start = test_block_job_run,
 };
 

index 8946bfd37b..b82026180a 100644
@@ -17,7 +17,9 @@
 #include "sysemu/block-backend.h"
 
 static const BlockJobDriver test_block_job_driver = {
-    .instance_size = sizeof(BlockJob),
+    .job_driver = {
+        .instance_size = sizeof(BlockJob),
+    },
 };
 
 static void block_job_cb(void *opaque, int ret)

@@ -38,9 +40,9 @@ static BlockJob *mk_job(BlockBackend *blk, const char *id,
         g_assert_null(errp);
         g_assert_nonnull(job);
         if (id) {
-            g_assert_cmpstr(job->id, ==, id);
+            g_assert_cmpstr(job->job.id, ==, id);
         } else {
-            g_assert_cmpstr(job->id, ==, blk_name(blk));
+            g_assert_cmpstr(job->job.id, ==, blk_name(blk));
         }
     } else {
         g_assert_nonnull(errp);

@@ -192,7 +194,9 @@ static void coroutine_fn cancel_job_start(void *opaque)
 }
 
 static const BlockJobDriver test_cancel_driver = {
-    .instance_size = sizeof(CancelJob),
+    .job_driver = {
+        .instance_size = sizeof(CancelJob),
+    },
     .start         = cancel_job_start,
     .complete      = cancel_job_complete,
 };

index e8eca44747..8e7e0a2f57 100644
@@ -39,7 +39,7 @@ struct BlockJobDriver {
     JobDriver job_driver;
 
     /** String describing the operation, part of query-block-jobs QMP API */
-    BlockJobType job_type;
+    JobType job_type;
 
     /** Mandatory: Entrypoint for the Coroutine. */
     CoroutineEntry *start;

index 9e672bbd5e..c49ea92dca 100644
@@ -526,7 +526,7 @@ static const BlockJobDriver backup_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(BackupBlockJob),
     },
-    .job_type               = BLOCK_JOB_TYPE_BACKUP,
+    .job_type               = JOB_TYPE_BACKUP,
     .start                  = backup_run,
     .commit                 = backup_commit,
     .abort                  = backup_abort,

index 18cbb2f9c4..afa2b2bacf 100644
@@ -218,7 +218,7 @@ static const BlockJobDriver commit_job_driver = {
     .job_driver = {
         .instance_size = sizeof(CommitBlockJob),
     },
-    .job_type      = BLOCK_JOB_TYPE_COMMIT,
+    .job_type      = JOB_TYPE_COMMIT,
     .start         = commit_run,
 };
 

index c75c7100fb..243a58f465 100644
@@ -989,7 +989,7 @@ static const BlockJobDriver mirror_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(MirrorBlockJob),
     },
-    .job_type               = BLOCK_JOB_TYPE_MIRROR,
+    .job_type               = JOB_TYPE_MIRROR,
     .start                  = mirror_run,
     .complete               = mirror_complete,
     .pause                  = mirror_pause,

@@ -1001,7 +1001,7 @@ static const BlockJobDriver commit_active_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(MirrorBlockJob),
     },
-    .job_type               = BLOCK_JOB_TYPE_COMMIT,
+    .job_type               = JOB_TYPE_COMMIT,
     .start                  = mirror_run,
     .complete               = mirror_complete,
     .pause                  = mirror_pause,

index f88fc75141..048bceb5d0 100644
@@ -212,7 +212,7 @@ static const BlockJobDriver stream_job_driver = {
     .job_driver = {
         .instance_size = sizeof(StreamBlockJob),
     },
-    .job_type      = BLOCK_JOB_TYPE_STREAM,
+    .job_type      = JOB_TYPE_STREAM,
     .start         = stream_run,
 };
 

index 35d604d05a..788b8cf686 100644
@@ -296,7 +296,7 @@ static char *child_job_get_parent_desc(BdrvChild *c)
 {
     BlockJob *job = c->opaque;
     return g_strdup_printf("%s job '%s'",
-                           BlockJobType_str(job->driver->job_type),
+                           JobType_str(job->driver->job_type),
                            job->job.id);
 }
 

@@ -839,7 +839,7 @@ BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
         return NULL;
     }
     info = g_new0(BlockJobInfo, 1);
-    info->type      = g_strdup(BlockJobType_str(job->driver->job_type));
+    info->type      = g_strdup(JobType_str(job->driver->job_type));
     info->device    = g_strdup(job->job.id);
     info->len       = job->len;
     info->busy      = atomic_read(&job->busy);

@@ -972,7 +972,7 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
                    block_job_sleep_timer_cb, job);
 
     error_setg(&job->blocker, "block device is in use by block job: %s",
-               BlockJobType_str(driver->job_type));
+               JobType_str(driver->job_type));
     block_job_add_bdrv(job, "main node", bs, 0, BLK_PERM_ALL, &error_abort);
     bs->job = job;
 

index 8e7e0a2f57..1e62d6dd30 100644
@@ -38,9 +38,6 @@ struct BlockJobDriver {
     /** Generic JobDriver callbacks and settings */
     JobDriver job_driver;
 
-    /** String describing the operation, part of query-block-jobs QMP API */
-    JobType job_type;
-
     /** Mandatory: Entrypoint for the Coroutine. */
     CoroutineEntry *start;
 

index b4b49f19e1..c87e951c8a 100644
@@ -26,6 +26,8 @@
 #ifndef JOB_H
 #define JOB_H
 
+#include "qapi/qapi-types-block-core.h"
+
 typedef struct JobDriver JobDriver;
 
 /**

@@ -45,6 +47,9 @@ typedef struct Job {
 struct JobDriver {
     /** Derived Job struct size */
     size_t instance_size;
+
+    /** Enum describing the operation */
+    JobType job_type;
 };
 
 

@@ -57,4 +62,10 @@ struct JobDriver {
  */
 void *job_create(const char *job_id, const JobDriver *driver, Error **errp);
 
+/** Returns the JobType of a given Job. */
+JobType job_type(Job *job);
+
+/** Returns the enum string for the JobType of a given Job. */
+const char *job_type_str(Job *job);
+
 #endif

index c49ea92dca..baf8d432da 100644
@@ -525,8 +525,8 @@ static void coroutine_fn backup_run(void *opaque)
 static const BlockJobDriver backup_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(BackupBlockJob),
+        .job_type               = JOB_TYPE_BACKUP,
     },
-    .job_type               = JOB_TYPE_BACKUP,
     .start                  = backup_run,
     .commit                 = backup_commit,
     .abort                  = backup_abort,

index afa2b2bacf..32d29c890e 100644
@@ -217,8 +217,8 @@ out:
 static const BlockJobDriver commit_job_driver = {
     .job_driver = {
         .instance_size = sizeof(CommitBlockJob),
+        .job_type      = JOB_TYPE_COMMIT,
     },
-    .job_type      = JOB_TYPE_COMMIT,
     .start         = commit_run,
 };
 

index 243a58f465..054972bdc1 100644
@@ -988,8 +988,8 @@ static void mirror_drain(BlockJob *job)
 static const BlockJobDriver mirror_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(MirrorBlockJob),
+        .job_type               = JOB_TYPE_MIRROR,
     },
-    .job_type               = JOB_TYPE_MIRROR,
     .start                  = mirror_run,
     .complete               = mirror_complete,
     .pause                  = mirror_pause,

@@ -1000,8 +1000,8 @@ static const BlockJobDriver mirror_job_driver = {
 static const BlockJobDriver commit_active_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(MirrorBlockJob),
+        .job_type               = JOB_TYPE_COMMIT,
     },
-    .job_type               = JOB_TYPE_COMMIT,
     .start                  = mirror_run,
     .complete               = mirror_complete,
     .pause                  = mirror_pause,

index 048bceb5d0..cb723f190a 100644
@@ -211,8 +211,8 @@ out:
 static const BlockJobDriver stream_job_driver = {
     .job_driver = {
         .instance_size = sizeof(StreamBlockJob),
+        .job_type      = JOB_TYPE_STREAM,
     },
-    .job_type      = JOB_TYPE_STREAM,
     .start         = stream_run,
 };
 

index 788b8cf686..3cf28eb730 100644
@@ -295,9 +295,7 @@ static void block_job_detach_aio_context(void *opaque)
 static char *child_job_get_parent_desc(BdrvChild *c)
 {
     BlockJob *job = c->opaque;
-    return g_strdup_printf("%s job '%s'",
-                           JobType_str(job->driver->job_type),
-                           job->job.id);
+    return g_strdup_printf("%s job '%s'", job_type_str(&job->job), job->job.id);
 }
 
 static void child_job_drained_begin(BdrvChild *c)

@@ -839,7 +837,7 @@ BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
         return NULL;
     }
     info = g_new0(BlockJobInfo, 1);
-    info->type      = g_strdup(JobType_str(job->driver->job_type));
+    info->type      = g_strdup(job_type_str(&job->job));
     info->device    = g_strdup(job->job.id);
     info->len       = job->len;
     info->busy      = atomic_read(&job->busy);

@@ -870,7 +868,7 @@ static void block_job_event_cancelled(BlockJob *job)
         return;
     }
 
-    qapi_event_send_block_job_cancelled(job->driver->job_type,
+    qapi_event_send_block_job_cancelled(job_type(&job->job),
                                         job->job.id,
                                         job->len,
                                         job->offset,

@@ -884,7 +882,7 @@ static void block_job_event_completed(BlockJob *job, const char *msg)
         return;
     }
 
-    qapi_event_send_block_job_completed(job->driver->job_type,
+    qapi_event_send_block_job_completed(job_type(&job->job),
                                         job->job.id,
                                         job->len,
                                         job->offset,

@@ -898,7 +896,7 @@ static int block_job_event_pending(BlockJob *job)
 {
     block_job_state_transition(job, BLOCK_JOB_STATUS_PENDING);
     if (!job->auto_finalize && !block_job_is_internal(job)) {
-        qapi_event_send_block_job_pending(job->driver->job_type,
+        qapi_event_send_block_job_pending(job_type(&job->job),
                                           job->job.id,
                                           &error_abort);
     }

@@ -972,7 +970,7 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
                    block_job_sleep_timer_cb, job);
 
     error_setg(&job->blocker, "block device is in use by block job: %s",
-               JobType_str(driver->job_type));
+               job_type_str(&job->job));
     block_job_add_bdrv(job, "main node", bs, 0, BLK_PERM_ALL, &error_abort);
     bs->job = job;
 

@@ -1176,7 +1174,7 @@ void block_job_event_ready(BlockJob *job)
         return;
     }
 
-    qapi_event_send_block_job_ready(job->driver->job_type,
+    qapi_event_send_block_job_ready(job_type(&job->job),
                                     job->job.id,
                                     job->len,
                                     job->offset,

index 87fd48468c..f00f401502 100644
@@ -29,6 +29,16 @@
 #include "qemu/job.h"
 #include "qemu/id.h"
 
+JobType job_type(Job *job)
+{
+    return job->driver->job_type;
+}
+
+const char *job_type_str(Job *job)
+{
+    return JobType_str(job_type(job));
+}
+
 void *job_create(const char *job_id, const JobDriver *driver, Error **errp)
 {
     Job *job;

index c87e951c8a..ee1f5d1ef4 100644
@@ -62,6 +62,9 @@ struct JobDriver {
  */
 void *job_create(const char *job_id, const JobDriver *driver, Error **errp);
 
+/** Frees the @job object. */
+void job_delete(Job *job);
+
 /** Returns the JobType of a given Job. */
 JobType job_type(Job *job);
 

index 3cf28eb730..f1bff2272e 100644
@@ -247,9 +247,8 @@ void block_job_unref(BlockJob *job)
                                         block_job_detach_aio_context, job);
         blk_unref(job->blk);
         error_free(job->blocker);
-        g_free(job->job.id);
         assert(!timer_pending(&job->sleep_timer));
-        g_free(job);
+        job_delete(&job->job);
     }
 }
 

index f00f401502..a36425498d 100644
@@ -56,3 +56,9 @@ void *job_create(const char *job_id, const JobDriver *driver, Error **errp)
 
     return job;
 }
+
+void job_delete(Job *job)
+{
+    g_free(job->id);
+    g_free(job);
+}

index 8acc1a236a..8946f118f3 100644
@@ -105,9 +105,6 @@ typedef struct BlockJob {
      */
     bool deferred_to_main_loop;
 
-    /** Element of the list of block jobs */
-    QLIST_ENTRY(BlockJob) job_list;
-
     /** Status that is published by the query-block-jobs QMP API */
     BlockDeviceIoStatus iostatus;
 

index ee1f5d1ef4..6555ab4c36 100644
@@ -27,6 +27,7 @@
 #define JOB_H
 
 #include "qapi/qapi-types-block-core.h"
+#include "qemu/queue.h"
 
 typedef struct JobDriver JobDriver;
 

@@ -39,6 +40,9 @@ typedef struct Job {
 
     /** The type of this job. */
     const JobDriver *driver;
+
+    /** Element of the list of jobs */
+    QLIST_ENTRY(Job) job_list;
 } Job;
 
 /**

@@ -71,4 +75,19 @@ JobType job_type(Job *job);
 /** Returns the enum string for the JobType of a given Job. */
 const char *job_type_str(Job *job);
 
+/**
+ * Get the next element from the list of block jobs after @job, or the
+ * first one if @job is %NULL.
+ *
+ * Returns the requested job, or %NULL if there are no more jobs left.
+ */
+Job *job_next(Job *job);
+
+/**
+ * Get the job identified by @id (which must not be %NULL).
+ *
+ * Returns the requested job, or %NULL if it doesn't exist.
+ */
+Job *job_get(const char *id);
+
 #endif

96 lines not on this page · raw message

index a36425498d..00f7968c50 100644
@@ -29,6 +29,8 @@
 #include "qemu/job.h"
 #include "qemu/id.h"
 
+static QLIST_HEAD(, Job) jobs = QLIST_HEAD_INITIALIZER(jobs);
+
 JobType job_type(Job *job)
 {
     return job->driver->job_type;

@@ -39,6 +41,27 @@ const char *job_type_str(Job *job)
     return JobType_str(job_type(job));
 }
 
+Job *job_next(Job *job)
+{
+    if (!job) {
+        return QLIST_FIRST(&jobs);
+    }
+    return QLIST_NEXT(job, job_list);
+}
+
+Job *job_get(const char *id)
+{
+    Job *job;
+
+    QLIST_FOREACH(job, &jobs, job_list) {
+        if (job->id && !strcmp(id, job->id)) {
+            return job;
+        }
+    }
+
+    return NULL;
+}
+
 void *job_create(const char *job_id, const JobDriver *driver, Error **errp)
 {
     Job *job;

@@ -48,17 +71,25 @@ void *job_create(const char *job_id, const JobDriver *driver, Error **errp)
             error_setg(errp, "Invalid job ID '%s'", job_id);
             return NULL;
         }
+        if (job_get(job_id)) {
+            error_setg(errp, "Job ID '%s' already in use", job_id);
+            return NULL;
+        }
     }
 
     job = g_malloc0(driver->instance_size);
     job->driver        = driver;
     job->id            = g_strdup(job_id);
 
+    QLIST_INSERT_HEAD(&jobs, job, job_list);
+
     return job;
 }
 
 void job_delete(Job *job)
 {
+    QLIST_REMOVE(job, job_list);
+
     g_free(job->id);
     g_free(job);
 }

index 8946f118f3..96e5f67ed7 100644
@@ -147,9 +147,6 @@ typedef struct BlockJob {
      */
     QEMUTimer sleep_timer;
 
-    /** Current state; See @BlockJobStatus for details. */
-    BlockJobStatus status;
-
     /** True if this job should automatically finalize itself */
     bool auto_finalize;
 

index 6555ab4c36..48a7dfc431 100644
@@ -41,6 +41,9 @@ typedef struct Job {
     /** The type of this job. */
     const JobDriver *driver;
 
+    /** Current state; See @JobStatus for details. */
+    JobStatus status;
+
     /** Element of the list of jobs */
     QLIST_ENTRY(Job) job_list;
 } Job;

@@ -90,4 +93,14 @@ Job *job_next(Job *job);
  */
 Job *job_get(const char *id);
 
+/**
+ * Check whether the verb @bv can be applied to @job in its current state.
+ * Returns 0 if the verb can be applied; otherwise errp is set and -EPERM
+ * returned.
+ */
+int job_apply_verb(Job *job, JobVerb bv, Error **errp);
+
+/* TODO To be removed from the public interface */
+void job_state_transition(Job *job, JobStatus s1);
+
 #endif

231 lines not on this page · raw message

71 lines not on this page · raw message

index b82026180a..6ccd585dee 100644
@@ -211,7 +211,7 @@ static CancelJob *create_common(BlockJob **pjob)
     job = mk_job(blk, "Steve", &test_cancel_driver, true,
                  BLOCK_JOB_MANUAL_FINALIZE | BLOCK_JOB_MANUAL_DISMISS);
     block_job_ref(job);
-    assert(job->status == BLOCK_JOB_STATUS_CREATED);
+    assert(job->job.status == JOB_STATUS_CREATED);
     s = container_of(job, CancelJob, common);
     s->blk = blk;
 

@@ -223,15 +223,14 @@ static void cancel_common(CancelJob *s)
 {
     BlockJob *job = &s->common;
     BlockBackend *blk = s->blk;
-    BlockJobStatus sts = job->status;
+    JobStatus sts = job->job.status;
 
     block_job_cancel_sync(job);
-    if ((sts != BLOCK_JOB_STATUS_CREATED) &&
-        (sts != BLOCK_JOB_STATUS_CONCLUDED)) {
+    if (sts != JOB_STATUS_CREATED && sts != JOB_STATUS_CONCLUDED) {
         BlockJob *dummy = job;
         block_job_dismiss(&dummy, &error_abort);
     }
-    assert(job->status == BLOCK_JOB_STATUS_NULL);
+    assert(job->job.status == JOB_STATUS_NULL);
     block_job_unref(job);
     destroy_blk(blk);
 }

@@ -253,7 +252,7 @@ static void test_cancel_running(void)
     s = create_common(&job);
 
     block_job_start(job);
-    assert(job->status == BLOCK_JOB_STATUS_RUNNING);
+    assert(job->job.status == JOB_STATUS_RUNNING);
 
     cancel_common(s);
 }

@@ -266,11 +265,11 @@ static void test_cancel_paused(void)
     s = create_common(&job);
 
     block_job_start(job);
-    assert(job->status == BLOCK_JOB_STATUS_RUNNING);
+    assert(job->job.status == JOB_STATUS_RUNNING);
 
     block_job_user_pause(job, &error_abort);
     block_job_enter(job);
-    assert(job->status == BLOCK_JOB_STATUS_PAUSED);
+    assert(job->job.status == JOB_STATUS_PAUSED);
 
     cancel_common(s);
 }

@@ -283,11 +282,11 @@ static void test_cancel_ready(void)
     s = create_common(&job);
 
     block_job_start(job);
-    assert(job->status == BLOCK_JOB_STATUS_RUNNING);
+    assert(job->job.status == JOB_STATUS_RUNNING);
 
     s->should_converge = true;
     block_job_enter(job);
-    assert(job->status == BLOCK_JOB_STATUS_READY);
+    assert(job->job.status == JOB_STATUS_READY);
 
     cancel_common(s);
 }

@@ -300,15 +299,15 @@ static void test_cancel_standby(void)
     s = create_common(&job);
 
     block_job_start(job);
-    assert(job->status == BLOCK_JOB_STATUS_RUNNING);
+    assert(job->job.status == JOB_STATUS_RUNNING);
 
     s->should_converge = true;
     block_job_enter(job);
-    assert(job->status == BLOCK_JOB_STATUS_READY);
+    assert(job->job.status == JOB_STATUS_READY);
 
     block_job_user_pause(job, &error_abort);
     block_job_enter(job);
-    assert(job->status == BLOCK_JOB_STATUS_STANDBY);
+    assert(job->job.status == JOB_STATUS_STANDBY);
 
     cancel_common(s);
 }

@@ -321,18 +320,18 @@ static void test_cancel_pending(void)
     s = create_common(&job);
 
     block_job_start(job);
-    assert(job->status == BLOCK_JOB_STATUS_RUNNING);
+    assert(job->job.status == JOB_STATUS_RUNNING);
 
     s->should_converge = true;
     block_job_enter(job);
-    assert(job->status == BLOCK_JOB_STATUS_READY);
+    assert(job->job.status == JOB_STATUS_READY);
 
     block_job_complete(job, &error_abort);
     block_job_enter(job);
     while (!s->completed) {
         aio_poll(qemu_get_aio_context(), true);
     }
-    assert(job->status == BLOCK_JOB_STATUS_PENDING);
+    assert(job->job.status == JOB_STATUS_PENDING);
 
     cancel_common(s);
 }

@@ -345,21 +344,21 @@ static void test_cancel_concluded(void)
     s = create_common(&job);
 
     block_job_start(job);
-    assert(job->status == BLOCK_JOB_STATUS_RUNNING);
+    assert(job->job.status == JOB_STATUS_RUNNING);
 
     s->should_converge = true;
     block_job_enter(job);
-    assert(job->status == BLOCK_JOB_STATUS_READY);
+    assert(job->job.status == JOB_STATUS_READY);
 
     block_job_complete(job, &error_abort);
     block_job_enter(job);
     while (!s->completed) {
         aio_poll(qemu_get_aio_context(), true);
     }
-    assert(job->status == BLOCK_JOB_STATUS_PENDING);
+    assert(job->job.status == JOB_STATUS_PENDING);
 
     block_job_finalize(job, &error_abort);
-    assert(job->status == BLOCK_JOB_STATUS_CONCLUDED);
+    assert(job->job.status == JOB_STATUS_CONCLUDED);
 
     cancel_common(s);
 }

index 96e5f67ed7..5e32e719b6 100644
@@ -132,9 +132,6 @@ typedef struct BlockJob {
     /** The opaque value that is passed to the completion function.  */
     void *opaque;
 
-    /** Reference count of the block job */
-    int refcnt;
-
     /** True when job has reported completion by calling block_job_completed. */
     bool completed;
 

@@ -400,24 +397,6 @@ void block_job_iostatus_reset(BlockJob *job);
 BlockJobTxn *block_job_txn_new(void);
 
 /**
- * block_job_ref:
- *
- * Add a reference to BlockJob refcnt, it will be decreased with
- * block_job_unref, and then be freed if it comes to be the last
- * reference.
- */
-void block_job_ref(BlockJob *job);
-
-/**
- * block_job_unref:
- *
- * Release a reference that was previously acquired with block_job_ref
- * or block_job_create. If it's the last reference to the object, it will be
- * freed.
- */
-void block_job_unref(BlockJob *job);
-
-/**
  * block_job_txn_unref:
  *
  * Release a reference that was previously acquired with block_job_txn_add_job

index 1e62d6dd30..6f0fe3c48d 100644
@@ -144,6 +144,13 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
                        BlockCompletionFunc *cb, void *opaque, Error **errp);
 
 /**
+ * block_job_free:
+ * Callback to be used for JobDriver.free in all block jobs. Frees block job
+ * specific resources in @job.
+ */
+void block_job_free(Job *job);
+
+/**
  * block_job_sleep_ns:
  * @job: The job that calls the function.
  * @ns: How many nanoseconds to stop for.

index 48a7dfc431..64e988f305 100644
@@ -41,6 +41,9 @@ typedef struct Job {
     /** The type of this job. */
     const JobDriver *driver;
 
+    /** Reference count of the block job */
+    int refcnt;
+
     /** Current state; See @JobStatus for details. */
     JobStatus status;
 

@@ -57,6 +60,9 @@ struct JobDriver {
 
     /** Enum describing the operation */
     JobType job_type;
+
+    /** Called when the job is freed */
+    void (*free)(Job *job);
 };
 
 

@@ -69,8 +75,17 @@ struct JobDriver {
  */
 void *job_create(const char *job_id, const JobDriver *driver, Error **errp);
 
-/** Frees the @job object. */
-void job_delete(Job *job);
+/**
+ * Add a reference to Job refcnt, it will be decreased with job_unref, and then
+ * be freed if it comes to be the last reference.
+ */
+void job_ref(Job *job);
+
+/**
+ * Release a reference that was previously acquired with job_ref() or
+ * job_create(). If it's the last reference to the object, it will be freed.
+ */
+void job_unref(Job *job);
 
 /** Returns the JobType of a given Job. */
 JobType job_type(Job *job);

index baf8d432da..cfdb89d977 100644
@@ -526,6 +526,7 @@ static const BlockJobDriver backup_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(BackupBlockJob),
         .job_type               = JOB_TYPE_BACKUP,
+        .free                   = block_job_free,
     },
     .start                  = backup_run,
     .commit                 = backup_commit,

index 32d29c890e..925c96abe7 100644
@@ -218,6 +218,7 @@ static const BlockJobDriver commit_job_driver = {
     .job_driver = {
         .instance_size = sizeof(CommitBlockJob),
         .job_type      = JOB_TYPE_COMMIT,
+        .free          = block_job_free,
     },
     .start         = commit_run,
 };

index 054972bdc1..9b1e5c0849 100644
@@ -989,6 +989,7 @@ static const BlockJobDriver mirror_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(MirrorBlockJob),
         .job_type               = JOB_TYPE_MIRROR,
+        .free                   = block_job_free,
     },
     .start                  = mirror_run,
     .complete               = mirror_complete,

@@ -1001,6 +1002,7 @@ static const BlockJobDriver commit_active_job_driver = {
     .job_driver = {
         .instance_size          = sizeof(MirrorBlockJob),
         .job_type               = JOB_TYPE_COMMIT,
+        .free                   = block_job_free,
     },
     .start                  = mirror_run,
     .complete               = mirror_complete,

index cb723f190a..7273d2213c 100644
@@ -212,6 +212,7 @@ static const BlockJobDriver stream_job_driver = {
     .job_driver = {
         .instance_size = sizeof(StreamBlockJob),
         .job_type      = JOB_TYPE_STREAM,
+        .free          = block_job_free,
     },
     .start         = stream_run,
 };

107 lines not on this page · raw message

index 93e846f7e2..67b09cb54d 100644
@@ -134,6 +134,7 @@ void *job_create(const char *job_id, const JobDriver *driver, Error **errp)
     job = g_malloc0(driver->instance_size);
     job->driver        = driver;
     job->id            = g_strdup(job_id);
+    job->refcnt        = 1;
 
     job_state_transition(job, JOB_STATUS_CREATED);
 

@@ -142,10 +143,23 @@ void *job_create(const char *job_id, const JobDriver *driver, Error **errp)
     return job;
 }
 
-void job_delete(Job *job)
+void job_ref(Job *job)
 {
-    QLIST_REMOVE(job, job_list);
+    ++job->refcnt;
+}
+
+void job_unref(Job *job)
+{
+    if (--job->refcnt == 0) {
+        assert(job->status == JOB_STATUS_NULL);
 
-    g_free(job->id);
-    g_free(job);
+        if (job->driver->free) {
+            job->driver->free(job);
+        }
+
+        QLIST_REMOVE(job, job_list);
+
+        g_free(job->id);
+        g_free(job);
+    }
 }

index ea62d2d61e..82f69269ae 100644
@@ -862,7 +862,7 @@ static void run_block_job(BlockJob *job, Error **errp)
     int ret = 0;
 
     aio_context_acquire(aio_context);
-    block_job_ref(job);
+    job_ref(&job->job);
     do {
         aio_poll(aio_context, true);
         qemu_progress_print(job->len ?

@@ -874,7 +874,7 @@ static void run_block_job(BlockJob *job, Error **errp)
     } else {
         ret = job->ret;
     }
-    block_job_unref(job);
+    job_unref(&job->job);
     aio_context_release(aio_context);
 
     /* publish completion progress only when success */

index fe9f412b39..f9e37d479c 100644
@@ -522,6 +522,7 @@ static void test_job_complete(BlockJob *job, Error **errp)
 BlockJobDriver test_job_driver = {
     .job_driver = {
         .instance_size  = sizeof(TestBlockJob),
+        .free           = block_job_free,
     },
     .start          = test_job_start,
     .complete       = test_job_complete,

index 48b12d1744..b49b28ca27 100644
@@ -76,6 +76,7 @@ static void test_block_job_cb(void *opaque, int ret)
 static const BlockJobDriver test_block_job_driver = {
     .job_driver = {
         .instance_size = sizeof(TestBlockJob),
+        .free          = block_job_free,
     },
     .start = test_block_job_run,
 };

index 6ccd585dee..e24fc3f140 100644
@@ -19,6 +19,7 @@
 static const BlockJobDriver test_block_job_driver = {
     .job_driver = {
         .instance_size = sizeof(BlockJob),
+        .free          = block_job_free,
     },
 };
 

@@ -196,6 +197,7 @@ static void coroutine_fn cancel_job_start(void *opaque)
 static const BlockJobDriver test_cancel_driver = {
     .job_driver = {
         .instance_size = sizeof(CancelJob),
+        .free          = block_job_free,
     },
     .start         = cancel_job_start,
     .complete      = cancel_job_complete,

@@ -210,7 +212,7 @@ static CancelJob *create_common(BlockJob **pjob)
     blk = create_blk(NULL);
     job = mk_job(blk, "Steve", &test_cancel_driver, true,
                  BLOCK_JOB_MANUAL_FINALIZE | BLOCK_JOB_MANUAL_DISMISS);
-    block_job_ref(job);
+    job_ref(&job->job);
     assert(job->job.status == JOB_STATUS_CREATED);
     s = container_of(job, CancelJob, common);
     s->blk = blk;

@@ -231,7 +233,7 @@ static void cancel_common(CancelJob *s)
         block_job_dismiss(&dummy, &error_abort);
     }
     assert(job->job.status == JOB_STATUS_NULL);
-    block_job_unref(job);
+    job_unref(&job->job);
     destroy_blk(blk);
 }
 

index 5e32e719b6..04efc94ffc 100644
@@ -57,14 +57,6 @@ typedef struct BlockJob {
     Coroutine *co;
 
     /**
-     * Set to true if the job should cancel itself.  The flag must
-     * always be tested just before toggling the busy flag from false
-     * to true.  After a job has been cancelled, it should only yield
-     * if #aio_poll will ("sooner or later") reenter the coroutine.
-     */
-    bool cancelled;
-
-    /**
      * Set to true if the job should abort immediately without waiting
      * for data to be in sync.
      */

index 6f0fe3c48d..d64f30e6b0 100644
@@ -196,14 +196,6 @@ void block_job_early_fail(BlockJob *job);
 void block_job_completed(BlockJob *job, int ret);
 
 /**
- * block_job_is_cancelled:
- * @job: The job being queried.
- *
- * Returns whether the job is scheduled for cancellation.
- */
-bool block_job_is_cancelled(BlockJob *job);
-
-/**
  * block_job_pause_point:
  * @job: The job that is ready to pause.
  *

index 64e988f305..14c6288517 100644
@@ -47,6 +47,14 @@ typedef struct Job {
     /** Current state; See @JobStatus for details. */
     JobStatus status;
 
+    /**
+     * Set to true if the job should cancel itself.  The flag must
+     * always be tested just before toggling the busy flag from false
+     * to true.  After a job has been cancelled, it should only yield
+     * if #aio_poll will ("sooner or later") reenter the coroutine.
+     */
+    bool cancelled;
+
     /** Element of the list of jobs */
     QLIST_ENTRY(Job) job_list;
 } Job;

@@ -93,6 +101,9 @@ JobType job_type(Job *job);
 /** Returns the enum string for the JobType of a given Job. */
 const char *job_type_str(Job *job);
 
+/** Returns whether the job is scheduled for cancellation. */
+bool job_is_cancelled(Job *job);
+
 /**
  * Get the next element from the list of block jobs after @job, or the
  * first one if @job is %NULL.

index cfdb89d977..ef0aa0e24e 100644
@@ -329,7 +329,7 @@ static bool coroutine_fn yield_and_check(BackupBlockJob *job)
 {
     uint64_t delay_ns;
 
-    if (block_job_is_cancelled(&job->common)) {
+    if (job_is_cancelled(&job->common.job)) {
         return true;
     }
 

@@ -339,7 +339,7 @@ static bool coroutine_fn yield_and_check(BackupBlockJob *job)
     job->bytes_read = 0;
     block_job_sleep_ns(&job->common, delay_ns);
 
-    if (block_job_is_cancelled(&job->common)) {
+    if (job_is_cancelled(&job->common.job)) {
         return true;
     }
 

@@ -441,7 +441,7 @@ static void coroutine_fn backup_run(void *opaque)
     if (job->sync_mode == MIRROR_SYNC_MODE_NONE) {
         /* All bits are set in copy_bitmap to allow any cluster to be copied.
          * This does not actually require them to be copied. */
-        while (!block_job_is_cancelled(&job->common)) {
+        while (!job_is_cancelled(&job->common.job)) {
             /* Yield until the job is cancelled.  We just let our before_write
              * notify callback service CoW requests. */
             block_job_yield(&job->common);

index 925c96abe7..85baea8f92 100644
@@ -90,7 +90,7 @@ static void commit_complete(BlockJob *job, void *opaque)
      * the normal backing chain can be restored. */
     blk_unref(s->base);
 
-    if (!block_job_is_cancelled(&s->common) && ret == 0) {
+    if (!job_is_cancelled(&s->common.job) && ret == 0) {
         /* success */
         ret = bdrv_drop_intermediate(s->commit_top_bs, base,
                                      s->backing_file_str);

@@ -172,7 +172,7 @@ static void coroutine_fn commit_run(void *opaque)
          * with no pending I/O here so that bdrv_drain_all() returns.
          */
         block_job_sleep_ns(&s->common, delay_ns);
-        if (block_job_is_cancelled(&s->common)) {
+        if (job_is_cancelled(&s->common.job)) {
             break;
         }
         /* Copy if allocated above the base */

index 9b1e5c0849..163d83e34a 100644
@@ -622,7 +622,7 @@ static int coroutine_fn mirror_dirty_init(MirrorBlockJob *s)
 
             mirror_throttle(s);
 
-            if (block_job_is_cancelled(&s->common)) {
+            if (job_is_cancelled(&s->common.job)) {
                 s->initial_zeroing_ongoing = false;
                 return 0;
             }

@@ -650,7 +650,7 @@ static int coroutine_fn mirror_dirty_init(MirrorBlockJob *s)
 
         mirror_throttle(s);
 
-        if (block_job_is_cancelled(&s->common)) {
+        if (job_is_cancelled(&s->common.job)) {
             return 0;
         }
 

@@ -695,7 +695,7 @@ static void coroutine_fn mirror_run(void *opaque)
                                  checking for a NULL string */
     int ret = 0;
 
-    if (block_job_is_cancelled(&s->common)) {
+    if (job_is_cancelled(&s->common.job)) {
         goto immediate_exit;
     }
 

@@ -729,10 +729,10 @@ static void coroutine_fn mirror_run(void *opaque)
         /* Report BLOCK_JOB_READY and wait for complete. */
         block_job_event_ready(&s->common);
         s->synced = true;
-        while (!block_job_is_cancelled(&s->common) && !s->should_complete) {
+        while (!job_is_cancelled(&s->common.job) && !s->should_complete) {
             block_job_yield(&s->common);
         }
-        s->common.cancelled = false;
+        s->common.job.cancelled = false;
         goto immediate_exit;
     }
 

@@ -768,7 +768,7 @@ static void coroutine_fn mirror_run(void *opaque)
     s->last_pause_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
     if (!s->is_none_mode) {
         ret = mirror_dirty_init(s);
-        if (ret < 0 || block_job_is_cancelled(&s->common)) {
+        if (ret < 0 || job_is_cancelled(&s->common.job)) {
             goto immediate_exit;
         }
     }

@@ -829,7 +829,7 @@ static void coroutine_fn mirror_run(void *opaque)
             }
 
             should_complete = s->should_complete ||
-                block_job_is_cancelled(&s->common);
+                job_is_cancelled(&s->common.job);
             cnt = bdrv_get_dirty_count(s->dirty_bitmap);
         }
 

@@ -857,7 +857,7 @@ static void coroutine_fn mirror_run(void *opaque)
              * completion.
              */
             assert(QLIST_EMPTY(&bs->tracked_requests));
-            s->common.cancelled = false;
+            s->common.job.cancelled = false;
             need_drain = false;
             break;
         }

@@ -869,7 +869,7 @@ static void coroutine_fn mirror_run(void *opaque)
         }
         trace_mirror_before_sleep(s, cnt, s->synced, delay_ns);
         block_job_sleep_ns(&s->common, delay_ns);
-        if (block_job_is_cancelled(&s->common) &&
+        if (job_is_cancelled(&s->common.job) &&
             (!s->synced || s->common.force))
         {
             break;

@@ -884,7 +884,7 @@ immediate_exit:
          * the target is a copy of the source.
          */
         assert(ret < 0 || ((s->common.force || !s->synced) &&
-               block_job_is_cancelled(&s->common)));
+               job_is_cancelled(&s->common.job)));
         assert(need_drain);
         mirror_wait_for_all_io(s);
     }

index 7273d2213c..22c71ae100 100644
@@ -66,7 +66,7 @@ static void stream_complete(BlockJob *job, void *opaque)
     BlockDriverState *base = s->base;
     Error *local_err = NULL;
 
-    if (!block_job_is_cancelled(&s->common) && bs->backing &&
+    if (!job_is_cancelled(&s->common.job) && bs->backing &&
         data->ret == 0) {
         const char *base_id = NULL, *base_fmt = NULL;
         if (base) {