Add queuelock_t, an MCS lock implementation.

This patch adds a queuelock_t spinlock. This lock
is a bit more efficient on systems with large
distributed caches (less bouncing of lock
cache lines). The ticketlock_t is remains
the default spinlock choice, however.

7 files changed, 323 insertions, 28 deletions

diff --git a/include/condvar.h b/include/condvar.h
index cba7678..8ee3859 100644
--- a/include/condvar.h
+++ b/include/condvar.h
@@ -6,7 +6,7 @@
 #include <sys/syscall.h>
 #include <linux/futex.h>
 
-#include "ticketlock.h"
+#include "spinlock.h"
 
 typedef int seq_t;
 
@@ -36,32 +36,32 @@ static inline void cv_init_shared(cv_t* cv)
         *cv = CV_INITIALIZER_PUBLIC;
 }
 
-static inline int cv_wait(cv_t* cv, ticketlock_t* l)
+static inline int cv_wait(cv_t* cv, spinlock_t* l)
 {
         int ret;
         int wait_mode = (cv->mode == CV_PRIVATE) ? FUTEX_WAIT_PRIVATE : FUTEX_WAIT;
         seq_t seq = cv->seq;
 
         ++cv->waiters;
-        tl_unlock(l); /* memory barrier */
+        spin_unlock(l); /* memory barrier */
         ret = syscall(SYS_futex, &cv->seq, wait_mode, seq, NULL, NULL, 0);
-        tl_lock(l);
+        spin_lock(l);
         --cv->waiters;
 
         return ret;
 }
 
 #ifndef PGM_PREEMPTIVE
