cryptospore / rpms / qemu-kvm

Forked from rpms/qemu-kvm 2 years ago
Clone
26ba25
From 72dcce395974c2cc4ae59c9ba11243abf04f1bbd Mon Sep 17 00:00:00 2001
26ba25
From: Kevin Wolf <kwolf@redhat.com>
26ba25
Date: Tue, 26 Jun 2018 09:48:09 +0200
26ba25
Subject: [PATCH 101/268] job: Move coroutine and related code to Job
26ba25
26ba25
RH-Author: Kevin Wolf <kwolf@redhat.com>
26ba25
Message-id: <20180626094856.6924-27-kwolf@redhat.com>
26ba25
Patchwork-id: 81113
26ba25
O-Subject: [RHV-7.6 qemu-kvm-rhev PATCH v2 26/73] job: Move coroutine and related code to Job
26ba25
Bugzilla: 1513543
26ba25
RH-Acked-by: Jeffrey Cody <jcody@redhat.com>
26ba25
RH-Acked-by: Max Reitz <mreitz@redhat.com>
26ba25
RH-Acked-by: Fam Zheng <famz@redhat.com>
26ba25
26ba25
This commit moves some core functions for dealing with the job coroutine
26ba25
from BlockJob to Job. This includes primarily entering the coroutine
26ba25
(both for the first and reentering) and yielding explicitly and at pause
26ba25
points.
26ba25
26ba25
Signed-off-by: Kevin Wolf <kwolf@redhat.com>
26ba25
Reviewed-by: John Snow <jsnow@redhat.com>
26ba25
(cherry picked from commit da01ff7f38f52791f93fc3ca59afcfbb220f15af)
26ba25
Signed-off-by: Kevin Wolf <kwolf@redhat.com>
26ba25
Signed-off-by: Miroslav Rezanina <mrezanin@redhat.com>
26ba25
---
26ba25
 block/backup.c               |   2 +-
26ba25
 block/commit.c               |   4 +-
26ba25
 block/mirror.c               |  22 ++---
26ba25
 block/replication.c          |   2 +-
26ba25
 block/stream.c               |   4 +-
26ba25
 blockdev.c                   |   8 +-
26ba25
 blockjob.c                   | 219 ++++++++-----------------------------------
26ba25
 include/block/blockjob.h     |  40 --------
26ba25
 include/block/blockjob_int.h |  26 -----
26ba25
 include/qemu/job.h           |  76 +++++++++++++++
26ba25
 job.c                        | 137 +++++++++++++++++++++++++++
26ba25
 tests/test-bdrv-drain.c      |  38 ++++----
26ba25
 tests/test-blockjob-txn.c    |  12 +--
26ba25
 tests/test-blockjob.c        |  14 +--
26ba25
 14 files changed, 305 insertions(+), 299 deletions(-)
26ba25
26ba25
diff --git a/block/backup.c b/block/backup.c
26ba25
index 22dd368..7d9aad9 100644
26ba25
--- a/block/backup.c
26ba25
+++ b/block/backup.c
26ba25
@@ -528,8 +528,8 @@ static const BlockJobDriver backup_job_driver = {
26ba25
         .instance_size          = sizeof(BackupBlockJob),
26ba25
         .job_type               = JOB_TYPE_BACKUP,
26ba25
         .free                   = block_job_free,
26ba25
+        .start                  = backup_run,
26ba25
     },
26ba25
-    .start                  = backup_run,
26ba25
     .commit                 = backup_commit,
26ba25
     .abort                  = backup_abort,
26ba25
     .clean                  = backup_clean,
26ba25
diff --git a/block/commit.c b/block/commit.c
26ba25
index d326766..2fbc310 100644
26ba25
--- a/block/commit.c
26ba25
+++ b/block/commit.c
26ba25
@@ -220,8 +220,8 @@ static const BlockJobDriver commit_job_driver = {
26ba25
         .instance_size = sizeof(CommitBlockJob),
26ba25
         .job_type      = JOB_TYPE_COMMIT,
26ba25
         .free          = block_job_free,
26ba25
+        .start         = commit_run,
26ba25
     },
26ba25
-    .start         = commit_run,
26ba25
 };
26ba25
 
26ba25
 static int coroutine_fn bdrv_commit_top_preadv(BlockDriverState *bs,
26ba25
@@ -371,7 +371,7 @@ void commit_start(const char *job_id, BlockDriverState *bs,
26ba25
     s->on_error = on_error;
26ba25
 
26ba25
     trace_commit_start(bs, base, top, s);
26ba25
-    block_job_start(&s->common);
26ba25
+    job_start(&s->common.job);
26ba25
     return;
26ba25
 
26ba25
 fail:
26ba25
diff --git a/block/mirror.c b/block/mirror.c
26ba25
index 90d4ac9..95fc807 100644
26ba25
--- a/block/mirror.c
26ba25
+++ b/block/mirror.c
26ba25
@@ -126,7 +126,7 @@ static void mirror_iteration_done(MirrorOp *op, int ret)
26ba25
     g_free(op);
26ba25
 
26ba25
     if (s->waiting_for_io) {
26ba25
-        qemu_coroutine_enter(s->common.co);
26ba25
+        qemu_coroutine_enter(s->common.job.co);
26ba25
     }
26ba25
 }
26ba25
 
26ba25
@@ -345,7 +345,7 @@ static uint64_t coroutine_fn mirror_iteration(MirrorBlockJob *s)
26ba25
         mirror_wait_for_io(s);
26ba25
     }
26ba25
 
26ba25
-    block_job_pause_point(&s->common);
26ba25
+    job_pause_point(&s->common.job);
26ba25
 
26ba25
     /* Find the number of consective dirty chunks following the first dirty
26ba25
      * one, and wait for in flight requests in them. */
26ba25
@@ -597,7 +597,7 @@ static void mirror_throttle(MirrorBlockJob *s)
26ba25
         s->last_pause_ns = now;
26ba25
         block_job_sleep_ns(&s->common, 0);
26ba25
     } else {
26ba25
-        block_job_pause_point(&s->common);
26ba25
+        job_pause_point(&s->common.job);
26ba25
     }
26ba25
 }
26ba25
 
26ba25
@@ -786,7 +786,7 @@ static void coroutine_fn mirror_run(void *opaque)
26ba25
             goto immediate_exit;
26ba25
         }
26ba25
 
26ba25
-        block_job_pause_point(&s->common);
26ba25
+        job_pause_point(&s->common.job);
26ba25
 
26ba25
         cnt = bdrv_get_dirty_count(s->dirty_bitmap);
