Re: [Qemu-devel] [PATCH 06/11] coroutine-lock: make CoMutex thread-safe

[Fam Zheng <famz@redhat.com>]

On Fri, 04/15 13:32, Paolo Bonzini wrote:
> +/* The wait records are handled with a multiple-producer, single-consumer
> + * lock-free queue.  There cannot be two concurrent pop_waiter() calls
> + * because pop_waiter() can only come when mutex->handoff is zero.  This can
> + * happen in three cases:
> + * - in qemu_co_mutex_unlock, before the hand-off protocol has started.
> + *   In this case, qemu_co_mutex_lock will see mutex->handoff == 0 and
> + *   not take part in the handoff.
> + * - in qemu_co_mutex_lock, if it steals the hand-off responsibility from
> + *   qemu_co_mutex_unlock.  In this case, qemu_co_mutex_unlock will fail
> + *   the cmpxchg (it will see either 0 or the next sequence value) and
> + *   exit.  The next hand-off cannot begin until qemu_co_mutex_lock has
> + *   woken up someone.
> + * - in qemu_co_mutex_unlock, if it takes the hand-off token itself.
> + *   In this case another iterations starts with mutex->handoff == 0;

s/iterations/iteration/

> + *   a concurrent qemu_co_mutex_lock will fail the cmpxchg, and
> + *   qemu_co_mutex_unlock will go back to case (1).
> + *
> + * The following functions manage this queue.
> + */
> +typedef struct CoWaitRecord {
> +    Coroutine *co;
> +    QSLIST_ENTRY(CoWaitRecord) next;
> +} CoWaitRecord;
> +

<snip>

> @@ -132,12 +223,50 @@ void coroutine_fn qemu_co_mutex_unlock(CoMutex *mutex)
>  
>      trace_qemu_co_mutex_unlock_entry(mutex, self);
>  
> -    assert(mutex->locked == true);
> +    assert(mutex->locked);
>      assert(qemu_in_coroutine());
>  
> -    mutex->locked = false;
> -    qemu_co_queue_next(&mutex->queue);
> +    if (atomic_fetch_dec(&mutex->locked) == 1) {
> +        /* No waiting qemu_co_mutex_lock().  Pfew, that was easy!  */
> +        return;
> +    }
> +
> +    for (;;) {
> +        CoWaitRecord *to_wake = pop_waiter(mutex);
> +        unsigned our_handoff;
> +
> +        if (to_wake) {
> +            Coroutine *co = to_wake->co;
> +            qemu_coroutine_wake(co->ctx, co);
> +            goto out;
> +        }
> +
> +        /* Some concurrent lock() is in progress (we know this because of
> +         * count) but it hasn't yet put itself on the wait queue.

Unlike OSv's lfmutex.cc, we don't seem to have count. Should the comment say
"locked" instead?

> +         * Pick a sequence number for the handoff protocol (not 0).
> +         */
> +        if (++mutex->sequence == 0) {
> +            mutex->sequence = 1;
> +        }
> +
> +        our_handoff = mutex->sequence;
> +        atomic_mb_set(&mutex->handoff, our_handoff);
> +        if (!has_waiters(mutex)) {
> +            /* The concurrent lock has not added itself yet, so it
> +             * will be able to pick our handoff.
> +             */
> +            goto out;
> +        }
> +
> +        /* Try to do the handoff protocol ourselves; if somebody else has
> +         * already taken it, however, we're done and they're responsible.
> +         */
> +        if (atomic_cmpxchg(&mutex->handoff, our_handoff, 0) != our_handoff) {
> +            goto out;
> +        }
> +    }
>  
> +out:
>      trace_qemu_co_mutex_unlock_return(mutex, self);
>  }
>  
> -- 
> 2.5.5
> 
>