[PATCHv1 0/4] spinlock: add qrwlocks

[PATCHv1 0/4] spinlock: add qrwlocks

[David Vrabel]

This series adds the qrwlocks from Linux.  These are fair; under
contention both readers and writers will be queued and obtain the lock
in FIFO order (due to the fairness of the internal ticket lock).

The implementation is all in C and thus architecture independent.

Compared to the Linux implementation some of the memory barrier
primitives were changed.  The ARM maintainers may want to give this
area a careful review.

David

[PATCHv1 1/4] atomic: replace atomic_compareandswap() with atomic_cmpxchg()

[David Vrabel]

atomic_compareandswap() used atomic_t as the new, old and returned
values which is less convinient than using just int.

Signed-off-by: David Vrabel <david.vrabel@citrix.com>
---
 xen/common/domain.c          |  5 +----
 xen/include/asm-arm/atomic.h |  8 --------
 xen/include/asm-x86/atomic.h |  8 --------
 xen/include/xen/atomic.h     | 43 +++++++++++++++++++++++++++++++++++++++++++
 xen/include/xen/sched.h      | 11 +++++------
 5 files changed, 49 insertions(+), 26 deletions(-)
 create mode 100644 xen/include/xen/atomic.h

diff --git a/xen/common/domain.c b/xen/common/domain.c
index c4661d8..3960f06 100644
--- a/xen/common/domain.c
+++ b/xen/common/domain.c
@@ -851,14 +851,11 @@ static void complete_domain_destroy(struct rcu_head *head)
 void domain_destroy(struct domain *d)
 {
     struct domain **pd;
-    atomic_t old = ATOMIC_INIT(0);
-    atomic_t new = ATOMIC_INIT(DOMAIN_DESTROYED);
 
     BUG_ON(!d->is_dying);
 
     /* May be already destroyed, or get_domain() can race us. */
-    old = atomic_compareandswap(old, new, &d->refcnt);
-    if ( _atomic_read(old) != 0 )
+    if ( atomic_cmpxchg(&d->refcnt, 0, DOMAIN_DESTROYED) != 0 )
         return;
 
     cpupool_rm_domain(d);
diff --git a/xen/include/asm-arm/atomic.h b/xen/include/asm-arm/atomic.h
index 5a38c67..29ab265 100644
--- a/xen/include/asm-arm/atomic.h
+++ b/xen/include/asm-arm/atomic.h
@@ -138,14 +138,6 @@ static inline void _atomic_set(atomic_t *v, int i)
 # error "unknown ARM variant"
 #endif
 
-static inline atomic_t atomic_compareandswap(
-    atomic_t old, atomic_t new, atomic_t *v)
-{
-    atomic_t rc;
-    rc.counter = __cmpxchg(&v->counter, old.counter, new.counter, sizeof(int));
-    return rc;
-}
-
 #endif /* __ARCH_ARM_ATOMIC__ */
 /*
  * Local variables:
diff --git a/xen/include/asm-x86/atomic.h b/xen/include/asm-x86/atomic.h
index 2b8c877..a40319e 100644
--- a/xen/include/asm-x86/atomic.h
+++ b/xen/include/asm-x86/atomic.h
@@ -272,12 +272,4 @@ static inline int atomic_add_negative(int i, atomic_t *v)
     return c;
 }
 
-static inline atomic_t atomic_compareandswap(
-    atomic_t old, atomic_t new, atomic_t *v)
-{
-    atomic_t rc;
-    rc.counter = __cmpxchg(&v->counter, old.counter, new.counter, sizeof(int));
-    return rc;
-}
-
 #endif /* __ARCH_X86_ATOMIC__ */
diff --git a/xen/include/xen/atomic.h b/xen/include/xen/atomic.h
new file mode 100644
index 0000000..387ac04
--- /dev/null
+++ b/xen/include/xen/atomic.h
@@ -0,0 +1,43 @@
+/******************************************************************************
+ * atomic.h
+ *
+ * Generic atomic variables.
+ *
+ * Copyright (c) 2015 Citrix R&D Ltd.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+#ifndef __ATOMIC_H__
+#define __ATOMIC_H__
+
+#include <asm/atomic.h>
+
+static inline int atomic_cmpxchg(atomic_t *v, int old, int new)
+{
+    return cmpxchg(&v->counter, old, new);
+}
+
+/**
+ * atomic_add_return - add integer and return
+ * @i: integer value to add
+ * @v: pointer of type atomic_t
+ *
+ * Atomically adds @i to @v and returns @i + @v
+ */
+static inline int atomic_add_return(int i, atomic_t *v)
+{
+    return i + arch_fetch_and_add(&v->counter, i);
+}
+
+#endif /* __ATOMIC_H__ */
diff --git a/xen/include/xen/sched.h b/xen/include/xen/sched.h
index 3729b0f..cca5547 100644
--- a/xen/include/xen/sched.h
+++ b/xen/include/xen/sched.h
@@ -17,7 +17,7 @@
 #include <xen/mm.h>
 #include <xen/smp.h>
 #include <xen/perfc.h>
-#include <asm/atomic.h>
+#include <xen/atomic.h>
 #include <xen/wait.h>
 #include <public/xen.h>
 #include <public/domctl.h>
@@ -481,16 +481,15 @@ extern struct vcpu *idle_vcpu[NR_CPUS];
  */
 static always_inline int get_domain(struct domain *d)
 {
-    atomic_t old, new, seen = d->refcnt;
+    int old, seen = atomic_read(&d->refcnt);
     do
     {
         old = seen;
-        if ( unlikely(_atomic_read(old) & DOMAIN_DESTROYED) )
+        if ( unlikely(old & DOMAIN_DESTROYED) )
             return 0;
-        _atomic_set(&new, _atomic_read(old) + 1);
-        seen = atomic_compareandswap(old, new, &d->refcnt);
+        seen = atomic_cmpxchg(&d->refcnt, old, old + 1);
     }
-    while ( unlikely(_atomic_read(seen) != _atomic_read(old)) );
+    while ( unlikely(seen != old) );
     return 1;
 }
 
-- 
2.1.4

Re: [PATCHv1 1/4] atomic: replace atomic_compareandswap() with atomic_cmpxchg()

[Jan Beulich]

>>> On 18.12.15 at 15:09, <david.vrabel@citrix.com> wrote:
> atomic_compareandswap() used atomic_t as the new, old and returned
> values which is less convinient than using just int.
> 
> Signed-off-by: David Vrabel <david.vrabel@citrix.com>

Reviewed-by: Jan Beulich <jbeulich@suse.com>

Re: [PATCHv1 1/4] atomic: replace atomic_compareandswap() with atomic_cmpxchg()

[Ian Campbell]

On Fri, 2015-12-18 at 14:09 +0000, David Vrabel wrote:
> atomic_compareandswap() used atomic_t as the new, old and returned
> values which is less convinient than using just int.

"convenient"

diff --git a/xen/include/asm-arm/atomic.h b/xen/include/asm-arm/atomic.h
> index 5a38c67..29ab265 100644
> --- a/xen/include/asm-arm/atomic.h
> +++ b/xen/include/asm-arm/atomic.h
> @@ -138,14 +138,6 @@ static inline void _atomic_set(atomic_t *v, int i)
>  # error "unknown ARM variant"
>  #endif
>  
> -static inline atomic_t atomic_compareandswap(
> -    atomic_t old, atomic_t new, atomic_t *v)
> -{
> -    atomic_t rc;
> -    rc.counter = __cmpxchg(&v->counter, old.counter, new.counter,
> sizeof(int));
> -    return rc;
> -}
> -

ARM: Acked-by: Ian Campbell <ian.campbell@citrix.com>


_______________________________________________
Xen-devel mailing list
Xen-devel@lists.xen.org
http://lists.xen.org/xen-devel

[PATCHv1 2/4] x86/domain: Compile with lock_profile=y enabled.

[David Vrabel]

From: Konrad Rzeszutek Wilk <konrad.wilk@oracle.com>

Our 'struct domain' has when lock profiling is enabled is bigger than
one page.

We can't use vmap nor vzalloc as both of those stash the
physical address in struct page which makes the assumptions
in 'arch_init_memory' trip over ASSERTs.

Signed-off-by: Konrad Rzeszutek Wilk <konrad.wilk@oracle.com>
Signed-off-by: David Vrabel <david.vrabel@citrix.com>
---
 xen/arch/x86/domain.c | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/xen/arch/x86/domain.c b/xen/arch/x86/domain.c
index 2c3bb09..40d9d7c 100644
--- a/xen/arch/x86/domain.c
+++ b/xen/arch/x86/domain.c
@@ -231,6 +231,7 @@ static unsigned int __init noinline _domain_struct_bits(void)
 struct domain *alloc_domain_struct(void)
 {
     struct domain *d;
+    unsigned int order = get_order_from_bytes(sizeof(*d));
 #ifdef CONFIG_BIGMEM
     const unsigned int bits = 0;
 #else
@@ -244,10 +245,18 @@ struct domain *alloc_domain_struct(void)
          bits = _domain_struct_bits();
 #endif
 
+
+#ifndef LOCK_PROFILE
     BUILD_BUG_ON(sizeof(*d) > PAGE_SIZE);
-    d = alloc_xenheap_pages(0, MEMF_bits(bits));
+#endif
+    d = alloc_xenheap_pages(order, MEMF_bits(bits));
     if ( d != NULL )
-        clear_page(d);
+    {
+        unsigned int sz;
+
+        for ( sz = 0; sz < (PAGE_SIZE << order); sz += PAGE_SIZE )
+            clear_page((void *)d + sz);
+    }
     return d;
 }
 
-- 
2.1.4

[PATCHv1 3/4] spinlock: shrink struct lock_debug

[David Vrabel]

From: Jennifer Herbert <jennifer.herbert@citrix.com>

Reduce the size of struct lock_debug so increases in other lock
structures don't increase the size of struct domain too much.

Signed-off-by: Jennifer Herbert <jennifer.herbert@citrix.com>
Signed-off-by: David Vrabel <david.vrabel@citrix.com>
---
 xen/common/spinlock.c      | 2 +-
 xen/include/xen/spinlock.h | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/xen/common/spinlock.c b/xen/common/spinlock.c
index 7f89694..c129a88 100644
--- a/xen/common/spinlock.c
+++ b/xen/common/spinlock.c
@@ -16,7 +16,7 @@ static atomic_t spin_debug __read_mostly = ATOMIC_INIT(0);
 
 static void check_lock(struct lock_debug *debug)
 {
-    int irq_safe = !local_irq_is_enabled();
+    s16 irq_safe = !local_irq_is_enabled();
 
     if ( unlikely(atomic_read(&spin_debug) <= 0) )
         return;
diff --git a/xen/include/xen/spinlock.h b/xen/include/xen/spinlock.h
index fb0438e..5e54407 100644
--- a/xen/include/xen/spinlock.h
+++ b/xen/include/xen/spinlock.h
@@ -6,7 +6,7 @@
 
 #ifndef NDEBUG
 struct lock_debug {
-    int irq_safe; /* +1: IRQ-safe; 0: not IRQ-safe; -1: don't know yet */
+    s16 irq_safe; /* +1: IRQ-safe; 0: not IRQ-safe; -1: don't know yet */
 };
 #define _LOCK_DEBUG { -1 }
 void spin_debug_enable(void);
-- 
2.1.4

Re: [PATCHv1 3/4] spinlock: shrink struct lock_debug

[Jan Beulich]

>>> On 18.12.15 at 15:09, <david.vrabel@citrix.com> wrote:

> --- a/xen/common/spinlock.c
> +++ b/xen/common/spinlock.c
> @@ -16,7 +16,7 @@ static atomic_t spin_debug __read_mostly = ATOMIC_INIT(0);
>  
>  static void check_lock(struct lock_debug *debug)
>  {
> -    int irq_safe = !local_irq_is_enabled();
> +    s16 irq_safe = !local_irq_is_enabled();
>  
>      if ( unlikely(atomic_read(&spin_debug) <= 0) )
>          return;

I can't figure out why this odd looking change is needed. 

Jan

Re: [PATCHv1 3/4] spinlock: shrink struct lock_debug

[Jennifer Herbert]

On 18/12/15 16:58, Jan Beulich wrote:
>>>> On 18.12.15 at 15:09, <david.vrabel@citrix.com> wrote:
>> --- a/xen/common/spinlock.c
>> +++ b/xen/common/spinlock.c
>> @@ -16,7 +16,7 @@ static atomic_t spin_debug __read_mostly = ATOMIC_INIT(0);
>>   
>>   static void check_lock(struct lock_debug *debug)
>>   {
>> -    int irq_safe = !local_irq_is_enabled();
>> +    s16 irq_safe = !local_irq_is_enabled();
>>   
>>       if ( unlikely(atomic_read(&spin_debug) <= 0) )
>>           return;
> I can't figure out why this odd looking change is needed.
>
> Jan
>


This patch shrinks struct lock_debug's irq_safe member.  Since you end 
up with many instances of this embedded in other structures such as 
struct domain, this become a useful saving in memory.  At one point, I 
had an issue with struct domain exceeding 4k, which caused trouble.
Given the change to struct_lock debug,  this local variable irq_safe, is 
compared with it, and used in cmpxchg with it it, and therefore should 
match type.

- Jenny

Re: [PATCHv1 3/4] spinlock: shrink struct lock_debug

[Jan Beulich]

>>> On 19.01.16 at 13:31, <Jennifer.Herbert@citrix.com> wrote:
> On 18/12/15 16:58, Jan Beulich wrote:
>>>>> On 18.12.15 at 15:09, <david.vrabel@citrix.com> wrote:
>>> --- a/xen/common/spinlock.c
>>> +++ b/xen/common/spinlock.c
>>> @@ -16,7 +16,7 @@ static atomic_t spin_debug __read_mostly = ATOMIC_INIT(0);
>>>   
>>>   static void check_lock(struct lock_debug *debug)
>>>   {
>>> -    int irq_safe = !local_irq_is_enabled();
>>> +    s16 irq_safe = !local_irq_is_enabled();
>>>   
>>>       if ( unlikely(atomic_read(&spin_debug) <= 0) )
>>>           return;
>> I can't figure out why this odd looking change is needed.
> 
> This patch shrinks struct lock_debug's irq_safe member.  Since you end 
> up with many instances of this embedded in other structures such as 
> struct domain, this become a useful saving in memory.  At one point, I 
> had an issue with struct domain exceeding 4k, which caused trouble.
> Given the change to struct_lock debug,  this local variable irq_safe, is 
> compared with it, and used in cmpxchg with it it, and therefore should 
> match type.

I don't think the type of the local variable matters for the cmpxchg();
note how the literal -1 is also not of type s16. Also if what you say
was to be consistent, the type of "seen" would need to be changed
too. Afaict all that can happen from the proposed type change is
worse generated code (which doesn't matter much as it's debug-only
code, but anyway).

Jan

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

[David Vrabel]

From: Jennifer Herbert <jennifer.herbert@citrix.com>

The current rwlocks are write-biased and unfair.  This allows writers
to starve readers in situations where there are many writers (e.g.,
p2m type changes from log dirty updates during domain save).

Introduce queued read-write locks which use a fair spinlock (a ticket
lock in this case) to ensure fairness between readers and writers when
they are contended.

This implementation is from the Linux commit 70af2f8a4f48 by Waiman
Long and Peter Zijlstra.

    locking/rwlocks: Introduce 'qrwlocks' - fair, queued rwlocks

    This rwlock uses the arch_spin_lock_t as a waitqueue, and assuming
    the arch_spin_lock_t is a fair lock (ticket,mcs etc..) the
    resulting rwlock is a fair lock.

    It fits in the same 8 bytes as the regular rwlock_t by folding the
    reader and writer count into a single integer, using the remaining
    4 bytes for the arch_spinlock_t.

    Architectures that can single-copy adress bytes can optimize
    queue_write_unlock() with a 0 write to the LSB (the write count).

We do not yet make use of the architecture-specific optimization noted
above.

Signed-off-by: Jennifer Herbert <jennifer.herbert@citrix.com>
Signed-off-by: David Vrabel <david.vrabel@citrix.com>
---
 xen/common/spinlock.c      | 268 +++++++++++++++------------------------------
 xen/include/xen/spinlock.h | 195 +++++++++++++++++++++++++++++----
 2 files changed, 260 insertions(+), 203 deletions(-)

diff --git a/xen/common/spinlock.c b/xen/common/spinlock.c
index c129a88..9cf4136 100644
--- a/xen/common/spinlock.c
+++ b/xen/common/spinlock.c
@@ -288,208 +288,116 @@ void _spin_unlock_recursive(spinlock_t *lock)
     }
 }
 
-void _read_lock(rwlock_t *lock)
-{
-    uint32_t x;
-
-    check_lock(&lock->debug);
-    do {
-        while ( (x = lock->lock) & RW_WRITE_FLAG )
-            cpu_relax();
-    } while ( cmpxchg(&lock->lock, x, x+1) != x );
-    preempt_disable();
-}
-
-void _read_lock_irq(rwlock_t *lock)
-{
-    uint32_t x;
 
-    ASSERT(local_irq_is_enabled());
-    local_irq_disable();
-    check_lock(&lock->debug);
-    do {
-        if ( (x = lock->lock) & RW_WRITE_FLAG )
-        {
-            local_irq_enable();
-            while ( (x = lock->lock) & RW_WRITE_FLAG )
-                cpu_relax();
-            local_irq_disable();
-        }
-    } while ( cmpxchg(&lock->lock, x, x+1) != x );
-    preempt_disable();
+/**
+ * rspin_until_writer_unlock - inc reader count & spin until writer is gone
+ * @lock  : Pointer to queue rwlock structure
+ * @writer: Current queue rwlock writer status byte
+ *
+ * In interrupt context or at the head of the queue, the reader will just
+ * increment the reader count & wait until the writer releases the lock.
+ */
+static inline void rspin_until_writer_unlock(rwlock_t *lock, u32 cnts)
+{
+    while ( (cnts & _QW_WMASK) == _QW_LOCKED )
+    {
+        cpu_relax();
+        smp_rmb();
+        cnts = atomic_read(&lock->cnts);
+    }
 }
 
-unsigned long _read_lock_irqsave(rwlock_t *lock)
+/**
+ * queue_read_lock_slowpath - acquire read lock of a queue rwlock
+ * @lock: Pointer to queue rwlock structure
+ */
+void queue_read_lock_slowpath(rwlock_t *lock)
 {
-    uint32_t x;
-    unsigned long flags;
-
-    local_irq_save(flags);
-    check_lock(&lock->debug);
-    do {
-        if ( (x = lock->lock) & RW_WRITE_FLAG )
-        {
-            local_irq_restore(flags);
-            while ( (x = lock->lock) & RW_WRITE_FLAG )
-                cpu_relax();
-            local_irq_disable();
-        }
-    } while ( cmpxchg(&lock->lock, x, x+1) != x );
-    preempt_disable();
-    return flags;
-}
+    u32 cnts;
+    /*
+     * 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;
+    }
+    atomic_sub(_QR_BIAS, &lock->cnts);
 
-int _read_trylock(rwlock_t *lock)
-{
-    uint32_t x;
+    /*
+     * Put the reader into the wait queue
+     */
+    spin_lock(&lock->lock);
 
-    check_lock(&lock->debug);
-    do {
-        if ( (x = lock->lock) & RW_WRITE_FLAG )
-            return 0;
-    } while ( cmpxchg(&lock->lock, x, x+1) != x );
-    preempt_disable();
-    return 1;
-}
+    /*
+     * At the head of the wait queue now, wait until the writer state
+     * goes to 0 and then try to increment the reader count and get
+     * the lock. It is possible that an incoming writer may steal the
+     * lock in the interim, so it is necessary to check the writer byte
+     * to make sure that the write lock isn't taken.
+     */
+    while ( atomic_read(&lock->cnts) & _QW_WMASK )
+        cpu_relax();
 
-#ifndef _raw_read_unlock
-# define _raw_read_unlock(l) do {                      \
-    uint32_t x = (l)->lock, y;                         \
-    while ( (y = cmpxchg(&(l)->lock, x, x - 1)) != x ) \
-        x = y;                                         \
-} while (0)
-#endif
+    cnts = atomic_add_return(_QR_BIAS, &lock->cnts) - _QR_BIAS;
+    rspin_until_writer_unlock(lock, cnts);
 
-inline void _read_unlock(rwlock_t *lock)
-{
-    preempt_enable();
-    _raw_read_unlock(lock);
+    /*
+     * Signal the next one in queue to become queue head
+     */
+    spin_unlock(&lock->lock);
 }
 
-void _read_unlock_irq(rwlock_t *lock)
+/**
+ * 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;
 
-    ASSERT(local_irq_is_enabled());
-    local_irq_disable();
-    check_lock(&lock->debug);
-    do {
-        if ( (x = lock->lock) & RW_WRITE_FLAG )
-        {
-            local_irq_enable();
-            while ( (x = lock->lock) & RW_WRITE_FLAG )
-                cpu_relax();
-            local_irq_disable();
-        }
-    } while ( cmpxchg(&lock->lock, x, x|RW_WRITE_FLAG) != x );
-    while ( x != 0 )
-    {
         cpu_relax();
-        x = lock->lock & ~RW_WRITE_FLAG;
     }
-    preempt_disable();
-}
 
-unsigned long _write_lock_irqsave(rwlock_t *lock)
-{
-    uint32_t x;
-    unsigned long flags;
-
-    local_irq_save(flags);
-    check_lock(&lock->debug);
-    do {
-        if ( (x = lock->lock) & RW_WRITE_FLAG )
-        {
-            local_irq_restore(flags);
-            while ( (x = lock->lock) & RW_WRITE_FLAG )
-                cpu_relax();
-            local_irq_disable();
-        }
-    } while ( cmpxchg(&lock->lock, x, x|RW_WRITE_FLAG) != x );
-    while ( x != 0 )
+    /* When no more readers, set the locked flag */
+    for (;;)
     {
+        cnts = atomic_read(&lock->cnts);
+        if ( (cnts == _QW_WAITING) &&
+             (atomic_cmpxchg(&lock->cnts, _QW_WAITING,
+                             _QW_LOCKED) == _QW_WAITING) )
+            break;
+
         cpu_relax();
-        x = lock->lock & ~RW_WRITE_FLAG;
     }
-    preempt_disable();
-    return flags;
-}
-
-int _write_trylock(rwlock_t *lock)
-{
-    uint32_t x;
-
-    check_lock(&lock->debug);
-    do {
-        if ( (x = lock->lock) != 0 )
-            return 0;
-    } while ( cmpxchg(&lock->lock, x, x|RW_WRITE_FLAG) != x );
-    preempt_disable();
-    return 1;
-}
-
-#ifndef _raw_write_unlock
-# define _raw_write_unlock(l) xchg(&(l)->lock, 0)
-#endif
-
-inline void _write_unlock(rwlock_t *lock)
-{
-    preempt_enable();
-    if ( _raw_write_unlock(lock) != RW_WRITE_FLAG )
-        BUG();
-}
-
-void _write_unlock_irq(rwlock_t *lock)
-{
-    _write_unlock(lock);
-    local_irq_enable();
-}
-
-void _write_unlock_irqrestore(rwlock_t *lock, unsigned long flags)
-{
-    _write_unlock(lock);
-    local_irq_restore(flags);
-}
-
-int _rw_is_locked(rwlock_t *lock)
-{
-    check_lock(&lock->debug);
-    return (lock->lock != 0); /* anyone in critical section? */
-}
-
-int _rw_is_write_locked(rwlock_t *lock)
-{
-    check_lock(&lock->debug);
-    return (lock->lock == RW_WRITE_FLAG); /* writer in critical section? */
+unlock:
+    spin_unlock(&lock->lock);
 }
 
 #ifdef LOCK_PROFILE
diff --git a/xen/include/xen/spinlock.h b/xen/include/xen/spinlock.h
index 5e54407..98f242c 100644
--- a/xen/include/xen/spinlock.h
+++ b/xen/include/xen/spinlock.h
@@ -1,6 +1,7 @@
 #ifndef __SPINLOCK_H__
 #define __SPINLOCK_H__
 
+#include <xen/atomic.h>
 #include <asm/system.h>
 #include <asm/spinlock.h>
 
@@ -148,17 +149,31 @@ typedef struct spinlock {
 
 #define spin_lock_init(l) (*(l) = (spinlock_t)SPIN_LOCK_UNLOCKED)
 
+
 typedef struct {
-    volatile uint32_t lock;
-    struct lock_debug debug;
+    atomic_t cnts;
+    spinlock_t lock;
+    /* FIXME: struct lock_debug debug; */
 } rwlock_t;
 
-#define RW_WRITE_FLAG (1u<<31)
+#define    RW_LOCK_UNLOCKED {           \
+    .cnts = ATOMIC_INIT(0),             \
+    .lock = SPIN_LOCK_UNLOCKED,         \
+/*    debug  */                         \
+}
 
-#define RW_LOCK_UNLOCKED { 0, _LOCK_DEBUG }
 #define DEFINE_RWLOCK(l) rwlock_t l = RW_LOCK_UNLOCKED
 #define rwlock_init(l) (*(l) = (rwlock_t)RW_LOCK_UNLOCKED)
 
+/*
+ * 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)
+
 void _spin_lock(spinlock_t *lock);
 void _spin_lock_irq(spinlock_t *lock);
 unsigned long _spin_lock_irqsave(spinlock_t *lock);
@@ -175,26 +190,160 @@ int _spin_trylock_recursive(spinlock_t *lock);
 void _spin_lock_recursive(spinlock_t *lock);
 void _spin_unlock_recursive(spinlock_t *lock);
 
-void _read_lock(rwlock_t *lock);
-void _read_lock_irq(rwlock_t *lock);
-unsigned long _read_lock_irqsave(rwlock_t *lock);
-
-void _read_unlock(rwlock_t *lock);
-void _read_unlock_irq(rwlock_t *lock);
-void _read_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
-int _read_trylock(rwlock_t *lock);
-
-void _write_lock(rwlock_t *lock);
-void _write_lock_irq(rwlock_t *lock);
-unsigned long _write_lock_irqsave(rwlock_t *lock);
-int _write_trylock(rwlock_t *lock);
+void queue_read_lock_slowpath(rwlock_t *lock);
+void queue_write_lock_slowpath(rwlock_t *lock);
 
-void _write_unlock(rwlock_t *lock);
-void _write_unlock_irq(rwlock_t *lock);
-void _write_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
-
-int _rw_is_locked(rwlock_t *lock);
-int _rw_is_write_locked(rwlock_t *lock);
+/**
+ * queue_read_trylock - try to acquire read lock of a queue rwlock
+ * @lock : Pointer to queue rwlock structure
+ * Return: 1 if lock acquired, 0 if failed
+ */
+static inline int _read_trylock(rwlock_t *lock)
+{
+    u32 cnts;
+
+    cnts = atomic_read(&lock->cnts);
+    if ( likely(!(cnts & _QW_WMASK)) )
+    {
+        cnts = (u32)atomic_add_return(_QR_BIAS, &lock->cnts);
+        if ( likely(!(cnts & _QW_WMASK)) )
+            return 1;
+        atomic_sub(_QR_BIAS, &lock->cnts);
+    }
+    return 0;
+}
+
+/**
+ * queue_read_lock - acquire read lock of a queue rwlock
+ * @lock: Pointer to queue rwlock structure
+ */
+static inline void _read_lock(rwlock_t *lock)
+{
+    u32 cnts;
+
+    cnts = atomic_add_return(_QR_BIAS, &lock->cnts);
+    if ( likely(!(cnts & _QW_WMASK)) )
+        return;
+
+    /* The slowpath will decrement the reader count, if necessary. */
+    queue_read_lock_slowpath(lock);
+}
+
+static inline void _read_lock_irq(rwlock_t *lock)
+{
+    ASSERT(local_irq_is_enabled());
+    local_irq_disable();
+    _read_lock(lock);
+}
+
+static inline unsigned long _read_lock_irqsave(rwlock_t *lock)
+{
+    unsigned long flags;
+    local_irq_save(flags);
+    _read_lock(lock);
+    return flags;
+}
+
+/**
+ * queue_read_unlock - release read lock of a queue rwlock
+ * @lock : Pointer to queue rwlock structure
+ */
+static inline void _read_unlock(rwlock_t *lock)
+{
+    /*
+     * Atomically decrement the reader count
+     */
+    atomic_sub(_QR_BIAS, &lock->cnts);
+}
+
+static inline void _read_unlock_irq(rwlock_t *lock)
+{
+    _read_unlock(lock);
+    local_irq_enable();
+}
+
+static inline void _read_unlock_irqrestore(rwlock_t *lock, unsigned long flags)
+{
+    _read_unlock(lock);
+    local_irq_restore(flags);
+}
+
+static inline int _rw_is_locked(rwlock_t *lock)
+{
+    return atomic_read(&lock->cnts);
+}
+
+/**
+ * queue_write_lock - acquire write lock of a queue rwlock
+ * @lock : Pointer to queue rwlock structure
+ */
+static inline void _write_lock(rwlock_t *lock)
+{
+    /* Optimize for the unfair lock case where the fair flag is 0. */
+    if ( atomic_cmpxchg(&lock->cnts, 0, _QW_LOCKED) == 0 )
+        return;
+
+    queue_write_lock_slowpath(lock);
+}
+
+static inline void _write_lock_irq(rwlock_t *lock)
+{
+    ASSERT(local_irq_is_enabled());
+    local_irq_disable();
+    _write_lock(lock);
+}
+
+static inline unsigned long _write_lock_irqsave(rwlock_t *lock)
+{
+    unsigned long flags;
+
+    local_irq_save(flags);
+    _write_lock(lock);
+    return flags;
+}
+
+/**
+ * queue_write_trylock - try to acquire write lock of a queue rwlock
+ * @lock : Pointer to queue rwlock structure
+ * Return: 1 if lock acquired, 0 if failed
+ */
+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);
+}
+
+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);
+}
+
+static inline void _write_unlock_irq(rwlock_t *lock)
+{
+    _write_unlock(lock);
+    local_irq_enable();
+}
+
+static inline void _write_unlock_irqrestore(rwlock_t *lock, unsigned long flags)
+{
+    _write_unlock(lock);
+    local_irq_restore(flags);
+}
+
+static inline int _rw_is_write_locked(rwlock_t *lock)
+{
+    return atomic_read(&lock->cnts) & _QW_WMASK;
+}
 
 #define spin_lock(l)                  _spin_lock(l)
 #define spin_lock_irq(l)              _spin_lock_irq(l)