26ba25
         /* cnt is the number of dirty bytes remaining and s->bytes_in_flight is
26ba25
@@ -957,9 +957,9 @@ static void mirror_complete(BlockJob *job, Error **errp)
26ba25
     block_job_enter(&s->common);
26ba25
 }
26ba25
 
26ba25
-static void mirror_pause(BlockJob *job)
26ba25
+static void mirror_pause(Job *job)
26ba25
 {
26ba25
-    MirrorBlockJob *s = container_of(job, MirrorBlockJob, common);
26ba25
+    MirrorBlockJob *s = container_of(job, MirrorBlockJob, common.job);
26ba25
 
26ba25
     mirror_wait_for_all_io(s);
26ba25
 }
26ba25
@@ -991,10 +991,10 @@ static const BlockJobDriver mirror_job_driver = {
26ba25
         .instance_size          = sizeof(MirrorBlockJob),
26ba25
         .job_type               = JOB_TYPE_MIRROR,
26ba25
         .free                   = block_job_free,
26ba25
+        .start                  = mirror_run,
26ba25
+        .pause                  = mirror_pause,
26ba25
     },
26ba25
-    .start                  = mirror_run,
26ba25
     .complete               = mirror_complete,
26ba25
-    .pause                  = mirror_pause,
26ba25
     .attached_aio_context   = mirror_attached_aio_context,
26ba25
     .drain                  = mirror_drain,
26ba25
 };
26ba25
@@ -1004,10 +1004,10 @@ static const BlockJobDriver commit_active_job_driver = {
26ba25
         .instance_size          = sizeof(MirrorBlockJob),
26ba25
         .job_type               = JOB_TYPE_COMMIT,
26ba25
         .free                   = block_job_free,
26ba25
+        .start                  = mirror_run,
26ba25
+        .pause                  = mirror_pause,
26ba25
     },
26ba25
-    .start                  = mirror_run,
26ba25
     .complete               = mirror_complete,
26ba25
-    .pause                  = mirror_pause,
26ba25
     .attached_aio_context   = mirror_attached_aio_context,
26ba25
     .drain                  = mirror_drain,
26ba25
 };
26ba25
@@ -1244,7 +1244,7 @@ static void mirror_start_job(const char *job_id, BlockDriverState *bs,
26ba25
     }
26ba25
 
26ba25
     trace_mirror_start(bs, s, opaque);
26ba25
-    block_job_start(&s->common);
26ba25
+    job_start(&s->common.job);
26ba25
     return;
26ba25
 
26ba25
 fail:
26ba25
diff --git a/block/replication.c b/block/replication.c
26ba25
index 6c0c718..3f7500e 100644
26ba25
--- a/block/replication.c
26ba25
+++ b/block/replication.c
26ba25
@@ -574,7 +574,7 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
26ba25
             aio_context_release(aio_context);
26ba25
             return;
26ba25
         }
26ba25
-        block_job_start(job);
26ba25
+        job_start(&job->job);
26ba25
         break;
26ba25
     default:
26ba25
         aio_context_release(aio_context);
26ba25
diff --git a/block/stream.c b/block/stream.c
26ba25
index 0bba816..6d8b7b6 100644
26ba25
--- a/block/stream.c
26ba25
+++ b/block/stream.c
26ba25
@@ -213,8 +213,8 @@ static const BlockJobDriver stream_job_driver = {
26ba25
         .instance_size = sizeof(StreamBlockJob),
26ba25
         .job_type      = JOB_TYPE_STREAM,
26ba25
         .free          = block_job_free,
26ba25
+        .start         = stream_run,
26ba25
     },
26ba25
-    .start         = stream_run,
26ba25
 };
26ba25
 
26ba25
 void stream_start(const char *job_id, BlockDriverState *bs,
26ba25
@@ -262,7 +262,7 @@ void stream_start(const char *job_id, BlockDriverState *bs,
26ba25
 
26ba25
     s->on_error = on_error;
26ba25
     trace_stream_start(bs, base, s);
26ba25
-    block_job_start(&s->common);
26ba25
+    job_start(&s->common.job);
26ba25
     return;
26ba25
 
26ba25
 fail:
26ba25
diff --git a/blockdev.c b/blockdev.c
26ba25
index 19c04d9..93a4cdf 100644
26ba25
--- a/blockdev.c
26ba25
+++ b/blockdev.c
26ba25
@@ -1911,7 +1911,7 @@ static void drive_backup_commit(BlkActionState *common)
26ba25
     aio_context_acquire(aio_context);
26ba25
 
26ba25
     assert(state->job);
26ba25
-    block_job_start(state->job);
26ba25
+    job_start(&state->job->job);
26ba25
 
26ba25
     aio_context_release(aio_context);
26ba25
 }
26ba25
@@ -2009,7 +2009,7 @@ static void blockdev_backup_commit(BlkActionState *common)
26ba25
     aio_context_acquire(aio_context);
26ba25
 
26ba25
     assert(state->job);
26ba25
-    block_job_start(state->job);
26ba25
+    job_start(&state->job->job);
26ba25
 
26ba25
     aio_context_release(aio_context);
26ba25
 }
26ba25
@@ -3426,7 +3426,7 @@ void qmp_drive_backup(DriveBackup *arg, Error **errp)
26ba25
     BlockJob *job;
26ba25
     job = do_drive_backup(arg, NULL, errp);
26ba25
     if (job) {
26ba25
-        block_job_start(job);
26ba25
+        job_start(&job->job);
26ba25
     }
26ba25
 }
26ba25
 
26ba25
@@ -3514,7 +3514,7 @@ void qmp_blockdev_backup(BlockdevBackup *arg, Error **errp)
26ba25
     BlockJob *job;
26ba25
     job = do_blockdev_backup(arg, NULL, errp);
26ba25
     if (job) {
26ba25
-        block_job_start(job);
26ba25
+        job_start(&job->job);
26ba25
     }
26ba25
 }
26ba25
 
26ba25
diff --git a/blockjob.c b/blockjob.c
26ba25
index 3ede511..313b1ff 100644
26ba25
--- a/blockjob.c
26ba25
+++ b/blockjob.c
26ba25
@@ -36,30 +36,9 @@
26ba25
 #include "qemu/coroutine.h"
26ba25
 #include "qemu/timer.h"
26ba25
 
26ba25
-/* Right now, this mutex is only needed to synchronize accesses to job->busy
26ba25
- * and job->sleep_timer, such as concurrent calls to block_job_do_yield and
26ba25
- * block_job_enter. */
26ba25
-static QemuMutex block_job_mutex;
26ba25
-
26ba25
-static void block_job_lock(void)
26ba25
-{
26ba25
-    qemu_mutex_lock(&block_job_mutex);
26ba25
-}
26ba25
-
26ba25
-static void block_job_unlock(void)
26ba25
-{
26ba25
-    qemu_mutex_unlock(&block_job_mutex);
26ba25
-}
26ba25
-
26ba25
-static void __attribute__((__constructor__)) block_job_init(void)
26ba25
-{
26ba25
-    qemu_mutex_init(&block_job_mutex);
26ba25
-}
26ba25
-
26ba25
 static void block_job_event_cancelled(BlockJob *job);
26ba25
 static void block_job_event_completed(BlockJob *job, const char *msg);
26ba25
 static int block_job_event_pending(BlockJob *job);
26ba25
-static void block_job_enter_cond(BlockJob *job, bool(*fn)(BlockJob *job));
26ba25
 
26ba25
 /* Transactional group of block jobs */
26ba25
 struct BlockJobTxn {
26ba25
@@ -161,33 +140,27 @@ static void block_job_txn_del_job(BlockJob *job)
26ba25
     }
26ba25
 }
26ba25
 
26ba25
-/* Assumes the block_job_mutex is held */
26ba25
-static bool block_job_timer_pending(BlockJob *job)
26ba25
-{
26ba25
-    return timer_pending(&job->sleep_timer);
26ba25
-}
26ba25
-
26ba25
-/* Assumes the block_job_mutex is held */
26ba25
-static bool block_job_timer_not_pending(BlockJob *job)
26ba25
+/* Assumes the job_mutex is held */
26ba25
+static bool job_timer_not_pending(Job *job)
26ba25
 {
26ba25
-    return !block_job_timer_pending(job);
26ba25
+    return !timer_pending(&job->sleep_timer);
26ba25
 }
