On 09/15/2015 02:11 AM, Fam Zheng wrote:
> Sometimes block jobs must execute as a transaction group. Finishing
> jobs wait until all other jobs are ready to complete successfully.
> Failure or cancellation of one job cancels the other jobs in the group.
>
> Signed-off-by: Stefan Hajnoczi <stefa...@redhat.com>
> [Rewrite the implementation which is now contained in block_job_completed.
> --Fam]
> Signed-off-by: Fam Zheng <f...@redhat.com>
> ---
> blockjob.c | 135
> ++++++++++++++++++++++++++++++++++++++++++++++-
> include/block/block.h | 1 +
> include/block/blockjob.h | 38 +++++++++++++
> 3 files changed, 172 insertions(+), 2 deletions(-)
>
> diff --git a/blockjob.c b/blockjob.c
> index 36c18e0..91e8d3c 100644
> --- a/blockjob.c
> +++ b/blockjob.c
> @@ -36,6 +36,19 @@
> #include "qemu/timer.h"
> #include "qapi-event.h"
>
> +/* 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;
> +};
> +
> void *block_job_create(const BlockJobDriver *driver, BlockDriverState *bs,
> int64_t speed, BlockCompletionFunc *cb,
> void *opaque, Error **errp)
> @@ -90,6 +103,86 @@ void block_job_unref(BlockJob *job)
> }
> }
>
> +static void block_job_completed_single(BlockJob *job)
> +{
> + if (!job->ret) {
> + if (job->driver->commit) {
> + job->driver->commit(job);
> + }
> + } else {
> + if (job->driver->abort) {
> + job->driver->abort(job);
> + }
> + }
> + job->cb(job->opaque, job->ret);
> + if (job->txn) {
> + block_job_txn_unref(job->txn);
> + }
> + block_job_unref(job);
> +}
> +
> +static void block_job_completed_txn_abort(BlockJob *job)
> +{
> + AioContext *ctx;
> + BlockJobTxn *txn = job->txn;
> + BlockJob *other_job, *next;
> +
> + if (txn->aborting) {
> + /*
> + * We are cancelled by another job, which will handle everything.
> + */
> + return;
> + }
> + txn->aborting = true;
> + /* We are the first failed job. Cancel other jobs. */
> + QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
> + ctx = bdrv_get_aio_context(other_job->bs);
> + aio_context_acquire(ctx);
> + }
> + 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;
> + }
This loop reads strangely to me due to this structure:
if (other_job == job || ...) {
if (other_job != job) {
...
}
}
why not:
foreach(...) {
if (other_job == job) {
continue;
}
if (other_job->completed) {
other_job->cancelled = true;
continue;
}
/* job is neither ours, nor completed */
block_job_cancel_sync(other_job);
assert(other_job->completed);
}
> + block_job_cancel_sync(other_job);
> + assert(other_job->completed);
> + }
> + QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
> + ctx = bdrv_get_aio_context(other_job->bs);
> + block_job_completed_single(other_job);
> + aio_context_release(ctx);
> + }
> +}
> +
> +static void block_job_completed_txn_success(BlockJob *job)
> +{
> + AioContext *ctx;
> + BlockJobTxn *txn = job->txn;
> + BlockJob *other_job, *next;
> + /*
> + * 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;
> + }
> + }
> + /* We are the last completed job, commit the transaction. */
> + QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
> + ctx = bdrv_get_aio_context(other_job->bs);
> + aio_context_acquire(ctx);
> + assert(other_job->ret == 0);
Here we assert that all jobs have a retcode of zero to be in the success
callback, but ...
> + block_job_completed_single(other_job);
> + aio_context_release(ctx);
> + }
> +}
> +
> void block_job_completed(BlockJob *job, int ret)
> {
> BlockDriverState *bs = job->bs;
> @@ -98,8 +191,13 @@ void block_job_completed(BlockJob *job, int ret)
> assert(!job->completed);
> job->completed = true;
> job->ret = ret;
> - job->cb(job->opaque, ret);
> - block_job_unref(job);
> + if (!job->txn) {
> + block_job_completed_single(job);
> + } else if (ret < 0 || block_job_is_cancelled(job)) {
> + block_job_completed_txn_abort(job);
> + } else {
over here it just appears we assert that ret is simply greater than or
equal to zero.
> + block_job_completed_txn_success(job);
> + }
> }
>
> void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
> @@ -398,3 +496,36 @@ void block_job_defer_to_main_loop(BlockJob *job,
>
> qemu_bh_schedule(data->bh);
> }
> +
> +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->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);
> +}
I guess we don't consider the list insertion to be a reference to the
job that we need to pick up, which also keeps txn_unref minimalist.
> diff --git a/include/block/block.h b/include/block/block.h
> index f70bec4..514b233 100644
> --- a/include/block/block.h
> +++ b/include/block/block.h
> @@ -14,6 +14,7 @@ typedef struct BlockDriver BlockDriver;
> typedef struct BlockJob BlockJob;
> typedef struct BdrvChild BdrvChild;
> typedef struct BdrvChildRole BdrvChildRole;
> +typedef struct BlockJobTxn BlockJobTxn;
>
> typedef struct BlockDriverInfo {
> /* in bytes, 0 if irrelevant */
> diff --git a/include/block/blockjob.h b/include/block/blockjob.h
> index f6e4c86..4999682 100644
> --- a/include/block/blockjob.h
> +++ b/include/block/blockjob.h
> @@ -152,6 +152,9 @@ struct BlockJob {
> */
> int ret;
>
> + /** Non-NULL if this job is part of a transaction */
> + BlockJobTxn *txn;
> + QLIST_ENTRY(BlockJob) txn_list;
> };
>
> /**
> @@ -395,4 +398,39 @@ void block_job_defer_to_main_loop(BlockJob *job,
> BlockJobDeferToMainLoopFn *fn,
> void *opaque);
>
> +/**
> + * block_job_txn_new:
> + *
> + * Allocate and return a new block job transaction. Jobs can be added to the
> + * transaction using block_job_txn_add_job().
> + *
> + * The transaction is automatically freed when the last job completes or is
> + * cancelled.
> + *
> + * All jobs in the transaction either complete successfully or fail/cancel
> as a
> + * group. Jobs wait for each other before completing. Cancelling one job
> + * cancels all jobs in the transaction.
> + */
> +BlockJobTxn *block_job_txn_new(void);
> +
> +/**
> + * block_job_txn_unref:
> + *
> + * Release a reference that was previously acquired with
> block_job_txn_add_job
> + * or block_job_txn_new. If it's the last reference to the object, it will be
> + * freed.
> + */
> +void block_job_txn_unref(BlockJobTxn *txn);
> +
> +/**
> + * block_job_txn_add_job:
> + * @txn: The transaction (may be NULL)
> + * @job: Job to add to the transaction
> + *
> + * Add @job to the transaction. The @job must not already be in a
> transaction.
> + * The block job driver must call block_job_txn_unref() in the end to release
> + * the reference that is automatically grabbed here.
> + */
> +void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job);
> +
> #endif
>
