2 * linux/kernel/workqueue.c
4 * Generic mechanism for defining kernel helper threads for running
5 * arbitrary tasks in process context.
7 * Started by Ingo Molnar, Copyright (C) 2002
9 * Derived from the taskqueue/keventd code by:
11 * David Woodhouse <dwmw2@infradead.org>
12 * Andrew Morton <andrewm@uow.edu.au>
13 * Kai Petzke <wpp@marie.physik.tu-berlin.de>
14 * Theodore Ts'o <tytso@mit.edu>
16 * Made to use alloc_percpu by Christoph Lameter <clameter@sgi.com>.
19 #include <linux/module.h>
20 #include <linux/kernel.h>
21 #include <linux/sched.h>
22 #include <linux/init.h>
23 #include <linux/signal.h>
24 #include <linux/completion.h>
25 #include <linux/workqueue.h>
26 #include <linux/slab.h>
27 #include <linux/cpu.h>
28 #include <linux/notifier.h>
29 #include <linux/kthread.h>
30 #include <linux/hardirq.h>
31 #include <linux/mempolicy.h>
32 #include <linux/freezer.h>
33 #include <linux/kallsyms.h>
34 #include <linux/debug_locks.h>
37 * The per-CPU workqueue (if single thread, we always use the first
40 struct cpu_workqueue_struct {
44 struct list_head worklist;
45 wait_queue_head_t more_work;
46 struct work_struct *current_work;
48 struct workqueue_struct *wq;
49 struct task_struct *thread;
52 int run_depth; /* Detect run_workqueue() recursion depth */
53 } ____cacheline_aligned;
56 * The externally visible workqueue abstraction is an array of
59 struct workqueue_struct {
60 struct cpu_workqueue_struct *cpu_wq;
61 struct list_head list;
64 int freezeable; /* Freeze threads during suspend */
67 /* All the per-cpu workqueues on the system, for hotplug cpu to add/remove
68 threads to each one as cpus come/go. */
69 static DEFINE_MUTEX(workqueue_mutex);
70 static LIST_HEAD(workqueues);
72 static int singlethread_cpu __read_mostly;
73 static cpumask_t cpu_singlethread_map __read_mostly;
74 /* optimization, we could use cpu_possible_map */
75 static cpumask_t cpu_populated_map __read_mostly;
77 /* If it's single threaded, it isn't in the list of workqueues. */
78 static inline int is_single_threaded(struct workqueue_struct *wq)
80 return wq->singlethread;
83 static const cpumask_t *wq_cpu_map(struct workqueue_struct *wq)
85 return is_single_threaded(wq)
86 ? &cpu_singlethread_map : &cpu_populated_map;
90 struct cpu_workqueue_struct *wq_per_cpu(struct workqueue_struct *wq, int cpu)
92 if (unlikely(is_single_threaded(wq)))
93 cpu = singlethread_cpu;
94 return per_cpu_ptr(wq->cpu_wq, cpu);
98 * Set the workqueue on which a work item is to be run
99 * - Must *only* be called if the pending flag is set
101 static inline void set_wq_data(struct work_struct *work,
102 struct cpu_workqueue_struct *cwq)
106 BUG_ON(!work_pending(work));
108 new = (unsigned long) cwq | (1UL << WORK_STRUCT_PENDING);
109 new |= WORK_STRUCT_FLAG_MASK & *work_data_bits(work);
110 atomic_long_set(&work->data, new);
114 struct cpu_workqueue_struct *get_wq_data(struct work_struct *work)
116 return (void *) (atomic_long_read(&work->data) & WORK_STRUCT_WQ_DATA_MASK);
119 static void insert_work(struct cpu_workqueue_struct *cwq,
120 struct work_struct *work, int tail)
122 set_wq_data(work, cwq);
124 * Ensure that we get the right work->data if we see the
125 * result of list_add() below, see try_to_grab_pending().
129 list_add_tail(&work->entry, &cwq->worklist);
131 list_add(&work->entry, &cwq->worklist);
132 wake_up(&cwq->more_work);
135 /* Preempt must be disabled. */
136 static void __queue_work(struct cpu_workqueue_struct *cwq,
137 struct work_struct *work)
141 spin_lock_irqsave(&cwq->lock, flags);
142 insert_work(cwq, work, 1);
143 spin_unlock_irqrestore(&cwq->lock, flags);
147 * queue_work - queue work on a workqueue
148 * @wq: workqueue to use
149 * @work: work to queue
151 * Returns 0 if @work was already on a queue, non-zero otherwise.
153 * We queue the work to the CPU it was submitted, but there is no
154 * guarantee that it will be processed by that CPU.
156 int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
160 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
161 BUG_ON(!list_empty(&work->entry));
162 __queue_work(wq_per_cpu(wq, get_cpu()), work);
168 EXPORT_SYMBOL_GPL(queue_work);
170 void delayed_work_timer_fn(unsigned long __data)
172 struct delayed_work *dwork = (struct delayed_work *)__data;
173 struct cpu_workqueue_struct *cwq = get_wq_data(&dwork->work);
174 struct workqueue_struct *wq = cwq->wq;
176 __queue_work(wq_per_cpu(wq, smp_processor_id()), &dwork->work);
180 * queue_delayed_work - queue work on a workqueue after delay
181 * @wq: workqueue to use
182 * @dwork: delayable work to queue
183 * @delay: number of jiffies to wait before queueing
185 * Returns 0 if @work was already on a queue, non-zero otherwise.
187 int fastcall queue_delayed_work(struct workqueue_struct *wq,
188 struct delayed_work *dwork, unsigned long delay)
190 timer_stats_timer_set_start_info(&dwork->timer);
192 return queue_work(wq, &dwork->work);
194 return queue_delayed_work_on(-1, wq, dwork, delay);
196 EXPORT_SYMBOL_GPL(queue_delayed_work);
199 * queue_delayed_work_on - queue work on specific CPU after delay
200 * @cpu: CPU number to execute work on
201 * @wq: workqueue to use
202 * @dwork: work to queue
203 * @delay: number of jiffies to wait before queueing
205 * Returns 0 if @work was already on a queue, non-zero otherwise.
207 int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
208 struct delayed_work *dwork, unsigned long delay)
211 struct timer_list *timer = &dwork->timer;
212 struct work_struct *work = &dwork->work;
214 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
215 BUG_ON(timer_pending(timer));
216 BUG_ON(!list_empty(&work->entry));
218 /* This stores cwq for the moment, for the timer_fn */
219 set_wq_data(work, wq_per_cpu(wq, raw_smp_processor_id()));
220 timer->expires = jiffies + delay;
221 timer->data = (unsigned long)dwork;
222 timer->function = delayed_work_timer_fn;
224 if (unlikely(cpu >= 0))
225 add_timer_on(timer, cpu);
232 EXPORT_SYMBOL_GPL(queue_delayed_work_on);
234 static void run_workqueue(struct cpu_workqueue_struct *cwq)
236 spin_lock_irq(&cwq->lock);
238 if (cwq->run_depth > 3) {
239 /* morton gets to eat his hat */
240 printk("%s: recursion depth exceeded: %d\n",
241 __FUNCTION__, cwq->run_depth);
244 while (!list_empty(&cwq->worklist)) {
245 struct work_struct *work = list_entry(cwq->worklist.next,
246 struct work_struct, entry);
247 work_func_t f = work->func;
249 cwq->current_work = work;
250 list_del_init(cwq->worklist.next);
251 spin_unlock_irq(&cwq->lock);
253 BUG_ON(get_wq_data(work) != cwq);
254 work_clear_pending(work);
257 if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
258 printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
260 current->comm, preempt_count(),
262 printk(KERN_ERR " last function: ");
263 print_symbol("%s\n", (unsigned long)f);
264 debug_show_held_locks(current);
268 spin_lock_irq(&cwq->lock);
269 cwq->current_work = NULL;
272 spin_unlock_irq(&cwq->lock);
276 * NOTE: the caller must not touch *cwq if this func returns true
278 static int cwq_should_stop(struct cpu_workqueue_struct *cwq)
280 int should_stop = cwq->should_stop;
282 if (unlikely(should_stop)) {
283 spin_lock_irq(&cwq->lock);
284 should_stop = cwq->should_stop && list_empty(&cwq->worklist);
287 spin_unlock_irq(&cwq->lock);
293 static int worker_thread(void *__cwq)
295 struct cpu_workqueue_struct *cwq = __cwq;
298 if (!cwq->wq->freezeable)
299 current->flags |= PF_NOFREEZE;
301 set_user_nice(current, -5);
304 prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE);
305 if (!freezing(current) && !cwq->should_stop
306 && list_empty(&cwq->worklist))
308 finish_wait(&cwq->more_work, &wait);
312 if (cwq_should_stop(cwq))
322 struct work_struct work;
323 struct completion done;
326 static void wq_barrier_func(struct work_struct *work)
328 struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
329 complete(&barr->done);
332 static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
333 struct wq_barrier *barr, int tail)
335 INIT_WORK(&barr->work, wq_barrier_func);
336 __set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
338 init_completion(&barr->done);
340 insert_work(cwq, &barr->work, tail);
343 static void flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
345 if (cwq->thread == current) {
347 * Probably keventd trying to flush its own queue. So simply run
348 * it by hand rather than deadlocking.
352 struct wq_barrier barr;
355 spin_lock_irq(&cwq->lock);
356 if (!list_empty(&cwq->worklist) || cwq->current_work != NULL) {
357 insert_wq_barrier(cwq, &barr, 1);
360 spin_unlock_irq(&cwq->lock);
363 wait_for_completion(&barr.done);
368 * flush_workqueue - ensure that any scheduled work has run to completion.
369 * @wq: workqueue to flush
371 * Forces execution of the workqueue and blocks until its completion.
372 * This is typically used in driver shutdown handlers.
374 * We sleep until all works which were queued on entry have been handled,
375 * but we are not livelocked by new incoming ones.
377 * This function used to run the workqueues itself. Now we just wait for the
378 * helper threads to do it.
380 void fastcall flush_workqueue(struct workqueue_struct *wq)
382 const cpumask_t *cpu_map = wq_cpu_map(wq);
386 for_each_cpu_mask(cpu, *cpu_map)
387 flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
389 EXPORT_SYMBOL_GPL(flush_workqueue);
392 * Upon a successful return, the caller "owns" WORK_STRUCT_PENDING bit,
393 * so this work can't be re-armed in any way.
395 static int try_to_grab_pending(struct work_struct *work)
397 struct cpu_workqueue_struct *cwq;
400 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work)))
404 * The queueing is in progress, or it is already queued. Try to
405 * steal it from ->worklist without clearing WORK_STRUCT_PENDING.
408 cwq = get_wq_data(work);
412 spin_lock_irq(&cwq->lock);
413 if (!list_empty(&work->entry)) {
415 * This work is queued, but perhaps we locked the wrong cwq.
416 * In that case we must see the new value after rmb(), see
417 * insert_work()->wmb().
420 if (cwq == get_wq_data(work)) {
421 list_del_init(&work->entry);
425 spin_unlock_irq(&cwq->lock);
430 static void wait_on_cpu_work(struct cpu_workqueue_struct *cwq,
431 struct work_struct *work)
433 struct wq_barrier barr;
436 spin_lock_irq(&cwq->lock);
437 if (unlikely(cwq->current_work == work)) {
438 insert_wq_barrier(cwq, &barr, 0);
441 spin_unlock_irq(&cwq->lock);
443 if (unlikely(running))
444 wait_for_completion(&barr.done);
447 static void wait_on_work(struct work_struct *work)
449 struct cpu_workqueue_struct *cwq;
450 struct workqueue_struct *wq;
451 const cpumask_t *cpu_map;
456 cwq = get_wq_data(work);
461 cpu_map = wq_cpu_map(wq);
463 for_each_cpu_mask(cpu, *cpu_map)
464 wait_on_cpu_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
468 * cancel_work_sync - block until a work_struct's callback has terminated
469 * @work: the work which is to be flushed
471 * cancel_work_sync() will cancel the work if it is queued. If the work's
472 * callback appears to be running, cancel_work_sync() will block until it
475 * It is possible to use this function if the work re-queues itself. It can
476 * cancel the work even if it migrates to another workqueue, however in that
477 * case it only guarantees that work->func() has completed on the last queued
480 * cancel_work_sync(&delayed_work->work) should be used only if ->timer is not
481 * pending, otherwise it goes into a busy-wait loop until the timer expires.
483 * The caller must ensure that workqueue_struct on which this work was last
484 * queued can't be destroyed before this function returns.
486 void cancel_work_sync(struct work_struct *work)
488 while (!try_to_grab_pending(work))
491 work_clear_pending(work);
493 EXPORT_SYMBOL_GPL(cancel_work_sync);
496 * cancel_rearming_delayed_work - reliably kill off a delayed work.
497 * @dwork: the delayed work struct
499 * It is possible to use this function if @dwork rearms itself via queue_work()
500 * or queue_delayed_work(). See also the comment for cancel_work_sync().
502 void cancel_rearming_delayed_work(struct delayed_work *dwork)
504 while (!del_timer(&dwork->timer) &&
505 !try_to_grab_pending(&dwork->work))
507 wait_on_work(&dwork->work);
508 work_clear_pending(&dwork->work);
510 EXPORT_SYMBOL(cancel_rearming_delayed_work);
512 static struct workqueue_struct *keventd_wq __read_mostly;
515 * schedule_work - put work task in global workqueue
516 * @work: job to be done
518 * This puts a job in the kernel-global workqueue.
520 int fastcall schedule_work(struct work_struct *work)
522 return queue_work(keventd_wq, work);
524 EXPORT_SYMBOL(schedule_work);
527 * schedule_delayed_work - put work task in global workqueue after delay
528 * @dwork: job to be done
529 * @delay: number of jiffies to wait or 0 for immediate execution
531 * After waiting for a given time this puts a job in the kernel-global
534 int fastcall schedule_delayed_work(struct delayed_work *dwork,
537 timer_stats_timer_set_start_info(&dwork->timer);
538 return queue_delayed_work(keventd_wq, dwork, delay);
540 EXPORT_SYMBOL(schedule_delayed_work);
543 * schedule_delayed_work_on - queue work in global workqueue on CPU after delay
545 * @dwork: job to be done
546 * @delay: number of jiffies to wait
548 * After waiting for a given time this puts a job in the kernel-global
549 * workqueue on the specified CPU.
551 int schedule_delayed_work_on(int cpu,
552 struct delayed_work *dwork, unsigned long delay)
554 return queue_delayed_work_on(cpu, keventd_wq, dwork, delay);
556 EXPORT_SYMBOL(schedule_delayed_work_on);
559 * schedule_on_each_cpu - call a function on each online CPU from keventd
560 * @func: the function to call
562 * Returns zero on success.
563 * Returns -ve errno on failure.
565 * Appears to be racy against CPU hotplug.
567 * schedule_on_each_cpu() is very slow.
569 int schedule_on_each_cpu(work_func_t func)
572 struct work_struct *works;
574 works = alloc_percpu(struct work_struct);
578 preempt_disable(); /* CPU hotplug */
579 for_each_online_cpu(cpu) {
580 struct work_struct *work = per_cpu_ptr(works, cpu);
582 INIT_WORK(work, func);
583 set_bit(WORK_STRUCT_PENDING, work_data_bits(work));
584 __queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu), work);
587 flush_workqueue(keventd_wq);
592 void flush_scheduled_work(void)
594 flush_workqueue(keventd_wq);
596 EXPORT_SYMBOL(flush_scheduled_work);
599 * execute_in_process_context - reliably execute the routine with user context
600 * @fn: the function to execute
601 * @ew: guaranteed storage for the execute work structure (must
602 * be available when the work executes)
604 * Executes the function immediately if process context is available,
605 * otherwise schedules the function for delayed execution.
607 * Returns: 0 - function was executed
608 * 1 - function was scheduled for execution
610 int execute_in_process_context(work_func_t fn, struct execute_work *ew)
612 if (!in_interrupt()) {
617 INIT_WORK(&ew->work, fn);
618 schedule_work(&ew->work);
622 EXPORT_SYMBOL_GPL(execute_in_process_context);
626 return keventd_wq != NULL;
629 int current_is_keventd(void)
631 struct cpu_workqueue_struct *cwq;
632 int cpu = smp_processor_id(); /* preempt-safe: keventd is per-cpu */
637 cwq = per_cpu_ptr(keventd_wq->cpu_wq, cpu);
638 if (current == cwq->thread)
645 static struct cpu_workqueue_struct *
646 init_cpu_workqueue(struct workqueue_struct *wq, int cpu)
648 struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
651 spin_lock_init(&cwq->lock);
652 INIT_LIST_HEAD(&cwq->worklist);
653 init_waitqueue_head(&cwq->more_work);
658 static int create_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
660 struct workqueue_struct *wq = cwq->wq;
661 const char *fmt = is_single_threaded(wq) ? "%s" : "%s/%d";
662 struct task_struct *p;
664 p = kthread_create(worker_thread, cwq, fmt, wq->name, cpu);
666 * Nobody can add the work_struct to this cwq,
667 * if (caller is __create_workqueue)
668 * nobody should see this wq
669 * else // caller is CPU_UP_PREPARE
670 * cpu is not on cpu_online_map
671 * so we can abort safely.
677 cwq->should_stop = 0;
682 static void start_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
684 struct task_struct *p = cwq->thread;
688 kthread_bind(p, cpu);
693 struct workqueue_struct *__create_workqueue(const char *name,
694 int singlethread, int freezeable)
696 struct workqueue_struct *wq;
697 struct cpu_workqueue_struct *cwq;
700 wq = kzalloc(sizeof(*wq), GFP_KERNEL);
704 wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
711 wq->singlethread = singlethread;
712 wq->freezeable = freezeable;
713 INIT_LIST_HEAD(&wq->list);
716 cwq = init_cpu_workqueue(wq, singlethread_cpu);
717 err = create_workqueue_thread(cwq, singlethread_cpu);
718 start_workqueue_thread(cwq, -1);
720 mutex_lock(&workqueue_mutex);
721 list_add(&wq->list, &workqueues);
723 for_each_possible_cpu(cpu) {
724 cwq = init_cpu_workqueue(wq, cpu);
725 if (err || !cpu_online(cpu))
727 err = create_workqueue_thread(cwq, cpu);
728 start_workqueue_thread(cwq, cpu);
730 mutex_unlock(&workqueue_mutex);
734 destroy_workqueue(wq);
739 EXPORT_SYMBOL_GPL(__create_workqueue);
741 static void cleanup_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
743 struct wq_barrier barr;
746 spin_lock_irq(&cwq->lock);
747 if (cwq->thread != NULL) {
748 insert_wq_barrier(cwq, &barr, 1);
749 cwq->should_stop = 1;
752 spin_unlock_irq(&cwq->lock);
755 wait_for_completion(&barr.done);
757 while (unlikely(cwq->thread != NULL))
760 * Wait until cwq->thread unlocks cwq->lock,
761 * it won't touch *cwq after that.
764 spin_unlock_wait(&cwq->lock);
769 * destroy_workqueue - safely terminate a workqueue
770 * @wq: target workqueue
772 * Safely destroy a workqueue. All work currently pending will be done first.
774 void destroy_workqueue(struct workqueue_struct *wq)
776 const cpumask_t *cpu_map = wq_cpu_map(wq);
777 struct cpu_workqueue_struct *cwq;
780 mutex_lock(&workqueue_mutex);
782 mutex_unlock(&workqueue_mutex);
784 for_each_cpu_mask(cpu, *cpu_map) {
785 cwq = per_cpu_ptr(wq->cpu_wq, cpu);
786 cleanup_workqueue_thread(cwq, cpu);
789 free_percpu(wq->cpu_wq);
792 EXPORT_SYMBOL_GPL(destroy_workqueue);
794 static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
795 unsigned long action,
798 unsigned int cpu = (unsigned long)hcpu;
799 struct cpu_workqueue_struct *cwq;
800 struct workqueue_struct *wq;
802 action &= ~CPU_TASKS_FROZEN;
805 case CPU_LOCK_ACQUIRE:
806 mutex_lock(&workqueue_mutex);
809 case CPU_LOCK_RELEASE:
810 mutex_unlock(&workqueue_mutex);
814 cpu_set(cpu, cpu_populated_map);
817 list_for_each_entry(wq, &workqueues, list) {
818 cwq = per_cpu_ptr(wq->cpu_wq, cpu);
822 if (!create_workqueue_thread(cwq, cpu))
824 printk(KERN_ERR "workqueue for %i failed\n", cpu);
828 start_workqueue_thread(cwq, cpu);
831 case CPU_UP_CANCELED:
832 start_workqueue_thread(cwq, -1);
834 cleanup_workqueue_thread(cwq, cpu);
842 void __init init_workqueues(void)
844 cpu_populated_map = cpu_online_map;
845 singlethread_cpu = first_cpu(cpu_possible_map);
846 cpu_singlethread_map = cpumask_of_cpu(singlethread_cpu);
847 hotcpu_notifier(workqueue_cpu_callback, 0);
848 keventd_wq = create_workqueue("events");