SRCU updates for 4.11

From: Lance Roy <[email protected]>

SRCU uses two per-cpu counters: a nesting counter to count the number of active critical sections, and a sequence counter to ensure that the nesting counters don't change while they are being added together in srcu_readers_active_idx_check().

This patch instead uses per-cpu lock and unlock counters. Because the both counters only increase and srcu_readers_active_idx_check() reads the unlock counter before the lock counter, this achieves the same end without having to increment two different counters in srcu_read_lock(). This also saves a smp_mb() in srcu_readers_active_idx_check().

A possible problem with this patch is that it can only handle ULONG_MAX - NR_CPUS simultaneous readers, whereas the old version could handle up to ULONG_MAX.

Suggested-by Mathieu Desnoyers <[email protected]>

Signed-off-by Lance Roy <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

Cc Lai Jiangshan <[email protected]>

Cc Peter Zijlstra <[email protected]>

---
 include/linux/srcu.h    |   4 +-
 kernel/rcu/rcutorture.c |  18 +++++++-
 kernel/rcu/srcu.c       | 117 ++++++++++++++++++------------------------------
 3 files changed, 62 insertions(+), 77 deletions(-)

include/linux/srcu.h +2 -2

index dc8eb63c6568..0caea34d8c5f 100644
@@ -34,8 +34,8 @@
 #include <linux/workqueue.h>
 
 struct srcu_struct_array {
-	unsigned long c[2];
-	unsigned long seq[2];
+	unsigned long lock_count[2];
+	unsigned long unlock_count[2];
 };
 
 struct rcu_batch {

kernel/rcu/rcutorture.c +0 -0

index 87c51225ceec..6e4fd7680c70 100644

kernel/rcu/rcutorture.c +2 -1

 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->seq[idx]);
+		struct srcu_struct_array* cpu_counts =
+			per_cpu_ptr(sp->per_cpu_ref, cpu);

kernel/rcu/rcutorture.c +4 -0

+	 * If the locks are the same as the unlocks, then there must of have
+	 * been no readers on this index at some time in between. This does not
+	 * mean that there are no more readers, as one could have read the
+	 * current index but have incremented the lock counter yet.

Noticed a few minor nits:

Quoted 8 lines

From: Lance Roy <[email protected]>

SRCU uses two per-cpu counters: a nesting counter to count the number of active critical sections, and a sequence counter to ensure that the nesting counters don't change while they are being added together in srcu_readers_active_idx_check().

Quoted 5 lines

This patch instead uses per-cpu lock and unlock counters. Because the both counters only increase and srcu_readers_active_idx_check() reads the unlock counter before the lock counter, this achieves the same end without having to increment two different counters in srcu_read_lock(). This also saves a smp_mb() in srcu_readers_active_idx_check().

typo:

s/Because the both counters
Because both counters

Quoted 3 lines

A possible problem with this patch is that it can only handle ULONG_MAX - NR_CPUS simultaneous readers, whereas the old version could handle up to ULONG_MAX.

I don't think this is a problem! :-)

Quoted 35 lines

Suggested-by Mathieu Desnoyers <[email protected]>

Signed-off-by Lance Roy <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

Cc Lai Jiangshan <[email protected]>

Cc Peter Zijlstra <[email protected]>

---
 include/linux/srcu.h    |   4 +-
 kernel/rcu/rcutorture.c |  18 +++++++-
 kernel/rcu/srcu.c       | 117 ++++++++++++++++++------------------------------
 3 files changed, 62 insertions(+), 77 deletions(-)

include/linux/srcu.h +2 -2

index dc8eb63c6568..0caea34d8c5f 100644
@@ -34,8 +34,8 @@
 #include <linux/workqueue.h>
 
 struct srcu_struct_array {
-	unsigned long c[2];
-	unsigned long seq[2];
+	unsigned long lock_count[2];
+	unsigned long unlock_count[2];
 };
 
 struct rcu_batch {

kernel/rcu/rcutorture.c +1 -0

index 87c51225ceec..6e4fd7680c70 100644
@@ -564,10 +564,24 @@ static void srcu_torture_stats(void)
 	pr_alert("%s%s per-CPU(idx=%d):",
 		 torture_type, TORTURE_FLAG, idx);
 	for_each_possible_cpu(cpu) {
+		unsigned long l0, l1;

kernel/rcu/rcutorture.c static void srcu_torture_stats(void) · line 571

Quoted 4 lines

kernel/rcu/rcutorture.c +3 -0

+		unsigned long u0, u1;
 		long c0, c1;
+		struct srcu_struct_array* counts =
+			per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu);

Please don't break the line to pacify checkpatch - if the line is too long then maybe split out the loop body into a helper function - but keeping it a bit longer than 80 cols is fine as well.

kernel/rcu/rcutorture.c static void srcu_torture_stats(void) · line 578

Quoted 8 lines

kernel/rcu/rcutorture.c +6 -2

-		c0 = (long)per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu)->c[!idx];
-		c1 = (long)per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu)->c[idx];
+		u0 = counts->unlock_count[!idx];
+		u1 = counts->unlock_count[idx];
+
+		/* Make sure that a lock is always counted if the corresponding
+		   unlock is counted. */
+		smp_rmb();

That's not the standard kernel code comment style.

kernel/rcu/rcutorture.c static void srcu_torture_stats(void) · line 584

Quoted 6 lines

kernel/rcu/rcutorture.c +6 -0

+
+		l0 = counts->lock_count[!idx];
+		l1 = counts->lock_count[idx];
+
+		c0 = (long)(l0 - u0);
+		c1 = (long)(l1 - u1);

These type casts look unnecessary to me.

Quoted 4 lines

kernel/rcu/rcutorture.c +2 -1

 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->seq[idx]);
+		struct srcu_struct_array* cpu_counts =
+			per_cpu_ptr(sp->per_cpu_ref, cpu);

kernel/rcu/rcutorture.c static void srcu_torture_stats(void)

Quoted 10 lines

kernel/rcu/rcutorture.c +4 -1

+		t = READ_ONCE(cpu_counts->lock_count[idx]);
 		sum += t;
 
 
 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[idx]);
+		struct srcu_struct_array* cpu_counts =
+			per_cpu_ptr(sp->per_cpu_ref, cpu);
+		t = READ_ONCE(cpu_counts->unlock_count[idx]);
 		sum += t;

These linebreak look ugly as well. Some abbreviation of types and variables might help:

s/srcu_struct_array/srcu_array
s/cpu_counts/cpuc

?

Quoted 4 lines

kernel/rcu/rcutorture.c +4 -0

+	 * If the locks are the same as the unlocks, then there must of have
+	 * been no readers on this index at some time in between. This does not
+	 * mean that there are no more readers, as one could have read the
+	 * current index but have incremented the lock counter yet.

kernel/rcu/rcutorture.c

Quoted 10 lines

kernel/rcu/rcutorture.c +9 -0

 
+	 * Note that there can be at most NR_CPUS worth of readers using the old
+	 * index that haven't incremented ->lock_count[] yet.  Therefore, the
+	 * sum of the ->lock_count[]s cannot increment enough times to overflow
+	 * and end up equal the sum of the ->unlock_count[]s, as long as there
+	 * are at most ULONG_MAX - NR_CPUS readers at a time.  (Yes, this does
+	 * mean that systems having more than a billion or so CPUs need to be
+	 * 64-bit systems.)  Therefore, the only way that the return values of
+	 * the two calls to srcu_readers_(un)lock_idx() can be equal is if there
+	 * are no active readers using this index.

typo:

s/must of have been no readers/
must have been no readers

Also, maybe I'm misreading it, but shouldn't it be:

s/as one could have read the current index but have incremented the lock counter yet.
/as one could have read the current index but not have incremented the lock counter yet.

?

Also, the title:

srcu: More efficient reader counts.

should have a verb and no full stop, i.e. something like:

srcu: Implement more efficient reader counts

Thanks,

Ingo

If a process invokes synchronize_srcu(), is delayed just the right amount of time, and thus does not sleep when waiting for the grace period to complete, there is no ordering between the end of the grace period and the code following the synchronize_srcu(). Similarly, there can be a lack of ordering between the end of the SRCU grace period and callback invocation.

This commit adds the necessary ordering.

Reported-by Lance Roy <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

---
 include/linux/rcupdate.h | 12 ++++++++++++
 kernel/rcu/srcu.c        |  5 +++++
 kernel/rcu/tree.h        | 12 ------------
 3 files changed, 17 insertions(+), 12 deletions(-)

include/linux/rcupdate.h +0 -0

index 01f71e1d2e94..608d56f908f2 100644

Quoted 22 lines · Paul E. McKenney

If a process invokes synchronize_srcu(), is delayed just the right amount of time, and thus does not sleep when waiting for the grace period to complete, there is no ordering between the end of the grace period and the code following the synchronize_srcu(). Similarly, there can be a lack of ordering between the end of the SRCU grace period and callback invocation.

This commit adds the necessary ordering.

Reported-by Lance Roy <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

---
 include/linux/rcupdate.h | 12 ++++++++++++
 kernel/rcu/srcu.c        |  5 +++++
 kernel/rcu/tree.h        | 12 ------------
 3 files changed, 17 insertions(+), 12 deletions(-)

include/linux/rcupdate.h +0 -0

index 01f71e1d2e94..608d56f908f2 100644

