@@ -18,12 +18,12 @@ static __always_inline int queued_spin_value_unlocked(struct qspinlock lock)
static __always_inline int queued_spin_is_contended(struct qspinlock *lock)
{
- return 0;
+ return !!(atomic_read(&lock->val) & _Q_TAIL_CPU_MASK);
}
static __always_inline int queued_spin_trylock(struct qspinlock *lock)
{
- return atomic_cmpxchg_acquire(&lock->val, 0, 1) == 0;
+ return atomic_cmpxchg_acquire(&lock->val, 0, _Q_LOCKED_VAL) == 0;
}
void queued_spin_lock_slowpath(struct qspinlock *lock);
@@ -36,7 +36,11 @@ static __always_inline void queued_spin_lock(struct qspinlock *lock)
static inline void queued_spin_unlock(struct qspinlock *lock)
{
- atomic_set_release(&lock->val, 0);
+ for (;;) {
+ int val = atomic_read(&lock->val);
+ if (atomic_cmpxchg_release(&lock->val, val, val & ~_Q_LOCKED_VAL) == val)
+ return;
+ }
}
#define arch_spin_is_locked(l) queued_spin_is_locked(l)
@@ -10,4 +10,25 @@ typedef struct qspinlock {
#define __ARCH_SPIN_LOCK_UNLOCKED { .val = ATOMIC_INIT(0) }
+/*
+ * Bitfields in the atomic value:
+ *
+ * 0: locked bit
+ * 16-31: tail cpu (+1)
+ */
+#define _Q_SET_MASK(type) (((1U << _Q_ ## type ## _BITS) - 1)\
+ << _Q_ ## type ## _OFFSET)
+#define _Q_LOCKED_OFFSET 0
+#define _Q_LOCKED_BITS 1
+#define _Q_LOCKED_MASK _Q_SET_MASK(LOCKED)
+#define _Q_LOCKED_VAL (1U << _Q_LOCKED_OFFSET)
+
+#define _Q_TAIL_CPU_OFFSET 16
+#define _Q_TAIL_CPU_BITS (32 - _Q_TAIL_CPU_OFFSET)
+#define _Q_TAIL_CPU_MASK _Q_SET_MASK(TAIL_CPU)
+
+#if CONFIG_NR_CPUS >= (1U << _Q_TAIL_CPU_BITS)
+#error "qspinlock does not support such large CONFIG_NR_CPUS"
+#endif
+
#endif /* _ASM_POWERPC_QSPINLOCK_TYPES_H */
@@ -1,12 +1,186 @@
// SPDX-License-Identifier: GPL-2.0-or-later
+#include <linux/atomic.h>
+#include <linux/bug.h>
+#include <linux/compiler.h>
#include <linux/export.h>
-#include <linux/processor.h>
+#include <linux/percpu.h>
+#include <linux/smp.h>
#include <asm/qspinlock.h>
-void queued_spin_lock_slowpath(struct qspinlock *lock)
+#define MAX_NODES 4
+
+struct qnode {
+ struct qnode *next;
+ struct qspinlock *lock;
+ u8 locked; /* 1 if lock acquired */
+};
+
+struct qnodes {
+ int count;
+ struct qnode nodes[MAX_NODES];
+};
+
+static DEFINE_PER_CPU_ALIGNED(struct qnodes, qnodes);
+
+static inline int encode_tail_cpu(int cpu)
+{
+ return (cpu + 1) << _Q_TAIL_CPU_OFFSET;
+}
+
+static inline int decode_tail_cpu(int val)
+{
+ return (val >> _Q_TAIL_CPU_OFFSET) - 1;
+}
+
+/* Take the lock by setting the bit, no other CPUs may concurrently lock it. */
+static __always_inline void set_locked(struct qspinlock *lock)
+{
+ atomic_or(_Q_LOCKED_VAL, &lock->val);
+ __atomic_acquire_fence();
+}
+
+/* Take lock, clearing tail, cmpxchg with val (which must not be locked) */
+static __always_inline int trylock_clear_tail_cpu(struct qspinlock *lock, int val)
+{
+ int newval = _Q_LOCKED_VAL;
+
+ BUG_ON(val & _Q_LOCKED_VAL);
+
+ return atomic_cmpxchg_acquire(&lock->val, val, newval) == val;
+}
+
+/*
+ * Publish our tail, replacing previous tail. Return previous value.
+ *
+ * This provides a release barrier for publishing node, this pairs with the
+ * acquire barrier in get_tail_qnode() when the next CPU finds this tail
+ * value.
+ */
+static __always_inline int publish_tail_cpu(struct qspinlock *lock, int tail)
+{
+ for (;;) {
+ int val = atomic_read(&lock->val);
+ int newval = (val & ~_Q_TAIL_CPU_MASK) | tail;
+ int old;
+
+ old = atomic_cmpxchg_release(&lock->val, val, newval);
+ if (old == val)
+ return old;
+ }
+}
+
+static struct qnode *get_tail_qnode(struct qspinlock *lock, int val)
+{
+ int cpu = decode_tail_cpu(val);
+ struct qnodes *qnodesp = per_cpu_ptr(&qnodes, cpu);
+ int idx;
+
+ /*
+ * After publishing the new tail and finding a previous tail in the
+ * previous val (which is the control dependency), this barrier
+ * orders the release barrier in publish_tail_cpu performed by the
+ * last CPU, with subsequently looking at its qnode structures
+ * after the barrier.
+ */
+ smp_acquire__after_ctrl_dep();
+
+ for (idx = 0; idx < MAX_NODES; idx++) {
+ struct qnode *qnode = &qnodesp->nodes[idx];
+ if (qnode->lock == lock)
+ return qnode;
+ }
+
+ BUG();
+}
+
+static inline void queued_spin_lock_mcs_queue(struct qspinlock *lock)
{
- while (!queued_spin_trylock(lock))
+ struct qnodes *qnodesp;
+ struct qnode *next, *node;
+ int val, old, tail;
+ int idx;
+
+ BUILD_BUG_ON(CONFIG_NR_CPUS >= (1U << _Q_TAIL_CPU_BITS));
+
+ qnodesp = this_cpu_ptr(&qnodes);
+ if (unlikely(qnodesp->count >= MAX_NODES)) {
+ while (!queued_spin_trylock(lock))
+ cpu_relax();
+ return;
+ }
+
+ idx = qnodesp->count++;
+ /*
+ * Ensure that we increment the head node->count before initialising
+ * the actual node. If the compiler is kind enough to reorder these
+ * stores, then an IRQ could overwrite our assignments.
+ */
+ barrier();
+ node = &qnodesp->nodes[idx];
+ node->next = NULL;
+ node->lock = lock;
+ node->locked = 0;
+
+ tail = encode_tail_cpu(smp_processor_id());
+
+ old = publish_tail_cpu(lock, tail);
+
+ /*
+ * If there was a previous node; link it and wait until reaching the
+ * head of the waitqueue.
+ */
+ if (old & _Q_TAIL_CPU_MASK) {
+ struct qnode *prev = get_tail_qnode(lock, old);
+
+ /* Link @node into the waitqueue. */
+ WRITE_ONCE(prev->next, node);
+
+ /* Wait for mcs node lock to be released */
+ while (!node->locked)
+ cpu_relax();
+
+ smp_rmb(); /* acquire barrier for the mcs lock */
+ }
+
+ /* We're at the head of the waitqueue, wait for the lock. */
+ for (;;) {
+ val = atomic_read(&lock->val);
+ if (!(val & _Q_LOCKED_VAL))
+ break;
+
+ cpu_relax();
+ }
+
+ /* If we're the last queued, must clean up the tail. */
+ if ((val & _Q_TAIL_CPU_MASK) == tail) {
+ if (trylock_clear_tail_cpu(lock, val))
+ goto release;
+ /* Another waiter must have enqueued */
+ }
+
+ /* We must be the owner, just set the lock bit and acquire */
+ set_locked(lock);
+
+ /* contended path; must wait for next != NULL (MCS protocol) */
+ while (!(next = READ_ONCE(node->next)))
cpu_relax();
+
+ /*
+ * Unlock the next mcs waiter node. Release barrier is not required
+ * here because the acquirer is only accessing the lock word, and
+ * the acquire barrier we took the lock with orders that update vs
+ * this store to locked. The corresponding barrier is the smp_rmb()
+ * acquire barrier for mcs lock, above.
+ */
+ WRITE_ONCE(next->locked, 1);
+
+release:
+ qnodesp->count--; /* release the node */
+}
+
+void queued_spin_lock_slowpath(struct qspinlock *lock)
+{
+ queued_spin_lock_mcs_queue(lock);
}
EXPORT_SYMBOL(queued_spin_lock_slowpath);
This forms the basis of the qspinlock slow path. Like generic qspinlocks and unlike the vanilla MCS algorithm, the lock owner does not participate in the queue, only waiters. The first waiter spins on the lock word, then when the lock is released it takes ownership and unqueues the next waiter. This is how qspinlocks can be implemented with the spinlock API -- lock owners don't need a node, only waiters do. Signed-off-by: Nicholas Piggin <npiggin@gmail.com> --- arch/powerpc/include/asm/qspinlock.h | 10 +- arch/powerpc/include/asm/qspinlock_types.h | 21 +++ arch/powerpc/lib/qspinlock.c | 180 ++++++++++++++++++++- 3 files changed, 205 insertions(+), 6 deletions(-)