1 files changed, 39 insertions, 180 deletions

diff --git a/blockjob.c b/blockjob.c
index 3ede511da0..313b1ff7ce 100644
--- a/blockjob.c
+++ b/blockjob.c
@@ -36,30 +36,9 @@
 #include "qemu/coroutine.h"
 #include "qemu/timer.h"
 
-/* Right now, this mutex is only needed to synchronize accesses to job->busy
- * and job->sleep_timer, such as concurrent calls to block_job_do_yield and
- * block_job_enter. */
-static QemuMutex block_job_mutex;
-
-static void block_job_lock(void)
-{
-    qemu_mutex_lock(&block_job_mutex);
-}
-
-static void block_job_unlock(void)
-{
-    qemu_mutex_unlock(&block_job_mutex);
-}
-
-static void __attribute__((__constructor__)) block_job_init(void)
-{
-    qemu_mutex_init(&block_job_mutex);
-}
-
 static void block_job_event_cancelled(BlockJob *job);
 static void block_job_event_completed(BlockJob *job, const char *msg);
 static int block_job_event_pending(BlockJob *job);
-static void block_job_enter_cond(BlockJob *job, bool(*fn)(BlockJob *job));
 
 /* Transactional group of block jobs */
 struct BlockJobTxn {
@@ -161,33 +140,27 @@ static void block_job_txn_del_job(BlockJob *job)
     }
 }
 