include/linux/rcupdate.h do { \ · line 1168

Quoted 8 lines

include/linux/rcupdate.h +5 -0

@@ -1161,5 +1161,17 @@ do { \
 		ftrace_dump(oops_dump_mode); \
 } while (0)
 
+/*
+ * Place this after a lock-acquisition primitive to guarantee that
+ * an UNLOCK+LOCK pair act as a full barrier.  This guarantee applies
+ * if the UNLOCK and LOCK are executed by the same CPU or if the
+ * UNLOCK and LOCK operate on the same lock variable.

minor typo:

s/an UNLOCK+LOCK pair act as
an UNLOCK+LOCK pair acts as

include/linux/rcupdate.h do { \ · line 1174

Quoted 6 lines

include/linux/rcupdate.h +6 -0

+ */
+#ifdef CONFIG_PPC
+#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for lock. */
+#else /* #ifdef CONFIG_PPC */
+#define smp_mb__after_unlock_lock()	do { } while (0)
+#endif /* #else #ifdef CONFIG_PPC */

Yeah, so I realize that this was pre-existing code, but putting CONFIG_$ARCH #ifdefs into generic headers is generally frowned upon.

The canonical approach would be either to define a helper Kconfig variable that can be set by PPC (but other architectures don't need to set it), or to expose a suitable macro (function) for architectures to define in their barrier.h arch header file.

Quoted 1 line · Ingo Molnar

* Paul E. McKenney <[email protected]> wrote:

Quoted 10 lines

Review

Quoted 6 lines

+ */

(quoted diff) +5 -0

+#ifdef CONFIG_PPC
+#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for lock. */
+#else /* #ifdef CONFIG_PPC */
+#define smp_mb__after_unlock_lock()	do { } while (0)
+#endif /* #else #ifdef CONFIG_PPC */

Yeah, so I realize that this was pre-existing code, but putting CONFIG_$ARCH #ifdefs into generic headers is generally frowned upon.

Quoted 4 lines

The canonical approach would be either to define a helper Kconfig variable that can be set by PPC (but other architectures don't need to set it), or to expose a suitable macro (function) for architectures to define in their barrier.h arch header file.

Very well, I will add a separate commit for this. 4.11 OK?

include/linux/rcupdate.h +0 -0

index 357b32aaea48..5fdfe874229e 100644
@@ -1175,11 +1175,11 @@ do { \
  * if the UNLOCK and LOCK are executed by the same CPU or if the
  * UNLOCK and LOCK operate on the same lock variable.
  */

Quoted 8 lines · Paul E. McKenney

include/linux/rcupdate.h +0 -0

index 357b32aaea48..5fdfe874229e 100644
@@ -1175,11 +1175,11 @@ do { \
  * if the UNLOCK and LOCK are executed by the same CPU or if the
  * UNLOCK and LOCK operate on the same lock variable.
  */

include/linux/rcupdate.h do { \ · line 1184

Quoted 10 lines

include/linux/rcupdate.h +3 -3

-#ifdef CONFIG_PPC
+#ifdef CONFIG_ARCH_WEAK_RELACQ
 #define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for lock. */
-#else /* #ifdef CONFIG_PPC */
+#else /* #ifdef CONFIG_ARCH_WEAK_RELACQ */
 #define smp_mb__after_unlock_lock()	do { } while (0)
-#endif /* #else #ifdef CONFIG_PPC */
+#endif /* #else #ifdef CONFIG_ARCH_WEAK_RELACQ */
 
 

So at the risk of sounding totally pedantic, why not structure it like the existing smp_mb__before/after*() primitives in barrier.h?

That allows asm-generic/barrier.h to pick up the definition - for example in the case of smp_acquire__after_ctrl_dep() we do:

#ifndef smp_acquire__after_ctrl_dep
#define smp_acquire__after_ctrl_dep() smp_rmb()
#endif

Which allows Tile to relax it:

arch/tile/include/asm/barrier.h:#define smp_acquire__after_ctrl_dep() barrier()

I.e. I'd move the API definition out of rcupdate.h and into barrier.h - even though tree-RCU is the only user of this barrier type.

Quoted 34 lines · Ingo Molnar

Quoted 8 lines · Paul E. McKenney

include/linux/rcupdate.h +0 -0

index 357b32aaea48..5fdfe874229e 100644
@@ -1175,11 +1175,11 @@ do { \
  * if the UNLOCK and LOCK are executed by the same CPU or if the
  * UNLOCK and LOCK operate on the same lock variable.
  */

include/linux/rcupdate.h do { \ · line 1184

Quoted 10 lines

include/linux/rcupdate.h +3 -3

-#ifdef CONFIG_PPC
+#ifdef CONFIG_ARCH_WEAK_RELACQ
 #define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for lock. */
-#else /* #ifdef CONFIG_PPC */
+#else /* #ifdef CONFIG_ARCH_WEAK_RELACQ */
 #define smp_mb__after_unlock_lock()	do { } while (0)
-#endif /* #else #ifdef CONFIG_PPC */
+#endif /* #else #ifdef CONFIG_ARCH_WEAK_RELACQ */
 
 

So at the risk of sounding totally pedantic, why not structure it like the existing smp_mb__before/after*() primitives in barrier.h?

That allows asm-generic/barrier.h to pick up the definition - for example in the case of smp_acquire__after_ctrl_dep() we do:

#ifndef smp_acquire__after_ctrl_dep
#define smp_acquire__after_ctrl_dep() smp_rmb()
#endif

Which allows Tile to relax it:

Quoted 4 lines

arch/tile/include/asm/barrier.h:#define smp_acquire__after_ctrl_dep() barrier()

I.e. I'd move the API definition out of rcupdate.h and into barrier.h - even though tree-RCU is the only user of this barrier type.

I wouldn't have any problem with that, however, some time back it was moved into RCU because (you guessed it!) RCU is the only user. ;-)

Quoted 35 lines · Paul E. McKenney

Quoted 34 lines · Ingo Molnar

Quoted 8 lines · Paul E. McKenney

include/linux/rcupdate.h +0 -0

index 357b32aaea48..5fdfe874229e 100644
@@ -1175,11 +1175,11 @@ do { \
  * if the UNLOCK and LOCK are executed by the same CPU or if the
  * UNLOCK and LOCK operate on the same lock variable.
  */

include/linux/rcupdate.h do { \ · line 1184

Quoted 10 lines

include/linux/rcupdate.h +3 -3

-#ifdef CONFIG_PPC
+#ifdef CONFIG_ARCH_WEAK_RELACQ
 #define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for lock. */
-#else /* #ifdef CONFIG_PPC */
+#else /* #ifdef CONFIG_ARCH_WEAK_RELACQ */
 #define smp_mb__after_unlock_lock()	do { } while (0)
-#endif /* #else #ifdef CONFIG_PPC */
+#endif /* #else #ifdef CONFIG_ARCH_WEAK_RELACQ */
 
 

So at the risk of sounding totally pedantic, why not structure it like the existing smp_mb__before/after*() primitives in barrier.h?

That allows asm-generic/barrier.h to pick up the definition - for example in the case of smp_acquire__after_ctrl_dep() we do:

#ifndef smp_acquire__after_ctrl_dep
#define smp_acquire__after_ctrl_dep() smp_rmb()
#endif

Which allows Tile to relax it:

Quoted 7 lines

Quoted 4 lines

arch/tile/include/asm/barrier.h:#define smp_acquire__after_ctrl_dep() barrier()

I.e. I'd move the API definition out of rcupdate.h and into barrier.h - even though tree-RCU is the only user of this barrier type.

I wouldn't have any problem with that, however, some time back it was moved into RCU because (you guessed it!) RCU is the only user. ;-)

Indeed ...

[sounds of rummaging around in the Git tree]

I found this commit of yours from ancient history (more than a year ago!):

commit 12d560f4ea87030667438a169912380be00cea4b
Author: Paul E. McKenney <[email protected]>
Date: Tue Jul 14 18:35:23 2015 -0700

rcu,locking: Privatize smp_mb__after_unlock_lock()

RCU is the only thing that uses smp_mb__after_unlock_lock(), and is likely the only thing that ever will use it, so this commit makes this macro private to RCU.

Signed-off-by Paul E. McKenney <[email protected]>

Cc Will Deacon <[email protected]>

Cc Peter Zijlstra <[email protected]>

Cc Benjamin Herrenschmidt <[email protected]>

Cc "[email protected]" <[email protected]>

So I concur and I'm fine with your patch - or with the status quo code as well.

Quoted 20 lines

[sounds of rummaging around in the Git tree]

I found this commit of yours from ancient history (more than a year ago!):

commit 12d560f4ea87030667438a169912380be00cea4b
Author: Paul E. McKenney <[email protected]>
Date: Tue Jul 14 18:35:23 2015 -0700

rcu,locking: Privatize smp_mb__after_unlock_lock()

RCU is the only thing that uses smp_mb__after_unlock_lock(), and is likely the only thing that ever will use it, so this commit makes this macro private to RCU.

Signed-off-by Paul E. McKenney <[email protected]>

Cc Will Deacon <[email protected]>

Cc Peter Zijlstra <[email protected]>

Cc Benjamin Herrenschmidt <[email protected]>

Cc "[email protected]" <[email protected]>

Quoted 1 line

So I concur and I'm fine with your patch - or with the status quo code as well.

I already have the patch queued, so how about I keep it if I get an ack from the powerpc guys and drop it otherwise?

Quoted 20 lines · Paul E. McKenney

Quoted 20 lines

[sounds of rummaging around in the Git tree]

I found this commit of yours from ancient history (more than a year ago!):

commit 12d560f4ea87030667438a169912380be00cea4b
Author: Paul E. McKenney <[email protected]>
Date: Tue Jul 14 18:35:23 2015 -0700

rcu,locking: Privatize smp_mb__after_unlock_lock()

RCU is the only thing that uses smp_mb__after_unlock_lock(), and is likely the only thing that ever will use it, so this commit makes this macro private to RCU.

Signed-off-by Paul E. McKenney <[email protected]>

Cc Will Deacon <[email protected]>

Cc Peter Zijlstra <[email protected]>

Cc Benjamin Herrenschmidt <[email protected]>

Cc "[email protected]" <[email protected]>

Quoted 4 lines

Quoted 1 line

So I concur and I'm fine with your patch - or with the status quo code as well.

I already have the patch queued, so how about I keep it if I get an ack from the powerpc guys and drop it otherwise?

Yeah, sounds good! Your patch made me look up 'RelAcq' so it has documentation value as well ;-)

Quoted 23 lines · Ingo Molnar

Quoted 20 lines · Paul E. McKenney

Quoted 20 lines

[sounds of rummaging around in the Git tree]

I found this commit of yours from ancient history (more than a year ago!):

commit 12d560f4ea87030667438a169912380be00cea4b
Author: Paul E. McKenney <[email protected]>
Date: Tue Jul 14 18:35:23 2015 -0700

rcu,locking: Privatize smp_mb__after_unlock_lock()

RCU is the only thing that uses smp_mb__after_unlock_lock(), and is likely the only thing that ever will use it, so this commit makes this macro private to RCU.

Signed-off-by Paul E. McKenney <[email protected]>

Cc Will Deacon <[email protected]>

Cc Peter Zijlstra <[email protected]>

Cc Benjamin Herrenschmidt <[email protected]>

Cc "[email protected]" <[email protected]>

Quoted 7 lines

Quoted 4 lines

Quoted 1 line

So I concur and I'm fine with your patch - or with the status quo code as well.

I already have the patch queued, so how about I keep it if I get an ack from the powerpc guys and drop it otherwise?

Yeah, sounds good! Your patch made me look up 'RelAcq' so it has documentation value as well ;-)

;-) ;-) ;-)

Looking forward, my guess would be that if some other code needs smp_mb__after_unlock_lock() or if some other architecture needs non-smb_mb() special handling, I should consider making it work the same as smp_mb__after_atomic() and friends. Does that seem like a reasonable thought?

Quoted 2 lines · Paul E. McKenney

Quoted 1 line · Ingo Molnar

* Paul E. McKenney <[email protected]> wrote:

[ . . . ]

Quoted 10 lines

Quoted 10 lines

Review

Quoted 6 lines

+ */

(quoted diff) +5 -0

+#ifdef CONFIG_PPC
+#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for lock. */
+#else /* #ifdef CONFIG_PPC */
+#define smp_mb__after_unlock_lock()	do { } while (0)
+#endif /* #else #ifdef CONFIG_PPC */

Yeah, so I realize that this was pre-existing code, but putting CONFIG_$ARCH #ifdefs into generic headers is generally frowned upon.

Quoted 6 lines

Quoted 4 lines

The canonical approach would be either to define a helper Kconfig variable that can be set by PPC (but other architectures don't need to set it), or to expose a suitable macro (function) for architectures to define in their barrier.h arch header file.

Very well, I will add a separate commit for this. 4.11 OK?

Does the patch below seem reasonable?

Thanx, Paul

------------------------------------------------------------------------

commit 271c0601237c41a279f975563e13837bace0df03

Author: Paul E. McKenney <[email protected]>
Date:   Sat Jan 14 13:32:50 2017 -0800

rcu: Make arch select smp_mb__after_unlock_lock() strength

The definition of smp_mb__after_unlock_lock() is currently smp_mb() for CONFIG_PPC and a no-op otherwise. It would be better to instead provide an architecture-selectable Kconfig option, and select the strength of smp_mb__after_unlock_lock() based on that option. This commit therefore creates CONFIG_ARCH_WEAK_RELACQ, has PPC select it, and bases the definition of smp_mb__after_unlock_lock() on this new CONFIG_ARCH_WEAK_RELACQ Kconfig option.

Reported-by Ingo Molnar <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

Cc Peter Zijlstra <[email protected]>

Cc Will Deacon <[email protected]>

Cc Boqun Feng <[email protected]>

Cc Benjamin Herrenschmidt <[email protected]>

Cc Paul Mackerras <[email protected]>

Cc Michael Ellerman <[email protected]>

Cc <[email protected]>

Quoted 43 lines · Paul E. McKenney

Quoted 2 lines · Paul E. McKenney

Quoted 1 line · Ingo Molnar

* Paul E. McKenney <[email protected]> wrote:

[ . . . ]

Quoted 10 lines

Quoted 10 lines

Review

Quoted 6 lines

+ */

(quoted diff) +5 -0

+#ifdef CONFIG_PPC
+#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for lock. */
+#else /* #ifdef CONFIG_PPC */
+#define smp_mb__after_unlock_lock()	do { } while (0)
+#endif /* #else #ifdef CONFIG_PPC */

Yeah, so I realize that this was pre-existing code, but putting CONFIG_$ARCH #ifdefs into generic headers is generally frowned upon.

Quoted 6 lines

Quoted 4 lines

The canonical approach would be either to define a helper Kconfig variable that can be set by PPC (but other architectures don't need to set it), or to expose a suitable macro (function) for architectures to define in their barrier.h arch header file.

Very well, I will add a separate commit for this. 4.11 OK?

Does the patch below seem reasonable?

Thanx, Paul

------------------------------------------------------------------------

commit 271c0601237c41a279f975563e13837bace0df03

Author: Paul E. McKenney <[email protected]>
Date:   Sat Jan 14 13:32:50 2017 -0800

rcu: Make arch select smp_mb__after_unlock_lock() strength

The definition of smp_mb__after_unlock_lock() is currently smp_mb() for CONFIG_PPC and a no-op otherwise. It would be better to instead provide an architecture-selectable Kconfig option, and select the strength of smp_mb__after_unlock_lock() based on that option. This commit therefore creates CONFIG_ARCH_WEAK_RELACQ, has PPC select it, and bases the definition of smp_mb__after_unlock_lock() on this new CONFIG_ARCH_WEAK_RELACQ Kconfig option.

Quoted 9 lines

Reported-by Ingo Molnar <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

Cc Peter Zijlstra <[email protected]>

Cc Will Deacon <[email protected]>

Cc Boqun Feng <[email protected]>

Cc Benjamin Herrenschmidt <[email protected]>

Cc Paul Mackerras <[email protected]>

Cc Michael Ellerman <[email protected]>

Cc <[email protected]>

Personally I'd call it ARCH_WEAK_RELEASE_ACQUIRE, which is longer but clearer I think. But it's not a big deal, so which ever you prefer.

Acked-by Michael Ellerman <[email protected]>

Quoted 10 lines

include/linux/rcupdate.h do { \ · line 1174

Quoted 6 lines

+ */

include/linux/rcupdate.h +5 -0

+#ifdef CONFIG_PPC
+#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for lock. */
+#else /* #ifdef CONFIG_PPC */
+#define smp_mb__after_unlock_lock()	do { } while (0)
+#endif /* #else #ifdef CONFIG_PPC */

Yeah, so I realize that this was pre-existing code, but putting CONFIG_$ARCH #ifdefs into generic headers is generally frowned upon.

Quoted 4 lines

The canonical approach would be either to define a helper Kconfig variable that can be set by PPC (but other architectures don't need to set it), or to expose a suitable macro (function) for architectures to define in their barrier.h arch header file.

Very well, I will add a separate commit for this. 4.11 OK?

SRCU uses two per-cpu counters: a nesting counter to count the number of active critical sections, and a sequence counter to ensure that the nesting counters don't change while they are being added together in srcu_readers_active_idx_check().

This patch instead uses per-cpu lock and unlock counters. Because the both counters only increase and srcu_readers_active_idx_check() reads the unlock counter before the lock counter, this achieves the same end without having to increment two different counters in srcu_read_lock(). This also saves a smp_mb() in srcu_readers_active_idx_check().

Possible bug: There is no guarantee that the lock counter won't overflow during srcu_readers_active_idx_check(), as there are no memory barriers around srcu_flip() (see comment in srcu_readers_active_idx_check() for details). However, this problem was already present before this patch.

Suggested-by Mathieu Desnoyers <[email protected]>

Signed-off-by Lance Roy <[email protected]>

---
 include/linux/srcu.h    |   4 +-
 kernel/rcu/rcutorture.c |  20 ++++++++-
 kernel/rcu/srcu.c       | 116 ++++++++++++++++++------------------------------
 3 files changed, 63 insertions(+), 77 deletions(-)

include/linux/srcu.h +2 -2

index dc8eb63..0caea34 100644
@@ -34,8 +34,8 @@
 #include <linux/workqueue.h>
 
 struct srcu_struct_array {
-	unsigned long c[2];
-	unsigned long seq[2];
+	unsigned long lock_count[2];
+	unsigned long unlock_count[2];
 };
 
 struct rcu_batch {

kernel/rcu/rcutorture.c +18 -2

index 87c5122..c3f25d1 100644
@@ -564,10 +564,26 @@ static void srcu_torture_stats(void)
 	pr_alert("%s%s per-CPU(idx=%d):",
 		 torture_type, TORTURE_FLAG, idx);
 	for_each_possible_cpu(cpu) {
+		unsigned long l0, l1;
+		unsigned long u0, u1;
 		long c0, c1;
+		struct srcu_struct_array *counts =
+			per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu);
 
-		c0 = (long)per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu)->c[!idx];
-		c1 = (long)per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu)->c[idx];
+		u0 = counts->unlock_count[!idx];
+		u1 = counts->unlock_count[idx];
+
+		/*
+		 * Make sure that a lock is always counted if the corresponding
+		 * unlock is counted.
+		 */
+		smp_rmb();
+
+		l0 = counts->lock_count[!idx];
+		l1 = counts->lock_count[idx];
+
+		c0 = (long)(l0 - u0);
+		c1 = (long)(l1 - u1);
 		pr_cont(" %d(%ld,%ld)", cpu, c0, c1);
 	}
 	pr_cont("\n");

kernel/rcu/srcu.c +43 -73

index 9b9cdd5..38e9aae 100644
@@ -141,34 +141,38 @@ EXPORT_SYMBOL_GPL(init_srcu_struct);
 #endif /* #else #ifdef CONFIG_DEBUG_LOCK_ALLOC */
 
 /*
- * Returns approximate total of the readers' ->seq[] values for the
+ * Returns approximate total of the readers' ->lock_count[] values for the
  * rank of per-CPU counters specified by idx.
  */
-static unsigned long srcu_readers_seq_idx(struct srcu_struct *sp, int idx)
+static unsigned long srcu_readers_lock_idx(struct srcu_struct *sp, int idx)
 {
 	int cpu;
 	unsigned long sum = 0;
 	unsigned long t;
 
 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->seq[idx]);
+		struct srcu_struct_array *cpu_counts =
+			per_cpu_ptr(sp->per_cpu_ref, cpu);
+		t = READ_ONCE(cpu_counts->lock_count[idx]);
 		sum += t;
 	}
 	return sum;
 }
 
 /*
- * Returns approximate number of readers active on the specified rank
- * of the per-CPU ->c[] counters.
+ * Returns approximate total of the readers' ->unlock_count[] values for the
+ * rank of per-CPU counters specified by idx.
  */
-static unsigned long srcu_readers_active_idx(struct srcu_struct *sp, int idx)
+static unsigned long srcu_readers_unlock_idx(struct srcu_struct *sp, int idx)
 {
 	int cpu;
 	unsigned long sum = 0;
 	unsigned long t;
 
 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[idx]);
+		struct srcu_struct_array *cpu_counts =
+			per_cpu_ptr(sp->per_cpu_ref, cpu);
+		t = READ_ONCE(cpu_counts->unlock_count[idx]);
 		sum += t;
 	}
 	return sum;

@@ -176,79 +180,42 @@ static unsigned long srcu_readers_active_idx(struct srcu_struct *sp, int idx)
 
 /*
  * Return true if the number of pre-existing readers is determined to
- * be stably zero.  An example unstable zero can occur if the call
- * to srcu_readers_active_idx() misses an __srcu_read_lock() increment,
- * but due to task migration, sees the corresponding __srcu_read_unlock()
- * decrement.  This can happen because srcu_readers_active_idx() takes
- * time to sum the array, and might in fact be interrupted or preempted
- * partway through the summation.
+ * be zero.
  */
 static bool srcu_readers_active_idx_check(struct srcu_struct *sp, int idx)
 {
-	unsigned long seq;
+	unsigned long unlocks;
 
-	seq = srcu_readers_seq_idx(sp, idx);
+	unlocks = srcu_readers_unlock_idx(sp, idx);
 
 	/*
-	 * The following smp_mb() A pairs with the smp_mb() B located in
-	 * __srcu_read_lock().  This pairing ensures that if an
-	 * __srcu_read_lock() increments its counter after the summation
-	 * in srcu_readers_active_idx(), then the corresponding SRCU read-side
-	 * critical section will see any changes made prior to the start
-	 * of the current SRCU grace period.
+	 * Make sure that a lock is always counted if the corresponding unlock
+	 * is counted. Needs to be a smp_mb() as the read side may contain a
+	 * read from a variable that is written to before the synchronize_srcu()
+	 * in the write side. In this case smp_mb()s A and B act like the store
+	 * buffering pattern.
 	 *
-	 * Also, if the above call to srcu_readers_seq_idx() saw the
-	 * increment of ->seq[], then the call to srcu_readers_active_idx()
-	 * must see the increment of ->c[].
+	 * This smp_mb() also pairs with smp_mb() C to prevent writes after the
+	 * synchronize_srcu() from being executed before the grace period ends.
 	 */
 	smp_mb(); /* A */
 
 	/*
-	 * Note that srcu_readers_active_idx() can incorrectly return
-	 * zero even though there is a pre-existing reader throughout.
-	 * To see this, suppose that task A is in a very long SRCU
-	 * read-side critical section that started on CPU 0, and that
-	 * no other reader exists, so that the sum of the counters
-	 * is equal to one.  Then suppose that task B starts executing
-	 * srcu_readers_active_idx(), summing up to CPU 1, and then that
-	 * task C starts reading on CPU 0, so that its increment is not
-	 * summed, but finishes reading on CPU 2, so that its decrement
-	 * -is- summed.  Then when task B completes its sum, it will
-	 * incorrectly get zero, despite the fact that task A has been
-	 * in its SRCU read-side critical section the whole time.
-	 *
-	 * We therefore do a validation step should srcu_readers_active_idx()
-	 * return zero.
-	 */
-	if (srcu_readers_active_idx(sp, idx) != 0)
-		return false;
-
-	/*
-	 * The remainder of this function is the validation step.
-	 * The following smp_mb() D pairs with the smp_mb() C in
-	 * __srcu_read_unlock().  If the __srcu_read_unlock() was seen
-	 * by srcu_readers_active_idx() above, then any destructive
-	 * operation performed after the grace period will happen after
-	 * the corresponding SRCU read-side critical section.
+	 * If the locks are the same as the unlocks, then there must of have
+	 * been no readers on this index at some time in between. This does not
+	 * mean that there are no more readers, as one could have read the
+	 * current index but have incremented the lock counter yet.
 	 *
-	 * Note that there can be at most NR_CPUS worth of readers using
-	 * the old index, which is not enough to overflow even a 32-bit
-	 * integer.  (Yes, this does mean that systems having more than
-	 * a billion or so CPUs need to be 64-bit systems.)  Therefore,
-	 * the sum of the ->seq[] counters cannot possibly overflow.
-	 * Therefore, the only way that the return values of the two
-	 * calls to srcu_readers_seq_idx() can be equal is if there were
-	 * no increments of the corresponding rank of ->seq[] counts
-	 * in the interim.  But the missed-increment scenario laid out
-	 * above includes an increment of the ->seq[] counter by
-	 * the corresponding __srcu_read_lock().  Therefore, if this
-	 * scenario occurs, the return values from the two calls to
-	 * srcu_readers_seq_idx() will differ, and thus the validation
-	 * step below suffices.
+	 * Possible bug: There is no guarantee that there haven't been ULONG_MAX
+	 * increments of ->lock_count[] since the unlocks were counted, meaning
+	 * that this could return true even if there are still active readers.
+	 * Since there are no memory barriers around srcu_flip(), the CPU is not
+	 * required to increment ->completed before running
+	 * srcu_readers_unlock_idx(), which means that there could be an
+	 * arbitrarily large number of critical sections that execute after
+	 * srcu_readers_unlock_idx() but use the old value of ->completed.
 	 */
-	smp_mb(); /* D */
-
-	return srcu_readers_seq_idx(sp, idx) == seq;
+	return srcu_readers_lock_idx(sp, idx) == unlocks;
 }
 
 /**

@@ -266,8 +233,12 @@ static bool srcu_readers_active(struct srcu_struct *sp)
 	unsigned long sum = 0;
 
 	for_each_possible_cpu(cpu) {
-		sum += READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[0]);
-		sum += READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[1]);
+		struct srcu_struct_array *cpu_counts =
+			per_cpu_ptr(sp->per_cpu_ref, cpu);
+		sum += READ_ONCE(cpu_counts->lock_count[0]);
+		sum += READ_ONCE(cpu_counts->lock_count[1]);
+		sum -= READ_ONCE(cpu_counts->unlock_count[0]);
+		sum -= READ_ONCE(cpu_counts->unlock_count[1]);
 	}
 	return sum;
 }

@@ -298,9 +269,8 @@ int __srcu_read_lock(struct srcu_struct *sp)
 	int idx;
 
 	idx = READ_ONCE(sp->completed) & 0x1;
-	__this_cpu_inc(sp->per_cpu_ref->c[idx]);
+	__this_cpu_inc(sp->per_cpu_ref->lock_count[idx]);
 	smp_mb(); /* B */  /* Avoid leaking the critical section. */
-	__this_cpu_inc(sp->per_cpu_ref->seq[idx]);
 	return idx;
 }
 EXPORT_SYMBOL_GPL(__srcu_read_lock);

@@ -314,7 +284,7 @@ EXPORT_SYMBOL_GPL(__srcu_read_lock);
 void __srcu_read_unlock(struct srcu_struct *sp, int idx)
 {
 	smp_mb(); /* C */  /* Avoid leaking the critical section. */
-	this_cpu_dec(sp->per_cpu_ref->c[idx]);
+	this_cpu_inc(sp->per_cpu_ref->unlock_count[idx]);
 }
 EXPORT_SYMBOL_GPL(__srcu_read_unlock);
 

@@ -349,7 +319,7 @@ static bool try_check_zero(struct srcu_struct *sp, int idx, int trycount)
 
 /*
  * Increment the ->completed counter so that future SRCU readers will
- * use the other rank of the ->c[] and ->seq[] arrays.  This allows
+ * use the other rank of the ->(un)lock_count[] arrays.  This allows
  * us to wait for pre-existing readers in a starvation-free manner.
  */
 static void srcu_flip(struct srcu_struct *sp)

Signature · 1 line

2.9.0

Quoted 11 lines · Lance Roy

SRCU uses two per-cpu counters: a nesting counter to count the number of active critical sections, and a sequence counter to ensure that the nesting counters don't change while they are being added together in srcu_readers_active_idx_check().

This patch instead uses per-cpu lock and unlock counters. Because the both counters only increase and srcu_readers_active_idx_check() reads the unlock counter before the lock counter, this achieves the same end without having to increment two different counters in srcu_read_lock(). This also saves a smp_mb() in srcu_readers_active_idx_check().

Quoted 7 lines

Possible bug: There is no guarantee that the lock counter won't overflow during srcu_readers_active_idx_check(), as there are no memory barriers around srcu_flip() (see comment in srcu_readers_active_idx_check() for details). However, this problem was already present before this patch.

Suggested-by Mathieu Desnoyers <[email protected]>

Signed-off-by Lance Roy <[email protected]>

In general, the comment update looks good, but this patch undoes my application of review feedback to your original patch. Could you please submit the comment update as a separate patch on top of branch rcu/next of my -rcu tree?

git://git.kernel.org/pub/scm/linux/kernel/git/paulmck/linux-rcu.git

Of course, if any of the changes from the review feedback are in any way problematic, please do let me know.

SRCU uses two per-cpu counters: a nesting counter to count the number of active critical sections, and a sequence counter to ensure that the nesting counters don't change while they are being added together in srcu_readers_active_idx_check().

This patch instead uses per-cpu lock and unlock counters. Because both counters only increase and srcu_readers_active_idx_check() reads the unlock counter before the lock counter, this achieves the same end without having to increment two different counters in srcu_read_lock(). This also saves a smp_mb() in srcu_readers_active_idx_check().

Possible bug: There is no guarantee that the lock counter won't overflow during srcu_readers_active_idx_check(), as there are no memory barriers around srcu_flip() (see comment in srcu_readers_active_idx_check() for details). However, this problem was already present before this patch.

Suggested-by Mathieu Desnoyers <[email protected]>

Signed-off-by Lance Roy <[email protected]>

Cc Paul E. McKenney <[email protected]>

Cc Lai Jiangshan <[email protected]>

Cc Peter Zijlstra <[email protected]>

---
 include/linux/srcu.h    |  10 ++--
 kernel/rcu/rcutorture.c |  19 +++++++-
 kernel/rcu/srcu.c       | 122 +++++++++++++++++-------------------------------
 3 files changed, 66 insertions(+), 85 deletions(-)

include/linux/srcu.h +5 -5

index dc8eb63..a598cf3 100644
@@ -33,9 +33,9 @@
 #include <linux/rcupdate.h>
 #include <linux/workqueue.h>
 
-struct srcu_struct_array {
-	unsigned long c[2];
-	unsigned long seq[2];
+struct srcu_array {
+	unsigned long lock_count[2];
+	unsigned long unlock_count[2];
 };
 
 struct rcu_batch {

@@ -46,7 +46,7 @@ struct rcu_batch {
 
 struct srcu_struct {
 	unsigned long completed;
-	struct srcu_struct_array __percpu *per_cpu_ref;
+	struct srcu_array __percpu *per_cpu_ref;
 	spinlock_t queue_lock; /* protect ->batch_queue, ->running */
 	bool running;
 	/* callbacks just queued */

@@ -118,7 +118,7 @@ void process_srcu(struct work_struct *work);
  * See include/linux/percpu-defs.h for the rules on per-CPU variables.
  */
 #define __DEFINE_SRCU(name, is_static)					\
-	static DEFINE_PER_CPU(struct srcu_struct_array, name##_srcu_array);\
+	static DEFINE_PER_CPU(struct srcu_array, name##_srcu_array);\
 	is_static struct srcu_struct name = __SRCU_STRUCT_INIT(name)
 #define DEFINE_SRCU(name)		__DEFINE_SRCU(name, /* not static */)
 #define DEFINE_STATIC_SRCU(name)	__DEFINE_SRCU(name, static)

kernel/rcu/rcutorture.c +17 -2

index 87c5122..d81345b 100644
@@ -564,10 +564,25 @@ static void srcu_torture_stats(void)
 	pr_alert("%s%s per-CPU(idx=%d):",
 		 torture_type, TORTURE_FLAG, idx);
 	for_each_possible_cpu(cpu) {
+		unsigned long l0, l1;
+		unsigned long u0, u1;
 		long c0, c1;
+		struct srcu_array *counts = per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu);
 
-		c0 = (long)per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu)->c[!idx];
-		c1 = (long)per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu)->c[idx];
+		u0 = counts->unlock_count[!idx];
+		u1 = counts->unlock_count[idx];
+
+		/*
+		 * Make sure that a lock is always counted if the corresponding
+		 * unlock is counted.
+		 */
+		smp_rmb();
+
+		l0 = counts->lock_count[!idx];
+		l1 = counts->lock_count[idx];
+
+		c0 = l0 - u0;
+		c1 = l1 - u1;
 		pr_cont(" %d(%ld,%ld)", cpu, c0, c1);
 	}
 	pr_cont("\n");

kernel/rcu/srcu.c +44 -78

index 9b9cdd5..c9a0015 100644
@@ -106,7 +106,7 @@ static int init_srcu_struct_fields(struct srcu_struct *sp)
 	rcu_batch_init(&sp->batch_check1);
 	rcu_batch_init(&sp->batch_done);
 	INIT_DELAYED_WORK(&sp->work, process_srcu);
-	sp->per_cpu_ref = alloc_percpu(struct srcu_struct_array);
+	sp->per_cpu_ref = alloc_percpu(struct srcu_array);
 	return sp->per_cpu_ref ? 0 : -ENOMEM;
 }
 

@@ -141,114 +141,77 @@ EXPORT_SYMBOL_GPL(init_srcu_struct);
 #endif /* #else #ifdef CONFIG_DEBUG_LOCK_ALLOC */
 
 /*
- * Returns approximate total of the readers' ->seq[] values for the
+ * Returns approximate total of the readers' ->lock_count[] values for the
  * rank of per-CPU counters specified by idx.
  */
-static unsigned long srcu_readers_seq_idx(struct srcu_struct *sp, int idx)
+static unsigned long srcu_readers_lock_idx(struct srcu_struct *sp, int idx)
 {
 	int cpu;
 	unsigned long sum = 0;
-	unsigned long t;
 
 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->seq[idx]);
-		sum += t;
+		struct srcu_array *cpuc = per_cpu_ptr(sp->per_cpu_ref, cpu);
+
+		sum += READ_ONCE(cpuc->lock_count[idx]);
 	}
 	return sum;
 }
 
 /*
- * Returns approximate number of readers active on the specified rank
- * of the per-CPU ->c[] counters.
+ * Returns approximate total of the readers' ->unlock_count[] values for the
+ * rank of per-CPU counters specified by idx.
  */
-static unsigned long srcu_readers_active_idx(struct srcu_struct *sp, int idx)
+static unsigned long srcu_readers_unlock_idx(struct srcu_struct *sp, int idx)
 {
 	int cpu;
 	unsigned long sum = 0;
-	unsigned long t;
 
 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[idx]);
-		sum += t;
+		struct srcu_array *cpuc = per_cpu_ptr(sp->per_cpu_ref, cpu);
+
+		sum += READ_ONCE(cpuc->unlock_count[idx]);
 	}
 	return sum;
 }
 
 /*
  * Return true if the number of pre-existing readers is determined to
- * be stably zero.  An example unstable zero can occur if the call
- * to srcu_readers_active_idx() misses an __srcu_read_lock() increment,
- * but due to task migration, sees the corresponding __srcu_read_unlock()
- * decrement.  This can happen because srcu_readers_active_idx() takes
- * time to sum the array, and might in fact be interrupted or preempted
- * partway through the summation.
+ * be zero.
  */
 static bool srcu_readers_active_idx_check(struct srcu_struct *sp, int idx)
 {
-	unsigned long seq;
+	unsigned long unlocks;
 
-	seq = srcu_readers_seq_idx(sp, idx);
+	unlocks = srcu_readers_unlock_idx(sp, idx);
 
 	/*
-	 * The following smp_mb() A pairs with the smp_mb() B located in
-	 * __srcu_read_lock().  This pairing ensures that if an
-	 * __srcu_read_lock() increments its counter after the summation
-	 * in srcu_readers_active_idx(), then the corresponding SRCU read-side
-	 * critical section will see any changes made prior to the start
-	 * of the current SRCU grace period.
+	 * Make sure that a lock is always counted if the corresponding unlock
+	 * is counted. Needs to be a smp_mb() as the read side may contain a
+	 * read from a variable that is written to before the synchronize_srcu()
+	 * in the write side. In this case smp_mb()s A and B act like the store
+	 * buffering pattern.
 	 *
-	 * Also, if the above call to srcu_readers_seq_idx() saw the
-	 * increment of ->seq[], then the call to srcu_readers_active_idx()
-	 * must see the increment of ->c[].
+	 * This smp_mb() also pairs with smp_mb() C to prevent accesses after the
+	 * synchronize_srcu() from being executed before the grace period ends.
 	 */
 	smp_mb(); /* A */
 
 	/*
-	 * Note that srcu_readers_active_idx() can incorrectly return
-	 * zero even though there is a pre-existing reader throughout.
-	 * To see this, suppose that task A is in a very long SRCU
-	 * read-side critical section that started on CPU 0, and that
-	 * no other reader exists, so that the sum of the counters
-	 * is equal to one.  Then suppose that task B starts executing
-	 * srcu_readers_active_idx(), summing up to CPU 1, and then that
-	 * task C starts reading on CPU 0, so that its increment is not
-	 * summed, but finishes reading on CPU 2, so that its decrement
-	 * -is- summed.  Then when task B completes its sum, it will
-	 * incorrectly get zero, despite the fact that task A has been
-	 * in its SRCU read-side critical section the whole time.
+	 * If the locks are the same as the unlocks, then there must have
+	 * been no readers on this index at some time in between. This does not
+	 * mean that there are no more readers, as one could have read the
+	 * current index but not have incremented the lock counter yet.
 	 *
-	 * We therefore do a validation step should srcu_readers_active_idx()
-	 * return zero.
+	 * Possible bug: There is no guarantee that there haven't been ULONG_MAX
+	 * increments of ->lock_count[] since the unlocks were counted, meaning
+	 * that this could return true even if there are still active readers.
+	 * Since there are no memory barriers around srcu_flip(), the CPU is not
+	 * required to increment ->completed before running
+	 * srcu_readers_unlock_idx(), which means that there could be an
+	 * arbitrarily large number of critical sections that execute after
+	 * srcu_readers_unlock_idx() but use the old value of ->completed.
 	 */
-	if (srcu_readers_active_idx(sp, idx) != 0)
-		return false;
-
-	/*
-	 * The remainder of this function is the validation step.
-	 * The following smp_mb() D pairs with the smp_mb() C in
-	 * __srcu_read_unlock().  If the __srcu_read_unlock() was seen
-	 * by srcu_readers_active_idx() above, then any destructive
-	 * operation performed after the grace period will happen after
-	 * the corresponding SRCU read-side critical section.
-	 *
-	 * Note that there can be at most NR_CPUS worth of readers using
-	 * the old index, which is not enough to overflow even a 32-bit
-	 * integer.  (Yes, this does mean that systems having more than
-	 * a billion or so CPUs need to be 64-bit systems.)  Therefore,
-	 * the sum of the ->seq[] counters cannot possibly overflow.
-	 * Therefore, the only way that the return values of the two
-	 * calls to srcu_readers_seq_idx() can be equal is if there were
-	 * no increments of the corresponding rank of ->seq[] counts
-	 * in the interim.  But the missed-increment scenario laid out
-	 * above includes an increment of the ->seq[] counter by
-	 * the corresponding __srcu_read_lock().  Therefore, if this
-	 * scenario occurs, the return values from the two calls to
-	 * srcu_readers_seq_idx() will differ, and thus the validation
-	 * step below suffices.
-	 */
-	smp_mb(); /* D */
-
-	return srcu_readers_seq_idx(sp, idx) == seq;
+	return srcu_readers_lock_idx(sp, idx) == unlocks;
 }
 
 /**

@@ -266,8 +229,12 @@ static bool srcu_readers_active(struct srcu_struct *sp)
 	unsigned long sum = 0;
 
 	for_each_possible_cpu(cpu) {
-		sum += READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[0]);
-		sum += READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[1]);
+		struct srcu_array *cpuc = per_cpu_ptr(sp->per_cpu_ref, cpu);
+
+		sum += READ_ONCE(cpuc->lock_count[0]);
+		sum += READ_ONCE(cpuc->lock_count[1]);
+		sum -= READ_ONCE(cpuc->unlock_count[0]);
+		sum -= READ_ONCE(cpuc->unlock_count[1]);
 	}
 	return sum;
 }

@@ -298,9 +265,8 @@ int __srcu_read_lock(struct srcu_struct *sp)
 	int idx;
 
 	idx = READ_ONCE(sp->completed) & 0x1;
-	__this_cpu_inc(sp->per_cpu_ref->c[idx]);
+	__this_cpu_inc(sp->per_cpu_ref->lock_count[idx]);
 	smp_mb(); /* B */  /* Avoid leaking the critical section. */
-	__this_cpu_inc(sp->per_cpu_ref->seq[idx]);
 	return idx;
 }
 EXPORT_SYMBOL_GPL(__srcu_read_lock);

