[people/pmueller/ipfire-2.x.git] / src / patches / suse-2.6.27.31 / patches.arch / s390-05-05-stp-etr-stop-machine.patch

From: Gerald Schaefer <geraldsc@de.ibm.com>
Subject: stp/etr: smp_call_function races.
References: bnc#450096

Symptom:     System hangs when setting stp/etr online.
Problem:     The stp/etr code uses smp_call_function() with blocking
             function on the signalled cpus. This is prone to cause
             deadlocks.
Solution:    Replace the smp_call_function() synchronization with one
             based on stop_machine_run.

Acked-by: John Jolly <jjolly@suse.de>
---
 arch/s390/kernel/time.c |  210 +++++++++++++++++++++++++++++-------------------
 1 file changed, 130 insertions(+), 80 deletions(-)

Index: linux-2.6.27/arch/s390/kernel/time.c
===================================================================
--- linux-2.6.27.orig/arch/s390/kernel/time.c
+++ linux-2.6.27/arch/s390/kernel/time.c
@@ -22,6 +22,8 @@
 #include <linux/string.h>
 #include <linux/mm.h>
 #include <linux/interrupt.h>
+#include <linux/cpu.h>
+#include <linux/stop_machine.h>
 #include <linux/time.h>
 #include <linux/sysdev.h>
 #include <linux/delay.h>
@@ -365,6 +367,15 @@ static void enable_sync_clock(void)
 	atomic_set_mask(0x80000000, sw_ptr);
 }
 
+/* Single threaded workqueue used for etr and stp sync events */
+static struct workqueue_struct *time_sync_wq;
+
+static void __init time_init_wq(void)
+{
+	if (!time_sync_wq)
+		time_sync_wq = create_singlethread_workqueue("timesync");
+}
+
 /*
  * External Time Reference (ETR) code.
  */
@@ -457,17 +468,18 @@ static int __init etr_init(void)
 
 	if (!test_bit(CLOCK_SYNC_HAS_ETR, &clock_sync_flags))
 		return 0;
+	time_init_wq();
 	/* Check if this machine has the steai instruction. */
 	if (etr_steai(&aib, ETR_STEAI_STEPPING_PORT) == 0)
 		etr_steai_available = 1;
 	setup_timer(&etr_timer, etr_timeout, 0UL);
 	if (etr_port0_online) {
 		set_bit(ETR_EVENT_PORT0_CHANGE, &etr_events);
-		schedule_work(&etr_work);
+		queue_work(time_sync_wq, &etr_work);
 	}
 	if (etr_port1_online) {
 		set_bit(ETR_EVENT_PORT1_CHANGE, &etr_events);
-		schedule_work(&etr_work);
+		queue_work(time_sync_wq, &etr_work);
 	}
 	return 0;
 }
@@ -494,7 +506,7 @@ void etr_switch_to_local(void)
 	if (test_bit(CLOCK_SYNC_ETR, &clock_sync_flags))
 		disable_sync_clock(NULL);
 	set_bit(ETR_EVENT_SWITCH_LOCAL, &etr_events);
-	schedule_work(&etr_work);
+	queue_work(time_sync_wq, &etr_work);
 }
 
 /*
@@ -510,7 +522,7 @@ void etr_sync_check(void)
 	if (test_bit(CLOCK_SYNC_ETR, &clock_sync_flags))
 		disable_sync_clock(NULL);
 	set_bit(ETR_EVENT_SYNC_CHECK, &etr_events);
-	schedule_work(&etr_work);
+	queue_work(time_sync_wq, &etr_work);
 }
 
 /*
@@ -534,13 +546,13 @@ static void etr_timing_alert(struct etr_
 		 * Both ports are not up-to-date now.
 		 */
 		set_bit(ETR_EVENT_PORT_ALERT, &etr_events);
