1 files changed, 139 insertions, 43 deletions

diff --git a/kernel/workqueue.c b/kernel/workqueue.c
index ce7799540c91..4048e92aa04f 100644
--- a/kernel/workqueue.c
+++ b/kernel/workqueue.c
@@ -125,7 +125,7 @@ struct cpu_workqueue_struct *get_wq_data(struct work_struct *work)
 }
 
 static void insert_work(struct cpu_workqueue_struct *cwq,
-                                struct work_struct *work, int tail)
+                        struct work_struct *work, struct list_head *head)
 {
         set_wq_data(work, cwq);
         /*
@@ -133,21 +133,17 @@ static void insert_work(struct cpu_workqueue_struct *cwq,
          * result of list_add() below, see try_to_grab_pending().
          */
         smp_wmb();
-        if (tail)
-                list_add_tail(&work->entry, &cwq->worklist);
-        else
-                list_add(&work->entry, &cwq->worklist);
+        list_add_tail(&work->entry, head);
         wake_up(&cwq->more_work);
 }
 
-/* Preempt must be disabled. */
 static void __queue_work(struct cpu_workqueue_struct *cwq,
                          struct work_struct *work)
 {
         unsigned long flags;
 
         spin_lock_irqsave(&cwq->lock, flags);
-        insert_work(cwq, work, 1);
+        insert_work(cwq, work, &cwq->worklist);
         spin_unlock_irqrestore(&cwq->lock, flags);
 }
 
@@ -163,17 +159,39 @@ static void __queue_work(struct cpu_workqueue_struct *cwq,
  */
 int queue_work(struct workqueue_struct *wq, struct work_struct *work)
 {
+        int ret;
+
+        ret = queue_work_on(get_cpu(), wq, work);
+        put_cpu();
+
+        return ret;
+}
+EXPORT_SYMBOL_GPL(queue_work);
+
+/**
+ * queue_work_on - queue work on specific cpu
+ * @cpu: CPU number to execute work on
+ * @wq: workqueue to use
+ * @work: work to queue
+ *
+ * Returns 0 if @work was already on a queue, non-zero otherwise.
+ *
+ * We queue the work to a specific CPU, the caller must ensure it
+ * can't go away.
+ */
+int
+queue_work_on(int cpu, struct workqueue_struct *wq, struct work_struct *work)
+{
         int ret = 0;
 
         if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
                 BUG_ON(!list_empty(&work->entry));
-                __queue_work(wq_per_cpu(wq, get_cpu()), work);
-                put_cpu();
+                __queue_work(wq_per_cpu(wq, cpu), work);
                 ret = 1;
         }
         return ret;
 }
-EXPORT_SYMBOL_GPL(queue_work);
+EXPORT_SYMBOL_GPL(queue_work_on);
 
 static void delayed_work_timer_fn(unsigned long __data)
 {
@@ -272,11 +290,11 @@ static void run_workqueue(struct cpu_workqueue_struct *cwq)
 
                 BUG_ON(get_wq_data(work) != cwq);
                 work_clear_pending(work);
-                lock_acquire(&cwq->wq->lockdep_map, 0, 0, 0, 2, _THIS_IP_);
-                lock_acquire(&lockdep_map, 0, 0, 0, 2, _THIS_IP_);
+                lock_map_acquire(&cwq->wq->lockdep_map);
+                lock_map_acquire(&lockdep_map);
                 f(work);
-                lock_release(&lockdep_map, 1, _THIS_IP_);
-                lock_release(&cwq->wq->lockdep_map, 1, _THIS_IP_);
+                lock_map_release(&lockdep_map);
+                lock_map_release(&cwq->wq->lockdep_map);
 
                 if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
                         printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
@@ -337,14 +355,14 @@ static void wq_barrier_func(struct work_struct *work)
 }
 
 static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
-                                        struct wq_barrier *barr, int tail)
+                        struct wq_barrier *barr, struct list_head *head)
 {
         INIT_WORK(&barr->work, wq_barrier_func);
         __set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
 
         init_completion(&barr->done);
 
-        insert_work(cwq, &barr->work, tail);
+        insert_work(cwq, &barr->work, head);
 }
 
 static int flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
@@ -364,7 +382,7 @@ static int flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
                 active = 0;
                 spin_lock_irq(&cwq->lock);
                 if (!list_empty(&cwq->worklist) || cwq->current_work != NULL) {
-                        insert_wq_barrier(cwq, &barr, 1);
+                        insert_wq_barrier(cwq, &barr, &cwq->worklist);
                         active = 1;
                 }
                 spin_unlock_irq(&cwq->lock);
@@ -395,13 +413,64 @@ void flush_workqueue(struct workqueue_struct *wq)
         int cpu;
 
         might_sleep();