@@ -314,7 +280,7 @@ EXPORT_SYMBOL_GPL(__srcu_read_lock);
 void __srcu_read_unlock(struct srcu_struct *sp, int idx)
 {
 	smp_mb(); /* C */  /* Avoid leaking the critical section. */
-	this_cpu_dec(sp->per_cpu_ref->c[idx]);
+	this_cpu_inc(sp->per_cpu_ref->unlock_count[idx]);
 }
 EXPORT_SYMBOL_GPL(__srcu_read_unlock);
 

@@ -349,7 +315,7 @@ static bool try_check_zero(struct srcu_struct *sp, int idx, int trycount)
 
 /*
  * Increment the ->completed counter so that future SRCU readers will
- * use the other rank of the ->c[] and ->seq[] arrays.  This allows
+ * use the other rank of the ->(un)lock_count[] arrays.  This allows
  * us to wait for pre-existing readers in a starvation-free manner.
  */
 static void srcu_flip(struct srcu_struct *sp)

Signature · 1 line

2.9.0

Quoted 16 lines · Lance Roy

SRCU uses two per-cpu counters: a nesting counter to count the number of active critical sections, and a sequence counter to ensure that the nesting counters don't change while they are being added together in srcu_readers_active_idx_check().

This patch instead uses per-cpu lock and unlock counters. Because both counters only increase and srcu_readers_active_idx_check() reads the unlock counter before the lock counter, this achieves the same end without having to increment two different counters in srcu_read_lock(). This also saves a smp_mb() in srcu_readers_active_idx_check().

