Since its inception, our understanding of ACQUIRE, esp. as applied to
spinlocks, has changed somewhat. Also, I wonder if, with a simple
change, we cannot make it provide more.

The problem with the comment is that the STORE done by spin_lock isn't
itself ordered by the ACQUIRE, and therefore a later LOAD can pass over
it and cross with any prior STORE, rendering the default WMB
insufficient (pointed out by Alan).

Now, this is only really a problem on PowerPC and ARM64, both of
which already defined smp_mb__before_spinlock() as a smp_mb().

At the same time, we can get a much stronger construct if we place
that same barrier _inside_ the spin_lock(). In that case we upgrade
the RCpc spinlock to an RCsc.  That would make all schedule() calls
fully transitive against one another.

Cc: Alan Stern <st...@rowland.harvard.edu>
Cc: Nicholas Piggin <npig...@gmail.com>
Cc: Ingo Molnar <mi...@kernel.org>
Cc: Will Deacon <will.dea...@arm.com>
Cc: Linus Torvalds <torva...@linux-foundation.org>
Cc: Michael Ellerman <m...@ellerman.id.au>
Cc: Oleg Nesterov <o...@redhat.com>
Cc: Benjamin Herrenschmidt <b...@kernel.crashing.org>
Cc: Paul McKenney <paul...@linux.vnet.ibm.com>
Signed-off-by: Peter Zijlstra (Intel) <pet...@infradead.org>
---
 arch/arm64/include/asm/spinlock.h   |    2 ++
 arch/powerpc/include/asm/spinlock.h |    3 +++
 include/linux/spinlock.h            |   36 ++++++++++++++++++++++++++++++++++++
 kernel/sched/core.c                 |    4 ++--
 4 files changed, 43 insertions(+), 2 deletions(-)

--- a/arch/arm64/include/asm/spinlock.h
+++ b/arch/arm64/include/asm/spinlock.h
@@ -367,5 +367,7 @@ static inline int arch_read_trylock(arch
  * smp_mb__before_spinlock() can restore the required ordering.
  */
 #define smp_mb__before_spinlock()      smp_mb()
+/* See include/linux/spinlock.h */
+#define smp_mb__after_spinlock()       smp_mb()
 
 #endif /* __ASM_SPINLOCK_H */
--- a/arch/powerpc/include/asm/spinlock.h
+++ b/arch/powerpc/include/asm/spinlock.h
@@ -342,5 +342,8 @@ static inline void arch_write_unlock(arc
 #define arch_read_relax(lock)  __rw_yield(lock)
 #define arch_write_relax(lock) __rw_yield(lock)
 
+/* See include/linux/spinlock.h */
+#define smp_mb__after_spinlock()   smp_mb()
+
 #endif /* __KERNEL__ */
 #endif /* __ASM_SPINLOCK_H */
--- a/include/linux/spinlock.h
+++ b/include/linux/spinlock.h
@@ -130,6 +130,42 @@ do {                                                       
        \
 #define smp_mb__before_spinlock()      smp_wmb()
 #endif
 
+/*
+ * This barrier must provide two things:
+ *
+ *   - it must guarantee a STORE before the spin_lock() is ordered against a
+ *     LOAD after it, see the comments at its two usage sites.
+ *
+ *   - it must ensure the critical section is RCsc.
+ *
+ * The latter is important for cases where we observe values written by other
+ * CPUs in spin-loops, without barriers, while being subject to scheduling.
+ *
+ * CPU0                        CPU1                    CPU2
+ *
+ *                     for (;;) {
+ *                       if (READ_ONCE(X))
+ *                         break;
+ *                     }
+ * X=1
+ *                     <sched-out>
+ *                                             <sched-in>
+ *                                             r = X;
+ *
+ * without transitivity it could be that CPU1 observes X!=0 breaks the loop,
+ * we get migrated and CPU2 sees X==0.
+ *
+ * Since most load-store architectures implement ACQUIRE with an smp_mb() after
+ * the LL/SC loop, they need no further barriers. Similarly all our TSO
+ * architectures imply an smp_mb() for each atomic instruction and equally 
don't
+ * need more.
+ *
+ * Architectures that can implement ACQUIRE better need to take care.
+ */
+#ifndef smp_mb__after_spinlock
+#define smp_mb__after_spinlock()       do { } while (0)
+#endif
+
 /**
  * raw_spin_unlock_wait - wait until the spinlock gets unlocked
  * @lock: the spinlock in question.
--- a/kernel/sched/core.c
+++ b/kernel/sched/core.c
@@ -1969,8 +1969,8 @@ try_to_wake_up(struct task_struct *p, un
         * reordered with p->state check below. This pairs with mb() in
         * set_current_state() the waiting thread does.
         */
-       smp_mb__before_spinlock();
        raw_spin_lock_irqsave(&p->pi_lock, flags);
+       smp_mb__after_spinlock();
        if (!(p->state & state))
                goto out;
 
@@ -3383,8 +3383,8 @@ static void __sched notrace __schedule(b
         * can't be reordered with __set_current_state(TASK_INTERRUPTIBLE)
         * done by the caller to avoid the race with signal_wake_up().
         */
-       smp_mb__before_spinlock();
        rq_lock(rq, &rf);
+       smp_mb__after_spinlock();
 
        /* Promote REQ to ACT */
        rq->clock_update_flags <<= 1;


Reply via email to