diff options
Diffstat (limited to 'blockjob.c')
| -rw-r--r-- | blockjob.c | 1094 |
1 files changed, 149 insertions, 945 deletions
diff --git a/blockjob.c b/blockjob.c index 112672a68b..0306533a2e 100644 --- a/blockjob.c +++ b/blockjob.c @@ -34,103 +34,8 @@ #include "qapi/qapi-events-block-core.h" #include "qapi/qmp/qerror.h" #include "qemu/coroutine.h" -#include "qemu/id.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; - -/* BlockJob State Transition Table */ -bool BlockJobSTT[BLOCK_JOB_STATUS__MAX][BLOCK_JOB_STATUS__MAX] = { - /* U, C, R, P, Y, S, W, D, X, E, N */ - /* U: */ [BLOCK_JOB_STATUS_UNDEFINED] = {0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0}, - /* C: */ [BLOCK_JOB_STATUS_CREATED] = {0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1}, - /* R: */ [BLOCK_JOB_STATUS_RUNNING] = {0, 0, 0, 1, 1, 0, 1, 0, 1, 0, 0}, - /* P: */ [BLOCK_JOB_STATUS_PAUSED] = {0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0}, - /* Y: */ [BLOCK_JOB_STATUS_READY] = {0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0}, - /* S: */ [BLOCK_JOB_STATUS_STANDBY] = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0}, - /* W: */ [BLOCK_JOB_STATUS_WAITING] = {0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0}, - /* D: */ [BLOCK_JOB_STATUS_PENDING] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0}, - /* X: */ [BLOCK_JOB_STATUS_ABORTING] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0}, - /* E: */ [BLOCK_JOB_STATUS_CONCLUDED] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1}, - /* N: */ [BLOCK_JOB_STATUS_NULL] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, -}; - -bool BlockJobVerbTable[BLOCK_JOB_VERB__MAX][BLOCK_JOB_STATUS__MAX] = { - /* U, C, R, P, Y, S, W, D, X, E, N */ - [BLOCK_JOB_VERB_CANCEL] = {0, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0}, - [BLOCK_JOB_VERB_PAUSE] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0}, - [BLOCK_JOB_VERB_RESUME] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0}, - [BLOCK_JOB_VERB_SET_SPEED] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0}, - [BLOCK_JOB_VERB_COMPLETE] = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0}, - [BLOCK_JOB_VERB_FINALIZE] = {0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0}, - [BLOCK_JOB_VERB_DISMISS] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0}, -}; - -static void block_job_state_transition(BlockJob *job, BlockJobStatus s1) -{ - BlockJobStatus s0 = job->status; - assert(s1 >= 0 && s1 <= BLOCK_JOB_STATUS__MAX); - trace_block_job_state_transition(job, job->ret, BlockJobSTT[s0][s1] ? - "allowed" : "disallowed", - BlockJobStatus_str(s0), - BlockJobStatus_str(s1)); - assert(BlockJobSTT[s0][s1]); - job->status = s1; -} - -static int block_job_apply_verb(BlockJob *job, BlockJobVerb bv, Error **errp) -{ - assert(bv >= 0 && bv <= BLOCK_JOB_VERB__MAX); - trace_block_job_apply_verb(job, BlockJobStatus_str(job->status), - BlockJobVerb_str(bv), - BlockJobVerbTable[bv][job->status] ? - "allowed" : "prohibited"); - if (BlockJobVerbTable[bv][job->status]) { - return 0; - } - error_setg(errp, "Job '%s' in state '%s' cannot accept command verb '%s'", - job->id, BlockJobStatus_str(job->status), BlockJobVerb_str(bv)); - return -EPERM; -} - -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 { - - /* Is this txn being cancelled? */ - bool aborting; - - /* List of jobs */ - QLIST_HEAD(, BlockJob) jobs; - - /* Reference count */ - int refcnt; -}; - -static QLIST_HEAD(, BlockJob) block_jobs = QLIST_HEAD_INITIALIZER(block_jobs); - /* * The block job API is composed of two categories of functions. * @@ -146,147 +51,78 @@ static QLIST_HEAD(, BlockJob) block_jobs = QLIST_HEAD_INITIALIZER(block_jobs); * blockjob_int.h. */ -BlockJob *block_job_next(BlockJob *job) +static bool is_block_job(Job *job) { - if (!job) { - return QLIST_FIRST(&block_jobs); - } - return QLIST_NEXT(job, job_list); -} - -BlockJob *block_job_get(const char *id) -{ - BlockJob *job; - - QLIST_FOREACH(job, &block_jobs, job_list) { - if (job->id && !strcmp(id, job->id)) { - return job; - } - } - - return NULL; + return job_type(job) == JOB_TYPE_BACKUP || + job_type(job) == JOB_TYPE_COMMIT || + job_type(job) == JOB_TYPE_MIRROR || + job_type(job) == JOB_TYPE_STREAM; } -BlockJobTxn *block_job_txn_new(void) +BlockJob *block_job_next(BlockJob *bjob) { - BlockJobTxn *txn = g_new0(BlockJobTxn, 1); - QLIST_INIT(&txn->jobs); - txn->refcnt = 1; - return txn; -} - -static void block_job_txn_ref(BlockJobTxn *txn) -{ - txn->refcnt++; -} - -void block_job_txn_unref(BlockJobTxn *txn) -{ - if (txn && --txn->refcnt == 0) { - g_free(txn); - } -} - -void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job) -{ - if (!txn) { - return; - } + Job *job = bjob ? &bjob->job : NULL; - assert(!job->txn); - job->txn = txn; + do { + job = job_next(job); + } while (job && !is_block_job(job)); - QLIST_INSERT_HEAD(&txn->jobs, job, txn_list); - block_job_txn_ref(txn); + return job ? container_of(job, BlockJob, job) : NULL; } -static void block_job_txn_del_job(BlockJob *job) -{ - if (job->txn) { - QLIST_REMOVE(job, txn_list); - block_job_txn_unref(job->txn); - job->txn = NULL; - } -} - -/* 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) -{ - return !block_job_timer_pending(job); -} - -static void block_job_pause(BlockJob *job) +BlockJob *block_job_get(const char *id) { - job->pause_count++; -} + Job *job = job_get(id); -static void block_job_resume(BlockJob *job) -{ - assert(job->pause_count > 0); - job->pause_count--; - if (job->pause_count) { - return; + if (job && is_block_job(job)) { + return container_of(job, BlockJob, job); + } else { + return NULL; } - - /* kick only if no timer is pending */ - block_job_enter_cond(job, block_job_timer_not_pending); -} - -void block_job_ref(BlockJob *job) -{ - ++job->refcnt; } static void block_job_attached_aio_context(AioContext *new_context, void *opaque); static void block_job_detach_aio_context(void *opaque); -void block_job_unref(BlockJob *job) +void block_job_free(Job *job) { - if (--job->refcnt == 0) { - assert(job->status == BLOCK_JOB_STATUS_NULL); - assert(!job->txn); - BlockDriverState *bs = blk_bs(job->blk); - QLIST_REMOVE(job, job_list); - bs->job = NULL; - block_job_remove_all_bdrv(job); - blk_remove_aio_context_notifier(job->blk, - block_job_attached_aio_context, - block_job_detach_aio_context, job); - blk_unref(job->blk); - error_free(job->blocker); - g_free(job->id); - assert(!timer_pending(&job->sleep_timer)); - g_free(job); - } + BlockJob *bjob = container_of(job, BlockJob, job); + BlockDriverState *bs = blk_bs(bjob->blk); + + bs->job = NULL; + block_job_remove_all_bdrv(bjob); + blk_remove_aio_context_notifier(bjob->blk, + block_job_attached_aio_context, + block_job_detach_aio_context, bjob); + blk_unref(bjob->blk); + error_free(bjob->blocker); } static void block_job_attached_aio_context(AioContext *new_context, void *opaque) { BlockJob *job = opaque; + const JobDriver *drv = job->job.driver; + BlockJobDriver *bjdrv = container_of(drv, BlockJobDriver, job_driver); - if (job->driver->attached_aio_context) { - job->driver->attached_aio_context(job, new_context); + job->job.aio_context = new_context; + if (bjdrv->attached_aio_context) { + bjdrv->attached_aio_context(job, new_context); } - block_job_resume(job); + job_resume(&job->job); } -static void block_job_drain(BlockJob *job) +void block_job_drain(Job *job) { - /* If job is !job->busy this kicks it into the next pause point. */ - block_job_enter(job); + BlockJob *bjob = container_of(job, BlockJob, job); + const JobDriver *drv = job->driver; + BlockJobDriver *bjdrv = container_of(drv, BlockJobDriver, job_driver); - blk_drain(job->blk); - if (job->driver->drain) { - job->driver->drain(job); + blk_drain(bjob->blk); + if (bjdrv->drain) { + bjdrv->drain(bjob); } } @@ -295,35 +131,34 @@ static void block_job_detach_aio_context(void *opaque) BlockJob *job = opaque; /* In case the job terminates during aio_poll()... */ - block_job_ref(job); + job_ref(&job->job); - block_job_pause(job); + job_pause(&job->job); - while (!job->paused && !job->completed) { - block_job_drain(job); + while (!job->job.paused && !job_is_completed(&job->job)) { + job_drain(&job->job); } - block_job_unref(job); + job->job.aio_context = NULL; + job_unref(&job->job); } 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), - 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) { BlockJob *job = c->opaque; - block_job_pause(job); + job_pause(&job->job); } static void child_job_drained_end(BdrvChild *c) { BlockJob *job = c->opaque; - block_job_resume(job); + job_resume(&job->job); } static const BdrvChildRole child_job = { @@ -365,316 +200,25 @@ int block_job_add_bdrv(BlockJob *job, const char *name, BlockDriverState *bs, bool block_job_is_internal(BlockJob *job) { - return (job->id == NULL); -} - -static bool block_job_started(BlockJob *job) -{ - return job->co; + return (job->job.id == NULL); } 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; - - 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; - block_job_state_transition(job, BLOCK_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->deferred_to_main_loop = true; - block_job_txn_del_job(job); - block_job_state_transition(job, BLOCK_JOB_STATUS_NULL); - block_job_unref(job); + return container_of(job->job.driver, BlockJobDriver, job_driver); } -static void block_job_do_dismiss(BlockJob *job) +/* Assumes the job_mutex is held */ +static bool job_timer_pending(Job *job) { - block_job_decommission(job); -} - -static void block_job_conclude(BlockJob *job) -{ - block_job_state_transition(job, BLOCK_JOB_STATUS_CONCLUDED); - if (job->auto_dismiss || !block_job_started(job)) { - block_job_do_dismiss(job); - } -} - -static void block_job_update_rc(BlockJob *job) -{ - if (!job->ret && block_job_is_cancelled(job)) { - job->ret = -ECANCELED; - } - if (job->ret) { - block_job_state_transition(job, BLOCK_JOB_STATUS_ABORTING); - } -} - -static int block_job_prepare(BlockJob *job) -{ - if (job->ret == 0 && job->driver->prepare) { - job->ret = job->driver->prepare(job); - } - return job->ret; -} - -static void block_job_commit(BlockJob *job) -{ - assert(!job->ret); - if (job->driver->commit) { - job->driver->commit(job); - } -} - -static void block_job_abort(BlockJob *job) -{ - assert(job->ret); - if (job->driver->abort) { - job->driver->abort(job); - } -} - -static void block_job_clean(BlockJob *job) -{ - if (job->driver->clean) { - job->driver->clean(job); - } -} - -static int block_job_finalize_single(BlockJob *job) -{ - assert(job->completed); - - /* Ensure abort is called for late-transactional failures */ - block_job_update_rc(job); - - if (!job->ret) { - block_job_commit(job); - } else { - block_job_abort(job); - } - block_job_clean(job); - - if (job->cb) { - job->cb(job->opaque, job->ret); - } - - /* Emit events only if we actually started */ - if (block_job_started(job)) { - if (block_job_is_cancelled(job)) { - block_job_event_cancelled(job); - } else { - const char *msg = NULL; - if (job->ret < 0) { - msg = strerror(-job->ret); - } - block_job_event_completed(job, msg); - } - } - - block_job_txn_del_job(job); - block_job_conclude(job); - return 0; -} - -static void block_job_cancel_async(BlockJob *job, bool force) -{ - if (job->iostatus != BLOCK_DEVICE_IO_STATUS_OK) { - block_job_iostatus_reset(job); - } - if (job->user_paused) { - /* Do not call block_job_enter here, the caller will handle it. */ - job->user_paused = false; - job->pause_count--; - } - job->cancelled = true; - /* To prevent 'force == false' overriding a previous 'force == true' */ - job->force |= force; -} - -static int block_job_txn_apply(BlockJobTxn *txn, int fn(BlockJob *), bool lock) -{ - AioContext *ctx; - BlockJob *job, *next; - int rc = 0; - - QLIST_FOREACH_SAFE(job, &txn->jobs, txn_list, next) { - if (lock) { - ctx = blk_get_aio_context(job->blk); - aio_context_acquire(ctx); - } - rc = fn(job); - if (lock) { - aio_context_release(ctx); - } - if (rc) { - break; - } - } - return rc; -} - -static int block_job_finish_sync(BlockJob *job, - void (*finish)(BlockJob *, Error **errp), - Error **errp) -{ - Error *local_err = NULL; - int ret; - - assert(blk_bs(job->blk)->job == job); - - block_job_ref(job); - - if (finish) { - finish(job, &local_err); - } - if (local_err) { - error_propagate(errp, local_err); - block_job_unref(job); - return -EBUSY; - } - /* block_job_drain calls block_job_enter, and it should be enough to - * induce progress until the job completes or moves to the main thread. - */ - while (!job->deferred_to_main_loop && !job->completed) { - block_job_drain(job); - } - while (!job->completed) { - aio_poll(qemu_get_aio_context(), true); - } - ret = (job->cancelled && job->ret == 0) ? -ECANCELED : job->ret; - block_job_unref(job); - return ret; -} - -static void block_job_completed_txn_abort(BlockJob *job) -{ - AioContext *ctx; - BlockJobTxn *txn = job->txn; - BlockJob *other_job; - - if (txn->aborting) { - /* - * We are cancelled by another job, which will handle everything. - */ - return; - } - txn->aborting = true; - block_job_txn_ref(txn); - - /* We are the first failed job. Cancel other jobs. */ - QLIST_FOREACH(other_job, &txn->jobs, txn_list) { - ctx = blk_get_aio_context(other_job->blk); - aio_context_acquire(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) { - block_job_cancel_async(other_job, false); - } - } - while (!QLIST_EMPTY(&txn->jobs)) { - other_job = QLIST_FIRST(&txn->jobs); - ctx = blk_get_aio_context(other_job->blk); - if (!other_job->completed) { - assert(other_job->cancelled); - block_job_finish_sync(other_job, NULL, NULL); - } - block_job_finalize_single(other_job); - aio_context_release(ctx); - } - - block_job_txn_unref(txn); -} - -static int block_job_needs_finalize(BlockJob *job) -{ - return !job->auto_finalize; -} - -static void block_job_do_finalize(BlockJob *job) -{ - int rc; - assert(job && job->txn); - - /* prepare the transaction to complete */ - rc = block_job_txn_apply(job->txn, block_job_prepare, true); - if (rc) { - block_job_completed_txn_abort(job); - } else { - block_job_txn_apply(job->txn, block_job_finalize_single, true); - } -} - -static void block_job_completed_txn_success(BlockJob *job) -{ - BlockJobTxn *txn = job->txn; - BlockJob *other_job; - - block_job_state_transition(job, BLOCK_JOB_STATUS_WAITING); - - /* - * Successful completion, see if there are other running jobs in this - * txn. - */ - QLIST_FOREACH(other_job, &txn->jobs, txn_list) { - if (!other_job->completed) { - return; - } - assert(other_job->ret == 0); - } - - block_job_txn_apply(txn, block_job_event_pending, false); - - /* If no jobs need manual finalization, automatically do so */ - if (block_job_txn_apply(txn, block_job_needs_finalize, false) == 0) { - block_job_do_finalize(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; - if (block_job_apply_verb(job, BLOCK_JOB_VERB_SET_SPEED, errp)) { + if (job_apply_verb(&job->job, JOB_VERB_SET_SPEED, errp)) { return; } if (speed < 0) { @@ -690,7 +234,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) @@ -702,142 +246,6 @@ int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n) 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 */ - assert(job->id); - if (block_job_apply_verb(job, BLOCK_JOB_VERB_COMPLETE, errp)) { - return; - } - if (job->pause_count || job->cancelled || !job->driver->complete) { - error_setg(errp, "The active block job '%s' cannot be completed", - job->id); - return; - } - - job->driver->complete(job, errp); -} - -void block_job_finalize(BlockJob *job, Error **errp) -{ - assert(job && job->id); - if (block_job_apply_verb(job, BLOCK_JOB_VERB_FINALIZE, errp)) { - return; - } - block_job_do_finalize(job); -} - -void block_job_dismiss(BlockJob **jobptr, Error **errp) -{ - BlockJob *job = *jobptr; - /* similarly to _complete, this is QMP-interface only. */ - assert(job->id); - if (block_job_apply_verb(job, BLOCK_JOB_VERB_DISMISS, errp)) { - return; - } - - block_job_do_dismiss(job); - *jobptr = NULL; -} - -void block_job_user_pause(BlockJob *job, Error **errp) -{ - if (block_job_apply_verb(job, BLOCK_JOB_VERB_PAUSE, errp)) { - return; - } - if (job->user_paused) { - error_setg(errp, "Job is already paused"); - return; - } - job->user_paused = true; - block_job_pause(job); -} - -bool block_job_user_paused(BlockJob *job) -{ - return job->user_paused; -} - -void block_job_user_resume(BlockJob *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 (block_job_apply_verb(job, BLOCK_JOB_VERB_RESUME, errp)) { - return; - } - block_job_iostatus_reset(job); - job->user_paused = false; - block_job_resume(job); -} - -void block_job_cancel(BlockJob *job, bool force) -{ - if (job->status == BLOCK_JOB_STATUS_CONCLUDED) { - block_job_do_dismiss(job); - return; - } - block_job_cancel_async(job, force); - if (!block_job_started(job)) { - block_job_completed(job, -ECANCELED); - } else if (job->deferred_to_main_loop) { - block_job_completed_txn_abort(job); - } else { - block_job_enter(job); - } -} - -void block_job_user_cancel(BlockJob *job, bool force, Error **errp) -{ - if (block_job_apply_verb(job, BLOCK_JOB_VERB_CANCEL, errp)) { - return; - } - block_job_cancel(job, force); -} - -/* A wrapper around block_job_cancel() taking an Error ** parameter so it may be - * used with block_job_finish_sync() without the need for (rather nasty) - * function pointer casts there. */ -static void block_job_cancel_err(BlockJob *job, Error **errp) -{ - block_job_cancel(job, false); -} - -int block_job_cancel_sync(BlockJob *job) -{ - return block_job_finish_sync(job, &block_job_cancel_err, NULL); -} - -void block_job_cancel_sync_all(void) -{ - BlockJob *job; - AioContext *aio_context; - - while ((job = QLIST_FIRST(&block_jobs))) { - aio_context = blk_get_aio_context(job->blk); - aio_context_acquire(aio_context); - block_job_cancel_sync(job); - aio_context_release(aio_context); - } -} - -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; @@ -847,20 +255,20 @@ 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->device = g_strdup(job->id); - info->len = job->len; - info->busy = atomic_read(&job->busy); - info->paused = job->pause_count > 0; - info->offset = job->offset; + info->type = g_strdup(job_type_str(&job->job)); + info->device = g_strdup(job->job.id); + info->busy = atomic_read(&job->job.busy); + info->paused = job->job.pause_count > 0; + info->offset = job->job.progress_current; + info->len = job->job.progress_total; info->speed = job->speed; info->io_status = job->iostatus; - info->ready = job->ready; - info->status = job->status; - info->auto_finalize = job->auto_finalize; - info->auto_dismiss = job->auto_dismiss; - info->has_error = job->ret != 0; - info->error = job->ret ? g_strdup(strerror(-job->ret)) : NULL; + info->ready = job_is_ready(&job->job), + info->status = job->job.status; + info->auto_finalize = job->job.auto_finalize; + info->auto_dismiss = job->job.auto_dismiss; + info->has_error = job->job.ret != 0; + info->error = job->job.ret ? g_strdup(strerror(-job->job.ret)) : NULL; return info; } @@ -872,54 +280,81 @@ static void block_job_iostatus_set_err(BlockJob *job, int error) } } -static void block_job_event_cancelled(BlockJob *job) +static void block_job_event_cancelled(Notifier *n, void *opaque) { + BlockJob *job = opaque; + if (block_job_is_internal(job)) { return; } - qapi_event_send_block_job_cancelled(job->driver->job_type, - job->id, - job->len, - job->offset, + qapi_event_send_block_job_cancelled(job_type(&job->job), + job->job.id, + job->job.progress_total, + job->job.progress_current, job->speed, &error_abort); } -static void block_job_event_completed(BlockJob *job, const char *msg) +static void block_job_event_completed(Notifier *n, void *opaque) { + BlockJob *job = opaque; + const char *msg = NULL; + if (block_job_is_internal(job)) { return; } - qapi_event_send_block_job_completed(job->driver->job_type, - job->id, - job->len, - job->offset, + if (job->job.ret < 0) { + msg = strerror(-job->job.ret); + } + + qapi_event_send_block_job_completed(job_type(&job->job), + job->job.id, + job->job.progress_total, + job->job.progress_current, job->speed, !!msg, msg, &error_abort); } -static int block_job_event_pending(BlockJob *job) +static void block_job_event_pending(Notifier *n, void *opaque) { - 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, - job->id, - &error_abort); + BlockJob *job = opaque; + + if (block_job_is_internal(job)) { + return; } - return 0; + + qapi_event_send_block_job_pending(job_type(&job->job), + job->job.id, + &error_abort); } +static void block_job_event_ready(Notifier *n, void *opaque) +{ + BlockJob *job = opaque; + + if (block_job_is_internal(job)) { + return; + } + + qapi_event_send_block_job_ready(job_type(&job->job), + job->job.id, + job->job.progress_total, + job->job.progress_current, + job->speed, &error_abort); +} + + /* * API for block job drivers and the block layer. These functions are * declared in blockjob_int.h. */ void *block_job_create(const char *job_id, const BlockJobDriver *driver, - BlockJobTxn *txn, BlockDriverState *bs, uint64_t perm, + JobTxn *txn, BlockDriverState *bs, uint64_t perm, uint64_t shared_perm, int64_t speed, int flags, BlockCompletionFunc *cb, void *opaque, Error **errp) { @@ -932,29 +367,8 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver, return NULL; } - if (job_id == NULL && !(flags & BLOCK_JOB_INTERNAL)) { + if (job_id == NULL && !(flags & JOB_INTERNAL)) { job_id = bdrv_get_device_name(bs); - if (!*job_id) { - error_setg(errp, "An explicit job ID is required for this node"); - return NULL; - } - } - - if (job_id) { - if (flags & BLOCK_JOB_INTERNAL) { - error_setg(errp, "Cannot specify job ID for internal block job"); - return NULL; - } - - if (!id_wellformed(job_id)) { - error_setg(errp, "Invalid job ID '%s'", job_id); - return NULL; - } - - if (block_job_get(job_id)) { - error_setg(errp, "Job ID '%s' already in use", job_id); - return NULL; - } } blk = blk_new(perm, shared_perm); @@ -964,32 +378,39 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver, return NULL; } - job = g_malloc0(driver->instance_size); - job->driver = driver; - job->id = g_strdup(job_id); - job->blk = blk; - job->cb = cb; - job->opaque = opaque; - job->busy = false; - job->paused = true; - job->pause_count = 1; - job->refcnt = 1; - job->auto_finalize = !(flags & BLOCK_JOB_MANUAL_FINALIZE); - job->auto_dismiss = !(flags & BLOCK_JOB_MANUAL_DISMISS); - block_job_state_transition(job, BLOCK_JOB_STATUS_CREATED); - aio_timer_init(qemu_get_aio_context(), &job->sleep_timer, - QEMU_CLOCK_REALTIME, SCALE_NS, - block_job_sleep_timer_cb, job); + job = job_create(job_id, &driver->job_driver, txn, blk_get_aio_context(blk), + flags, cb, opaque, errp); + if (job == NULL) { + blk_unref(blk); + return NULL; + } + + assert(is_block_job(&job->job)); + assert(job->job.driver->free == &block_job_free); + assert(job->job.driver->user_resume == &block_job_user_resume); + assert(job->job.driver->drain == &block_job_drain); + + job->blk = blk; + + job->finalize_cancelled_notifier.notify = block_job_event_cancelled; + job->finalize_completed_notifier.notify = block_job_event_completed; + job->pending_notifier.notify = block_job_event_pending; + job->ready_notifier.notify = block_job_event_ready; + + notifier_list_add(&job->job.on_finalize_cancelled, + &job->finalize_cancelled_notifier); + notifier_list_add(&job->job.on_finalize_completed, + &job->finalize_completed_notifier); + notifier_list_add(&job->job.on_pending, &job->pending_notifier); + notifier_list_add(&job->job.on_ready, &job->ready_notifier); error_setg(&job->blocker, "block device is in use by block job: %s", - BlockJobType_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; bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker); - QLIST_INSERT_HEAD(&block_jobs, job, job_list); - blk_add_aio_context_notifier(blk, block_job_attached_aio_context, block_job_detach_aio_context, job); @@ -999,198 +420,28 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver, block_job_set_speed(job, speed, &local_err); if (local_err) { - block_job_early_fail(job); + job_early_fail(&job->job); error_propagate(errp, local_err); return NULL; } } - /* Single jobs are modeled as single-job transactions for sake of - * consolidating the job management logic */ - if (!txn) { - txn = block_job_txn_new(); - block_job_txn_add_job(txn, job); - block_job_txn_unref(txn); - } else { - block_job_txn_add_job(txn, job); - } - return job; } -void block_job_early_fail(BlockJob *job) -{ - assert(job->status == BLOCK_JOB_STATUS_CREATED); - block_job_decommission(job); -} - -void block_job_completed(BlockJob *job, int ret) -{ - assert(job && job->txn && !job->completed); - assert(blk_bs(job->blk)->job == job); - job->completed = true; - job->ret = ret; - block_job_update_rc(job); - trace_block_job_completed(job, ret, job->ret); - if (job->ret) { - block_job_completed_txn_abort(job); - } else { - block_job_completed_txn_success(job); - } -} - -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 (block_job_is_cancelled(job)) { - return; - } - - if (job->driver->pause) { - job->driver->pause(job); - } - - if (block_job_should_pause(job) && !block_job_is_cancelled(job)) { - BlockJobStatus status = job->status; - block_job_state_transition(job, status == BLOCK_JOB_STATUS_READY ? \ - BLOCK_JOB_STATUS_STANDBY : \ - BLOCK_JOB_STATUS_PAUSED); - job->paused = true; - block_job_do_yield(job, -1); - job->paused = false; - block_job_state_transition(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->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->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); -} - -bool block_job_is_cancelled(BlockJob *job) -{ - return job->cancelled; -} - -void block_job_sleep_ns(BlockJob *job, int64_t ns) -{ - assert(job->busy); - - /* Check cancellation *before* setting busy = false, too! */ - if (block_job_is_cancelled(job)) { - return; - } - - if (!block_job_should_pause(job)) { - block_job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns); - } - - block_job_pause_point(job); -} - -void block_job_yield(BlockJob *job) -{ - assert(job->busy); - - /* Check cancellation *before* setting busy = false, too! */ - if (block_job_is_cancelled(job)) { - return; - } - - if (!block_job_should_pause(job)) { - block_job_do_yield(job, -1); - } - - block_job_pause_point(job); -} - 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->job.user_paused && job->job.pause_count > 0); job->iostatus = BLOCK_DEVICE_IO_STATUS_OK; } -void block_job_event_ready(BlockJob *job) +void block_job_user_resume(Job *job) { - block_job_state_transition(job, BLOCK_JOB_STATUS_READY); - job->ready = true; - - if (block_job_is_internal(job)) { - return; - } - - qapi_event_send_block_job_ready(job->driver->job_type, - job->id, - job->len, - job->offset, - job->speed, &error_abort); + BlockJob *bjob = container_of(job, BlockJob, job); + block_job_iostatus_reset(bjob); } BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err, @@ -1217,63 +468,16 @@ BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err, abort(); } if (!block_job_is_internal(job)) { - qapi_event_send_block_job_error(job->id, + qapi_event_send_block_job_error(job->job.id, is_read ? IO_OPERATION_TYPE_READ : IO_OPERATION_TYPE_WRITE, action, &error_abort); } if (action == BLOCK_ERROR_ACTION_STOP) { - block_job_pause(job); + job_pause(&job->job); /* make the pause user visible, which will be resumed from QMP. */ - job->user_paused = true; + job->job.user_paused = true; block_job_iostatus_set_err(job, error); } return action; } - -typedef struct { - BlockJob *job; - AioContext *aio_context; - BlockJobDeferToMainLoopFn *fn; - void *opaque; -} BlockJobDeferToMainLoopData; - -static void block_job_defer_to_main_loop_bh(void *opaque) -{ - BlockJobDeferToMainLoopData *data = opaque; - AioContext *aio_context; - - /* Prevent race with block_job_defer_to_main_loop() */ - aio_context_acquire(data->aio_context); - - /* Fetch BDS AioContext again, in case it has changed */ - aio_context = blk_get_aio_context(data->job->blk); - if (aio_context != data->aio_context) { - aio_context_acquire(aio_context); - } - - data->fn(data->job, data->opaque); - - if (aio_context != data->aio_context) { - aio_context_release(aio_context); - } - - aio_context_release(data->aio_context); - - g_free(data); -} - -void block_job_defer_to_main_loop(BlockJob *job, - BlockJobDeferToMainLoopFn *fn, - void *opaque) -{ - BlockJobDeferToMainLoopData *data = g_malloc(sizeof(*data)); - data->job = job; - data->aio_context = blk_get_aio_context(job->blk); - data->fn = fn; - data->opaque = opaque; - job->deferred_to_main_loop = true; - - aio_bh_schedule_oneshot(qemu_get_aio_context(), - block_job_defer_to_main_loop_bh, data); -} |