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;
62 struct list_head list; /* Empty if single thread */
63 int freezeable; /* Freeze threads during suspend */
66 /* All the per-cpu workqueues on the system, for hotplug cpu to add/remove
67 threads to each one as cpus come/go. */
68 static DEFINE_MUTEX(workqueue_mutex);
69 static LIST_HEAD(workqueues);
71 static int singlethread_cpu __read_mostly;
72 static cpumask_t cpu_singlethread_map __read_mostly;
73 /* optimization, we could use cpu_possible_map */
74 static cpumask_t cpu_populated_map __read_mostly;
76 /* If it's single threaded, it isn't in the list of workqueues. */
77 static inline int is_single_threaded(struct workqueue_struct *wq)
79 return list_empty(&wq->list);
82 static const cpumask_t *wq_cpu_map(struct workqueue_struct *wq)
84 return is_single_threaded(wq)
85 ? &cpu_singlethread_map : &cpu_populated_map;
89 * Set the workqueue on which a work item is to be run
90 * - Must *only* be called if the pending flag is set
92 static inline void set_wq_data(struct work_struct *work, void *wq)
96 BUG_ON(!work_pending(work));
98 new = (unsigned long) wq | (1UL << WORK_STRUCT_PENDING);
99 new |= WORK_STRUCT_FLAG_MASK & *work_data_bits(work);
100 atomic_long_set(&work->data, new);
103 static inline void *get_wq_data(struct work_struct *work)
105 return (void *) (atomic_long_read(&work->data) & WORK_STRUCT_WQ_DATA_MASK);
108 static void insert_work(struct cpu_workqueue_struct *cwq,
109 struct work_struct *work, int tail)
111 set_wq_data(work, cwq);
113 list_add_tail(&work->entry, &cwq->worklist);
115 list_add(&work->entry, &cwq->worklist);
116 wake_up(&cwq->more_work);
119 /* Preempt must be disabled. */
120 static void __queue_work(struct cpu_workqueue_struct *cwq,
121 struct work_struct *work)
125 spin_lock_irqsave(&cwq->lock, flags);
126 insert_work(cwq, work, 1);
127 spin_unlock_irqrestore(&cwq->lock, flags);
131 * queue_work - queue work on a workqueue
132 * @wq: workqueue to use
133 * @work: work to queue
135 * Returns 0 if @work was already on a queue, non-zero otherwise.
137 * We queue the work to the CPU it was submitted, but there is no
138 * guarantee that it will be processed by that CPU.
140 int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
142 int ret = 0, cpu = get_cpu();
144 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
145 if (unlikely(is_single_threaded(wq)))
146 cpu = singlethread_cpu;
147 BUG_ON(!list_empty(&work->entry));
148 __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
154 EXPORT_SYMBOL_GPL(queue_work);
156 void delayed_work_timer_fn(unsigned long __data)
158 struct delayed_work *dwork = (struct delayed_work *)__data;
159 struct workqueue_struct *wq = get_wq_data(&dwork->work);
160 int cpu = smp_processor_id();
162 if (unlikely(is_single_threaded(wq)))
163 cpu = singlethread_cpu;
165 __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), &dwork->work);
169 * queue_delayed_work - queue work on a workqueue after delay
170 * @wq: workqueue to use
171 * @dwork: delayable work to queue
172 * @delay: number of jiffies to wait before queueing
174 * Returns 0 if @work was already on a queue, non-zero otherwise.
176 int fastcall queue_delayed_work(struct workqueue_struct *wq,
177 struct delayed_work *dwork, unsigned long delay)
180 struct timer_list *timer = &dwork->timer;
181 struct work_struct *work = &dwork->work;
183 timer_stats_timer_set_start_info(timer);
185 return queue_work(wq, work);
187 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
188 BUG_ON(timer_pending(timer));
189 BUG_ON(!list_empty(&work->entry));
191 /* This stores wq for the moment, for the timer_fn */
192 set_wq_data(work, wq);
193 timer->expires = jiffies + delay;
194 timer->data = (unsigned long)dwork;
195 timer->function = delayed_work_timer_fn;
201 EXPORT_SYMBOL_GPL(queue_delayed_work);
204 * queue_delayed_work_on - queue work on specific CPU after delay
205 * @cpu: CPU number to execute work on
206 * @wq: workqueue to use
207 * @dwork: work to queue
208 * @delay: number of jiffies to wait before queueing
210 * Returns 0 if @work was already on a queue, non-zero otherwise.
212 int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
213 struct delayed_work *dwork, unsigned long delay)
216 struct timer_list *timer = &dwork->timer;
217 struct work_struct *work = &dwork->work;
219 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
220 BUG_ON(timer_pending(timer));
221 BUG_ON(!list_empty(&work->entry));
223 /* This stores wq for the moment, for the timer_fn */
224 set_wq_data(work, wq);
225 timer->expires = jiffies + delay;
226 timer->data = (unsigned long)dwork;
227 timer->function = delayed_work_timer_fn;
228 add_timer_on(timer, cpu);
233 EXPORT_SYMBOL_GPL(queue_delayed_work_on);
235 static void run_workqueue(struct cpu_workqueue_struct *cwq)
237 spin_lock_irq(&cwq->lock);
239 if (cwq->run_depth > 3) {
240 /* morton gets to eat his hat */
241 printk("%s: recursion depth exceeded: %d\n",
242 __FUNCTION__, cwq->run_depth);
245 while (!list_empty(&cwq->worklist)) {
246 struct work_struct *work = list_entry(cwq->worklist.next,
247 struct work_struct, entry);
248 work_func_t f = work->func;
250 cwq->current_work = work;
251 list_del_init(cwq->worklist.next);
252 spin_unlock_irq(&cwq->lock);
254 BUG_ON(get_wq_data(work) != cwq);
255 if (!test_bit(WORK_STRUCT_NOAUTOREL, work_data_bits(work)))
259 if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
260 printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
262 current->comm, preempt_count(),
264 printk(KERN_ERR " last function: ");
265 print_symbol("%s\n", (unsigned long)f);
266 debug_show_held_locks(current);
270 spin_lock_irq(&cwq->lock);
271 cwq->current_work = NULL;
274 spin_unlock_irq(&cwq->lock);
278 * NOTE: the caller must not touch *cwq if this func returns true
280 static int cwq_should_stop(struct cpu_workqueue_struct *cwq)
282 int should_stop = cwq->should_stop;
284 if (unlikely(should_stop)) {
285 spin_lock_irq(&cwq->lock);
286 should_stop = cwq->should_stop && list_empty(&cwq->worklist);
289 spin_unlock_irq(&cwq->lock);
295 static int worker_thread(void *__cwq)
297 struct cpu_workqueue_struct *cwq = __cwq;
299 struct k_sigaction sa;
302 if (!cwq->wq->freezeable)
303 current->flags |= PF_NOFREEZE;
305 set_user_nice(current, -5);
307 /* Block and flush all signals */
308 sigfillset(&blocked);
309 sigprocmask(SIG_BLOCK, &blocked, NULL);
310 flush_signals(current);
313 * We inherited MPOL_INTERLEAVE from the booting kernel.
314 * Set MPOL_DEFAULT to insure node local allocations.
316 numa_default_policy();
318 /* SIG_IGN makes children autoreap: see do_notify_parent(). */
319 sa.sa.sa_handler = SIG_IGN;
321 siginitset(&sa.sa.sa_mask, sigmask(SIGCHLD));
322 do_sigaction(SIGCHLD, &sa, (struct k_sigaction *)0);
325 if (cwq->wq->freezeable)
328 prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE);
329 if (!cwq->should_stop && list_empty(&cwq->worklist))
331 finish_wait(&cwq->more_work, &wait);
333 if (cwq_should_stop(cwq))
343 struct work_struct work;
344 struct completion done;
347 static void wq_barrier_func(struct work_struct *work)
349 struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
350 complete(&barr->done);
353 static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
354 struct wq_barrier *barr, int tail)
356 INIT_WORK(&barr->work, wq_barrier_func);
357 __set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
359 init_completion(&barr->done);
361 insert_work(cwq, &barr->work, tail);
364 static void flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
366 if (cwq->thread == current) {
368 * Probably keventd trying to flush its own queue. So simply run
369 * it by hand rather than deadlocking.
373 struct wq_barrier barr;
376 spin_lock_irq(&cwq->lock);
377 if (!list_empty(&cwq->worklist) || cwq->current_work != NULL) {
378 insert_wq_barrier(cwq, &barr, 1);
381 spin_unlock_irq(&cwq->lock);
384 wait_for_completion(&barr.done);
389 * flush_workqueue - ensure that any scheduled work has run to completion.
390 * @wq: workqueue to flush
392 * Forces execution of the workqueue and blocks until its completion.
393 * This is typically used in driver shutdown handlers.
395 * We sleep until all works which were queued on entry have been handled,
396 * but we are not livelocked by new incoming ones.
398 * This function used to run the workqueues itself. Now we just wait for the
399 * helper threads to do it.
401 void fastcall flush_workqueue(struct workqueue_struct *wq)
403 const cpumask_t *cpu_map = wq_cpu_map(wq);
407 for_each_cpu_mask(cpu, *cpu_map)
408 flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
410 EXPORT_SYMBOL_GPL(flush_workqueue);
412 static void wait_on_work(struct cpu_workqueue_struct *cwq,
413 struct work_struct *work)
415 struct wq_barrier barr;
418 spin_lock_irq(&cwq->lock);
419 if (unlikely(cwq->current_work == work)) {
420 insert_wq_barrier(cwq, &barr, 0);
423 spin_unlock_irq(&cwq->lock);
425 if (unlikely(running))
426 wait_for_completion(&barr.done);
430 * flush_work - block until a work_struct's callback has terminated
431 * @wq: the workqueue on which the work is queued
432 * @work: the work which is to be flushed
434 * flush_work() will attempt to cancel the work if it is queued. If the work's
435 * callback appears to be running, flush_work() will block until it has
438 * flush_work() is designed to be used when the caller is tearing down data
439 * structures which the callback function operates upon. It is expected that,
440 * prior to calling flush_work(), the caller has arranged for the work to not
443 void flush_work(struct workqueue_struct *wq, struct work_struct *work)
445 const cpumask_t *cpu_map = wq_cpu_map(wq);
446 struct cpu_workqueue_struct *cwq;
451 cwq = get_wq_data(work);
452 /* Was it ever queued ? */
457 * This work can't be re-queued, no need to re-check that
458 * get_wq_data() is still the same when we take cwq->lock.
460 spin_lock_irq(&cwq->lock);
461 list_del_init(&work->entry);
463 spin_unlock_irq(&cwq->lock);
465 for_each_cpu_mask(cpu, *cpu_map)
466 wait_on_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
468 EXPORT_SYMBOL_GPL(flush_work);
471 static struct workqueue_struct *keventd_wq;
474 * schedule_work - put work task in global workqueue
475 * @work: job to be done
477 * This puts a job in the kernel-global workqueue.
479 int fastcall schedule_work(struct work_struct *work)
481 return queue_work(keventd_wq, work);
483 EXPORT_SYMBOL(schedule_work);
486 * schedule_delayed_work - put work task in global workqueue after delay
487 * @dwork: job to be done
488 * @delay: number of jiffies to wait or 0 for immediate execution
490 * After waiting for a given time this puts a job in the kernel-global
493 int fastcall schedule_delayed_work(struct delayed_work *dwork,
496 timer_stats_timer_set_start_info(&dwork->timer);
497 return queue_delayed_work(keventd_wq, dwork, delay);
499 EXPORT_SYMBOL(schedule_delayed_work);
502 * schedule_delayed_work_on - queue work in global workqueue on CPU after delay
504 * @dwork: job to be done
505 * @delay: number of jiffies to wait
507 * After waiting for a given time this puts a job in the kernel-global
508 * workqueue on the specified CPU.
510 int schedule_delayed_work_on(int cpu,
511 struct delayed_work *dwork, unsigned long delay)
513 return queue_delayed_work_on(cpu, keventd_wq, dwork, delay);
515 EXPORT_SYMBOL(schedule_delayed_work_on);
518 * schedule_on_each_cpu - call a function on each online CPU from keventd
519 * @func: the function to call
521 * Returns zero on success.
522 * Returns -ve errno on failure.
524 * Appears to be racy against CPU hotplug.
526 * schedule_on_each_cpu() is very slow.
528 int schedule_on_each_cpu(work_func_t func)
531 struct work_struct *works;
533 works = alloc_percpu(struct work_struct);
537 preempt_disable(); /* CPU hotplug */
538 for_each_online_cpu(cpu) {
539 struct work_struct *work = per_cpu_ptr(works, cpu);
541 INIT_WORK(work, func);
542 set_bit(WORK_STRUCT_PENDING, work_data_bits(work));
543 __queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu), work);
546 flush_workqueue(keventd_wq);
551 void flush_scheduled_work(void)
553 flush_workqueue(keventd_wq);
555 EXPORT_SYMBOL(flush_scheduled_work);
557 void flush_work_keventd(struct work_struct *work)
559 flush_work(keventd_wq, work);
561 EXPORT_SYMBOL(flush_work_keventd);
564 * cancel_rearming_delayed_workqueue - reliably kill off a delayed work whose handler rearms the delayed work.
565 * @wq: the controlling workqueue structure
566 * @dwork: the delayed work struct
568 void cancel_rearming_delayed_workqueue(struct workqueue_struct *wq,
569 struct delayed_work *dwork)
571 /* Was it ever queued ? */
572 if (!get_wq_data(&dwork->work))
575 while (!cancel_delayed_work(dwork))
578 EXPORT_SYMBOL(cancel_rearming_delayed_workqueue);
581 * cancel_rearming_delayed_work - reliably kill off a delayed keventd work whose handler rearms the delayed work.
582 * @dwork: the delayed work struct
584 void cancel_rearming_delayed_work(struct delayed_work *dwork)
586 cancel_rearming_delayed_workqueue(keventd_wq, dwork);
588 EXPORT_SYMBOL(cancel_rearming_delayed_work);
591 * execute_in_process_context - reliably execute the routine with user context
592 * @fn: the function to execute
593 * @ew: guaranteed storage for the execute work structure (must
594 * be available when the work executes)
596 * Executes the function immediately if process context is available,
597 * otherwise schedules the function for delayed execution.
599 * Returns: 0 - function was executed
600 * 1 - function was scheduled for execution
602 int execute_in_process_context(work_func_t fn, struct execute_work *ew)
604 if (!in_interrupt()) {
609 INIT_WORK(&ew->work, fn);
610 schedule_work(&ew->work);
614 EXPORT_SYMBOL_GPL(execute_in_process_context);
618 return keventd_wq != NULL;
621 int current_is_keventd(void)
623 struct cpu_workqueue_struct *cwq;
624 int cpu = smp_processor_id(); /* preempt-safe: keventd is per-cpu */
629 cwq = per_cpu_ptr(keventd_wq->cpu_wq, cpu);
630 if (current == cwq->thread)
637 static struct cpu_workqueue_struct *
638 init_cpu_workqueue(struct workqueue_struct *wq, int cpu)
640 struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
643 spin_lock_init(&cwq->lock);
644 INIT_LIST_HEAD(&cwq->worklist);
645 init_waitqueue_head(&cwq->more_work);
650 static int create_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
652 struct workqueue_struct *wq = cwq->wq;
653 const char *fmt = is_single_threaded(wq) ? "%s" : "%s/%d";
654 struct task_struct *p;
656 p = kthread_create(worker_thread, cwq, fmt, wq->name, cpu);
658 * Nobody can add the work_struct to this cwq,
659 * if (caller is __create_workqueue)
660 * nobody should see this wq
661 * else // caller is CPU_UP_PREPARE
662 * cpu is not on cpu_online_map
663 * so we can abort safely.
669 cwq->should_stop = 0;
670 if (!is_single_threaded(wq))
671 kthread_bind(p, cpu);
673 if (is_single_threaded(wq) || cpu_online(cpu))
679 struct workqueue_struct *__create_workqueue(const char *name,
680 int singlethread, int freezeable)
682 struct workqueue_struct *wq;
683 struct cpu_workqueue_struct *cwq;
686 wq = kzalloc(sizeof(*wq), GFP_KERNEL);
690 wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
697 wq->freezeable = freezeable;
700 INIT_LIST_HEAD(&wq->list);
701 cwq = init_cpu_workqueue(wq, singlethread_cpu);
702 err = create_workqueue_thread(cwq, singlethread_cpu);
704 mutex_lock(&workqueue_mutex);
705 list_add(&wq->list, &workqueues);
707 for_each_possible_cpu(cpu) {
708 cwq = init_cpu_workqueue(wq, cpu);
709 if (err || !cpu_online(cpu))
711 err = create_workqueue_thread(cwq, cpu);
713 mutex_unlock(&workqueue_mutex);
717 destroy_workqueue(wq);
722 EXPORT_SYMBOL_GPL(__create_workqueue);
724 static void cleanup_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
726 struct wq_barrier barr;
729 spin_lock_irq(&cwq->lock);
730 if (cwq->thread != NULL) {
731 insert_wq_barrier(cwq, &barr, 1);
732 cwq->should_stop = 1;
735 spin_unlock_irq(&cwq->lock);
738 wait_for_completion(&barr.done);
740 while (unlikely(cwq->thread != NULL))
743 * Wait until cwq->thread unlocks cwq->lock,
744 * it won't touch *cwq after that.
747 spin_unlock_wait(&cwq->lock);
752 * destroy_workqueue - safely terminate a workqueue
753 * @wq: target workqueue
755 * Safely destroy a workqueue. All work currently pending will be done first.
757 void destroy_workqueue(struct workqueue_struct *wq)
759 const cpumask_t *cpu_map = wq_cpu_map(wq);
760 struct cpu_workqueue_struct *cwq;
763 mutex_lock(&workqueue_mutex);
765 mutex_unlock(&workqueue_mutex);
767 for_each_cpu_mask(cpu, *cpu_map) {
768 cwq = per_cpu_ptr(wq->cpu_wq, cpu);
769 cleanup_workqueue_thread(cwq, cpu);
772 free_percpu(wq->cpu_wq);
775 EXPORT_SYMBOL_GPL(destroy_workqueue);
777 static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
778 unsigned long action,
781 unsigned int cpu = (unsigned long)hcpu;
782 struct cpu_workqueue_struct *cwq;
783 struct workqueue_struct *wq;
786 case CPU_LOCK_ACQUIRE:
787 mutex_lock(&workqueue_mutex);
790 case CPU_LOCK_RELEASE:
791 mutex_unlock(&workqueue_mutex);
795 cpu_set(cpu, cpu_populated_map);
798 list_for_each_entry(wq, &workqueues, list) {
799 cwq = per_cpu_ptr(wq->cpu_wq, cpu);
803 if (!create_workqueue_thread(cwq, cpu))
805 printk(KERN_ERR "workqueue for %i failed\n", cpu);
809 wake_up_process(cwq->thread);
812 case CPU_UP_CANCELED:
814 wake_up_process(cwq->thread);
816 cleanup_workqueue_thread(cwq, cpu);
824 void init_workqueues(void)
826 cpu_populated_map = cpu_online_map;
827 singlethread_cpu = first_cpu(cpu_possible_map);
828 cpu_singlethread_map = cpumask_of_cpu(singlethread_cpu);
829 hotcpu_notifier(workqueue_cpu_callback, 0);
830 keventd_wq = create_workqueue("events");