Implement queue locks based on the primitives provided atomic.h

This re-implements the queue locks for the new LITMUS version. Major
differences to the previous implementation:

	1) platform independent implementation - no manual assembly code in
           the queue lock implementation required

	2) Recursive acquiring of the locks is not possible! None of the
           other Linux locking primitives allow it, so it is consistent
           and we weren't using it anyway as far as I know.

	3) The number of "processes" in the implementation is fixed to
           NR_CPUS. Anything else doesn't really make sense in the kernel
           anyway.

1 files changed, 88 insertions, 32 deletions

diff --git a/include/linux/queuelock.h b/include/linux/queuelock.h
index 044422b856..aaefdcaa4c 100644
--- a/include/linux/queuelock.h
+++ b/include/linux/queuelock.h
@@ -1,32 +1,88 @@
-#ifndef _UNC_QUEUE_LOCK_H_

-#define _UNC_QUEUE_LOCK_H_

-/**

-*       Queue lock

-*

-*

-*       CLEANUP: Q-Locks to be re-implemented properly for the new LITMUS.

-*                Until then, we just use spin locks to make the code compile.

-*/

-

-#include <linux/spinlock.h>

-

-typedef spinlock_t queuelock_t;

-

-

-

-/* The new linux-like queue lock API.

-

-  int init_queuelock(queuelock_t* lock, int num_procs);

-  void queue_lock_wipe(queuelock_t * lock);

-  void queue_lock(queuelock_t *);

-  void queue_unlock(queuelock_t *); 

-

-*/

-

-

-#define queue_lock_init(lock)           spin_lock_init(lock)

-#define queue_lock_wipe(lock)           spin_lock_init(lock)

-#define queue_lock(lock)                spin_lock(lock)

-#define queue_unlock(lock)              spin_unlock(lock)

-

-#endif /*       _UNC_QUEUE_LOCK_H_      */

+#ifndef _UNC_QUEUELOCK_H_
+#define _UNC_QUEUELOCK_H_
+/**
+* Queue lock
+*
+* This is an implementation of T. Anderson's queue lock.
+* It strives to follow the normal Linux locking conventions
+* as much as possible. The rules for acquiring a lock are:
+*
+*  1) The caller must ensure interrupts and preemptions are disabled.
+* 
+*  2) The caller _cannot_ recursively acquire the lock.
+* 
+*  3) The caller may not sleep while holding the lock. This is currently
+*     not enforced, but it will not work.
+*/
+
+#include <linux/spinlock.h>
+#include <linux/cache.h>
+#include <asm/atomic.h>
+#include <linux/cpumask.h>
+#include <linux/smp.h>
+
+typedef struct {
+        /* pad the values being spun on to make sure
+           that they are cache local
+         */
+        union {
+                enum {
+                        MUST_WAIT,
+                        HAS_LOCK
+                } val;
+                char    padding[SMP_CACHE_BYTES];
+        } slots[NR_CPUS];
+
+        /* since spin_slot is not being spun on it can be
+         * in a shared cache line. next_slot will be evicted
+         * anyway on every attempt to acquire the lock.
+         */
+        int             spin_slot[NR_CPUS];
+
+        /* The next slot that will be available.
+         */
+        atomic_t        next_slot;
+} queuelock_t;
+
+
+static inline void queue_lock_init(queuelock_t *lock)
+{
+        int i;
+        for_each_possible_cpu(i) {
+                lock->slots[i].val      = MUST_WAIT;
+                lock->spin_slot[i]      = i;
+        }
+        lock->slots[i].val      = HAS_LOCK;
+        atomic_set(&lock->next_slot, 0);
+}
+
+
+static inline void queue_lock(queuelock_t *lock)
+{
+        int me = smp_processor_id();
+        /* Get slot to spin on. atomic_inc_return() returns the incremented
+         * value, so take one of again
+         */
+        lock->spin_slot[me] = atomic_inc_return(&lock->next_slot) - 1;
+        /* check for wrap-around
+         * This could probably optimized away if we ensure that NR_CPUS divides 
+         * INT_MAX...
+         */
+        if (unlikely(lock->spin_slot[me] == NR_CPUS - 1))
+                atomic_add(-NR_CPUS, &lock->next_slot);
+        /* range limit*/
+        lock->spin_slot[me] %= NR_CPUS;
+        /* spin until you acquire the lock */
+        while (lock->slots[lock->spin_slot[me]].val == MUST_WAIT);
+        /* reset the lock */
+        lock->slots[lock->spin_slot[me]].val = MUST_WAIT;
+}
+
+
+static inline void queue_unlock(queuelock_t *lock)
+{
+        int me = smp_processor_id();
+        lock->slots[(lock->spin_slot[me] + 1) % NR_CPUS].val = HAS_LOCK;
+}
+
+#endif /*       _UNC_QUEUELOCK_H_       */