Possible bug: There is no guarantee that the lock counter won't overflow during srcu_readers_active_idx_check(), as there are no memory barriers around srcu_flip() (see comment in srcu_readers_active_idx_check() for details). However, this problem was already present before this patch.

Quoted 5 lines

Suggested-by Mathieu Desnoyers <[email protected]>

Signed-off-by Lance Roy <[email protected]>

Cc Paul E. McKenney <[email protected]>

Cc Lai Jiangshan <[email protected]>

Cc Peter Zijlstra <[email protected]>

OK, this only has differences only in the comments, so I can reasonably substitute it, even this near the next merge window.

But let's talk about the potential overflow. If I understand correctly, for this to happen, there needs to be ULONG_MAX/2 or thereabouts srcu_read_lock() calls without matching srcu_read_unlock() calls. Let's focus on 32-bit systems for the moment.

Lockdep allows at most 48 locks held at a given time by any single task, and RCU does not pass in a non-NULL nest_lock -- if you acquire more than that, a lockdep kernel will splat. Each task has at least one 4K page of kernel stack, so there can be at most 1,048,576 tasks (actually quite a bit fewer due to the size of the task_struct and so on). This allows at most 50,331,648 unmatched srcu_read_lock() calls in the system, which is not sufficient to cause overflow.