26ba25
 
26ba25
 static void block_job_pause(BlockJob *job)
26ba25
 {
26ba25
-    job->pause_count++;
26ba25
+    job->job.pause_count++;
26ba25
 }
26ba25
 
26ba25
 static void block_job_resume(BlockJob *job)
26ba25
 {
26ba25
-    assert(job->pause_count > 0);
26ba25
-    job->pause_count--;
26ba25
-    if (job->pause_count) {
26ba25
+    assert(job->job.pause_count > 0);
26ba25
+    job->job.pause_count--;
26ba25
+    if (job->job.pause_count) {
26ba25
         return;
26ba25
     }
26ba25
 
26ba25
     /* kick only if no timer is pending */
26ba25
-    block_job_enter_cond(job, block_job_timer_not_pending);
26ba25
+    job_enter_cond(&job->job, job_timer_not_pending);
26ba25
 }
26ba25
 
26ba25
 static void block_job_attached_aio_context(AioContext *new_context,
26ba25
@@ -208,7 +181,7 @@ void block_job_free(Job *job)
26ba25
                                     block_job_detach_aio_context, bjob);
26ba25
     blk_unref(bjob->blk);
26ba25
     error_free(bjob->blocker);
26ba25
-    assert(!timer_pending(&bjob->sleep_timer));
26ba25
+    assert(!timer_pending(&bjob->job.sleep_timer));
26ba25
 }
26ba25
 
