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