Re: [Qemu-devel] [PATCH 09/10] blockjob: reorganize block_job_completed_txn_abort

[Paolo Bonzini]

On 08/04/2017 09:15, John Snow wrote:
> 
> 
> On 03/23/2017 01:39 PM, Paolo Bonzini wrote:
>> This splits the part that touches job states from the part that invokes
>> callbacks.  It will be a bit simpler to understand once job states will
>> be protected by a different mutex than the AioContext lock.
>>
> 
> Maybe easier to review then, too :)

The idea is that the "touching job states" part (block_job_cancel_async)
doesn't release the mutex, while the "invokes callbacks" functions
(block_job_finish_sync, block_job_completed_single) part does.  It is
much simpler to understand if all block_job_cancel_asyncs are done first.

I should once more split code movement from changes, but I think this
patch is still reviewable on its own.  All the "introduce block job
mutex" does in block_job_completed_txn_abort is remove all
aio_context_acquire and release calls.

Paolo

>> Signed-off-by: Paolo Bonzini <address@hidden>
>> ---
>>  blockjob.c | 165 
>> ++++++++++++++++++++++++++++++++-----------------------------
>>  1 file changed, 88 insertions(+), 77 deletions(-)
>>
>> diff --git a/blockjob.c b/blockjob.c
>> index 093962b..3fa2885 100644
>> --- a/blockjob.c
>> +++ b/blockjob.c
>> @@ -76,6 +76,39 @@ 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++;
>> @@ -336,6 +369,8 @@ void block_job_start(BlockJob *job)
>>  
>>  static void block_job_completed_single(BlockJob *job)
>>  {
>> +    assert(job->completed);
>> +
>>      if (!job->ret) {
>>          if (job->driver->commit) {
>>              job->driver->commit(job);
>> @@ -376,14 +411,49 @@ static void block_job_completed_single(BlockJob *job)
>>  static void block_job_cancel_async(BlockJob *job)
>>  {
>>      job->cancelled = true;
>> -    block_job_iostatus_reset(job);
>> +    if (!job->completed) {
>> +        block_job_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);
>> +
>> +    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) {
>>          /*
>> @@ -392,29 +462,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) {
>> -                block_job_cancel_async(other_job);
>> -            }
>> -            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)
>> @@ -502,37 +577,6 @@ void block_job_cancel(BlockJob *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. */
>> @@ -856,36 +900,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);
>> -}
>>