-static inline int cv_wait_np(cv_t* cv, ticketlock_t* l, unsigned long *flags)
+static inline int cv_wait_np(cv_t* cv, spinlock_t* l, unsigned long *flags)
 {
         int ret;
         int wait_mode = (cv->mode == CV_PRIVATE) ? FUTEX_WAIT_PRIVATE : FUTEX_WAIT;
         seq_t seq = cv->seq;
 
         ++cv->waiters;
-        tl_unlock_np(l, *flags); /* memory barrier */
+        spin_unlock_np(l, *flags); /* memory barrier */
         ret = syscall(SYS_futex, &cv->seq, wait_mode, seq, NULL, NULL, 0);
-        tl_lock_np(l, flags);
+        spin_lock_np(l, flags);
         --cv->waiters;
 
         return ret;
diff --git a/include/config.h b/include/config.h
index af41d42..41bfdb4 100644
--- a/include/config.h
+++ b/include/config.h
@@ -22,6 +22,18 @@
 #endif
 
 /*
+ Select the spinlock type. ticketlock_t is simple,
+ but queuelock_t is more cache efficient on systems
+ with large distributed caches (less bouncing of
+ cache lines).
+ - ticketlock_t = 0
+ - queuelock_t  = 1
+ */
+#if (PGM_SYNC_METHOD == 1)
+        #define PGM_SPINLOCK_TYPE       0
+#endif
+
+/*
  Select the method tasks use for becoming non-preemptive
  while holding spinlocks.
  - To allow preemption                 = 0
@@ -32,7 +44,7 @@
   1) Options 1 and 2 require tasks to run as root.
   2) Disabling interrupts may be unsafe for IPCs
      that may block on write. (PGM uses non-blocking
-         writes whenever possible.)
+     writes whenever possible.)
  */
 #ifndef _USE_LITMUS
         #define PGM_NP_METHOD           0
diff --git a/include/interrupts.h b/include/interrupts.h
new file mode 100644
index 0000000..abd5346
--- /dev/null
+++ b/include/interrupts.h
@@ -0,0 +1,17 @@
+// Copyright (c) 2014, Glenn Elliott
+// All rights reserved.
+
+#pragma once
+
+static inline unsigned long save_flags(void)
+{
+        unsigned long flags;
+        asm volatile("pushf ; pop %0" : "=rm" (flags) : :"memory");
+        return flags;
+}
+
+static inline void restore_flags(unsigned long flags)
+{
+        asm volatile("push %0 ; popf" : : "g" (flags) : "memory", "cc");
+}
+
diff --git a/include/queuelock.h b/include/queuelock.h
new file mode 100644
index 0000000..1eab9a3
--- /dev/null
+++ b/include/queuelock.h
@@ -0,0 +1,184 @@
+// Copyright (c) 2014, Glenn Elliott
+// All rights reserved.
+
+// An implementation of an MCS (Mellor-Crummey and Scott) spinlock
+
+#pragma once
+
+#include <stdint.h>
+
+#if defined(PGM_NP_INTERRUPTS)
+#include <sys/io.h>
+#endif
+
+#if defined(PGM_NP_LITMUS)
+#include "litmus.h"
+#endif
+
+#include "atomic.h"
+#include "interrupts.h"
+
+#if ULONG_MAX == 0xffffffffffffffffUL
+typedef uint64_t ptr_t;
+#else
+typedef uint32_t ptr_t;
+#endif
+
+typedef struct __queuenode
+{
+        volatile struct __queuenode* next;
+        volatile int                 blocked;
+} queuenode_t;
+
+typedef struct
+{
+        ptr_t tail;
+} queuelock_t;
+
+static const queuelock_t QUEUELOCK_INITIALIZER = {.tail = 0};
+
+static inline void ql_init(queuelock_t* q)
+{
+        *q = QUEUELOCK_INITIALIZER;
+}
+
+static inline void __ql_lock(queuelock_t* q, queuenode_t* self)
+{
+        queuenode_t* prev;
+
+        /* prepare for enqueue */
+        self->next = NULL;
+        self->blocked = 1;
+
+        /* enqueue at the end of the line */
+        do
+        {
+                prev = (struct __queuenode*)((volatile ptr_t) q->tail);
+        } while(__sync_bool_compare_and_swap(&q->tail, (ptr_t)prev, (ptr_t)self));
+
+        /* was there someone ahead of us? */
+        if (prev)
+        {
+                /* tell the one ahead of us about ourselves */
+                prev->next = self;
+
+                /* wait until unlock */
+                while(self->blocked)
+                {
+                        __sync_pause();
+                }
+        }
+
+        /* make sure nothing leaks out (upward) of the critical section */
+        __sync_synchronize();
+}
+
+static inline void ql_lock(queuelock_t* q, queuenode_t* self)
+{
+        __ql_lock(q, self);
+}
+
+static inline void __ql_unlock(queuelock_t* q, queuenode_t* self)
+{
+        /* make sure nothing leaks out (downward) of the critical section */
+        __sync_synchronize();
+
+        /* try to dequeue from the end of the line */
+        if (!self->next && __sync_bool_compare_and_swap(&q->tail, (ptr_t)self, (ptr_t)0))
+        {
+                /* we really weren't at the end of the line! */
+
+                /* wait until the one behind us tells us where they are */
+                while(!self->next)
+                {
+                        __sync_pause();
+                }
+        }
+
+        /* tell the next one behind us that they may go */
+        if(self->next)
+        {
+                /* unlock */
+                self->next->blocked = 0;
+        }
+}
+
+static inline void ql_unlock(queuelock_t* q, queuenode_t* self)
+{
+        __ql_unlock(q, self);
+}
+
+#ifndef PGM_PREEMPTIVE
+
+static inline void ql_init_np(queuelock_t* q)
+{
+        ql_init(q);
+
+#if defined(PGM_NP_INTERRUPTS)
+        iopl(3);
+#endif
+}
+
+static inline void ql_lock_np(queuelock_t* q, queuenode_t* self, unsigned long* flags)
+{
+#if defined(PGM_NP_INTERRUPTS)
+        /* save flags and disable interrupts */
+        *flags = save_flags();
+        asm volatile("cli": : :"memory");
+#elif defined(PGM_NP_LITMUS)
+        /* start non-preemption */
+        __sync_synchronize();
+        enter_np();
+        __sync_synchronize();
+#endif
+
+        __ql_lock(q, self);
+}
+
+static inline void ql_unlock_np(queuelock_t* q, queuenode_t* self, unsigned long flags)
+{
+        __ql_unlock(q, self);
+
+#if defined(PGM_NP_INTERRUPTS)
+        /* re-enable flags before we exit */
+        restore_flags(flags);
+#elif defined(PGM_NP_LITMUS)
+        /* end non-preemption */
+        __sync_synchronize();
+        exit_np();
+        __sync_synchronize();
+#endif
+}
+#endif
+
+/*
+   Carrying around the queuenode_t is a pain for simple non-nested
+   critical sections. These varients of lock/unlock just puts the
+   wait node in thread-local storage. At most one non-nest queue
+   lock calll may be nested within other critical sections.
+*/
+static inline void ql_lock_no_nest(queuelock_t* q)
+{
+        extern __thread queuenode_t thread_qnode;
+        ql_lock(q, &thread_qnode);
+}
+
+static inline void ql_unlock_no_nest(queuelock_t* q)
+{
+        extern __thread queuenode_t thread_qnode;
+        ql_unlock(q, &thread_qnode);
+}
+
+#ifndef PGM_PREEMPTIVE
+static inline void ql_lock_no_nest_np(queuelock_t* q, unsigned long* flags)
+{
+        extern __thread queuenode_t thread_qnode;
+        ql_lock_np(q, &thread_qnode, flags);
+}
+
+static inline void ql_unlock_no_nest_np(queuelock_t* q, unsigned long flags)
+{
+        extern __thread queuenode_t thread_qnode;
+        ql_unlock_np(q, &thread_qnode, flags);
+}
+#endif
diff --git a/include/spinlock.h b/include/spinlock.h
new file mode 100644
index 0000000..7220353
--- /dev/null
+++ b/include/spinlock.h
@@ -0,0 +1,89 @@
+// Copyright (c) 2014, Glenn Elliott
+// All rights reserved.
+
+#pragma once
+
+#include "config.h"
+
+#if PGM_SYNC_METHOD == 1
+
+#if PGM_SPINLOCK_TYPE == 0 /* ticket lock */
+#include "ticketlock.h"
+typedef ticketlock_t spinlock_t;
+static const spinlock_t PGM_SPINLOCK_INITIALIZER = TICKETLOCK_INITIALIZER;
+
+static inline void spin_init(spinlock_t* l)
+{
+        tl_init(l);
+}
+
+static inline void spin_lock(spinlock_t* l)
+{
+        tl_lock(l);
+}
+
+static inline void spin_unlock(spinlock_t* l)
+{
+        tl_unlock(l);
+}
+
+#ifndef PGM_PREEMPTIVE
+static inline void spin_init_np(spinlock_t* l)
+{
+        tl_init_np(l);
+}
+
+static inline void spin_lock_np(spinlock_t* l, long unsigned int* flags)
+{
+        tl_lock_np(l, flags);
+}
+
+static inline void spin_unlock_np(spinlock_t* l, long unsigned int flags)
+{
+        tl_unlock_np(l, flags);
+}
+#endif /* end !PGM_PREEMPTIVE */
+
+#elif PGM_SPINLOCK_TYPE == 1 /* queue lock */
+
+#include "queuelock.h"
+typedef queuelock_t spinlock_t;
+static const spinlock_t PGM_SPINLOCK_INITIALIZER = QUEUELOCK_INITIALIZER;
+
+static inline void spin_init(spinlock_t* l)
+{
+        ql_init(l);
+}
+
+static inline void spin_lock(spinlock_t* l)
+{
+        ql_lock_no_nest(l);
+}
+
+static inline void spin_unlock(spinlock_t* l)
+{
+        ql_unlock_no_nest(l);
+}
+
+#ifndef PGM_PREEMPTIVE
+static inline void spin_init_np(spinlock_t* l)
+{
+        ql_init_np(l);
+}
+
+static inline void spin_lock_np(spinlock_t* l, long unsigned int* flags)
+{
+        ql_lock_no_nest_np(l, flags);
+}
+
+static inline void spin_unlock_np(spinlock_t* l, long unsigned int flags)
+{
+        ql_unlock_no_nest_np(l, flags);
+}
+#endif /* end !PGM_PREEMPTIVE */
+
+#else
+#error "Unknown spinlock configuration!"
+#endif
+
+#endif /* end PGM_SYNC_METHOD == 1 */
diff --git a/include/ticketlock.h b/include/ticketlock.h
index 4c1035c..faef7d9 100644
--- a/include/ticketlock.h
+++ b/include/ticketlock.h
@@ -14,6 +14,7 @@
 #endif
 
 #include "atomic.h"
+#include "interrupts.h"
 
 #if ULONG_MAX == 0xffffffffffffffffUL
 typedef unsigned long long ticketdata_t;
@@ -77,18 +78,6 @@ static inline void tl_init_np(ticketlock_t *t)
 #endif
 }
 
