diff options
Diffstat (limited to 'blockjob.c')
| -rw-r--r-- | blockjob.c | 750 |
1 files changed, 416 insertions, 334 deletions
@@ -55,35 +55,20 @@ struct BlockJobTxn { static QLIST_HEAD(, BlockJob) block_jobs = QLIST_HEAD_INITIALIZER(block_jobs); -static char *child_job_get_parent_desc(BdrvChild *c) -{ - BlockJob *job = c->opaque; - return g_strdup_printf("%s job '%s'", - BlockJobType_lookup[job->driver->job_type], - job->id); -} - -static const BdrvChildRole child_job = { - .get_parent_desc = child_job_get_parent_desc, - .stay_at_node = true, -}; - -static void block_job_drained_begin(void *opaque) -{ - BlockJob *job = opaque; - block_job_pause(job); -} - -static void block_job_drained_end(void *opaque) -{ - BlockJob *job = opaque; - block_job_resume(job); -} - -static const BlockDevOps block_job_dev_ops = { - .drained_begin = block_job_drained_begin, - .drained_end = block_job_drained_end, -}; +/* + * The block job API is composed of two categories of functions. + * + * The first includes functions used by the monitor. The monitor is + * peculiar in that it accesses the block job list with block_job_get, and + * therefore needs consistency across block_job_get and the actual operation + * (e.g. block_job_set_speed). The consistency is achieved with + * aio_context_acquire/release. These functions are declared in blockjob.h. + * + * The second includes functions used by the block job drivers and sometimes + * by the core block layer. These do not care about locking, because the + * whole coroutine runs under the AioContext lock, and are declared in + * blockjob_int.h. + */ BlockJob *block_job_next(BlockJob *job) { @@ -106,6 +91,80 @@ BlockJob *block_job_get(const char *id) return NULL; } +BlockJobTxn *block_job_txn_new(void) +{ + 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; + } + + assert(!job->txn); + job->txn = txn; + + QLIST_INSERT_HEAD(&txn->jobs, job, txn_list); + block_job_txn_ref(txn); +} + +static void block_job_pause(BlockJob *job) +{ + job->pause_count++; +} + +static void block_job_resume(BlockJob *job) +{ + assert(job->pause_count > 0); + job->pause_count--; + if (job->pause_count) { + return; + } + block_job_enter(job); +} + +static 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); + +static void block_job_unref(BlockJob *job) +{ + if (--job->refcnt == 0) { + BlockDriverState *bs = blk_bs(job->blk); + 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); + QLIST_REMOVE(job, job_list); + g_free(job); + } +} + static void block_job_attached_aio_context(AioContext *new_context, void *opaque) { @@ -145,6 +204,36 @@ static void block_job_detach_aio_context(void *opaque) block_job_unref(job); } +static char *child_job_get_parent_desc(BdrvChild *c) +{ + BlockJob *job = c->opaque; + return g_strdup_printf("%s job '%s'", + BlockJobType_lookup[job->driver->job_type], + job->id); +} + +static const BdrvChildRole child_job = { + .get_parent_desc = child_job_get_parent_desc, + .stay_at_node = true, +}; + +static void block_job_drained_begin(void *opaque) +{ + BlockJob *job = opaque; + block_job_pause(job); +} + +static void block_job_drained_end(void *opaque) +{ + BlockJob *job = opaque; + block_job_resume(job); +} + +static const BlockDevOps block_job_dev_ops = { + .drained_begin = block_job_drained_begin, + .drained_end = block_job_drained_end, +}; + void block_job_remove_all_bdrv(BlockJob *job) { GSList *l; @@ -175,90 +264,6 @@ int block_job_add_bdrv(BlockJob *job, const char *name, BlockDriverState *bs, return 0; } -void *block_job_create(const char *job_id, const BlockJobDriver *driver, - BlockDriverState *bs, uint64_t perm, - uint64_t shared_perm, int64_t speed, int flags, - BlockCompletionFunc *cb, void *opaque, Error **errp) -{ - BlockBackend *blk; - BlockJob *job; - int ret; - - if (bs->job) { - error_setg(errp, QERR_DEVICE_IN_USE, bdrv_get_device_name(bs)); - return NULL; - } - - if (job_id == NULL && !(flags & BLOCK_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); - ret = blk_insert_bs(blk, bs, errp); - if (ret < 0) { - blk_unref(blk); - 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; - - error_setg(&job->blocker, "block device is in use by block job: %s", - BlockJobType_lookup[driver->job_type]); - block_job_add_bdrv(job, "main node", bs, 0, BLK_PERM_ALL, &error_abort); - bs->job = job; - - blk_set_dev_ops(blk, &block_job_dev_ops, 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); - - /* Only set speed when necessary to avoid NotSupported error */ - if (speed != 0) { - Error *local_err = NULL; - - block_job_set_speed(job, speed, &local_err); - if (local_err) { - block_job_unref(job); - error_propagate(errp, local_err); - return NULL; - } - } - return job; -} - bool block_job_is_internal(BlockJob *job) { return (job->id == NULL); @@ -293,30 +298,10 @@ void block_job_start(BlockJob *job) bdrv_coroutine_enter(blk_bs(job->blk), job->co); } -void block_job_ref(BlockJob *job) -{ - ++job->refcnt; -} - -void block_job_unref(BlockJob *job) -{ - if (--job->refcnt == 0) { - BlockDriverState *bs = blk_bs(job->blk); - 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); - QLIST_REMOVE(job, job_list); - g_free(job); - } -} - static void block_job_completed_single(BlockJob *job) { + assert(job->completed); + if (!job->ret) { if (job->driver->commit) { job->driver->commit(job); @@ -354,11 +339,57 @@ static void block_job_completed_single(BlockJob *job) block_job_unref(job); } +static void block_job_cancel_async(BlockJob *job) +{ + 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; +} + +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, *next; + BlockJob *other_job; if (txn->aborting) { /* @@ -367,29 +398,34 @@ static void block_job_completed_txn_abort(BlockJob *job) 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 || other_job->completed) { - /* 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. */ - if (other_job != job) { - other_job->cancelled = true; - } - continue; + if (other_job != job) { + block_job_cancel_async(other_job); } - block_job_cancel_sync(other_job); - assert(other_job->completed); } - QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) { + 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_completed_single(other_job); aio_context_release(ctx); } + + block_job_txn_unref(txn); } static void block_job_completed_txn_success(BlockJob *job) @@ -416,21 +452,6 @@ static void block_job_completed_txn_success(BlockJob *job) } } -void block_job_completed(BlockJob *job, int ret) -{ - assert(blk_bs(job->blk)->job == job); - assert(!job->completed); - job->completed = true; - job->ret = ret; - if (!job->txn) { - block_job_completed_single(job); - } else if (ret < 0 || block_job_is_cancelled(job)) { - block_job_completed_txn_abort(job); - } else { - block_job_completed_txn_success(job); - } -} - void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp) { Error *local_err = NULL; @@ -462,135 +483,36 @@ void block_job_complete(BlockJob *job, Error **errp) job->driver->complete(job, errp); } -void block_job_pause(BlockJob *job) -{ - job->pause_count++; -} - void block_job_user_pause(BlockJob *job) { job->user_paused = true; block_job_pause(job); } -static bool block_job_should_pause(BlockJob *job) -{ - return job->pause_count > 0; -} - bool block_job_user_paused(BlockJob *job) { - return job ? job->user_paused : 0; -} - -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)) { - job->paused = true; - job->busy = false; - qemu_coroutine_yield(); /* wait for block_job_resume() */ - job->busy = true; - job->paused = false; - } - - if (job->driver->resume) { - job->driver->resume(job); - } -} - -void block_job_resume(BlockJob *job) -{ - assert(job->pause_count > 0); - job->pause_count--; - if (job->pause_count) { - return; - } - block_job_enter(job); + return job->user_paused; } void block_job_user_resume(BlockJob *job) { if (job && job->user_paused && job->pause_count > 0) { + block_job_iostatus_reset(job); job->user_paused = false; block_job_resume(job); } } -void block_job_enter(BlockJob *job) -{ - if (job->co && !job->busy) { - bdrv_coroutine_enter(blk_bs(job->blk), job->co); - } -} - void block_job_cancel(BlockJob *job) { if (block_job_started(job)) { - job->cancelled = true; - block_job_iostatus_reset(job); + block_job_cancel_async(job); block_job_enter(job); } else { block_job_completed(job, -ECANCELED); } } -bool block_job_is_cancelled(BlockJob *job) -{ - return job->cancelled; -} - -void block_job_iostatus_reset(BlockJob *job) -{ - job->iostatus = BLOCK_DEVICE_IO_STATUS_OK; - if (job->driver->iostatus_reset) { - job->driver->iostatus_reset(job); - } -} - -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); - - 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; -} - /* 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. */ @@ -622,42 +544,6 @@ int block_job_complete_sync(BlockJob *job, Error **errp) return block_job_finish_sync(job, &block_job_complete, errp); } -void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns) -{ - assert(job->busy); - - /* Check cancellation *before* setting busy = false, too! */ - if (block_job_is_cancelled(job)) { - return; - } - - job->busy = false; - if (!block_job_should_pause(job)) { - co_aio_sleep_ns(blk_get_aio_context(job->blk), type, ns); - } - job->busy = true; - - 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; - } - - job->busy = false; - if (!block_job_should_pause(job)) { - qemu_coroutine_yield(); - } - job->busy = true; - - block_job_pause_point(job); -} - BlockJobInfo *block_job_query(BlockJob *job, Error **errp) { BlockJobInfo *info; @@ -717,6 +603,236 @@ static void block_job_event_completed(BlockJob *job, const char *msg) &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, + BlockDriverState *bs, uint64_t perm, + uint64_t shared_perm, int64_t speed, int flags, + BlockCompletionFunc *cb, void *opaque, Error **errp) +{ + BlockBackend *blk; + BlockJob *job; + int ret; + + if (bs->job) { + error_setg(errp, QERR_DEVICE_IN_USE, bdrv_get_device_name(bs)); + return NULL; + } + + if (job_id == NULL && !(flags & BLOCK_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); + ret = blk_insert_bs(blk, bs, errp); + if (ret < 0) { + blk_unref(blk); + 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; + + error_setg(&job->blocker, "block device is in use by block job: %s", + BlockJobType_lookup[driver->job_type]); + block_job_add_bdrv(job, "main node", bs, 0, BLK_PERM_ALL, &error_abort); + bs->job = job; + + blk_set_dev_ops(blk, &block_job_dev_ops, 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); + + /* Only set speed when necessary to avoid NotSupported error */ + if (speed != 0) { + Error *local_err = NULL; + + block_job_set_speed(job, speed, &local_err); + if (local_err) { + block_job_unref(job); + error_propagate(errp, local_err); + return NULL; + } + } + return job; +} + +void block_job_pause_all(void) +{ + BlockJob *job = NULL; + while ((job = block_job_next(job))) { + AioContext *aio_context = blk_get_aio_context(job->blk); + + aio_context_acquire(aio_context); + block_job_pause(job); + aio_context_release(aio_context); + } +} + +void block_job_early_fail(BlockJob *job) +{ + block_job_unref(job); +} + +void block_job_completed(BlockJob *job, int ret) +{ + assert(blk_bs(job->blk)->job == job); + assert(!job->completed); + job->completed = true; + job->ret = ret; + if (!job->txn) { + block_job_completed_single(job); + } else if (ret < 0 || block_job_is_cancelled(job)) { + 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; +} + +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)) { + job->paused = true; + job->busy = false; + qemu_coroutine_yield(); /* wait for block_job_resume() */ + job->busy = true; + job->paused = false; + } + + if (job->driver->resume) { + job->driver->resume(job); + } +} + +void block_job_resume_all(void) +{ + BlockJob *job = NULL; + while ((job = block_job_next(job))) { + AioContext *aio_context = blk_get_aio_context(job->blk); + + aio_context_acquire(aio_context); + block_job_resume(job); + aio_context_release(aio_context); + } +} + +void block_job_enter(BlockJob *job) +{ + if (!block_job_started(job)) { + return; + } + if (job->deferred_to_main_loop) { + return; + } + + if (!job->busy) { + bdrv_coroutine_enter(blk_bs(job->blk), job->co); + } +} + +bool block_job_is_cancelled(BlockJob *job) +{ + return job->cancelled; +} + +void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns) +{ + assert(job->busy); + + /* Check cancellation *before* setting busy = false, too! */ + if (block_job_is_cancelled(job)) { + return; + } + + job->busy = false; + if (!block_job_should_pause(job)) { + co_aio_sleep_ns(blk_get_aio_context(job->blk), type, ns); + } + job->busy = true; + + 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; + } + + job->busy = false; + if (!block_job_should_pause(job)) { + qemu_coroutine_yield(); + } + job->busy = true; + + 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); + job->iostatus = BLOCK_DEVICE_IO_STATUS_OK; +} + void block_job_event_ready(BlockJob *job) { job->ready = true; @@ -790,7 +906,6 @@ static void block_job_defer_to_main_loop_bh(void *opaque) aio_context_acquire(aio_context); } - data->job->deferred_to_main_loop = false; data->fn(data->job, data->opaque); if (aio_context != data->aio_context) { @@ -816,36 +931,3 @@ void block_job_defer_to_main_loop(BlockJob *job, aio_bh_schedule_oneshot(qemu_get_aio_context(), block_job_defer_to_main_loop_bh, data); } - -BlockJobTxn *block_job_txn_new(void) -{ - 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; - } - - assert(!job->txn); - job->txn = txn; - - QLIST_INSERT_HEAD(&txn->jobs, job, txn_list); - block_job_txn_ref(txn); -} |