Re: [PATCHv1 4/4] spinlock: add fair read-write locks

From: "Jan Beulich" <JBeulich@suse.com>
To: David Vrabel <david.vrabel@citrix.com>,
	Jennifer Herbert <jennifer.herbert@citrix.com>
Cc: xen-devel@lists.xenproject.org, Ian Campbell <ian.campbell@citrix.com>
Subject: Re: [PATCHv1 4/4] spinlock: add fair read-write locks
Date: Tue, 22 Dec 2015 06:54:28 -0700	[thread overview]
Message-ID: <5679642402000078000C2428@prv-mh.provo.novell.com> (raw)
In-Reply-To: <1450447746-9305-5-git-send-email-david.vrabel@citrix.com>

>>> On 18.12.15 at 15:09, <david.vrabel@citrix.com> wrote:
> +/**
> + * rspin_until_writer_unlock - inc reader count & spin until writer is gone

Stale comment - the function doesn't increment anything.

Also throughout the file, with Linux coding style converted to Xen
style, comment style should be made Xen-like too.

> +    /*
> +     * Readers come here when they cannot get the lock without waiting
> +     */
> +    if ( unlikely(in_irq()) )
> +    {
> +        /*
> +         * Readers in interrupt context will spin until the lock is
> +         * available without waiting in the queue.
> +         */
> +        smp_rmb();
> +        cnts = atomic_read(&lock->cnts);
> +        rspin_until_writer_unlock(lock, cnts);
> +        return;
> +    }

I can't immediately see the reason for this re-introduction of
unfairness - can you say a word on this, or perhaps extend the
comment?

> +/**
> + * queue_write_lock_slowpath - acquire write lock of a queue rwlock
> + * @lock : Pointer to queue rwlock structure
> + */
> +void queue_write_lock_slowpath(rwlock_t *lock)
>  {
> -    _read_unlock(lock);
> -    local_irq_enable();
> -}
> +    u32 cnts;
>  
> -void _read_unlock_irqrestore(rwlock_t *lock, unsigned long flags)
> -{
> -    _read_unlock(lock);
> -    local_irq_restore(flags);
> -}
> +    /* Put the writer into the wait queue */
> +    spin_lock(&lock->lock);
>  
> -void _write_lock(rwlock_t *lock)
> -{
> -    uint32_t x;
> +    /* Try to acquire the lock directly if no reader is present */
> +    if ( !atomic_read(&lock->cnts) &&
> +         (atomic_cmpxchg(&lock->cnts, 0, _QW_LOCKED) == 0) )
> +        goto unlock;
>  
> -    check_lock(&lock->debug);
> -    do {
> -        while ( (x = lock->lock) & RW_WRITE_FLAG )
> -            cpu_relax();
> -    } while ( cmpxchg(&lock->lock, x, x|RW_WRITE_FLAG) != x );
> -    while ( x != 0 )
> +    /*
> +     * Set the waiting flag to notify readers that a writer is pending,
> +     * or wait for a previous writer to go away.
> +     */
> +    for (;;)
>      {
> -        cpu_relax();
> -        x = lock->lock & ~RW_WRITE_FLAG;
> -    }
> -    preempt_disable();
> -}
> -
> -void _write_lock_irq(rwlock_t *lock)
> -{
> -    uint32_t x;
> +        cnts = atomic_read(&lock->cnts);
> +        if ( !(cnts & _QW_WMASK) &&
> +             (atomic_cmpxchg(&lock->cnts, cnts,
> +                             cnts | _QW_WAITING) == cnts) )
> +            break;

Considering that (at least on x86) cmpxchg is relatively expensive,
and that atomic OR would be carried out by some cmpxchg-like
mechanism on most other arches anyway, can't this be an atomic
OR, followed by a read to check for another active writer?

> +unlock:
> +    spin_unlock(&lock->lock);

Labels indented by at least one space please.

Also - are you using any of the static functions in spinlock.c? If not,
putting the rwlock code in a new rwlock.c would help review quite a
bit, since code removal and code addition would then be separate
rather than intermixed.

> +/*
> + * Writer states & reader shift and bias
> + */
> +#define    _QW_WAITING  1               /* A writer is waiting     */
> +#define    _QW_LOCKED   0xff            /* A writer holds the lock */
> +#define    _QW_WMASK    0xff            /* Writer mask             */
> +#define    _QR_SHIFT    8               /* Reader count shift      */
> +#define    _QR_BIAS     (1U << _QR_SHIFT)

Is there a particular reason for the 8-bit writer mask (a 2-bit one
would seem to suffice)?

> +static inline int _write_trylock(rwlock_t *lock)
> +{
> +    u32 cnts;
> +
> +    cnts = atomic_read(&lock->cnts);
> +    if ( unlikely(cnts) )
> +        return 0;
> +
> +    return likely(atomic_cmpxchg(&lock->cnts,
> +                                 cnts, cnts | _QW_LOCKED) == cnts);

The | is pointless here considering that cnts is zero.

> +static inline void _write_unlock(rwlock_t *lock)
> +{
> +    /*
> +     * If the writer field is atomic, it can be cleared directly.
> +     * Otherwise, an atomic subtraction will be used to clear it.
> +     */
> +    atomic_sub(_QW_LOCKED, &lock->cnts);
> +}

Ah, I guess the comment here is the explanation for the 8-bit
write mask.

> +static inline int _rw_is_write_locked(rwlock_t *lock)
> +{
> +    return atomic_read(&lock->cnts) & _QW_WMASK;
> +}

This returns true for write-locked or writer-waiting - is this intended?

Jan