-	schedule_work(&etr_work);
+	queue_work(time_sync_wq, &etr_work);
 }
 
 static void etr_timeout(unsigned long dummy)
 {
 	set_bit(ETR_EVENT_UPDATE, &etr_events);
-	schedule_work(&etr_work);
+	queue_work(time_sync_wq, &etr_work);
 }
 
 /*
@@ -647,14 +659,16 @@ static int etr_aib_follows(struct etr_ai
 }
 
 struct clock_sync_data {
+	atomic_t cpus;
 	int in_sync;
 	unsigned long long fixup_cc;
+	int etr_port;
+	struct etr_aib *etr_aib;
 };
 
-static void clock_sync_cpu_start(void *dummy)
+static void clock_sync_cpu(struct clock_sync_data *sync)
 {
-	struct clock_sync_data *sync = dummy;
-
+	atomic_dec(&sync->cpus);
 	enable_sync_clock();
 	/*
 	 * This looks like a busy wait loop but it isn't. etr_sync_cpus
@@ -680,39 +694,35 @@ static void clock_sync_cpu_start(void *d
 	fixup_clock_comparator(sync->fixup_cc);
 }
 
-static void clock_sync_cpu_end(void *dummy)
-{
-}
-
 /*
  * Sync the TOD clock using the port refered to by aibp. This port
  * has to be enabled and the other port has to be disabled. The
  * last eacr update has to be more than 1.6 seconds in the past.
  */
