Re: [Qemu-devel] [PATCH v5 09/14] block: Add block job transactions

[Fam Zheng]

On Fri, 09/11 20:52, Max Reitz wrote:
> On 07.09.2015 09:34, 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 <address@hidden>
> > [Rewrite the implementation which is now contained in block_job_completed.
> > --Fam]
> > Signed-off-by: Fam Zheng <address@hidden>
> > ---
> >  blockjob.c               | 134 
> > ++++++++++++++++++++++++++++++++++++++++++++++-
> >  include/block/block.h    |   1 +
> >  include/block/blockjob.h |  38 ++++++++++++++
> >  3 files changed, 171 insertions(+), 2 deletions(-)
> > 
> > diff --git a/blockjob.c b/blockjob.c
> > index 36c18e0..1b4eb20 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,85 @@ 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);
> > +    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.
> > +         */
> 
> After having reviewed patch 11, I think we need to call
> block_job_txn_unref(txn) here.
> 
> > +        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;
> > +        }
> > +        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);
> > +    }
> > +    block_job_txn_unref(txn);
> > +}
> > +
> > +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) {
> 
> Same here.

Yes. Will add at ...

> > +            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);
> > +        block_job_completed_single(other_job);
> > +        aio_context_release(ctx);
> > +    }
> > +    block_job_txn_unref(txn);
> > +}
> > +
> >  void block_job_completed(BlockJob *job, int ret)
> >  {
> >      BlockDriverState *bs = job->bs;
> > @@ -98,8 +190,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 {
> > +        block_job_completed_txn_success(job);
> > +    }

Here in an "if (job->txn)" branch.

Fam

> >  }