[PATCH] RCU - cpu-offline-cleanup [1/3]

[PATCH] RCU - cpu-offline-cleanup [1/3]

[Dipankar Sarma]

Andrew,

There is a series of patches in my tree and these 3 are the first
ones that should probably be merged down the road. Descriptions are on 
top of the patches. Please include them in -mm.

A lot of RCU code will be cleaned up later in order to support
call_rcu_bh(), the separate RCU interface that considers softirq
handler completion a quiescent state.

Thanks
Dipankar



Signed-off-by: Dipankar Sarma <dipankar@in.ibm.com>

Minor cleanup of the hotplug code to remove #ifdef in cpu
event notifier handler. If CONFIG_HOTPLUG_CPU is not defined,
CPU_DEAD case will be optimized off.


 kernel/rcupdate.c |    8 ++++++--
 1 files changed, 6 insertions(+), 2 deletions(-)

diff -puN kernel/rcupdate.c~cpu-offline-cleanup kernel/rcupdate.c
--- linux-2.6.8-rc2-rcu/kernel/rcupdate.c~cpu-offline-cleanup	2004-07-29 13:35:18.000000000 +0530
+++ linux-2.6.8-rc2-rcu-dipankar/kernel/rcupdate.c	2004-07-29 13:41:18.000000000 +0530
@@ -242,6 +242,12 @@ unlock:
 	tasklet_kill_immediate(&RCU_tasklet(cpu), cpu);
 }
 
+#else
+
+static void rcu_offline_cpu(int cpu)
+{
+}
+
 #endif
 
 void rcu_restart_cpu(int cpu)
@@ -325,11 +331,9 @@ static int __devinit rcu_cpu_notify(stru
 	case CPU_UP_PREPARE:
 		rcu_online_cpu(cpu);
 		break;
-#ifdef CONFIG_HOTPLUG_CPU
 	case CPU_DEAD:
 		rcu_offline_cpu(cpu);
 		break;
-#endif
 	default:
 		break;
 	}

_

Re: [PATCH] RCU - rcu-cpu-offline-fix [2/3]

[Dipankar Sarma]

Andrew,

This patch replaces fixes-for-rcu_offline_cpu-rcu_move_batch-268-rc2.patch
in -mm tree. Same thing, but I needed to add the tail pointers anyway
for the RCU lists. That allows callbacks lists to be added to the
tail without iterating, so an optimized version of Nathan's patch.
I have sanity tested cpu hotplug on a ppc64 box with this patch.

Thanks
Dipankar

This fixes the RCU cpu offline code which was broken by singly-linked
RCU changes. Nathan pointed out the problems and submitted a patch
for this. This is an optimal fix - no need to iterate through
the list of callbacks, just use the tail pointers and attach the
list from the dead cpu.

Signed-off-by: Nathan Lynch <nathanl@austin.ibm.com>
Signed-off-by: Dipankar Sarma <dipankar@in.ibm.com>



 include/linux/rcupdate.h |    2 ++
 kernel/rcupdate.c        |   22 +++++++++++-----------
 2 files changed, 13 insertions(+), 11 deletions(-)

diff -puN kernel/rcupdate.c~rcu-cpu-offline-fix kernel/rcupdate.c
--- linux-2.6.8-rc2-rcu/kernel/rcupdate.c~rcu-cpu-offline-fix	2004-07-29 16:06:03.000000000 +0530
+++ linux-2.6.8-rc2-rcu-dipankar/kernel/rcupdate.c	2004-07-31 12:25:57.000000000 +0530
@@ -210,17 +210,15 @@ static void rcu_check_quiescent_state(vo
  * locking requirements, the list it's pulling from has to belong to a cpu
  * which is dead and hence not processing interrupts.
  */
-static void rcu_move_batch(struct list_head *list)
+static void rcu_move_batch(struct rcu_head *list, struct rcu_head **tail)
 {
-	struct list_head *entry;
-	int cpu = smp_processor_id();
+	int cpu;
 
 	local_irq_disable();
-	while (!list_empty(list)) {
-		entry = list->next;
-		list_del(entry);
-		list_add_tail(entry, &RCU_nxtlist(cpu));
-	}
+	cpu = smp_processor_id();
+	*RCU_nxttail(cpu) = list;
+	if (list)
+		RCU_nxttail(cpu) = tail;
 	local_irq_enable();
 }
 
@@ -233,11 +231,10 @@ static void rcu_offline_cpu(int cpu)
 	spin_lock_bh(&rcu_state.mutex);
 	if (rcu_ctrlblk.cur != rcu_ctrlblk.completed)
 		cpu_quiet(cpu);
-unlock:
 	spin_unlock_bh(&rcu_state.mutex);
 
-	rcu_move_batch(&RCU_curlist(cpu));
-	rcu_move_batch(&RCU_nxtlist(cpu));
+	rcu_move_batch(RCU_curlist(cpu), RCU_curtail(cpu));
+	rcu_move_batch(RCU_nxtlist(cpu), RCU_nxttail(cpu));
 
 	tasklet_kill_immediate(&RCU_tasklet(cpu), cpu);
 }
