Re: [Qemu-devel] [PATCH v3 10/15] block: add block job transactions

[John Snow]

On 07/09/2015 11:46 PM, Fam Zheng wrote:
> From: Stefan Hajnoczi <address@hidden>
> 
> 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 <address@hidden>
> [Rewrite the implementation which is now contained in block_job_completed.
> --Fam]
> Signed-off-by: Fam Zheng <address@hidden>
> ---
>  blockjob.c               | 86 
> ++++++++++++++++++++++++++++++++++++++++++++++++
>  include/block/block.h    |  1 +
>  include/block/blockjob.h | 26 +++++++++++++++
>  3 files changed, 113 insertions(+)
> 
> diff --git a/blockjob.c b/blockjob.c
> index e057dd5..7b59b53 100644
> --- a/blockjob.c
> +++ b/blockjob.c
> @@ -36,6 +36,16 @@
>  #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;
> +};
> +
>  void *block_job_create(const BlockJobDriver *driver, BlockDriverState *bs,
>                         int64_t speed, BlockCompletionFunc *cb,
>                         void *opaque, Error **errp)
> @@ -84,6 +94,59 @@ void block_job_release(BlockDriverState *bs)
>      g_free(job);
>  }
>  
> +static void block_job_completed_txn(BlockJobTxn *txn, BlockJob *job, int ret)
> +{
> +    AioContext *ctx;
> +    BlockJob *other_job, *next;
> +    if (ret < 0 || block_job_is_cancelled(job)) {
> +        if (!txn->aborting) {
> +            /* We are the first failed job. Cancel other jobs. */

What guarantee against a race is there at this point? Something to do
with deferring back to the main loop before we call completion?

Do jobs always complete in the main thread?

> +            txn->aborting = true;
> +            QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
> +                if (other_job == job || other_job->completed) {
> +                    continue;
> +                }
> +                ctx = bdrv_get_aio_context(other_job->bs);
> +                aio_context_acquire(ctx);
> +                block_job_cancel_sync(other_job);
> +                assert(other_job->completed);
> +                aio_context_release(ctx);
> +            }
> +            QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
> +                if (other_job->driver->txn_abort) {
> +                    other_job->driver->txn_abort(other_job);
> +                }
> +                other_job->cb(other_job->opaque,
> +                              other_job->ret ? : -ECANCELED);
> +                block_job_release(other_job->bs);
> +            }
> +        } else {
> +            /*
> +             * We are cancelled by another job, who will handle everything.
> +             */
> +            return;
> +        }
> +    } else {
> +        /*
> +         * Successful completion, see if there is other running jobs in this
> +         * txn. */

s/is/are/

> +        QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
> +            if (!other_job->completed) {
> +                return;
> +            }
> +        }

Clever (though quadratic!) :~)

> +        /* We are the last completed job, commit the transaction. */
> +        QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
> +            if (other_job->driver->txn_commit) {
> +                other_job->driver->txn_commit(other_job);
> +            }
> +            assert(other_job->ret == 0);
> +            other_job->cb(other_job->opaque, 0);
> +            block_job_release(other_job->bs);

Worth creating some static local/inline routine and sharing it with
block_job_completed?

> +        }
> +    }
> +}
> +
>  void block_job_completed(BlockJob *job, int ret)
>  {
>      BlockDriverState *bs = job->bs;
> @@ -96,6 +159,10 @@ void block_job_completed(BlockJob *job, int ret)
>          qemu_bh_delete(job->defer_to_main_loop_data.bh);
>          job->defer_to_main_loop_data.bh = NULL;
>      }
> +    if (job->txn) {
> +        block_job_completed_txn(job->txn, job, ret);
> +        return;
> +    }
>      job->cb(job->opaque, ret);
>      block_job_release(bs);
>  }
> @@ -384,3 +451,22 @@ 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);
> +    return 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);
> +}
> diff --git a/include/block/block.h b/include/block/block.h
> index 7437590..c7fc5b6 100644
> --- a/include/block/block.h
> +++ b/include/block/block.h
> @@ -13,6 +13,7 @@
>  typedef struct BlockDriver BlockDriver;
>  typedef struct BlockJob BlockJob;
>  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 5bac2e2..d854ee0 100644
> --- a/include/block/blockjob.h
> +++ b/include/block/blockjob.h
> @@ -157,6 +157,9 @@ struct BlockJob {
>       */
>      int ret;
>  
> +    /** Non-NULL if this job is part of a transaction */
> +    BlockJobTxn *txn;
> +    QLIST_ENTRY(BlockJob) txn_list;
>  };
>  
>  /**
> @@ -389,4 +392,27 @@ 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().
> + *
> + * 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_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_prepare_to_complete() before
> + * final cleanup and completion.
> + */
> +void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job);
> +
>  #endif
> 

This definitely combines the best of our approaches AND has less code to
boot. Great!

Nitpicks and comment spelling aside:
Reviewed-by: John Snow <address@hidden>

--
(linguistic postscript: I don't know how to say "Stefan and I's patches"
because I think 'I' is incorrect here because the subject of the
sentence is the patches, not Stefan or myself. I think it's some variant
on "me" "my" or "mine," but can't quite work it out. English is hard, or
I'm stupid, or both.)