-static inline unsigned long save_flags(void)
-{
-        unsigned long flags;
-        asm volatile("pushf ; pop %0" : "=rm" (flags) : :"memory");
-        return flags;
-}
-
-static inline void restore_flags(unsigned long flags)
-{
-        asm volatile("push %0 ; popf" : : "g" (flags) : "memory", "cc");
-}
-
 static inline void tl_lock_np(ticketlock_t* t, unsigned long* flags)
 {
 #if defined(PGM_NP_INTERRUPTS)
diff --git a/src/pgm.cpp b/src/pgm.cpp
index 122775e..ff6c2f9 100644
--- a/src/pgm.cpp
+++ b/src/pgm.cpp
@@ -33,10 +33,14 @@
 typedef pthread_mutex_t pgm_lock_t;
 typedef pthread_cond_t  pgm_cv_t;
 #elif defined(PGM_USE_PGM_SYNC)
-#include "ticketlock.h"
+#include "spinlock.h"
 #include "condvar.h"
-typedef ticketlock_t    pgm_lock_t;
+typedef spinlock_t      pgm_lock_t;
 typedef cv_t            pgm_cv_t;
+
+#if PGM_SPINLOCK_TYPE == 1
+__thread queuenode_t thread_qnode;
+#endif
 #endif
 
 #include "ring.h"
@@ -1420,13 +1424,13 @@ out:
 #elif defined(PGM_USE_PGM_SYNC)
 
         #ifdef PGM_PREEMPTIVE
-                #define pgm_lock_init(l)    tl_init((l))
-                #define pgm_lock(l, flags)   do { ((void)(flags)); tl_lock((l)); } while(0)
-                #define pgm_unlock(l, flags) do { ((void)(flags)); tl_unlock((l)); } while(0)
+                #define pgm_lock_init(l)     spin_init((l))
+                #define pgm_lock(l, flags)   do { ((void)(flags)); spin_lock((l)); } while(0)
+                #define pgm_unlock(l, flags) do { ((void)(flags)); spin_unlock((l)); } while(0)
         #else
-                #define pgm_lock_init(l)    tl_init_np((l))
-                #define pgm_lock(l, flags)   tl_lock_np((l), &(flags))
-                #define pgm_unlock(l, flags) tl_unlock_np((l), (flags))
+                #define pgm_lock_init(l)     spin_init_np((l))
+                #define pgm_lock(l, flags)   spin_lock_np((l), &(flags))
+                #define pgm_unlock(l, flags) spin_unlock_np((l), (flags))
         #endif
         #define pgm_lock_destroy(l) (void)(l)