Or am I missing something here?

Thanx, Paul

Quoted 39 lines · On Mon, 23 Jan 2017 16:42:52 -0800 "Paul E. McKenney"…

Quoted 16 lines · Lance Roy

SRCU uses two per-cpu counters: a nesting counter to count the number of active critical sections, and a sequence counter to ensure that the nesting counters don't change while they are being added together in srcu_readers_active_idx_check().

This patch instead uses per-cpu lock and unlock counters. Because both counters only increase and srcu_readers_active_idx_check() reads the unlock counter before the lock counter, this achieves the same end without having to increment two different counters in srcu_read_lock(). This also saves a smp_mb() in srcu_readers_active_idx_check().

Possible bug: There is no guarantee that the lock counter won't overflow during srcu_readers_active_idx_check(), as there are no memory barriers around srcu_flip() (see comment in srcu_readers_active_idx_check() for details). However, this problem was already present before this patch.

Quoted 5 lines

Suggested-by Mathieu Desnoyers <[email protected]>

Signed-off-by Lance Roy <[email protected]>

Cc Paul E. McKenney <[email protected]>

Cc Lai Jiangshan <[email protected]>

Cc Peter Zijlstra <[email protected]>

OK, this only has differences only in the comments, so I can reasonably substitute it, even this near the next merge window.

