aboutsummaryrefslogtreecommitdiffstats
path: root/kernel/workqueue.c
diff options
context:
space:
mode:
Diffstat (limited to 'kernel/workqueue.c')
-rw-r--r--kernel/workqueue.c3478
1 files changed, 2981 insertions, 497 deletions
diff --git a/kernel/workqueue.c b/kernel/workqueue.c
index 327d2deb4451..90db1bd1a978 100644
--- a/kernel/workqueue.c
+++ b/kernel/workqueue.c
@@ -1,19 +1,26 @@
1/* 1/*
2 * linux/kernel/workqueue.c 2 * kernel/workqueue.c - generic async execution with shared worker pool
3 * 3 *
4 * Generic mechanism for defining kernel helper threads for running 4 * Copyright (C) 2002 Ingo Molnar
5 * arbitrary tasks in process context.
6 * 5 *
7 * Started by Ingo Molnar, Copyright (C) 2002 6 * Derived from the taskqueue/keventd code by:
7 * David Woodhouse <dwmw2@infradead.org>
8 * Andrew Morton
9 * Kai Petzke <wpp@marie.physik.tu-berlin.de>
10 * Theodore Ts'o <tytso@mit.edu>
8 * 11 *
9 * Derived from the taskqueue/keventd code by: 12 * Made to use alloc_percpu by Christoph Lameter.
10 * 13 *
11 * David Woodhouse <dwmw2@infradead.org> 14 * Copyright (C) 2010 SUSE Linux Products GmbH
12 * Andrew Morton 15 * Copyright (C) 2010 Tejun Heo <tj@kernel.org>
13 * Kai Petzke <wpp@marie.physik.tu-berlin.de>
14 * Theodore Ts'o <tytso@mit.edu>
15 * 16 *
16 * Made to use alloc_percpu by Christoph Lameter. 17 * This is the generic async execution mechanism. Work items as are
18 * executed in process context. The worker pool is shared and
19 * automatically managed. There is one worker pool for each CPU and
20 * one extra for works which are better served by workers which are
21 * not bound to any specific CPU.
22 *
23 * Please read Documentation/workqueue.txt for details.
17 */ 24 */
18 25
19#include <linux/module.h> 26#include <linux/module.h>
@@ -33,41 +40,276 @@
33#include <linux/kallsyms.h> 40#include <linux/kallsyms.h>
34#include <linux/debug_locks.h> 41#include <linux/debug_locks.h>
35#include <linux/lockdep.h> 42#include <linux/lockdep.h>
36#define CREATE_TRACE_POINTS 43#include <linux/idr.h>
37#include <trace/events/workqueue.h> 44
45#include "workqueue_sched.h"
46
47enum {
48 /* global_cwq flags */
49 GCWQ_MANAGE_WORKERS = 1 << 0, /* need to manage workers */
50 GCWQ_MANAGING_WORKERS = 1 << 1, /* managing workers */
51 GCWQ_DISASSOCIATED = 1 << 2, /* cpu can't serve workers */
52 GCWQ_FREEZING = 1 << 3, /* freeze in progress */
53 GCWQ_HIGHPRI_PENDING = 1 << 4, /* highpri works on queue */
54
55 /* worker flags */
56 WORKER_STARTED = 1 << 0, /* started */
57 WORKER_DIE = 1 << 1, /* die die die */
58 WORKER_IDLE = 1 << 2, /* is idle */
59 WORKER_PREP = 1 << 3, /* preparing to run works */
60 WORKER_ROGUE = 1 << 4, /* not bound to any cpu */
61 WORKER_REBIND = 1 << 5, /* mom is home, come back */
62 WORKER_CPU_INTENSIVE = 1 << 6, /* cpu intensive */
63 WORKER_UNBOUND = 1 << 7, /* worker is unbound */
64
65 WORKER_NOT_RUNNING = WORKER_PREP | WORKER_ROGUE | WORKER_REBIND |
66 WORKER_CPU_INTENSIVE | WORKER_UNBOUND,
67
68 /* gcwq->trustee_state */
69 TRUSTEE_START = 0, /* start */
70 TRUSTEE_IN_CHARGE = 1, /* trustee in charge of gcwq */
71 TRUSTEE_BUTCHER = 2, /* butcher workers */
72 TRUSTEE_RELEASE = 3, /* release workers */
73 TRUSTEE_DONE = 4, /* trustee is done */
74
75 BUSY_WORKER_HASH_ORDER = 6, /* 64 pointers */
76 BUSY_WORKER_HASH_SIZE = 1 << BUSY_WORKER_HASH_ORDER,
77 BUSY_WORKER_HASH_MASK = BUSY_WORKER_HASH_SIZE - 1,
78
79 MAX_IDLE_WORKERS_RATIO = 4, /* 1/4 of busy can be idle */
80 IDLE_WORKER_TIMEOUT = 300 * HZ, /* keep idle ones for 5 mins */
81
82 MAYDAY_INITIAL_TIMEOUT = HZ / 100, /* call for help after 10ms */
83 MAYDAY_INTERVAL = HZ / 10, /* and then every 100ms */
84 CREATE_COOLDOWN = HZ, /* time to breath after fail */
85 TRUSTEE_COOLDOWN = HZ / 10, /* for trustee draining */
86
87 /*
88 * Rescue workers are used only on emergencies and shared by
89 * all cpus. Give -20.
90 */
91 RESCUER_NICE_LEVEL = -20,
92};
38 93
39/* 94/*
40 * The per-CPU workqueue (if single thread, we always use the first 95 * Structure fields follow one of the following exclusion rules.
41 * possible cpu). 96 *
97 * I: Modifiable by initialization/destruction paths and read-only for
98 * everyone else.
99 *
100 * P: Preemption protected. Disabling preemption is enough and should
101 * only be modified and accessed from the local cpu.
102 *
103 * L: gcwq->lock protected. Access with gcwq->lock held.
104 *
105 * X: During normal operation, modification requires gcwq->lock and
106 * should be done only from local cpu. Either disabling preemption
107 * on local cpu or grabbing gcwq->lock is enough for read access.
108 * If GCWQ_DISASSOCIATED is set, it's identical to L.
109 *
110 * F: wq->flush_mutex protected.
111 *
112 * W: workqueue_lock protected.
42 */ 113 */
43struct cpu_workqueue_struct {
44 114
45 spinlock_t lock; 115struct global_cwq;
46 116
47 struct list_head worklist; 117/*
48 wait_queue_head_t more_work; 118 * The poor guys doing the actual heavy lifting. All on-duty workers
49 struct work_struct *current_work; 119 * are either serving the manager role, on idle list or on busy hash.
120 */
121struct worker {
122 /* on idle list while idle, on busy hash table while busy */
123 union {
124 struct list_head entry; /* L: while idle */
125 struct hlist_node hentry; /* L: while busy */
126 };
50 127
51 struct workqueue_struct *wq; 128 struct work_struct *current_work; /* L: work being processed */
52 struct task_struct *thread; 129 struct cpu_workqueue_struct *current_cwq; /* L: current_work's cwq */
53} ____cacheline_aligned; 130 struct list_head scheduled; /* L: scheduled works */
131 struct task_struct *task; /* I: worker task */
132 struct global_cwq *gcwq; /* I: the associated gcwq */
133 /* 64 bytes boundary on 64bit, 32 on 32bit */
134 unsigned long last_active; /* L: last active timestamp */
135 unsigned int flags; /* X: flags */
136 int id; /* I: worker id */
137 struct work_struct rebind_work; /* L: rebind worker to cpu */
138};
139
140/*
141 * Global per-cpu workqueue. There's one and only one for each cpu
142 * and all works are queued and processed here regardless of their
143 * target workqueues.
144 */
145struct global_cwq {
146 spinlock_t lock; /* the gcwq lock */
147 struct list_head worklist; /* L: list of pending works */
148 unsigned int cpu; /* I: the associated cpu */
149 unsigned int flags; /* L: GCWQ_* flags */
150
151 int nr_workers; /* L: total number of workers */
152 int nr_idle; /* L: currently idle ones */
153
154 /* workers are chained either in the idle_list or busy_hash */
155 struct list_head idle_list; /* X: list of idle workers */
156 struct hlist_head busy_hash[BUSY_WORKER_HASH_SIZE];
157 /* L: hash of busy workers */
158
159 struct timer_list idle_timer; /* L: worker idle timeout */
160 struct timer_list mayday_timer; /* L: SOS timer for dworkers */
161
162 struct ida worker_ida; /* L: for worker IDs */
163
164 struct task_struct *trustee; /* L: for gcwq shutdown */
165 unsigned int trustee_state; /* L: trustee state */
166 wait_queue_head_t trustee_wait; /* trustee wait */
167 struct worker *first_idle; /* L: first idle worker */
168} ____cacheline_aligned_in_smp;
169
170/*
171 * The per-CPU workqueue. The lower WORK_STRUCT_FLAG_BITS of
172 * work_struct->data are used for flags and thus cwqs need to be
173 * aligned at two's power of the number of flag bits.
174 */
175struct cpu_workqueue_struct {
176 struct global_cwq *gcwq; /* I: the associated gcwq */
177 struct workqueue_struct *wq; /* I: the owning workqueue */
178 int work_color; /* L: current color */
179 int flush_color; /* L: flushing color */
180 int nr_in_flight[WORK_NR_COLORS];
181 /* L: nr of in_flight works */
182 int nr_active; /* L: nr of active works */
183 int max_active; /* L: max active works */
184 struct list_head delayed_works; /* L: delayed works */
185};
186
187/*
188 * Structure used to wait for workqueue flush.
189 */
190struct wq_flusher {
191 struct list_head list; /* F: list of flushers */
192 int flush_color; /* F: flush color waiting for */
193 struct completion done; /* flush completion */
194};
195
196/*
197 * All cpumasks are assumed to be always set on UP and thus can't be
198 * used to determine whether there's something to be done.
199 */
200#ifdef CONFIG_SMP
201typedef cpumask_var_t mayday_mask_t;
202#define mayday_test_and_set_cpu(cpu, mask) \
203 cpumask_test_and_set_cpu((cpu), (mask))
204#define mayday_clear_cpu(cpu, mask) cpumask_clear_cpu((cpu), (mask))
205#define for_each_mayday_cpu(cpu, mask) for_each_cpu((cpu), (mask))
206#define alloc_mayday_mask(maskp, gfp) zalloc_cpumask_var((maskp), (gfp))
207#define free_mayday_mask(mask) free_cpumask_var((mask))
208#else
209typedef unsigned long mayday_mask_t;
210#define mayday_test_and_set_cpu(cpu, mask) test_and_set_bit(0, &(mask))
211#define mayday_clear_cpu(cpu, mask) clear_bit(0, &(mask))
212#define for_each_mayday_cpu(cpu, mask) if ((cpu) = 0, (mask))
213#define alloc_mayday_mask(maskp, gfp) true
214#define free_mayday_mask(mask) do { } while (0)
215#endif
54 216
55/* 217/*
56 * The externally visible workqueue abstraction is an array of 218 * The externally visible workqueue abstraction is an array of
57 * per-CPU workqueues: 219 * per-CPU workqueues:
58 */ 220 */
59struct workqueue_struct { 221struct workqueue_struct {
60 struct cpu_workqueue_struct *cpu_wq; 222 unsigned int flags; /* I: WQ_* flags */
61 struct list_head list; 223 union {
62 const char *name; 224 struct cpu_workqueue_struct __percpu *pcpu;
63 int singlethread; 225 struct cpu_workqueue_struct *single;
64 int freezeable; /* Freeze threads during suspend */ 226 unsigned long v;
65 int rt; 227 } cpu_wq; /* I: cwq's */
228 struct list_head list; /* W: list of all workqueues */
229
230 struct mutex flush_mutex; /* protects wq flushing */
231 int work_color; /* F: current work color */
232 int flush_color; /* F: current flush color */
233 atomic_t nr_cwqs_to_flush; /* flush in progress */
234 struct wq_flusher *first_flusher; /* F: first flusher */
235 struct list_head flusher_queue; /* F: flush waiters */
236 struct list_head flusher_overflow; /* F: flush overflow list */
237
238 mayday_mask_t mayday_mask; /* cpus requesting rescue */
239 struct worker *rescuer; /* I: rescue worker */
240
241 int saved_max_active; /* W: saved cwq max_active */
242 const char *name; /* I: workqueue name */
66#ifdef CONFIG_LOCKDEP 243#ifdef CONFIG_LOCKDEP
67 struct lockdep_map lockdep_map; 244 struct lockdep_map lockdep_map;
68#endif 245#endif
69}; 246};
70 247
248struct workqueue_struct *system_wq __read_mostly;
249struct workqueue_struct *system_long_wq __read_mostly;
250struct workqueue_struct *system_nrt_wq __read_mostly;
251struct workqueue_struct *system_unbound_wq __read_mostly;
252EXPORT_SYMBOL_GPL(system_wq);
253EXPORT_SYMBOL_GPL(system_long_wq);
254EXPORT_SYMBOL_GPL(system_nrt_wq);
255EXPORT_SYMBOL_GPL(system_unbound_wq);
256
257#define CREATE_TRACE_POINTS
258#include <trace/events/workqueue.h>
259
260#define for_each_busy_worker(worker, i, pos, gcwq) \
261 for (i = 0; i < BUSY_WORKER_HASH_SIZE; i++) \
262 hlist_for_each_entry(worker, pos, &gcwq->busy_hash[i], hentry)
263
264static inline int __next_gcwq_cpu(int cpu, const struct cpumask *mask,
265 unsigned int sw)
266{
267 if (cpu < nr_cpu_ids) {
268 if (sw & 1) {
269 cpu = cpumask_next(cpu, mask);
270 if (cpu < nr_cpu_ids)
271 return cpu;
272 }
273 if (sw & 2)
274 return WORK_CPU_UNBOUND;
275 }
276 return WORK_CPU_NONE;
277}
278
279static inline int __next_wq_cpu(int cpu, const struct cpumask *mask,
280 struct workqueue_struct *wq)
281{
282 return __next_gcwq_cpu(cpu, mask, !(wq->flags & WQ_UNBOUND) ? 1 : 2);
283}
284
285/*
286 * CPU iterators
287 *
288 * An extra gcwq is defined for an invalid cpu number
289 * (WORK_CPU_UNBOUND) to host workqueues which are not bound to any
290 * specific CPU. The following iterators are similar to
291 * for_each_*_cpu() iterators but also considers the unbound gcwq.
292 *
293 * for_each_gcwq_cpu() : possible CPUs + WORK_CPU_UNBOUND
294 * for_each_online_gcwq_cpu() : online CPUs + WORK_CPU_UNBOUND
295 * for_each_cwq_cpu() : possible CPUs for bound workqueues,
296 * WORK_CPU_UNBOUND for unbound workqueues
297 */
298#define for_each_gcwq_cpu(cpu) \
299 for ((cpu) = __next_gcwq_cpu(-1, cpu_possible_mask, 3); \
300 (cpu) < WORK_CPU_NONE; \
301 (cpu) = __next_gcwq_cpu((cpu), cpu_possible_mask, 3))
302
303#define for_each_online_gcwq_cpu(cpu) \
304 for ((cpu) = __next_gcwq_cpu(-1, cpu_online_mask, 3); \
305 (cpu) < WORK_CPU_NONE; \
306 (cpu) = __next_gcwq_cpu((cpu), cpu_online_mask, 3))
307
308#define for_each_cwq_cpu(cpu, wq) \
309 for ((cpu) = __next_wq_cpu(-1, cpu_possible_mask, (wq)); \
310 (cpu) < WORK_CPU_NONE; \
311 (cpu) = __next_wq_cpu((cpu), cpu_possible_mask, (wq)))
312
71#ifdef CONFIG_DEBUG_OBJECTS_WORK 313#ifdef CONFIG_DEBUG_OBJECTS_WORK
72 314
73static struct debug_obj_descr work_debug_descr; 315static struct debug_obj_descr work_debug_descr;
@@ -107,7 +349,7 @@ static int work_fixup_activate(void *addr, enum debug_obj_state state)
107 * statically initialized. We just make sure that it 349 * statically initialized. We just make sure that it
108 * is tracked in the object tracker. 350 * is tracked in the object tracker.
109 */ 351 */
110 if (test_bit(WORK_STRUCT_STATIC, work_data_bits(work))) { 352 if (test_bit(WORK_STRUCT_STATIC_BIT, work_data_bits(work))) {
111 debug_object_init(work, &work_debug_descr); 353 debug_object_init(work, &work_debug_descr);
112 debug_object_activate(work, &work_debug_descr); 354 debug_object_activate(work, &work_debug_descr);
113 return 0; 355 return 0;
@@ -181,94 +423,586 @@ static inline void debug_work_deactivate(struct work_struct *work) { }
181/* Serializes the accesses to the list of workqueues. */ 423/* Serializes the accesses to the list of workqueues. */
182static DEFINE_SPINLOCK(workqueue_lock); 424static DEFINE_SPINLOCK(workqueue_lock);
183static LIST_HEAD(workqueues); 425static LIST_HEAD(workqueues);
426static bool workqueue_freezing; /* W: have wqs started freezing? */
184 427
185static int singlethread_cpu __read_mostly;
186static const struct cpumask *cpu_singlethread_map __read_mostly;
187/* 428/*
188 * _cpu_down() first removes CPU from cpu_online_map, then CPU_DEAD 429 * The almighty global cpu workqueues. nr_running is the only field
189 * flushes cwq->worklist. This means that flush_workqueue/wait_on_work 430 * which is expected to be used frequently by other cpus via
190 * which comes in between can't use for_each_online_cpu(). We could 431 * try_to_wake_up(). Put it in a separate cacheline.
191 * use cpu_possible_map, the cpumask below is more a documentation
192 * than optimization.
193 */ 432 */
194static cpumask_var_t cpu_populated_map __read_mostly; 433static DEFINE_PER_CPU(struct global_cwq, global_cwq);
434static DEFINE_PER_CPU_SHARED_ALIGNED(atomic_t, gcwq_nr_running);
195 435
196/* If it's single threaded, it isn't in the list of workqueues. */ 436/*
197static inline int is_wq_single_threaded(struct workqueue_struct *wq) 437 * Global cpu workqueue and nr_running counter for unbound gcwq. The
438 * gcwq is always online, has GCWQ_DISASSOCIATED set, and all its
439 * workers have WORKER_UNBOUND set.
440 */
441static struct global_cwq unbound_global_cwq;
442static atomic_t unbound_gcwq_nr_running = ATOMIC_INIT(0); /* always 0 */
443
444static int worker_thread(void *__worker);
445
446static struct global_cwq *get_gcwq(unsigned int cpu)
447{
448 if (cpu != WORK_CPU_UNBOUND)
449 return &per_cpu(global_cwq, cpu);
450 else
451 return &unbound_global_cwq;
452}
453
454static atomic_t *get_gcwq_nr_running(unsigned int cpu)
455{
456 if (cpu != WORK_CPU_UNBOUND)
457 return &per_cpu(gcwq_nr_running, cpu);
458 else
459 return &unbound_gcwq_nr_running;
460}
461
462static struct cpu_workqueue_struct *get_cwq(unsigned int cpu,
463 struct workqueue_struct *wq)
464{
465 if (!(wq->flags & WQ_UNBOUND)) {
466 if (likely(cpu < nr_cpu_ids)) {
467#ifdef CONFIG_SMP
468 return per_cpu_ptr(wq->cpu_wq.pcpu, cpu);
469#else
470 return wq->cpu_wq.single;
471#endif
472 }
473 } else if (likely(cpu == WORK_CPU_UNBOUND))
474 return wq->cpu_wq.single;
475 return NULL;
476}
477
478static unsigned int work_color_to_flags(int color)
198{ 479{
199 return wq->singlethread; 480 return color << WORK_STRUCT_COLOR_SHIFT;
200} 481}
201 482
202static const struct cpumask *wq_cpu_map(struct workqueue_struct *wq) 483static int get_work_color(struct work_struct *work)
203{ 484{
204 return is_wq_single_threaded(wq) 485 return (*work_data_bits(work) >> WORK_STRUCT_COLOR_SHIFT) &
205 ? cpu_singlethread_map : cpu_populated_map; 486 ((1 << WORK_STRUCT_COLOR_BITS) - 1);
206} 487}
207 488
208static 489static int work_next_color(int color)
209struct cpu_workqueue_struct *wq_per_cpu(struct workqueue_struct *wq, int cpu)
210{ 490{
211 if (unlikely(is_wq_single_threaded(wq))) 491 return (color + 1) % WORK_NR_COLORS;
212 cpu = singlethread_cpu;
213 return per_cpu_ptr(wq->cpu_wq, cpu);
214} 492}
215 493
216/* 494/*
217 * Set the workqueue on which a work item is to be run 495 * A work's data points to the cwq with WORK_STRUCT_CWQ set while the
218 * - Must *only* be called if the pending flag is set 496 * work is on queue. Once execution starts, WORK_STRUCT_CWQ is
497 * cleared and the work data contains the cpu number it was last on.
498 *
499 * set_work_{cwq|cpu}() and clear_work_data() can be used to set the
500 * cwq, cpu or clear work->data. These functions should only be
501 * called while the work is owned - ie. while the PENDING bit is set.
502 *
503 * get_work_[g]cwq() can be used to obtain the gcwq or cwq
504 * corresponding to a work. gcwq is available once the work has been
505 * queued anywhere after initialization. cwq is available only from
506 * queueing until execution starts.
219 */ 507 */
220static inline void set_wq_data(struct work_struct *work, 508static inline void set_work_data(struct work_struct *work, unsigned long data,
221 struct cpu_workqueue_struct *cwq) 509 unsigned long flags)
222{ 510{
223 unsigned long new;
224
225 BUG_ON(!work_pending(work)); 511 BUG_ON(!work_pending(work));
512 atomic_long_set(&work->data, data | flags | work_static(work));
513}
514
515static void set_work_cwq(struct work_struct *work,
516 struct cpu_workqueue_struct *cwq,
517 unsigned long extra_flags)
518{
519 set_work_data(work, (unsigned long)cwq,
520 WORK_STRUCT_PENDING | WORK_STRUCT_CWQ | extra_flags);
521}
522
523static void set_work_cpu(struct work_struct *work, unsigned int cpu)
524{
525 set_work_data(work, cpu << WORK_STRUCT_FLAG_BITS, WORK_STRUCT_PENDING);
526}
527
528static void clear_work_data(struct work_struct *work)
529{
530 set_work_data(work, WORK_STRUCT_NO_CPU, 0);
531}
532
533static struct cpu_workqueue_struct *get_work_cwq(struct work_struct *work)
534{
535 unsigned long data = atomic_long_read(&work->data);
536
537 if (data & WORK_STRUCT_CWQ)
538 return (void *)(data & WORK_STRUCT_WQ_DATA_MASK);
539 else
540 return NULL;
541}
542
543static struct global_cwq *get_work_gcwq(struct work_struct *work)
544{
545 unsigned long data = atomic_long_read(&work->data);
546 unsigned int cpu;
547
548 if (data & WORK_STRUCT_CWQ)
549 return ((struct cpu_workqueue_struct *)
550 (data & WORK_STRUCT_WQ_DATA_MASK))->gcwq;
551
552 cpu = data >> WORK_STRUCT_FLAG_BITS;
553 if (cpu == WORK_CPU_NONE)
554 return NULL;
555
556 BUG_ON(cpu >= nr_cpu_ids && cpu != WORK_CPU_UNBOUND);
557 return get_gcwq(cpu);
558}
559
560/*
561 * Policy functions. These define the policies on how the global
562 * worker pool is managed. Unless noted otherwise, these functions
563 * assume that they're being called with gcwq->lock held.
564 */
565
566static bool __need_more_worker(struct global_cwq *gcwq)
567{
568 return !atomic_read(get_gcwq_nr_running(gcwq->cpu)) ||
569 gcwq->flags & GCWQ_HIGHPRI_PENDING;
570}
571
572/*
573 * Need to wake up a worker? Called from anything but currently
574 * running workers.
575 */
576static bool need_more_worker(struct global_cwq *gcwq)
577{
578 return !list_empty(&gcwq->worklist) && __need_more_worker(gcwq);
579}
580
581/* Can I start working? Called from busy but !running workers. */
582static bool may_start_working(struct global_cwq *gcwq)
583{
584 return gcwq->nr_idle;
585}
586
587/* Do I need to keep working? Called from currently running workers. */
588static bool keep_working(struct global_cwq *gcwq)
589{
590 atomic_t *nr_running = get_gcwq_nr_running(gcwq->cpu);
591
592 return !list_empty(&gcwq->worklist) &&
593 (atomic_read(nr_running) <= 1 ||
594 gcwq->flags & GCWQ_HIGHPRI_PENDING);
595}
596
597/* Do we need a new worker? Called from manager. */
598static bool need_to_create_worker(struct global_cwq *gcwq)
599{
600 return need_more_worker(gcwq) && !may_start_working(gcwq);
601}
602
603/* Do I need to be the manager? */
604static bool need_to_manage_workers(struct global_cwq *gcwq)
605{
606 return need_to_create_worker(gcwq) || gcwq->flags & GCWQ_MANAGE_WORKERS;
607}
226 608
227 new = (unsigned long) cwq | (1UL << WORK_STRUCT_PENDING); 609/* Do we have too many workers and should some go away? */
228 new |= WORK_STRUCT_FLAG_MASK & *work_data_bits(work); 610static bool too_many_workers(struct global_cwq *gcwq)
229 atomic_long_set(&work->data, new); 611{
612 bool managing = gcwq->flags & GCWQ_MANAGING_WORKERS;
613 int nr_idle = gcwq->nr_idle + managing; /* manager is considered idle */
614 int nr_busy = gcwq->nr_workers - nr_idle;
615
616 return nr_idle > 2 && (nr_idle - 2) * MAX_IDLE_WORKERS_RATIO >= nr_busy;
230} 617}
231 618
232/* 619/*
233 * Clear WORK_STRUCT_PENDING and the workqueue on which it was queued. 620 * Wake up functions.
621 */
622
623/* Return the first worker. Safe with preemption disabled */
624static struct worker *first_worker(struct global_cwq *gcwq)
625{
626 if (unlikely(list_empty(&gcwq->idle_list)))
627 return NULL;
628
629 return list_first_entry(&gcwq->idle_list, struct worker, entry);
630}
631
632/**
633 * wake_up_worker - wake up an idle worker
634 * @gcwq: gcwq to wake worker for
635 *
636 * Wake up the first idle worker of @gcwq.
637 *
638 * CONTEXT:
639 * spin_lock_irq(gcwq->lock).
640 */
641static void wake_up_worker(struct global_cwq *gcwq)
642{
643 struct worker *worker = first_worker(gcwq);
644
645 if (likely(worker))
646 wake_up_process(worker->task);
647}
648
649/**
650 * wq_worker_waking_up - a worker is waking up
651 * @task: task waking up
652 * @cpu: CPU @task is waking up to
653 *
654 * This function is called during try_to_wake_up() when a worker is
655 * being awoken.
656 *
657 * CONTEXT:
658 * spin_lock_irq(rq->lock)
659 */
660void wq_worker_waking_up(struct task_struct *task, unsigned int cpu)
661{
662 struct worker *worker = kthread_data(task);
663
664 if (likely(!(worker->flags & WORKER_NOT_RUNNING)))
665 atomic_inc(get_gcwq_nr_running(cpu));
666}
667
668/**
669 * wq_worker_sleeping - a worker is going to sleep
670 * @task: task going to sleep
671 * @cpu: CPU in question, must be the current CPU number
672 *
673 * This function is called during schedule() when a busy worker is
674 * going to sleep. Worker on the same cpu can be woken up by
675 * returning pointer to its task.
676 *
677 * CONTEXT:
678 * spin_lock_irq(rq->lock)
679 *
680 * RETURNS:
681 * Worker task on @cpu to wake up, %NULL if none.
682 */
683struct task_struct *wq_worker_sleeping(struct task_struct *task,
684 unsigned int cpu)
685{
686 struct worker *worker = kthread_data(task), *to_wakeup = NULL;
687 struct global_cwq *gcwq = get_gcwq(cpu);
688 atomic_t *nr_running = get_gcwq_nr_running(cpu);
689
690 if (unlikely(worker->flags & WORKER_NOT_RUNNING))
691 return NULL;
692
693 /* this can only happen on the local cpu */
694 BUG_ON(cpu != raw_smp_processor_id());
695
696 /*
697 * The counterpart of the following dec_and_test, implied mb,
698 * worklist not empty test sequence is in insert_work().
699 * Please read comment there.
700 *
701 * NOT_RUNNING is clear. This means that trustee is not in
702 * charge and we're running on the local cpu w/ rq lock held
703 * and preemption disabled, which in turn means that none else
704 * could be manipulating idle_list, so dereferencing idle_list
705 * without gcwq lock is safe.
706 */
707 if (atomic_dec_and_test(nr_running) && !list_empty(&gcwq->worklist))
708 to_wakeup = first_worker(gcwq);
709 return to_wakeup ? to_wakeup->task : NULL;
710}
711
712/**
713 * worker_set_flags - set worker flags and adjust nr_running accordingly
714 * @worker: self
715 * @flags: flags to set
716 * @wakeup: wakeup an idle worker if necessary
717 *
718 * Set @flags in @worker->flags and adjust nr_running accordingly. If
719 * nr_running becomes zero and @wakeup is %true, an idle worker is
720 * woken up.
721 *
722 * CONTEXT:
723 * spin_lock_irq(gcwq->lock)
234 */ 724 */
235static inline void clear_wq_data(struct work_struct *work) 725static inline void worker_set_flags(struct worker *worker, unsigned int flags,
726 bool wakeup)
236{ 727{
237 unsigned long flags = *work_data_bits(work) & 728 struct global_cwq *gcwq = worker->gcwq;
238 (1UL << WORK_STRUCT_STATIC); 729
239 atomic_long_set(&work->data, flags); 730 WARN_ON_ONCE(worker->task != current);
731
732 /*
733 * If transitioning into NOT_RUNNING, adjust nr_running and
734 * wake up an idle worker as necessary if requested by
735 * @wakeup.
736 */
737 if ((flags & WORKER_NOT_RUNNING) &&
738 !(worker->flags & WORKER_NOT_RUNNING)) {
739 atomic_t *nr_running = get_gcwq_nr_running(gcwq->cpu);
740
741 if (wakeup) {
742 if (atomic_dec_and_test(nr_running) &&
743 !list_empty(&gcwq->worklist))
744 wake_up_worker(gcwq);
745 } else
746 atomic_dec(nr_running);
747 }
748
749 worker->flags |= flags;
240} 750}
241 751
242static inline 752/**
243struct cpu_workqueue_struct *get_wq_data(struct work_struct *work) 753 * worker_clr_flags - clear worker flags and adjust nr_running accordingly
754 * @worker: self
755 * @flags: flags to clear
756 *
757 * Clear @flags in @worker->flags and adjust nr_running accordingly.
758 *
759 * CONTEXT:
760 * spin_lock_irq(gcwq->lock)
761 */
762static inline void worker_clr_flags(struct worker *worker, unsigned int flags)
244{ 763{
245 return (void *) (atomic_long_read(&work->data) & WORK_STRUCT_WQ_DATA_MASK); 764 struct global_cwq *gcwq = worker->gcwq;
765 unsigned int oflags = worker->flags;
766
767 WARN_ON_ONCE(worker->task != current);
768
769 worker->flags &= ~flags;
770
771 /* if transitioning out of NOT_RUNNING, increment nr_running */
772 if ((flags & WORKER_NOT_RUNNING) && (oflags & WORKER_NOT_RUNNING))
773 if (!(worker->flags & WORKER_NOT_RUNNING))
774 atomic_inc(get_gcwq_nr_running(gcwq->cpu));
246} 775}
247 776
777/**
778 * busy_worker_head - return the busy hash head for a work
779 * @gcwq: gcwq of interest
780 * @work: work to be hashed
781 *
782 * Return hash head of @gcwq for @work.
783 *
784 * CONTEXT:
785 * spin_lock_irq(gcwq->lock).
786 *
787 * RETURNS:
788 * Pointer to the hash head.
789 */
790static struct hlist_head *busy_worker_head(struct global_cwq *gcwq,
791 struct work_struct *work)
792{
793 const int base_shift = ilog2(sizeof(struct work_struct));
794 unsigned long v = (unsigned long)work;
795
796 /* simple shift and fold hash, do we need something better? */
797 v >>= base_shift;
798 v += v >> BUSY_WORKER_HASH_ORDER;
799 v &= BUSY_WORKER_HASH_MASK;
800
801 return &gcwq->busy_hash[v];
802}
803
804/**
805 * __find_worker_executing_work - find worker which is executing a work
806 * @gcwq: gcwq of interest
807 * @bwh: hash head as returned by busy_worker_head()
808 * @work: work to find worker for
809 *
810 * Find a worker which is executing @work on @gcwq. @bwh should be
811 * the hash head obtained by calling busy_worker_head() with the same
812 * work.
813 *
814 * CONTEXT:
815 * spin_lock_irq(gcwq->lock).
816 *
817 * RETURNS:
818 * Pointer to worker which is executing @work if found, NULL
819 * otherwise.
820 */
821static struct worker *__find_worker_executing_work(struct global_cwq *gcwq,
822 struct hlist_head *bwh,
823 struct work_struct *work)
824{
825 struct worker *worker;
826 struct hlist_node *tmp;
827
828 hlist_for_each_entry(worker, tmp, bwh, hentry)
829 if (worker->current_work == work)
830 return worker;
831 return NULL;
832}
833
834/**
835 * find_worker_executing_work - find worker which is executing a work
836 * @gcwq: gcwq of interest
837 * @work: work to find worker for
838 *
839 * Find a worker which is executing @work on @gcwq. This function is
840 * identical to __find_worker_executing_work() except that this
841 * function calculates @bwh itself.
842 *
843 * CONTEXT:
844 * spin_lock_irq(gcwq->lock).
845 *
846 * RETURNS:
847 * Pointer to worker which is executing @work if found, NULL
848 * otherwise.
849 */
850static struct worker *find_worker_executing_work(struct global_cwq *gcwq,
851 struct work_struct *work)
852{
853 return __find_worker_executing_work(gcwq, busy_worker_head(gcwq, work),
854 work);
855}
856
857/**
858 * gcwq_determine_ins_pos - find insertion position
859 * @gcwq: gcwq of interest
860 * @cwq: cwq a work is being queued for
861 *
862 * A work for @cwq is about to be queued on @gcwq, determine insertion
863 * position for the work. If @cwq is for HIGHPRI wq, the work is
864 * queued at the head of the queue but in FIFO order with respect to
865 * other HIGHPRI works; otherwise, at the end of the queue. This
866 * function also sets GCWQ_HIGHPRI_PENDING flag to hint @gcwq that
867 * there are HIGHPRI works pending.
868 *
869 * CONTEXT:
870 * spin_lock_irq(gcwq->lock).
871 *
872 * RETURNS:
873 * Pointer to inserstion position.
874 */
875static inline struct list_head *gcwq_determine_ins_pos(struct global_cwq *gcwq,
876 struct cpu_workqueue_struct *cwq)
877{
878 struct work_struct *twork;
879
880 if (likely(!(cwq->wq->flags & WQ_HIGHPRI)))
881 return &gcwq->worklist;
882
883 list_for_each_entry(twork, &gcwq->worklist, entry) {
884 struct cpu_workqueue_struct *tcwq = get_work_cwq(twork);
885
886 if (!(tcwq->wq->flags & WQ_HIGHPRI))
887 break;
888 }
889
890 gcwq->flags |= GCWQ_HIGHPRI_PENDING;
891 return &twork->entry;
892}
893
894/**
895 * insert_work - insert a work into gcwq
896 * @cwq: cwq @work belongs to
897 * @work: work to insert
898 * @head: insertion point
899 * @extra_flags: extra WORK_STRUCT_* flags to set
900 *
901 * Insert @work which belongs to @cwq into @gcwq after @head.
902 * @extra_flags is or'd to work_struct flags.
903 *
904 * CONTEXT:
905 * spin_lock_irq(gcwq->lock).
906 */
248static void insert_work(struct cpu_workqueue_struct *cwq, 907static void insert_work(struct cpu_workqueue_struct *cwq,
249 struct work_struct *work, struct list_head *head) 908 struct work_struct *work, struct list_head *head,
909 unsigned int extra_flags)
250{ 910{
251 trace_workqueue_insertion(cwq->thread, work); 911 struct global_cwq *gcwq = cwq->gcwq;
912
913 /* we own @work, set data and link */
914 set_work_cwq(work, cwq, extra_flags);
252 915
253 set_wq_data(work, cwq);
254 /* 916 /*
255 * Ensure that we get the right work->data if we see the 917 * Ensure that we get the right work->data if we see the
256 * result of list_add() below, see try_to_grab_pending(). 918 * result of list_add() below, see try_to_grab_pending().
257 */ 919 */
258 smp_wmb(); 920 smp_wmb();
921
259 list_add_tail(&work->entry, head); 922 list_add_tail(&work->entry, head);
260 wake_up(&cwq->more_work); 923
924 /*
925 * Ensure either worker_sched_deactivated() sees the above
926 * list_add_tail() or we see zero nr_running to avoid workers
927 * lying around lazily while there are works to be processed.
928 */
929 smp_mb();
930
931 if (__need_more_worker(gcwq))
932 wake_up_worker(gcwq);
261} 933}
262 934
263static void __queue_work(struct cpu_workqueue_struct *cwq, 935static void __queue_work(unsigned int cpu, struct workqueue_struct *wq,
264 struct work_struct *work) 936 struct work_struct *work)
265{ 937{
938 struct global_cwq *gcwq;
939 struct cpu_workqueue_struct *cwq;
940 struct list_head *worklist;
941 unsigned int work_flags;
266 unsigned long flags; 942 unsigned long flags;
267 943
268 debug_work_activate(work); 944 debug_work_activate(work);
269 spin_lock_irqsave(&cwq->lock, flags); 945
270 insert_work(cwq, work, &cwq->worklist); 946 if (WARN_ON_ONCE(wq->flags & WQ_DYING))
271 spin_unlock_irqrestore(&cwq->lock, flags); 947 return;
948
949 /* determine gcwq to use */
950 if (!(wq->flags & WQ_UNBOUND)) {
951 struct global_cwq *last_gcwq;
952
953 if (unlikely(cpu == WORK_CPU_UNBOUND))
954 cpu = raw_smp_processor_id();
955
956 /*
957 * It's multi cpu. If @wq is non-reentrant and @work
958 * was previously on a different cpu, it might still
959 * be running there, in which case the work needs to
960 * be queued on that cpu to guarantee non-reentrance.
961 */
962 gcwq = get_gcwq(cpu);
963 if (wq->flags & WQ_NON_REENTRANT &&
964 (last_gcwq = get_work_gcwq(work)) && last_gcwq != gcwq) {
965 struct worker *worker;
966
967 spin_lock_irqsave(&last_gcwq->lock, flags);
968
969 worker = find_worker_executing_work(last_gcwq, work);
970
971 if (worker && worker->current_cwq->wq == wq)
972 gcwq = last_gcwq;
973 else {
974 /* meh... not running there, queue here */
975 spin_unlock_irqrestore(&last_gcwq->lock, flags);
976 spin_lock_irqsave(&gcwq->lock, flags);
977 }
978 } else
979 spin_lock_irqsave(&gcwq->lock, flags);
980 } else {
981 gcwq = get_gcwq(WORK_CPU_UNBOUND);
982 spin_lock_irqsave(&gcwq->lock, flags);
983 }
984
985 /* gcwq determined, get cwq and queue */
986 cwq = get_cwq(gcwq->cpu, wq);
987 trace_workqueue_queue_work(cpu, cwq, work);
988
989 BUG_ON(!list_empty(&work->entry));
990
991 cwq->nr_in_flight[cwq->work_color]++;
992 work_flags = work_color_to_flags(cwq->work_color);
993
994 if (likely(cwq->nr_active < cwq->max_active)) {
995 trace_workqueue_activate_work(work);
996 cwq->nr_active++;
997 worklist = gcwq_determine_ins_pos(gcwq, cwq);
998 } else {
999 work_flags |= WORK_STRUCT_DELAYED;
1000 worklist = &cwq->delayed_works;
1001 }
1002
1003 insert_work(cwq, work, worklist, work_flags);
1004
1005 spin_unlock_irqrestore(&gcwq->lock, flags);
272} 1006}
273 1007
274/** 1008/**
@@ -308,9 +1042,8 @@ queue_work_on(int cpu, struct workqueue_struct *wq, struct work_struct *work)
308{ 1042{
309 int ret = 0; 1043 int ret = 0;
310 1044
311 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) { 1045 if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
312 BUG_ON(!list_empty(&work->entry)); 1046 __queue_work(cpu, wq, work);
313 __queue_work(wq_per_cpu(wq, cpu), work);
314 ret = 1; 1047 ret = 1;
315 } 1048 }
316 return ret; 1049 return ret;
@@ -320,10 +1053,9 @@ EXPORT_SYMBOL_GPL(queue_work_on);
320static void delayed_work_timer_fn(unsigned long __data) 1053static void delayed_work_timer_fn(unsigned long __data)
321{ 1054{
322 struct delayed_work *dwork = (struct delayed_work *)__data; 1055 struct delayed_work *dwork = (struct delayed_work *)__data;
323 struct cpu_workqueue_struct *cwq = get_wq_data(&dwork->work); 1056 struct cpu_workqueue_struct *cwq = get_work_cwq(&dwork->work);
324 struct workqueue_struct *wq = cwq->wq;
325 1057
326 __queue_work(wq_per_cpu(wq, smp_processor_id()), &dwork->work); 1058 __queue_work(smp_processor_id(), cwq->wq, &dwork->work);
327} 1059}
328 1060
329/** 1061/**
@@ -360,14 +1092,31 @@ int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
360 struct timer_list *timer = &dwork->timer; 1092 struct timer_list *timer = &dwork->timer;
361 struct work_struct *work = &dwork->work; 1093 struct work_struct *work = &dwork->work;
362 1094
363 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) { 1095 if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
1096 unsigned int lcpu;
1097
364 BUG_ON(timer_pending(timer)); 1098 BUG_ON(timer_pending(timer));
365 BUG_ON(!list_empty(&work->entry)); 1099 BUG_ON(!list_empty(&work->entry));
366 1100
367 timer_stats_timer_set_start_info(&dwork->timer); 1101 timer_stats_timer_set_start_info(&dwork->timer);
368 1102
369 /* This stores cwq for the moment, for the timer_fn */ 1103 /*
370 set_wq_data(work, wq_per_cpu(wq, raw_smp_processor_id())); 1104 * This stores cwq for the moment, for the timer_fn.
1105 * Note that the work's gcwq is preserved to allow
1106 * reentrance detection for delayed works.
1107 */
1108 if (!(wq->flags & WQ_UNBOUND)) {
1109 struct global_cwq *gcwq = get_work_gcwq(work);
1110
1111 if (gcwq && gcwq->cpu != WORK_CPU_UNBOUND)
1112 lcpu = gcwq->cpu;
1113 else
1114 lcpu = raw_smp_processor_id();
1115 } else
1116 lcpu = WORK_CPU_UNBOUND;
1117
1118 set_work_cwq(work, get_cwq(lcpu, wq), 0);
1119
371 timer->expires = jiffies + delay; 1120 timer->expires = jiffies + delay;
372 timer->data = (unsigned long)dwork; 1121 timer->data = (unsigned long)dwork;
373 timer->function = delayed_work_timer_fn; 1122 timer->function = delayed_work_timer_fn;
@@ -382,80 +1131,889 @@ int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
382} 1131}
383EXPORT_SYMBOL_GPL(queue_delayed_work_on); 1132EXPORT_SYMBOL_GPL(queue_delayed_work_on);
384 1133
385static void run_workqueue(struct cpu_workqueue_struct *cwq) 1134/**
1135 * worker_enter_idle - enter idle state
1136 * @worker: worker which is entering idle state
1137 *
1138 * @worker is entering idle state. Update stats and idle timer if
1139 * necessary.
1140 *
1141 * LOCKING:
1142 * spin_lock_irq(gcwq->lock).
1143 */
1144static void worker_enter_idle(struct worker *worker)
386{ 1145{
387 spin_lock_irq(&cwq->lock); 1146 struct global_cwq *gcwq = worker->gcwq;
388 while (!list_empty(&cwq->worklist)) { 1147
389 struct work_struct *work = list_entry(cwq->worklist.next, 1148 BUG_ON(worker->flags & WORKER_IDLE);
390 struct work_struct, entry); 1149 BUG_ON(!list_empty(&worker->entry) &&
391 work_func_t f = work->func; 1150 (worker->hentry.next || worker->hentry.pprev));
392#ifdef CONFIG_LOCKDEP 1151
1152 /* can't use worker_set_flags(), also called from start_worker() */
1153 worker->flags |= WORKER_IDLE;
1154 gcwq->nr_idle++;
1155 worker->last_active = jiffies;
1156
1157 /* idle_list is LIFO */
1158 list_add(&worker->entry, &gcwq->idle_list);
1159
1160 if (likely(!(worker->flags & WORKER_ROGUE))) {
1161 if (too_many_workers(gcwq) && !timer_pending(&gcwq->idle_timer))
1162 mod_timer(&gcwq->idle_timer,
1163 jiffies + IDLE_WORKER_TIMEOUT);
1164 } else
1165 wake_up_all(&gcwq->trustee_wait);
1166
1167 /* sanity check nr_running */
1168 WARN_ON_ONCE(gcwq->nr_workers == gcwq->nr_idle &&
1169 atomic_read(get_gcwq_nr_running(gcwq->cpu)));
1170}
1171
1172/**
1173 * worker_leave_idle - leave idle state
1174 * @worker: worker which is leaving idle state
1175 *
1176 * @worker is leaving idle state. Update stats.
1177 *
1178 * LOCKING:
1179 * spin_lock_irq(gcwq->lock).
1180 */
1181static void worker_leave_idle(struct worker *worker)
1182{
1183 struct global_cwq *gcwq = worker->gcwq;
1184
1185 BUG_ON(!(worker->flags & WORKER_IDLE));
1186 worker_clr_flags(worker, WORKER_IDLE);
1187 gcwq->nr_idle--;
1188 list_del_init(&worker->entry);
1189}
1190
1191/**
1192 * worker_maybe_bind_and_lock - bind worker to its cpu if possible and lock gcwq
1193 * @worker: self
1194 *
1195 * Works which are scheduled while the cpu is online must at least be
1196 * scheduled to a worker which is bound to the cpu so that if they are
1197 * flushed from cpu callbacks while cpu is going down, they are
1198 * guaranteed to execute on the cpu.
1199 *
1200 * This function is to be used by rogue workers and rescuers to bind
1201 * themselves to the target cpu and may race with cpu going down or
1202 * coming online. kthread_bind() can't be used because it may put the
1203 * worker to already dead cpu and set_cpus_allowed_ptr() can't be used
1204 * verbatim as it's best effort and blocking and gcwq may be
1205 * [dis]associated in the meantime.
1206 *
1207 * This function tries set_cpus_allowed() and locks gcwq and verifies
1208 * the binding against GCWQ_DISASSOCIATED which is set during
1209 * CPU_DYING and cleared during CPU_ONLINE, so if the worker enters
1210 * idle state or fetches works without dropping lock, it can guarantee
1211 * the scheduling requirement described in the first paragraph.
1212 *
1213 * CONTEXT:
1214 * Might sleep. Called without any lock but returns with gcwq->lock
1215 * held.
1216 *
1217 * RETURNS:
1218 * %true if the associated gcwq is online (@worker is successfully
1219 * bound), %false if offline.
1220 */
1221static bool worker_maybe_bind_and_lock(struct worker *worker)
1222__acquires(&gcwq->lock)
1223{
1224 struct global_cwq *gcwq = worker->gcwq;
1225 struct task_struct *task = worker->task;
1226
1227 while (true) {
393 /* 1228 /*
394 * It is permissible to free the struct work_struct 1229 * The following call may fail, succeed or succeed
395 * from inside the function that is called from it, 1230 * without actually migrating the task to the cpu if
396 * this we need to take into account for lockdep too. 1231 * it races with cpu hotunplug operation. Verify
397 * To avoid bogus "held lock freed" warnings as well 1232 * against GCWQ_DISASSOCIATED.
398 * as problems when looking into work->lockdep_map,
399 * make a copy and use that here.
400 */ 1233 */
401 struct lockdep_map lockdep_map = work->lockdep_map; 1234 if (!(gcwq->flags & GCWQ_DISASSOCIATED))
402#endif 1235 set_cpus_allowed_ptr(task, get_cpu_mask(gcwq->cpu));
403 trace_workqueue_execution(cwq->thread, work); 1236
404 debug_work_deactivate(work); 1237 spin_lock_irq(&gcwq->lock);
405 cwq->current_work = work; 1238 if (gcwq->flags & GCWQ_DISASSOCIATED)
406 list_del_init(cwq->worklist.next); 1239 return false;
407 spin_unlock_irq(&cwq->lock); 1240 if (task_cpu(task) == gcwq->cpu &&
408 1241 cpumask_equal(&current->cpus_allowed,
409 BUG_ON(get_wq_data(work) != cwq); 1242 get_cpu_mask(gcwq->cpu)))
410 work_clear_pending(work); 1243 return true;
411 lock_map_acquire(&cwq->wq->lockdep_map); 1244 spin_unlock_irq(&gcwq->lock);
412 lock_map_acquire(&lockdep_map); 1245
413 f(work); 1246 /* CPU has come up inbetween, retry migration */
414 lock_map_release(&lockdep_map); 1247 cpu_relax();
415 lock_map_release(&cwq->wq->lockdep_map); 1248 }
416 1249}
417 if (unlikely(in_atomic() || lockdep_depth(current) > 0)) { 1250
418 printk(KERN_ERR "BUG: workqueue leaked lock or atomic: " 1251/*
419 "%s/0x%08x/%d\n", 1252 * Function for worker->rebind_work used to rebind rogue busy workers
420 current->comm, preempt_count(), 1253 * to the associated cpu which is coming back online. This is
421 task_pid_nr(current)); 1254 * scheduled by cpu up but can race with other cpu hotplug operations
422 printk(KERN_ERR " last function: "); 1255 * and may be executed twice without intervening cpu down.
423 print_symbol("%s\n", (unsigned long)f); 1256 */
424 debug_show_held_locks(current); 1257static void worker_rebind_fn(struct work_struct *work)
425 dump_stack(); 1258{
1259 struct worker *worker = container_of(work, struct worker, rebind_work);
1260 struct global_cwq *gcwq = worker->gcwq;
1261
1262 if (worker_maybe_bind_and_lock(worker))
1263 worker_clr_flags(worker, WORKER_REBIND);
1264
1265 spin_unlock_irq(&gcwq->lock);
1266}
1267
1268static struct worker *alloc_worker(void)
1269{
1270 struct worker *worker;
1271
1272 worker = kzalloc(sizeof(*worker), GFP_KERNEL);
1273 if (worker) {
1274 INIT_LIST_HEAD(&worker->entry);
1275 INIT_LIST_HEAD(&worker->scheduled);
1276 INIT_WORK(&worker->rebind_work, worker_rebind_fn);
1277 /* on creation a worker is in !idle && prep state */
1278 worker->flags = WORKER_PREP;
1279 }
1280 return worker;
1281}
1282
1283/**
1284 * create_worker - create a new workqueue worker
1285 * @gcwq: gcwq the new worker will belong to
1286 * @bind: whether to set affinity to @cpu or not
1287 *
1288 * Create a new worker which is bound to @gcwq. The returned worker
1289 * can be started by calling start_worker() or destroyed using
1290 * destroy_worker().
1291 *
1292 * CONTEXT:
1293 * Might sleep. Does GFP_KERNEL allocations.
1294 *
1295 * RETURNS:
1296 * Pointer to the newly created worker.
1297 */
1298static struct worker *create_worker(struct global_cwq *gcwq, bool bind)
1299{
1300 bool on_unbound_cpu = gcwq->cpu == WORK_CPU_UNBOUND;
1301 struct worker *worker = NULL;
1302 int id = -1;
1303
1304 spin_lock_irq(&gcwq->lock);
1305 while (ida_get_new(&gcwq->worker_ida, &id)) {
1306 spin_unlock_irq(&gcwq->lock);
1307 if (!ida_pre_get(&gcwq->worker_ida, GFP_KERNEL))
1308 goto fail;
1309 spin_lock_irq(&gcwq->lock);
1310 }
1311 spin_unlock_irq(&gcwq->lock);
1312
1313 worker = alloc_worker();
1314 if (!worker)
1315 goto fail;
1316
1317 worker->gcwq = gcwq;
1318 worker->id = id;
1319
1320 if (!on_unbound_cpu)
1321 worker->task = kthread_create(worker_thread, worker,
1322 "kworker/%u:%d", gcwq->cpu, id);
1323 else
1324 worker->task = kthread_create(worker_thread, worker,
1325 "kworker/u:%d", id);
1326 if (IS_ERR(worker->task))
1327 goto fail;
1328
1329 /*
1330 * A rogue worker will become a regular one if CPU comes
1331 * online later on. Make sure every worker has
1332 * PF_THREAD_BOUND set.
1333 */
1334 if (bind && !on_unbound_cpu)
1335 kthread_bind(worker->task, gcwq->cpu);
1336 else {
1337 worker->task->flags |= PF_THREAD_BOUND;
1338 if (on_unbound_cpu)
1339 worker->flags |= WORKER_UNBOUND;
1340 }
1341
1342 return worker;
1343fail:
1344 if (id >= 0) {
1345 spin_lock_irq(&gcwq->lock);
1346 ida_remove(&gcwq->worker_ida, id);
1347 spin_unlock_irq(&gcwq->lock);
1348 }
1349 kfree(worker);
1350 return NULL;
1351}
1352
1353/**
1354 * start_worker - start a newly created worker
1355 * @worker: worker to start
1356 *
1357 * Make the gcwq aware of @worker and start it.
1358 *
1359 * CONTEXT:
1360 * spin_lock_irq(gcwq->lock).
1361 */
1362static void start_worker(struct worker *worker)
1363{
1364 worker->flags |= WORKER_STARTED;
1365 worker->gcwq->nr_workers++;
1366 worker_enter_idle(worker);
1367 wake_up_process(worker->task);
1368}
1369
1370/**
1371 * destroy_worker - destroy a workqueue worker
1372 * @worker: worker to be destroyed
1373 *
1374 * Destroy @worker and adjust @gcwq stats accordingly.
1375 *
1376 * CONTEXT:
1377 * spin_lock_irq(gcwq->lock) which is released and regrabbed.
1378 */
1379static void destroy_worker(struct worker *worker)
1380{
1381 struct global_cwq *gcwq = worker->gcwq;
1382 int id = worker->id;
1383
1384 /* sanity check frenzy */
1385 BUG_ON(worker->current_work);
1386 BUG_ON(!list_empty(&worker->scheduled));
1387
1388 if (worker->flags & WORKER_STARTED)
1389 gcwq->nr_workers--;
1390 if (worker->flags & WORKER_IDLE)
1391 gcwq->nr_idle--;
1392
1393 list_del_init(&worker->entry);
1394 worker->flags |= WORKER_DIE;
1395
1396 spin_unlock_irq(&gcwq->lock);
1397
1398 kthread_stop(worker->task);
1399 kfree(worker);
1400
1401 spin_lock_irq(&gcwq->lock);
1402 ida_remove(&gcwq->worker_ida, id);
1403}
1404
1405static void idle_worker_timeout(unsigned long __gcwq)
1406{
1407 struct global_cwq *gcwq = (void *)__gcwq;
1408
1409 spin_lock_irq(&gcwq->lock);
1410
1411 if (too_many_workers(gcwq)) {
1412 struct worker *worker;
1413 unsigned long expires;
1414
1415 /* idle_list is kept in LIFO order, check the last one */
1416 worker = list_entry(gcwq->idle_list.prev, struct worker, entry);
1417 expires = worker->last_active + IDLE_WORKER_TIMEOUT;
1418
1419 if (time_before(jiffies, expires))
1420 mod_timer(&gcwq->idle_timer, expires);
1421 else {
1422 /* it's been idle for too long, wake up manager */
1423 gcwq->flags |= GCWQ_MANAGE_WORKERS;
1424 wake_up_worker(gcwq);
426 } 1425 }
1426 }
1427
1428 spin_unlock_irq(&gcwq->lock);
1429}
1430
1431static bool send_mayday(struct work_struct *work)
1432{
1433 struct cpu_workqueue_struct *cwq = get_work_cwq(work);
1434 struct workqueue_struct *wq = cwq->wq;
1435 unsigned int cpu;
1436
1437 if (!(wq->flags & WQ_RESCUER))
1438 return false;
1439
1440 /* mayday mayday mayday */
1441 cpu = cwq->gcwq->cpu;
1442 /* WORK_CPU_UNBOUND can't be set in cpumask, use cpu 0 instead */
1443 if (cpu == WORK_CPU_UNBOUND)
1444 cpu = 0;
1445 if (!mayday_test_and_set_cpu(cpu, wq->mayday_mask))
1446 wake_up_process(wq->rescuer->task);
1447 return true;
1448}
1449
1450static void gcwq_mayday_timeout(unsigned long __gcwq)
1451{
1452 struct global_cwq *gcwq = (void *)__gcwq;
1453 struct work_struct *work;
1454
1455 spin_lock_irq(&gcwq->lock);
427 1456
428 spin_lock_irq(&cwq->lock); 1457 if (need_to_create_worker(gcwq)) {
429 cwq->current_work = NULL; 1458 /*
1459 * We've been trying to create a new worker but
1460 * haven't been successful. We might be hitting an
1461 * allocation deadlock. Send distress signals to
1462 * rescuers.
1463 */
1464 list_for_each_entry(work, &gcwq->worklist, entry)
1465 send_mayday(work);
430 } 1466 }
431 spin_unlock_irq(&cwq->lock); 1467
1468 spin_unlock_irq(&gcwq->lock);
1469
1470 mod_timer(&gcwq->mayday_timer, jiffies + MAYDAY_INTERVAL);
432} 1471}
433 1472
434static int worker_thread(void *__cwq) 1473/**
1474 * maybe_create_worker - create a new worker if necessary
1475 * @gcwq: gcwq to create a new worker for
1476 *
1477 * Create a new worker for @gcwq if necessary. @gcwq is guaranteed to
1478 * have at least one idle worker on return from this function. If
1479 * creating a new worker takes longer than MAYDAY_INTERVAL, mayday is
1480 * sent to all rescuers with works scheduled on @gcwq to resolve
1481 * possible allocation deadlock.
1482 *
1483 * On return, need_to_create_worker() is guaranteed to be false and
1484 * may_start_working() true.
1485 *
1486 * LOCKING:
1487 * spin_lock_irq(gcwq->lock) which may be released and regrabbed
1488 * multiple times. Does GFP_KERNEL allocations. Called only from
1489 * manager.
1490 *
1491 * RETURNS:
1492 * false if no action was taken and gcwq->lock stayed locked, true
1493 * otherwise.
1494 */
1495static bool maybe_create_worker(struct global_cwq *gcwq)
1496__releases(&gcwq->lock)
1497__acquires(&gcwq->lock)
435{ 1498{
436 struct cpu_workqueue_struct *cwq = __cwq; 1499 if (!need_to_create_worker(gcwq))
437 DEFINE_WAIT(wait); 1500 return false;
1501restart:
1502 spin_unlock_irq(&gcwq->lock);
1503
1504 /* if we don't make progress in MAYDAY_INITIAL_TIMEOUT, call for help */
1505 mod_timer(&gcwq->mayday_timer, jiffies + MAYDAY_INITIAL_TIMEOUT);
1506
1507 while (true) {
1508 struct worker *worker;
1509
1510 worker = create_worker(gcwq, true);
1511 if (worker) {
1512 del_timer_sync(&gcwq->mayday_timer);
1513 spin_lock_irq(&gcwq->lock);
1514 start_worker(worker);
1515 BUG_ON(need_to_create_worker(gcwq));
1516 return true;
1517 }
1518
1519 if (!need_to_create_worker(gcwq))
1520 break;
1521
1522 __set_current_state(TASK_INTERRUPTIBLE);
1523 schedule_timeout(CREATE_COOLDOWN);
1524
1525 if (!need_to_create_worker(gcwq))
1526 break;
1527 }
1528
1529 del_timer_sync(&gcwq->mayday_timer);
1530 spin_lock_irq(&gcwq->lock);
1531 if (need_to_create_worker(gcwq))
1532 goto restart;
1533 return true;
1534}
438 1535
439 if (cwq->wq->freezeable) 1536/**
440 set_freezable(); 1537 * maybe_destroy_worker - destroy workers which have been idle for a while
1538 * @gcwq: gcwq to destroy workers for
1539 *
1540 * Destroy @gcwq workers which have been idle for longer than
1541 * IDLE_WORKER_TIMEOUT.
1542 *
1543 * LOCKING:
1544 * spin_lock_irq(gcwq->lock) which may be released and regrabbed
1545 * multiple times. Called only from manager.
1546 *
1547 * RETURNS:
1548 * false if no action was taken and gcwq->lock stayed locked, true
1549 * otherwise.
1550 */
1551static bool maybe_destroy_workers(struct global_cwq *gcwq)
1552{
1553 bool ret = false;
441 1554
442 for (;;) { 1555 while (too_many_workers(gcwq)) {
443 prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE); 1556 struct worker *worker;
444 if (!freezing(current) && 1557 unsigned long expires;
445 !kthread_should_stop() &&
446 list_empty(&cwq->worklist))
447 schedule();
448 finish_wait(&cwq->more_work, &wait);
449 1558
450 try_to_freeze(); 1559 worker = list_entry(gcwq->idle_list.prev, struct worker, entry);
1560 expires = worker->last_active + IDLE_WORKER_TIMEOUT;
451 1561
452 if (kthread_should_stop()) 1562 if (time_before(jiffies, expires)) {
1563 mod_timer(&gcwq->idle_timer, expires);
453 break; 1564 break;
1565 }
454 1566
455 run_workqueue(cwq); 1567 destroy_worker(worker);
1568 ret = true;
456 } 1569 }
457 1570
458 return 0; 1571 return ret;
1572}
1573
1574/**
1575 * manage_workers - manage worker pool
1576 * @worker: self
1577 *
1578 * Assume the manager role and manage gcwq worker pool @worker belongs
1579 * to. At any given time, there can be only zero or one manager per
1580 * gcwq. The exclusion is handled automatically by this function.
1581 *
1582 * The caller can safely start processing works on false return. On
1583 * true return, it's guaranteed that need_to_create_worker() is false
1584 * and may_start_working() is true.
1585 *
1586 * CONTEXT:
1587 * spin_lock_irq(gcwq->lock) which may be released and regrabbed
1588 * multiple times. Does GFP_KERNEL allocations.
1589 *
1590 * RETURNS:
1591 * false if no action was taken and gcwq->lock stayed locked, true if
1592 * some action was taken.
1593 */
1594static bool manage_workers(struct worker *worker)
1595{
1596 struct global_cwq *gcwq = worker->gcwq;
1597 bool ret = false;
1598
1599 if (gcwq->flags & GCWQ_MANAGING_WORKERS)
1600 return ret;
1601
1602 gcwq->flags &= ~GCWQ_MANAGE_WORKERS;
1603 gcwq->flags |= GCWQ_MANAGING_WORKERS;
1604
1605 /*
1606 * Destroy and then create so that may_start_working() is true
1607 * on return.
1608 */
1609 ret |= maybe_destroy_workers(gcwq);
1610 ret |= maybe_create_worker(gcwq);
1611
1612 gcwq->flags &= ~GCWQ_MANAGING_WORKERS;
1613
1614 /*
1615 * The trustee might be waiting to take over the manager
1616 * position, tell it we're done.
1617 */
1618 if (unlikely(gcwq->trustee))
1619 wake_up_all(&gcwq->trustee_wait);
1620
1621 return ret;
1622}
1623
1624/**
1625 * move_linked_works - move linked works to a list
1626 * @work: start of series of works to be scheduled
1627 * @head: target list to append @work to
1628 * @nextp: out paramter for nested worklist walking
1629 *
1630 * Schedule linked works starting from @work to @head. Work series to
1631 * be scheduled starts at @work and includes any consecutive work with
1632 * WORK_STRUCT_LINKED set in its predecessor.
1633 *
1634 * If @nextp is not NULL, it's updated to point to the next work of
1635 * the last scheduled work. This allows move_linked_works() to be
1636 * nested inside outer list_for_each_entry_safe().
1637 *
1638 * CONTEXT:
1639 * spin_lock_irq(gcwq->lock).
1640 */
1641static void move_linked_works(struct work_struct *work, struct list_head *head,
1642 struct work_struct **nextp)
1643{
1644 struct work_struct *n;
1645
1646 /*
1647 * Linked worklist will always end before the end of the list,
1648 * use NULL for list head.
1649 */
1650 list_for_each_entry_safe_from(work, n, NULL, entry) {
1651 list_move_tail(&work->entry, head);
1652 if (!(*work_data_bits(work) & WORK_STRUCT_LINKED))
1653 break;
1654 }
1655
1656 /*
1657 * If we're already inside safe list traversal and have moved
1658 * multiple works to the scheduled queue, the next position
1659 * needs to be updated.
1660 */
1661 if (nextp)
1662 *nextp = n;
1663}
1664
1665static void cwq_activate_first_delayed(struct cpu_workqueue_struct *cwq)
1666{
1667 struct work_struct *work = list_first_entry(&cwq->delayed_works,
1668 struct work_struct, entry);
1669 struct list_head *pos = gcwq_determine_ins_pos(cwq->gcwq, cwq);
1670
1671 trace_workqueue_activate_work(work);
1672 move_linked_works(work, pos, NULL);
1673 __clear_bit(WORK_STRUCT_DELAYED_BIT, work_data_bits(work));
1674 cwq->nr_active++;
1675}
1676
1677/**
1678 * cwq_dec_nr_in_flight - decrement cwq's nr_in_flight
1679 * @cwq: cwq of interest
1680 * @color: color of work which left the queue
1681 * @delayed: for a delayed work
1682 *
1683 * A work either has completed or is removed from pending queue,
1684 * decrement nr_in_flight of its cwq and handle workqueue flushing.
1685 *
1686 * CONTEXT:
1687 * spin_lock_irq(gcwq->lock).
1688 */
1689static void cwq_dec_nr_in_flight(struct cpu_workqueue_struct *cwq, int color,
1690 bool delayed)
1691{
1692 /* ignore uncolored works */
1693 if (color == WORK_NO_COLOR)
1694 return;
1695
1696 cwq->nr_in_flight[color]--;
1697
1698 if (!delayed) {
1699 cwq->nr_active--;
1700 if (!list_empty(&cwq->delayed_works)) {
1701 /* one down, submit a delayed one */
1702 if (cwq->nr_active < cwq->max_active)
1703 cwq_activate_first_delayed(cwq);
1704 }
1705 }
1706
1707 /* is flush in progress and are we at the flushing tip? */
1708 if (likely(cwq->flush_color != color))
1709 return;
1710
1711 /* are there still in-flight works? */
1712 if (cwq->nr_in_flight[color])
1713 return;
1714
1715 /* this cwq is done, clear flush_color */
1716 cwq->flush_color = -1;
1717
1718 /*
1719 * If this was the last cwq, wake up the first flusher. It
1720 * will handle the rest.
1721 */
1722 if (atomic_dec_and_test(&cwq->wq->nr_cwqs_to_flush))
1723 complete(&cwq->wq->first_flusher->done);
1724}
1725
1726/**
1727 * process_one_work - process single work
1728 * @worker: self
1729 * @work: work to process
1730 *
1731 * Process @work. This function contains all the logics necessary to
1732 * process a single work including synchronization against and
1733 * interaction with other workers on the same cpu, queueing and
1734 * flushing. As long as context requirement is met, any worker can
1735 * call this function to process a work.
1736 *
1737 * CONTEXT:
1738 * spin_lock_irq(gcwq->lock) which is released and regrabbed.
1739 */
1740static void process_one_work(struct worker *worker, struct work_struct *work)
1741__releases(&gcwq->lock)
1742__acquires(&gcwq->lock)
1743{
1744 struct cpu_workqueue_struct *cwq = get_work_cwq(work);
1745 struct global_cwq *gcwq = cwq->gcwq;
1746 struct hlist_head *bwh = busy_worker_head(gcwq, work);
1747 bool cpu_intensive = cwq->wq->flags & WQ_CPU_INTENSIVE;
1748 work_func_t f = work->func;
1749 int work_color;
1750 struct worker *collision;
1751#ifdef CONFIG_LOCKDEP
1752 /*
1753 * It is permissible to free the struct work_struct from
1754 * inside the function that is called from it, this we need to
1755 * take into account for lockdep too. To avoid bogus "held
1756 * lock freed" warnings as well as problems when looking into
1757 * work->lockdep_map, make a copy and use that here.
1758 */
1759 struct lockdep_map lockdep_map = work->lockdep_map;
1760#endif
1761 /*
1762 * A single work shouldn't be executed concurrently by
1763 * multiple workers on a single cpu. Check whether anyone is
1764 * already processing the work. If so, defer the work to the
1765 * currently executing one.
1766 */
1767 collision = __find_worker_executing_work(gcwq, bwh, work);
1768 if (unlikely(collision)) {
1769 move_linked_works(work, &collision->scheduled, NULL);
1770 return;
1771 }
1772
1773 /* claim and process */
1774 debug_work_deactivate(work);
1775 hlist_add_head(&worker->hentry, bwh);
1776 worker->current_work = work;
1777 worker->current_cwq = cwq;
1778 work_color = get_work_color(work);
1779
1780 /* record the current cpu number in the work data and dequeue */
1781 set_work_cpu(work, gcwq->cpu);
1782 list_del_init(&work->entry);
1783
1784 /*
1785 * If HIGHPRI_PENDING, check the next work, and, if HIGHPRI,
1786 * wake up another worker; otherwise, clear HIGHPRI_PENDING.
1787 */
1788 if (unlikely(gcwq->flags & GCWQ_HIGHPRI_PENDING)) {
1789 struct work_struct *nwork = list_first_entry(&gcwq->worklist,
1790 struct work_struct, entry);
1791
1792 if (!list_empty(&gcwq->worklist) &&
1793 get_work_cwq(nwork)->wq->flags & WQ_HIGHPRI)
1794 wake_up_worker(gcwq);
1795 else
1796 gcwq->flags &= ~GCWQ_HIGHPRI_PENDING;
1797 }
1798
1799 /*
1800 * CPU intensive works don't participate in concurrency
1801 * management. They're the scheduler's responsibility.
1802 */
1803 if (unlikely(cpu_intensive))
1804 worker_set_flags(worker, WORKER_CPU_INTENSIVE, true);
1805
1806 spin_unlock_irq(&gcwq->lock);
1807
1808 work_clear_pending(work);
1809 lock_map_acquire(&cwq->wq->lockdep_map);
1810 lock_map_acquire(&lockdep_map);
1811 trace_workqueue_execute_start(work);
1812 f(work);
1813 /*
1814 * While we must be careful to not use "work" after this, the trace
1815 * point will only record its address.
1816 */
1817 trace_workqueue_execute_end(work);
1818 lock_map_release(&lockdep_map);
1819 lock_map_release(&cwq->wq->lockdep_map);
1820
1821 if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
1822 printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
1823 "%s/0x%08x/%d\n",
1824 current->comm, preempt_count(), task_pid_nr(current));
1825 printk(KERN_ERR " last function: ");
1826 print_symbol("%s\n", (unsigned long)f);
1827 debug_show_held_locks(current);
1828 dump_stack();
1829 }
1830
1831 spin_lock_irq(&gcwq->lock);
1832
1833 /* clear cpu intensive status */
1834 if (unlikely(cpu_intensive))
1835 worker_clr_flags(worker, WORKER_CPU_INTENSIVE);
1836
1837 /* we're done with it, release */
1838 hlist_del_init(&worker->hentry);
1839 worker->current_work = NULL;
1840 worker->current_cwq = NULL;
1841 cwq_dec_nr_in_flight(cwq, work_color, false);
1842}
1843
1844/**
1845 * process_scheduled_works - process scheduled works
1846 * @worker: self
1847 *
1848 * Process all scheduled works. Please note that the scheduled list
1849 * may change while processing a work, so this function repeatedly
1850 * fetches a work from the top and executes it.
1851 *
1852 * CONTEXT:
1853 * spin_lock_irq(gcwq->lock) which may be released and regrabbed
1854 * multiple times.
1855 */
1856static void process_scheduled_works(struct worker *worker)
1857{
1858 while (!list_empty(&worker->scheduled)) {
1859 struct work_struct *work = list_first_entry(&worker->scheduled,
1860 struct work_struct, entry);
1861 process_one_work(worker, work);
1862 }
1863}
1864
1865/**
1866 * worker_thread - the worker thread function
1867 * @__worker: self
1868 *
1869 * The gcwq worker thread function. There's a single dynamic pool of
1870 * these per each cpu. These workers process all works regardless of
1871 * their specific target workqueue. The only exception is works which
1872 * belong to workqueues with a rescuer which will be explained in
1873 * rescuer_thread().
1874 */
1875static int worker_thread(void *__worker)
1876{
1877 struct worker *worker = __worker;
1878 struct global_cwq *gcwq = worker->gcwq;
1879
1880 /* tell the scheduler that this is a workqueue worker */
1881 worker->task->flags |= PF_WQ_WORKER;
1882woke_up:
1883 spin_lock_irq(&gcwq->lock);
1884
1885 /* DIE can be set only while we're idle, checking here is enough */
1886 if (worker->flags & WORKER_DIE) {
1887 spin_unlock_irq(&gcwq->lock);
1888 worker->task->flags &= ~PF_WQ_WORKER;
1889 return 0;
1890 }
1891
1892 worker_leave_idle(worker);
1893recheck:
1894 /* no more worker necessary? */
1895 if (!need_more_worker(gcwq))
1896 goto sleep;
1897
1898 /* do we need to manage? */
1899 if (unlikely(!may_start_working(gcwq)) && manage_workers(worker))
1900 goto recheck;
1901
1902 /*
1903 * ->scheduled list can only be filled while a worker is
1904 * preparing to process a work or actually processing it.
1905 * Make sure nobody diddled with it while I was sleeping.
1906 */
1907 BUG_ON(!list_empty(&worker->scheduled));
1908
1909 /*
1910 * When control reaches this point, we're guaranteed to have
1911 * at least one idle worker or that someone else has already
1912 * assumed the manager role.
1913 */
1914 worker_clr_flags(worker, WORKER_PREP);
1915
1916 do {
1917 struct work_struct *work =
1918 list_first_entry(&gcwq->worklist,
1919 struct work_struct, entry);
1920
1921 if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
1922 /* optimization path, not strictly necessary */
1923 process_one_work(worker, work);
1924 if (unlikely(!list_empty(&worker->scheduled)))
1925 process_scheduled_works(worker);
1926 } else {
1927 move_linked_works(work, &worker->scheduled, NULL);
1928 process_scheduled_works(worker);
1929 }
1930 } while (keep_working(gcwq));
1931
1932 worker_set_flags(worker, WORKER_PREP, false);
1933sleep:
1934 if (unlikely(need_to_manage_workers(gcwq)) && manage_workers(worker))
1935 goto recheck;
1936
1937 /*
1938 * gcwq->lock is held and there's no work to process and no
1939 * need to manage, sleep. Workers are woken up only while
1940 * holding gcwq->lock or from local cpu, so setting the
1941 * current state before releasing gcwq->lock is enough to
1942 * prevent losing any event.
1943 */
1944 worker_enter_idle(worker);
1945 __set_current_state(TASK_INTERRUPTIBLE);
1946 spin_unlock_irq(&gcwq->lock);
1947 schedule();
1948 goto woke_up;
1949}
1950
1951/**
1952 * rescuer_thread - the rescuer thread function
1953 * @__wq: the associated workqueue
1954 *
1955 * Workqueue rescuer thread function. There's one rescuer for each
1956 * workqueue which has WQ_RESCUER set.
1957 *
1958 * Regular work processing on a gcwq may block trying to create a new
1959 * worker which uses GFP_KERNEL allocation which has slight chance of
1960 * developing into deadlock if some works currently on the same queue
1961 * need to be processed to satisfy the GFP_KERNEL allocation. This is
1962 * the problem rescuer solves.
1963 *
1964 * When such condition is possible, the gcwq summons rescuers of all
1965 * workqueues which have works queued on the gcwq and let them process
1966 * those works so that forward progress can be guaranteed.
1967 *
1968 * This should happen rarely.
1969 */
1970static int rescuer_thread(void *__wq)
1971{
1972 struct workqueue_struct *wq = __wq;
1973 struct worker *rescuer = wq->rescuer;
1974 struct list_head *scheduled = &rescuer->scheduled;
1975 bool is_unbound = wq->flags & WQ_UNBOUND;
1976 unsigned int cpu;
1977
1978 set_user_nice(current, RESCUER_NICE_LEVEL);
1979repeat:
1980 set_current_state(TASK_INTERRUPTIBLE);
1981
1982 if (kthread_should_stop())
1983 return 0;
1984
1985 /*
1986 * See whether any cpu is asking for help. Unbounded
1987 * workqueues use cpu 0 in mayday_mask for CPU_UNBOUND.
1988 */
1989 for_each_mayday_cpu(cpu, wq->mayday_mask) {
1990 unsigned int tcpu = is_unbound ? WORK_CPU_UNBOUND : cpu;
1991 struct cpu_workqueue_struct *cwq = get_cwq(tcpu, wq);
1992 struct global_cwq *gcwq = cwq->gcwq;
1993 struct work_struct *work, *n;
1994
1995 __set_current_state(TASK_RUNNING);
1996 mayday_clear_cpu(cpu, wq->mayday_mask);
1997
1998 /* migrate to the target cpu if possible */
1999 rescuer->gcwq = gcwq;
2000 worker_maybe_bind_and_lock(rescuer);
2001
2002 /*
2003 * Slurp in all works issued via this workqueue and
2004 * process'em.
2005 */
2006 BUG_ON(!list_empty(&rescuer->scheduled));
2007 list_for_each_entry_safe(work, n, &gcwq->worklist, entry)
2008 if (get_work_cwq(work) == cwq)
2009 move_linked_works(work, scheduled, &n);
2010
2011 process_scheduled_works(rescuer);
2012 spin_unlock_irq(&gcwq->lock);
2013 }
2014
2015 schedule();
2016 goto repeat;
459} 2017}
460 2018
461struct wq_barrier { 2019struct wq_barrier {
@@ -469,44 +2027,137 @@ static void wq_barrier_func(struct work_struct *work)
469 complete(&barr->done); 2027 complete(&barr->done);
470} 2028}
471 2029
2030/**
2031 * insert_wq_barrier - insert a barrier work
2032 * @cwq: cwq to insert barrier into
2033 * @barr: wq_barrier to insert
2034 * @target: target work to attach @barr to
2035 * @worker: worker currently executing @target, NULL if @target is not executing
2036 *
2037 * @barr is linked to @target such that @barr is completed only after
2038 * @target finishes execution. Please note that the ordering
2039 * guarantee is observed only with respect to @target and on the local
2040 * cpu.
2041 *
2042 * Currently, a queued barrier can't be canceled. This is because
2043 * try_to_grab_pending() can't determine whether the work to be
2044 * grabbed is at the head of the queue and thus can't clear LINKED
2045 * flag of the previous work while there must be a valid next work
2046 * after a work with LINKED flag set.
2047 *
2048 * Note that when @worker is non-NULL, @target may be modified
2049 * underneath us, so we can't reliably determine cwq from @target.
2050 *
2051 * CONTEXT:
2052 * spin_lock_irq(gcwq->lock).
2053 */
472static void insert_wq_barrier(struct cpu_workqueue_struct *cwq, 2054static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
473 struct wq_barrier *barr, struct list_head *head) 2055 struct wq_barrier *barr,
2056 struct work_struct *target, struct worker *worker)
474{ 2057{
2058 struct list_head *head;
2059 unsigned int linked = 0;
2060
475 /* 2061 /*
476 * debugobject calls are safe here even with cwq->lock locked 2062 * debugobject calls are safe here even with gcwq->lock locked
477 * as we know for sure that this will not trigger any of the 2063 * as we know for sure that this will not trigger any of the
478 * checks and call back into the fixup functions where we 2064 * checks and call back into the fixup functions where we
479 * might deadlock. 2065 * might deadlock.
480 */ 2066 */
481 INIT_WORK_ON_STACK(&barr->work, wq_barrier_func); 2067 INIT_WORK_ONSTACK(&barr->work, wq_barrier_func);
482 __set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work)); 2068 __set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(&barr->work));
483
484 init_completion(&barr->done); 2069 init_completion(&barr->done);
485 2070
2071 /*
2072 * If @target is currently being executed, schedule the
2073 * barrier to the worker; otherwise, put it after @target.
2074 */
2075 if (worker)
2076 head = worker->scheduled.next;
2077 else {
2078 unsigned long *bits = work_data_bits(target);
2079
2080 head = target->entry.next;
2081 /* there can already be other linked works, inherit and set */
2082 linked = *bits & WORK_STRUCT_LINKED;
2083 __set_bit(WORK_STRUCT_LINKED_BIT, bits);
2084 }
2085
486 debug_work_activate(&barr->work); 2086 debug_work_activate(&barr->work);
487 insert_work(cwq, &barr->work, head); 2087 insert_work(cwq, &barr->work, head,
2088 work_color_to_flags(WORK_NO_COLOR) | linked);
488} 2089}
489 2090
490static int flush_cpu_workqueue(struct cpu_workqueue_struct *cwq) 2091/**
2092 * flush_workqueue_prep_cwqs - prepare cwqs for workqueue flushing
2093 * @wq: workqueue being flushed
2094 * @flush_color: new flush color, < 0 for no-op
2095 * @work_color: new work color, < 0 for no-op
2096 *
2097 * Prepare cwqs for workqueue flushing.
2098 *
2099 * If @flush_color is non-negative, flush_color on all cwqs should be
2100 * -1. If no cwq has in-flight commands at the specified color, all
2101 * cwq->flush_color's stay at -1 and %false is returned. If any cwq
2102 * has in flight commands, its cwq->flush_color is set to
2103 * @flush_color, @wq->nr_cwqs_to_flush is updated accordingly, cwq
2104 * wakeup logic is armed and %true is returned.
2105 *
2106 * The caller should have initialized @wq->first_flusher prior to
2107 * calling this function with non-negative @flush_color. If
2108 * @flush_color is negative, no flush color update is done and %false
2109 * is returned.
2110 *
2111 * If @work_color is non-negative, all cwqs should have the same
2112 * work_color which is previous to @work_color and all will be
2113 * advanced to @work_color.
2114 *
2115 * CONTEXT:
2116 * mutex_lock(wq->flush_mutex).
2117 *
2118 * RETURNS:
2119 * %true if @flush_color >= 0 and there's something to flush. %false
2120 * otherwise.
2121 */
2122static bool flush_workqueue_prep_cwqs(struct workqueue_struct *wq,
2123 int flush_color, int work_color)
491{ 2124{
492 int active = 0; 2125 bool wait = false;
493 struct wq_barrier barr; 2126 unsigned int cpu;
494
495 WARN_ON(cwq->thread == current);
496 2127
497 spin_lock_irq(&cwq->lock); 2128 if (flush_color >= 0) {
498 if (!list_empty(&cwq->worklist) || cwq->current_work != NULL) { 2129 BUG_ON(atomic_read(&wq->nr_cwqs_to_flush));
499 insert_wq_barrier(cwq, &barr, &cwq->worklist); 2130 atomic_set(&wq->nr_cwqs_to_flush, 1);
500 active = 1;
501 } 2131 }
502 spin_unlock_irq(&cwq->lock);
503 2132
504 if (active) { 2133 for_each_cwq_cpu(cpu, wq) {
505 wait_for_completion(&barr.done); 2134 struct cpu_workqueue_struct *cwq = get_cwq(cpu, wq);
506 destroy_work_on_stack(&barr.work); 2135 struct global_cwq *gcwq = cwq->gcwq;
2136
2137 spin_lock_irq(&gcwq->lock);
2138
2139 if (flush_color >= 0) {
2140 BUG_ON(cwq->flush_color != -1);
2141
2142 if (cwq->nr_in_flight[flush_color]) {
2143 cwq->flush_color = flush_color;
2144 atomic_inc(&wq->nr_cwqs_to_flush);
2145 wait = true;
2146 }
2147 }
2148
2149 if (work_color >= 0) {
2150 BUG_ON(work_color != work_next_color(cwq->work_color));
2151 cwq->work_color = work_color;
2152 }
2153
2154 spin_unlock_irq(&gcwq->lock);
507 } 2155 }
508 2156
509 return active; 2157 if (flush_color >= 0 && atomic_dec_and_test(&wq->nr_cwqs_to_flush))
2158 complete(&wq->first_flusher->done);
2159
2160 return wait;
510} 2161}
511 2162
512/** 2163/**
@@ -518,158 +2169,340 @@ static int flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
518 * 2169 *
519 * We sleep until all works which were queued on entry have been handled, 2170 * We sleep until all works which were queued on entry have been handled,
520 * but we are not livelocked by new incoming ones. 2171 * but we are not livelocked by new incoming ones.
521 *
522 * This function used to run the workqueues itself. Now we just wait for the
523 * helper threads to do it.
524 */ 2172 */
525void flush_workqueue(struct workqueue_struct *wq) 2173void flush_workqueue(struct workqueue_struct *wq)
526{ 2174{
527 const struct cpumask *cpu_map = wq_cpu_map(wq); 2175 struct wq_flusher this_flusher = {
528 int cpu; 2176 .list = LIST_HEAD_INIT(this_flusher.list),
2177 .flush_color = -1,
2178 .done = COMPLETION_INITIALIZER_ONSTACK(this_flusher.done),
2179 };
2180 int next_color;
529 2181
530 might_sleep();
531 lock_map_acquire(&wq->lockdep_map); 2182 lock_map_acquire(&wq->lockdep_map);
532 lock_map_release(&wq->lockdep_map); 2183 lock_map_release(&wq->lockdep_map);
533 for_each_cpu(cpu, cpu_map) 2184
534 flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu)); 2185 mutex_lock(&wq->flush_mutex);
2186
2187 /*
2188 * Start-to-wait phase
2189 */
2190 next_color = work_next_color(wq->work_color);
2191
2192 if (next_color != wq->flush_color) {
2193 /*
2194 * Color space is not full. The current work_color
2195 * becomes our flush_color and work_color is advanced
2196 * by one.
2197 */
2198 BUG_ON(!list_empty(&wq->flusher_overflow));
2199 this_flusher.flush_color = wq->work_color;
2200 wq->work_color = next_color;
2201
2202 if (!wq->first_flusher) {
2203 /* no flush in progress, become the first flusher */
2204 BUG_ON(wq->flush_color != this_flusher.flush_color);
2205
2206 wq->first_flusher = &this_flusher;
2207
2208 if (!flush_workqueue_prep_cwqs(wq, wq->flush_color,
2209 wq->work_color)) {
2210 /* nothing to flush, done */
2211 wq->flush_color = next_color;
2212 wq->first_flusher = NULL;
2213 goto out_unlock;
2214 }
2215 } else {
2216 /* wait in queue */
2217 BUG_ON(wq->flush_color == this_flusher.flush_color);
2218 list_add_tail(&this_flusher.list, &wq->flusher_queue);
2219 flush_workqueue_prep_cwqs(wq, -1, wq->work_color);
2220 }
2221 } else {
2222 /*
2223 * Oops, color space is full, wait on overflow queue.
2224 * The next flush completion will assign us
2225 * flush_color and transfer to flusher_queue.
2226 */
2227 list_add_tail(&this_flusher.list, &wq->flusher_overflow);
2228 }
2229
2230 mutex_unlock(&wq->flush_mutex);
2231
2232 wait_for_completion(&this_flusher.done);
2233
2234 /*
2235 * Wake-up-and-cascade phase
2236 *
2237 * First flushers are responsible for cascading flushes and
2238 * handling overflow. Non-first flushers can simply return.
2239 */
2240 if (wq->first_flusher != &this_flusher)
2241 return;
2242
2243 mutex_lock(&wq->flush_mutex);
2244
2245 /* we might have raced, check again with mutex held */
2246 if (wq->first_flusher != &this_flusher)
2247 goto out_unlock;
2248
2249 wq->first_flusher = NULL;
2250
2251 BUG_ON(!list_empty(&this_flusher.list));
2252 BUG_ON(wq->flush_color != this_flusher.flush_color);
2253
2254 while (true) {
2255 struct wq_flusher *next, *tmp;
2256
2257 /* complete all the flushers sharing the current flush color */
2258 list_for_each_entry_safe(next, tmp, &wq->flusher_queue, list) {
2259 if (next->flush_color != wq->flush_color)
2260 break;
2261 list_del_init(&next->list);
2262 complete(&next->done);
2263 }
2264
2265 BUG_ON(!list_empty(&wq->flusher_overflow) &&
2266 wq->flush_color != work_next_color(wq->work_color));
2267
2268 /* this flush_color is finished, advance by one */
2269 wq->flush_color = work_next_color(wq->flush_color);
2270
2271 /* one color has been freed, handle overflow queue */
2272 if (!list_empty(&wq->flusher_overflow)) {
2273 /*
2274 * Assign the same color to all overflowed
2275 * flushers, advance work_color and append to
2276 * flusher_queue. This is the start-to-wait
2277 * phase for these overflowed flushers.
2278 */
2279 list_for_each_entry(tmp, &wq->flusher_overflow, list)
2280 tmp->flush_color = wq->work_color;
2281
2282 wq->work_color = work_next_color(wq->work_color);
2283
2284 list_splice_tail_init(&wq->flusher_overflow,
2285 &wq->flusher_queue);
2286 flush_workqueue_prep_cwqs(wq, -1, wq->work_color);
2287 }
2288
2289 if (list_empty(&wq->flusher_queue)) {
2290 BUG_ON(wq->flush_color != wq->work_color);
2291 break;
2292 }
2293
2294 /*
2295 * Need to flush more colors. Make the next flusher
2296 * the new first flusher and arm cwqs.
2297 */
2298 BUG_ON(wq->flush_color == wq->work_color);
2299 BUG_ON(wq->flush_color != next->flush_color);
2300
2301 list_del_init(&next->list);
2302 wq->first_flusher = next;
2303
2304 if (flush_workqueue_prep_cwqs(wq, wq->flush_color, -1))
2305 break;
2306
2307 /*
2308 * Meh... this color is already done, clear first
2309 * flusher and repeat cascading.
2310 */
2311 wq->first_flusher = NULL;
2312 }
2313
2314out_unlock:
2315 mutex_unlock(&wq->flush_mutex);
535} 2316}
536EXPORT_SYMBOL_GPL(flush_workqueue); 2317EXPORT_SYMBOL_GPL(flush_workqueue);
537 2318
538/** 2319static bool start_flush_work(struct work_struct *work, struct wq_barrier *barr,
539 * flush_work - block until a work_struct's callback has terminated 2320 bool wait_executing)
540 * @work: the work which is to be flushed
541 *
542 * Returns false if @work has already terminated.
543 *
544 * It is expected that, prior to calling flush_work(), the caller has
545 * arranged for the work to not be requeued, otherwise it doesn't make
546 * sense to use this function.
547 */
548int flush_work(struct work_struct *work)
549{ 2321{
2322 struct worker *worker = NULL;
2323 struct global_cwq *gcwq;
550 struct cpu_workqueue_struct *cwq; 2324 struct cpu_workqueue_struct *cwq;
551 struct list_head *prev;
552 struct wq_barrier barr;
553 2325
554 might_sleep(); 2326 might_sleep();
555 cwq = get_wq_data(work); 2327 gcwq = get_work_gcwq(work);
556 if (!cwq) 2328 if (!gcwq)
557 return 0; 2329 return false;
558
559 lock_map_acquire(&cwq->wq->lockdep_map);
560 lock_map_release(&cwq->wq->lockdep_map);
561 2330
562 prev = NULL; 2331 spin_lock_irq(&gcwq->lock);
563 spin_lock_irq(&cwq->lock);
564 if (!list_empty(&work->entry)) { 2332 if (!list_empty(&work->entry)) {
565 /* 2333 /*
566 * See the comment near try_to_grab_pending()->smp_rmb(). 2334 * See the comment near try_to_grab_pending()->smp_rmb().
567 * If it was re-queued under us we are not going to wait. 2335 * If it was re-queued to a different gcwq under us, we
2336 * are not going to wait.
568 */ 2337 */
569 smp_rmb(); 2338 smp_rmb();
570 if (unlikely(cwq != get_wq_data(work))) 2339 cwq = get_work_cwq(work);
571 goto out; 2340 if (unlikely(!cwq || gcwq != cwq->gcwq))
572 prev = &work->entry; 2341 goto already_gone;
573 } else { 2342 } else if (wait_executing) {
574 if (cwq->current_work != work) 2343 worker = find_worker_executing_work(gcwq, work);
575 goto out; 2344 if (!worker)
576 prev = &cwq->worklist; 2345 goto already_gone;
577 } 2346 cwq = worker->current_cwq;
578 insert_wq_barrier(cwq, &barr, prev->next); 2347 } else
579out: 2348 goto already_gone;
580 spin_unlock_irq(&cwq->lock); 2349
581 if (!prev) 2350 insert_wq_barrier(cwq, barr, work, worker);
582 return 0; 2351 spin_unlock_irq(&gcwq->lock);
583 2352
584 wait_for_completion(&barr.done); 2353 lock_map_acquire(&cwq->wq->lockdep_map);
585 destroy_work_on_stack(&barr.work); 2354 lock_map_release(&cwq->wq->lockdep_map);
586 return 1; 2355 return true;
2356already_gone:
2357 spin_unlock_irq(&gcwq->lock);
2358 return false;
2359}
2360
2361/**
2362 * flush_work - wait for a work to finish executing the last queueing instance
2363 * @work: the work to flush
2364 *
2365 * Wait until @work has finished execution. This function considers
2366 * only the last queueing instance of @work. If @work has been
2367 * enqueued across different CPUs on a non-reentrant workqueue or on
2368 * multiple workqueues, @work might still be executing on return on
2369 * some of the CPUs from earlier queueing.
2370 *
2371 * If @work was queued only on a non-reentrant, ordered or unbound
2372 * workqueue, @work is guaranteed to be idle on return if it hasn't
2373 * been requeued since flush started.
2374 *
2375 * RETURNS:
2376 * %true if flush_work() waited for the work to finish execution,
2377 * %false if it was already idle.
2378 */
2379bool flush_work(struct work_struct *work)
2380{
2381 struct wq_barrier barr;
2382
2383 if (start_flush_work(work, &barr, true)) {
2384 wait_for_completion(&barr.done);
2385 destroy_work_on_stack(&barr.work);
2386 return true;
2387 } else
2388 return false;
587} 2389}
588EXPORT_SYMBOL_GPL(flush_work); 2390EXPORT_SYMBOL_GPL(flush_work);
589 2391
2392static bool wait_on_cpu_work(struct global_cwq *gcwq, struct work_struct *work)
2393{
2394 struct wq_barrier barr;
2395 struct worker *worker;
2396
2397 spin_lock_irq(&gcwq->lock);
2398
2399 worker = find_worker_executing_work(gcwq, work);
2400 if (unlikely(worker))
2401 insert_wq_barrier(worker->current_cwq, &barr, work, worker);
2402
2403 spin_unlock_irq(&gcwq->lock);
2404
2405 if (unlikely(worker)) {
2406 wait_for_completion(&barr.done);
2407 destroy_work_on_stack(&barr.work);
2408 return true;
2409 } else
2410 return false;
2411}
2412
2413static bool wait_on_work(struct work_struct *work)
2414{
2415 bool ret = false;
2416 int cpu;
2417
2418 might_sleep();
2419
2420 lock_map_acquire(&work->lockdep_map);
2421 lock_map_release(&work->lockdep_map);
2422
2423 for_each_gcwq_cpu(cpu)
2424 ret |= wait_on_cpu_work(get_gcwq(cpu), work);
2425 return ret;
2426}
2427
2428/**
2429 * flush_work_sync - wait until a work has finished execution
2430 * @work: the work to flush
2431 *
2432 * Wait until @work has finished execution. On return, it's
2433 * guaranteed that all queueing instances of @work which happened
2434 * before this function is called are finished. In other words, if
2435 * @work hasn't been requeued since this function was called, @work is
2436 * guaranteed to be idle on return.
2437 *
2438 * RETURNS:
2439 * %true if flush_work_sync() waited for the work to finish execution,
2440 * %false if it was already idle.
2441 */
2442bool flush_work_sync(struct work_struct *work)
2443{
2444 struct wq_barrier barr;
2445 bool pending, waited;
2446
2447 /* we'll wait for executions separately, queue barr only if pending */
2448 pending = start_flush_work(work, &barr, false);
2449
2450 /* wait for executions to finish */
2451 waited = wait_on_work(work);
2452
2453 /* wait for the pending one */
2454 if (pending) {
2455 wait_for_completion(&barr.done);
2456 destroy_work_on_stack(&barr.work);
2457 }
2458
2459 return pending || waited;
2460}
2461EXPORT_SYMBOL_GPL(flush_work_sync);
2462
590/* 2463/*
591 * Upon a successful return (>= 0), the caller "owns" WORK_STRUCT_PENDING bit, 2464 * Upon a successful return (>= 0), the caller "owns" WORK_STRUCT_PENDING bit,
592 * so this work can't be re-armed in any way. 2465 * so this work can't be re-armed in any way.
593 */ 2466 */
594static int try_to_grab_pending(struct work_struct *work) 2467static int try_to_grab_pending(struct work_struct *work)
595{ 2468{
596 struct cpu_workqueue_struct *cwq; 2469 struct global_cwq *gcwq;
597 int ret = -1; 2470 int ret = -1;
598 2471
599 if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) 2472 if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work)))
600 return 0; 2473 return 0;
601 2474
602 /* 2475 /*
603 * The queueing is in progress, or it is already queued. Try to 2476 * The queueing is in progress, or it is already queued. Try to
604 * steal it from ->worklist without clearing WORK_STRUCT_PENDING. 2477 * steal it from ->worklist without clearing WORK_STRUCT_PENDING.
605 */ 2478 */
606 2479 gcwq = get_work_gcwq(work);
607 cwq = get_wq_data(work); 2480 if (!gcwq)
608 if (!cwq)
609 return ret; 2481 return ret;
610 2482
611 spin_lock_irq(&cwq->lock); 2483 spin_lock_irq(&gcwq->lock);
612 if (!list_empty(&work->entry)) { 2484 if (!list_empty(&work->entry)) {
613 /* 2485 /*
614 * This work is queued, but perhaps we locked the wrong cwq. 2486 * This work is queued, but perhaps we locked the wrong gcwq.
615 * In that case we must see the new value after rmb(), see 2487 * In that case we must see the new value after rmb(), see
616 * insert_work()->wmb(). 2488 * insert_work()->wmb().
617 */ 2489 */
618 smp_rmb(); 2490 smp_rmb();
619 if (cwq == get_wq_data(work)) { 2491 if (gcwq == get_work_gcwq(work)) {
620 debug_work_deactivate(work); 2492 debug_work_deactivate(work);
621 list_del_init(&work->entry); 2493 list_del_init(&work->entry);
2494 cwq_dec_nr_in_flight(get_work_cwq(work),
2495 get_work_color(work),
2496 *work_data_bits(work) & WORK_STRUCT_DELAYED);
622 ret = 1; 2497 ret = 1;
623 } 2498 }
624 } 2499 }
625 spin_unlock_irq(&cwq->lock); 2500 spin_unlock_irq(&gcwq->lock);
626 2501
627 return ret; 2502 return ret;
628} 2503}
629 2504
630static void wait_on_cpu_work(struct cpu_workqueue_struct *cwq, 2505static bool __cancel_work_timer(struct work_struct *work,
631 struct work_struct *work)
632{
633 struct wq_barrier barr;
634 int running = 0;
635
636 spin_lock_irq(&cwq->lock);
637 if (unlikely(cwq->current_work == work)) {
638 insert_wq_barrier(cwq, &barr, cwq->worklist.next);
639 running = 1;
640 }
641 spin_unlock_irq(&cwq->lock);
642
643 if (unlikely(running)) {
644 wait_for_completion(&barr.done);
645 destroy_work_on_stack(&barr.work);
646 }
647}
648
649static void wait_on_work(struct work_struct *work)
650{
651 struct cpu_workqueue_struct *cwq;
652 struct workqueue_struct *wq;
653 const struct cpumask *cpu_map;
654 int cpu;
655
656 might_sleep();
657
658 lock_map_acquire(&work->lockdep_map);
659 lock_map_release(&work->lockdep_map);
660
661 cwq = get_wq_data(work);
662 if (!cwq)
663 return;
664
665 wq = cwq->wq;
666 cpu_map = wq_cpu_map(wq);
667
668 for_each_cpu(cpu, cpu_map)
669 wait_on_cpu_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
670}
671
672static int __cancel_work_timer(struct work_struct *work,
673 struct timer_list* timer) 2506 struct timer_list* timer)
674{ 2507{
675 int ret; 2508 int ret;
@@ -681,54 +2514,91 @@ static int __cancel_work_timer(struct work_struct *work,
681 wait_on_work(work); 2514 wait_on_work(work);
682 } while (unlikely(ret < 0)); 2515 } while (unlikely(ret < 0));
683 2516
684 clear_wq_data(work); 2517 clear_work_data(work);
685 return ret; 2518 return ret;
686} 2519}
687 2520
688/** 2521/**
689 * cancel_work_sync - block until a work_struct's callback has terminated 2522 * cancel_work_sync - cancel a work and wait for it to finish
690 * @work: the work which is to be flushed 2523 * @work: the work to cancel
691 *
692 * Returns true if @work was pending.
693 * 2524 *
694 * cancel_work_sync() will cancel the work if it is queued. If the work's 2525 * Cancel @work and wait for its execution to finish. This function
695 * callback appears to be running, cancel_work_sync() will block until it 2526 * can be used even if the work re-queues itself or migrates to
696 * has completed. 2527 * another workqueue. On return from this function, @work is
2528 * guaranteed to be not pending or executing on any CPU.
697 * 2529 *
698 * It is possible to use this function if the work re-queues itself. It can 2530 * cancel_work_sync(&delayed_work->work) must not be used for
699 * cancel the work even if it migrates to another workqueue, however in that 2531 * delayed_work's. Use cancel_delayed_work_sync() instead.
700 * case it only guarantees that work->func() has completed on the last queued
701 * workqueue.
702 *
703 * cancel_work_sync(&delayed_work->work) should be used only if ->timer is not
704 * pending, otherwise it goes into a busy-wait loop until the timer expires.
705 * 2532 *
706 * The caller must ensure that workqueue_struct on which this work was last 2533 * The caller must ensure that the workqueue on which @work was last
707 * queued can't be destroyed before this function returns. 2534 * queued can't be destroyed before this function returns.
2535 *
2536 * RETURNS:
2537 * %true if @work was pending, %false otherwise.
708 */ 2538 */
709int cancel_work_sync(struct work_struct *work) 2539bool cancel_work_sync(struct work_struct *work)
710{ 2540{
711 return __cancel_work_timer(work, NULL); 2541 return __cancel_work_timer(work, NULL);
712} 2542}
713EXPORT_SYMBOL_GPL(cancel_work_sync); 2543EXPORT_SYMBOL_GPL(cancel_work_sync);
714 2544
715/** 2545/**
716 * cancel_delayed_work_sync - reliably kill off a delayed work. 2546 * flush_delayed_work - wait for a dwork to finish executing the last queueing
717 * @dwork: the delayed work struct 2547 * @dwork: the delayed work to flush
718 * 2548 *
719 * Returns true if @dwork was pending. 2549 * Delayed timer is cancelled and the pending work is queued for
2550 * immediate execution. Like flush_work(), this function only
2551 * considers the last queueing instance of @dwork.
720 * 2552 *
721 * It is possible to use this function if @dwork rearms itself via queue_work() 2553 * RETURNS:
722 * or queue_delayed_work(). See also the comment for cancel_work_sync(). 2554 * %true if flush_work() waited for the work to finish execution,
2555 * %false if it was already idle.
723 */ 2556 */
724int cancel_delayed_work_sync(struct delayed_work *dwork) 2557bool flush_delayed_work(struct delayed_work *dwork)
2558{
2559 if (del_timer_sync(&dwork->timer))
2560 __queue_work(raw_smp_processor_id(),
2561 get_work_cwq(&dwork->work)->wq, &dwork->work);
2562 return flush_work(&dwork->work);
2563}
2564EXPORT_SYMBOL(flush_delayed_work);
2565
2566/**
2567 * flush_delayed_work_sync - wait for a dwork to finish
2568 * @dwork: the delayed work to flush
2569 *
2570 * Delayed timer is cancelled and the pending work is queued for
2571 * execution immediately. Other than timer handling, its behavior
2572 * is identical to flush_work_sync().
2573 *
2574 * RETURNS:
2575 * %true if flush_work_sync() waited for the work to finish execution,
2576 * %false if it was already idle.
2577 */
2578bool flush_delayed_work_sync(struct delayed_work *dwork)
2579{
2580 if (del_timer_sync(&dwork->timer))
2581 __queue_work(raw_smp_processor_id(),
2582 get_work_cwq(&dwork->work)->wq, &dwork->work);
2583 return flush_work_sync(&dwork->work);
2584}
2585EXPORT_SYMBOL(flush_delayed_work_sync);
2586
2587/**
2588 * cancel_delayed_work_sync - cancel a delayed work and wait for it to finish
2589 * @dwork: the delayed work cancel
2590 *
2591 * This is cancel_work_sync() for delayed works.
2592 *
2593 * RETURNS:
2594 * %true if @dwork was pending, %false otherwise.
2595 */
2596bool cancel_delayed_work_sync(struct delayed_work *dwork)
725{ 2597{
726 return __cancel_work_timer(&dwork->work, &dwork->timer); 2598 return __cancel_work_timer(&dwork->work, &dwork->timer);
727} 2599}
728EXPORT_SYMBOL(cancel_delayed_work_sync); 2600EXPORT_SYMBOL(cancel_delayed_work_sync);
729 2601
730static struct workqueue_struct *keventd_wq __read_mostly;
731
732/** 2602/**
733 * schedule_work - put work task in global workqueue 2603 * schedule_work - put work task in global workqueue
734 * @work: job to be done 2604 * @work: job to be done
@@ -742,7 +2612,7 @@ static struct workqueue_struct *keventd_wq __read_mostly;
742 */ 2612 */
743int schedule_work(struct work_struct *work) 2613int schedule_work(struct work_struct *work)
744{ 2614{
745 return queue_work(keventd_wq, work); 2615 return queue_work(system_wq, work);
746} 2616}
747EXPORT_SYMBOL(schedule_work); 2617EXPORT_SYMBOL(schedule_work);
748 2618
@@ -755,7 +2625,7 @@ EXPORT_SYMBOL(schedule_work);
755 */ 2625 */
756int schedule_work_on(int cpu, struct work_struct *work) 2626int schedule_work_on(int cpu, struct work_struct *work)
757{ 2627{
758 return queue_work_on(cpu, keventd_wq, work); 2628 return queue_work_on(cpu, system_wq, work);
759} 2629}
760EXPORT_SYMBOL(schedule_work_on); 2630EXPORT_SYMBOL(schedule_work_on);
761 2631
@@ -770,29 +2640,11 @@ EXPORT_SYMBOL(schedule_work_on);
770int schedule_delayed_work(struct delayed_work *dwork, 2640int schedule_delayed_work(struct delayed_work *dwork,
771 unsigned long delay) 2641 unsigned long delay)
772{ 2642{
773 return queue_delayed_work(keventd_wq, dwork, delay); 2643 return queue_delayed_work(system_wq, dwork, delay);
774} 2644}
775EXPORT_SYMBOL(schedule_delayed_work); 2645EXPORT_SYMBOL(schedule_delayed_work);
776 2646
777/** 2647/**
778 * flush_delayed_work - block until a dwork_struct's callback has terminated
779 * @dwork: the delayed work which is to be flushed
780 *
781 * Any timeout is cancelled, and any pending work is run immediately.
782 */
783void flush_delayed_work(struct delayed_work *dwork)
784{
785 if (del_timer_sync(&dwork->timer)) {
786 struct cpu_workqueue_struct *cwq;
787 cwq = wq_per_cpu(get_wq_data(&dwork->work)->wq, get_cpu());
788 __queue_work(cwq, &dwork->work);
789 put_cpu();
790 }
791 flush_work(&dwork->work);
792}
793EXPORT_SYMBOL(flush_delayed_work);
794
795/**
796 * schedule_delayed_work_on - queue work in global workqueue on CPU after delay 2648 * schedule_delayed_work_on - queue work in global workqueue on CPU after delay
797 * @cpu: cpu to use 2649 * @cpu: cpu to use
798 * @dwork: job to be done 2650 * @dwork: job to be done
@@ -804,24 +2656,25 @@ EXPORT_SYMBOL(flush_delayed_work);
804int schedule_delayed_work_on(int cpu, 2656int schedule_delayed_work_on(int cpu,
805 struct delayed_work *dwork, unsigned long delay) 2657 struct delayed_work *dwork, unsigned long delay)
806{ 2658{
807 return queue_delayed_work_on(cpu, keventd_wq, dwork, delay); 2659 return queue_delayed_work_on(cpu, system_wq, dwork, delay);
808} 2660}
809EXPORT_SYMBOL(schedule_delayed_work_on); 2661EXPORT_SYMBOL(schedule_delayed_work_on);
810 2662
811/** 2663/**
812 * schedule_on_each_cpu - call a function on each online CPU from keventd 2664 * schedule_on_each_cpu - execute a function synchronously on each online CPU
813 * @func: the function to call 2665 * @func: the function to call
814 * 2666 *
815 * Returns zero on success. 2667 * schedule_on_each_cpu() executes @func on each online CPU using the
816 * Returns -ve errno on failure. 2668 * system workqueue and blocks until all CPUs have completed.
817 *
818 * schedule_on_each_cpu() is very slow. 2669 * schedule_on_each_cpu() is very slow.
2670 *
2671 * RETURNS:
2672 * 0 on success, -errno on failure.
819 */ 2673 */
820int schedule_on_each_cpu(work_func_t func) 2674int schedule_on_each_cpu(work_func_t func)
821{ 2675{
822 int cpu; 2676 int cpu;
823 int orig = -1; 2677 struct work_struct __percpu *works;
824 struct work_struct *works;
825 2678
826 works = alloc_percpu(struct work_struct); 2679 works = alloc_percpu(struct work_struct);
827 if (!works) 2680 if (!works)
@@ -829,23 +2682,12 @@ int schedule_on_each_cpu(work_func_t func)
829 2682
830 get_online_cpus(); 2683 get_online_cpus();
831 2684
832 /*
833 * When running in keventd don't schedule a work item on
834 * itself. Can just call directly because the work queue is
835 * already bound. This also is faster.
836 */
837 if (current_is_keventd())
838 orig = raw_smp_processor_id();
839
840 for_each_online_cpu(cpu) { 2685 for_each_online_cpu(cpu) {
841 struct work_struct *work = per_cpu_ptr(works, cpu); 2686 struct work_struct *work = per_cpu_ptr(works, cpu);
842 2687
843 INIT_WORK(work, func); 2688 INIT_WORK(work, func);
844 if (cpu != orig) 2689 schedule_work_on(cpu, work);
845 schedule_work_on(cpu, work);
846 } 2690 }
847 if (orig >= 0)
848 func(per_cpu_ptr(works, orig));
849 2691
850 for_each_online_cpu(cpu) 2692 for_each_online_cpu(cpu)
851 flush_work(per_cpu_ptr(works, cpu)); 2693 flush_work(per_cpu_ptr(works, cpu));
@@ -881,7 +2723,7 @@ int schedule_on_each_cpu(work_func_t func)
881 */ 2723 */
882void flush_scheduled_work(void) 2724void flush_scheduled_work(void)
883{ 2725{
884 flush_workqueue(keventd_wq); 2726 flush_workqueue(system_wq);
885} 2727}
886EXPORT_SYMBOL(flush_scheduled_work); 2728EXPORT_SYMBOL(flush_scheduled_work);
887 2729
@@ -913,170 +2755,178 @@ EXPORT_SYMBOL_GPL(execute_in_process_context);
913 2755
914int keventd_up(void) 2756int keventd_up(void)
915{ 2757{
916 return keventd_wq != NULL; 2758 return system_wq != NULL;
917} 2759}
918 2760
919int current_is_keventd(void) 2761static int alloc_cwqs(struct workqueue_struct *wq)
920{ 2762{
921 struct cpu_workqueue_struct *cwq; 2763 /*
922 int cpu = raw_smp_processor_id(); /* preempt-safe: keventd is per-cpu */ 2764 * cwqs are forced aligned according to WORK_STRUCT_FLAG_BITS.
923 int ret = 0; 2765 * Make sure that the alignment isn't lower than that of
924 2766 * unsigned long long.
925 BUG_ON(!keventd_wq); 2767 */
2768 const size_t size = sizeof(struct cpu_workqueue_struct);
2769 const size_t align = max_t(size_t, 1 << WORK_STRUCT_FLAG_BITS,
2770 __alignof__(unsigned long long));
2771#ifdef CONFIG_SMP
2772 bool percpu = !(wq->flags & WQ_UNBOUND);
2773#else
2774 bool percpu = false;
2775#endif
926 2776
927 cwq = per_cpu_ptr(keventd_wq->cpu_wq, cpu); 2777 if (percpu)
928 if (current == cwq->thread) 2778 wq->cpu_wq.pcpu = __alloc_percpu(size, align);
929 ret = 1; 2779 else {
2780 void *ptr;
930 2781
931 return ret; 2782 /*
2783 * Allocate enough room to align cwq and put an extra
2784 * pointer at the end pointing back to the originally
2785 * allocated pointer which will be used for free.
2786 */
2787 ptr = kzalloc(size + align + sizeof(void *), GFP_KERNEL);
2788 if (ptr) {
2789 wq->cpu_wq.single = PTR_ALIGN(ptr, align);
2790 *(void **)(wq->cpu_wq.single + 1) = ptr;
2791 }
2792 }
932 2793
2794 /* just in case, make sure it's actually aligned
2795 * - this is affected by PERCPU() alignment in vmlinux.lds.S
2796 */
2797 BUG_ON(!IS_ALIGNED(wq->cpu_wq.v, align));
2798 return wq->cpu_wq.v ? 0 : -ENOMEM;
933} 2799}
934 2800
935static struct cpu_workqueue_struct * 2801static void free_cwqs(struct workqueue_struct *wq)
936init_cpu_workqueue(struct workqueue_struct *wq, int cpu)
937{ 2802{
938 struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu); 2803#ifdef CONFIG_SMP
939 2804 bool percpu = !(wq->flags & WQ_UNBOUND);
940 cwq->wq = wq; 2805#else
941 spin_lock_init(&cwq->lock); 2806 bool percpu = false;
942 INIT_LIST_HEAD(&cwq->worklist); 2807#endif
943 init_waitqueue_head(&cwq->more_work);
944 2808
945 return cwq; 2809 if (percpu)
2810 free_percpu(wq->cpu_wq.pcpu);
2811 else if (wq->cpu_wq.single) {
2812 /* the pointer to free is stored right after the cwq */
2813 kfree(*(void **)(wq->cpu_wq.single + 1));
2814 }
946} 2815}
947 2816
948static int create_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu) 2817static int wq_clamp_max_active(int max_active, unsigned int flags,
2818 const char *name)
949{ 2819{
950 struct sched_param param = { .sched_priority = MAX_RT_PRIO-1 }; 2820 int lim = flags & WQ_UNBOUND ? WQ_UNBOUND_MAX_ACTIVE : WQ_MAX_ACTIVE;
951 struct workqueue_struct *wq = cwq->wq;
952 const char *fmt = is_wq_single_threaded(wq) ? "%s" : "%s/%d";
953 struct task_struct *p;
954 2821
955 p = kthread_create(worker_thread, cwq, fmt, wq->name, cpu); 2822 if (max_active < 1 || max_active > lim)
956 /* 2823 printk(KERN_WARNING "workqueue: max_active %d requested for %s "
957 * Nobody can add the work_struct to this cwq, 2824 "is out of range, clamping between %d and %d\n",
958 * if (caller is __create_workqueue) 2825 max_active, name, 1, lim);
959 * nobody should see this wq
960 * else // caller is CPU_UP_PREPARE
961 * cpu is not on cpu_online_map
962 * so we can abort safely.
963 */
964 if (IS_ERR(p))
965 return PTR_ERR(p);
966 if (cwq->wq->rt)
967 sched_setscheduler_nocheck(p, SCHED_FIFO, &param);
968 cwq->thread = p;
969 2826
970 trace_workqueue_creation(cwq->thread, cpu); 2827 return clamp_val(max_active, 1, lim);
971
972 return 0;
973} 2828}
974 2829
975static void start_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu) 2830struct workqueue_struct *__alloc_workqueue_key(const char *name,
2831 unsigned int flags,
2832 int max_active,
2833 struct lock_class_key *key,
2834 const char *lock_name)
976{ 2835{
977 struct task_struct *p = cwq->thread; 2836 struct workqueue_struct *wq;
2837 unsigned int cpu;
978 2838
979 if (p != NULL) { 2839 /*
980 if (cpu >= 0) 2840 * Workqueues which may be used during memory reclaim should
981 kthread_bind(p, cpu); 2841 * have a rescuer to guarantee forward progress.
982 wake_up_process(p); 2842 */
983 } 2843 if (flags & WQ_MEM_RECLAIM)
984} 2844 flags |= WQ_RESCUER;
985 2845
986struct workqueue_struct *__create_workqueue_key(const char *name, 2846 /*
987 int singlethread, 2847 * Unbound workqueues aren't concurrency managed and should be
988 int freezeable, 2848 * dispatched to workers immediately.
989 int rt, 2849 */
990 struct lock_class_key *key, 2850 if (flags & WQ_UNBOUND)
991 const char *lock_name) 2851 flags |= WQ_HIGHPRI;
992{ 2852
993 struct workqueue_struct *wq; 2853 max_active = max_active ?: WQ_DFL_ACTIVE;
994 struct cpu_workqueue_struct *cwq; 2854 max_active = wq_clamp_max_active(max_active, flags, name);
995 int err = 0, cpu;
996 2855
997 wq = kzalloc(sizeof(*wq), GFP_KERNEL); 2856 wq = kzalloc(sizeof(*wq), GFP_KERNEL);
998 if (!wq) 2857 if (!wq)
999 return NULL; 2858 goto err;
1000 2859
1001 wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct); 2860 wq->flags = flags;
1002 if (!wq->cpu_wq) { 2861 wq->saved_max_active = max_active;
1003 kfree(wq); 2862 mutex_init(&wq->flush_mutex);
1004 return NULL; 2863 atomic_set(&wq->nr_cwqs_to_flush, 0);
1005 } 2864 INIT_LIST_HEAD(&wq->flusher_queue);
2865 INIT_LIST_HEAD(&wq->flusher_overflow);
1006 2866
1007 wq->name = name; 2867 wq->name = name;
1008 lockdep_init_map(&wq->lockdep_map, lock_name, key, 0); 2868 lockdep_init_map(&wq->lockdep_map, lock_name, key, 0);
1009 wq->singlethread = singlethread;
1010 wq->freezeable = freezeable;
1011 wq->rt = rt;
1012 INIT_LIST_HEAD(&wq->list); 2869 INIT_LIST_HEAD(&wq->list);
1013 2870
1014 if (singlethread) { 2871 if (alloc_cwqs(wq) < 0)
1015 cwq = init_cpu_workqueue(wq, singlethread_cpu); 2872 goto err;
1016 err = create_workqueue_thread(cwq, singlethread_cpu); 2873
1017 start_workqueue_thread(cwq, -1); 2874 for_each_cwq_cpu(cpu, wq) {
1018 } else { 2875 struct cpu_workqueue_struct *cwq = get_cwq(cpu, wq);
1019 cpu_maps_update_begin(); 2876 struct global_cwq *gcwq = get_gcwq(cpu);
1020 /* 2877
1021 * We must place this wq on list even if the code below fails. 2878 BUG_ON((unsigned long)cwq & WORK_STRUCT_FLAG_MASK);
1022 * cpu_down(cpu) can remove cpu from cpu_populated_map before 2879 cwq->gcwq = gcwq;
1023 * destroy_workqueue() takes the lock, in that case we leak 2880 cwq->wq = wq;
1024 * cwq[cpu]->thread. 2881 cwq->flush_color = -1;
1025 */ 2882 cwq->max_active = max_active;
1026 spin_lock(&workqueue_lock); 2883 INIT_LIST_HEAD(&cwq->delayed_works);
1027 list_add(&wq->list, &workqueues);
1028 spin_unlock(&workqueue_lock);
1029 /*
1030 * We must initialize cwqs for each possible cpu even if we
1031 * are going to call destroy_workqueue() finally. Otherwise
1032 * cpu_up() can hit the uninitialized cwq once we drop the
1033 * lock.
1034 */
1035 for_each_possible_cpu(cpu) {
1036 cwq = init_cpu_workqueue(wq, cpu);
1037 if (err || !cpu_online(cpu))
1038 continue;
1039 err = create_workqueue_thread(cwq, cpu);
1040 start_workqueue_thread(cwq, cpu);
1041 }
1042 cpu_maps_update_done();
1043 } 2884 }
1044 2885
1045 if (err) { 2886 if (flags & WQ_RESCUER) {
1046 destroy_workqueue(wq); 2887 struct worker *rescuer;
1047 wq = NULL; 2888
2889 if (!alloc_mayday_mask(&wq->mayday_mask, GFP_KERNEL))
2890 goto err;
2891
2892 wq->rescuer = rescuer = alloc_worker();
2893 if (!rescuer)
2894 goto err;
2895
2896 rescuer->task = kthread_create(rescuer_thread, wq, "%s", name);
2897 if (IS_ERR(rescuer->task))
2898 goto err;
2899
2900 rescuer->task->flags |= PF_THREAD_BOUND;
2901 wake_up_process(rescuer->task);
1048 } 2902 }
1049 return wq;
1050}
1051EXPORT_SYMBOL_GPL(__create_workqueue_key);
1052 2903
1053static void cleanup_workqueue_thread(struct cpu_workqueue_struct *cwq)
1054{
1055 /* 2904 /*
1056 * Our caller is either destroy_workqueue() or CPU_POST_DEAD, 2905 * workqueue_lock protects global freeze state and workqueues
1057 * cpu_add_remove_lock protects cwq->thread. 2906 * list. Grab it, set max_active accordingly and add the new
2907 * workqueue to workqueues list.
1058 */ 2908 */
1059 if (cwq->thread == NULL) 2909 spin_lock(&workqueue_lock);
1060 return;
1061 2910
1062 lock_map_acquire(&cwq->wq->lockdep_map); 2911 if (workqueue_freezing && wq->flags & WQ_FREEZEABLE)
1063 lock_map_release(&cwq->wq->lockdep_map); 2912 for_each_cwq_cpu(cpu, wq)
2913 get_cwq(cpu, wq)->max_active = 0;
1064 2914
1065 flush_cpu_workqueue(cwq); 2915 list_add(&wq->list, &workqueues);
1066 /* 2916
1067 * If the caller is CPU_POST_DEAD and cwq->worklist was not empty, 2917 spin_unlock(&workqueue_lock);
1068 * a concurrent flush_workqueue() can insert a barrier after us. 2918
1069 * However, in that case run_workqueue() won't return and check 2919 return wq;
1070 * kthread_should_stop() until it flushes all work_struct's. 2920err:
1071 * When ->worklist becomes empty it is safe to exit because no 2921 if (wq) {
1072 * more work_structs can be queued on this cwq: flush_workqueue 2922 free_cwqs(wq);
1073 * checks list_empty(), and a "normal" queue_work() can't use 2923 free_mayday_mask(wq->mayday_mask);
1074 * a dead CPU. 2924 kfree(wq->rescuer);
1075 */ 2925 kfree(wq);
1076 trace_workqueue_destruction(cwq->thread); 2926 }
1077 kthread_stop(cwq->thread); 2927 return NULL;
1078 cwq->thread = NULL;
1079} 2928}
2929EXPORT_SYMBOL_GPL(__alloc_workqueue_key);
1080 2930
1081/** 2931/**
1082 * destroy_workqueue - safely terminate a workqueue 2932 * destroy_workqueue - safely terminate a workqueue
@@ -1086,72 +2936,520 @@ static void cleanup_workqueue_thread(struct cpu_workqueue_struct *cwq)
1086 */ 2936 */
1087void destroy_workqueue(struct workqueue_struct *wq) 2937void destroy_workqueue(struct workqueue_struct *wq)
1088{ 2938{
1089 const struct cpumask *cpu_map = wq_cpu_map(wq); 2939 unsigned int cpu;
1090 int cpu; 2940
2941 wq->flags |= WQ_DYING;
2942 flush_workqueue(wq);
1091 2943
1092 cpu_maps_update_begin(); 2944 /*
2945 * wq list is used to freeze wq, remove from list after
2946 * flushing is complete in case freeze races us.
2947 */
1093 spin_lock(&workqueue_lock); 2948 spin_lock(&workqueue_lock);
1094 list_del(&wq->list); 2949 list_del(&wq->list);
1095 spin_unlock(&workqueue_lock); 2950 spin_unlock(&workqueue_lock);
1096 2951
1097 for_each_cpu(cpu, cpu_map) 2952 /* sanity check */
1098 cleanup_workqueue_thread(per_cpu_ptr(wq->cpu_wq, cpu)); 2953 for_each_cwq_cpu(cpu, wq) {
1099 cpu_maps_update_done(); 2954 struct cpu_workqueue_struct *cwq = get_cwq(cpu, wq);
2955 int i;
2956
2957 for (i = 0; i < WORK_NR_COLORS; i++)
2958 BUG_ON(cwq->nr_in_flight[i]);
2959 BUG_ON(cwq->nr_active);
2960 BUG_ON(!list_empty(&cwq->delayed_works));
2961 }
2962
2963 if (wq->flags & WQ_RESCUER) {
2964 kthread_stop(wq->rescuer->task);
2965 free_mayday_mask(wq->mayday_mask);
2966 kfree(wq->rescuer);
2967 }
1100 2968
1101 free_percpu(wq->cpu_wq); 2969 free_cwqs(wq);
1102 kfree(wq); 2970 kfree(wq);
1103} 2971}
1104EXPORT_SYMBOL_GPL(destroy_workqueue); 2972EXPORT_SYMBOL_GPL(destroy_workqueue);
1105 2973
2974/**
2975 * workqueue_set_max_active - adjust max_active of a workqueue
2976 * @wq: target workqueue
2977 * @max_active: new max_active value.
2978 *
2979 * Set max_active of @wq to @max_active.
2980 *
2981 * CONTEXT:
2982 * Don't call from IRQ context.
2983 */
2984void workqueue_set_max_active(struct workqueue_struct *wq, int max_active)
2985{
2986 unsigned int cpu;
2987
2988 max_active = wq_clamp_max_active(max_active, wq->flags, wq->name);
2989
2990 spin_lock(&workqueue_lock);
2991
2992 wq->saved_max_active = max_active;
2993
2994 for_each_cwq_cpu(cpu, wq) {
2995 struct global_cwq *gcwq = get_gcwq(cpu);
2996
2997 spin_lock_irq(&gcwq->lock);
2998
2999 if (!(wq->flags & WQ_FREEZEABLE) ||
3000 !(gcwq->flags & GCWQ_FREEZING))
3001 get_cwq(gcwq->cpu, wq)->max_active = max_active;
3002
3003 spin_unlock_irq(&gcwq->lock);
3004 }
3005
3006 spin_unlock(&workqueue_lock);
3007}
3008EXPORT_SYMBOL_GPL(workqueue_set_max_active);
3009
3010/**
3011 * workqueue_congested - test whether a workqueue is congested
3012 * @cpu: CPU in question
3013 * @wq: target workqueue
3014 *
3015 * Test whether @wq's cpu workqueue for @cpu is congested. There is
3016 * no synchronization around this function and the test result is
3017 * unreliable and only useful as advisory hints or for debugging.
3018 *
3019 * RETURNS:
3020 * %true if congested, %false otherwise.
3021 */
3022bool workqueue_congested(unsigned int cpu, struct workqueue_struct *wq)
3023{
3024 struct cpu_workqueue_struct *cwq = get_cwq(cpu, wq);
3025
3026 return !list_empty(&cwq->delayed_works);
3027}
3028EXPORT_SYMBOL_GPL(workqueue_congested);
3029
3030/**
3031 * work_cpu - return the last known associated cpu for @work
3032 * @work: the work of interest
3033 *
3034 * RETURNS:
3035 * CPU number if @work was ever queued. WORK_CPU_NONE otherwise.
3036 */
3037unsigned int work_cpu(struct work_struct *work)
3038{
3039 struct global_cwq *gcwq = get_work_gcwq(work);
3040
3041 return gcwq ? gcwq->cpu : WORK_CPU_NONE;
3042}
3043EXPORT_SYMBOL_GPL(work_cpu);
3044
3045/**
3046 * work_busy - test whether a work is currently pending or running
3047 * @work: the work to be tested
3048 *
3049 * Test whether @work is currently pending or running. There is no
3050 * synchronization around this function and the test result is
3051 * unreliable and only useful as advisory hints or for debugging.
3052 * Especially for reentrant wqs, the pending state might hide the
3053 * running state.
3054 *
3055 * RETURNS:
3056 * OR'd bitmask of WORK_BUSY_* bits.
3057 */
3058unsigned int work_busy(struct work_struct *work)
3059{
3060 struct global_cwq *gcwq = get_work_gcwq(work);
3061 unsigned long flags;
3062 unsigned int ret = 0;
3063
3064 if (!gcwq)
3065 return false;
3066
3067 spin_lock_irqsave(&gcwq->lock, flags);
3068
3069 if (work_pending(work))
3070 ret |= WORK_BUSY_PENDING;
3071 if (find_worker_executing_work(gcwq, work))
3072 ret |= WORK_BUSY_RUNNING;
3073
3074 spin_unlock_irqrestore(&gcwq->lock, flags);
3075
3076 return ret;
3077}
3078EXPORT_SYMBOL_GPL(work_busy);
3079
3080/*
3081 * CPU hotplug.
3082 *
3083 * There are two challenges in supporting CPU hotplug. Firstly, there
3084 * are a lot of assumptions on strong associations among work, cwq and
3085 * gcwq which make migrating pending and scheduled works very
3086 * difficult to implement without impacting hot paths. Secondly,
3087 * gcwqs serve mix of short, long and very long running works making
3088 * blocked draining impractical.
3089 *
3090 * This is solved by allowing a gcwq to be detached from CPU, running
3091 * it with unbound (rogue) workers and allowing it to be reattached
3092 * later if the cpu comes back online. A separate thread is created
3093 * to govern a gcwq in such state and is called the trustee of the
3094 * gcwq.
3095 *
3096 * Trustee states and their descriptions.
3097 *
3098 * START Command state used on startup. On CPU_DOWN_PREPARE, a
3099 * new trustee is started with this state.
3100 *
3101 * IN_CHARGE Once started, trustee will enter this state after
3102 * assuming the manager role and making all existing
3103 * workers rogue. DOWN_PREPARE waits for trustee to
3104 * enter this state. After reaching IN_CHARGE, trustee
3105 * tries to execute the pending worklist until it's empty
3106 * and the state is set to BUTCHER, or the state is set
3107 * to RELEASE.
3108 *
3109 * BUTCHER Command state which is set by the cpu callback after
3110 * the cpu has went down. Once this state is set trustee
3111 * knows that there will be no new works on the worklist
3112 * and once the worklist is empty it can proceed to
3113 * killing idle workers.
3114 *
3115 * RELEASE Command state which is set by the cpu callback if the
3116 * cpu down has been canceled or it has come online
3117 * again. After recognizing this state, trustee stops
3118 * trying to drain or butcher and clears ROGUE, rebinds
3119 * all remaining workers back to the cpu and releases
3120 * manager role.
3121 *
3122 * DONE Trustee will enter this state after BUTCHER or RELEASE
3123 * is complete.
3124 *
3125 * trustee CPU draining
3126 * took over down complete
3127 * START -----------> IN_CHARGE -----------> BUTCHER -----------> DONE
3128 * | | ^
3129 * | CPU is back online v return workers |
3130 * ----------------> RELEASE --------------
3131 */
3132
3133/**
3134 * trustee_wait_event_timeout - timed event wait for trustee
3135 * @cond: condition to wait for
3136 * @timeout: timeout in jiffies
3137 *
3138 * wait_event_timeout() for trustee to use. Handles locking and
3139 * checks for RELEASE request.
3140 *
3141 * CONTEXT:
3142 * spin_lock_irq(gcwq->lock) which may be released and regrabbed
3143 * multiple times. To be used by trustee.
3144 *
3145 * RETURNS:
3146 * Positive indicating left time if @cond is satisfied, 0 if timed
3147 * out, -1 if canceled.
3148 */
3149#define trustee_wait_event_timeout(cond, timeout) ({ \
3150 long __ret = (timeout); \
3151 while (!((cond) || (gcwq->trustee_state == TRUSTEE_RELEASE)) && \
3152 __ret) { \
3153 spin_unlock_irq(&gcwq->lock); \
3154 __wait_event_timeout(gcwq->trustee_wait, (cond) || \
3155 (gcwq->trustee_state == TRUSTEE_RELEASE), \
3156 __ret); \
3157 spin_lock_irq(&gcwq->lock); \
3158 } \
3159 gcwq->trustee_state == TRUSTEE_RELEASE ? -1 : (__ret); \
3160})
3161
3162/**
3163 * trustee_wait_event - event wait for trustee
3164 * @cond: condition to wait for
3165 *
3166 * wait_event() for trustee to use. Automatically handles locking and
3167 * checks for CANCEL request.
3168 *
3169 * CONTEXT:
3170 * spin_lock_irq(gcwq->lock) which may be released and regrabbed
3171 * multiple times. To be used by trustee.
3172 *
3173 * RETURNS:
3174 * 0 if @cond is satisfied, -1 if canceled.
3175 */
3176#define trustee_wait_event(cond) ({ \
3177 long __ret1; \
3178 __ret1 = trustee_wait_event_timeout(cond, MAX_SCHEDULE_TIMEOUT);\
3179 __ret1 < 0 ? -1 : 0; \
3180})
3181
3182static int __cpuinit trustee_thread(void *__gcwq)
3183{
3184 struct global_cwq *gcwq = __gcwq;
3185 struct worker *worker;
3186 struct work_struct *work;
3187 struct hlist_node *pos;
3188 long rc;
3189 int i;
3190
3191 BUG_ON(gcwq->cpu != smp_processor_id());
3192
3193 spin_lock_irq(&gcwq->lock);
3194 /*
3195 * Claim the manager position and make all workers rogue.
3196 * Trustee must be bound to the target cpu and can't be
3197 * cancelled.
3198 */
3199 BUG_ON(gcwq->cpu != smp_processor_id());
3200 rc = trustee_wait_event(!(gcwq->flags & GCWQ_MANAGING_WORKERS));
3201 BUG_ON(rc < 0);
3202
3203 gcwq->flags |= GCWQ_MANAGING_WORKERS;
3204
3205 list_for_each_entry(worker, &gcwq->idle_list, entry)
3206 worker->flags |= WORKER_ROGUE;
3207
3208 for_each_busy_worker(worker, i, pos, gcwq)
3209 worker->flags |= WORKER_ROGUE;
3210
3211 /*
3212 * Call schedule() so that we cross rq->lock and thus can
3213 * guarantee sched callbacks see the rogue flag. This is
3214 * necessary as scheduler callbacks may be invoked from other
3215 * cpus.
3216 */
3217 spin_unlock_irq(&gcwq->lock);
3218 schedule();
3219 spin_lock_irq(&gcwq->lock);
3220
3221 /*
3222 * Sched callbacks are disabled now. Zap nr_running. After
3223 * this, nr_running stays zero and need_more_worker() and
3224 * keep_working() are always true as long as the worklist is
3225 * not empty.
3226 */
3227 atomic_set(get_gcwq_nr_running(gcwq->cpu), 0);
3228
3229 spin_unlock_irq(&gcwq->lock);
3230 del_timer_sync(&gcwq->idle_timer);
3231 spin_lock_irq(&gcwq->lock);
3232
3233 /*
3234 * We're now in charge. Notify and proceed to drain. We need
3235 * to keep the gcwq running during the whole CPU down
3236 * procedure as other cpu hotunplug callbacks may need to
3237 * flush currently running tasks.
3238 */
3239 gcwq->trustee_state = TRUSTEE_IN_CHARGE;
3240 wake_up_all(&gcwq->trustee_wait);
3241
3242 /*
3243 * The original cpu is in the process of dying and may go away
3244 * anytime now. When that happens, we and all workers would
3245 * be migrated to other cpus. Try draining any left work. We
3246 * want to get it over with ASAP - spam rescuers, wake up as
3247 * many idlers as necessary and create new ones till the
3248 * worklist is empty. Note that if the gcwq is frozen, there
3249 * may be frozen works in freezeable cwqs. Don't declare
3250 * completion while frozen.
3251 */
3252 while (gcwq->nr_workers != gcwq->nr_idle ||
3253 gcwq->flags & GCWQ_FREEZING ||
3254 gcwq->trustee_state == TRUSTEE_IN_CHARGE) {
3255 int nr_works = 0;
3256
3257 list_for_each_entry(work, &gcwq->worklist, entry) {
3258 send_mayday(work);
3259 nr_works++;
3260 }
3261
3262 list_for_each_entry(worker, &gcwq->idle_list, entry) {
3263 if (!nr_works--)
3264 break;
3265 wake_up_process(worker->task);
3266 }
3267
3268 if (need_to_create_worker(gcwq)) {
3269 spin_unlock_irq(&gcwq->lock);
3270 worker = create_worker(gcwq, false);
3271 spin_lock_irq(&gcwq->lock);
3272 if (worker) {
3273 worker->flags |= WORKER_ROGUE;
3274 start_worker(worker);
3275 }
3276 }
3277
3278 /* give a breather */
3279 if (trustee_wait_event_timeout(false, TRUSTEE_COOLDOWN) < 0)
3280 break;
3281 }
3282
3283 /*
3284 * Either all works have been scheduled and cpu is down, or
3285 * cpu down has already been canceled. Wait for and butcher
3286 * all workers till we're canceled.
3287 */
3288 do {
3289 rc = trustee_wait_event(!list_empty(&gcwq->idle_list));
3290 while (!list_empty(&gcwq->idle_list))
3291 destroy_worker(list_first_entry(&gcwq->idle_list,
3292 struct worker, entry));
3293 } while (gcwq->nr_workers && rc >= 0);
3294
3295 /*
3296 * At this point, either draining has completed and no worker
3297 * is left, or cpu down has been canceled or the cpu is being
3298 * brought back up. There shouldn't be any idle one left.
3299 * Tell the remaining busy ones to rebind once it finishes the
3300 * currently scheduled works by scheduling the rebind_work.
3301 */
3302 WARN_ON(!list_empty(&gcwq->idle_list));
3303
3304 for_each_busy_worker(worker, i, pos, gcwq) {
3305 struct work_struct *rebind_work = &worker->rebind_work;
3306
3307 /*
3308 * Rebind_work may race with future cpu hotplug
3309 * operations. Use a separate flag to mark that
3310 * rebinding is scheduled.
3311 */
3312 worker->flags |= WORKER_REBIND;
3313 worker->flags &= ~WORKER_ROGUE;
3314
3315 /* queue rebind_work, wq doesn't matter, use the default one */
3316 if (test_and_set_bit(WORK_STRUCT_PENDING_BIT,
3317 work_data_bits(rebind_work)))
3318 continue;
3319
3320 debug_work_activate(rebind_work);
3321 insert_work(get_cwq(gcwq->cpu, system_wq), rebind_work,
3322 worker->scheduled.next,
3323 work_color_to_flags(WORK_NO_COLOR));
3324 }
3325
3326 /* relinquish manager role */
3327 gcwq->flags &= ~GCWQ_MANAGING_WORKERS;
3328
3329 /* notify completion */
3330 gcwq->trustee = NULL;
3331 gcwq->trustee_state = TRUSTEE_DONE;
3332 wake_up_all(&gcwq->trustee_wait);
3333 spin_unlock_irq(&gcwq->lock);
3334 return 0;
3335}
3336
3337/**
3338 * wait_trustee_state - wait for trustee to enter the specified state
3339 * @gcwq: gcwq the trustee of interest belongs to
3340 * @state: target state to wait for
3341 *
3342 * Wait for the trustee to reach @state. DONE is already matched.
3343 *
3344 * CONTEXT:
3345 * spin_lock_irq(gcwq->lock) which may be released and regrabbed
3346 * multiple times. To be used by cpu_callback.
3347 */
3348static void __cpuinit wait_trustee_state(struct global_cwq *gcwq, int state)
3349__releases(&gcwq->lock)
3350__acquires(&gcwq->lock)
3351{
3352 if (!(gcwq->trustee_state == state ||
3353 gcwq->trustee_state == TRUSTEE_DONE)) {
3354 spin_unlock_irq(&gcwq->lock);
3355 __wait_event(gcwq->trustee_wait,
3356 gcwq->trustee_state == state ||
3357 gcwq->trustee_state == TRUSTEE_DONE);
3358 spin_lock_irq(&gcwq->lock);
3359 }
3360}
3361
1106static int __devinit workqueue_cpu_callback(struct notifier_block *nfb, 3362static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
1107 unsigned long action, 3363 unsigned long action,
1108 void *hcpu) 3364 void *hcpu)
1109{ 3365{
1110 unsigned int cpu = (unsigned long)hcpu; 3366 unsigned int cpu = (unsigned long)hcpu;
1111 struct cpu_workqueue_struct *cwq; 3367 struct global_cwq *gcwq = get_gcwq(cpu);
1112 struct workqueue_struct *wq; 3368 struct task_struct *new_trustee = NULL;
1113 int err = 0; 3369 struct worker *uninitialized_var(new_worker);
3370 unsigned long flags;
1114 3371
1115 action &= ~CPU_TASKS_FROZEN; 3372 action &= ~CPU_TASKS_FROZEN;
1116 3373
1117 switch (action) { 3374 switch (action) {
3375 case CPU_DOWN_PREPARE:
3376 new_trustee = kthread_create(trustee_thread, gcwq,
3377 "workqueue_trustee/%d\n", cpu);
3378 if (IS_ERR(new_trustee))
3379 return notifier_from_errno(PTR_ERR(new_trustee));
3380 kthread_bind(new_trustee, cpu);
3381 /* fall through */
1118 case CPU_UP_PREPARE: 3382 case CPU_UP_PREPARE:
1119 cpumask_set_cpu(cpu, cpu_populated_map); 3383 BUG_ON(gcwq->first_idle);
1120 } 3384 new_worker = create_worker(gcwq, false);
1121undo: 3385 if (!new_worker) {
1122 list_for_each_entry(wq, &workqueues, list) { 3386 if (new_trustee)
1123 cwq = per_cpu_ptr(wq->cpu_wq, cpu); 3387 kthread_stop(new_trustee);
1124 3388 return NOTIFY_BAD;
1125 switch (action) {
1126 case CPU_UP_PREPARE:
1127 err = create_workqueue_thread(cwq, cpu);
1128 if (!err)
1129 break;
1130 printk(KERN_ERR "workqueue [%s] for %i failed\n",
1131 wq->name, cpu);
1132 action = CPU_UP_CANCELED;
1133 err = -ENOMEM;
1134 goto undo;
1135
1136 case CPU_ONLINE:
1137 start_workqueue_thread(cwq, cpu);
1138 break;
1139
1140 case CPU_UP_CANCELED:
1141 start_workqueue_thread(cwq, -1);
1142 case CPU_POST_DEAD:
1143 cleanup_workqueue_thread(cwq);
1144 break;
1145 } 3389 }
1146 } 3390 }
1147 3391
3392 /* some are called w/ irq disabled, don't disturb irq status */
3393 spin_lock_irqsave(&gcwq->lock, flags);
3394
1148 switch (action) { 3395 switch (action) {
1149 case CPU_UP_CANCELED: 3396 case CPU_DOWN_PREPARE:
3397 /* initialize trustee and tell it to acquire the gcwq */
3398 BUG_ON(gcwq->trustee || gcwq->trustee_state != TRUSTEE_DONE);
3399 gcwq->trustee = new_trustee;
3400 gcwq->trustee_state = TRUSTEE_START;
3401 wake_up_process(gcwq->trustee);
3402 wait_trustee_state(gcwq, TRUSTEE_IN_CHARGE);
3403 /* fall through */
3404 case CPU_UP_PREPARE:
3405 BUG_ON(gcwq->first_idle);
3406 gcwq->first_idle = new_worker;
3407 break;
3408
3409 case CPU_DYING:
3410 /*
3411 * Before this, the trustee and all workers except for
3412 * the ones which are still executing works from
3413 * before the last CPU down must be on the cpu. After
3414 * this, they'll all be diasporas.
3415 */
3416 gcwq->flags |= GCWQ_DISASSOCIATED;
3417 break;
3418
1150 case CPU_POST_DEAD: 3419 case CPU_POST_DEAD:
1151 cpumask_clear_cpu(cpu, cpu_populated_map); 3420 gcwq->trustee_state = TRUSTEE_BUTCHER;
3421 /* fall through */
3422 case CPU_UP_CANCELED:
3423 destroy_worker(gcwq->first_idle);
3424 gcwq->first_idle = NULL;
3425 break;
3426
3427 case CPU_DOWN_FAILED:
3428 case CPU_ONLINE:
3429 gcwq->flags &= ~GCWQ_DISASSOCIATED;
3430 if (gcwq->trustee_state != TRUSTEE_DONE) {
3431 gcwq->trustee_state = TRUSTEE_RELEASE;
3432 wake_up_process(gcwq->trustee);
3433 wait_trustee_state(gcwq, TRUSTEE_DONE);
3434 }
3435
3436 /*
3437 * Trustee is done and there might be no worker left.
3438 * Put the first_idle in and request a real manager to
3439 * take a look.
3440 */
3441 spin_unlock_irq(&gcwq->lock);
3442 kthread_bind(gcwq->first_idle->task, cpu);
3443 spin_lock_irq(&gcwq->lock);
3444 gcwq->flags |= GCWQ_MANAGE_WORKERS;
3445 start_worker(gcwq->first_idle);
3446 gcwq->first_idle = NULL;
3447 break;
1152 } 3448 }
1153 3449
1154 return notifier_from_errno(err); 3450 spin_unlock_irqrestore(&gcwq->lock, flags);
3451
3452 return notifier_from_errno(0);
1155} 3453}
1156 3454
1157#ifdef CONFIG_SMP 3455#ifdef CONFIG_SMP
@@ -1201,14 +3499,200 @@ long work_on_cpu(unsigned int cpu, long (*fn)(void *), void *arg)
1201EXPORT_SYMBOL_GPL(work_on_cpu); 3499EXPORT_SYMBOL_GPL(work_on_cpu);
1202#endif /* CONFIG_SMP */ 3500#endif /* CONFIG_SMP */
1203 3501
1204void __init init_workqueues(void) 3502#ifdef CONFIG_FREEZER
3503
3504/**
3505 * freeze_workqueues_begin - begin freezing workqueues
3506 *
3507 * Start freezing workqueues. After this function returns, all
3508 * freezeable workqueues will queue new works to their frozen_works
3509 * list instead of gcwq->worklist.
3510 *
3511 * CONTEXT:
3512 * Grabs and releases workqueue_lock and gcwq->lock's.
3513 */
3514void freeze_workqueues_begin(void)
3515{
3516 unsigned int cpu;
3517
3518 spin_lock(&workqueue_lock);
3519
3520 BUG_ON(workqueue_freezing);
3521 workqueue_freezing = true;
3522
3523 for_each_gcwq_cpu(cpu) {
3524 struct global_cwq *gcwq = get_gcwq(cpu);
3525 struct workqueue_struct *wq;
3526
3527 spin_lock_irq(&gcwq->lock);
3528
3529 BUG_ON(gcwq->flags & GCWQ_FREEZING);
3530 gcwq->flags |= GCWQ_FREEZING;
3531
3532 list_for_each_entry(wq, &workqueues, list) {
3533 struct cpu_workqueue_struct *cwq = get_cwq(cpu, wq);
3534
3535 if (cwq && wq->flags & WQ_FREEZEABLE)
3536 cwq->max_active = 0;
3537 }
3538
3539 spin_unlock_irq(&gcwq->lock);
3540 }
3541
3542 spin_unlock(&workqueue_lock);
3543}
3544
3545/**
3546 * freeze_workqueues_busy - are freezeable workqueues still busy?
3547 *
3548 * Check whether freezing is complete. This function must be called
3549 * between freeze_workqueues_begin() and thaw_workqueues().
3550 *
3551 * CONTEXT:
3552 * Grabs and releases workqueue_lock.
3553 *
3554 * RETURNS:
3555 * %true if some freezeable workqueues are still busy. %false if
3556 * freezing is complete.
3557 */
3558bool freeze_workqueues_busy(void)
3559{
3560 unsigned int cpu;
3561 bool busy = false;
3562
3563 spin_lock(&workqueue_lock);
3564
3565 BUG_ON(!workqueue_freezing);
3566
3567 for_each_gcwq_cpu(cpu) {
3568 struct workqueue_struct *wq;
3569 /*
3570 * nr_active is monotonically decreasing. It's safe
3571 * to peek without lock.
3572 */
3573 list_for_each_entry(wq, &workqueues, list) {
3574 struct cpu_workqueue_struct *cwq = get_cwq(cpu, wq);
3575
3576 if (!cwq || !(wq->flags & WQ_FREEZEABLE))
3577 continue;
3578
3579 BUG_ON(cwq->nr_active < 0);
3580 if (cwq->nr_active) {
3581 busy = true;
3582 goto out_unlock;
3583 }
3584 }
3585 }
3586out_unlock:
3587 spin_unlock(&workqueue_lock);
3588 return busy;
3589}
3590
3591/**
3592 * thaw_workqueues - thaw workqueues
3593 *
3594 * Thaw workqueues. Normal queueing is restored and all collected
3595 * frozen works are transferred to their respective gcwq worklists.
3596 *
3597 * CONTEXT:
3598 * Grabs and releases workqueue_lock and gcwq->lock's.
3599 */
3600void thaw_workqueues(void)
1205{ 3601{
1206 alloc_cpumask_var(&cpu_populated_map, GFP_KERNEL); 3602 unsigned int cpu;
1207 3603
1208 cpumask_copy(cpu_populated_map, cpu_online_mask); 3604 spin_lock(&workqueue_lock);
1209 singlethread_cpu = cpumask_first(cpu_possible_mask); 3605
1210 cpu_singlethread_map = cpumask_of(singlethread_cpu); 3606 if (!workqueue_freezing)
1211 hotcpu_notifier(workqueue_cpu_callback, 0); 3607 goto out_unlock;
1212 keventd_wq = create_workqueue("events"); 3608
1213 BUG_ON(!keventd_wq); 3609 for_each_gcwq_cpu(cpu) {
3610 struct global_cwq *gcwq = get_gcwq(cpu);
3611 struct workqueue_struct *wq;
3612
3613 spin_lock_irq(&gcwq->lock);
3614
3615 BUG_ON(!(gcwq->flags & GCWQ_FREEZING));
3616 gcwq->flags &= ~GCWQ_FREEZING;
3617
3618 list_for_each_entry(wq, &workqueues, list) {
3619 struct cpu_workqueue_struct *cwq = get_cwq(cpu, wq);
3620
3621 if (!cwq || !(wq->flags & WQ_FREEZEABLE))
3622 continue;
3623
3624 /* restore max_active and repopulate worklist */
3625 cwq->max_active = wq->saved_max_active;
3626
3627 while (!list_empty(&cwq->delayed_works) &&
3628 cwq->nr_active < cwq->max_active)
3629 cwq_activate_first_delayed(cwq);
3630 }
3631
3632 wake_up_worker(gcwq);
3633
3634 spin_unlock_irq(&gcwq->lock);
3635 }
3636
3637 workqueue_freezing = false;
3638out_unlock:
3639 spin_unlock(&workqueue_lock);
3640}
3641#endif /* CONFIG_FREEZER */
3642
3643static int __init init_workqueues(void)
3644{
3645 unsigned int cpu;
3646 int i;
3647
3648 cpu_notifier(workqueue_cpu_callback, CPU_PRI_WORKQUEUE);
3649
3650 /* initialize gcwqs */
3651 for_each_gcwq_cpu(cpu) {
3652 struct global_cwq *gcwq = get_gcwq(cpu);
3653
3654 spin_lock_init(&gcwq->lock);
3655 INIT_LIST_HEAD(&gcwq->worklist);
3656 gcwq->cpu = cpu;
3657 gcwq->flags |= GCWQ_DISASSOCIATED;
3658
3659 INIT_LIST_HEAD(&gcwq->idle_list);
3660 for (i = 0; i < BUSY_WORKER_HASH_SIZE; i++)
3661 INIT_HLIST_HEAD(&gcwq->busy_hash[i]);
3662
3663 init_timer_deferrable(&gcwq->idle_timer);
3664 gcwq->idle_timer.function = idle_worker_timeout;
3665 gcwq->idle_timer.data = (unsigned long)gcwq;
3666
3667 setup_timer(&gcwq->mayday_timer, gcwq_mayday_timeout,
3668 (unsigned long)gcwq);
3669
3670 ida_init(&gcwq->worker_ida);
3671
3672 gcwq->trustee_state = TRUSTEE_DONE;
3673 init_waitqueue_head(&gcwq->trustee_wait);
3674 }
3675
3676 /* create the initial worker */
3677 for_each_online_gcwq_cpu(cpu) {
3678 struct global_cwq *gcwq = get_gcwq(cpu);
3679 struct worker *worker;
3680
3681 if (cpu != WORK_CPU_UNBOUND)
3682 gcwq->flags &= ~GCWQ_DISASSOCIATED;
3683 worker = create_worker(gcwq, true);
3684 BUG_ON(!worker);
3685 spin_lock_irq(&gcwq->lock);
3686 start_worker(worker);
3687 spin_unlock_irq(&gcwq->lock);
3688 }
3689
3690 system_wq = alloc_workqueue("events", 0, 0);
3691 system_long_wq = alloc_workqueue("events_long", 0, 0);
3692 system_nrt_wq = alloc_workqueue("events_nrt", WQ_NON_REENTRANT, 0);
3693 system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
3694 WQ_UNBOUND_MAX_ACTIVE);
3695 BUG_ON(!system_wq || !system_long_wq || !system_nrt_wq);
3696 return 0;
1214} 3697}
3698early_initcall(init_workqueues);
generated by cgit v1.2.2 (git 2.25.0) at 2025-06-13 04:14:31 -0400