@@ -270,6 +267,7 @@ static void rcu_process_callbacks(unsign
 	    !rcu_batch_before(rcu_ctrlblk.completed, RCU_batch(cpu))) {
 		rcu_list = RCU_curlist(cpu);
 		RCU_curlist(cpu) = NULL;
+		RCU_curtail(cpu) = &RCU_curlist(cpu);
 	}
 
 	local_irq_disable();
@@ -277,6 +275,7 @@ static void rcu_process_callbacks(unsign
 		int next_pending, seq;
 
 		RCU_curlist(cpu) = RCU_nxtlist(cpu);
+		RCU_curtail(cpu) = RCU_nxttail(cpu);
 		RCU_nxtlist(cpu) = NULL;
 		RCU_nxttail(cpu) = &RCU_nxtlist(cpu);
 		local_irq_enable();
@@ -318,6 +317,7 @@ static void __devinit rcu_online_cpu(int
 {
 	memset(&per_cpu(rcu_data, cpu), 0, sizeof(struct rcu_data));
 	tasklet_init(&RCU_tasklet(cpu), rcu_process_callbacks, 0UL);
+	RCU_curtail(cpu) = &RCU_curlist(cpu);
 	RCU_nxttail(cpu) = &RCU_nxtlist(cpu);
 	RCU_quiescbatch(cpu) = rcu_ctrlblk.completed;
 	RCU_qs_pending(cpu) = 0;
diff -puN include/linux/rcupdate.h~rcu-cpu-offline-fix include/linux/rcupdate.h
--- linux-2.6.8-rc2-rcu/include/linux/rcupdate.h~rcu-cpu-offline-fix	2004-07-29 16:12:51.000000000 +0530
+++ linux-2.6.8-rc2-rcu-dipankar/include/linux/rcupdate.h	2004-07-31 12:25:44.000000000 +0530
@@ -98,6 +98,7 @@ struct rcu_data {
         struct rcu_head *nxtlist;
 	struct rcu_head **nxttail;
         struct rcu_head *curlist;
+        struct rcu_head **curtail;
 };
 
 DECLARE_PER_CPU(struct rcu_data, rcu_data);
@@ -111,6 +112,7 @@ extern struct rcu_ctrlblk rcu_ctrlblk;
 #define RCU_nxtlist(cpu) 	(per_cpu(rcu_data, (cpu)).nxtlist)
 #define RCU_curlist(cpu) 	(per_cpu(rcu_data, (cpu)).curlist)
 #define RCU_nxttail(cpu) 	(per_cpu(rcu_data, (cpu)).nxttail)
+#define RCU_curtail(cpu) 	(per_cpu(rcu_data, (cpu)).curtail)
 
 static inline int rcu_pending(int cpu) 
 {

_

Re: [PATCH] RCU - low-latency-rcu [3/3]

[Dipankar Sarma]

As promised in OLS, this patch introduces a throttling mechanism
for RCU callbacks. We should not be executing thousands of callbacks
from a single tasklet handler, that is not scheduler friendly.
I introduced a kernel paramenter (rcupdate.maxbatch) which
allows us to control how many callbacks per tasklet handler
are invoked. I have sanity tested this on both UP and SMP
x86 box with dcache heavy workload, there doesn't seem to be
any problem, leak or otherwise.

Thanks
Dipankar

This patch makes RCU callbacks friendly to scheduler. It helps
low latency by limiting the number of callbacks invoked per
tasklet handler. Since we cannot schedule during a single softirq
handler, this reduces size of non-preemptible section significantly,
specially under heavy RCU updates. The limiting is done
through a kernel parameter rcupdate.maxbatch which is the
maximum number of RCU callbacks to invoke during a single
tasklet handler.

Signed-off-by: Dipankar Sarma <dipankar@in.ibm.com>


 include/linux/rcupdate.h |    7 +++++++
 kernel/rcupdate.c        |   27 +++++++++++++++++++--------
 2 files changed, 26 insertions(+), 8 deletions(-)

diff -puN kernel/rcupdate.c~low-latency-rcu kernel/rcupdate.c
--- linux-2.6.8-rc2-rcu/kernel/rcupdate.c~low-latency-rcu	2004-07-31 12:26:19.000000000 +0530
+++ linux-2.6.8-rc2-rcu-dipankar/kernel/rcupdate.c	2004-07-31 12:26:19.000000000 +0530
@@ -40,6 +40,7 @@
 #include <asm/bitops.h>
 #include <linux/module.h>
 #include <linux/completion.h>
+#include <linux/moduleparam.h>
 #include <linux/percpu.h>
 #include <linux/notifier.h>
 #include <linux/rcupdate.h>
@@ -63,6 +64,7 @@ DEFINE_PER_CPU(struct rcu_data, rcu_data
 /* Fake initialization required by compiler */
 static DEFINE_PER_CPU(struct tasklet_struct, rcu_tasklet) = {NULL};
 #define RCU_tasklet(cpu) (per_cpu(rcu_tasklet, cpu))
+static int maxbatch = 10;
 
 /**
  * call_rcu - Queue an RCU update request.
@@ -93,15 +95,23 @@ void fastcall call_rcu(struct rcu_head *
  * Invoke the completed RCU callbacks. They are expected to be in
  * a per-cpu list.
  */
-static void rcu_do_batch(struct rcu_head *list)
+static void rcu_do_batch(int cpu)
 {
-	struct rcu_head *next;
+	struct rcu_head *next, *list;
+	int count = 0;
 
+	list = RCU_donelist(cpu);
 	while (list) {
-		next = list->next;
+		next = RCU_donelist(cpu) = list->next;
 		list->func(list);
 		list = next;
+		if (++count >= maxbatch)
+			break;
 	}
+	if (!RCU_donelist(cpu))
+		RCU_donetail(cpu) = &RCU_donelist(cpu);
+	else
+		tasklet_schedule(&RCU_tasklet(cpu));
 }
 
 /*
@@ -261,11 +271,11 @@ void rcu_restart_cpu(int cpu)
 static void rcu_process_callbacks(unsigned long unused)
 {
 	int cpu = smp_processor_id();
-	struct rcu_head *rcu_list = NULL;
 
 	if (RCU_curlist(cpu) &&
 	    !rcu_batch_before(rcu_ctrlblk.completed, RCU_batch(cpu))) {
-		rcu_list = RCU_curlist(cpu);
+		*RCU_donetail(cpu) = RCU_curlist(cpu);
+		RCU_donetail(cpu) = RCU_curtail(cpu);
 		RCU_curlist(cpu) = NULL;
 		RCU_curtail(cpu) = &RCU_curlist(cpu);
 	}
@@ -300,8 +310,8 @@ static void rcu_process_callbacks(unsign
 		local_irq_enable();
 	}
 	rcu_check_quiescent_state();
-	if (rcu_list)
-		rcu_do_batch(rcu_list);
+	if (RCU_donelist(cpu))
+		rcu_do_batch(cpu);
 }
 
 void rcu_check_callbacks(int cpu, int user)
@@ -319,6 +329,7 @@ static void __devinit rcu_online_cpu(int
 	tasklet_init(&RCU_tasklet(cpu), rcu_process_callbacks, 0UL);
 	RCU_curtail(cpu) = &RCU_curlist(cpu);
 	RCU_nxttail(cpu) = &RCU_nxtlist(cpu);
+	RCU_donetail(cpu) = &RCU_donelist(cpu);
 	RCU_quiescbatch(cpu) = rcu_ctrlblk.completed;
 	RCU_qs_pending(cpu) = 0;
 }
@@ -388,6 +399,6 @@ void synchronize_kernel(void)
 	wait_for_completion(&rcu.completion);
 }
 
-
+module_param(maxbatch, int, 0);
 EXPORT_SYMBOL(call_rcu);
 EXPORT_SYMBOL(synchronize_kernel);
diff -puN include/linux/rcupdate.h~low-latency-rcu include/linux/rcupdate.h
--- linux-2.6.8-rc2-rcu/include/linux/rcupdate.h~low-latency-rcu	2004-07-31 12:26:19.000000000 +0530
+++ linux-2.6.8-rc2-rcu-dipankar/include/linux/rcupdate.h	2004-07-31 12:26:19.000000000 +0530
@@ -99,6 +99,8 @@ struct rcu_data {
 	struct rcu_head **nxttail;
         struct rcu_head *curlist;
         struct rcu_head **curtail;
+        struct rcu_head *donelist;
+        struct rcu_head **donetail;
 };
 
 DECLARE_PER_CPU(struct rcu_data, rcu_data);
@@ -113,6 +115,8 @@ extern struct rcu_ctrlblk rcu_ctrlblk;
 #define RCU_curlist(cpu) 	(per_cpu(rcu_data, (cpu)).curlist)
 #define RCU_nxttail(cpu) 	(per_cpu(rcu_data, (cpu)).nxttail)
 #define RCU_curtail(cpu) 	(per_cpu(rcu_data, (cpu)).curtail)
+#define RCU_donelist(cpu) 	(per_cpu(rcu_data, (cpu)).donelist)
+#define RCU_donetail(cpu) 	(per_cpu(rcu_data, (cpu)).donetail)
 
 static inline int rcu_pending(int cpu) 
 {
@@ -127,6 +131,9 @@ static inline int rcu_pending(int cpu) 
 	if (!RCU_curlist(cpu) && RCU_nxtlist(cpu))
 		return 1;
 
+	if (RCU_donelist(cpu))
+		return 1;
+
 	/* The rcu core waits for a quiescent state from the cpu */
 	if (RCU_quiescbatch(cpu) != rcu_ctrlblk.cur || RCU_qs_pending(cpu))
 		return 1;

_