But let's talk about the potential overflow. If I understand correctly, for this to happen, there needs to be ULONG_MAX/2 or thereabouts srcu_read_lock() calls without matching srcu_read_unlock() calls. Let's focus on 32-bit systems for the moment.

Lockdep allows at most 48 locks held at a given time by any single task, and RCU does not pass in a non-NULL nest_lock -- if you acquire more than that, a lockdep kernel will splat. Each task has at least one 4K page of kernel stack, so there can be at most 1,048,576 tasks (actually quite a bit fewer due to the size of the task_struct and so on). This allows at most 50,331,648 unmatched srcu_read_lock() calls in the system, which is not sufficient to cause overflow.

Quoted 3 lines

Or am I missing something here?

Thanx, Paul

It isn't the nest that is the problem. Here is a previous email I wrote describing the problem:

The trouble is that disabling preemption is not enough to ensure that there is at most one srcu_read_lock() call per CPU that missed the srcu_flip().

Define a reader to be an SRCU lock+unlock pair. A reader is called active if it has incremented ->lock_count[] but hasn't incremented ->unlock_count[] yet, and completed if it has incremented ->unlock_count[]. I think that we only want to limit the number of active readers and the number of CPUs. In particular, I don't think there should be a limit on the rate of completion of read side critical section.

The goal of srcu_readers_active_idx_check() is to verify that there were zero active readers on the inactive index at some time during its execution. To do this, it totals the unlock counts, executes a memory barrier, totals the lock counts, and takes the difference. This difference counts the readers that are active when srcu_readers_lock_idx() gets to their CPU, plus the readers that completed after srcu_readers_unlock_idx() and before srcu_readers_lock_idx(). If the true (infinite precision) value of the difference is zero, then there were no active readers at some point while srcu_readers_lock_idx() is running. However, the difference is only stored in a long, so there is a potential for overflow if too many readers complete during srcu_readers_active_idx_check().

For example, let's say there are three threads, each running on their own CPU:

int data, flag;
struct srcu_struct *sp = /* ... */;

Thread 0:

	data = 1;
	synchronize_srcu(sp);
	flag = 1;

Thread 1:

	int data_r, flag_r;
	int idx = srcu_read_lock(sp);
	data_r = data;
	flag_r = flag;
	srcu_read_unlock(sp, idx);
	BUG_ON(flag_r == 1 && data_r == 0);

Thread 2:

	while (true) {
		int idx = srcu_read_lock(sp);
		srcu_read_unlock(sp, idx);
	}

Let's take the following execution order. Thread 1 increments
the CPU 1 version of sp->lock_count[0], sets idx to zero, and loads data (0) into data_r. Thread 0 then sets data to be 1, verifies that there are no readers on index 1, and increments sp->completed, but the CPU actually doesn't preform the last operation, putting it off until the next memory barrier. Thread 0 then calls srcu_readers_active_idx_check() on index 0, which runs srcu_readers_unlock_idx() (returning 0). Right after srcu_readers_unlock_idx() completes, thread 2 starts running. Since Thread 0 hasn't actually incremented sp->completed in any way that is visible to thread 2, srcu_read_lock() will still return 0. Thread 2 can then run for ULONG_MAX iterations, setting the CPU 2 version of sp->unlock_count[0] to ULONG_MAX. CPU 0 then finally gets around to incrementing sp->completed, runs its memory barrier, and then reads the lock counts: 1, 0, ULONG_MAX. The total of ULONG_MAX+1 will overflow to 0 and compare equal with earlier unlock count. Thread 0 will then think that the grace period is over and set flag to one. Thread 1 can then read flag (1) into flag_r and run srcu_read_unlock(). The BUG_ON statement will then fail.

Although ULONG_MAX readers completed during srcu_readers_active_idx_check(), there were at most 2 active readers at any time, so this program doesn't run into any limit.

I hope that was clear enough.

Yeah, we did have this same conversation awhile back, didn't we?

Back then, did I think to ask if this could be minimized or even prevented by adding memory barriers appropriately? ;-)

Thanx, Paul

Quoted 69 lines · Lance Roy

The trouble is that disabling preemption is not enough to ensure that there is at most one srcu_read_lock() call per CPU that missed the srcu_flip().

Define a reader to be an SRCU lock+unlock pair. A reader is called active if it has incremented ->lock_count[] but hasn't incremented ->unlock_count[] yet, and completed if it has incremented ->unlock_count[]. I think that we only want to limit the number of active readers and the number of CPUs. In particular, I don't think there should be a limit on the rate of completion of read side critical section.

The goal of srcu_readers_active_idx_check() is to verify that there were zero active readers on the inactive index at some time during its execution. To do this, it totals the unlock counts, executes a memory barrier, totals the lock counts, and takes the difference. This difference counts the readers that are active when srcu_readers_lock_idx() gets to their CPU, plus the readers that completed after srcu_readers_unlock_idx() and before srcu_readers_lock_idx(). If the true (infinite precision) value of the difference is zero, then there were no active readers at some point while srcu_readers_lock_idx() is running. However, the difference is only stored in a long, so there is a potential for overflow if too many readers complete during srcu_readers_active_idx_check().

For example, let's say there are three threads, each running on their own CPU:

int data, flag;
struct srcu_struct *sp = /* ... */;

Thread 0:

	data = 1;
	synchronize_srcu(sp);
	flag = 1;

Thread 1:

	int data_r, flag_r;
	int idx = srcu_read_lock(sp);
	data_r = data;
	flag_r = flag;
	srcu_read_unlock(sp, idx);
	BUG_ON(flag_r == 1 && data_r == 0);

Thread 2:

	while (true) {
		int idx = srcu_read_lock(sp);
		srcu_read_unlock(sp, idx);
	}

Let's take the following execution order. Thread 1 increments the CPU 1 version of sp->lock_count[0], sets idx to zero, and loads data (0) into data_r. Thread 0 then sets data to be 1, verifies that there are no readers on index 1, and increments sp->completed, but the CPU actually doesn't preform the last operation, putting it off until the next memory barrier. Thread 0 then calls srcu_readers_active_idx_check() on index 0, which runs srcu_readers_unlock_idx() (returning 0). Right after srcu_readers_unlock_idx() completes, thread 2 starts running. Since Thread 0 hasn't actually incremented sp->completed in any way that is visible to thread 2, srcu_read_lock() will still return 0. Thread 2 can then run for ULONG_MAX iterations, setting the CPU 2 version of sp->unlock_count[0] to ULONG_MAX. CPU 0 then finally gets around to incrementing sp->completed, runs its memory barrier, and then reads the lock counts: 1, 0, ULONG_MAX. The total of ULONG_MAX+1 will overflow to 0 and compare equal with earlier unlock count. Thread 0 will then think that the grace period is over and set flag to one. Thread 1 can then read flag (1) into flag_r and run srcu_read_unlock(). The BUG_ON statement will then fail.

Although ULONG_MAX readers completed during srcu_readers_active_idx_check(), there were at most 2 active readers at any time, so this program doesn't run into any limit.

Quoted 1 line

I hope that was clear enough.

Indeed it is!

So adding a full memory barrier immediately after the srcu_flip() should prevent this, because if the updater failed to see an unlock increment, the second following lock for that CPU/task would be guaranteed to see the flip. Or am I still missing something?

Is there a sequence of events that requires a full memory barrier before the srcu_flip()?

Quoted 5 lines

Yeah, we did have this same conversation awhile back, didn't we?

Back then, did I think to ask if this could be minimized or even prevented by adding memory barriers appropriately? ;-)

Quoted 1 line

Thanx, Paul

Yes, it can be fixed by adding a memory barrier after incrementing ->completed inside srcu_flip(). The upper limit on NR_CPUS turns out to be more complicated than this, as it needs to deal with highly nested read side critical sections mixed with the critical section loops, but only the one memory barrier should be necessary.

If a process invokes synchronize_srcu(), is delayed just the right amount of time, and thus does not sleep when waiting for the grace period to complete, there is no ordering between the end of the grace period and the code following the synchronize_srcu(). Similarly, there can be a lack of ordering between the end of the SRCU grace period and callback invocation.

This commit adds the necessary ordering.

Reported-by Lance Roy <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

---
 include/linux/rcupdate.h | 12 ++++++++++++
 kernel/rcu/srcu.c        |  5 +++++
 kernel/rcu/tree.h        | 12 ------------
 3 files changed, 17 insertions(+), 12 deletions(-)

include/linux/rcupdate.h +8 -0

index 01f71e1d2e94..6ade6a52d9d4 100644
@@ -1161,5 +1161,17 @@ do { \
 		ftrace_dump(oops_dump_mode); \
 } while (0)
 
+/*
+ * Place this after a lock-acquisition primitive to guarantee that
+ * an UNLOCK+LOCK pair acts as a full barrier.  This guarantee applies
+ * if the UNLOCK and LOCK are executed by the same CPU or if the
+ * UNLOCK and LOCK operate on the same lock variable.
+ */
+#ifdef CONFIG_PPC
+#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for

lock. */ +#else /* #ifdef CONFIG_PPC */

include/linux/rcupdate.h +3 -0

+#define smp_mb__after_unlock_lock()	do { } while (0)
+#endif /* #else #ifdef CONFIG_PPC */
+
 
 #endif /* __LINUX_RCUPDATE_H */

kernel/rcu/srcu.c +2 -0