-        lock_acquire(&wq->lockdep_map, 0, 0, 0, 2, _THIS_IP_);
-        lock_release(&wq->lockdep_map, 1, _THIS_IP_);
-        for_each_cpu_mask(cpu, *cpu_map)
+        lock_map_acquire(&wq->lockdep_map);
+        lock_map_release(&wq->lockdep_map);
+        for_each_cpu_mask_nr(cpu, *cpu_map)
                 flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
 }
 EXPORT_SYMBOL_GPL(flush_workqueue);
 
+/**
+ * flush_work - block until a work_struct's callback has terminated
+ * @work: the work which is to be flushed
+ *
+ * Returns false if @work has already terminated.
+ *
+ * It is expected that, prior to calling flush_work(), the caller has
+ * arranged for the work to not be requeued, otherwise it doesn't make
+ * sense to use this function.
+ */
+int flush_work(struct work_struct *work)
+{
+        struct cpu_workqueue_struct *cwq;
+        struct list_head *prev;
+        struct wq_barrier barr;
+
+        might_sleep();
+        cwq = get_wq_data(work);
+        if (!cwq)
+                return 0;
+
+        lock_map_acquire(&cwq->wq->lockdep_map);
+        lock_map_release(&cwq->wq->lockdep_map);
+
+        prev = NULL;
+        spin_lock_irq(&cwq->lock);
+        if (!list_empty(&work->entry)) {
+                /*
+                 * See the comment near try_to_grab_pending()->smp_rmb().
+                 * If it was re-queued under us we are not going to wait.
+                 */
+                smp_rmb();
+                if (unlikely(cwq != get_wq_data(work)))
+                        goto out;
+                prev = &work->entry;
+        } else {
+                if (cwq->current_work != work)
+                        goto out;
+                prev = &cwq->worklist;
+        }
+        insert_wq_barrier(cwq, &barr, prev->next);
+out:
+        spin_unlock_irq(&cwq->lock);
+        if (!prev)
+                return 0;
+
+        wait_for_completion(&barr.done);
+        return 1;
+}
+EXPORT_SYMBOL_GPL(flush_work);
+
 /*
  * Upon a successful return (>= 0), the caller "owns" WORK_STRUCT_PENDING bit,
  * so this work can't be re-armed in any way.
@@ -449,7 +518,7 @@ static void wait_on_cpu_work(struct cpu_workqueue_struct *cwq,
 
         spin_lock_irq(&cwq->lock);
         if (unlikely(cwq->current_work == work)) {
-                insert_wq_barrier(cwq, &barr, 0);
+                insert_wq_barrier(cwq, &barr, cwq->worklist.next);
                 running = 1;
         }
         spin_unlock_irq(&cwq->lock);
@@ -467,8 +536,8 @@ static void wait_on_work(struct work_struct *work)
 
         might_sleep();
 
-        lock_acquire(&work->lockdep_map, 0, 0, 0, 2, _THIS_IP_);
-        lock_release(&work->lockdep_map, 1, _THIS_IP_);
+        lock_map_acquire(&work->lockdep_map);
+        lock_map_release(&work->lockdep_map);
 
         cwq = get_wq_data(work);
         if (!cwq)
@@ -477,7 +546,7 @@ static void wait_on_work(struct work_struct *work)
         wq = cwq->wq;
         cpu_map = wq_cpu_map(wq);
 
-        for_each_cpu_mask(cpu, *cpu_map)
+        for_each_cpu_mask_nr(cpu, *cpu_map)
                 wait_on_cpu_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
 }
 
@@ -553,6 +622,19 @@ int schedule_work(struct work_struct *work)
 }
 EXPORT_SYMBOL(schedule_work);
 
+/*
+ * schedule_work_on - put work task on a specific cpu
+ * @cpu: cpu to put the work task on
+ * @work: job to be done
+ *
+ * This puts a job on a specific cpu
+ */
+int schedule_work_on(int cpu, struct work_struct *work)
+{
+        return queue_work_on(cpu, keventd_wq, work);
+}
+EXPORT_SYMBOL(schedule_work_on);
+
 /**
  * schedule_delayed_work - put work task in global workqueue after delay
  * @dwork: job to be done
@@ -607,10 +689,10 @@ int schedule_on_each_cpu(work_func_t func)
                 struct work_struct *work = per_cpu_ptr(works, cpu);
 
                 INIT_WORK(work, func);
-                set_bit(WORK_STRUCT_PENDING, work_data_bits(work));
-                __queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu), work);
+                schedule_work_on(cpu, work);
         }
-        flush_workqueue(keventd_wq);
+        for_each_online_cpu(cpu)
+                flush_work(per_cpu_ptr(works, cpu));
         put_online_cpus();
         free_percpu(works);
         return 0;
@@ -747,11 +829,22 @@ struct workqueue_struct *__create_workqueue_key(const char *name,
                 err = create_workqueue_thread(cwq, singlethread_cpu);
                 start_workqueue_thread(cwq, -1);
         } else {
-                get_online_cpus();
+                cpu_maps_update_begin();
+                /*
+                 * We must place this wq on list even if the code below fails.
+                 * cpu_down(cpu) can remove cpu from cpu_populated_map before
+                 * destroy_workqueue() takes the lock, in that case we leak
+                 * cwq[cpu]->thread.
+                 */
                 spin_lock(&workqueue_lock);
                 list_add(&wq->list, &workqueues);
                 spin_unlock(&workqueue_lock);
