mirrors/qemu
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
41f4f92260
qemu/job.c
Emanuele Giuseppe Esposito a7b4f8fc09 progressmeter: protect with a mutex 
Progressmeter is protected by the AioContext mutex, which
is taken by the block jobs and their caller (like blockdev).

We would like to remove the dependency of block layer code on the
AioContext mutex, since most drivers and the core I/O code are already
not relying on it.

Create a new C file to implement the ProgressMeter API, but keep the
struct as public, to avoid forcing allocation on the heap.

Also add a mutex to be able to provide an accurate snapshot of the
progress values to the caller.

Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Message-Id: <20210614081130.22134-5-eesposit@redhat.com>
Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
2021-06-25 14:24:24 +03:00

1029 lines
25 KiB
C
Raw Blame History

/*
* Background jobs (long-running operations)
*
* Copyright (c) 2011 IBM Corp.
* Copyright (c) 2012, 2018 Red Hat, Inc.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
#include "qemu/osdep.h"
#include "qapi/error.h"
#include "qemu/job.h"
#include "qemu/id.h"
#include "qemu/main-loop.h"
#include "block/aio-wait.h"
#include "trace/trace-root.h"
#include "qapi/qapi-events-job.h"
static QLIST_HEAD(, Job) jobs = QLIST_HEAD_INITIALIZER(jobs);
/* Job State Transition Table */
bool JobSTT[JOB_STATUS__MAX][JOB_STATUS__MAX] = {
/* U, C, R, P, Y, S, W, D, X, E, N */
/* U: */ [JOB_STATUS_UNDEFINED] = {0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0},
/* C: */ [JOB_STATUS_CREATED] = {0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1},
/* R: */ [JOB_STATUS_RUNNING] = {0, 0, 0, 1, 1, 0, 1, 0, 1, 0, 0},
/* P: */ [JOB_STATUS_PAUSED] = {0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0},
/* Y: */ [JOB_STATUS_READY] = {0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0},
/* S: */ [JOB_STATUS_STANDBY] = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0},
/* W: */ [JOB_STATUS_WAITING] = {0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0},
/* D: */ [JOB_STATUS_PENDING] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
/* X: */ [JOB_STATUS_ABORTING] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
/* E: */ [JOB_STATUS_CONCLUDED] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1},
/* N: */ [JOB_STATUS_NULL] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
};
bool JobVerbTable[JOB_VERB__MAX][JOB_STATUS__MAX] = {
/* U, C, R, P, Y, S, W, D, X, E, N */
[JOB_VERB_CANCEL] = {0, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0},
[JOB_VERB_PAUSE] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
[JOB_VERB_RESUME] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
[JOB_VERB_SET_SPEED] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
[JOB_VERB_COMPLETE] = {0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 0},
[JOB_VERB_FINALIZE] = {0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0},
[JOB_VERB_DISMISS] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0},
};
/* Transactional group of jobs */
struct JobTxn {
/* Is this txn being cancelled? */
bool aborting;
/* List of jobs */
QLIST_HEAD(, Job) jobs;
/* Reference count */
int refcnt;
};
/* Right now, this mutex is only needed to synchronize accesses to job->busy
* and job->sleep_timer, such as concurrent calls to job_do_yield and
* job_enter. */
static QemuMutex job_mutex;
static void job_lock(void)
{
qemu_mutex_lock(&job_mutex);
}
static void job_unlock(void)
{
qemu_mutex_unlock(&job_mutex);
}
static void __attribute__((__constructor__)) job_init(void)
{
qemu_mutex_init(&job_mutex);
}
JobTxn *job_txn_new(void)
{
JobTxn *txn = g_new0(JobTxn, 1);
QLIST_INIT(&txn->jobs);
txn->refcnt = 1;
return txn;
}
static void job_txn_ref(JobTxn *txn)
{
txn->refcnt++;
}
void job_txn_unref(JobTxn *txn)
{
if (txn && --txn->refcnt == 0) {
g_free(txn);
}
}
void job_txn_add_job(JobTxn *txn, Job *job)
{
if (!txn) {
return;
}
assert(!job->txn);
job->txn = txn;
QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
job_txn_ref(txn);
}
static void job_txn_del_job(Job *job)
{
if (job->txn) {
QLIST_REMOVE(job, txn_list);
job_txn_unref(job->txn);
job->txn = NULL;
}
}
static int job_txn_apply(Job *job, int fn(Job *))
{
AioContext *inner_ctx;
Job *other_job, *next;
JobTxn *txn = job->txn;
int rc = 0;
/*
* Similar to job_completed_txn_abort, we take each job's lock before
* applying fn, but since we assume that outer_ctx is held by the caller,
* we need to release it here to avoid holding the lock twice - which would
* break AIO_WAIT_WHILE from within fn.
*/
job_ref(job);
aio_context_release(job->aio_context);
QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
inner_ctx = other_job->aio_context;
aio_context_acquire(inner_ctx);
rc = fn(other_job);
aio_context_release(inner_ctx);
if (rc) {
break;
}
}
/*
* Note that job->aio_context might have been changed by calling fn, so we
* can't use a local variable to cache it.
*/
aio_context_acquire(job->aio_context);
job_unref(job);
return rc;
}
bool job_is_internal(Job *job)
{
return (job->id == NULL);
}
static void job_state_transition(Job *job, JobStatus s1)
{
JobStatus s0 = job->status;
assert(s1 >= 0 && s1 < JOB_STATUS__MAX);
trace_job_state_transition(job, job->ret,
JobSTT[s0][s1] ? "allowed" : "disallowed",
JobStatus_str(s0), JobStatus_str(s1));
assert(JobSTT[s0][s1]);
job->status = s1;
if (!job_is_internal(job) && s1 != s0) {
qapi_event_send_job_status_change(job->id, job->status);
}
}
int job_apply_verb(Job *job, JobVerb verb, Error **errp)
{
JobStatus s0 = job->status;
assert(verb >= 0 && verb < JOB_VERB__MAX);
trace_job_apply_verb(job, JobStatus_str(s0), JobVerb_str(verb),
JobVerbTable[verb][s0] ? "allowed" : "prohibited");
if (JobVerbTable[verb][s0]) {
return 0;
}
error_setg(errp, "Job '%s' in state '%s' cannot accept command verb '%s'",
job->id, JobStatus_str(s0), JobVerb_str(verb));
return -EPERM;
}
JobType job_type(const Job *job)
{
return job->driver->job_type;
}
const char *job_type_str(const Job *job)
{
return JobType_str(job_type(job));
}
bool job_is_cancelled(Job *job)
{
return job->cancelled;
}
bool job_is_ready(Job *job)
{
switch (job->status) {
case JOB_STATUS_UNDEFINED:
case JOB_STATUS_CREATED:
case JOB_STATUS_RUNNING:
case JOB_STATUS_PAUSED:
case JOB_STATUS_WAITING:
case JOB_STATUS_PENDING:
case JOB_STATUS_ABORTING:
case JOB_STATUS_CONCLUDED:
case JOB_STATUS_NULL:
return false;
case JOB_STATUS_READY:
case JOB_STATUS_STANDBY:
return true;
default:
g_assert_not_reached();
}
return false;
}
bool job_is_completed(Job *job)
{
switch (job->status) {
case JOB_STATUS_UNDEFINED:
case JOB_STATUS_CREATED:
case JOB_STATUS_RUNNING:
case JOB_STATUS_PAUSED:
case JOB_STATUS_READY:
case JOB_STATUS_STANDBY:
return false;
case JOB_STATUS_WAITING:
case JOB_STATUS_PENDING:
case JOB_STATUS_ABORTING:
case JOB_STATUS_CONCLUDED:
case JOB_STATUS_NULL:
return true;
default:
g_assert_not_reached();
}
return false;
}
static bool job_started(Job *job)
{
return job->co;
}
static bool job_should_pause(Job *job)
{
return job->pause_count > 0;
}
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;
}
static void job_sleep_timer_cb(void *opaque)
{
Job *job = opaque;
job_enter(job);
}
void *job_create(const char *job_id, const JobDriver *driver, JobTxn *txn,
AioContext *ctx, int flags, BlockCompletionFunc *cb,
void *opaque, Error **errp)
{
Job *job;
if (job_id) {
if (flags & JOB_INTERNAL) {
error_setg(errp, "Cannot specify job ID for internal job");
return NULL;
}
if (!id_wellformed(job_id)) {
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;
}
} else if (!(flags & JOB_INTERNAL)) {
error_setg(errp, "An explicit job ID is required");
return NULL;
}
job = g_malloc0(driver->instance_size);
job->driver = driver;
job->id = g_strdup(job_id);
job->refcnt = 1;
job->aio_context = ctx;
job->busy = false;
job->paused = true;
job->pause_count = 1;
job->auto_finalize = !(flags & JOB_MANUAL_FINALIZE);
job->auto_dismiss = !(flags & JOB_MANUAL_DISMISS);
job->cb = cb;
job->opaque = opaque;
progress_init(&job->progress);
notifier_list_init(&job->on_finalize_cancelled);
notifier_list_init(&job->on_finalize_completed);
notifier_list_init(&job->on_pending);
notifier_list_init(&job->on_ready);
job_state_transition(job, JOB_STATUS_CREATED);
aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
QEMU_CLOCK_REALTIME, SCALE_NS,
job_sleep_timer_cb, job);
QLIST_INSERT_HEAD(&jobs, job, job_list);
/* Single jobs are modeled as single-job transactions for sake of
* consolidating the job management logic */
if (!txn) {
txn = job_txn_new();
job_txn_add_job(txn, job);
job_txn_unref(txn);
} else {
job_txn_add_job(txn, job);
}
return job;
}
void job_ref(Job *job)
{
++job->refcnt;
}
void job_unref(Job *job)
{
if (--job->refcnt == 0) {
assert(job->status == JOB_STATUS_NULL);
assert(!timer_pending(&job->sleep_timer));
assert(!job->txn);
if (job->driver->free) {
job->driver->free(job);
}
QLIST_REMOVE(job, job_list);
progress_destroy(&job->progress);
error_free(job->err);
g_free(job->id);
g_free(job);
}
}
void job_progress_update(Job *job, uint64_t done)
{
progress_work_done(&job->progress, done);
}
void job_progress_set_remaining(Job *job, uint64_t remaining)
{
progress_set_remaining(&job->progress, remaining);
}
void job_progress_increase_remaining(Job *job, uint64_t delta)
{
progress_increase_remaining(&job->progress, delta);
}
void job_event_cancelled(Job *job)
{
notifier_list_notify(&job->on_finalize_cancelled, job);
}
void job_event_completed(Job *job)
{
notifier_list_notify(&job->on_finalize_completed, job);
}
static void job_event_pending(Job *job)
{
notifier_list_notify(&job->on_pending, job);
}
static void job_event_ready(Job *job)
{
notifier_list_notify(&job->on_ready, job);
}
static void job_event_idle(Job *job)
{
notifier_list_notify(&job->on_idle, job);
}
void job_enter_cond(Job *job, bool(*fn)(Job *job))
{
if (!job_started(job)) {
return;
}
if (job->deferred_to_main_loop) {
return;
}
job_lock();
if (job->busy) {
job_unlock();
return;
}
if (fn && !fn(job)) {
job_unlock();
return;
}
assert(!job->deferred_to_main_loop);
timer_del(&job->sleep_timer);
job->busy = true;
job_unlock();
aio_co_enter(job->aio_context, job->co);
}
void job_enter(Job *job)
{
job_enter_cond(job, NULL);
}
/* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
* Reentering the job coroutine with job_enter() before the timer has expired
* is allowed and cancels the timer.
*
* If @ns is (uint64_t) -1, no timer is scheduled and job_enter() must be
* called explicitly. */
static void coroutine_fn job_do_yield(Job *job, uint64_t ns)
{
job_lock();
if (ns != -1) {
timer_mod(&job->sleep_timer, ns);
}
job->busy = false;
job_event_idle(job);
job_unlock();
qemu_coroutine_yield();
/* Set by job_enter_cond() before re-entering the coroutine. */
assert(job->busy);
}
void coroutine_fn job_pause_point(Job *job)
{
assert(job && job_started(job));
if (!job_should_pause(job)) {
return;
}
if (job_is_cancelled(job)) {
return;
}
if (job->driver->pause) {
job->driver->pause(job);
}
if (job_should_pause(job) && !job_is_cancelled(job)) {
JobStatus status = job->status;
job_state_transition(job, status == JOB_STATUS_READY
? JOB_STATUS_STANDBY
: JOB_STATUS_PAUSED);
job->paused = true;
job_do_yield(job, -1);
job->paused = false;
job_state_transition(job, status);
}
if (job->driver->resume) {
job->driver->resume(job);
}
}
void job_yield(Job *job)
{
assert(job->busy);
/* Check cancellation *before* setting busy = false, too! */
if (job_is_cancelled(job)) {
return;
}
if (!job_should_pause(job)) {
job_do_yield(job, -1);
}
job_pause_point(job);
}
void coroutine_fn job_sleep_ns(Job *job, int64_t ns)
{
assert(job->busy);
/* Check cancellation *before* setting busy = false, too! */
if (job_is_cancelled(job)) {
return;
}
if (!job_should_pause(job)) {
job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
}
job_pause_point(job);
}
/* Assumes the block_job_mutex is held */
static bool job_timer_not_pending(Job *job)
{
return !timer_pending(&job->sleep_timer);
}
void job_pause(Job *job)
{
job->pause_count++;
if (!job->paused) {
job_enter(job);
}
}
void job_resume(Job *job)
{
assert(job->pause_count > 0);
job->pause_count--;
if (job->pause_count) {
return;
}
/* kick only if no timer is pending */
job_enter_cond(job, job_timer_not_pending);
}
void job_user_pause(Job *job, Error **errp)
{
if (job_apply_verb(job, JOB_VERB_PAUSE, errp)) {
return;
}
if (job->user_paused) {
error_setg(errp, "Job is already paused");
return;
}
job->user_paused = true;
job_pause(job);
}
bool job_user_paused(Job *job)
{
return job->user_paused;
}
void job_user_resume(Job *job, Error **errp)
{
assert(job);
if (!job->user_paused || job->pause_count <= 0) {
error_setg(errp, "Can't resume a job that was not paused");
return;
}
if (job_apply_verb(job, JOB_VERB_RESUME, errp)) {
return;
}
if (job->driver->user_resume) {
job->driver->user_resume(job);
}
job->user_paused = false;
job_resume(job);
}
static void job_do_dismiss(Job *job)
{
assert(job);
job->busy = false;
job->paused = false;
job->deferred_to_main_loop = true;
job_txn_del_job(job);
job_state_transition(job, JOB_STATUS_NULL);
job_unref(job);
}
void job_dismiss(Job **jobptr, Error **errp)
{
Job *job = *jobptr;
/* similarly to _complete, this is QMP-interface only. */
assert(job->id);
if (job_apply_verb(job, JOB_VERB_DISMISS, errp)) {
return;
}
job_do_dismiss(job);
*jobptr = NULL;
}
void job_early_fail(Job *job)
{
assert(job->status == JOB_STATUS_CREATED);
job_do_dismiss(job);
}
static void job_conclude(Job *job)
{
job_state_transition(job, JOB_STATUS_CONCLUDED);
if (job->auto_dismiss || !job_started(job)) {
job_do_dismiss(job);
}
}
static void job_update_rc(Job *job)
{
if (!job->ret && job_is_cancelled(job)) {
job->ret = -ECANCELED;
}
if (job->ret) {
if (!job->err) {
error_setg(&job->err, "%s", strerror(-job->ret));
}
job_state_transition(job, JOB_STATUS_ABORTING);
}
}
static void job_commit(Job *job)
{
assert(!job->ret);
if (job->driver->commit) {
job->driver->commit(job);
}
}
static void job_abort(Job *job)
{
assert(job->ret);
if (job->driver->abort) {
job->driver->abort(job);
}
}
static void job_clean(Job *job)
{
if (job->driver->clean) {
job->driver->clean(job);
}
}
static int job_finalize_single(Job *job)
{
assert(job_is_completed(job));
/* Ensure abort is called for late-transactional failures */
job_update_rc(job);
if (!job->ret) {
job_commit(job);
} else {
job_abort(job);
}
job_clean(job);
if (job->cb) {
job->cb(job->opaque, job->ret);
}
/* Emit events only if we actually started */
if (job_started(job)) {
if (job_is_cancelled(job)) {
job_event_cancelled(job);
} else {
job_event_completed(job);
}
}
job_txn_del_job(job);
job_conclude(job);
return 0;
}
static void job_cancel_async(Job *job, bool force)
{
if (job->driver->cancel) {
job->driver->cancel(job, force);
}
if (job->user_paused) {
/* Do not call job_enter here, the caller will handle it. */
if (job->driver->user_resume) {
job->driver->user_resume(job);
}
job->user_paused = false;
assert(job->pause_count > 0);
job->pause_count--;
}
job->cancelled = true;
/* To prevent 'force == false' overriding a previous 'force == true' */
job->force_cancel |= force;
}
static void job_completed_txn_abort(Job *job)
{
AioContext *outer_ctx = job->aio_context;
AioContext *ctx;
JobTxn *txn = job->txn;
Job *other_job;
if (txn->aborting) {
/*
* We are cancelled by another job, which will handle everything.
*/
return;
}
txn->aborting = true;
job_txn_ref(txn);
/* We can only hold the single job's AioContext lock while calling
* job_finalize_single() because the finalization callbacks can involve
* calls of AIO_WAIT_WHILE(), which could deadlock otherwise. */
aio_context_release(outer_ctx);
/* Other jobs are effectively cancelled by us, set the status for
* them; this job, however, may or may not be cancelled, depending
* on the caller, so leave it. */
QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
if (other_job != job) {
ctx = other_job->aio_context;
aio_context_acquire(ctx);
job_cancel_async(other_job, false);
aio_context_release(ctx);
}
}
while (!QLIST_EMPTY(&txn->jobs)) {
other_job = QLIST_FIRST(&txn->jobs);
ctx = other_job->aio_context;
aio_context_acquire(ctx);
if (!job_is_completed(other_job)) {
assert(job_is_cancelled(other_job));
job_finish_sync(other_job, NULL, NULL);
}
job_finalize_single(other_job);
aio_context_release(ctx);
}
aio_context_acquire(outer_ctx);
job_txn_unref(txn);
}
static int job_prepare(Job *job)
{
if (job->ret == 0 && job->driver->prepare) {
job->ret = job->driver->prepare(job);
job_update_rc(job);
}
return job->ret;
}
static int job_needs_finalize(Job *job)
{
return !job->auto_finalize;
}
static void job_do_finalize(Job *job)
{
int rc;
assert(job && job->txn);
/* prepare the transaction to complete */
rc = job_txn_apply(job, job_prepare);
if (rc) {
job_completed_txn_abort(job);
} else {
job_txn_apply(job, job_finalize_single);
}
}
void job_finalize(Job *job, Error **errp)
{
assert(job && job->id);
if (job_apply_verb(job, JOB_VERB_FINALIZE, errp)) {
return;
}
job_do_finalize(job);
}
static int job_transition_to_pending(Job *job)
{
job_state_transition(job, JOB_STATUS_PENDING);
if (!job->auto_finalize) {
job_event_pending(job);
}
return 0;
}
void job_transition_to_ready(Job *job)
{
job_state_transition(job, JOB_STATUS_READY);
job_event_ready(job);
}
static void job_completed_txn_success(Job *job)
{
JobTxn *txn = job->txn;
Job *other_job;
job_state_transition(job, JOB_STATUS_WAITING);
/*
* Successful completion, see if there are other running jobs in this
* txn.
*/
QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
if (!job_is_completed(other_job)) {
return;
}
assert(other_job->ret == 0);
}
job_txn_apply(job, job_transition_to_pending);
/* If no jobs need manual finalization, automatically do so */
if (job_txn_apply(job, job_needs_finalize) == 0) {
job_do_finalize(job);
}
}
static void job_completed(Job *job)
{
assert(job && job->txn && !job_is_completed(job));
job_update_rc(job);
trace_job_completed(job, job->ret);
if (job->ret) {
job_completed_txn_abort(job);
} else {
job_completed_txn_success(job);
}
}
/** Useful only as a type shim for aio_bh_schedule_oneshot. */
static void job_exit(void *opaque)
{
Job *job = (Job *)opaque;
AioContext *ctx;
job_ref(job);
aio_context_acquire(job->aio_context);
/* This is a lie, we're not quiescent, but still doing the completion
* callbacks. However, completion callbacks tend to involve operations that
* drain block nodes, and if .drained_poll still returned true, we would
* deadlock. */
job->busy = false;
job_event_idle(job);
job_completed(job);
/*
* Note that calling job_completed can move the job to a different
* aio_context, so we cannot cache from above. job_txn_apply takes care of
* acquiring the new lock, and we ref/unref to avoid job_completed freeing
* the job underneath us.
*/
ctx = job->aio_context;
job_unref(job);
aio_context_release(ctx);
}
/**
* 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 job_co_entry(void *opaque)
{
Job *job = opaque;
assert(job && job->driver && job->driver->run);
job_pause_point(job);
job->ret = job->driver->run(job, &job->err);
job->deferred_to_main_loop = true;
job->busy = true;
aio_bh_schedule_oneshot(qemu_get_aio_context(), job_exit, job);
}
void job_start(Job *job)
{
assert(job && !job_started(job) && job->paused &&
job->driver && job->driver->run);
job->co = qemu_coroutine_create(job_co_entry, job);
job->pause_count--;
job->busy = true;
job->paused = false;
job_state_transition(job, JOB_STATUS_RUNNING);
aio_co_enter(job->aio_context, job->co);
}
void job_cancel(Job *job, bool force)
{
if (job->status == JOB_STATUS_CONCLUDED) {
job_do_dismiss(job);
return;
}
job_cancel_async(job, force);
if (!job_started(job)) {
job_completed(job);
} else if (job->deferred_to_main_loop) {
job_completed_txn_abort(job);
} else {
job_enter(job);
}
}
void job_user_cancel(Job *job, bool force, Error **errp)
{
if (job_apply_verb(job, JOB_VERB_CANCEL, errp)) {
return;
}
job_cancel(job, force);
}
/* A wrapper around job_cancel() taking an Error ** parameter so it may be
* used with job_finish_sync() without the need for (rather nasty) function
* pointer casts there. */
static void job_cancel_err(Job *job, Error **errp)
{
job_cancel(job, false);
}
int job_cancel_sync(Job *job)
{
return job_finish_sync(job, &job_cancel_err, NULL);
}
void job_cancel_sync_all(void)
{
Job *job;
AioContext *aio_context;
while ((job = job_next(NULL))) {
aio_context = job->aio_context;
aio_context_acquire(aio_context);
job_cancel_sync(job);
aio_context_release(aio_context);
}
}
int job_complete_sync(Job *job, Error **errp)
{
return job_finish_sync(job, job_complete, errp);
}
void job_complete(Job *job, Error **errp)
{
/* Should not be reachable via external interface for internal jobs */
assert(job->id);
if (job_apply_verb(job, JOB_VERB_COMPLETE, errp)) {
return;
}
if (job_is_cancelled(job) || !job->driver->complete) {
error_setg(errp, "The active block job '%s' cannot be completed",
job->id);
return;
}
job->driver->complete(job, errp);
}
int job_finish_sync(Job *job, void (*finish)(Job *, Error **errp), Error **errp)
{
Error *local_err = NULL;
int ret;
job_ref(job);
if (finish) {
finish(job, &local_err);
}
if (local_err) {
error_propagate(errp, local_err);
job_unref(job);
return -EBUSY;
}
AIO_WAIT_WHILE(job->aio_context,
(job_enter(job), !job_is_completed(job)));
ret = (job_is_cancelled(job) && job->ret == 0) ? -ECANCELED : job->ret;
job_unref(job);
return ret;
}
Reference in New Issue View Git Blame Copy Permalink