-- 
2.1.4

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

[Jan Beulich]

>>> 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

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

[Jennifer Herbert]

Hi Jan,

Thank you for your comments.  Sorry for the slow response - xmas and all...
Responses below...


On 22/12/15 13:54, Jan Beulich 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.

Oh yes, missed that - will fix.

>
>> +    /*
>> +     * 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?

We haven't found a reason this was introduced to Linux, but assume this 
was to reduce  interrupt latency.  I had thought to leave it there, in 
case we want to use it in the future, but now feel it would probably 
better to remove it, and deal with any irq related issues, if and when 
we use rw locks from irq handlers.


>> -        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?

I wanted to port know proven code.  We have now run this code in xen for 
quite a while, and while I think you may well be correct, I'd rather get 
this version in, and then consider further optimisation testing in 
future patches.

>> +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.

Great idea. I think the reasons for keeping them together must just be 
historical.  I'll try and split that out.


>> +/*
>> + * 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)?

You picked up on optimisation in your later comment - I should add a 
comment here.

>> +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.

I theorised that this was like this to aid the readability of the code, 
although I don't know if it does.  I'd happily change it over, and 
replace the cnts with 0s.

>> +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.

Yes. A comment on the declaration would no doubt help too.

>> +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?

It was, but having thought about it a bit more, it would only have been 
"more useful" like this in unsafe (or at lest ugly) code, and so for 
that reason I should change it.  I don't think this would have affected 
the safe cases.
> Jan

Cheers

-jenny

Re: [PATCHv1 0/4] spinlock: add qrwlocks

[Ian Campbell]

On Fri, 2015-12-18 at 14:09 +0000, David Vrabel wrote:
> This series adds the qrwlocks from Linux.

This really means "replace existing rwlock implementation with qrwlocks",
right? i.e. you aren't actually adding a new set of rwlock primitives here,
you are just switching out the implementation behind the existing ones.

Ian.

Re: [PATCHv1 0/4] spinlock: add qrwlocks

[David Vrabel]

On 18/12/15 15:47, Ian Campbell wrote:
> On Fri, 2015-12-18 at 14:09 +0000, David Vrabel wrote:
>> This series adds the qrwlocks from Linux.
> 
> This really means "replace existing rwlock implementation with qrwlocks",
> right? i.e. you aren't actually adding a new set of rwlock primitives here,
> you are just switching out the implementation behind the existing ones.

Yes. Sorry for not being clearer.

David