-/* Assumes the block_job_mutex is held */
-static bool block_job_timer_pending(BlockJob *job)
-{
-    return timer_pending(&job->sleep_timer);
-}
-
-/* Assumes the block_job_mutex is held */
-static bool block_job_timer_not_pending(BlockJob *job)
+/* Assumes the job_mutex is held */
+static bool job_timer_not_pending(Job *job)
 {
-    return !block_job_timer_pending(job);
+    return !timer_pending(&job->sleep_timer);
 }
 
 static void block_job_pause(BlockJob *job)
 {
-    job->pause_count++;
+    job->job.pause_count++;
 }
 
 static void block_job_resume(BlockJob *job)
 {
-    assert(job->pause_count > 0);
-    job->pause_count--;
-    if (job->pause_count) {
+    assert(job->job.pause_count > 0);
+    job->job.pause_count--;
+    if (job->job.pause_count) {
         return;
     }
 
     /* kick only if no timer is pending */
-    block_job_enter_cond(job, block_job_timer_not_pending);
+    job_enter_cond(&job->job, job_timer_not_pending);
 }
 
 static void block_job_attached_aio_context(AioContext *new_context,
@@ -208,7 +181,7 @@ void block_job_free(Job *job)
                                     block_job_detach_aio_context, bjob);
     blk_unref(bjob->blk);
     error_free(bjob->blocker);
-    assert(!timer_pending(&bjob->sleep_timer));
+    assert(!timer_pending(&bjob->job.sleep_timer));
 }
 
 static void block_job_attached_aio_context(AioContext *new_context,
@@ -226,7 +199,7 @@ static void block_job_attached_aio_context(AioContext *new_context,
 
 static void block_job_drain(BlockJob *job)
 {
-    /* If job is !job->busy this kicks it into the next pause point. */
+    /* If job is !job->job.busy this kicks it into the next pause point. */
     block_job_enter(job);
 
     blk_drain(job->blk);
@@ -244,7 +217,7 @@ static void block_job_detach_aio_context(void *opaque)
 
     block_job_pause(job);
 
-    while (!job->paused && !job->completed) {
+    while (!job->job.paused && !job->completed) {
         block_job_drain(job);
     }
 
@@ -312,29 +285,11 @@ bool block_job_is_internal(BlockJob *job)
     return (job->job.id == NULL);
 }
 
-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.
- */
-static void coroutine_fn block_job_co_entry(void *opaque)
-{
-    BlockJob *job = opaque;
-
-    assert(job && job->driver && job->driver->start);
-    block_job_pause_point(job);
-    job->driver->start(job);
-}
-
 static void block_job_sleep_timer_cb(void *opaque)
 {
     BlockJob *job = opaque;
@@ -342,24 +297,12 @@ static void block_job_sleep_timer_cb(void *opaque)
     block_job_enter(job);
 }
 
-void block_job_start(BlockJob *job)
-{
-    assert(job && !block_job_started(job) && job->paused &&
-           job->driver && job->driver->start);
-    job->co = qemu_coroutine_create(block_job_co_entry, job);
-    job->pause_count--;
-    job->busy = true;
-    job->paused = false;
-    job_state_transition(&job->job, JOB_STATUS_RUNNING);
-    bdrv_coroutine_enter(blk_bs(job->blk), job->co);
-}
-
 static void block_job_decommission(BlockJob *job)
 {
     assert(job);
     job->completed = true;
-    job->busy = false;
-    job->paused = false;
+    job->job.busy = false;
+    job->job.paused = false;
     job->job.deferred_to_main_loop = true;
     block_job_txn_del_job(job);
     job_state_transition(&job->job, JOB_STATUS_NULL);
@@ -374,7 +317,7 @@ static void block_job_do_dismiss(BlockJob *job)
 static void block_job_conclude(BlockJob *job)
 {
     job_state_transition(&job->job, JOB_STATUS_CONCLUDED);
-    if (job->auto_dismiss || !block_job_started(job)) {
+    if (job->auto_dismiss || !job_started(&job->job)) {
         block_job_do_dismiss(job);
     }
 }
@@ -439,7 +382,7 @@ static int block_job_finalize_single(BlockJob *job)
     }
 
     /* Emit events only if we actually started */
-    if (block_job_started(job)) {
+    if (job_started(&job->job)) {
         if (job_is_cancelled(&job->job)) {
             block_job_event_cancelled(job);
         } else {
@@ -464,7 +407,7 @@ static void block_job_cancel_async(BlockJob *job, bool force)
     if (job->user_paused) {
         /* Do not call block_job_enter here, the caller will handle it.  */
         job->user_paused = false;
-        job->pause_count--;
+        job->job.pause_count--;
     }
     job->job.cancelled = true;
     /* To prevent 'force == false' overriding a previous 'force == true' */
@@ -615,6 +558,12 @@ static void block_job_completed_txn_success(BlockJob *job)
     }
 }
 
+/* Assumes the job_mutex is held */
+static bool job_timer_pending(Job *job)
+{
+    return timer_pending(&job->sleep_timer);
+}
+
 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
 {
     int64_t old_speed = job->speed;
@@ -635,7 +584,7 @@ void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
     }
 
     /* kick only if a timer is pending */
-    block_job_enter_cond(job, block_job_timer_pending);
+    job_enter_cond(&job->job, job_timer_pending);
 }
 
 int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
@@ -654,7 +603,7 @@ void block_job_complete(BlockJob *job, Error **errp)
     if (job_apply_verb(&job->job, JOB_VERB_COMPLETE, errp)) {
         return;
     }
-    if (job->pause_count || job_is_cancelled(&job->job) ||
+    if (job->job.pause_count || job_is_cancelled(&job->job) ||
         !job->driver->complete)
     {
         error_setg(errp, "The active block job '%s' cannot be completed",
@@ -708,7 +657,7 @@ bool block_job_user_paused(BlockJob *job)
 void block_job_user_resume(BlockJob *job, Error **errp)
 {
     assert(job);
-    if (!job->user_paused || job->pause_count <= 0) {
+    if (!job->user_paused || job->job.pause_count <= 0) {
         error_setg(errp, "Can't resume a job that was not paused");
         return;
     }
@@ -727,7 +676,7 @@ void block_job_cancel(BlockJob *job, bool force)
         return;
     }
     block_job_cancel_async(job, force);
-    if (!block_job_started(job)) {
+    if (!job_started(&job->job)) {
         block_job_completed(job, -ECANCELED);
     } else if (job->job.deferred_to_main_loop) {
         block_job_completed_txn_abort(job);
@@ -797,8 +746,8 @@ BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
     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);
-    info->paused    = job->pause_count > 0;
+    info->busy      = atomic_read(&job->job.busy);
+    info->paused    = job->job.pause_count > 0;
     info->offset    = job->offset;
     info->speed     = job->speed;
     info->io_status = job->iostatus;
@@ -915,12 +864,9 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
     job->blk           = blk;
     job->cb            = cb;
     job->opaque        = opaque;
-    job->busy          = false;
-    job->paused        = true;
-    job->pause_count   = 1;
     job->auto_finalize = !(flags & BLOCK_JOB_MANUAL_FINALIZE);
     job->auto_dismiss  = !(flags & BLOCK_JOB_MANUAL_DISMISS);
-    aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
+    aio_timer_init(qemu_get_aio_context(), &job->job.sleep_timer,
                    QEMU_CLOCK_REALTIME, SCALE_NS,
                    block_job_sleep_timer_cb, job);
 
@@ -980,128 +926,41 @@ void block_job_completed(BlockJob *job, int ret)
     }
 }
 
-static bool block_job_should_pause(BlockJob *job)
-{
-    return job->pause_count > 0;
-}
-
-/* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
- * Reentering the job coroutine with block_job_enter() before the timer has
- * expired is allowed and cancels the timer.
- *
- * If @ns is (uint64_t) -1, no timer is scheduled and block_job_enter() must be
- * called explicitly. */
-static void block_job_do_yield(BlockJob *job, uint64_t ns)
-{
-    block_job_lock();
-    if (ns != -1) {
-        timer_mod(&job->sleep_timer, ns);
-    }
-    job->busy = false;
-    block_job_unlock();
-    qemu_coroutine_yield();
-
-    /* Set by block_job_enter before re-entering the coroutine.  */
-    assert(job->busy);
-}
-
-void coroutine_fn block_job_pause_point(BlockJob *job)
-{
-    assert(job && block_job_started(job));
-
-    if (!block_job_should_pause(job)) {
-        return;
-    }
-    if (job_is_cancelled(&job->job)) {
-        return;
-    }
-
-    if (job->driver->pause) {
-        job->driver->pause(job);
-    }
-
-    if (block_job_should_pause(job) && !job_is_cancelled(&job->job)) {
-        JobStatus status = job->job.status;
-        job_state_transition(&job->job, status == JOB_STATUS_READY
-                                        ? JOB_STATUS_STANDBY
-                                        : JOB_STATUS_PAUSED);
-        job->paused = true;
-        block_job_do_yield(job, -1);
-        job->paused = false;
-        job_state_transition(&job->job, status);
-    }
-
-    if (job->driver->resume) {
-        job->driver->resume(job);
-    }
-}
-
-/*
- * Conditionally enter a block_job pending a call to fn() while
- * under the block_job_lock critical section.
- */
-static void block_job_enter_cond(BlockJob *job, bool(*fn)(BlockJob *job))
-{
-    if (!block_job_started(job)) {
-        return;
-    }
-    if (job->job.deferred_to_main_loop) {
-        return;
-    }
-
-    block_job_lock();
-    if (job->busy) {
-        block_job_unlock();
-        return;
-    }
-
-    if (fn && !fn(job)) {
-        block_job_unlock();
-        return;
-    }
-
-    assert(!job->job.deferred_to_main_loop);
-    timer_del(&job->sleep_timer);
-    job->busy = true;
-    block_job_unlock();
-    aio_co_wake(job->co);
-}
-
 void block_job_enter(BlockJob *job)
 {
-    block_job_enter_cond(job, NULL);
+    job_enter_cond(&job->job, NULL);
 }
 
 void block_job_sleep_ns(BlockJob *job, int64_t ns)
 {
-    assert(job->busy);
+    assert(job->job.busy);
 
     /* Check cancellation *before* setting busy = false, too!  */
     if (job_is_cancelled(&job->job)) {
         return;
     }
 
-    if (!block_job_should_pause(job)) {
-        block_job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
+    if (!job_should_pause(&job->job)) {
+        job_do_yield(&job->job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
     }
 
-    block_job_pause_point(job);
+    job_pause_point(&job->job);
 }
 
 void block_job_yield(BlockJob *job)
 {
-    assert(job->busy);
+    assert(job->job.busy);
 
     /* Check cancellation *before* setting busy = false, too!  */
     if (job_is_cancelled(&job->job)) {
         return;
     }
 
-    if (!block_job_should_pause(job)) {
-        block_job_do_yield(job, -1);
+    if (!job_should_pause(&job->job)) {
+        job_do_yield(&job->job, -1);
     }
 
-    block_job_pause_point(job);
+    job_pause_point(&job->job);
 }
 
 void block_job_iostatus_reset(BlockJob *job)
@@ -1109,7 +968,7 @@ void block_job_iostatus_reset(BlockJob *job)
     if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
         return;
     }
-    assert(job->user_paused && job->pause_count > 0);
+    assert(job->user_paused && job->job.pause_count > 0);
     job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
 }