-
+                /*
+                 * We must initialize cwqs for each possible cpu even if we
+                 * are going to call destroy_workqueue() finally. Otherwise
+                 * cpu_up() can hit the uninitialized cwq once we drop the
+                 * lock.
+                 */
                 for_each_possible_cpu(cpu) {
                         cwq = init_cpu_workqueue(wq, cpu);
                         if (err || !cpu_online(cpu))
@@ -759,7 +852,7 @@ struct workqueue_struct *__create_workqueue_key(const char *name,
                         err = create_workqueue_thread(cwq, cpu);
                         start_workqueue_thread(cwq, cpu);
                 }
-                put_online_cpus();
+                cpu_maps_update_done();
         }
 
         if (err) {
@@ -773,18 +866,18 @@ EXPORT_SYMBOL_GPL(__create_workqueue_key);
 static void cleanup_workqueue_thread(struct cpu_workqueue_struct *cwq)
 {
         /*
-         * Our caller is either destroy_workqueue() or CPU_DEAD,
-         * get_online_cpus() protects cwq->thread.
+         * Our caller is either destroy_workqueue() or CPU_POST_DEAD,
+         * cpu_add_remove_lock protects cwq->thread.
          */
         if (cwq->thread == NULL)
                 return;
 
-        lock_acquire(&cwq->wq->lockdep_map, 0, 0, 0, 2, _THIS_IP_);
-        lock_release(&cwq->wq->lockdep_map, 1, _THIS_IP_);
+        lock_map_acquire(&cwq->wq->lockdep_map);
+        lock_map_release(&cwq->wq->lockdep_map);
 
         flush_cpu_workqueue(cwq);
         /*
-         * If the caller is CPU_DEAD and cwq->worklist was not empty,
+         * If the caller is CPU_POST_DEAD and cwq->worklist was not empty,
          * a concurrent flush_workqueue() can insert a barrier after us.
          * However, in that case run_workqueue() won't return and check
          * kthread_should_stop() until it flushes all work_struct's.
@@ -808,14 +901,14 @@ void destroy_workqueue(struct workqueue_struct *wq)
         const cpumask_t *cpu_map = wq_cpu_map(wq);
         int cpu;
 
-        get_online_cpus();
+        cpu_maps_update_begin();
         spin_lock(&workqueue_lock);
         list_del(&wq->list);
         spin_unlock(&workqueue_lock);
 
-        for_each_cpu_mask(cpu, *cpu_map)
+        for_each_cpu_mask_nr(cpu, *cpu_map)
                 cleanup_workqueue_thread(per_cpu_ptr(wq->cpu_wq, cpu));
-        put_online_cpus();
+        cpu_maps_update_done();
 
         free_percpu(wq->cpu_wq);
         kfree(wq);
@@ -829,6 +922,7 @@ static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
         unsigned int cpu = (unsigned long)hcpu;
         struct cpu_workqueue_struct *cwq;
         struct workqueue_struct *wq;
+        int ret = NOTIFY_OK;
 
         action &= ~CPU_TASKS_FROZEN;
 
@@ -836,7 +930,7 @@ static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
         case CPU_UP_PREPARE:
                 cpu_set(cpu, cpu_populated_map);
         }
-
+undo:
         list_for_each_entry(wq, &workqueues, list) {
                 cwq = per_cpu_ptr(wq->cpu_wq, cpu);
 
@@ -846,7 +940,9 @@ static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
                                 break;
                         printk(KERN_ERR "workqueue [%s] for %i failed\n",
                                 wq->name, cpu);
-                        return NOTIFY_BAD;
+                        action = CPU_UP_CANCELED;
+                        ret = NOTIFY_BAD;
+                        goto undo;
 
                 case CPU_ONLINE:
                         start_workqueue_thread(cwq, cpu);
@@ -854,7 +950,7 @@ static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
 
                 case CPU_UP_CANCELED:
                         start_workqueue_thread(cwq, -1);
-                case CPU_DEAD:
+                case CPU_POST_DEAD:
                         cleanup_workqueue_thread(cwq);
                         break;
                 }
@@ -862,11 +958,11 @@ static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
 
         switch (action) {
         case CPU_UP_CANCELED:
-        case CPU_DEAD:
+        case CPU_POST_DEAD:
                 cpu_clear(cpu, cpu_populated_map);
         }
 
-        return NOTIFY_OK;
+        return ret;
 }
 
 void __init init_workqueues(void)