26ba25
 static void block_job_attached_aio_context(AioContext *new_context,
26ba25
@@ -226,7 +199,7 @@ static void block_job_attached_aio_context(AioContext *new_context,
26ba25
 
26ba25
 static void block_job_drain(BlockJob *job)
26ba25
 {
26ba25
-    /* If job is !job->busy this kicks it into the next pause point. */
26ba25
+    /* If job is !job->job.busy this kicks it into the next pause point. */
26ba25
     block_job_enter(job);
26ba25
 
26ba25
     blk_drain(job->blk);
26ba25
@@ -244,7 +217,7 @@ static void block_job_detach_aio_context(void *opaque)
26ba25
 
26ba25
     block_job_pause(job);
26ba25
 
26ba25
-    while (!job->paused && !job->completed) {
26ba25
+    while (!job->job.paused && !job->completed) {
26ba25
         block_job_drain(job);
26ba25
     }
26ba25
 
26ba25
@@ -312,29 +285,11 @@ bool block_job_is_internal(BlockJob *job)
26ba25
     return (job->job.id == NULL);
26ba25
 }
26ba25
 
26ba25
-static bool block_job_started(BlockJob *job)
26ba25
-{
26ba25
-    return job->co;
26ba25
-}
26ba25
-
26ba25
 const BlockJobDriver *block_job_driver(BlockJob *job)
26ba25
 {
26ba25
     return job->driver;
26ba25
 }
26ba25
 
26ba25
-/**
26ba25
- * All jobs must allow a pause point before entering their job proper. This
26ba25
- * ensures that jobs can be paused prior to being started, then resumed later.
26ba25
- */
26ba25
-static void coroutine_fn block_job_co_entry(void *opaque)
26ba25
-{
26ba25
-    BlockJob *job = opaque;
26ba25
-
26ba25
-    assert(job && job->driver && job->driver->start);
26ba25
-    block_job_pause_point(job);
26ba25
-    job->driver->start(job);
26ba25
-}
26ba25
-
26ba25
 static void block_job_sleep_timer_cb(void *opaque)
26ba25
 {
26ba25
     BlockJob *job = opaque;
26ba25
@@ -342,24 +297,12 @@ static void block_job_sleep_timer_cb(void *opaque)
26ba25
     block_job_enter(job);
26ba25
 }
26ba25
 
26ba25
-void block_job_start(BlockJob *job)
26ba25
-{
26ba25
-    assert(job && !block_job_started(job) && job->paused &&
26ba25
-           job->driver && job->driver->start);
26ba25
-    job->co = qemu_coroutine_create(block_job_co_entry, job);
26ba25
-    job->pause_count--;
26ba25
-    job->busy = true;
26ba25
-    job->paused = false;
26ba25
-    job_state_transition(&job->job, JOB_STATUS_RUNNING);
26ba25
-    bdrv_coroutine_enter(blk_bs(job->blk), job->co);
26ba25
-}
26ba25
-
26ba25
 static void block_job_decommission(BlockJob *job)
26ba25
 {
26ba25
     assert(job);
26ba25
     job->completed = true;
26ba25
-    job->busy = false;
26ba25
-    job->paused = false;
26ba25
+    job->job.busy = false;
26ba25
+    job->job.paused = false;
26ba25
     job->job.deferred_to_main_loop = true;
26ba25
     block_job_txn_del_job(job);
26ba25
     job_state_transition(&job->job, JOB_STATUS_NULL);
26ba25
@@ -374,7 +317,7 @@ static void block_job_do_dismiss(BlockJob *job)
26ba25
 static void block_job_conclude(BlockJob *job)
26ba25
 {
26ba25
     job_state_transition(&job->job, JOB_STATUS_CONCLUDED);
26ba25
-    if (job->auto_dismiss || !block_job_started(job)) {
26ba25
+    if (job->auto_dismiss || !job_started(&job->job)) {
26ba25
         block_job_do_dismiss(job);
26ba25
     }
26ba25
 }
26ba25
@@ -439,7 +382,7 @@ static int block_job_finalize_single(BlockJob *job)
26ba25
     }
26ba25
 
26ba25
     /* Emit events only if we actually started */
26ba25
-    if (block_job_started(job)) {
26ba25
+    if (job_started(&job->job)) {
26ba25
         if (job_is_cancelled(&job->job)) {
26ba25
             block_job_event_cancelled(job);
26ba25
         } else {
26ba25
@@ -464,7 +407,7 @@ static void block_job_cancel_async(BlockJob *job, bool force)
26ba25
     if (job->user_paused) {
26ba25
         /* Do not call block_job_enter here, the caller will handle it.  */
26ba25
         job->user_paused = false;
26ba25
-        job->pause_count--;
26ba25
+        job->job.pause_count--;
26ba25
     }
26ba25
     job->job.cancelled = true;
26ba25
     /* To prevent 'force == false' overriding a previous 'force == true' */
26ba25
@@ -615,6 +558,12 @@ static void block_job_completed_txn_success(BlockJob *job)
26ba25
     }
26ba25
 }
26ba25
 
26ba25
+/* Assumes the job_mutex is held */
26ba25
+static bool job_timer_pending(Job *job)
26ba25
+{
26ba25
+    return timer_pending(&job->sleep_timer);
26ba25
+}
26ba25
+
26ba25
 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
26ba25
 {
26ba25
     int64_t old_speed = job->speed;
26ba25
@@ -635,7 +584,7 @@ void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
26ba25
     }
26ba25
 
26ba25
     /* kick only if a timer is pending */
26ba25
-    block_job_enter_cond(job, block_job_timer_pending);
26ba25
+    job_enter_cond(&job->job, job_timer_pending);
26ba25
 }
26ba25
 
26ba25
 int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
26ba25
@@ -654,7 +603,7 @@ void block_job_complete(BlockJob *job, Error **errp)
26ba25
     if (job_apply_verb(&job->job, JOB_VERB_COMPLETE, errp)) {
26ba25
         return;
26ba25
     }
26ba25
-    if (job->pause_count || job_is_cancelled(&job->job) ||
26ba25
+    if (job->job.pause_count || job_is_cancelled(&job->job) ||
26ba25
         !job->driver->complete)
26ba25
     {
26ba25
         error_setg(errp, "The active block job '%s' cannot be completed",
26ba25
@@ -708,7 +657,7 @@ bool block_job_user_paused(BlockJob *job)
26ba25
 void block_job_user_resume(BlockJob *job, Error **errp)
26ba25
 {
26ba25
     assert(job);
26ba25
-    if (!job->user_paused || job->pause_count <= 0) {
26ba25
+    if (!job->user_paused || job->job.pause_count <= 0) {
26ba25
         error_setg(errp, "Can't resume a job that was not paused");
26ba25
         return;
26ba25
     }
26ba25
@@ -727,7 +676,7 @@ void block_job_cancel(BlockJob *job, bool force)
26ba25
         return;
26ba25
     }
26ba25
     block_job_cancel_async(job, force);
26ba25
-    if (!block_job_started(job)) {
26ba25
+    if (!job_started(&job->job)) {
26ba25
         block_job_completed(job, -ECANCELED);
26ba25
     } else if (job->job.deferred_to_main_loop) {
26ba25
         block_job_completed_txn_abort(job);
26ba25
@@ -797,8 +746,8 @@ BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
26ba25
     info->type      = g_strdup(job_type_str(&job->job));
26ba25
     info->device    = g_strdup(job->job.id);
26ba25
     info->len       = job->len;
26ba25
-    info->busy      = atomic_read(&job->busy);
26ba25
-    info->paused    = job->pause_count > 0;
26ba25
+    info->busy      = atomic_read(&job->job.busy);
26ba25
+    info->paused    = job->job.pause_count > 0;
26ba25
     info->offset    = job->offset;
26ba25
     info->speed     = job->speed;
26ba25
     info->io_status = job->iostatus;
26ba25
@@ -915,12 +864,9 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
26ba25
     job->blk           = blk;
26ba25
     job->cb            = cb;
26ba25
     job->opaque        = opaque;
26ba25
-    job->busy          = false;
26ba25
-    job->paused        = true;
26ba25
-    job->pause_count   = 1;
26ba25
     job->auto_finalize = !(flags & BLOCK_JOB_MANUAL_FINALIZE);
26ba25
     job->auto_dismiss  = !(flags & BLOCK_JOB_MANUAL_DISMISS);
26ba25
-    aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
26ba25
+    aio_timer_init(qemu_get_aio_context(), &job->job.sleep_timer,
26ba25
                    QEMU_CLOCK_REALTIME, SCALE_NS,
26ba25
                    block_job_sleep_timer_cb, job);
26ba25
 
26ba25
@@ -980,128 +926,41 @@ void block_job_completed(BlockJob *job, int ret)
26ba25
     }
26ba25
 }
26ba25
 
26ba25
-static bool block_job_should_pause(BlockJob *job)
26ba25
-{
26ba25
-    return job->pause_count > 0;
26ba25
-}
26ba25
-
26ba25
-/* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
26ba25
- * Reentering the job coroutine with block_job_enter() before the timer has
26ba25
- * expired is allowed and cancels the timer.
26ba25
- *
26ba25
- * If @ns is (uint64_t) -1, no timer is scheduled and block_job_enter() must be
26ba25
- * called explicitly. */
26ba25
-static void block_job_do_yield(BlockJob *job, uint64_t ns)
26ba25
-{
26ba25
-    block_job_lock();
26ba25
-    if (ns != -1) {
26ba25
-        timer_mod(&job->sleep_timer, ns);
26ba25
-    }
26ba25
-    job->busy = false;
26ba25
-    block_job_unlock();
26ba25
-    qemu_coroutine_yield();
26ba25
-
26ba25
-    /* Set by block_job_enter before re-entering the coroutine.  */
26ba25
-    assert(job->busy);
26ba25
-}
26ba25
-
26ba25
-void coroutine_fn block_job_pause_point(BlockJob *job)
26ba25
-{
26ba25
-    assert(job && block_job_started(job));
26ba25
-
26ba25
-    if (!block_job_should_pause(job)) {
26ba25
-        return;
26ba25
-    }
26ba25
-    if (job_is_cancelled(&job->job)) {
26ba25
-        return;
26ba25
-    }
26ba25
-
26ba25
-    if (job->driver->pause) {
26ba25
-        job->driver->pause(job);
26ba25
-    }
26ba25
-
26ba25
-    if (block_job_should_pause(job) && !job_is_cancelled(&job->job)) {
26ba25
-        JobStatus status = job->job.status;
26ba25
-        job_state_transition(&job->job, status == JOB_STATUS_READY
26ba25
-                                        ? JOB_STATUS_STANDBY
26ba25
-                                        : JOB_STATUS_PAUSED);
26ba25
-        job->paused = true;
26ba25
-        block_job_do_yield(job, -1);
26ba25
-        job->paused = false;
26ba25
-        job_state_transition(&job->job, status);
26ba25
-    }
26ba25
-
26ba25
-    if (job->driver->resume) {
26ba25
-        job->driver->resume(job);
26ba25
-    }
26ba25
-}
26ba25
-
26ba25
-/*
26ba25
- * Conditionally enter a block_job pending a call to fn() while
26ba25
- * under the block_job_lock critical section.
26ba25
- */
26ba25
-static void block_job_enter_cond(BlockJob *job, bool(*fn)(BlockJob *job))
26ba25
-{
26ba25
-    if (!block_job_started(job)) {
26ba25
-        return;
26ba25
-    }
26ba25
-    if (job->job.deferred_to_main_loop) {
26ba25
-        return;
26ba25
-    }
26ba25
-
26ba25
-    block_job_lock();
26ba25
-    if (job->busy) {
26ba25
-        block_job_unlock();
26ba25
-        return;
26ba25
-    }
26ba25
-
26ba25
-    if (fn && !fn(job)) {
26ba25
-        block_job_unlock();
26ba25
-        return;
26ba25
-    }
26ba25
-
26ba25
-    assert(!job->job.deferred_to_main_loop);
26ba25
-    timer_del(&job->sleep_timer);
26ba25
-    job->busy = true;
26ba25
-    block_job_unlock();
26ba25
-    aio_co_wake(job->co);
26ba25
-}
26ba25
-
26ba25
 void block_job_enter(BlockJob *job)
26ba25
 {
26ba25
-    block_job_enter_cond(job, NULL);
26ba25
+    job_enter_cond(&job->job, NULL);
26ba25
 }
26ba25
 
26ba25
 void block_job_sleep_ns(BlockJob *job, int64_t ns)
26ba25
 {
26ba25
-    assert(job->busy);
26ba25
+    assert(job->job.busy);
26ba25
 
26ba25
     /* Check cancellation *before* setting busy = false, too!  */
26ba25
     if (job_is_cancelled(&job->job)) {
26ba25
         return;
26ba25
     }
26ba25
 
26ba25
-    if (!block_job_should_pause(job)) {
26ba25
-        block_job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
26ba25
+    if (!job_should_pause(&job->job)) {
26ba25
+        job_do_yield(&job->job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
26ba25
     }
26ba25
 
26ba25
-    block_job_pause_point(job);
26ba25
+    job_pause_point(&job->job);
26ba25
 }
26ba25
 
26ba25
 void block_job_yield(BlockJob *job)
26ba25
 {
26ba25
-    assert(job->busy);
26ba25
+    assert(job->job.busy);
26ba25
 
26ba25
     /* Check cancellation *before* setting busy = false, too!  */
26ba25
     if (job_is_cancelled(&job->job)) {
26ba25
         return;
26ba25
     }
26ba25
 
26ba25
-    if (!block_job_should_pause(job)) {
26ba25
-        block_job_do_yield(job, -1);
26ba25
+    if (!job_should_pause(&job->job)) {
26ba25
+        job_do_yield(&job->job, -1);
26ba25
     }
26ba25
 
26ba25
-    block_job_pause_point(job);
26ba25
+    job_pause_point(&job->job);
26ba25
 }
26ba25
 
26ba25
 void block_job_iostatus_reset(BlockJob *job)
26ba25
@@ -1109,7 +968,7 @@ void block_job_iostatus_reset(BlockJob *job)
26ba25
     if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
26ba25
         return;
26ba25
     }
26ba25
-    assert(job->user_paused && job->pause_count > 0);
26ba25
+    assert(job->user_paused && job->job.pause_count > 0);
26ba25
     job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
26ba25
 }
26ba25
 
26ba25
diff --git a/include/block/blockjob.h b/include/block/blockjob.h
26ba25
index 2a9e865..b60d919 100644
26ba25
--- a/include/block/blockjob.h
26ba25
+++ b/include/block/blockjob.h
26ba25
@@ -51,43 +51,18 @@ typedef struct BlockJob {
26ba25
     BlockBackend *blk;
26ba25
 
26ba25
     /**
26ba25
-     * The coroutine that executes the job.  If not NULL, it is
26ba25
-     * reentered when busy is false and the job is cancelled.
26ba25
-     */
26ba25
-    Coroutine *co;
26ba25
-
26ba25
-    /**
26ba25
      * Set to true if the job should abort immediately without waiting
26ba25
      * for data to be in sync.
26ba25
      */
26ba25
     bool force;
26ba25
 
26ba25
     /**
26ba25
-     * Counter for pause request. If non-zero, the block job is either paused,
26ba25
-     * or if busy == true will pause itself as soon as possible.
26ba25
-     */
26ba25
-    int pause_count;
26ba25
-
26ba25
-    /**
26ba25
      * Set to true if the job is paused by user.  Can be unpaused with the
26ba25
      * block-job-resume QMP command.
26ba25
      */
26ba25
     bool user_paused;
26ba25
 
26ba25
     /**
26ba25
-     * Set to false by the job while the coroutine has yielded and may be
26ba25
-     * re-entered by block_job_enter().  There may still be I/O or event loop
26ba25
-     * activity pending.  Accessed under block_job_mutex (in blockjob.c).
26ba25
-     */
26ba25
-    bool busy;
26ba25
-
26ba25
-    /**
26ba25
-     * Set to true by the job while it is in a quiescent state, where
26ba25
-     * no I/O or event loop activity is pending.
26ba25
-     */
26ba25
-    bool paused;
26ba25
-
26ba25
-    /**
26ba25
      * Set to true when the job is ready to be completed.
26ba25
      */
26ba25
     bool ready;
26ba25
@@ -125,12 +100,6 @@ typedef struct BlockJob {
26ba25
     /** ret code passed to block_job_completed. */
26ba25
     int ret;
26ba25
 
26ba25
-    /**
26ba25
-     * Timer that is used by @block_job_sleep_ns. Accessed under
26ba25
-     * block_job_mutex (in blockjob.c).
26ba25
-     */
26ba25
-    QEMUTimer sleep_timer;
26ba25
-
26ba25
     /** True if this job should automatically finalize itself */
26ba25
     bool auto_finalize;
26ba25
 
26ba25
@@ -208,15 +177,6 @@ void block_job_remove_all_bdrv(BlockJob *job);
26ba25
 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp);
26ba25
 
26ba25
 /**
26ba25
- * block_job_start:
26ba25
- * @job: A job that has not yet been started.
26ba25
- *
26ba25
- * Begins execution of a block job.
26ba25
- * Takes ownership of one reference to the job object.
26ba25
- */
26ba25
-void block_job_start(BlockJob *job);
26ba25
-
26ba25
-/**
26ba25
  * block_job_cancel:
26ba25
  * @job: The job to be canceled.
26ba25
  * @force: Quit a job without waiting for data to be in sync.
26ba25
diff --git a/include/block/blockjob_int.h b/include/block/blockjob_int.h
26ba25
index 0c2f8de..0a614a8 100644
26ba25
--- a/include/block/blockjob_int.h
26ba25
+++ b/include/block/blockjob_int.h
26ba25
@@ -38,9 +38,6 @@ struct BlockJobDriver {
26ba25
     /** Generic JobDriver callbacks and settings */
26ba25
     JobDriver job_driver;
26ba25
 
26ba25
-    /** Mandatory: Entrypoint for the Coroutine. */
26ba25
-    CoroutineEntry *start;
26ba25
-
26ba25
     /**
26ba25
      * Optional callback for job types whose completion must be triggered
26ba25
      * manually.
26ba25
@@ -85,20 +82,6 @@ struct BlockJobDriver {
26ba25
      */
26ba25
     void (*clean)(BlockJob *job);
26ba25
 
26ba25
-    /**
26ba25
-     * If the callback is not NULL, it will be invoked when the job transitions
26ba25
-     * into the paused state.  Paused jobs must not perform any asynchronous
26ba25
-     * I/O or event loop activity.  This callback is used to quiesce jobs.
26ba25
-     */
26ba25
-    void coroutine_fn (*pause)(BlockJob *job);
26ba25
-
26ba25
-    /**
26ba25
-     * If the callback is not NULL, it will be invoked when the job transitions
26ba25
-     * out of the paused state.  Any asynchronous I/O or event loop activity
26ba25
-     * should be restarted from this callback.
26ba25
-     */
26ba25
-    void coroutine_fn (*resume)(BlockJob *job);
26ba25
-
26ba25
     /*
26ba25
      * If the callback is not NULL, it will be invoked before the job is
26ba25
      * resumed in a new AioContext.  This is the place to move any resources
26ba25
@@ -196,15 +179,6 @@ void block_job_early_fail(BlockJob *job);
26ba25
 void block_job_completed(BlockJob *job, int ret);
26ba25
 
26ba25
 /**
26ba25
- * block_job_pause_point:
26ba25
- * @job: The job that is ready to pause.
26ba25
- *
26ba25
- * Pause now if block_job_pause() has been called.  Block jobs that perform
26ba25
- * lots of I/O must call this between requests so that the job can be paused.
26ba25
- */
26ba25
-void coroutine_fn block_job_pause_point(BlockJob *job);
26ba25
-
26ba25
-/**
26ba25
  * block_job_enter:
26ba25
  * @job: The job to enter.
26ba25
  *
26ba25
diff --git a/include/qemu/job.h b/include/qemu/job.h
26ba25
index 933e0ab..9dcff12 100644
26ba25
--- a/include/qemu/job.h
26ba25
+++ b/include/qemu/job.h
26ba25
@@ -28,6 +28,7 @@
26ba25
 
26ba25
 #include "qapi/qapi-types-block-core.h"
26ba25
 #include "qemu/queue.h"
26ba25
+#include "qemu/coroutine.h"
26ba25
 
26ba25
 typedef struct JobDriver JobDriver;
26ba25
 
26ba25
@@ -51,6 +52,37 @@ typedef struct Job {
26ba25
     AioContext *aio_context;
26ba25
 
26ba25
     /**
26ba25
+     * The coroutine that executes the job.  If not NULL, it is reentered when
26ba25
+     * busy is false and the job is cancelled.
26ba25
+     */
26ba25
+    Coroutine *co;
26ba25
+
26ba25
+    /**
26ba25
+     * Timer that is used by @block_job_sleep_ns. Accessed under job_mutex (in
26ba25
+     * job.c).
26ba25
+     */
26ba25
+    QEMUTimer sleep_timer;
26ba25
+
26ba25
+    /**
26ba25
+     * Counter for pause request. If non-zero, the block job is either paused,
26ba25
+     * or if busy == true will pause itself as soon as possible.
26ba25
+     */
26ba25
+    int pause_count;
26ba25
+
26ba25
+    /**
26ba25
+     * Set to false by the job while the coroutine has yielded and may be
26ba25
+     * re-entered by block_job_enter().  There may still be I/O or event loop
26ba25
+     * activity pending.  Accessed under block_job_mutex (in blockjob.c).
26ba25
+     */
26ba25
+    bool busy;
26ba25
+
26ba25
+    /**
26ba25
+     * Set to true by the job while it is in a quiescent state, where
26ba25
+     * no I/O or event loop activity is pending.
26ba25
+     */
26ba25
+    bool paused;
26ba25
+
26ba25
+    /**
26ba25
      * Set to true if the job should cancel itself.  The flag must
26ba25
      * always be tested just before toggling the busy flag from false
26ba25
      * to true.  After a job has been cancelled, it should only yield
26ba25
@@ -75,6 +107,23 @@ struct JobDriver {
26ba25
     /** Enum describing the operation */
26ba25
     JobType job_type;
26ba25
 
26ba25
+    /** Mandatory: Entrypoint for the Coroutine. */
26ba25
+    CoroutineEntry *start;
26ba25
+
26ba25
+    /**
26ba25
+     * If the callback is not NULL, it will be invoked when the job transitions
26ba25
+     * into the paused state.  Paused jobs must not perform any asynchronous
26ba25
+     * I/O or event loop activity.  This callback is used to quiesce jobs.
26ba25
+     */
26ba25
+    void coroutine_fn (*pause)(Job *job);
26ba25
+
26ba25
+    /**
26ba25
+     * If the callback is not NULL, it will be invoked when the job transitions
26ba25
+     * out of the paused state.  Any asynchronous I/O or event loop activity
26ba25
+     * should be restarted from this callback.
26ba25
+     */
26ba25
+    void coroutine_fn (*resume)(Job *job);
26ba25
+
26ba25
     /** Called when the job is freed */
26ba25
     void (*free)(Job *job);
26ba25
 };
26ba25
@@ -103,6 +152,30 @@ void job_ref(Job *job);
26ba25
  */
26ba25
 void job_unref(Job *job);
26ba25
 
26ba25
+/**
26ba25
+ * Conditionally enter the job coroutine if the job is ready to run, not
26ba25
+ * already busy and fn() returns true. fn() is called while under the job_lock
26ba25
+ * critical section.
26ba25
+ */
26ba25
+void job_enter_cond(Job *job, bool(*fn)(Job *job));
26ba25
+
26ba25
+/**
26ba25
+ * @job: A job that has not yet been started.
26ba25
+ *
26ba25
+ * Begins execution of a job.
26ba25
+ * Takes ownership of one reference to the job object.
26ba25
+ */
26ba25
+void job_start(Job *job);
26ba25
+
26ba25
+/**
26ba25
+ * @job: The job that is ready to pause.
26ba25
+ *
26ba25
+ * Pause now if job_pause() has been called. Jobs that perform lots of I/O
26ba25
+ * must call this between requests so that the job can be paused.
26ba25
+ */
26ba25
+void coroutine_fn job_pause_point(Job *job);
26ba25
+
26ba25
+
26ba25
 /** Returns the JobType of a given Job. */
26ba25
 JobType job_type(const Job *job);
26ba25
 
26ba25
@@ -153,5 +226,8 @@ void job_defer_to_main_loop(Job *job, JobDeferToMainLoopFn *fn, void *opaque);
26ba25
 
26ba25
 /* TODO To be removed from the public interface */
26ba25
 void job_state_transition(Job *job, JobStatus s1);
26ba25
+void coroutine_fn job_do_yield(Job *job, uint64_t ns);
26ba25
+bool job_should_pause(Job *job);
26ba25
+bool job_started(Job *job);
26ba25
 
26ba25
 #endif
26ba25
diff --git a/job.c b/job.c
26ba25
index c5a37fb..78497fd 100644
26ba25
--- a/job.c
26ba25
+++ b/job.c
26ba25
@@ -60,6 +60,26 @@ bool JobVerbTable[JOB_VERB__MAX][JOB_STATUS__MAX] = {
26ba25
     [JOB_VERB_DISMISS]              = {0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0},
26ba25
 };
26ba25
 
26ba25
+/* Right now, this mutex is only needed to synchronize accesses to job->busy
26ba25
+ * and job->sleep_timer, such as concurrent calls to job_do_yield and
26ba25
+ * job_enter. */
26ba25
+static QemuMutex job_mutex;
26ba25
+
26ba25
+static void job_lock(void)
26ba25
+{
26ba25
+    qemu_mutex_lock(&job_mutex);
26ba25
+}
26ba25
+
26ba25
+static void job_unlock(void)
26ba25
+{
26ba25
+    qemu_mutex_unlock(&job_mutex);
26ba25
+}
26ba25
+
26ba25
+static void __attribute__((__constructor__)) job_init(void)
26ba25
+{
26ba25
+    qemu_mutex_init(&job_mutex);
26ba25
+}
26ba25
+
26ba25
 /* TODO Make static once the whole state machine is in job.c */
26ba25
 void job_state_transition(Job *job, JobStatus s1)
26ba25
 {
26ba25
@@ -101,6 +121,16 @@ bool job_is_cancelled(Job *job)
26ba25
     return job->cancelled;
26ba25
 }
26ba25
 
26ba25
+bool job_started(Job *job)
26ba25
+{
26ba25
+    return job->co;
26ba25
+}
26ba25
+
26ba25
+bool job_should_pause(Job *job)
26ba25
+{
26ba25
+    return job->pause_count > 0;
26ba25
+}
26ba25
+
26ba25
 Job *job_next(Job *job)
26ba25
 {
26ba25
     if (!job) {
26ba25
@@ -143,6 +173,9 @@ void *job_create(const char *job_id, const JobDriver *driver, AioContext *ctx,
26ba25
     job->id            = g_strdup(job_id);
26ba25
     job->refcnt        = 1;
26ba25
     job->aio_context   = ctx;
26ba25
+    job->busy          = false;
26ba25
+    job->paused        = true;
26ba25
+    job->pause_count   = 1;
26ba25
 
26ba25
     job_state_transition(job, JOB_STATUS_CREATED);
26ba25
 
26ba25
@@ -172,6 +205,110 @@ void job_unref(Job *job)
26ba25
     }
26ba25
 }
26ba25
 
26ba25
+void job_enter_cond(Job *job, bool(*fn)(Job *job))
26ba25
+{
26ba25
+    if (!job_started(job)) {
26ba25
+        return;
26ba25
+    }
26ba25
+    if (job->deferred_to_main_loop) {
26ba25
+        return;
26ba25
+    }
26ba25
+
26ba25
+    job_lock();
26ba25
+    if (job->busy) {
26ba25
+        job_unlock();
26ba25
+        return;
26ba25
+    }
26ba25
+
26ba25
+    if (fn && !fn(job)) {
26ba25
+        job_unlock();
26ba25
+        return;
26ba25
+    }
26ba25
+
26ba25
+    assert(!job->deferred_to_main_loop);
26ba25
+    timer_del(&job->sleep_timer);
26ba25
+    job->busy = true;
26ba25
+    job_unlock();
26ba25
+    aio_co_wake(job->co);
26ba25
+}
26ba25
+
26ba25
+/* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
26ba25
+ * Reentering the job coroutine with block_job_enter() before the timer has
26ba25
+ * expired is allowed and cancels the timer.
26ba25
+ *
26ba25
+ * If @ns is (uint64_t) -1, no timer is scheduled and block_job_enter() must be
26ba25
+ * called explicitly. */
26ba25
+void coroutine_fn job_do_yield(Job *job, uint64_t ns)
26ba25
+{
26ba25
+    job_lock();
26ba25
+    if (ns != -1) {
26ba25
+        timer_mod(&job->sleep_timer, ns);
26ba25
+    }
26ba25
+    job->busy = false;
26ba25
+    job_unlock();
26ba25
+    qemu_coroutine_yield();
26ba25
+
26ba25
+    /* Set by job_enter_cond() before re-entering the coroutine.  */
26ba25
+    assert(job->busy);
26ba25
+}
26ba25
+
26ba25
+void coroutine_fn job_pause_point(Job *job)
26ba25
+{
26ba25
+    assert(job && job_started(job));
26ba25
+
26ba25
+    if (!job_should_pause(job)) {
26ba25
+        return;
26ba25
+    }
26ba25
+    if (job_is_cancelled(job)) {
26ba25
+        return;
26ba25
+    }
26ba25
+
26ba25
+    if (job->driver->pause) {
26ba25
+        job->driver->pause(job);
26ba25
+    }
26ba25
+
26ba25
+    if (job_should_pause(job) && !job_is_cancelled(job)) {
26ba25
+        JobStatus status = job->status;
26ba25
+        job_state_transition(job, status == JOB_STATUS_READY
26ba25
+                                  ? JOB_STATUS_STANDBY
26ba25
+                                  : JOB_STATUS_PAUSED);
26ba25
+        job->paused = true;
26ba25
+        job_do_yield(job, -1);
26ba25
+        job->paused = false;
26ba25
+        job_state_transition(job, status);
26ba25
+    }
26ba25
+
26ba25
+    if (job->driver->resume) {
26ba25
+        job->driver->resume(job);
26ba25
+    }
26ba25
+}
26ba25
+
26ba25
+/**
26ba25
+ * All jobs must allow a pause point before entering their job proper. This
26ba25
+ * ensures that jobs can be paused prior to being started, then resumed later.
26ba25
+ */
26ba25
+static void coroutine_fn job_co_entry(void *opaque)
26ba25
+{
26ba25
+    Job *job = opaque;
26ba25
+
26ba25
+    assert(job && job->driver && job->driver->start);
26ba25
+    job_pause_point(job);
26ba25
+    job->driver->start(job);
26ba25
+}
26ba25
+
26ba25
+
26ba25
+void job_start(Job *job)
26ba25
+{
26ba25
+    assert(job && !job_started(job) && job->paused &&
26ba25
+           job->driver && job->driver->start);
26ba25
+    job->co = qemu_coroutine_create(job_co_entry, job);
26ba25
+    job->pause_count--;
26ba25
+    job->busy = true;
26ba25
+    job->paused = false;
26ba25
+    job_state_transition(job, JOB_STATUS_RUNNING);
26ba25
+    aio_co_enter(job->aio_context, job->co);
26ba25
+}
26ba25
+
26ba25
 typedef struct {
26ba25
     Job *job;
26ba25
     JobDeferToMainLoopFn *fn;
26ba25
diff --git a/tests/test-bdrv-drain.c b/tests/test-bdrv-drain.c
26ba25
index 4f8cba8..c9f2f9b 100644
26ba25
--- a/tests/test-bdrv-drain.c
26ba25
+++ b/tests/test-bdrv-drain.c
26ba25
@@ -524,8 +524,8 @@ BlockJobDriver test_job_driver = {
26ba25
     .job_driver = {
26ba25
         .instance_size  = sizeof(TestBlockJob),
26ba25
         .free           = block_job_free,
26ba25
+        .start          = test_job_start,
26ba25
     },
26ba25
-    .start          = test_job_start,
26ba25
     .complete       = test_job_complete,
26ba25
 };
26ba25
 
26ba25
@@ -549,47 +549,47 @@ static void test_blockjob_common(enum drain_type drain_type)
26ba25
     job = block_job_create("job0", &test_job_driver, NULL, src, 0, BLK_PERM_ALL,
26ba25
                            0, 0, NULL, NULL, &error_abort);
26ba25
     block_job_add_bdrv(job, "target", target, 0, BLK_PERM_ALL, &error_abort);
26ba25
-    block_job_start(job);
26ba25
+    job_start(&job->job);
26ba25
 
26ba25
-    g_assert_cmpint(job->pause_count, ==, 0);
26ba25
-    g_assert_false(job->paused);
26ba25
-    g_assert_false(job->busy); /* We're in block_job_sleep_ns() */
26ba25
+    g_assert_cmpint(job->job.pause_count, ==, 0);
26ba25
+    g_assert_false(job->job.paused);
26ba25
+    g_assert_false(job->job.busy); /* We're in block_job_sleep_ns() */
26ba25
 
26ba25
     do_drain_begin(drain_type, src);
26ba25
 
26ba25
     if (drain_type == BDRV_DRAIN_ALL) {
26ba25
         /* bdrv_drain_all() drains both src and target */
26ba25
-        g_assert_cmpint(job->pause_count, ==, 2);
26ba25
+        g_assert_cmpint(job->job.pause_count, ==, 2);
26ba25
     } else {
26ba25
-        g_assert_cmpint(job->pause_count, ==, 1);
26ba25
+        g_assert_cmpint(job->job.pause_count, ==, 1);
26ba25
     }
26ba25
     /* XXX We don't wait until the job is actually paused. Is this okay? */
26ba25
-    /* g_assert_true(job->paused); */
26ba25
-    g_assert_false(job->busy); /* The job is paused */
26ba25
+    /* g_assert_true(job->job.paused); */
26ba25
+    g_assert_false(job->job.busy); /* The job is paused */
26ba25
 
26ba25
     do_drain_end(drain_type, src);
26ba25
 
26ba25
-    g_assert_cmpint(job->pause_count, ==, 0);
26ba25
-    g_assert_false(job->paused);
26ba25
-    g_assert_false(job->busy); /* We're in block_job_sleep_ns() */
26ba25
+    g_assert_cmpint(job->job.pause_count, ==, 0);
26ba25
+    g_assert_false(job->job.paused);
26ba25
+    g_assert_false(job->job.busy); /* We're in block_job_sleep_ns() */
26ba25
 
26ba25
     do_drain_begin(drain_type, target);
26ba25
 
26ba25
     if (drain_type == BDRV_DRAIN_ALL) {
26ba25
         /* bdrv_drain_all() drains both src and target */
26ba25
-        g_assert_cmpint(job->pause_count, ==, 2);
26ba25
+        g_assert_cmpint(job->job.pause_count, ==, 2);
26ba25
     } else {
26ba25
-        g_assert_cmpint(job->pause_count, ==, 1);
26ba25
+        g_assert_cmpint(job->job.pause_count, ==, 1);
26ba25
     }
26ba25
     /* XXX We don't wait until the job is actually paused. Is this okay? */
26ba25
-    /* g_assert_true(job->paused); */
26ba25
-    g_assert_false(job->busy); /* The job is paused */
26ba25
+    /* g_assert_true(job->job.paused); */
26ba25
+    g_assert_false(job->job.busy); /* The job is paused */
26ba25
 
26ba25
     do_drain_end(drain_type, target);
26ba25
 
26ba25
-    g_assert_cmpint(job->pause_count, ==, 0);
26ba25
-    g_assert_false(job->paused);
26ba25
-    g_assert_false(job->busy); /* We're in block_job_sleep_ns() */
26ba25
+    g_assert_cmpint(job->job.pause_count, ==, 0);
26ba25
+    g_assert_false(job->job.paused);
26ba25
+    g_assert_false(job->job.busy); /* We're in block_job_sleep_ns() */
26ba25
 
26ba25
     ret = block_job_complete_sync(job, &error_abort);
26ba25
     g_assert_cmpint(ret, ==, 0);
26ba25
diff --git a/tests/test-blockjob-txn.c b/tests/test-blockjob-txn.c
26ba25
index c03f966..323e154 100644
26ba25
--- a/tests/test-blockjob-txn.c
26ba25
+++ b/tests/test-blockjob-txn.c
26ba25
@@ -78,8 +78,8 @@ static const BlockJobDriver test_block_job_driver = {
26ba25
     .job_driver = {
26ba25
         .instance_size = sizeof(TestBlockJob),
26ba25
         .free          = block_job_free,
26ba25
+        .start         = test_block_job_run,
26ba25
     },
26ba25
-    .start = test_block_job_run,
26ba25
 };
26ba25
 
26ba25
 /* Create a block job that completes with a given return code after a given
26ba25
@@ -125,7 +125,7 @@ static void test_single_job(int expected)
26ba25
 
26ba25
     txn = block_job_txn_new();
26ba25
     job = test_block_job_start(1, true, expected, &result, txn);
26ba25
-    block_job_start(job);
26ba25
+    job_start(&job->job);
26ba25
 
26ba25
     if (expected == -ECANCELED) {
26ba25
         block_job_cancel(job, false);
26ba25
@@ -165,8 +165,8 @@ static void test_pair_jobs(int expected1, int expected2)
26ba25
     txn = block_job_txn_new();
26ba25
     job1 = test_block_job_start(1, true, expected1, &result1, txn);
26ba25
     job2 = test_block_job_start(2, true, expected2, &result2, txn);
26ba25
-    block_job_start(job1);
26ba25
-    block_job_start(job2);
26ba25
+    job_start(&job1->job);
26ba25
+    job_start(&job2->job);
26ba25
 
26ba25
     /* Release our reference now to trigger as many nice
26ba25
      * use-after-free bugs as possible.
26ba25
@@ -227,8 +227,8 @@ static void test_pair_jobs_fail_cancel_race(void)
26ba25
     txn = block_job_txn_new();
26ba25
     job1 = test_block_job_start(1, true, -ECANCELED, &result1, txn);
26ba25
     job2 = test_block_job_start(2, false, 0, &result2, txn);
26ba25
-    block_job_start(job1);
26ba25
-    block_job_start(job2);
26ba25
+    job_start(&job1->job);
26ba25
+    job_start(&job2->job);
26ba25
 
26ba25
     block_job_cancel(job1, false);
26ba25
 
26ba25
diff --git a/tests/test-blockjob.c b/tests/test-blockjob.c
26ba25
index 5f43bd7..1d18325 100644
26ba25
--- a/tests/test-blockjob.c
26ba25
+++ b/tests/test-blockjob.c
26ba25
@@ -199,8 +199,8 @@ static const BlockJobDriver test_cancel_driver = {
26ba25
     .job_driver = {
26ba25
         .instance_size = sizeof(CancelJob),
26ba25
         .free          = block_job_free,
26ba25
+        .start         = cancel_job_start,
26ba25
     },
26ba25
-    .start         = cancel_job_start,
26ba25
     .complete      = cancel_job_complete,
26ba25
 };
26ba25
 
26ba25
@@ -254,7 +254,7 @@ static void test_cancel_running(void)
26ba25
 
26ba25
     s = create_common(&job;;
26ba25
 
26ba25
-    block_job_start(job);
26ba25
+    job_start(&job->job);
26ba25
     assert(job->job.status == JOB_STATUS_RUNNING);
26ba25
 
26ba25
     cancel_common(s);
26ba25
@@ -267,7 +267,7 @@ static void test_cancel_paused(void)
26ba25
 
26ba25
     s = create_common(&job;;
26ba25
 
26ba25
-    block_job_start(job);
26ba25
+    job_start(&job->job);
26ba25
     assert(job->job.status == JOB_STATUS_RUNNING);
26ba25
 
26ba25
     block_job_user_pause(job, &error_abort);
26ba25
@@ -284,7 +284,7 @@ static void test_cancel_ready(void)
26ba25
 
26ba25
     s = create_common(&job;;
26ba25
 
26ba25
-    block_job_start(job);
26ba25
+    job_start(&job->job);
26ba25
     assert(job->job.status == JOB_STATUS_RUNNING);
26ba25
 
26ba25
     s->should_converge = true;
26ba25
@@ -301,7 +301,7 @@ static void test_cancel_standby(void)
26ba25
 
26ba25
     s = create_common(&job;;
26ba25
 
26ba25
-    block_job_start(job);
26ba25
+    job_start(&job->job);
26ba25
     assert(job->job.status == JOB_STATUS_RUNNING);
26ba25
 
26ba25
     s->should_converge = true;
26ba25
@@ -322,7 +322,7 @@ static void test_cancel_pending(void)
26ba25
 
26ba25
     s = create_common(&job;;
26ba25
 
26ba25
-    block_job_start(job);
26ba25
+    job_start(&job->job);
26ba25
     assert(job->job.status == JOB_STATUS_RUNNING);
26ba25
 
26ba25
     s->should_converge = true;
26ba25
@@ -346,7 +346,7 @@ static void test_cancel_concluded(void)
26ba25
 
26ba25
     s = create_common(&job;;
26ba25
 
26ba25
-    block_job_start(job);
26ba25
+    job_start(&job->job);
26ba25
     assert(job->job.status == JOB_STATUS_RUNNING);
26ba25
 
26ba25
     s->should_converge = true;
26ba25
-- 
26ba25
1.8.3.1
26ba25