[PATCH] rcu: eliminate rcu_ctrlblk.lock

rcu_ctrlblk.lock is used to read the ->cur and ->next_pending
atomically in __rcu_process_callbacks(). It can be replaced
by a couple of memory barriers.

rcu_start_batch:
	rcp->next_pending = 0;
	smp_wmb();
	rcp->cur++;

__rcu_process_callbacks:
	rdp->batch = rcp->cur + 1;
	smp_rmb();
	if (!rcp->next_pending)
		rcu_start_batch(rcp, rsp, 1);

This way, if __rcu_process_callbacks() sees incremented ->cur value,
it must also see that ->next_pending == 0 (or rcu_start_batch() is
already in progress on another cpu).
Signed-off-by: Oleg Nesterov <oleg@tv-sign.ru>
Signed-off-by: Andrew Morton <akpm@osdl.org>
Signed-off-by: Linus Torvalds <torvalds@osdl.org>

include/linux/rcupdate.h

@@ -65,7 +65,6 @@ struct rcu_ctrlblk {
 	long	cur;		/* Current batch number.                      */
 	long	completed;	/* Number of the last completed batch         */
 	int	next_pending;	/* Is the next batch already waiting?         */
-	seqcount_t lock;	/* For atomic reads of cur and next_pending.  */
 } ____cacheline_maxaligned_in_smp;
 
 /* Is batch a before batch b ? */

kernel/rcupdate.c

@@ -49,9 +49,9 @@
 
 /* Definition for rcupdate control block. */
 struct rcu_ctrlblk rcu_ctrlblk = 
-	{ .cur = -300, .completed = -300 , .lock = SEQCNT_ZERO };
+	{ .cur = -300, .completed = -300 };
 struct rcu_ctrlblk rcu_bh_ctrlblk =
-	{ .cur = -300, .completed = -300 , .lock = SEQCNT_ZERO };
+	{ .cur = -300, .completed = -300 };
 
 /* Bookkeeping of the progress of the grace period */
 struct rcu_state {
@@ -185,10 +185,13 @@ static void rcu_start_batch(struct rcu_ctrlblk *rcp, struct rcu_state *rsp,
 			rcp->completed == rcp->cur) {
 		/* Can't change, since spin lock held. */
 		cpus_andnot(rsp->cpumask, cpu_online_map, nohz_cpu_mask);
-		write_seqcount_begin(&rcp->lock);
+
 		rcp->next_pending = 0;
+		/* next_pending == 0 must be visible in __rcu_process_callbacks()
+		 * before it can see new value of cur.
+		 */
+		smp_wmb();
 		rcp->cur++;
-		write_seqcount_end(&rcp->lock);
 	}
 }
 
@@ -319,8 +322,6 @@ static void __rcu_process_callbacks(struct rcu_ctrlblk *rcp,
 
 	local_irq_disable();
 	if (rdp->nxtlist && !rdp->curlist) {
-		int next_pending, seq;
-
 		rdp->curlist = rdp->nxtlist;
 		rdp->curtail = rdp->nxttail;
 		rdp->nxtlist = NULL;
@@ -330,14 +331,15 @@ static void __rcu_process_callbacks(struct rcu_ctrlblk *rcp,
 		/*
 		 * start the next batch of callbacks
 		 */
-		do {
-			seq = read_seqcount_begin(&rcp->lock);
-			/* determine batch number */
-			rdp->batch = rcp->cur + 1;
-			next_pending = rcp->next_pending;
-		} while (read_seqcount_retry(&rcp->lock, seq));
-
-		if (!next_pending) {
+
+		/* determine batch number */
+		rdp->batch = rcp->cur + 1;
+		/* see the comment and corresponding wmb() in
+		 * the rcu_start_batch()
+		 */
+		smp_rmb();
+
+		if (!rcp->next_pending) {
 			/* and start it/schedule start if it's a new batch */
 			spin_lock(&rsp->lock);
 			rcu_start_batch(rcp, rsp, 1);