index ddabf5fbf562..f2abfbae258c 100644
@@ -359,6 +359,7 @@ void call_srcu(struct srcu_struct *sp, struct rcu_head
 *head, head->next = NULL;
 	head->func = func;
 	spin_lock_irqsave(&sp->queue_lock, flags);
+	smp_mb__after_unlock_lock(); /* Caller's prior accesses before GP. */
 	rcu_batch_queue(&sp->batch_queue, head);
 	if (!sp->running) {
 		sp->running = true;

@@ -392,6 +393,7 @@ static void __synchronize_srcu(struct srcu_struct *sp,
 int trycount) head->next = NULL;
 	head->func = wakeme_after_rcu;
 	spin_lock_irq(&sp->queue_lock);
+	smp_mb__after_unlock_lock(); /* Caller's prior accesses before GP. */
 	if (!sp->running) {
 		/* steal the processing owner */
 		sp->running = true;

kernel/rcu/srcu.c +0 -0

 	for (i = 0; i < SRCU_CALLBACK_BATCH; i++) {
 		head = rcu_batch_dequeue(&sp->batch_done);
 		if (!head)

kernel/rcu/tree.h +0 -7

index fe98dd24adf8..abcc25bdcb29 100644
@@ -688,18 +688,6 @@ static inline void rcu_nocb_q_lengths(struct rcu_data
 *rdp, long *ql, long *qll) #endif /* #ifdef CONFIG_RCU_TRACE */
 
 /*
- * Place this after a lock-acquisition primitive to guarantee that
- * an UNLOCK+LOCK pair act as a full barrier.  This guarantee applies
- * if the UNLOCK and LOCK are executed by the same CPU or if the
- * UNLOCK and LOCK operate on the same lock variable.
- */
-#ifdef CONFIG_PPC
-#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for

lock. */ -#else /* #ifdef CONFIG_PPC */

Quoted 60 lines · On Sun, 15 Jan 2017 14:42:34 -0800 "Paul E. McKenney"…

If a process invokes synchronize_srcu(), is delayed just the right amount of time, and thus does not sleep when waiting for the grace period to complete, there is no ordering between the end of the grace period and the code following the synchronize_srcu(). Similarly, there can be a lack of ordering between the end of the SRCU grace period and callback invocation.

This commit adds the necessary ordering.

Reported-by Lance Roy <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

---
 include/linux/rcupdate.h | 12 ++++++++++++
 kernel/rcu/srcu.c        |  5 +++++
 kernel/rcu/tree.h        | 12 ------------
 3 files changed, 17 insertions(+), 12 deletions(-)

include/linux/rcupdate.h +8 -0

index 01f71e1d2e94..6ade6a52d9d4 100644
@@ -1161,5 +1161,17 @@ do { \
 		ftrace_dump(oops_dump_mode); \
 } while (0)
 
+/*
+ * Place this after a lock-acquisition primitive to guarantee that
+ * an UNLOCK+LOCK pair acts as a full barrier.  This guarantee applies
+ * if the UNLOCK and LOCK are executed by the same CPU or if the
+ * UNLOCK and LOCK operate on the same lock variable.
+ */
+#ifdef CONFIG_PPC
+#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for

lock. */ +#else /* #ifdef CONFIG_PPC */

include/linux/rcupdate.h +3 -0

+#define smp_mb__after_unlock_lock()	do { } while (0)
+#endif /* #else #ifdef CONFIG_PPC */
+
 
 #endif /* __LINUX_RCUPDATE_H */

kernel/rcu/srcu.c +2 -0

index ddabf5fbf562..f2abfbae258c 100644
@@ -359,6 +359,7 @@ void call_srcu(struct srcu_struct *sp, struct rcu_head
 *head, head->next = NULL;
 	head->func = func;
 	spin_lock_irqsave(&sp->queue_lock, flags);
+	smp_mb__after_unlock_lock(); /* Caller's prior accesses before GP. */
 	rcu_batch_queue(&sp->batch_queue, head);
 	if (!sp->running) {
 		sp->running = true;

@@ -392,6 +393,7 @@ static void __synchronize_srcu(struct srcu_struct *sp,
 int trycount) head->next = NULL;
 	head->func = wakeme_after_rcu;
 	spin_lock_irq(&sp->queue_lock);
+	smp_mb__after_unlock_lock(); /* Caller's prior accesses before GP. */
 	if (!sp->running) {
 		/* steal the processing owner */
 		sp->running = true;

@@ -413,6 +415,8 @@ static void __synchronize_srcu(struct srcu_struct *sp,

kernel/rcu/srcu.c static void __synchronize_srcu(struct srcu_struct *sp, int trycount) · line 419

Quoted 5 lines

kernel/rcu/srcu.c +2 -0

 int trycount) 
 	if (!done)
 		wait_for_completion(&rcu.completion);
+
+	smp_mb(); /* Caller's later accesses after GP. */

I think that this memory barrier is only necessary when done == false, as otherwise srcu_advance_batches() should provide sufficient memory ordering.

kernel/rcu/srcu.c static void srcu_invoke_callbacks(struct srcu_struct *sp) · line 594

Quoted 8 lines

kernel/rcu/srcu.c +1 -0

 }
 
 /**

@@ -587,6 +591,7 @@ static void srcu_invoke_callbacks(struct srcu_struct *sp)
 	int i;
 	struct rcu_head *head;
 
+	smp_mb(); /* Callback accesses after GP. */

Shouldn't srcu_advance_batches() have already run all necessary memory barriers?

Quoted 10 lines

kernel/rcu/srcu.c +0 -0

 	for (i = 0; i < SRCU_CALLBACK_BATCH; i++) {
 		head = rcu_batch_dequeue(&sp->batch_done);
 		if (!head)

kernel/rcu/tree.h +0 -0

index fe98dd24adf8..abcc25bdcb29 100644
@@ -688,18 +688,6 @@ static inline void rcu_nocb_q_lengths(struct rcu_data
 *rdp, long *ql, long *qll) #endif /* #ifdef CONFIG_RCU_TRACE */
 

Quoted 9 lines

kernel/rcu/tree.h +0 -7

 /*
- * Place this after a lock-acquisition primitive to guarantee that
- * an UNLOCK+LOCK pair act as a full barrier.  This guarantee applies
- * if the UNLOCK and LOCK are executed by the same CPU or if the
- * UNLOCK and LOCK operate on the same lock variable.
- */
-#ifdef CONFIG_PPC
-#define smp_mb__after_unlock_lock()	smp_mb()  /* Full ordering for

lock. */ -#else /* #ifdef CONFIG_PPC */

kernel/rcu/tree.h static inline void rcu_nocb_q_lengths(struct rcu_data *rdp, long *ql, long *qll) · line 693

Quoted 7 lines

kernel/rcu/tree.h +0 -4

-#define smp_mb__after_unlock_lock()	do { } while (0)
-#endif /* #else #ifdef CONFIG_PPC */
-
-/*
  * Wrappers for the rcu_node::lock acquire and release.
  *
  * Because the rcu_nodes form a tree, the tree traversal locking will observe

Thanks,
Lance

Quoted 13 lines · Lance Roy

kernel/rcu/srcu.c static void __synchronize_srcu(struct srcu_struct *sp, · line 419

Quoted 6 lines · On Sun, 15 Jan 2017 14:42:34 -0800 "Paul E. McKenney"…

kernel/rcu/srcu.c +2 -0

@@ -413,6 +415,8 @@ static void __synchronize_srcu(struct srcu_struct *sp,
 int trycount) 
 	if (!done)
 		wait_for_completion(&rcu.completion);
+
+	smp_mb(); /* Caller's later accesses after GP. */  

I think that this memory barrier is only necessary when done == false, as otherwise srcu_advance_batches() should provide sufficient memory ordering.

Let me make sure that I understand your rationale here.

The idea is that although srcu_readers_active_idx_check() did a full memory barrier, this might have happened on some other CPU, which would not have provided ordering to the current CPU in the race case where current CPU didn't actually sleep. (This can happen where the current task is preempted, and then is resumed just as the grace period completes.)

Or are you concerned about some other sequence of events?

(I have moved the smp_mb() inside the "if (!done)" in the meantime.)

Quoted 8 lines

kernel/rcu/srcu.c static void srcu_invoke_callbacks(struct srcu_struct · line 594

Quoted 5 lines

kernel/rcu/srcu.c +1 -0

@@ -587,6 +591,7 @@ static void srcu_invoke_callbacks(struct srcu_struct
 *sp) int i;
 	struct rcu_head *head;
 
+	smp_mb(); /* Callback accesses after GP. */  

Shouldn't srcu_advance_batches() have already run all necessary memory barriers?

It does look that way:

o process_srcu() is the only thing that invokes
srcu_invoke_callbacks().

o process_srcu() invokes srcu_advance_batches() immediately before
srcu_invoke_callbacks(), so any memory barriers invoked from srcu_advance_batches() affect process_srcu() (unlike the earlier example where srcu_advance_batches() might be executed in the context of some other task).

o srcu_advance_batches() unconditionally invokes try_check_zero(),
which in turn unconditionally invokes srcu_readers_active_idx_check(), which in turn invokes smp_mb().

This smp_mb() precedes a successful check that all pre-existing readers are done, otherwise srcu_advance_batches() won't have returned (or won't have advanced the callbacks, which in turn will prevent them from being invoked).

I have removed this memory barrier and added a comment.

And thank you for your review and comments!!!

From: Lance Roy <[email protected]>

SRCU uses two per-cpu counters: a nesting counter to count the number of active critical sections, and a sequence counter to ensure that the nesting counters don't change while they are being added together in srcu_readers_active_idx_check().

This patch instead uses per-cpu lock and unlock counters. Because both counters only increase and srcu_readers_active_idx_check() reads the unlock counter before the lock counter, this achieves the same end without having to increment two different counters in srcu_read_lock(). This also saves a smp_mb() in srcu_readers_active_idx_check().

A possible problem with this patch is that it can only handle ULONG_MAX - NR_CPUS simultaneous readers, whereas the old version could handle up to ULONG_MAX.

Suggested-by Mathieu Desnoyers <[email protected]>

Signed-off-by Lance Roy <[email protected]>

Signed-off-by Paul E. McKenney <[email protected]>

Cc Lai Jiangshan <[email protected]>

Cc Peter Zijlstra <[email protected]>

---
 include/linux/srcu.h    |  10 ++--
 kernel/rcu/rcutorture.c |  19 +++++++-
 kernel/rcu/srcu.c       | 123

++++++++++++++++++------------------------------ 3 files changed, 67 insertions(+), 85 deletions(-)

include/linux/srcu.h +5 -5

index dc8eb63c6568..a598cf3ac70c 100644
@@ -33,9 +33,9 @@
 #include <linux/rcupdate.h>
 #include <linux/workqueue.h>
 
-struct srcu_struct_array {
-	unsigned long c[2];
-	unsigned long seq[2];
+struct srcu_array {
+	unsigned long lock_count[2];
+	unsigned long unlock_count[2];
 };
 
 struct rcu_batch {

@@ -46,7 +46,7 @@ struct rcu_batch {
 
 struct srcu_struct {
 	unsigned long completed;
-	struct srcu_struct_array __percpu *per_cpu_ref;
+	struct srcu_array __percpu *per_cpu_ref;
 	spinlock_t queue_lock; /* protect ->batch_queue, ->running */
 	bool running;
 	/* callbacks just queued */

@@ -118,7 +118,7 @@ void process_srcu(struct work_struct *work);
  * See include/linux/percpu-defs.h for the rules on per-CPU variables.
  */
 #define __DEFINE_SRCU(name,
 is_static)					\
-	static DEFINE_PER_CPU(struct srcu_struct_array, name##_srcu_array);\
+	static DEFINE_PER_CPU(struct srcu_array, name##_srcu_array);\
 	is_static struct srcu_struct name = __SRCU_STRUCT_INIT(name)
 #define DEFINE_SRCU(name)		__DEFINE_SRCU(name, /* not static

*/) #define DEFINE_STATIC_SRCU(name) __DEFINE_SRCU(name, static)

kernel/rcu/rcutorture.c +8 -2

index 87c51225ceec..d81345be730e 100644
@@ -564,10 +564,25 @@ static void srcu_torture_stats(void)
 	pr_alert("%s%s per-CPU(idx=%d):",
 		 torture_type, TORTURE_FLAG, idx);
 	for_each_possible_cpu(cpu) {
+		unsigned long l0, l1;
+		unsigned long u0, u1;
 		long c0, c1;
+		struct srcu_array *counts =
 per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu);
-		c0 = (long)per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu)->c[!idx];
-		c1 = (long)per_cpu_ptr(srcu_ctlp->per_cpu_ref, cpu)->c[idx];
+		u0 = counts->unlock_count[!idx];
+		u1 = counts->unlock_count[idx];
+
+		/*
+		 * Make sure that a lock is always counted if the

corresponding

kernel/rcu/rcutorture.c +9 -0

+		 * unlock is counted.
+		 */
+		smp_rmb();
+
+		l0 = counts->lock_count[!idx];
+		l1 = counts->lock_count[idx];
+
+		c0 = l0 - u0;
+		c1 = l1 - u1;
 		pr_cont(" %d(%ld,%ld)", cpu, c0, c1);
 	}
 	pr_cont("\n");

kernel/rcu/srcu.c +15 -24

index 9b9cdd549caa..ddabf5fbf562 100644
@@ -106,7 +106,7 @@ static int init_srcu_struct_fields(struct srcu_struct *sp)
 	rcu_batch_init(&sp->batch_check1);
 	rcu_batch_init(&sp->batch_done);
 	INIT_DELAYED_WORK(&sp->work, process_srcu);
-	sp->per_cpu_ref = alloc_percpu(struct srcu_struct_array);
+	sp->per_cpu_ref = alloc_percpu(struct srcu_array);
 	return sp->per_cpu_ref ? 0 : -ENOMEM;
 }
 

@@ -141,114 +141,78 @@ EXPORT_SYMBOL_GPL(init_srcu_struct);
 #endif /* #else #ifdef CONFIG_DEBUG_LOCK_ALLOC */
 
 /*
- * Returns approximate total of the readers' ->seq[] values for the
+ * Returns approximate total of the readers' ->lock_count[] values for the
  * rank of per-CPU counters specified by idx.
  */
-static unsigned long srcu_readers_seq_idx(struct srcu_struct *sp, int idx)
+static unsigned long srcu_readers_lock_idx(struct srcu_struct *sp, int idx)
 {
 	int cpu;
 	unsigned long sum = 0;
-	unsigned long t;
 
 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->seq[idx]);
-		sum += t;
+		struct srcu_array *cpuc = per_cpu_ptr(sp->per_cpu_ref, cpu);
+
+		sum += READ_ONCE(cpuc->lock_count[idx]);
 	}
 	return sum;
 }
 
 /*
- * Returns approximate number of readers active on the specified rank
- * of the per-CPU ->c[] counters.
+ * Returns approximate total of the readers' ->unlock_count[] values for the
+ * rank of per-CPU counters specified by idx.
  */
-static unsigned long srcu_readers_active_idx(struct srcu_struct *sp, int idx)
+static unsigned long srcu_readers_unlock_idx(struct srcu_struct *sp, int idx)
 {
 	int cpu;
 	unsigned long sum = 0;
-	unsigned long t;
 
 	for_each_possible_cpu(cpu) {
-		t = READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[idx]);
-		sum += t;
+		struct srcu_array *cpuc = per_cpu_ptr(sp->per_cpu_ref, cpu);
+
+		sum += READ_ONCE(cpuc->unlock_count[idx]);
 	}
 	return sum;
 }
 
 /*
  * Return true if the number of pre-existing readers is determined to
- * be stably zero.  An example unstable zero can occur if the call
- * to srcu_readers_active_idx() misses an __srcu_read_lock() increment,
- * but due to task migration, sees the corresponding __srcu_read_unlock()
- * decrement.  This can happen because srcu_readers_active_idx() takes
- * time to sum the array, and might in fact be interrupted or preempted
- * partway through the summation.
+ * be zero.
  */
 static bool srcu_readers_active_idx_check(struct srcu_struct *sp, int idx)
 {
-	unsigned long seq;
+	unsigned long unlocks;
 
-	seq = srcu_readers_seq_idx(sp, idx);
+	unlocks = srcu_readers_unlock_idx(sp, idx);
 
 	/*
-	 * The following smp_mb() A pairs with the smp_mb() B located in
-	 * __srcu_read_lock().  This pairing ensures that if an
-	 * __srcu_read_lock() increments its counter after the summation
-	 * in srcu_readers_active_idx(), then the corresponding SRCU

read-side

kernel/rcu/srcu.c +1 -2

-	 * critical section will see any changes made prior to the start
-	 * of the current SRCU grace period.
+	 * Make sure that a lock is always counted if the corresponding

unlock

kernel/rcu/srcu.c +2 -0

+	 * is counted. Needs to be a smp_mb() as the read side may contain a
+	 * read from a variable that is written to before the

synchronize_srcu()

kernel/rcu/srcu.c +1 -0

+	 * in the write side. In this case smp_mb()s A and B act like the

store

kernel/rcu/srcu.c +2 -3

+	 * buffering pattern.
 	 *
-	 * Also, if the above call to srcu_readers_seq_idx() saw the
-	 * increment of ->seq[], then the call to srcu_readers_active_idx()
-	 * must see the increment of ->c[].
+	 * This smp_mb() also pairs with smp_mb() C to prevent writes after

the

kernel/rcu/srcu.c +1 -0

+	 * synchronize_srcu() from being executed before the grace period

ends. */

kernel/rcu/srcu.c +2 -12

 	smp_mb(); /* A */
 
 	/*
-	 * Note that srcu_readers_active_idx() can incorrectly return
-	 * zero even though there is a pre-existing reader throughout.
-	 * To see this, suppose that task A is in a very long SRCU
-	 * read-side critical section that started on CPU 0, and that
-	 * no other reader exists, so that the sum of the counters
-	 * is equal to one.  Then suppose that task B starts executing
-	 * srcu_readers_active_idx(), summing up to CPU 1, and then that
-	 * task C starts reading on CPU 0, so that its increment is not
-	 * summed, but finishes reading on CPU 2, so that its decrement
-	 * -is- summed.  Then when task B completes its sum, it will
-	 * incorrectly get zero, despite the fact that task A has been
-	 * in its SRCU read-side critical section the whole time.
+	 * If the locks are the same as the unlocks, then there must have
+	 * been no readers on this index at some time in between. This does

not

kernel/rcu/srcu.c +3 -2

+	 * mean that there are no more readers, as one could have read the
+	 * current index but not have incremented the lock counter yet.
 	 *
-	 * We therefore do a validation step should srcu_readers_active_idx()
-	 * return zero.
+	 * Note that there can be at most NR_CPUS worth of readers using the

old

kernel/rcu/srcu.c +2 -0

+	 * index that haven't incremented ->lock_count[] yet.  Therefore, the
+	 * sum of the ->lock_count[]s cannot increment enough times to

overflow

kernel/rcu/srcu.c +1 -0

+	 * and end up equal the sum of the ->unlock_count[]s, as long as

there

kernel/rcu/srcu.c +1 -0

+	 * are at most ULONG_MAX - NR_CPUS readers at a time.  (Yes, this

does

kernel/rcu/srcu.c +2 -0

+	 * mean that systems having more than a billion or so CPUs need to be
+	 * 64-bit systems.)  Therefore, the only way that the return values

of

kernel/rcu/srcu.c +1 -0

+	 * the two calls to srcu_readers_(un)lock_idx() can be equal is if

there

kernel/rcu/srcu.c +11 -35

+	 * are no active readers using this index.
 	 */
-	if (srcu_readers_active_idx(sp, idx) != 0)
-		return false;
-
-	/*
-	 * The remainder of this function is the validation step.
-	 * The following smp_mb() D pairs with the smp_mb() C in
-	 * __srcu_read_unlock().  If the __srcu_read_unlock() was seen
-	 * by srcu_readers_active_idx() above, then any destructive
-	 * operation performed after the grace period will happen after
-	 * the corresponding SRCU read-side critical section.
-	 *
-	 * Note that there can be at most NR_CPUS worth of readers using
-	 * the old index, which is not enough to overflow even a 32-bit
-	 * integer.  (Yes, this does mean that systems having more than
-	 * a billion or so CPUs need to be 64-bit systems.)  Therefore,
-	 * the sum of the ->seq[] counters cannot possibly overflow.
-	 * Therefore, the only way that the return values of the two
-	 * calls to srcu_readers_seq_idx() can be equal is if there were
-	 * no increments of the corresponding rank of ->seq[] counts
-	 * in the interim.  But the missed-increment scenario laid out
-	 * above includes an increment of the ->seq[] counter by
-	 * the corresponding __srcu_read_lock().  Therefore, if this
-	 * scenario occurs, the return values from the two calls to
-	 * srcu_readers_seq_idx() will differ, and thus the validation
-	 * step below suffices.
-	 */
-	smp_mb(); /* D */
-
-	return srcu_readers_seq_idx(sp, idx) == seq;
+	return srcu_readers_lock_idx(sp, idx) == unlocks;
 }
 
 /**

@@ -266,8 +230,12 @@ static bool srcu_readers_active(struct srcu_struct *sp)
 	unsigned long sum = 0;
 
 	for_each_possible_cpu(cpu) {
-		sum += READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[0]);
-		sum += READ_ONCE(per_cpu_ptr(sp->per_cpu_ref, cpu)->c[1]);
+		struct srcu_array *cpuc = per_cpu_ptr(sp->per_cpu_ref, cpu);
+
+		sum += READ_ONCE(cpuc->lock_count[0]);
+		sum += READ_ONCE(cpuc->lock_count[1]);
+		sum -= READ_ONCE(cpuc->unlock_count[0]);
+		sum -= READ_ONCE(cpuc->unlock_count[1]);
 	}
 	return sum;
 }

@@ -298,9 +266,8 @@ int __srcu_read_lock(struct srcu_struct *sp)
 	int idx;
 
 	idx = READ_ONCE(sp->completed) & 0x1;
-	__this_cpu_inc(sp->per_cpu_ref->c[idx]);
+	__this_cpu_inc(sp->per_cpu_ref->lock_count[idx]);
 	smp_mb(); /* B */  /* Avoid leaking the critical section. */
-	__this_cpu_inc(sp->per_cpu_ref->seq[idx]);
 	return idx;
 }
 EXPORT_SYMBOL_GPL(__srcu_read_lock);

@@ -314,7 +281,7 @@ EXPORT_SYMBOL_GPL(__srcu_read_lock);
 void __srcu_read_unlock(struct srcu_struct *sp, int idx)
 {
 	smp_mb(); /* C */  /* Avoid leaking the critical section. */
-	this_cpu_dec(sp->per_cpu_ref->c[idx]);
+	this_cpu_inc(sp->per_cpu_ref->unlock_count[idx]);
 }
 EXPORT_SYMBOL_GPL(__srcu_read_unlock);
 

@@ -349,7 +316,7 @@ static bool try_check_zero(struct srcu_struct *sp, int
 idx, int trycount)
 /*
  * Increment the ->completed counter so that future SRCU readers will
- * use the other rank of the ->c[] and ->seq[] arrays.  This allows
+ * use the other rank of the ->(un)lock_count[] arrays.  This allows
  * us to wait for pre-existing readers in a starvation-free manner.
  */
 static void srcu_flip(struct srcu_struct *sp)