[PATCH] workqueue.c subtle fix and core extraction

From: Rusty Russell <rusty@rustcorp.com.au>

A barrier is needed on workqueue shutdown: there's a chance that the thead
could see the wq->thread set to NULL before the completion is initialized.

Also extracts functions which actually create and destroy workqueues, for
use by hotplug CPU patch.

kernel/workqueue.c

@@ -259,24 +259,13 @@ void flush_workqueue(struct workqueue_struct *wq)
 	}
 }
 
-struct workqueue_struct *create_workqueue(const char *name)
+static int create_workqueue_thread(struct workqueue_struct *wq,
+				   const char *name,
+				   int cpu)
 {
-	int ret, cpu, destroy = 0;
-	struct cpu_workqueue_struct *cwq;
 	startup_t startup;
-	struct workqueue_struct *wq;
-
-	BUG_ON(strlen(name) > 10);
-	startup.name = name;
-
-	wq = kmalloc(sizeof(*wq), GFP_KERNEL);
-	if (!wq)
-		return NULL;
-
-	for (cpu = 0; cpu < NR_CPUS; cpu++) {
-		if (!cpu_online(cpu))
-			continue;
-		cwq = wq->cpu_wq + cpu;
+	struct cpu_workqueue_struct *cwq = wq->cpu_wq + cpu;
+	int ret;
 
 	spin_lock_init(&cwq->lock);
 	cwq->wq = wq;
@@ -289,14 +278,31 @@ struct workqueue_struct *create_workqueue(const char *name)
 
 	init_completion(&startup.done);
 	startup.cwq = cwq;
-		ret = kernel_thread(worker_thread, &startup,
-						CLONE_FS | CLONE_FILES);
-		if (ret < 0)
-			destroy = 1;
-		else {
+	startup.name = name;
+	ret = kernel_thread(worker_thread, &startup, CLONE_FS | CLONE_FILES);
+	if (ret >= 0) {
 		wait_for_completion(&startup.done);
 		BUG_ON(!cwq->thread);
 	}
+	return ret;
+}
+
+struct workqueue_struct *create_workqueue(const char *name)
+{
+	int cpu, destroy = 0;
+	struct workqueue_struct *wq;
+
+	BUG_ON(strlen(name) > 10);
+
+	wq = kmalloc(sizeof(*wq), GFP_KERNEL);
+	if (!wq)
+		return NULL;
+
+	for (cpu = 0; cpu < NR_CPUS; cpu++) {
+		if (!cpu_online(cpu))
+			continue;
+		if (create_workqueue_thread(wq, name, cpu) < 0)
+			destroy = 1;
 	}
 	/*
 	 * Was there any error during startup? If yes then clean up:
@@ -308,28 +314,33 @@ struct workqueue_struct *create_workqueue(const char *name)
 	return wq;
 }
 
-void destroy_workqueue(struct workqueue_struct *wq)
+static void cleanup_workqueue_thread(struct workqueue_struct *wq, int cpu)
 {
 	struct cpu_workqueue_struct *cwq;
-	int cpu;
-
-	flush_workqueue(wq);
 
-	for (cpu = 0; cpu < NR_CPUS; cpu++) {
-		if (!cpu_online(cpu))
-			continue;
 	cwq = wq->cpu_wq + cpu;
-		if (!cwq->thread)
-			continue;
-		/*
-		 * Initiate an exit and wait for it:
-		 */
+	if (cwq->thread) {
+		printk("Cleaning up workqueue thread for %i\n", cpu);
+		/* Initiate an exit and wait for it: */
 		init_completion(&cwq->exit);
+		wmb(); /* Thread must see !cwq->thread after completion init */
 		cwq->thread = NULL;
 		wake_up(&cwq->more_work);
 
 		wait_for_completion(&cwq->exit);
 	}
+}
+
+void destroy_workqueue(struct workqueue_struct *wq)
+{
+	int cpu;
+
+	flush_workqueue(wq);
+
+	for (cpu = 0; cpu < NR_CPUS; cpu++) {
+		if (cpu_online(cpu))
+			cleanup_workqueue_thread(wq, cpu);
+	}
 	kfree(wq);
 }