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

PATCH In review Dipankar Sarma 1 participant 3 messages Jul 31, 2004 RCU

lore

Share & tools

lore

Dipankar Sarma

author

2004-07-31 08:57 UTC #0 lore raw

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

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 <[email protected]>

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(-)

kernel/rcupdate.c renamed from kernel/rcupdate.c~cpu-offline-cleanup +6 -2

diff -puN kernel/rcupdate.c~cpu-offline-cleanup kernel/rcupdate.c
@@ -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;
 	}

Dipankar Sarma

author

2004-07-31 09:04 UTC #1 lore raw

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

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 <[email protected]>

Signed-off-by Dipankar Sarma <[email protected]>

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

kernel/rcupdate.c renamed from kernel/rcupdate.c~rcu-cpu-offline-fix +11 ~~-11~~

diff -puN kernel/rcupdate.c~rcu-cpu-offline-fix kernel/rcupdate.c
@@ -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;

include/linux/rcupdate.h renamed from include/linux/rcupdate.h~rcu-cpu-offline-fix +2 -0

diff -puN include/linux/rcupdate.h~rcu-cpu-offline-fix include/linux/rcupdate.h
@@ -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) 
 {

Dipankar Sarma

author

2004-07-31 09:08 UTC #2 lore raw

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

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 <[email protected]>

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

kernel/rcupdate.c renamed from kernel/rcupdate.c~low-latency-rcu +19 -8

diff -puN kernel/rcupdate.c~low-latency-rcu kernel/rcupdate.c
@@ -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);

include/linux/rcupdate.h renamed from include/linux/rcupdate.h~low-latency-rcu +7 -0

diff -puN include/linux/rcupdate.h~low-latency-rcu include/linux/rcupdate.h
@@ -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;