-static int etr_sync_clock(struct etr_aib *aib, int port)
+static int etr_sync_clock(void *data)
 {
-	struct etr_aib *sync_port;
-	struct clock_sync_data etr_sync;
+	static int first;
 	unsigned long long clock, old_clock, delay, delta;
-	int follows;
+	struct clock_sync_data *etr_sync;
+	struct etr_aib *sync_port, *aib;
+	int port;
 	int rc;
 
-	/* Check if the current aib is adjacent to the sync port aib. */
-	sync_port = (port == 0) ? &etr_port0 : &etr_port1;
-	follows = etr_aib_follows(sync_port, aib, port);
-	memcpy(sync_port, aib, sizeof(*aib));
-	if (!follows)
-		return -EAGAIN;
+	etr_sync = data;
 
-	/*
-	 * Catch all other cpus and make them wait until we have
-	 * successfully synced the clock. smp_call_function will
-	 * return after all other cpus are in etr_sync_cpu_start.
-	 */
-	memset(&etr_sync, 0, sizeof(etr_sync));
-	preempt_disable();
-	smp_call_function(clock_sync_cpu_start, &etr_sync, 0);
-	local_irq_disable();
+	if (xchg(&first, 1) == 1) {
+		/* Slave */
+		clock_sync_cpu(etr_sync);
+		return 0;
+	}
+
+	/* Wait until all other cpus entered the sync function. */
+	while (atomic_read(&etr_sync->cpus) != 0)
+		cpu_relax();
+
+	port = etr_sync->etr_port;
+	aib = etr_sync->etr_aib;
+	sync_port = (port == 0) ? &etr_port0 : &etr_port1;
 	enable_sync_clock();
 
 	/* Set clock to next OTE. */
@@ -729,16 +739,16 @@ static int etr_sync_clock(struct etr_aib
 		delay = (unsigned long long)
 			(aib->edf2.etv - sync_port->edf2.etv) << 32;
 		delta = adjust_time(old_clock, clock, delay);
-		etr_sync.fixup_cc = delta;
+		etr_sync->fixup_cc = delta;
 		fixup_clock_comparator(delta);
 		/* Verify that the clock is properly set. */
 		if (!etr_aib_follows(sync_port, aib, port)) {
 			/* Didn't work. */
 			disable_sync_clock(NULL);
-			etr_sync.in_sync = -EAGAIN;
+			etr_sync->in_sync = -EAGAIN;
 			rc = -EAGAIN;
 		} else {
-			etr_sync.in_sync = 1;
+			etr_sync->in_sync = 1;
 			rc = 0;
 		}
 	} else {
@@ -746,12 +756,33 @@ static int etr_sync_clock(struct etr_aib
 		__ctl_clear_bit(0, 29);
 		__ctl_clear_bit(14, 21);
 		disable_sync_clock(NULL);
-		etr_sync.in_sync = -EAGAIN;
+		etr_sync->in_sync = -EAGAIN;
 		rc = -EAGAIN;
 	}
-	local_irq_enable();
-	smp_call_function(clock_sync_cpu_end, NULL, 0);
-	preempt_enable();
+	xchg(&first, 0);
+	return rc;
+}
+
+static int etr_sync_clock_stop(struct etr_aib *aib, int port)
+{
+	struct clock_sync_data etr_sync;
+	struct etr_aib *sync_port;
+	int follows;
+	int rc;
+
+	/* Check if the current aib is adjacent to the sync port aib. */
+	sync_port = (port == 0) ? &etr_port0 : &etr_port1;
+	follows = etr_aib_follows(sync_port, aib, port);
+	memcpy(sync_port, aib, sizeof(*aib));
+	if (!follows)
+		return -EAGAIN;
+	memset(&etr_sync, 0, sizeof(etr_sync));
+	etr_sync.etr_aib = aib;
+	etr_sync.etr_port = port;
+	get_online_cpus();
+	atomic_set(&etr_sync.cpus, num_online_cpus() - 1);
+	rc = stop_machine(etr_sync_clock, &etr_sync, &cpu_online_map);
+	put_online_cpus();
 	return rc;
 }
 
@@ -908,7 +939,7 @@ static void etr_update_eacr(struct etr_e
 }
 
 /*
- * ETR tasklet. In this function you'll find the main logic. In
+ * ETR work. In this function you'll find the main logic. In
  * particular this is the only function that calls etr_update_eacr(),
  * it "controls" the etr control register.
  */
@@ -1041,7 +1072,7 @@ static void etr_work_fn(struct work_stru
 	etr_update_eacr(eacr);
 	set_bit(CLOCK_SYNC_ETR, &clock_sync_flags);
 	if (now < etr_tolec + (1600000 << 12) ||
-	    etr_sync_clock(&aib, sync_port) != 0) {
+	    etr_sync_clock_stop(&aib, sync_port) != 0) {
 		/* Sync failed. Try again in 1/2 second. */
 		eacr.es = 0;
 		etr_update_eacr(eacr);
@@ -1130,13 +1161,13 @@ static ssize_t etr_online_store(struct s
 			return count;	/* Nothing to do. */
 		etr_port0_online = value;
 		set_bit(ETR_EVENT_PORT0_CHANGE, &etr_events);
-		schedule_work(&etr_work);
+		queue_work(time_sync_wq, &etr_work);
 	} else {
 		if (etr_port1_online == value)
 			return count;	/* Nothing to do. */
 		etr_port1_online = value;
 		set_bit(ETR_EVENT_PORT1_CHANGE, &etr_events);
-		schedule_work(&etr_work);
+		queue_work(time_sync_wq, &etr_work);
 	}
 	return count;
 }
@@ -1371,8 +1402,12 @@ static void __init stp_reset(void)
 
 static int __init stp_init(void)
 {
-	if (test_bit(CLOCK_SYNC_HAS_STP, &clock_sync_flags) && stp_online)
-		schedule_work(&stp_work);
+	if (!test_bit(CLOCK_SYNC_HAS_STP, &clock_sync_flags))
+		return 0;
+	time_init_wq();
+	if (!stp_online)
+		return 0;
+	queue_work(time_sync_wq, &stp_work);
 	return 0;
 }
 
@@ -1389,7 +1424,7 @@ arch_initcall(stp_init);
 static void stp_timing_alert(struct stp_irq_parm *intparm)
 {
 	if (intparm->tsc || intparm->lac || intparm->tcpc)
-		schedule_work(&stp_work);
+		queue_work(time_sync_wq, &stp_work);
 }
 
 /*
@@ -1417,46 +1452,34 @@ void stp_island_check(void)
 	if (!test_bit(CLOCK_SYNC_STP, &clock_sync_flags))
 		return;
 	disable_sync_clock(NULL);
-	schedule_work(&stp_work);
+	queue_work(time_sync_wq, &stp_work);
 }
 
-/*
- * STP tasklet. Check for the STP state and take over the clock
- * synchronization if the STP clock source is usable.
- */
-static void stp_work_fn(struct work_struct *work)
+
+static int stp_sync_clock(void *data)
 {
-	struct clock_sync_data stp_sync;
+	static int first;
 	unsigned long long old_clock, delta;
+	struct clock_sync_data *stp_sync;
 	int rc;
 
-	if (!stp_online) {
-		chsc_sstpc(stp_page, STP_OP_CTRL, 0x0000);
-		return;
-	}
+	stp_sync = data;
 
-	rc = chsc_sstpc(stp_page, STP_OP_CTRL, 0xb0e0);
-	if (rc)
-		return;
+	if (xchg(&first, 1) == 1) {
+		/* Slave */
+		clock_sync_cpu(stp_sync);
+		return 0;
+	}
 
-	rc = chsc_sstpi(stp_page, &stp_info, sizeof(struct stp_sstpi));
-	if (rc || stp_info.c == 0)
-		return;
+	/* Wait until all other cpus entered the sync function. */
+	while (atomic_read(&stp_sync->cpus) != 0)
+		cpu_relax();
 
-	/*
-	 * Catch all other cpus and make them wait until we have
-	 * successfully synced the clock. smp_call_function will
-	 * return after all other cpus are in clock_sync_cpu_start.
-	 */
-	memset(&stp_sync, 0, sizeof(stp_sync));
-	preempt_disable();
-	smp_call_function(clock_sync_cpu_start, &stp_sync, 0);
-	local_irq_disable();
 	enable_sync_clock();
 
 	set_bit(CLOCK_SYNC_STP, &clock_sync_flags);
 	if (test_and_clear_bit(CLOCK_SYNC_ETR, &clock_sync_flags))
-		schedule_work(&etr_work);
+		queue_work(time_sync_wq, &etr_work);
 
 	rc = 0;
 	if (stp_info.todoff[0] || stp_info.todoff[1] ||
@@ -1475,16 +1498,43 @@ static void stp_work_fn(struct work_stru
 	}
 	if (rc) {
 		disable_sync_clock(NULL);
-		stp_sync.in_sync = -EAGAIN;
+		stp_sync->in_sync = -EAGAIN;
 		clear_bit(CLOCK_SYNC_STP, &clock_sync_flags);
 		if (etr_port0_online || etr_port1_online)
-			schedule_work(&etr_work);
+			queue_work(time_sync_wq, &etr_work);
 	} else
-		stp_sync.in_sync = 1;
+		stp_sync->in_sync = 1;
+	xchg(&first, 0);
+	return 0;
+}
 
-	local_irq_enable();
-	smp_call_function(clock_sync_cpu_end, NULL, 0);
-	preempt_enable();
+/*
+ * STP work. Check for the STP state and take over the clock
+ * synchronization if the STP clock source is usable.
+ */
+static void stp_work_fn(struct work_struct *work)
+{
+	struct clock_sync_data stp_sync;
+	int rc;
+
+	if (!stp_online) {
+		chsc_sstpc(stp_page, STP_OP_CTRL, 0x0000);
+		return;
+	}
+
+	rc = chsc_sstpc(stp_page, STP_OP_CTRL, 0xb0e0);
+	if (rc)
+		return;
+
+	rc = chsc_sstpi(stp_page, &stp_info, sizeof(struct stp_sstpi));
+	if (rc || stp_info.c == 0)
+		return;
+
+	memset(&stp_sync, 0, sizeof(stp_sync));
+	get_online_cpus();
+	atomic_set(&stp_sync.cpus, num_online_cpus() - 1);
+	stop_machine(stp_sync_clock, &stp_sync, &cpu_online_map);
+	put_online_cpus();
 }
 
 /*
@@ -1593,7 +1643,7 @@ static ssize_t stp_online_store(struct s
 	if (!test_bit(CLOCK_SYNC_HAS_STP, &clock_sync_flags))
 		return -EOPNOTSUPP;
 	stp_online = value;
-	schedule_work(&stp_work);
+	queue_work(time_sync_wq, &stp_work);
 	return count;
 }
Commit	Line	Data
00e5a55c BS	1	From: Gerald Schaefer <geraldsc@de.ibm.com>
	2	Subject: stp/etr: smp_call_function races.
	3	References: bnc#450096
	4
	5	Symptom: System hangs when setting stp/etr online.
	6	Problem: The stp/etr code uses smp_call_function() with blocking
	7	function on the signalled cpus. This is prone to cause
	8	deadlocks.
	9	Solution: Replace the smp_call_function() synchronization with one
	10	based on stop_machine_run.
	11
	12	Acked-by: John Jolly <jjolly@suse.de>
	13	---
	14	arch/s390/kernel/time.c \| 210 +++++++++++++++++++++++++++++-------------------
	15	1 file changed, 130 insertions(+), 80 deletions(-)
	16
	17	Index: linux-2.6.27/arch/s390/kernel/time.c
	18	===================================================================
	19	--- linux-2.6.27.orig/arch/s390/kernel/time.c
	20	+++ linux-2.6.27/arch/s390/kernel/time.c
	21	@@ -22,6 +22,8 @@
	22	#include <linux/string.h>
	23	#include <linux/mm.h>
	24	#include <linux/interrupt.h>
	25	+#include <linux/cpu.h>
	26	+#include <linux/stop_machine.h>
	27	#include <linux/time.h>
	28	#include <linux/sysdev.h>
	29	#include <linux/delay.h>
	30	@@ -365,6 +367,15 @@ static void enable_sync_clock(void)
	31	atomic_set_mask(0x80000000, sw_ptr);
	32	}
	33
	34	+/* Single threaded workqueue used for etr and stp sync events */
	35	+static struct workqueue_struct *time_sync_wq;
	36	+
	37	+static void __init time_init_wq(void)
	38	+{
	39	+ if (!time_sync_wq)
	40	+ time_sync_wq = create_singlethread_workqueue("timesync");
	41	+}
	42	+
	43	/*
	44	* External Time Reference (ETR) code.
	45	*/
	46	@@ -457,17 +468,18 @@ static int __init etr_init(void)
	47
	48	if (!test_bit(CLOCK_SYNC_HAS_ETR, &clock_sync_flags))
	49	return 0;
	50	+ time_init_wq();
	51	/* Check if this machine has the steai instruction. */
	52	if (etr_steai(&aib, ETR_STEAI_STEPPING_PORT) == 0)
	53	etr_steai_available = 1;
	54	setup_timer(&etr_timer, etr_timeout, 0UL);
	55	if (etr_port0_online) {
	56	set_bit(ETR_EVENT_PORT0_CHANGE, &etr_events);
	57	- schedule_work(&etr_work);
	58	+ queue_work(time_sync_wq, &etr_work);
	59	}
	60	if (etr_port1_online) {
	61	set_bit(ETR_EVENT_PORT1_CHANGE, &etr_events);
	62	- schedule_work(&etr_work);
	63	+ queue_work(time_sync_wq, &etr_work);
	64	}
65	return 0;
66	}
67	@@ -494,7 +506,7 @@ void etr_switch_to_local(void)
68	if (test_bit(CLOCK_SYNC_ETR, &clock_sync_flags))
69	disable_sync_clock(NULL);
70	set_bit(ETR_EVENT_SWITCH_LOCAL, &etr_events);
71	- schedule_work(&etr_work);
72	+ queue_work(time_sync_wq, &etr_work);
73	}
74
75	/*
76	@@ -510,7 +522,7 @@ void etr_sync_check(void)
77	if (test_bit(CLOCK_SYNC_ETR, &clock_sync_flags))
78	disable_sync_clock(NULL);
79	set_bit(ETR_EVENT_SYNC_CHECK, &etr_events);
80	- schedule_work(&etr_work);
81	+ queue_work(time_sync_wq, &etr_work);
82	}
83
84	/*
85	@@ -534,13 +546,13 @@ static void etr_timing_alert(struct etr_
86	* Both ports are not up-to-date now.
87	*/
88	set_bit(ETR_EVENT_PORT_ALERT, &etr_events);
89	- schedule_work(&etr_work);
90	+ queue_work(time_sync_wq, &etr_work);
91	}
92
93	static void etr_timeout(unsigned long dummy)
94	{
95	set_bit(ETR_EVENT_UPDATE, &etr_events);
96	- schedule_work(&etr_work);
97	+ queue_work(time_sync_wq, &etr_work);
98	}
99
100	/*
101	@@ -647,14 +659,16 @@ static int etr_aib_follows(struct etr_ai
102	}
103
104	struct clock_sync_data {
105	+ atomic_t cpus;
106	int in_sync;
107	unsigned long long fixup_cc;
108	+ int etr_port;
109	+ struct etr_aib *etr_aib;
110	};
111
112	-static void clock_sync_cpu_start(void *dummy)
113	+static void clock_sync_cpu(struct clock_sync_data *sync)
114	{
115	- struct clock_sync_data *sync = dummy;
116	-
117	+ atomic_dec(&sync->cpus);
118	enable_sync_clock();
119	/*
120	* This looks like a busy wait loop but it isn't. etr_sync_cpus
121	@@ -680,39 +694,35 @@ static void clock_sync_cpu_start(void *d
122	fixup_clock_comparator(sync->fixup_cc);
123	}
124
125	-static void clock_sync_cpu_end(void *dummy)
126	-{
127	-}
128	-
129	/*
130	* Sync the TOD clock using the port refered to by aibp. This port
131	* has to be enabled and the other port has to be disabled. The
132	* last eacr update has to be more than 1.6 seconds in the past.
133	*/
134	-static int etr_sync_clock(struct etr_aib *aib, int port)
135	+static int etr_sync_clock(void *data)
136	{
137	- struct etr_aib *sync_port;
138	- struct clock_sync_data etr_sync;
139	+ static int first;
140	unsigned long long clock, old_clock, delay, delta;
141	- int follows;
142	+ struct clock_sync_data *etr_sync;
143	+ struct etr_aib sync_port, aib;
144	+ int port;
145	int rc;
146
147	- /* Check if the current aib is adjacent to the sync port aib. */
148	- sync_port = (port == 0) ? &etr_port0 : &etr_port1;
149	- follows = etr_aib_follows(sync_port, aib, port);
150	- memcpy(sync_port, aib, sizeof(*aib));
151	- if (!follows)
152	- return -EAGAIN;
153	+ etr_sync = data;
154
155	- /*
156	- * Catch all other cpus and make them wait until we have
157	- * successfully synced the clock. smp_call_function will
158	- * return after all other cpus are in etr_sync_cpu_start.
159	- */
160	- memset(&etr_sync, 0, sizeof(etr_sync));
161	- preempt_disable();
162	- smp_call_function(clock_sync_cpu_start, &etr_sync, 0);
163	- local_irq_disable();
164	+ if (xchg(&first, 1) == 1) {
165	+ /* Slave */
166	+ clock_sync_cpu(etr_sync);
167	+ return 0;
168	+ }
169	+
170	+ /* Wait until all other cpus entered the sync function. */
171	+ while (atomic_read(&etr_sync->cpus) != 0)
172	+ cpu_relax();
173	+
174	+ port = etr_sync->etr_port;
175	+ aib = etr_sync->etr_aib;
176	+ sync_port = (port == 0) ? &etr_port0 : &etr_port1;
177	enable_sync_clock();
178
179	/* Set clock to next OTE. */
180	@@ -729,16 +739,16 @@ static int etr_sync_clock(struct etr_aib
181	delay = (unsigned long long)
182	(aib->edf2.etv - sync_port->edf2.etv) << 32;
183	delta = adjust_time(old_clock, clock, delay);
184	- etr_sync.fixup_cc = delta;
185	+ etr_sync->fixup_cc = delta;
186	fixup_clock_comparator(delta);
187	/* Verify that the clock is properly set. */
188	if (!etr_aib_follows(sync_port, aib, port)) {
189	/* Didn't work. */
190	disable_sync_clock(NULL);
191	- etr_sync.in_sync = -EAGAIN;
192	+ etr_sync->in_sync = -EAGAIN;
193	rc = -EAGAIN;
194	} else {
195	- etr_sync.in_sync = 1;
196	+ etr_sync->in_sync = 1;
197	rc = 0;
198	}
199	} else {
200	@@ -746,12 +756,33 @@ static int etr_sync_clock(struct etr_aib
201	__ctl_clear_bit(0, 29);
202	__ctl_clear_bit(14, 21);
203	disable_sync_clock(NULL);
204	- etr_sync.in_sync = -EAGAIN;
205	+ etr_sync->in_sync = -EAGAIN;
206	rc = -EAGAIN;
207	}
208	- local_irq_enable();
209	- smp_call_function(clock_sync_cpu_end, NULL, 0);
210	- preempt_enable();
211	+ xchg(&first, 0);
212	+ return rc;
213	+}
214	+
215	+static int etr_sync_clock_stop(struct etr_aib *aib, int port)
216	+{
217	+ struct clock_sync_data etr_sync;
218	+ struct etr_aib *sync_port;
219	+ int follows;
220	+ int rc;
221	+
222	+ /* Check if the current aib is adjacent to the sync port aib. */
223	+ sync_port = (port == 0) ? &etr_port0 : &etr_port1;
224	+ follows = etr_aib_follows(sync_port, aib, port);
225	+ memcpy(sync_port, aib, sizeof(*aib));
226	+ if (!follows)
227	+ return -EAGAIN;
228	+ memset(&etr_sync, 0, sizeof(etr_sync));
229	+ etr_sync.etr_aib = aib;
230	+ etr_sync.etr_port = port;
231	+ get_online_cpus();
232	+ atomic_set(&etr_sync.cpus, num_online_cpus() - 1);
233	+ rc = stop_machine(etr_sync_clock, &etr_sync, &cpu_online_map);
234	+ put_online_cpus();
235	return rc;
236	}
237
238	@@ -908,7 +939,7 @@ static void etr_update_eacr(struct etr_e
239	}
240
241	/*
242	- * ETR tasklet. In this function you'll find the main logic. In
243	+ * ETR work. In this function you'll find the main logic. In
244	* particular this is the only function that calls etr_update_eacr(),
245	* it "controls" the etr control register.
246	*/
247	@@ -1041,7 +1072,7 @@ static void etr_work_fn(struct work_stru
248	etr_update_eacr(eacr);
249	set_bit(CLOCK_SYNC_ETR, &clock_sync_flags);
250	if (now < etr_tolec + (1600000 << 12) \|\|
251	- etr_sync_clock(&aib, sync_port) != 0) {
252	+ etr_sync_clock_stop(&aib, sync_port) != 0) {
253	/* Sync failed. Try again in 1/2 second. */
254	eacr.es = 0;
255	etr_update_eacr(eacr);
256	@@ -1130,13 +1161,13 @@ static ssize_t etr_online_store(struct s
257	return count; /* Nothing to do. */
258	etr_port0_online = value;
259	set_bit(ETR_EVENT_PORT0_CHANGE, &etr_events);
260	- schedule_work(&etr_work);
261	+ queue_work(time_sync_wq, &etr_work);
262	} else {
263	if (etr_port1_online == value)
264	return count; /* Nothing to do. */
265	etr_port1_online = value;
266	set_bit(ETR_EVENT_PORT1_CHANGE, &etr_events);
267	- schedule_work(&etr_work);
268	+ queue_work(time_sync_wq, &etr_work);
269	}
270	return count;
271	}
272	@@ -1371,8 +1402,12 @@ static void __init stp_reset(void)
273
274	static int __init stp_init(void)
275	{
276	- if (test_bit(CLOCK_SYNC_HAS_STP, &clock_sync_flags) && stp_online)
277	- schedule_work(&stp_work);
278	+ if (!test_bit(CLOCK_SYNC_HAS_STP, &clock_sync_flags))
279	+ return 0;
280	+ time_init_wq();
281	+ if (!stp_online)
282	+ return 0;
283	+ queue_work(time_sync_wq, &stp_work);
284	return 0;
285	}
286
287	@@ -1389,7 +1424,7 @@ arch_initcall(stp_init);
288	static void stp_timing_alert(struct stp_irq_parm *intparm)
289	{
290	if (intparm->tsc \|\| intparm->lac \|\| intparm->tcpc)
291	- schedule_work(&stp_work);
292	+ queue_work(time_sync_wq, &stp_work);
293	}
294
295	/*
296	@@ -1417,46 +1452,34 @@ void stp_island_check(void)
297	if (!test_bit(CLOCK_SYNC_STP, &clock_sync_flags))
298	return;
299	disable_sync_clock(NULL);
300	- schedule_work(&stp_work);
301	+ queue_work(time_sync_wq, &stp_work);
302	}
303
304	-/*
305	- * STP tasklet. Check for the STP state and take over the clock
306	- * synchronization if the STP clock source is usable.
307	- */
308	-static void stp_work_fn(struct work_struct *work)
309	+
310	+static int stp_sync_clock(void *data)
311	{
312	- struct clock_sync_data stp_sync;
313	+ static int first;
314	unsigned long long old_clock, delta;
315	+ struct clock_sync_data *stp_sync;
316	int rc;
317
318	- if (!stp_online) {
319	- chsc_sstpc(stp_page, STP_OP_CTRL, 0x0000);
320	- return;
321	- }
322	+ stp_sync = data;
323
324	- rc = chsc_sstpc(stp_page, STP_OP_CTRL, 0xb0e0);
325	- if (rc)
326	- return;
327	+ if (xchg(&first, 1) == 1) {
328	+ /* Slave */
329	+ clock_sync_cpu(stp_sync);
330	+ return 0;
331	+ }
332
333	- rc = chsc_sstpi(stp_page, &stp_info, sizeof(struct stp_sstpi));
334	- if (rc \|\| stp_info.c == 0)
335	- return;
336	+ /* Wait until all other cpus entered the sync function. */
337	+ while (atomic_read(&stp_sync->cpus) != 0)
338	+ cpu_relax();
339
340	- /*
341	- * Catch all other cpus and make them wait until we have
342	- * successfully synced the clock. smp_call_function will
343	- * return after all other cpus are in clock_sync_cpu_start.
344	- */
345	- memset(&stp_sync, 0, sizeof(stp_sync));
346	- preempt_disable();
347	- smp_call_function(clock_sync_cpu_start, &stp_sync, 0);
348	- local_irq_disable();
349	enable_sync_clock();
350
351	set_bit(CLOCK_SYNC_STP, &clock_sync_flags);
352	if (test_and_clear_bit(CLOCK_SYNC_ETR, &clock_sync_flags))
353	- schedule_work(&etr_work);
354	+ queue_work(time_sync_wq, &etr_work);
355
356	rc = 0;
357	if (stp_info.todoff[0] \|\| stp_info.todoff[1] \|\|
358	@@ -1475,16 +1498,43 @@ static void stp_work_fn(struct work_stru
359	}
360	if (rc) {
361	disable_sync_clock(NULL);
362	- stp_sync.in_sync = -EAGAIN;
363	+ stp_sync->in_sync = -EAGAIN;
364	clear_bit(CLOCK_SYNC_STP, &clock_sync_flags);
365	if (etr_port0_online \|\| etr_port1_online)
366	- schedule_work(&etr_work);
367	+ queue_work(time_sync_wq, &etr_work);
368	} else
369	- stp_sync.in_sync = 1;
370	+ stp_sync->in_sync = 1;
371	+ xchg(&first, 0);
372	+ return 0;
373	+}
374
375	- local_irq_enable();
376	- smp_call_function(clock_sync_cpu_end, NULL, 0);
377	- preempt_enable();
378	+/*
379	+ * STP work. Check for the STP state and take over the clock
380	+ * synchronization if the STP clock source is usable.
381	+ */
382	+static void stp_work_fn(struct work_struct *work)
383	+{
384	+ struct clock_sync_data stp_sync;
385	+ int rc;
386	+
387	+ if (!stp_online) {
388	+ chsc_sstpc(stp_page, STP_OP_CTRL, 0x0000);
389	+ return;
390	+ }
391	+
392	+ rc = chsc_sstpc(stp_page, STP_OP_CTRL, 0xb0e0);
393	+ if (rc)
394	+ return;
395	+
396	+ rc = chsc_sstpi(stp_page, &stp_info, sizeof(struct stp_sstpi));
397	+ if (rc \|\| stp_info.c == 0)
398	+ return;
399	+
400	+ memset(&stp_sync, 0, sizeof(stp_sync));
401	+ get_online_cpus();
402	+ atomic_set(&stp_sync.cpus, num_online_cpus() - 1);
403	+ stop_machine(stp_sync_clock, &stp_sync, &cpu_online_map);
404	+ put_online_cpus();
405	}
406
407	/*
408	@@ -1593,7 +1643,7 @@ static ssize_t stp_online_store(struct s
409	if (!test_bit(CLOCK_SYNC_HAS_STP, &clock_sync_flags))
410	return -EOPNOTSUPP;
411	stp_online = value;
412	- schedule_work(&stp_work);
413	+ queue_work(time_sync_wq, &stp_work);
414	return count;
415	}
416