reboot: checkpatch.pl the new kernel/reboot.c file
[linux-3.10.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 #include <linux/jhash.h>
45 #include <linux/hashtable.h>
46 #include <linux/rculist.h>
47 #include <linux/nodemask.h>
48 #include <linux/moduleparam.h>
49 #include <linux/uaccess.h>
50
51 #include "workqueue_internal.h"
52
53 enum {
54         /*
55          * worker_pool flags
56          *
57          * A bound pool is either associated or disassociated with its CPU.
58          * While associated (!DISASSOCIATED), all workers are bound to the
59          * CPU and none has %WORKER_UNBOUND set and concurrency management
60          * is in effect.
61          *
62          * While DISASSOCIATED, the cpu may be offline and all workers have
63          * %WORKER_UNBOUND set and concurrency management disabled, and may
64          * be executing on any CPU.  The pool behaves as an unbound one.
65          *
66          * Note that DISASSOCIATED should be flipped only while holding
67          * manager_mutex to avoid changing binding state while
68          * create_worker() is in progress.
69          */
70         POOL_MANAGE_WORKERS     = 1 << 0,       /* need to manage workers */
71         POOL_DISASSOCIATED      = 1 << 2,       /* cpu can't serve workers */
72         POOL_FREEZING           = 1 << 3,       /* freeze in progress */
73
74         /* worker flags */
75         WORKER_STARTED          = 1 << 0,       /* started */
76         WORKER_DIE              = 1 << 1,       /* die die die */
77         WORKER_IDLE             = 1 << 2,       /* is idle */
78         WORKER_PREP             = 1 << 3,       /* preparing to run works */
79         WORKER_CPU_INTENSIVE    = 1 << 6,       /* cpu intensive */
80         WORKER_UNBOUND          = 1 << 7,       /* worker is unbound */
81         WORKER_REBOUND          = 1 << 8,       /* worker was rebound */
82
83         WORKER_NOT_RUNNING      = WORKER_PREP | WORKER_CPU_INTENSIVE |
84                                   WORKER_UNBOUND | WORKER_REBOUND,
85
86         NR_STD_WORKER_POOLS     = 2,            /* # standard pools per cpu */
87
88         UNBOUND_POOL_HASH_ORDER = 6,            /* hashed by pool->attrs */
89         BUSY_WORKER_HASH_ORDER  = 6,            /* 64 pointers */
90
91         MAX_IDLE_WORKERS_RATIO  = 4,            /* 1/4 of busy can be idle */
92         IDLE_WORKER_TIMEOUT     = 300 * HZ,     /* keep idle ones for 5 mins */
93
94         MAYDAY_INITIAL_TIMEOUT  = HZ / 100 >= 2 ? HZ / 100 : 2,
95                                                 /* call for help after 10ms
96                                                    (min two ticks) */
97         MAYDAY_INTERVAL         = HZ / 10,      /* and then every 100ms */
98         CREATE_COOLDOWN         = HZ,           /* time to breath after fail */
99
100         /*
101          * Rescue workers are used only on emergencies and shared by
102          * all cpus.  Give -20.
103          */
104         RESCUER_NICE_LEVEL      = -20,
105         HIGHPRI_NICE_LEVEL      = -20,
106
107         WQ_NAME_LEN             = 24,
108 };
109
110 /*
111  * Structure fields follow one of the following exclusion rules.
112  *
113  * I: Modifiable by initialization/destruction paths and read-only for
114  *    everyone else.
115  *
116  * P: Preemption protected.  Disabling preemption is enough and should
117  *    only be modified and accessed from the local cpu.
118  *
119  * L: pool->lock protected.  Access with pool->lock held.
120  *
121  * X: During normal operation, modification requires pool->lock and should
122  *    be done only from local cpu.  Either disabling preemption on local
123  *    cpu or grabbing pool->lock is enough for read access.  If
124  *    POOL_DISASSOCIATED is set, it's identical to L.
125  *
126  * MG: pool->manager_mutex and pool->lock protected.  Writes require both
127  *     locks.  Reads can happen under either lock.
128  *
129  * PL: wq_pool_mutex protected.
130  *
131  * PR: wq_pool_mutex protected for writes.  Sched-RCU protected for reads.
132  *
133  * WQ: wq->mutex protected.
134  *
135  * WR: wq->mutex protected for writes.  Sched-RCU protected for reads.
136  *
137  * MD: wq_mayday_lock protected.
138  */
139
140 /* struct worker is defined in workqueue_internal.h */
141
142 struct worker_pool {
143         spinlock_t              lock;           /* the pool lock */
144         int                     cpu;            /* I: the associated cpu */
145         int                     node;           /* I: the associated node ID */
146         int                     id;             /* I: pool ID */
147         unsigned int            flags;          /* X: flags */
148
149         struct list_head        worklist;       /* L: list of pending works */
150         int                     nr_workers;     /* L: total number of workers */
151
152         /* nr_idle includes the ones off idle_list for rebinding */
153         int                     nr_idle;        /* L: currently idle ones */
154
155         struct list_head        idle_list;      /* X: list of idle workers */
156         struct timer_list       idle_timer;     /* L: worker idle timeout */
157         struct timer_list       mayday_timer;   /* L: SOS timer for workers */
158
159         /* a workers is either on busy_hash or idle_list, or the manager */
160         DECLARE_HASHTABLE(busy_hash, BUSY_WORKER_HASH_ORDER);
161                                                 /* L: hash of busy workers */
162
163         /* see manage_workers() for details on the two manager mutexes */
164         struct mutex            manager_arb;    /* manager arbitration */
165         struct mutex            manager_mutex;  /* manager exclusion */
166         struct idr              worker_idr;     /* MG: worker IDs and iteration */
167
168         struct workqueue_attrs  *attrs;         /* I: worker attributes */
169         struct hlist_node       hash_node;      /* PL: unbound_pool_hash node */
170         int                     refcnt;         /* PL: refcnt for unbound pools */
171
172         /*
173          * The current concurrency level.  As it's likely to be accessed
174          * from other CPUs during try_to_wake_up(), put it in a separate
175          * cacheline.
176          */
177         atomic_t                nr_running ____cacheline_aligned_in_smp;
178
179         /*
180          * Destruction of pool is sched-RCU protected to allow dereferences
181          * from get_work_pool().
182          */
183         struct rcu_head         rcu;
184 } ____cacheline_aligned_in_smp;
185
186 /*
187  * The per-pool workqueue.  While queued, the lower WORK_STRUCT_FLAG_BITS
188  * of work_struct->data are used for flags and the remaining high bits
189  * point to the pwq; thus, pwqs need to be aligned at two's power of the
190  * number of flag bits.
191  */
192 struct pool_workqueue {
193         struct worker_pool      *pool;          /* I: the associated pool */
194         struct workqueue_struct *wq;            /* I: the owning workqueue */
195         int                     work_color;     /* L: current color */
196         int                     flush_color;    /* L: flushing color */
197         int                     refcnt;         /* L: reference count */
198         int                     nr_in_flight[WORK_NR_COLORS];
199                                                 /* L: nr of in_flight works */
200         int                     nr_active;      /* L: nr of active works */
201         int                     max_active;     /* L: max active works */
202         struct list_head        delayed_works;  /* L: delayed works */
203         struct list_head        pwqs_node;      /* WR: node on wq->pwqs */
204         struct list_head        mayday_node;    /* MD: node on wq->maydays */
205
206         /*
207          * Release of unbound pwq is punted to system_wq.  See put_pwq()
208          * and pwq_unbound_release_workfn() for details.  pool_workqueue
209          * itself is also sched-RCU protected so that the first pwq can be
210          * determined without grabbing wq->mutex.
211          */
212         struct work_struct      unbound_release_work;
213         struct rcu_head         rcu;
214 } __aligned(1 << WORK_STRUCT_FLAG_BITS);
215
216 /*
217  * Structure used to wait for workqueue flush.
218  */
219 struct wq_flusher {
220         struct list_head        list;           /* WQ: list of flushers */
221         int                     flush_color;    /* WQ: flush color waiting for */
222         struct completion       done;           /* flush completion */
223 };
224
225 struct wq_device;
226
227 /*
228  * The externally visible workqueue.  It relays the issued work items to
229  * the appropriate worker_pool through its pool_workqueues.
230  */
231 struct workqueue_struct {
232         struct list_head        pwqs;           /* WR: all pwqs of this wq */
233         struct list_head        list;           /* PL: list of all workqueues */
234
235         struct mutex            mutex;          /* protects this wq */
236         int                     work_color;     /* WQ: current work color */
237         int                     flush_color;    /* WQ: current flush color */
238         atomic_t                nr_pwqs_to_flush; /* flush in progress */
239         struct wq_flusher       *first_flusher; /* WQ: first flusher */
240         struct list_head        flusher_queue;  /* WQ: flush waiters */
241         struct list_head        flusher_overflow; /* WQ: flush overflow list */
242
243         struct list_head        maydays;        /* MD: pwqs requesting rescue */
244         struct worker           *rescuer;       /* I: rescue worker */
245
246         int                     nr_drainers;    /* WQ: drain in progress */
247         int                     saved_max_active; /* WQ: saved pwq max_active */
248
249         struct workqueue_attrs  *unbound_attrs; /* WQ: only for unbound wqs */
250         struct pool_workqueue   *dfl_pwq;       /* WQ: only for unbound wqs */
251
252 #ifdef CONFIG_SYSFS
253         struct wq_device        *wq_dev;        /* I: for sysfs interface */
254 #endif
255 #ifdef CONFIG_LOCKDEP
256         struct lockdep_map      lockdep_map;
257 #endif
258         char                    name[WQ_NAME_LEN]; /* I: workqueue name */
259
260         /* hot fields used during command issue, aligned to cacheline */
261         unsigned int            flags ____cacheline_aligned; /* WQ: WQ_* flags */
262         struct pool_workqueue __percpu *cpu_pwqs; /* I: per-cpu pwqs */
263         struct pool_workqueue __rcu *numa_pwq_tbl[]; /* FR: unbound pwqs indexed by node */
264 };
265
266 static struct kmem_cache *pwq_cache;
267
268 static int wq_numa_tbl_len;             /* highest possible NUMA node id + 1 */
269 static cpumask_var_t *wq_numa_possible_cpumask;
270                                         /* possible CPUs of each node */
271
272 static bool wq_disable_numa;
273 module_param_named(disable_numa, wq_disable_numa, bool, 0444);
274
275 static bool wq_numa_enabled;            /* unbound NUMA affinity enabled */
276
277 /* buf for wq_update_unbound_numa_attrs(), protected by CPU hotplug exclusion */
278 static struct workqueue_attrs *wq_update_unbound_numa_attrs_buf;
279
280 static DEFINE_MUTEX(wq_pool_mutex);     /* protects pools and workqueues list */
281 static DEFINE_SPINLOCK(wq_mayday_lock); /* protects wq->maydays list */
282
283 static LIST_HEAD(workqueues);           /* PL: list of all workqueues */
284 static bool workqueue_freezing;         /* PL: have wqs started freezing? */
285
286 /* the per-cpu worker pools */
287 static DEFINE_PER_CPU_SHARED_ALIGNED(struct worker_pool [NR_STD_WORKER_POOLS],
288                                      cpu_worker_pools);
289
290 static DEFINE_IDR(worker_pool_idr);     /* PR: idr of all pools */
291
292 /* PL: hash of all unbound pools keyed by pool->attrs */
293 static DEFINE_HASHTABLE(unbound_pool_hash, UNBOUND_POOL_HASH_ORDER);
294
295 /* I: attributes used when instantiating standard unbound pools on demand */
296 static struct workqueue_attrs *unbound_std_wq_attrs[NR_STD_WORKER_POOLS];
297
298 /* I: attributes used when instantiating ordered pools on demand */
299 static struct workqueue_attrs *ordered_wq_attrs[NR_STD_WORKER_POOLS];
300
301 struct workqueue_struct *system_wq __read_mostly;
302 EXPORT_SYMBOL(system_wq);
303 struct workqueue_struct *system_highpri_wq __read_mostly;
304 EXPORT_SYMBOL_GPL(system_highpri_wq);
305 struct workqueue_struct *system_long_wq __read_mostly;
306 EXPORT_SYMBOL_GPL(system_long_wq);
307 struct workqueue_struct *system_unbound_wq __read_mostly;
308 EXPORT_SYMBOL_GPL(system_unbound_wq);
309 struct workqueue_struct *system_freezable_wq __read_mostly;
310 EXPORT_SYMBOL_GPL(system_freezable_wq);
311
312 static int worker_thread(void *__worker);
313 static void copy_workqueue_attrs(struct workqueue_attrs *to,
314                                  const struct workqueue_attrs *from);
315
316 #define CREATE_TRACE_POINTS
317 #include <trace/events/workqueue.h>
318
319 #define assert_rcu_or_pool_mutex()                                      \
320         rcu_lockdep_assert(rcu_read_lock_sched_held() ||                \
321                            lockdep_is_held(&wq_pool_mutex),             \
322                            "sched RCU or wq_pool_mutex should be held")
323
324 #define assert_rcu_or_wq_mutex(wq)                                      \
325         rcu_lockdep_assert(rcu_read_lock_sched_held() ||                \
326                            lockdep_is_held(&wq->mutex),                 \
327                            "sched RCU or wq->mutex should be held")
328
329 #ifdef CONFIG_LOCKDEP
330 #define assert_manager_or_pool_lock(pool)                               \
331         WARN_ONCE(debug_locks &&                                        \
332                   !lockdep_is_held(&(pool)->manager_mutex) &&           \
333                   !lockdep_is_held(&(pool)->lock),                      \
334                   "pool->manager_mutex or ->lock should be held")
335 #else
336 #define assert_manager_or_pool_lock(pool)       do { } while (0)
337 #endif
338
339 #define for_each_cpu_worker_pool(pool, cpu)                             \
340         for ((pool) = &per_cpu(cpu_worker_pools, cpu)[0];               \
341              (pool) < &per_cpu(cpu_worker_pools, cpu)[NR_STD_WORKER_POOLS]; \
342              (pool)++)
343
344 /**
345  * for_each_pool - iterate through all worker_pools in the system
346  * @pool: iteration cursor
347  * @pi: integer used for iteration
348  *
349  * This must be called either with wq_pool_mutex held or sched RCU read
350  * locked.  If the pool needs to be used beyond the locking in effect, the
351  * caller is responsible for guaranteeing that the pool stays online.
352  *
353  * The if/else clause exists only for the lockdep assertion and can be
354  * ignored.
355  */
356 #define for_each_pool(pool, pi)                                         \
357         idr_for_each_entry(&worker_pool_idr, pool, pi)                  \
358                 if (({ assert_rcu_or_pool_mutex(); false; })) { }       \
359                 else
360
361 /**
362  * for_each_pool_worker - iterate through all workers of a worker_pool
363  * @worker: iteration cursor
364  * @wi: integer used for iteration
365  * @pool: worker_pool to iterate workers of
366  *
367  * This must be called with either @pool->manager_mutex or ->lock held.
368  *
369  * The if/else clause exists only for the lockdep assertion and can be
370  * ignored.
371  */
372 #define for_each_pool_worker(worker, wi, pool)                          \
373         idr_for_each_entry(&(pool)->worker_idr, (worker), (wi))         \
374                 if (({ assert_manager_or_pool_lock((pool)); false; })) { } \
375                 else
376
377 /**
378  * for_each_pwq - iterate through all pool_workqueues of the specified workqueue
379  * @pwq: iteration cursor
380  * @wq: the target workqueue
381  *
382  * This must be called either with wq->mutex held or sched RCU read locked.
383  * If the pwq needs to be used beyond the locking in effect, the caller is
384  * responsible for guaranteeing that the pwq stays online.
385  *
386  * The if/else clause exists only for the lockdep assertion and can be
387  * ignored.
388  */
389 #define for_each_pwq(pwq, wq)                                           \
390         list_for_each_entry_rcu((pwq), &(wq)->pwqs, pwqs_node)          \
391                 if (({ assert_rcu_or_wq_mutex(wq); false; })) { }       \
392                 else
393
394 #ifdef CONFIG_DEBUG_OBJECTS_WORK
395
396 static struct debug_obj_descr work_debug_descr;
397
398 static void *work_debug_hint(void *addr)
399 {
400         return ((struct work_struct *) addr)->func;
401 }
402
403 /*
404  * fixup_init is called when:
405  * - an active object is initialized
406  */
407 static int work_fixup_init(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_init(work, &work_debug_descr);
415                 return 1;
416         default:
417                 return 0;
418         }
419 }
420
421 /*
422  * fixup_activate is called when:
423  * - an active object is activated
424  * - an unknown object is activated (might be a statically initialized object)
425  */
426 static int work_fixup_activate(void *addr, enum debug_obj_state state)
427 {
428         struct work_struct *work = addr;
429
430         switch (state) {
431
432         case ODEBUG_STATE_NOTAVAILABLE:
433                 /*
434                  * This is not really a fixup. The work struct was
435                  * statically initialized. We just make sure that it
436                  * is tracked in the object tracker.
437                  */
438                 if (test_bit(WORK_STRUCT_STATIC_BIT, work_data_bits(work))) {
439                         debug_object_init(work, &work_debug_descr);
440                         debug_object_activate(work, &work_debug_descr);
441                         return 0;
442                 }
443                 WARN_ON_ONCE(1);
444                 return 0;
445
446         case ODEBUG_STATE_ACTIVE:
447                 WARN_ON(1);
448
449         default:
450                 return 0;
451         }
452 }
453
454 /*
455  * fixup_free is called when:
456  * - an active object is freed
457  */
458 static int work_fixup_free(void *addr, enum debug_obj_state state)
459 {
460         struct work_struct *work = addr;
461
462         switch (state) {
463         case ODEBUG_STATE_ACTIVE:
464                 cancel_work_sync(work);
465                 debug_object_free(work, &work_debug_descr);
466                 return 1;
467         default:
468                 return 0;
469         }
470 }
471
472 static struct debug_obj_descr work_debug_descr = {
473         .name           = "work_struct",
474         .debug_hint     = work_debug_hint,
475         .fixup_init     = work_fixup_init,
476         .fixup_activate = work_fixup_activate,
477         .fixup_free     = work_fixup_free,
478 };
479
480 static inline void debug_work_activate(struct work_struct *work)
481 {
482         debug_object_activate(work, &work_debug_descr);
483 }
484
485 static inline void debug_work_deactivate(struct work_struct *work)
486 {
487         debug_object_deactivate(work, &work_debug_descr);
488 }
489
490 void __init_work(struct work_struct *work, int onstack)
491 {
492         if (onstack)
493                 debug_object_init_on_stack(work, &work_debug_descr);
494         else
495                 debug_object_init(work, &work_debug_descr);
496 }
497 EXPORT_SYMBOL_GPL(__init_work);
498
499 void destroy_work_on_stack(struct work_struct *work)
500 {
501         debug_object_free(work, &work_debug_descr);
502 }
503 EXPORT_SYMBOL_GPL(destroy_work_on_stack);
504
505 #else
506 static inline void debug_work_activate(struct work_struct *work) { }
507 static inline void debug_work_deactivate(struct work_struct *work) { }
508 #endif
509
510 /* allocate ID and assign it to @pool */
511 static int worker_pool_assign_id(struct worker_pool *pool)
512 {
513         int ret;
514
515         lockdep_assert_held(&wq_pool_mutex);
516
517         ret = idr_alloc(&worker_pool_idr, pool, 0, 0, GFP_KERNEL);
518         if (ret >= 0) {
519                 pool->id = ret;
520                 return 0;
521         }
522         return ret;
523 }
524
525 /**
526  * unbound_pwq_by_node - return the unbound pool_workqueue for the given node
527  * @wq: the target workqueue
528  * @node: the node ID
529  *
530  * This must be called either with pwq_lock held or sched RCU read locked.
531  * If the pwq needs to be used beyond the locking in effect, the caller is
532  * responsible for guaranteeing that the pwq stays online.
533  */
534 static struct pool_workqueue *unbound_pwq_by_node(struct workqueue_struct *wq,
535                                                   int node)
536 {
537         assert_rcu_or_wq_mutex(wq);
538         return rcu_dereference_raw(wq->numa_pwq_tbl[node]);
539 }
540
541 static unsigned int work_color_to_flags(int color)
542 {
543         return color << WORK_STRUCT_COLOR_SHIFT;
544 }
545
546 static int get_work_color(struct work_struct *work)
547 {
548         return (*work_data_bits(work) >> WORK_STRUCT_COLOR_SHIFT) &
549                 ((1 << WORK_STRUCT_COLOR_BITS) - 1);
550 }
551
552 static int work_next_color(int color)
553 {
554         return (color + 1) % WORK_NR_COLORS;
555 }
556
557 /*
558  * While queued, %WORK_STRUCT_PWQ is set and non flag bits of a work's data
559  * contain the pointer to the queued pwq.  Once execution starts, the flag
560  * is cleared and the high bits contain OFFQ flags and pool ID.
561  *
562  * set_work_pwq(), set_work_pool_and_clear_pending(), mark_work_canceling()
563  * and clear_work_data() can be used to set the pwq, pool or clear
564  * work->data.  These functions should only be called while the work is
565  * owned - ie. while the PENDING bit is set.
566  *
567  * get_work_pool() and get_work_pwq() can be used to obtain the pool or pwq
568  * corresponding to a work.  Pool is available once the work has been
569  * queued anywhere after initialization until it is sync canceled.  pwq is
570  * available only while the work item is queued.
571  *
572  * %WORK_OFFQ_CANCELING is used to mark a work item which is being
573  * canceled.  While being canceled, a work item may have its PENDING set
574  * but stay off timer and worklist for arbitrarily long and nobody should
575  * try to steal the PENDING bit.
576  */
577 static inline void set_work_data(struct work_struct *work, unsigned long data,
578                                  unsigned long flags)
579 {
580         WARN_ON_ONCE(!work_pending(work));
581         atomic_long_set(&work->data, data | flags | work_static(work));
582 }
583
584 static void set_work_pwq(struct work_struct *work, struct pool_workqueue *pwq,
585                          unsigned long extra_flags)
586 {
587         set_work_data(work, (unsigned long)pwq,
588                       WORK_STRUCT_PENDING | WORK_STRUCT_PWQ | extra_flags);
589 }
590
591 static void set_work_pool_and_keep_pending(struct work_struct *work,
592                                            int pool_id)
593 {
594         set_work_data(work, (unsigned long)pool_id << WORK_OFFQ_POOL_SHIFT,
595                       WORK_STRUCT_PENDING);
596 }
597
598 static void set_work_pool_and_clear_pending(struct work_struct *work,
599                                             int pool_id)
600 {
601         /*
602          * The following wmb is paired with the implied mb in
603          * test_and_set_bit(PENDING) and ensures all updates to @work made
604          * here are visible to and precede any updates by the next PENDING
605          * owner.
606          */
607         smp_wmb();
608         set_work_data(work, (unsigned long)pool_id << WORK_OFFQ_POOL_SHIFT, 0);
609 }
610
611 static void clear_work_data(struct work_struct *work)
612 {
613         smp_wmb();      /* see set_work_pool_and_clear_pending() */
614         set_work_data(work, WORK_STRUCT_NO_POOL, 0);
615 }
616
617 static struct pool_workqueue *get_work_pwq(struct work_struct *work)
618 {
619         unsigned long data = atomic_long_read(&work->data);
620
621         if (data & WORK_STRUCT_PWQ)
622                 return (void *)(data & WORK_STRUCT_WQ_DATA_MASK);
623         else
624                 return NULL;
625 }
626
627 /**
628  * get_work_pool - return the worker_pool a given work was associated with
629  * @work: the work item of interest
630  *
631  * Return the worker_pool @work was last associated with.  %NULL if none.
632  *
633  * Pools are created and destroyed under wq_pool_mutex, and allows read
634  * access under sched-RCU read lock.  As such, this function should be
635  * called under wq_pool_mutex or with preemption disabled.
636  *
637  * All fields of the returned pool are accessible as long as the above
638  * mentioned locking is in effect.  If the returned pool needs to be used
639  * beyond the critical section, the caller is responsible for ensuring the
640  * returned pool is and stays online.
641  */
642 static struct worker_pool *get_work_pool(struct work_struct *work)
643 {
644         unsigned long data = atomic_long_read(&work->data);
645         int pool_id;
646
647         assert_rcu_or_pool_mutex();
648
649         if (data & WORK_STRUCT_PWQ)
650                 return ((struct pool_workqueue *)
651                         (data & WORK_STRUCT_WQ_DATA_MASK))->pool;
652
653         pool_id = data >> WORK_OFFQ_POOL_SHIFT;
654         if (pool_id == WORK_OFFQ_POOL_NONE)
655                 return NULL;
656
657         return idr_find(&worker_pool_idr, pool_id);
658 }
659
660 /**
661  * get_work_pool_id - return the worker pool ID a given work is associated with
662  * @work: the work item of interest
663  *
664  * Return the worker_pool ID @work was last associated with.
665  * %WORK_OFFQ_POOL_NONE if none.
666  */
667 static int get_work_pool_id(struct work_struct *work)
668 {
669         unsigned long data = atomic_long_read(&work->data);
670
671         if (data & WORK_STRUCT_PWQ)
672                 return ((struct pool_workqueue *)
673                         (data & WORK_STRUCT_WQ_DATA_MASK))->pool->id;
674
675         return data >> WORK_OFFQ_POOL_SHIFT;
676 }
677
678 static void mark_work_canceling(struct work_struct *work)
679 {
680         unsigned long pool_id = get_work_pool_id(work);
681
682         pool_id <<= WORK_OFFQ_POOL_SHIFT;
683         set_work_data(work, pool_id | WORK_OFFQ_CANCELING, WORK_STRUCT_PENDING);
684 }
685
686 static bool work_is_canceling(struct work_struct *work)
687 {
688         unsigned long data = atomic_long_read(&work->data);
689
690         return !(data & WORK_STRUCT_PWQ) && (data & WORK_OFFQ_CANCELING);
691 }
692
693 /*
694  * Policy functions.  These define the policies on how the global worker
695  * pools are managed.  Unless noted otherwise, these functions assume that
696  * they're being called with pool->lock held.
697  */
698
699 static bool __need_more_worker(struct worker_pool *pool)
700 {
701         return !atomic_read(&pool->nr_running);
702 }
703
704 /*
705  * Need to wake up a worker?  Called from anything but currently
706  * running workers.
707  *
708  * Note that, because unbound workers never contribute to nr_running, this
709  * function will always return %true for unbound pools as long as the
710  * worklist isn't empty.
711  */
712 static bool need_more_worker(struct worker_pool *pool)
713 {
714         return !list_empty(&pool->worklist) && __need_more_worker(pool);
715 }
716
717 /* Can I start working?  Called from busy but !running workers. */
718 static bool may_start_working(struct worker_pool *pool)
719 {
720         return pool->nr_idle;
721 }
722
723 /* Do I need to keep working?  Called from currently running workers. */
724 static bool keep_working(struct worker_pool *pool)
725 {
726         return !list_empty(&pool->worklist) &&
727                 atomic_read(&pool->nr_running) <= 1;
728 }
729
730 /* Do we need a new worker?  Called from manager. */
731 static bool need_to_create_worker(struct worker_pool *pool)
732 {
733         return need_more_worker(pool) && !may_start_working(pool);
734 }
735
736 /* Do I need to be the manager? */
737 static bool need_to_manage_workers(struct worker_pool *pool)
738 {
739         return need_to_create_worker(pool) ||
740                 (pool->flags & POOL_MANAGE_WORKERS);
741 }
742
743 /* Do we have too many workers and should some go away? */
744 static bool too_many_workers(struct worker_pool *pool)
745 {
746         bool managing = mutex_is_locked(&pool->manager_arb);
747         int nr_idle = pool->nr_idle + managing; /* manager is considered idle */
748         int nr_busy = pool->nr_workers - nr_idle;
749
750         /*
751          * nr_idle and idle_list may disagree if idle rebinding is in
752          * progress.  Never return %true if idle_list is empty.
753          */
754         if (list_empty(&pool->idle_list))
755                 return false;
756
757         return nr_idle > 2 && (nr_idle - 2) * MAX_IDLE_WORKERS_RATIO >= nr_busy;
758 }
759
760 /*
761  * Wake up functions.
762  */
763
764 /* Return the first worker.  Safe with preemption disabled */
765 static struct worker *first_worker(struct worker_pool *pool)
766 {
767         if (unlikely(list_empty(&pool->idle_list)))
768                 return NULL;
769
770         return list_first_entry(&pool->idle_list, struct worker, entry);
771 }
772
773 /**
774  * wake_up_worker - wake up an idle worker
775  * @pool: worker pool to wake worker from
776  *
777  * Wake up the first idle worker of @pool.
778  *
779  * CONTEXT:
780  * spin_lock_irq(pool->lock).
781  */
782 static void wake_up_worker(struct worker_pool *pool)
783 {
784         struct worker *worker = first_worker(pool);
785
786         if (likely(worker))
787                 wake_up_process(worker->task);
788 }
789
790 /**
791  * wq_worker_waking_up - a worker is waking up
792  * @task: task waking up
793  * @cpu: CPU @task is waking up to
794  *
795  * This function is called during try_to_wake_up() when a worker is
796  * being awoken.
797  *
798  * CONTEXT:
799  * spin_lock_irq(rq->lock)
800  */
801 void wq_worker_waking_up(struct task_struct *task, int cpu)
802 {
803         struct worker *worker = kthread_data(task);
804
805         if (!(worker->flags & WORKER_NOT_RUNNING)) {
806                 WARN_ON_ONCE(worker->pool->cpu != cpu);
807                 atomic_inc(&worker->pool->nr_running);
808         }
809 }
810
811 /**
812  * wq_worker_sleeping - a worker is going to sleep
813  * @task: task going to sleep
814  * @cpu: CPU in question, must be the current CPU number
815  *
816  * This function is called during schedule() when a busy worker is
817  * going to sleep.  Worker on the same cpu can be woken up by
818  * returning pointer to its task.
819  *
820  * CONTEXT:
821  * spin_lock_irq(rq->lock)
822  *
823  * RETURNS:
824  * Worker task on @cpu to wake up, %NULL if none.
825  */
826 struct task_struct *wq_worker_sleeping(struct task_struct *task, int cpu)
827 {
828         struct worker *worker = kthread_data(task), *to_wakeup = NULL;
829         struct worker_pool *pool;
830
831         /*
832          * Rescuers, which may not have all the fields set up like normal
833          * workers, also reach here, let's not access anything before
834          * checking NOT_RUNNING.
835          */
836         if (worker->flags & WORKER_NOT_RUNNING)
837                 return NULL;
838
839         pool = worker->pool;
840
841         /* this can only happen on the local cpu */
842         if (WARN_ON_ONCE(cpu != raw_smp_processor_id()))
843                 return NULL;
844
845         /*
846          * The counterpart of the following dec_and_test, implied mb,
847          * worklist not empty test sequence is in insert_work().
848          * Please read comment there.
849          *
850          * NOT_RUNNING is clear.  This means that we're bound to and
851          * running on the local cpu w/ rq lock held and preemption
852          * disabled, which in turn means that none else could be
853          * manipulating idle_list, so dereferencing idle_list without pool
854          * lock is safe.
855          */
856         if (atomic_dec_and_test(&pool->nr_running) &&
857             !list_empty(&pool->worklist))
858                 to_wakeup = first_worker(pool);
859         return to_wakeup ? to_wakeup->task : NULL;
860 }
861
862 /**
863  * worker_set_flags - set worker flags and adjust nr_running accordingly
864  * @worker: self
865  * @flags: flags to set
866  * @wakeup: wakeup an idle worker if necessary
867  *
868  * Set @flags in @worker->flags and adjust nr_running accordingly.  If
869  * nr_running becomes zero and @wakeup is %true, an idle worker is
870  * woken up.
871  *
872  * CONTEXT:
873  * spin_lock_irq(pool->lock)
874  */
875 static inline void worker_set_flags(struct worker *worker, unsigned int flags,
876                                     bool wakeup)
877 {
878         struct worker_pool *pool = worker->pool;
879
880         WARN_ON_ONCE(worker->task != current);
881
882         /*
883          * If transitioning into NOT_RUNNING, adjust nr_running and
884          * wake up an idle worker as necessary if requested by
885          * @wakeup.
886          */
887         if ((flags & WORKER_NOT_RUNNING) &&
888             !(worker->flags & WORKER_NOT_RUNNING)) {
889                 if (wakeup) {
890                         if (atomic_dec_and_test(&pool->nr_running) &&
891                             !list_empty(&pool->worklist))
892                                 wake_up_worker(pool);
893                 } else
894                         atomic_dec(&pool->nr_running);
895         }
896
897         worker->flags |= flags;
898 }
899
900 /**
901  * worker_clr_flags - clear worker flags and adjust nr_running accordingly
902  * @worker: self
903  * @flags: flags to clear
904  *
905  * Clear @flags in @worker->flags and adjust nr_running accordingly.
906  *
907  * CONTEXT:
908  * spin_lock_irq(pool->lock)
909  */
910 static inline void worker_clr_flags(struct worker *worker, unsigned int flags)
911 {
912         struct worker_pool *pool = worker->pool;
913         unsigned int oflags = worker->flags;
914
915         WARN_ON_ONCE(worker->task != current);
916
917         worker->flags &= ~flags;
918
919         /*
920          * If transitioning out of NOT_RUNNING, increment nr_running.  Note
921          * that the nested NOT_RUNNING is not a noop.  NOT_RUNNING is mask
922          * of multiple flags, not a single flag.
923          */
924         if ((flags & WORKER_NOT_RUNNING) && (oflags & WORKER_NOT_RUNNING))
925                 if (!(worker->flags & WORKER_NOT_RUNNING))
926                         atomic_inc(&pool->nr_running);
927 }
928
929 /**
930  * find_worker_executing_work - find worker which is executing a work
931  * @pool: pool of interest
932  * @work: work to find worker for
933  *
934  * Find a worker which is executing @work on @pool by searching
935  * @pool->busy_hash which is keyed by the address of @work.  For a worker
936  * to match, its current execution should match the address of @work and
937  * its work function.  This is to avoid unwanted dependency between
938  * unrelated work executions through a work item being recycled while still
939  * being executed.
940  *
941  * This is a bit tricky.  A work item may be freed once its execution
942  * starts and nothing prevents the freed area from being recycled for
943  * another work item.  If the same work item address ends up being reused
944  * before the original execution finishes, workqueue will identify the
945  * recycled work item as currently executing and make it wait until the
946  * current execution finishes, introducing an unwanted dependency.
947  *
948  * This function checks the work item address and work function to avoid
949  * false positives.  Note that this isn't complete as one may construct a
950  * work function which can introduce dependency onto itself through a
951  * recycled work item.  Well, if somebody wants to shoot oneself in the
952  * foot that badly, there's only so much we can do, and if such deadlock
953  * actually occurs, it should be easy to locate the culprit work function.
954  *
955  * CONTEXT:
956  * spin_lock_irq(pool->lock).
957  *
958  * RETURNS:
959  * Pointer to worker which is executing @work if found, NULL
960  * otherwise.
961  */
962 static struct worker *find_worker_executing_work(struct worker_pool *pool,
963                                                  struct work_struct *work)
964 {
965         struct worker *worker;
966
967         hash_for_each_possible(pool->busy_hash, worker, hentry,
968                                (unsigned long)work)
969                 if (worker->current_work == work &&
970                     worker->current_func == work->func)
971                         return worker;
972
973         return NULL;
974 }
975
976 /**
977  * move_linked_works - move linked works to a list
978  * @work: start of series of works to be scheduled
979  * @head: target list to append @work to
980  * @nextp: out paramter for nested worklist walking
981  *
982  * Schedule linked works starting from @work to @head.  Work series to
983  * be scheduled starts at @work and includes any consecutive work with
984  * WORK_STRUCT_LINKED set in its predecessor.
985  *
986  * If @nextp is not NULL, it's updated to point to the next work of
987  * the last scheduled work.  This allows move_linked_works() to be
988  * nested inside outer list_for_each_entry_safe().
989  *
990  * CONTEXT:
991  * spin_lock_irq(pool->lock).
992  */
993 static void move_linked_works(struct work_struct *work, struct list_head *head,
994                               struct work_struct **nextp)
995 {
996         struct work_struct *n;
997
998         /*
999          * Linked worklist will always end before the end of the list,
1000          * use NULL for list head.
1001          */
1002         list_for_each_entry_safe_from(work, n, NULL, entry) {
1003                 list_move_tail(&work->entry, head);
1004                 if (!(*work_data_bits(work) & WORK_STRUCT_LINKED))
1005                         break;
1006         }
1007
1008         /*
1009          * If we're already inside safe list traversal and have moved
1010          * multiple works to the scheduled queue, the next position
1011          * needs to be updated.
1012          */
1013         if (nextp)
1014                 *nextp = n;
1015 }
1016
1017 /**
1018  * get_pwq - get an extra reference on the specified pool_workqueue
1019  * @pwq: pool_workqueue to get
1020  *
1021  * Obtain an extra reference on @pwq.  The caller should guarantee that
1022  * @pwq has positive refcnt and be holding the matching pool->lock.
1023  */
1024 static void get_pwq(struct pool_workqueue *pwq)
1025 {
1026         lockdep_assert_held(&pwq->pool->lock);
1027         WARN_ON_ONCE(pwq->refcnt <= 0);
1028         pwq->refcnt++;
1029 }
1030
1031 /**
1032  * put_pwq - put a pool_workqueue reference
1033  * @pwq: pool_workqueue to put
1034  *
1035  * Drop a reference of @pwq.  If its refcnt reaches zero, schedule its
1036  * destruction.  The caller should be holding the matching pool->lock.
1037  */
1038 static void put_pwq(struct pool_workqueue *pwq)
1039 {
1040         lockdep_assert_held(&pwq->pool->lock);
1041         if (likely(--pwq->refcnt))
1042                 return;
1043         if (WARN_ON_ONCE(!(pwq->wq->flags & WQ_UNBOUND)))
1044                 return;
1045         /*
1046          * @pwq can't be released under pool->lock, bounce to
1047          * pwq_unbound_release_workfn().  This never recurses on the same
1048          * pool->lock as this path is taken only for unbound workqueues and
1049          * the release work item is scheduled on a per-cpu workqueue.  To
1050          * avoid lockdep warning, unbound pool->locks are given lockdep
1051          * subclass of 1 in get_unbound_pool().
1052          */
1053         schedule_work(&pwq->unbound_release_work);
1054 }
1055
1056 /**
1057  * put_pwq_unlocked - put_pwq() with surrounding pool lock/unlock
1058  * @pwq: pool_workqueue to put (can be %NULL)
1059  *
1060  * put_pwq() with locking.  This function also allows %NULL @pwq.
1061  */
1062 static void put_pwq_unlocked(struct pool_workqueue *pwq)
1063 {
1064         if (pwq) {
1065                 /*
1066                  * As both pwqs and pools are sched-RCU protected, the
1067                  * following lock operations are safe.
1068                  */
1069                 spin_lock_irq(&pwq->pool->lock);
1070                 put_pwq(pwq);
1071                 spin_unlock_irq(&pwq->pool->lock);
1072         }
1073 }
1074
1075 static void pwq_activate_delayed_work(struct work_struct *work)
1076 {
1077         struct pool_workqueue *pwq = get_work_pwq(work);
1078
1079         trace_workqueue_activate_work(work);
1080         move_linked_works(work, &pwq->pool->worklist, NULL);
1081         __clear_bit(WORK_STRUCT_DELAYED_BIT, work_data_bits(work));
1082         pwq->nr_active++;
1083 }
1084
1085 static void pwq_activate_first_delayed(struct pool_workqueue *pwq)
1086 {
1087         struct work_struct *work = list_first_entry(&pwq->delayed_works,
1088                                                     struct work_struct, entry);
1089
1090         pwq_activate_delayed_work(work);
1091 }
1092
1093 /**
1094  * pwq_dec_nr_in_flight - decrement pwq's nr_in_flight
1095  * @pwq: pwq of interest
1096  * @color: color of work which left the queue
1097  *
1098  * A work either has completed or is removed from pending queue,
1099  * decrement nr_in_flight of its pwq and handle workqueue flushing.
1100  *
1101  * CONTEXT:
1102  * spin_lock_irq(pool->lock).
1103  */
1104 static void pwq_dec_nr_in_flight(struct pool_workqueue *pwq, int color)
1105 {
1106         /* uncolored work items don't participate in flushing or nr_active */
1107         if (color == WORK_NO_COLOR)
1108                 goto out_put;
1109
1110         pwq->nr_in_flight[color]--;
1111
1112         pwq->nr_active--;
1113         if (!list_empty(&pwq->delayed_works)) {
1114                 /* one down, submit a delayed one */
1115                 if (pwq->nr_active < pwq->max_active)
1116                         pwq_activate_first_delayed(pwq);
1117         }
1118
1119         /* is flush in progress and are we at the flushing tip? */
1120         if (likely(pwq->flush_color != color))
1121                 goto out_put;
1122
1123         /* are there still in-flight works? */
1124         if (pwq->nr_in_flight[color])
1125                 goto out_put;
1126
1127         /* this pwq is done, clear flush_color */
1128         pwq->flush_color = -1;
1129
1130         /*
1131          * If this was the last pwq, wake up the first flusher.  It
1132          * will handle the rest.
1133          */
1134         if (atomic_dec_and_test(&pwq->wq->nr_pwqs_to_flush))
1135                 complete(&pwq->wq->first_flusher->done);
1136 out_put:
1137         put_pwq(pwq);
1138 }
1139
1140 /**
1141  * try_to_grab_pending - steal work item from worklist and disable irq
1142  * @work: work item to steal
1143  * @is_dwork: @work is a delayed_work
1144  * @flags: place to store irq state
1145  *
1146  * Try to grab PENDING bit of @work.  This function can handle @work in any
1147  * stable state - idle, on timer or on worklist.  Return values are
1148  *
1149  *  1           if @work was pending and we successfully stole PENDING
1150  *  0           if @work was idle and we claimed PENDING
1151  *  -EAGAIN     if PENDING couldn't be grabbed at the moment, safe to busy-retry
1152  *  -ENOENT     if someone else is canceling @work, this state may persist
1153  *              for arbitrarily long
1154  *
1155  * On >= 0 return, the caller owns @work's PENDING bit.  To avoid getting
1156  * interrupted while holding PENDING and @work off queue, irq must be
1157  * disabled on entry.  This, combined with delayed_work->timer being
1158  * irqsafe, ensures that we return -EAGAIN for finite short period of time.
1159  *
1160  * On successful return, >= 0, irq is disabled and the caller is
1161  * responsible for releasing it using local_irq_restore(*@flags).
1162  *
1163  * This function is safe to call from any context including IRQ handler.
1164  */
1165 static int try_to_grab_pending(struct work_struct *work, bool is_dwork,
1166                                unsigned long *flags)
1167 {
1168         struct worker_pool *pool;
1169         struct pool_workqueue *pwq;
1170
1171         local_irq_save(*flags);
1172
1173         /* try to steal the timer if it exists */
1174         if (is_dwork) {
1175                 struct delayed_work *dwork = to_delayed_work(work);
1176
1177                 /*
1178                  * dwork->timer is irqsafe.  If del_timer() fails, it's
1179                  * guaranteed that the timer is not queued anywhere and not
1180                  * running on the local CPU.
1181                  */
1182                 if (likely(del_timer(&dwork->timer)))
1183                         return 1;
1184         }
1185
1186         /* try to claim PENDING the normal way */
1187         if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work)))
1188                 return 0;
1189
1190         /*
1191          * The queueing is in progress, or it is already queued. Try to
1192          * steal it from ->worklist without clearing WORK_STRUCT_PENDING.
1193          */
1194         pool = get_work_pool(work);
1195         if (!pool)
1196                 goto fail;
1197
1198         spin_lock(&pool->lock);
1199         /*
1200          * work->data is guaranteed to point to pwq only while the work
1201          * item is queued on pwq->wq, and both updating work->data to point
1202          * to pwq on queueing and to pool on dequeueing are done under
1203          * pwq->pool->lock.  This in turn guarantees that, if work->data
1204          * points to pwq which is associated with a locked pool, the work
1205          * item is currently queued on that pool.
1206          */
1207         pwq = get_work_pwq(work);
1208         if (pwq && pwq->pool == pool) {
1209                 debug_work_deactivate(work);
1210
1211                 /*
1212                  * A delayed work item cannot be grabbed directly because
1213                  * it might have linked NO_COLOR work items which, if left
1214                  * on the delayed_list, will confuse pwq->nr_active
1215                  * management later on and cause stall.  Make sure the work
1216                  * item is activated before grabbing.
1217                  */
1218                 if (*work_data_bits(work) & WORK_STRUCT_DELAYED)
1219                         pwq_activate_delayed_work(work);
1220
1221                 list_del_init(&work->entry);
1222                 pwq_dec_nr_in_flight(get_work_pwq(work), get_work_color(work));
1223
1224                 /* work->data points to pwq iff queued, point to pool */
1225                 set_work_pool_and_keep_pending(work, pool->id);
1226
1227                 spin_unlock(&pool->lock);
1228                 return 1;
1229         }
1230         spin_unlock(&pool->lock);
1231 fail:
1232         local_irq_restore(*flags);
1233         if (work_is_canceling(work))
1234                 return -ENOENT;
1235         cpu_relax();
1236         return -EAGAIN;
1237 }
1238
1239 /**
1240  * insert_work - insert a work into a pool
1241  * @pwq: pwq @work belongs to
1242  * @work: work to insert
1243  * @head: insertion point
1244  * @extra_flags: extra WORK_STRUCT_* flags to set
1245  *
1246  * Insert @work which belongs to @pwq after @head.  @extra_flags is or'd to
1247  * work_struct flags.
1248  *
1249  * CONTEXT:
1250  * spin_lock_irq(pool->lock).
1251  */
1252 static void insert_work(struct pool_workqueue *pwq, struct work_struct *work,
1253                         struct list_head *head, unsigned int extra_flags)
1254 {
1255         struct worker_pool *pool = pwq->pool;
1256
1257         /* we own @work, set data and link */
1258         set_work_pwq(work, pwq, extra_flags);
1259         list_add_tail(&work->entry, head);
1260         get_pwq(pwq);
1261
1262         /*
1263          * Ensure either wq_worker_sleeping() sees the above
1264          * list_add_tail() or we see zero nr_running to avoid workers lying
1265          * around lazily while there are works to be processed.
1266          */
1267         smp_mb();
1268
1269         if (__need_more_worker(pool))
1270                 wake_up_worker(pool);
1271 }
1272
1273 /*
1274  * Test whether @work is being queued from another work executing on the
1275  * same workqueue.
1276  */
1277 static bool is_chained_work(struct workqueue_struct *wq)
1278 {
1279         struct worker *worker;
1280
1281         worker = current_wq_worker();
1282         /*
1283          * Return %true iff I'm a worker execuing a work item on @wq.  If
1284          * I'm @worker, it's safe to dereference it without locking.
1285          */
1286         return worker && worker->current_pwq->wq == wq;
1287 }
1288
1289 static void __queue_work(int cpu, struct workqueue_struct *wq,
1290                          struct work_struct *work)
1291 {
1292         struct pool_workqueue *pwq;
1293         struct worker_pool *last_pool;
1294         struct list_head *worklist;
1295         unsigned int work_flags;
1296         unsigned int req_cpu = cpu;
1297
1298         /*
1299          * While a work item is PENDING && off queue, a task trying to
1300          * steal the PENDING will busy-loop waiting for it to either get
1301          * queued or lose PENDING.  Grabbing PENDING and queueing should
1302          * happen with IRQ disabled.
1303          */
1304         WARN_ON_ONCE(!irqs_disabled());
1305
1306         debug_work_activate(work);
1307
1308         /* if dying, only works from the same workqueue are allowed */
1309         if (unlikely(wq->flags & __WQ_DRAINING) &&
1310             WARN_ON_ONCE(!is_chained_work(wq)))
1311                 return;
1312 retry:
1313         if (req_cpu == WORK_CPU_UNBOUND)
1314                 cpu = raw_smp_processor_id();
1315
1316         /* pwq which will be used unless @work is executing elsewhere */
1317         if (!(wq->flags & WQ_UNBOUND))
1318                 pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
1319         else
1320                 pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
1321
1322         /*
1323          * If @work was previously on a different pool, it might still be
1324          * running there, in which case the work needs to be queued on that
1325          * pool to guarantee non-reentrancy.
1326          */
1327         last_pool = get_work_pool(work);
1328         if (last_pool && last_pool != pwq->pool) {
1329                 struct worker *worker;
1330
1331                 spin_lock(&last_pool->lock);
1332
1333                 worker = find_worker_executing_work(last_pool, work);
1334
1335                 if (worker && worker->current_pwq->wq == wq) {
1336                         pwq = worker->current_pwq;
1337                 } else {
1338                         /* meh... not running there, queue here */
1339                         spin_unlock(&last_pool->lock);
1340                         spin_lock(&pwq->pool->lock);
1341                 }
1342         } else {
1343                 spin_lock(&pwq->pool->lock);
1344         }
1345
1346         /*
1347          * pwq is determined and locked.  For unbound pools, we could have
1348          * raced with pwq release and it could already be dead.  If its
1349          * refcnt is zero, repeat pwq selection.  Note that pwqs never die
1350          * without another pwq replacing it in the numa_pwq_tbl or while
1351          * work items are executing on it, so the retrying is guaranteed to
1352          * make forward-progress.
1353          */
1354         if (unlikely(!pwq->refcnt)) {
1355                 if (wq->flags & WQ_UNBOUND) {
1356                         spin_unlock(&pwq->pool->lock);
1357                         cpu_relax();
1358                         goto retry;
1359                 }
1360                 /* oops */
1361                 WARN_ONCE(true, "workqueue: per-cpu pwq for %s on cpu%d has 0 refcnt",
1362                           wq->name, cpu);
1363         }
1364
1365         /* pwq determined, queue */
1366         trace_workqueue_queue_work(req_cpu, pwq, work);
1367
1368         if (WARN_ON(!list_empty(&work->entry))) {
1369                 spin_unlock(&pwq->pool->lock);
1370                 return;
1371         }
1372
1373         pwq->nr_in_flight[pwq->work_color]++;
1374         work_flags = work_color_to_flags(pwq->work_color);
1375
1376         if (likely(pwq->nr_active < pwq->max_active)) {
1377                 trace_workqueue_activate_work(work);
1378                 pwq->nr_active++;
1379                 worklist = &pwq->pool->worklist;
1380         } else {
1381                 work_flags |= WORK_STRUCT_DELAYED;
1382                 worklist = &pwq->delayed_works;
1383         }
1384
1385         insert_work(pwq, work, worklist, work_flags);
1386
1387         spin_unlock(&pwq->pool->lock);
1388 }
1389
1390 /**
1391  * queue_work_on - queue work on specific cpu
1392  * @cpu: CPU number to execute work on
1393  * @wq: workqueue to use
1394  * @work: work to queue
1395  *
1396  * Returns %false if @work was already on a queue, %true otherwise.
1397  *
1398  * We queue the work to a specific CPU, the caller must ensure it
1399  * can't go away.
1400  */
1401 bool queue_work_on(int cpu, struct workqueue_struct *wq,
1402                    struct work_struct *work)
1403 {
1404         bool ret = false;
1405         unsigned long flags;
1406
1407         local_irq_save(flags);
1408
1409         if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
1410                 __queue_work(cpu, wq, work);
1411                 ret = true;
1412         }
1413
1414         local_irq_restore(flags);
1415         return ret;
1416 }
1417 EXPORT_SYMBOL(queue_work_on);
1418
1419 void delayed_work_timer_fn(unsigned long __data)
1420 {
1421         struct delayed_work *dwork = (struct delayed_work *)__data;
1422
1423         /* should have been called from irqsafe timer with irq already off */
1424         __queue_work(dwork->cpu, dwork->wq, &dwork->work);
1425 }
1426 EXPORT_SYMBOL(delayed_work_timer_fn);
1427
1428 static void __queue_delayed_work(int cpu, struct workqueue_struct *wq,
1429                                 struct delayed_work *dwork, unsigned long delay)
1430 {
1431         struct timer_list *timer = &dwork->timer;
1432         struct work_struct *work = &dwork->work;
1433
1434         WARN_ON_ONCE(timer->function != delayed_work_timer_fn ||
1435                      timer->data != (unsigned long)dwork);
1436         WARN_ON_ONCE(timer_pending(timer));
1437         WARN_ON_ONCE(!list_empty(&work->entry));
1438
1439         /*
1440          * If @delay is 0, queue @dwork->work immediately.  This is for
1441          * both optimization and correctness.  The earliest @timer can
1442          * expire is on the closest next tick and delayed_work users depend
1443          * on that there's no such delay when @delay is 0.
1444          */
1445         if (!delay) {
1446                 __queue_work(cpu, wq, &dwork->work);
1447                 return;
1448         }
1449
1450         timer_stats_timer_set_start_info(&dwork->timer);
1451
1452         dwork->wq = wq;
1453         dwork->cpu = cpu;
1454         timer->expires = jiffies + delay;
1455
1456         if (unlikely(cpu != WORK_CPU_UNBOUND))
1457                 add_timer_on(timer, cpu);
1458         else
1459                 add_timer(timer);
1460 }
1461
1462 /**
1463  * queue_delayed_work_on - queue work on specific CPU after delay
1464  * @cpu: CPU number to execute work on
1465  * @wq: workqueue to use
1466  * @dwork: work to queue
1467  * @delay: number of jiffies to wait before queueing
1468  *
1469  * Returns %false if @work was already on a queue, %true otherwise.  If
1470  * @delay is zero and @dwork is idle, it will be scheduled for immediate
1471  * execution.
1472  */
1473 bool queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
1474                            struct delayed_work *dwork, unsigned long delay)
1475 {
1476         struct work_struct *work = &dwork->work;
1477         bool ret = false;
1478         unsigned long flags;
1479
1480         /* read the comment in __queue_work() */
1481         local_irq_save(flags);
1482
1483         if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
1484                 __queue_delayed_work(cpu, wq, dwork, delay);
1485                 ret = true;
1486         }
1487
1488         local_irq_restore(flags);
1489         return ret;
1490 }
1491 EXPORT_SYMBOL(queue_delayed_work_on);
1492
1493 /**
1494  * mod_delayed_work_on - modify delay of or queue a delayed work on specific CPU
1495  * @cpu: CPU number to execute work on
1496  * @wq: workqueue to use
1497  * @dwork: work to queue
1498  * @delay: number of jiffies to wait before queueing
1499  *
1500  * If @dwork is idle, equivalent to queue_delayed_work_on(); otherwise,
1501  * modify @dwork's timer so that it expires after @delay.  If @delay is
1502  * zero, @work is guaranteed to be scheduled immediately regardless of its
1503  * current state.
1504  *
1505  * Returns %false if @dwork was idle and queued, %true if @dwork was
1506  * pending and its timer was modified.
1507  *
1508  * This function is safe to call from any context including IRQ handler.
1509  * See try_to_grab_pending() for details.
1510  */
1511 bool mod_delayed_work_on(int cpu, struct workqueue_struct *wq,
1512                          struct delayed_work *dwork, unsigned long delay)
1513 {
1514         unsigned long flags;
1515         int ret;
1516
1517         do {
1518                 ret = try_to_grab_pending(&dwork->work, true, &flags);
1519         } while (unlikely(ret == -EAGAIN));
1520
1521         if (likely(ret >= 0)) {
1522                 __queue_delayed_work(cpu, wq, dwork, delay);
1523                 local_irq_restore(flags);
1524         }
1525
1526         /* -ENOENT from try_to_grab_pending() becomes %true */
1527         return ret;
1528 }
1529 EXPORT_SYMBOL_GPL(mod_delayed_work_on);
1530
1531 /**
1532  * worker_enter_idle - enter idle state
1533  * @worker: worker which is entering idle state
1534  *
1535  * @worker is entering idle state.  Update stats and idle timer if
1536  * necessary.
1537  *
1538  * LOCKING:
1539  * spin_lock_irq(pool->lock).
1540  */
1541 static void worker_enter_idle(struct worker *worker)
1542 {
1543         struct worker_pool *pool = worker->pool;
1544
1545         if (WARN_ON_ONCE(worker->flags & WORKER_IDLE) ||
1546             WARN_ON_ONCE(!list_empty(&worker->entry) &&
1547                          (worker->hentry.next || worker->hentry.pprev)))
1548                 return;
1549
1550         /* can't use worker_set_flags(), also called from start_worker() */
1551         worker->flags |= WORKER_IDLE;
1552         pool->nr_idle++;
1553         worker->last_active = jiffies;
1554
1555         /* idle_list is LIFO */
1556         list_add(&worker->entry, &pool->idle_list);
1557
1558         if (too_many_workers(pool) && !timer_pending(&pool->idle_timer))
1559                 mod_timer(&pool->idle_timer, jiffies + IDLE_WORKER_TIMEOUT);
1560
1561         /*
1562          * Sanity check nr_running.  Because wq_unbind_fn() releases
1563          * pool->lock between setting %WORKER_UNBOUND and zapping
1564          * nr_running, the warning may trigger spuriously.  Check iff
1565          * unbind is not in progress.
1566          */
1567         WARN_ON_ONCE(!(pool->flags & POOL_DISASSOCIATED) &&
1568                      pool->nr_workers == pool->nr_idle &&
1569                      atomic_read(&pool->nr_running));
1570 }
1571
1572 /**
1573  * worker_leave_idle - leave idle state
1574  * @worker: worker which is leaving idle state
1575  *
1576  * @worker is leaving idle state.  Update stats.
1577  *
1578  * LOCKING:
1579  * spin_lock_irq(pool->lock).
1580  */
1581 static void worker_leave_idle(struct worker *worker)
1582 {
1583         struct worker_pool *pool = worker->pool;
1584
1585         if (WARN_ON_ONCE(!(worker->flags & WORKER_IDLE)))
1586                 return;
1587         worker_clr_flags(worker, WORKER_IDLE);
1588         pool->nr_idle--;
1589         list_del_init(&worker->entry);
1590 }
1591
1592 /**
1593  * worker_maybe_bind_and_lock - try to bind %current to worker_pool and lock it
1594  * @pool: target worker_pool
1595  *
1596  * Bind %current to the cpu of @pool if it is associated and lock @pool.
1597  *
1598  * Works which are scheduled while the cpu is online must at least be
1599  * scheduled to a worker which is bound to the cpu so that if they are
1600  * flushed from cpu callbacks while cpu is going down, they are
1601  * guaranteed to execute on the cpu.
1602  *
1603  * This function is to be used by unbound workers and rescuers to bind
1604  * themselves to the target cpu and may race with cpu going down or
1605  * coming online.  kthread_bind() can't be used because it may put the
1606  * worker to already dead cpu and set_cpus_allowed_ptr() can't be used
1607  * verbatim as it's best effort and blocking and pool may be
1608  * [dis]associated in the meantime.
1609  *
1610  * This function tries set_cpus_allowed() and locks pool and verifies the
1611  * binding against %POOL_DISASSOCIATED which is set during
1612  * %CPU_DOWN_PREPARE and cleared during %CPU_ONLINE, so if the worker
1613  * enters idle state or fetches works without dropping lock, it can
1614  * guarantee the scheduling requirement described in the first paragraph.
1615  *
1616  * CONTEXT:
1617  * Might sleep.  Called without any lock but returns with pool->lock
1618  * held.
1619  *
1620  * RETURNS:
1621  * %true if the associated pool is online (@worker is successfully
1622  * bound), %false if offline.
1623  */
1624 static bool worker_maybe_bind_and_lock(struct worker_pool *pool)
1625 __acquires(&pool->lock)
1626 {
1627         while (true) {
1628                 /*
1629                  * The following call may fail, succeed or succeed
1630                  * without actually migrating the task to the cpu if
1631                  * it races with cpu hotunplug operation.  Verify
1632                  * against POOL_DISASSOCIATED.
1633                  */
1634                 if (!(pool->flags & POOL_DISASSOCIATED))
1635                         set_cpus_allowed_ptr(current, pool->attrs->cpumask);
1636
1637                 spin_lock_irq(&pool->lock);
1638                 if (pool->flags & POOL_DISASSOCIATED)
1639                         return false;
1640                 if (task_cpu(current) == pool->cpu &&
1641                     cpumask_equal(&current->cpus_allowed, pool->attrs->cpumask))
1642                         return true;
1643                 spin_unlock_irq(&pool->lock);
1644
1645                 /*
1646                  * We've raced with CPU hot[un]plug.  Give it a breather
1647                  * and retry migration.  cond_resched() is required here;
1648                  * otherwise, we might deadlock against cpu_stop trying to
1649                  * bring down the CPU on non-preemptive kernel.
1650                  */
1651                 cpu_relax();
1652                 cond_resched();
1653         }
1654 }
1655
1656 static struct worker *alloc_worker(void)
1657 {
1658         struct worker *worker;
1659
1660         worker = kzalloc(sizeof(*worker), GFP_KERNEL);
1661         if (worker) {
1662                 INIT_LIST_HEAD(&worker->entry);
1663                 INIT_LIST_HEAD(&worker->scheduled);
1664                 /* on creation a worker is in !idle && prep state */
1665                 worker->flags = WORKER_PREP;
1666         }
1667         return worker;
1668 }
1669
1670 /**
1671  * create_worker - create a new workqueue worker
1672  * @pool: pool the new worker will belong to
1673  *
1674  * Create a new worker which is bound to @pool.  The returned worker
1675  * can be started by calling start_worker() or destroyed using
1676  * destroy_worker().
1677  *
1678  * CONTEXT:
1679  * Might sleep.  Does GFP_KERNEL allocations.
1680  *
1681  * RETURNS:
1682  * Pointer to the newly created worker.
1683  */
1684 static struct worker *create_worker(struct worker_pool *pool)
1685 {
1686         struct worker *worker = NULL;
1687         int id = -1;
1688         char id_buf[16];
1689
1690         lockdep_assert_held(&pool->manager_mutex);
1691
1692         /*
1693          * ID is needed to determine kthread name.  Allocate ID first
1694          * without installing the pointer.
1695          */
1696         idr_preload(GFP_KERNEL);
1697         spin_lock_irq(&pool->lock);
1698
1699         id = idr_alloc(&pool->worker_idr, NULL, 0, 0, GFP_NOWAIT);
1700
1701         spin_unlock_irq(&pool->lock);
1702         idr_preload_end();
1703         if (id < 0)
1704                 goto fail;
1705
1706         worker = alloc_worker();
1707         if (!worker)
1708                 goto fail;
1709
1710         worker->pool = pool;
1711         worker->id = id;
1712
1713         if (pool->cpu >= 0)
1714                 snprintf(id_buf, sizeof(id_buf), "%d:%d%s", pool->cpu, id,
1715                          pool->attrs->nice < 0  ? "H" : "");
1716         else
1717                 snprintf(id_buf, sizeof(id_buf), "u%d:%d", pool->id, id);
1718
1719         worker->task = kthread_create_on_node(worker_thread, worker, pool->node,
1720                                               "kworker/%s", id_buf);
1721         if (IS_ERR(worker->task))
1722                 goto fail;
1723
1724         /*
1725          * set_cpus_allowed_ptr() will fail if the cpumask doesn't have any
1726          * online CPUs.  It'll be re-applied when any of the CPUs come up.
1727          */
1728         set_user_nice(worker->task, pool->attrs->nice);
1729         set_cpus_allowed_ptr(worker->task, pool->attrs->cpumask);
1730
1731         /* prevent userland from meddling with cpumask of workqueue workers */
1732         worker->task->flags |= PF_NO_SETAFFINITY;
1733
1734         /*
1735          * The caller is responsible for ensuring %POOL_DISASSOCIATED
1736          * remains stable across this function.  See the comments above the
1737          * flag definition for details.
1738          */
1739         if (pool->flags & POOL_DISASSOCIATED)
1740                 worker->flags |= WORKER_UNBOUND;
1741
1742         /* successful, commit the pointer to idr */
1743         spin_lock_irq(&pool->lock);
1744         idr_replace(&pool->worker_idr, worker, worker->id);
1745         spin_unlock_irq(&pool->lock);
1746
1747         return worker;
1748
1749 fail:
1750         if (id >= 0) {
1751                 spin_lock_irq(&pool->lock);
1752                 idr_remove(&pool->worker_idr, id);
1753                 spin_unlock_irq(&pool->lock);
1754         }
1755         kfree(worker);
1756         return NULL;
1757 }
1758
1759 /**
1760  * start_worker - start a newly created worker
1761  * @worker: worker to start
1762  *
1763  * Make the pool aware of @worker and start it.
1764  *
1765  * CONTEXT:
1766  * spin_lock_irq(pool->lock).
1767  */
1768 static void start_worker(struct worker *worker)
1769 {
1770         worker->flags |= WORKER_STARTED;
1771         worker->pool->nr_workers++;
1772         worker_enter_idle(worker);
1773         wake_up_process(worker->task);
1774 }
1775
1776 /**
1777  * create_and_start_worker - create and start a worker for a pool
1778  * @pool: the target pool
1779  *
1780  * Grab the managership of @pool and create and start a new worker for it.
1781  */
1782 static int create_and_start_worker(struct worker_pool *pool)
1783 {
1784         struct worker *worker;
1785
1786         mutex_lock(&pool->manager_mutex);
1787
1788         worker = create_worker(pool);
1789         if (worker) {
1790                 spin_lock_irq(&pool->lock);
1791                 start_worker(worker);
1792                 spin_unlock_irq(&pool->lock);
1793         }
1794
1795         mutex_unlock(&pool->manager_mutex);
1796
1797         return worker ? 0 : -ENOMEM;
1798 }
1799
1800 /**
1801  * destroy_worker - destroy a workqueue worker
1802  * @worker: worker to be destroyed
1803  *
1804  * Destroy @worker and adjust @pool stats accordingly.
1805  *
1806  * CONTEXT:
1807  * spin_lock_irq(pool->lock) which is released and regrabbed.
1808  */
1809 static void destroy_worker(struct worker *worker)
1810 {
1811         struct worker_pool *pool = worker->pool;
1812
1813         lockdep_assert_held(&pool->manager_mutex);
1814         lockdep_assert_held(&pool->lock);
1815
1816         /* sanity check frenzy */
1817         if (WARN_ON(worker->current_work) ||
1818             WARN_ON(!list_empty(&worker->scheduled)))
1819                 return;
1820
1821         if (worker->flags & WORKER_STARTED)
1822                 pool->nr_workers--;
1823         if (worker->flags & WORKER_IDLE)
1824                 pool->nr_idle--;
1825
1826         /*
1827          * Once WORKER_DIE is set, the kworker may destroy itself at any
1828          * point.  Pin to ensure the task stays until we're done with it.
1829          */
1830         get_task_struct(worker->task);
1831
1832         list_del_init(&worker->entry);
1833         worker->flags |= WORKER_DIE;
1834
1835         idr_remove(&pool->worker_idr, worker->id);
1836
1837         spin_unlock_irq(&pool->lock);
1838
1839         kthread_stop(worker->task);
1840         put_task_struct(worker->task);
1841         kfree(worker);
1842
1843         spin_lock_irq(&pool->lock);
1844 }
1845
1846 static void idle_worker_timeout(unsigned long __pool)
1847 {
1848         struct worker_pool *pool = (void *)__pool;
1849
1850         spin_lock_irq(&pool->lock);
1851
1852         if (too_many_workers(pool)) {
1853                 struct worker *worker;
1854                 unsigned long expires;
1855
1856                 /* idle_list is kept in LIFO order, check the last one */
1857                 worker = list_entry(pool->idle_list.prev, struct worker, entry);
1858                 expires = worker->last_active + IDLE_WORKER_TIMEOUT;
1859
1860                 if (time_before(jiffies, expires))
1861                         mod_timer(&pool->idle_timer, expires);
1862                 else {
1863                         /* it's been idle for too long, wake up manager */
1864                         pool->flags |= POOL_MANAGE_WORKERS;
1865                         wake_up_worker(pool);
1866                 }
1867         }
1868
1869         spin_unlock_irq(&pool->lock);
1870 }
1871
1872 static void send_mayday(struct work_struct *work)
1873 {
1874         struct pool_workqueue *pwq = get_work_pwq(work);
1875         struct workqueue_struct *wq = pwq->wq;
1876
1877         lockdep_assert_held(&wq_mayday_lock);
1878
1879         if (!wq->rescuer)
1880                 return;
1881
1882         /* mayday mayday mayday */
1883         if (list_empty(&pwq->mayday_node)) {
1884                 /*
1885                  * If @pwq is for an unbound wq, its base ref may be put at
1886                  * any time due to an attribute change.  Pin @pwq until the
1887                  * rescuer is done with it.
1888                  */
1889                 get_pwq(pwq);
1890                 list_add_tail(&pwq->mayday_node, &wq->maydays);
1891                 wake_up_process(wq->rescuer->task);
1892         }
1893 }
1894
1895 static void pool_mayday_timeout(unsigned long __pool)
1896 {
1897         struct worker_pool *pool = (void *)__pool;
1898         struct work_struct *work;
1899
1900         spin_lock_irq(&wq_mayday_lock);         /* for wq->maydays */
1901         spin_lock(&pool->lock);
1902
1903         if (need_to_create_worker(pool)) {
1904                 /*
1905                  * We've been trying to create a new worker but
1906                  * haven't been successful.  We might be hitting an
1907                  * allocation deadlock.  Send distress signals to
1908                  * rescuers.
1909                  */
1910                 list_for_each_entry(work, &pool->worklist, entry)
1911                         send_mayday(work);
1912         }
1913
1914         spin_unlock(&pool->lock);
1915         spin_unlock_irq(&wq_mayday_lock);
1916
1917         mod_timer(&pool->mayday_timer, jiffies + MAYDAY_INTERVAL);
1918 }
1919
1920 /**
1921  * maybe_create_worker - create a new worker if necessary
1922  * @pool: pool to create a new worker for
1923  *
1924  * Create a new worker for @pool if necessary.  @pool is guaranteed to
1925  * have at least one idle worker on return from this function.  If
1926  * creating a new worker takes longer than MAYDAY_INTERVAL, mayday is
1927  * sent to all rescuers with works scheduled on @pool to resolve
1928  * possible allocation deadlock.
1929  *
1930  * On return, need_to_create_worker() is guaranteed to be %false and
1931  * may_start_working() %true.
1932  *
1933  * LOCKING:
1934  * spin_lock_irq(pool->lock) which may be released and regrabbed
1935  * multiple times.  Does GFP_KERNEL allocations.  Called only from
1936  * manager.
1937  *
1938  * RETURNS:
1939  * %false if no action was taken and pool->lock stayed locked, %true
1940  * otherwise.
1941  */
1942 static bool maybe_create_worker(struct worker_pool *pool)
1943 __releases(&pool->lock)
1944 __acquires(&pool->lock)
1945 {
1946         if (!need_to_create_worker(pool))
1947                 return false;
1948 restart:
1949         spin_unlock_irq(&pool->lock);
1950
1951         /* if we don't make progress in MAYDAY_INITIAL_TIMEOUT, call for help */
1952         mod_timer(&pool->mayday_timer, jiffies + MAYDAY_INITIAL_TIMEOUT);
1953
1954         while (true) {
1955                 struct worker *worker;
1956
1957                 worker = create_worker(pool);
1958                 if (worker) {
1959                         del_timer_sync(&pool->mayday_timer);
1960                         spin_lock_irq(&pool->lock);
1961                         start_worker(worker);
1962                         if (WARN_ON_ONCE(need_to_create_worker(pool)))
1963                                 goto restart;
1964                         return true;
1965                 }
1966
1967                 if (!need_to_create_worker(pool))
1968                         break;
1969
1970                 __set_current_state(TASK_INTERRUPTIBLE);
1971                 schedule_timeout(CREATE_COOLDOWN);
1972
1973                 if (!need_to_create_worker(pool))
1974                         break;
1975         }
1976
1977         del_timer_sync(&pool->mayday_timer);
1978         spin_lock_irq(&pool->lock);
1979         if (need_to_create_worker(pool))
1980                 goto restart;
1981         return true;
1982 }
1983
1984 /**
1985  * maybe_destroy_worker - destroy workers which have been idle for a while
1986  * @pool: pool to destroy workers for
1987  *
1988  * Destroy @pool workers which have been idle for longer than
1989  * IDLE_WORKER_TIMEOUT.
1990  *
1991  * LOCKING:
1992  * spin_lock_irq(pool->lock) which may be released and regrabbed
1993  * multiple times.  Called only from manager.
1994  *
1995  * RETURNS:
1996  * %false if no action was taken and pool->lock stayed locked, %true
1997  * otherwise.
1998  */
1999 static bool maybe_destroy_workers(struct worker_pool *pool)
2000 {
2001         bool ret = false;
2002
2003         while (too_many_workers(pool)) {
2004                 struct worker *worker;
2005                 unsigned long expires;
2006
2007                 worker = list_entry(pool->idle_list.prev, struct worker, entry);
2008                 expires = worker->last_active + IDLE_WORKER_TIMEOUT;
2009
2010                 if (time_before(jiffies, expires)) {
2011                         mod_timer(&pool->idle_timer, expires);
2012                         break;
2013                 }
2014
2015                 destroy_worker(worker);
2016                 ret = true;
2017         }
2018
2019         return ret;
2020 }
2021
2022 /**
2023  * manage_workers - manage worker pool
2024  * @worker: self
2025  *
2026  * Assume the manager role and manage the worker pool @worker belongs
2027  * to.  At any given time, there can be only zero or one manager per
2028  * pool.  The exclusion is handled automatically by this function.
2029  *
2030  * The caller can safely start processing works on false return.  On
2031  * true return, it's guaranteed that need_to_create_worker() is false
2032  * and may_start_working() is true.
2033  *
2034  * CONTEXT:
2035  * spin_lock_irq(pool->lock) which may be released and regrabbed
2036  * multiple times.  Does GFP_KERNEL allocations.
2037  *
2038  * RETURNS:
2039  * spin_lock_irq(pool->lock) which may be released and regrabbed
2040  * multiple times.  Does GFP_KERNEL allocations.
2041  */
2042 static bool manage_workers(struct worker *worker)
2043 {
2044         struct worker_pool *pool = worker->pool;
2045         bool ret = false;
2046
2047         /*
2048          * Managership is governed by two mutexes - manager_arb and
2049          * manager_mutex.  manager_arb handles arbitration of manager role.
2050          * Anyone who successfully grabs manager_arb wins the arbitration
2051          * and becomes the manager.  mutex_trylock() on pool->manager_arb
2052          * failure while holding pool->lock reliably indicates that someone
2053          * else is managing the pool and the worker which failed trylock
2054          * can proceed to executing work items.  This means that anyone
2055          * grabbing manager_arb is responsible for actually performing
2056          * manager duties.  If manager_arb is grabbed and released without
2057          * actual management, the pool may stall indefinitely.
2058          *
2059          * manager_mutex is used for exclusion of actual management
2060          * operations.  The holder of manager_mutex can be sure that none
2061          * of management operations, including creation and destruction of
2062          * workers, won't take place until the mutex is released.  Because
2063          * manager_mutex doesn't interfere with manager role arbitration,
2064          * it is guaranteed that the pool's management, while may be
2065          * delayed, won't be disturbed by someone else grabbing
2066          * manager_mutex.
2067          */
2068         if (!mutex_trylock(&pool->manager_arb))
2069                 return ret;
2070
2071         /*
2072          * With manager arbitration won, manager_mutex would be free in
2073          * most cases.  trylock first without dropping @pool->lock.
2074          */
2075         if (unlikely(!mutex_trylock(&pool->manager_mutex))) {
2076                 spin_unlock_irq(&pool->lock);
2077                 mutex_lock(&pool->manager_mutex);
2078                 spin_lock_irq(&pool->lock);
2079                 ret = true;
2080         }
2081
2082         pool->flags &= ~POOL_MANAGE_WORKERS;
2083
2084         /*
2085          * Destroy and then create so that may_start_working() is true
2086          * on return.
2087          */
2088         ret |= maybe_destroy_workers(pool);
2089         ret |= maybe_create_worker(pool);
2090
2091         mutex_unlock(&pool->manager_mutex);
2092         mutex_unlock(&pool->manager_arb);
2093         return ret;
2094 }
2095
2096 /**
2097  * process_one_work - process single work
2098  * @worker: self
2099  * @work: work to process
2100  *
2101  * Process @work.  This function contains all the logics necessary to
2102  * process a single work including synchronization against and
2103  * interaction with other workers on the same cpu, queueing and
2104  * flushing.  As long as context requirement is met, any worker can
2105  * call this function to process a work.
2106  *
2107  * CONTEXT:
2108  * spin_lock_irq(pool->lock) which is released and regrabbed.
2109  */
2110 static void process_one_work(struct worker *worker, struct work_struct *work)
2111 __releases(&pool->lock)
2112 __acquires(&pool->lock)
2113 {
2114         struct pool_workqueue *pwq = get_work_pwq(work);
2115         struct worker_pool *pool = worker->pool;
2116         bool cpu_intensive = pwq->wq->flags & WQ_CPU_INTENSIVE;
2117         int work_color;
2118         struct worker *collision;
2119 #ifdef CONFIG_LOCKDEP
2120         /*
2121          * It is permissible to free the struct work_struct from
2122          * inside the function that is called from it, this we need to
2123          * take into account for lockdep too.  To avoid bogus "held
2124          * lock freed" warnings as well as problems when looking into
2125          * work->lockdep_map, make a copy and use that here.
2126          */
2127         struct lockdep_map lockdep_map;
2128
2129         lockdep_copy_map(&lockdep_map, &work->lockdep_map);
2130 #endif
2131         /*
2132          * Ensure we're on the correct CPU.  DISASSOCIATED test is
2133          * necessary to avoid spurious warnings from rescuers servicing the
2134          * unbound or a disassociated pool.
2135          */
2136         WARN_ON_ONCE(!(worker->flags & WORKER_UNBOUND) &&
2137                      !(pool->flags & POOL_DISASSOCIATED) &&
2138                      raw_smp_processor_id() != pool->cpu);
2139
2140         /*
2141          * A single work shouldn't be executed concurrently by
2142          * multiple workers on a single cpu.  Check whether anyone is
2143          * already processing the work.  If so, defer the work to the
2144          * currently executing one.
2145          */
2146         collision = find_worker_executing_work(pool, work);
2147         if (unlikely(collision)) {
2148                 move_linked_works(work, &collision->scheduled, NULL);
2149                 return;
2150         }
2151
2152         /* claim and dequeue */
2153         debug_work_deactivate(work);
2154         hash_add(pool->busy_hash, &worker->hentry, (unsigned long)work);
2155         worker->current_work = work;
2156         worker->current_func = work->func;
2157         worker->current_pwq = pwq;
2158         work_color = get_work_color(work);
2159
2160         list_del_init(&work->entry);
2161
2162         /*
2163          * CPU intensive works don't participate in concurrency
2164          * management.  They're the scheduler's responsibility.
2165          */
2166         if (unlikely(cpu_intensive))
2167                 worker_set_flags(worker, WORKER_CPU_INTENSIVE, true);
2168
2169         /*
2170          * Unbound pool isn't concurrency managed and work items should be
2171          * executed ASAP.  Wake up another worker if necessary.
2172          */
2173         if ((worker->flags & WORKER_UNBOUND) && need_more_worker(pool))
2174                 wake_up_worker(pool);
2175
2176         /*
2177          * Record the last pool and clear PENDING which should be the last
2178          * update to @work.  Also, do this inside @pool->lock so that
2179          * PENDING and queued state changes happen together while IRQ is
2180          * disabled.
2181          */
2182         set_work_pool_and_clear_pending(work, pool->id);
2183
2184         spin_unlock_irq(&pool->lock);
2185
2186         lock_map_acquire_read(&pwq->wq->lockdep_map);
2187         lock_map_acquire(&lockdep_map);
2188         trace_workqueue_execute_start(work);
2189         worker->current_func(work);
2190         /*
2191          * While we must be careful to not use "work" after this, the trace
2192          * point will only record its address.
2193          */
2194         trace_workqueue_execute_end(work);
2195         lock_map_release(&lockdep_map);
2196         lock_map_release(&pwq->wq->lockdep_map);
2197
2198         if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
2199                 pr_err("BUG: workqueue leaked lock or atomic: %s/0x%08x/%d\n"
2200                        "     last function: %pf\n",
2201                        current->comm, preempt_count(), task_pid_nr(current),
2202                        worker->current_func);
2203                 debug_show_held_locks(current);
2204                 dump_stack();
2205         }
2206
2207         /*
2208          * The following prevents a kworker from hogging CPU on !PREEMPT
2209          * kernels, where a requeueing work item waiting for something to
2210          * happen could deadlock with stop_machine as such work item could
2211          * indefinitely requeue itself while all other CPUs are trapped in
2212          * stop_machine.
2213          */
2214         cond_resched();
2215
2216         spin_lock_irq(&pool->lock);
2217
2218         /* clear cpu intensive status */
2219         if (unlikely(cpu_intensive))
2220                 worker_clr_flags(worker, WORKER_CPU_INTENSIVE);
2221
2222         /* we're done with it, release */
2223         hash_del(&worker->hentry);
2224         worker->current_work = NULL;
2225         worker->current_func = NULL;
2226         worker->current_pwq = NULL;
2227         worker->desc_valid = false;
2228         pwq_dec_nr_in_flight(pwq, work_color);
2229 }
2230
2231 /**
2232  * process_scheduled_works - process scheduled works
2233  * @worker: self
2234  *
2235  * Process all scheduled works.  Please note that the scheduled list
2236  * may change while processing a work, so this function repeatedly
2237  * fetches a work from the top and executes it.
2238  *
2239  * CONTEXT:
2240  * spin_lock_irq(pool->lock) which may be released and regrabbed
2241  * multiple times.
2242  */
2243 static void process_scheduled_works(struct worker *worker)
2244 {
2245         while (!list_empty(&worker->scheduled)) {
2246                 struct work_struct *work = list_first_entry(&worker->scheduled,
2247                                                 struct work_struct, entry);
2248                 process_one_work(worker, work);
2249         }
2250 }
2251
2252 /**
2253  * worker_thread - the worker thread function
2254  * @__worker: self
2255  *
2256  * The worker thread function.  All workers belong to a worker_pool -
2257  * either a per-cpu one or dynamic unbound one.  These workers process all
2258  * work items regardless of their specific target workqueue.  The only
2259  * exception is work items which belong to workqueues with a rescuer which
2260  * will be explained in rescuer_thread().
2261  */
2262 static int worker_thread(void *__worker)
2263 {
2264         struct worker *worker = __worker;
2265         struct worker_pool *pool = worker->pool;
2266
2267         /* tell the scheduler that this is a workqueue worker */
2268         worker->task->flags |= PF_WQ_WORKER;
2269 woke_up:
2270         spin_lock_irq(&pool->lock);
2271
2272         /* am I supposed to die? */
2273         if (unlikely(worker->flags & WORKER_DIE)) {
2274                 spin_unlock_irq(&pool->lock);
2275                 WARN_ON_ONCE(!list_empty(&worker->entry));
2276                 worker->task->flags &= ~PF_WQ_WORKER;
2277                 return 0;
2278         }
2279
2280         worker_leave_idle(worker);
2281 recheck:
2282         /* no more worker necessary? */
2283         if (!need_more_worker(pool))
2284                 goto sleep;
2285
2286         /* do we need to manage? */
2287         if (unlikely(!may_start_working(pool)) && manage_workers(worker))
2288                 goto recheck;
2289
2290         /*
2291          * ->scheduled list can only be filled while a worker is
2292          * preparing to process a work or actually processing it.
2293          * Make sure nobody diddled with it while I was sleeping.
2294          */
2295         WARN_ON_ONCE(!list_empty(&worker->scheduled));
2296
2297         /*
2298          * Finish PREP stage.  We're guaranteed to have at least one idle
2299          * worker or that someone else has already assumed the manager
2300          * role.  This is where @worker starts participating in concurrency
2301          * management if applicable and concurrency management is restored
2302          * after being rebound.  See rebind_workers() for details.
2303          */
2304         worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);
2305
2306         do {
2307                 struct work_struct *work =
2308                         list_first_entry(&pool->worklist,
2309                                          struct work_struct, entry);
2310
2311                 if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
2312                         /* optimization path, not strictly necessary */
2313                         process_one_work(worker, work);
2314                         if (unlikely(!list_empty(&worker->scheduled)))
2315                                 process_scheduled_works(worker);
2316                 } else {
2317                         move_linked_works(work, &worker->scheduled, NULL);
2318                         process_scheduled_works(worker);
2319                 }
2320         } while (keep_working(pool));
2321
2322         worker_set_flags(worker, WORKER_PREP, false);
2323 sleep:
2324         if (unlikely(need_to_manage_workers(pool)) && manage_workers(worker))
2325                 goto recheck;
2326
2327         /*
2328          * pool->lock is held and there's no work to process and no need to
2329          * manage, sleep.  Workers are woken up only while holding
2330          * pool->lock or from local cpu, so setting the current state
2331          * before releasing pool->lock is enough to prevent losing any
2332          * event.
2333          */
2334         worker_enter_idle(worker);
2335         __set_current_state(TASK_INTERRUPTIBLE);
2336         spin_unlock_irq(&pool->lock);
2337         schedule();
2338         goto woke_up;
2339 }
2340
2341 /**
2342  * rescuer_thread - the rescuer thread function
2343  * @__rescuer: self
2344  *
2345  * Workqueue rescuer thread function.  There's one rescuer for each
2346  * workqueue which has WQ_MEM_RECLAIM set.
2347  *
2348  * Regular work processing on a pool may block trying to create a new
2349  * worker which uses GFP_KERNEL allocation which has slight chance of
2350  * developing into deadlock if some works currently on the same queue
2351  * need to be processed to satisfy the GFP_KERNEL allocation.  This is
2352  * the problem rescuer solves.
2353  *
2354  * When such condition is possible, the pool summons rescuers of all
2355  * workqueues which have works queued on the pool and let them process
2356  * those works so that forward progress can be guaranteed.
2357  *
2358  * This should happen rarely.
2359  */
2360 static int rescuer_thread(void *__rescuer)
2361 {
2362         struct worker *rescuer = __rescuer;
2363         struct workqueue_struct *wq = rescuer->rescue_wq;
2364         struct list_head *scheduled = &rescuer->scheduled;
2365         bool should_stop;
2366
2367         set_user_nice(current, RESCUER_NICE_LEVEL);
2368
2369         /*
2370          * Mark rescuer as worker too.  As WORKER_PREP is never cleared, it
2371          * doesn't participate in concurrency management.
2372          */
2373         rescuer->task->flags |= PF_WQ_WORKER;
2374 repeat:
2375         set_current_state(TASK_INTERRUPTIBLE);
2376
2377         /*
2378          * By the time the rescuer is requested to stop, the workqueue
2379          * shouldn't have any work pending, but @wq->maydays may still have
2380          * pwq(s) queued.  This can happen by non-rescuer workers consuming
2381          * all the work items before the rescuer got to them.  Go through
2382          * @wq->maydays processing before acting on should_stop so that the
2383          * list is always empty on exit.
2384          */
2385         should_stop = kthread_should_stop();
2386
2387         /* see whether any pwq is asking for help */
2388         spin_lock_irq(&wq_mayday_lock);
2389
2390         while (!list_empty(&wq->maydays)) {
2391                 struct pool_workqueue *pwq = list_first_entry(&wq->maydays,
2392                                         struct pool_workqueue, mayday_node);
2393                 struct worker_pool *pool = pwq->pool;
2394                 struct work_struct *work, *n;
2395
2396                 __set_current_state(TASK_RUNNING);
2397                 list_del_init(&pwq->mayday_node);
2398
2399                 spin_unlock_irq(&wq_mayday_lock);
2400
2401                 /* migrate to the target cpu if possible */
2402                 worker_maybe_bind_and_lock(pool);
2403                 rescuer->pool = pool;
2404
2405                 /*
2406                  * Slurp in all works issued via this workqueue and
2407                  * process'em.
2408                  */
2409                 WARN_ON_ONCE(!list_empty(&rescuer->scheduled));
2410                 list_for_each_entry_safe(work, n, &pool->worklist, entry)
2411                         if (get_work_pwq(work) == pwq)
2412                                 move_linked_works(work, scheduled, &n);
2413
2414                 process_scheduled_works(rescuer);
2415
2416                 /*
2417                  * Put the reference grabbed by send_mayday().  @pool won't
2418                  * go away while we're holding its lock.
2419                  */
2420                 put_pwq(pwq);
2421
2422                 /*
2423                  * Leave this pool.  If keep_working() is %true, notify a
2424                  * regular worker; otherwise, we end up with 0 concurrency
2425                  * and stalling the execution.
2426                  */
2427                 if (keep_working(pool))
2428                         wake_up_worker(pool);
2429
2430                 rescuer->pool = NULL;
2431                 spin_unlock(&pool->lock);
2432                 spin_lock(&wq_mayday_lock);
2433         }
2434
2435         spin_unlock_irq(&wq_mayday_lock);
2436
2437         if (should_stop) {
2438                 __set_current_state(TASK_RUNNING);
2439                 rescuer->task->flags &= ~PF_WQ_WORKER;
2440                 return 0;
2441         }
2442
2443         /* rescuers should never participate in concurrency management */
2444         WARN_ON_ONCE(!(rescuer->flags & WORKER_NOT_RUNNING));
2445         schedule();
2446         goto repeat;
2447 }
2448
2449 struct wq_barrier {
2450         struct work_struct      work;
2451         struct completion       done;
2452 };
2453
2454 static void wq_barrier_func(struct work_struct *work)
2455 {
2456         struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
2457         complete(&barr->done);
2458 }
2459
2460 /**
2461  * insert_wq_barrier - insert a barrier work
2462  * @pwq: pwq to insert barrier into
2463  * @barr: wq_barrier to insert
2464  * @target: target work to attach @barr to
2465  * @worker: worker currently executing @target, NULL if @target is not executing
2466  *
2467  * @barr is linked to @target such that @barr is completed only after
2468  * @target finishes execution.  Please note that the ordering
2469  * guarantee is observed only with respect to @target and on the local
2470  * cpu.
2471  *
2472  * Currently, a queued barrier can't be canceled.  This is because
2473  * try_to_grab_pending() can't determine whether the work to be
2474  * grabbed is at the head of the queue and thus can't clear LINKED
2475  * flag of the previous work while there must be a valid next work
2476  * after a work with LINKED flag set.
2477  *
2478  * Note that when @worker is non-NULL, @target may be modified
2479  * underneath us, so we can't reliably determine pwq from @target.
2480  *
2481  * CONTEXT:
2482  * spin_lock_irq(pool->lock).
2483  */
2484 static void insert_wq_barrier(struct pool_workqueue *pwq,
2485                               struct wq_barrier *barr,
2486                               struct work_struct *target, struct worker *worker)
2487 {
2488         struct list_head *head;
2489         unsigned int linked = 0;
2490
2491         /*
2492          * debugobject calls are safe here even with pool->lock locked
2493          * as we know for sure that this will not trigger any of the
2494          * checks and call back into the fixup functions where we
2495          * might deadlock.
2496          */
2497         INIT_WORK_ONSTACK(&barr->work, wq_barrier_func);
2498         __set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(&barr->work));
2499         init_completion(&barr->done);
2500
2501         /*
2502          * If @target is currently being executed, schedule the
2503          * barrier to the worker; otherwise, put it after @target.
2504          */
2505         if (worker)
2506                 head = worker->scheduled.next;
2507         else {
2508                 unsigned long *bits = work_data_bits(target);
2509
2510                 head = target->entry.next;
2511                 /* there can already be other linked works, inherit and set */
2512                 linked = *bits & WORK_STRUCT_LINKED;
2513                 __set_bit(WORK_STRUCT_LINKED_BIT, bits);
2514         }
2515
2516         debug_work_activate(&barr->work);
2517         insert_work(pwq, &barr->work, head,
2518                     work_color_to_flags(WORK_NO_COLOR) | linked);
2519 }
2520
2521 /**
2522  * flush_workqueue_prep_pwqs - prepare pwqs for workqueue flushing
2523  * @wq: workqueue being flushed
2524  * @flush_color: new flush color, < 0 for no-op
2525  * @work_color: new work color, < 0 for no-op
2526  *
2527  * Prepare pwqs for workqueue flushing.
2528  *
2529  * If @flush_color is non-negative, flush_color on all pwqs should be
2530  * -1.  If no pwq has in-flight commands at the specified color, all
2531  * pwq->flush_color's stay at -1 and %false is returned.  If any pwq
2532  * has in flight commands, its pwq->flush_color is set to
2533  * @flush_color, @wq->nr_pwqs_to_flush is updated accordingly, pwq
2534  * wakeup logic is armed and %true is returned.
2535  *
2536  * The caller should have initialized @wq->first_flusher prior to
2537  * calling this function with non-negative @flush_color.  If
2538  * @flush_color is negative, no flush color update is done and %false
2539  * is returned.
2540  *
2541  * If @work_color is non-negative, all pwqs should have the same
2542  * work_color which is previous to @work_color and all will be
2543  * advanced to @work_color.
2544  *
2545  * CONTEXT:
2546  * mutex_lock(wq->mutex).
2547  *
2548  * RETURNS:
2549  * %true if @flush_color >= 0 and there's something to flush.  %false
2550  * otherwise.
2551  */
2552 static bool flush_workqueue_prep_pwqs(struct workqueue_struct *wq,
2553                                       int flush_color, int work_color)
2554 {
2555         bool wait = false;
2556         struct pool_workqueue *pwq;
2557
2558         if (flush_color >= 0) {
2559                 WARN_ON_ONCE(atomic_read(&wq->nr_pwqs_to_flush));
2560                 atomic_set(&wq->nr_pwqs_to_flush, 1);
2561         }
2562
2563         for_each_pwq(pwq, wq) {
2564                 struct worker_pool *pool = pwq->pool;
2565
2566                 spin_lock_irq(&pool->lock);
2567
2568                 if (flush_color >= 0) {
2569                         WARN_ON_ONCE(pwq->flush_color != -1);
2570
2571                         if (pwq->nr_in_flight[flush_color]) {
2572                                 pwq->flush_color = flush_color;
2573                                 atomic_inc(&wq->nr_pwqs_to_flush);
2574                                 wait = true;
2575                         }
2576                 }
2577
2578                 if (work_color >= 0) {
2579                         WARN_ON_ONCE(work_color != work_next_color(pwq->work_color));
2580                         pwq->work_color = work_color;
2581                 }
2582
2583                 spin_unlock_irq(&pool->lock);
2584         }
2585
2586         if (flush_color >= 0 && atomic_dec_and_test(&wq->nr_pwqs_to_flush))
2587                 complete(&wq->first_flusher->done);
2588
2589         return wait;
2590 }
2591
2592 /**
2593  * flush_workqueue - ensure that any scheduled work has run to completion.
2594  * @wq: workqueue to flush
2595  *
2596  * This function sleeps until all work items which were queued on entry
2597  * have finished execution, but it is not livelocked by new incoming ones.
2598  */
2599 void flush_workqueue(struct workqueue_struct *wq)
2600 {
2601         struct wq_flusher this_flusher = {
2602                 .list = LIST_HEAD_INIT(this_flusher.list),
2603                 .flush_color = -1,
2604                 .done = COMPLETION_INITIALIZER_ONSTACK(this_flusher.done),
2605         };
2606         int next_color;
2607
2608         lock_map_acquire(&wq->lockdep_map);
2609         lock_map_release(&wq->lockdep_map);
2610
2611         mutex_lock(&wq->mutex);
2612
2613         /*
2614          * Start-to-wait phase
2615          */
2616         next_color = work_next_color(wq->work_color);
2617
2618         if (next_color != wq->flush_color) {
2619                 /*
2620                  * Color space is not full.  The current work_color
2621                  * becomes our flush_color and work_color is advanced
2622                  * by one.
2623                  */
2624                 WARN_ON_ONCE(!list_empty(&wq->flusher_overflow));
2625                 this_flusher.flush_color = wq->work_color;
2626                 wq->work_color = next_color;
2627
2628                 if (!wq->first_flusher) {
2629                         /* no flush in progress, become the first flusher */
2630                         WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);
2631
2632                         wq->first_flusher = &this_flusher;
2633
2634                         if (!flush_workqueue_prep_pwqs(wq, wq->flush_color,
2635                                                        wq->work_color)) {
2636                                 /* nothing to flush, done */
2637                                 wq->flush_color = next_color;
2638                                 wq->first_flusher = NULL;
2639                                 goto out_unlock;
2640                         }
2641                 } else {
2642                         /* wait in queue */
2643                         WARN_ON_ONCE(wq->flush_color == this_flusher.flush_color);
2644                         list_add_tail(&this_flusher.list, &wq->flusher_queue);
2645                         flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
2646                 }
2647         } else {
2648                 /*
2649                  * Oops, color space is full, wait on overflow queue.
2650                  * The next flush completion will assign us
2651                  * flush_color and transfer to flusher_queue.
2652                  */
2653                 list_add_tail(&this_flusher.list, &wq->flusher_overflow);
2654         }
2655
2656         mutex_unlock(&wq->mutex);
2657
2658         wait_for_completion(&this_flusher.done);
2659
2660         /*
2661          * Wake-up-and-cascade phase
2662          *
2663          * First flushers are responsible for cascading flushes and
2664          * handling overflow.  Non-first flushers can simply return.
2665          */
2666         if (wq->first_flusher != &this_flusher)
2667                 return;
2668
2669         mutex_lock(&wq->mutex);
2670
2671         /* we might have raced, check again with mutex held */
2672         if (wq->first_flusher != &this_flusher)
2673                 goto out_unlock;
2674
2675         wq->first_flusher = NULL;
2676
2677         WARN_ON_ONCE(!list_empty(&this_flusher.list));
2678         WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);
2679
2680         while (true) {
2681                 struct wq_flusher *next, *tmp;
2682
2683                 /* complete all the flushers sharing the current flush color */
2684                 list_for_each_entry_safe(next, tmp, &wq->flusher_queue, list) {
2685                         if (next->flush_color != wq->flush_color)
2686                                 break;
2687                         list_del_init(&next->list);
2688                         complete(&next->done);
2689                 }
2690
2691                 WARN_ON_ONCE(!list_empty(&wq->flusher_overflow) &&
2692                              wq->flush_color != work_next_color(wq->work_color));
2693
2694                 /* this flush_color is finished, advance by one */
2695                 wq->flush_color = work_next_color(wq->flush_color);
2696
2697                 /* one color has been freed, handle overflow queue */
2698                 if (!list_empty(&wq->flusher_overflow)) {
2699                         /*
2700                          * Assign the same color to all overflowed
2701                          * flushers, advance work_color and append to
2702                          * flusher_queue.  This is the start-to-wait
2703                          * phase for these overflowed flushers.
2704                          */
2705                         list_for_each_entry(tmp, &wq->flusher_overflow, list)
2706                                 tmp->flush_color = wq->work_color;
2707
2708                         wq->work_color = work_next_color(wq->work_color);
2709
2710                         list_splice_tail_init(&wq->flusher_overflow,
2711                                               &wq->flusher_queue);
2712                         flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
2713                 }
2714
2715                 if (list_empty(&wq->flusher_queue)) {
2716                         WARN_ON_ONCE(wq->flush_color != wq->work_color);
2717                         break;
2718                 }
2719
2720                 /*
2721                  * Need to flush more colors.  Make the next flusher
2722                  * the new first flusher and arm pwqs.
2723                  */
2724                 WARN_ON_ONCE(wq->flush_color == wq->work_color);
2725                 WARN_ON_ONCE(wq->flush_color != next->flush_color);
2726
2727                 list_del_init(&next->list);
2728                 wq->first_flusher = next;
2729
2730                 if (flush_workqueue_prep_pwqs(wq, wq->flush_color, -1))
2731                         break;
2732
2733                 /*
2734                  * Meh... this color is already done, clear first
2735                  * flusher and repeat cascading.
2736                  */
2737                 wq->first_flusher = NULL;
2738         }
2739
2740 out_unlock:
2741         mutex_unlock(&wq->mutex);
2742 }
2743 EXPORT_SYMBOL_GPL(flush_workqueue);
2744
2745 /**
2746  * drain_workqueue - drain a workqueue
2747  * @wq: workqueue to drain
2748  *
2749  * Wait until the workqueue becomes empty.  While draining is in progress,
2750  * only chain queueing is allowed.  IOW, only currently pending or running
2751  * work items on @wq can queue further work items on it.  @wq is flushed
2752  * repeatedly until it becomes empty.  The number of flushing is detemined
2753  * by the depth of chaining and should be relatively short.  Whine if it
2754  * takes too long.
2755  */
2756 void drain_workqueue(struct workqueue_struct *wq)
2757 {
2758         unsigned int flush_cnt = 0;
2759         struct pool_workqueue *pwq;
2760
2761         /*
2762          * __queue_work() needs to test whether there are drainers, is much
2763          * hotter than drain_workqueue() and already looks at @wq->flags.
2764          * Use __WQ_DRAINING so that queue doesn't have to check nr_drainers.
2765          */
2766         mutex_lock(&wq->mutex);
2767         if (!wq->nr_drainers++)
2768                 wq->flags |= __WQ_DRAINING;
2769         mutex_unlock(&wq->mutex);
2770 reflush:
2771         flush_workqueue(wq);
2772
2773         mutex_lock(&wq->mutex);
2774
2775         for_each_pwq(pwq, wq) {
2776                 bool drained;
2777
2778                 spin_lock_irq(&pwq->pool->lock);
2779                 drained = !pwq->nr_active && list_empty(&pwq->delayed_works);
2780                 spin_unlock_irq(&pwq->pool->lock);
2781
2782                 if (drained)
2783                         continue;
2784
2785                 if (++flush_cnt == 10 ||
2786                     (flush_cnt % 100 == 0 && flush_cnt <= 1000))
2787                         pr_warn("workqueue %s: drain_workqueue() isn't complete after %u tries\n",
2788                                 wq->name, flush_cnt);
2789
2790                 mutex_unlock(&wq->mutex);
2791                 goto reflush;
2792         }
2793
2794         if (!--wq->nr_drainers)
2795                 wq->flags &= ~__WQ_DRAINING;
2796         mutex_unlock(&wq->mutex);
2797 }
2798 EXPORT_SYMBOL_GPL(drain_workqueue);
2799
2800 static bool start_flush_work(struct work_struct *work, struct wq_barrier *barr)
2801 {
2802         struct worker *worker = NULL;
2803         struct worker_pool *pool;
2804         struct pool_workqueue *pwq;
2805
2806         might_sleep();
2807
2808         local_irq_disable();
2809         pool = get_work_pool(work);
2810         if (!pool) {
2811                 local_irq_enable();
2812                 return false;
2813         }
2814
2815         spin_lock(&pool->lock);
2816         /* see the comment in try_to_grab_pending() with the same code */
2817         pwq = get_work_pwq(work);
2818         if (pwq) {
2819                 if (unlikely(pwq->pool != pool))
2820                         goto already_gone;
2821         } else {
2822                 worker = find_worker_executing_work(pool, work);
2823                 if (!worker)
2824                         goto already_gone;
2825                 pwq = worker->current_pwq;
2826         }
2827
2828         insert_wq_barrier(pwq, barr, work, worker);
2829         spin_unlock_irq(&pool->lock);
2830
2831         /*
2832          * If @max_active is 1 or rescuer is in use, flushing another work
2833          * item on the same workqueue may lead to deadlock.  Make sure the
2834          * flusher is not running on the same workqueue by verifying write
2835          * access.
2836          */
2837         if (pwq->wq->saved_max_active == 1 || pwq->wq->rescuer)
2838                 lock_map_acquire(&pwq->wq->lockdep_map);
2839         else
2840                 lock_map_acquire_read(&pwq->wq->lockdep_map);
2841         lock_map_release(&pwq->wq->lockdep_map);
2842
2843         return true;
2844 already_gone:
2845         spin_unlock_irq(&pool->lock);
2846         return false;
2847 }
2848
2849 /**
2850  * flush_work - wait for a work to finish executing the last queueing instance
2851  * @work: the work to flush
2852  *
2853  * Wait until @work has finished execution.  @work is guaranteed to be idle
2854  * on return if it hasn't been requeued since flush started.
2855  *
2856  * RETURNS:
2857  * %true if flush_work() waited for the work to finish execution,
2858  * %false if it was already idle.
2859  */
2860 bool flush_work(struct work_struct *work)
2861 {
2862         struct wq_barrier barr;
2863
2864         lock_map_acquire(&work->lockdep_map);
2865         lock_map_release(&work->lockdep_map);
2866
2867         if (start_flush_work(work, &barr)) {
2868                 wait_for_completion(&barr.done);
2869                 destroy_work_on_stack(&barr.work);
2870                 return true;
2871         } else {
2872                 return false;
2873         }
2874 }
2875 EXPORT_SYMBOL_GPL(flush_work);
2876
2877 static bool __cancel_work_timer(struct work_struct *work, bool is_dwork)
2878 {
2879         unsigned long flags;
2880         int ret;
2881
2882         do {
2883                 ret = try_to_grab_pending(work, is_dwork, &flags);
2884                 /*
2885                  * If someone else is canceling, wait for the same event it
2886                  * would be waiting for before retrying.
2887                  */
2888                 if (unlikely(ret == -ENOENT))
2889                         flush_work(work);
2890         } while (unlikely(ret < 0));
2891
2892         /* tell other tasks trying to grab @work to back off */
2893         mark_work_canceling(work);
2894         local_irq_restore(flags);
2895
2896         flush_work(work);
2897         clear_work_data(work);
2898         return ret;
2899 }
2900
2901 /**
2902  * cancel_work_sync - cancel a work and wait for it to finish
2903  * @work: the work to cancel
2904  *
2905  * Cancel @work and wait for its execution to finish.  This function
2906  * can be used even if the work re-queues itself or migrates to
2907  * another workqueue.  On return from this function, @work is
2908  * guaranteed to be not pending or executing on any CPU.
2909  *
2910  * cancel_work_sync(&delayed_work->work) must not be used for
2911  * delayed_work's.  Use cancel_delayed_work_sync() instead.
2912  *
2913  * The caller must ensure that the workqueue on which @work was last
2914  * queued can't be destroyed before this function returns.
2915  *
2916  * RETURNS:
2917  * %true if @work was pending, %false otherwise.
2918  */
2919 bool cancel_work_sync(struct work_struct *work)
2920 {
2921         return __cancel_work_timer(work, false);
2922 }
2923 EXPORT_SYMBOL_GPL(cancel_work_sync);
2924
2925 /**
2926  * flush_delayed_work - wait for a dwork to finish executing the last queueing
2927  * @dwork: the delayed work to flush
2928  *
2929  * Delayed timer is cancelled and the pending work is queued for
2930  * immediate execution.  Like flush_work(), this function only
2931  * considers the last queueing instance of @dwork.
2932  *
2933  * RETURNS:
2934  * %true if flush_work() waited for the work to finish execution,
2935  * %false if it was already idle.
2936  */
2937 bool flush_delayed_work(struct delayed_work *dwork)
2938 {
2939         local_irq_disable();
2940         if (del_timer_sync(&dwork->timer))
2941                 __queue_work(dwork->cpu, dwork->wq, &dwork->work);
2942         local_irq_enable();
2943         return flush_work(&dwork->work);
2944 }
2945 EXPORT_SYMBOL(flush_delayed_work);
2946
2947 /**
2948  * cancel_delayed_work - cancel a delayed work
2949  * @dwork: delayed_work to cancel
2950  *
2951  * Kill off a pending delayed_work.  Returns %true if @dwork was pending
2952  * and canceled; %false if wasn't pending.  Note that the work callback
2953  * function may still be running on return, unless it returns %true and the
2954  * work doesn't re-arm itself.  Explicitly flush or use
2955  * cancel_delayed_work_sync() to wait on it.
2956  *
2957  * This function is safe to call from any context including IRQ handler.
2958  */
2959 bool cancel_delayed_work(struct delayed_work *dwork)
2960 {
2961         unsigned long flags;
2962         int ret;
2963
2964         do {
2965                 ret = try_to_grab_pending(&dwork->work, true, &flags);
2966         } while (unlikely(ret == -EAGAIN));
2967
2968         if (unlikely(ret < 0))
2969                 return false;
2970
2971         set_work_pool_and_clear_pending(&dwork->work,
2972                                         get_work_pool_id(&dwork->work));
2973         local_irq_restore(flags);
2974         return ret;
2975 }
2976 EXPORT_SYMBOL(cancel_delayed_work);
2977
2978 /**
2979  * cancel_delayed_work_sync - cancel a delayed work and wait for it to finish
2980  * @dwork: the delayed work cancel
2981  *
2982  * This is cancel_work_sync() for delayed works.
2983  *
2984  * RETURNS:
2985  * %true if @dwork was pending, %false otherwise.
2986  */
2987 bool cancel_delayed_work_sync(struct delayed_work *dwork)
2988 {
2989         return __cancel_work_timer(&dwork->work, true);
2990 }
2991 EXPORT_SYMBOL(cancel_delayed_work_sync);
2992
2993 /**
2994  * schedule_on_each_cpu - execute a function synchronously on each online CPU
2995  * @func: the function to call
2996  *
2997  * schedule_on_each_cpu() executes @func on each online CPU using the
2998  * system workqueue and blocks until all CPUs have completed.
2999  * schedule_on_each_cpu() is very slow.
3000  *
3001  * RETURNS:
3002  * 0 on success, -errno on failure.
3003  */
3004 int schedule_on_each_cpu(work_func_t func)
3005 {
3006         int cpu;
3007         struct work_struct __percpu *works;
3008
3009         works = alloc_percpu(struct work_struct);
3010         if (!works)
3011                 return -ENOMEM;
3012
3013         get_online_cpus();
3014
3015         for_each_online_cpu(cpu) {
3016                 struct work_struct *work = per_cpu_ptr(works, cpu);
3017
3018                 INIT_WORK(work, func);
3019                 schedule_work_on(cpu, work);
3020         }
3021
3022         for_each_online_cpu(cpu)
3023                 flush_work(per_cpu_ptr(works, cpu));
3024
3025         put_online_cpus();
3026         free_percpu(works);
3027         return 0;
3028 }
3029
3030 /**
3031  * flush_scheduled_work - ensure that any scheduled work has run to completion.
3032  *
3033  * Forces execution of the kernel-global workqueue and blocks until its
3034  * completion.
3035  *
3036  * Think twice before calling this function!  It's very easy to get into
3037  * trouble if you don't take great care.  Either of the following situations
3038  * will lead to deadlock:
3039  *
3040  *      One of the work items currently on the workqueue needs to acquire
3041  *      a lock held by your code or its caller.
3042  *
3043  *      Your code is running in the context of a work routine.
3044  *
3045  * They will be detected by lockdep when they occur, but the first might not
3046  * occur very often.  It depends on what work items are on the workqueue and
3047  * what locks they need, which you have no control over.
3048  *
3049  * In most situations flushing the entire workqueue is overkill; you merely
3050  * need to know that a particular work item isn't queued and isn't running.
3051  * In such cases you should use cancel_delayed_work_sync() or
3052  * cancel_work_sync() instead.
3053  */
3054 void flush_scheduled_work(void)
3055 {
3056         flush_workqueue(system_wq);
3057 }
3058 EXPORT_SYMBOL(flush_scheduled_work);
3059
3060 /**
3061  * execute_in_process_context - reliably execute the routine with user context
3062  * @fn:         the function to execute
3063  * @ew:         guaranteed storage for the execute work structure (must
3064  *              be available when the work executes)
3065  *
3066  * Executes the function immediately if process context is available,
3067  * otherwise schedules the function for delayed execution.
3068  *
3069  * Returns:     0 - function was executed
3070  *              1 - function was scheduled for execution
3071  */
3072 int execute_in_process_context(work_func_t fn, struct execute_work *ew)
3073 {
3074         if (!in_interrupt()) {
3075                 fn(&ew->work);
3076                 return 0;
3077         }
3078
3079         INIT_WORK(&ew->work, fn);
3080         schedule_work(&ew->work);
3081
3082         return 1;
3083 }
3084 EXPORT_SYMBOL_GPL(execute_in_process_context);
3085
3086 #ifdef CONFIG_SYSFS
3087 /*
3088  * Workqueues with WQ_SYSFS flag set is visible to userland via
3089  * /sys/bus/workqueue/devices/WQ_NAME.  All visible workqueues have the
3090  * following attributes.
3091  *
3092  *  per_cpu     RO bool : whether the workqueue is per-cpu or unbound
3093  *  max_active  RW int  : maximum number of in-flight work items
3094  *
3095  * Unbound workqueues have the following extra attributes.
3096  *
3097  *  id          RO int  : the associated pool ID
3098  *  nice        RW int  : nice value of the workers
3099  *  cpumask     RW mask : bitmask of allowed CPUs for the workers
3100  */
3101 struct wq_device {
3102         struct workqueue_struct         *wq;
3103         struct device                   dev;
3104 };
3105
3106 static struct workqueue_struct *dev_to_wq(struct device *dev)
3107 {
3108         struct wq_device *wq_dev = container_of(dev, struct wq_device, dev);
3109
3110         return wq_dev->wq;
3111 }
3112
3113 static ssize_t wq_per_cpu_show(struct device *dev,
3114                                struct device_attribute *attr, char *buf)
3115 {
3116         struct workqueue_struct *wq = dev_to_wq(dev);
3117
3118         return scnprintf(buf, PAGE_SIZE, "%d\n", (bool)!(wq->flags & WQ_UNBOUND));
3119 }
3120
3121 static ssize_t wq_max_active_show(struct device *dev,
3122                                   struct device_attribute *attr, char *buf)
3123 {
3124         struct workqueue_struct *wq = dev_to_wq(dev);
3125
3126         return scnprintf(buf, PAGE_SIZE, "%d\n", wq->saved_max_active);
3127 }
3128
3129 static ssize_t wq_max_active_store(struct device *dev,
3130                                    struct device_attribute *attr,
3131                                    const char *buf, size_t count)
3132 {
3133         struct workqueue_struct *wq = dev_to_wq(dev);
3134         int val;
3135
3136         if (sscanf(buf, "%d", &val) != 1 || val <= 0)
3137                 return -EINVAL;
3138
3139         workqueue_set_max_active(wq, val);
3140         return count;
3141 }
3142
3143 static struct device_attribute wq_sysfs_attrs[] = {
3144         __ATTR(per_cpu, 0444, wq_per_cpu_show, NULL),
3145         __ATTR(max_active, 0644, wq_max_active_show, wq_max_active_store),
3146         __ATTR_NULL,
3147 };
3148
3149 static ssize_t wq_pool_ids_show(struct device *dev,
3150                                 struct device_attribute *attr, char *buf)
3151 {
3152         struct workqueue_struct *wq = dev_to_wq(dev);
3153         const char *delim = "";
3154         int node, written = 0;
3155
3156         rcu_read_lock_sched();
3157         for_each_node(node) {
3158                 written += scnprintf(buf + written, PAGE_SIZE - written,
3159                                      "%s%d:%d", delim, node,
3160                                      unbound_pwq_by_node(wq, node)->pool->id);
3161                 delim = " ";
3162         }
3163         written += scnprintf(buf + written, PAGE_SIZE - written, "\n");
3164         rcu_read_unlock_sched();
3165
3166         return written;
3167 }
3168
3169 static ssize_t wq_nice_show(struct device *dev, struct device_attribute *attr,
3170                             char *buf)
3171 {
3172         struct workqueue_struct *wq = dev_to_wq(dev);
3173         int written;
3174
3175         mutex_lock(&wq->mutex);
3176         written = scnprintf(buf, PAGE_SIZE, "%d\n", wq->unbound_attrs->nice);
3177         mutex_unlock(&wq->mutex);
3178
3179         return written;
3180 }
3181
3182 /* prepare workqueue_attrs for sysfs store operations */
3183 static struct workqueue_attrs *wq_sysfs_prep_attrs(struct workqueue_struct *wq)
3184 {
3185         struct workqueue_attrs *attrs;
3186
3187         attrs = alloc_workqueue_attrs(GFP_KERNEL);
3188         if (!attrs)
3189                 return NULL;
3190
3191         mutex_lock(&wq->mutex);
3192         copy_workqueue_attrs(attrs, wq->unbound_attrs);
3193         mutex_unlock(&wq->mutex);
3194         return attrs;
3195 }
3196
3197 static ssize_t wq_nice_store(struct device *dev, struct device_attribute *attr,
3198                              const char *buf, size_t count)
3199 {
3200         struct workqueue_struct *wq = dev_to_wq(dev);
3201         struct workqueue_attrs *attrs;
3202         int ret;
3203
3204         attrs = wq_sysfs_prep_attrs(wq);
3205         if (!attrs)
3206                 return -ENOMEM;
3207
3208         if (sscanf(buf, "%d", &attrs->nice) == 1 &&
3209             attrs->nice >= -20 && attrs->nice <= 19)
3210                 ret = apply_workqueue_attrs(wq, attrs);
3211         else
3212                 ret = -EINVAL;
3213
3214         free_workqueue_attrs(attrs);
3215         return ret ?: count;
3216 }
3217
3218 static ssize_t wq_cpumask_show(struct device *dev,
3219                                struct device_attribute *attr, char *buf)
3220 {
3221         struct workqueue_struct *wq = dev_to_wq(dev);
3222         int written;
3223
3224         mutex_lock(&wq->mutex);
3225         written = cpumask_scnprintf(buf, PAGE_SIZE, wq->unbound_attrs->cpumask);
3226         mutex_unlock(&wq->mutex);
3227
3228         written += scnprintf(buf + written, PAGE_SIZE - written, "\n");
3229         return written;
3230 }
3231
3232 static ssize_t wq_cpumask_store(struct device *dev,
3233                                 struct device_attribute *attr,
3234                                 const char *buf, size_t count)
3235 {
3236         struct workqueue_struct *wq = dev_to_wq(dev);
3237         struct workqueue_attrs *attrs;
3238         int ret;
3239
3240         attrs = wq_sysfs_prep_attrs(wq);
3241         if (!attrs)
3242                 return -ENOMEM;
3243
3244         ret = cpumask_parse(buf, attrs->cpumask);
3245         if (!ret)
3246                 ret = apply_workqueue_attrs(wq, attrs);
3247
3248         free_workqueue_attrs(attrs);
3249         return ret ?: count;
3250 }
3251
3252 static ssize_t wq_numa_show(struct device *dev, struct device_attribute *attr,
3253                             char *buf)
3254 {
3255         struct workqueue_struct *wq = dev_to_wq(dev);
3256         int written;
3257
3258         mutex_lock(&wq->mutex);
3259         written = scnprintf(buf, PAGE_SIZE, "%d\n",
3260                             !wq->unbound_attrs->no_numa);
3261         mutex_unlock(&wq->mutex);
3262
3263         return written;
3264 }
3265
3266 static ssize_t wq_numa_store(struct device *dev, struct device_attribute *attr,
3267                              const char *buf, size_t count)
3268 {
3269         struct workqueue_struct *wq = dev_to_wq(dev);
3270         struct workqueue_attrs *attrs;
3271         int v, ret;
3272
3273         attrs = wq_sysfs_prep_attrs(wq);
3274         if (!attrs)
3275                 return -ENOMEM;
3276
3277         ret = -EINVAL;
3278         if (sscanf(buf, "%d", &v) == 1) {
3279                 attrs->no_numa = !v;
3280                 ret = apply_workqueue_attrs(wq, attrs);
3281         }
3282
3283         free_workqueue_attrs(attrs);
3284         return ret ?: count;
3285 }
3286
3287 static struct device_attribute wq_sysfs_unbound_attrs[] = {
3288         __ATTR(pool_ids, 0444, wq_pool_ids_show, NULL),
3289         __ATTR(nice, 0644, wq_nice_show, wq_nice_store),
3290         __ATTR(cpumask, 0644, wq_cpumask_show, wq_cpumask_store),
3291         __ATTR(numa, 0644, wq_numa_show, wq_numa_store),
3292         __ATTR_NULL,
3293 };
3294
3295 static struct bus_type wq_subsys = {
3296         .name                           = "workqueue",
3297         .dev_attrs                      = wq_sysfs_attrs,
3298 };
3299
3300 static int __init wq_sysfs_init(void)
3301 {
3302         return subsys_virtual_register(&wq_subsys, NULL);
3303 }
3304 core_initcall(wq_sysfs_init);
3305
3306 static void wq_device_release(struct device *dev)
3307 {
3308         struct wq_device *wq_dev = container_of(dev, struct wq_device, dev);
3309
3310         kfree(wq_dev);
3311 }
3312
3313 /**
3314  * workqueue_sysfs_register - make a workqueue visible in sysfs
3315  * @wq: the workqueue to register
3316  *
3317  * Expose @wq in sysfs under /sys/bus/workqueue/devices.
3318  * alloc_workqueue*() automatically calls this function if WQ_SYSFS is set
3319  * which is the preferred method.
3320  *
3321  * Workqueue user should use this function directly iff it wants to apply
3322  * workqueue_attrs before making the workqueue visible in sysfs; otherwise,
3323  * apply_workqueue_attrs() may race against userland updating the
3324  * attributes.
3325  *
3326  * Returns 0 on success, -errno on failure.
3327  */
3328 int workqueue_sysfs_register(struct workqueue_struct *wq)
3329 {
3330         struct wq_device *wq_dev;
3331         int ret;
3332
3333         /*
3334          * Adjusting max_active or creating new pwqs by applyting
3335          * attributes breaks ordering guarantee.  Disallow exposing ordered
3336          * workqueues.
3337          */
3338         if (WARN_ON(wq->flags & __WQ_ORDERED))
3339                 return -EINVAL;
3340
3341         wq->wq_dev = wq_dev = kzalloc(sizeof(*wq_dev), GFP_KERNEL);
3342         if (!wq_dev)
3343                 return -ENOMEM;
3344
3345         wq_dev->wq = wq;
3346         wq_dev->dev.bus = &wq_subsys;
3347         wq_dev->dev.init_name = wq->name;
3348         wq_dev->dev.release = wq_device_release;
3349
3350         /*
3351          * unbound_attrs are created separately.  Suppress uevent until
3352          * everything is ready.
3353          */
3354         dev_set_uevent_suppress(&wq_dev->dev, true);
3355
3356         ret = device_register(&wq_dev->dev);
3357         if (ret) {
3358                 kfree(wq_dev);
3359                 wq->wq_dev = NULL;
3360                 return ret;
3361         }
3362
3363         if (wq->flags & WQ_UNBOUND) {
3364                 struct device_attribute *attr;
3365
3366                 for (attr = wq_sysfs_unbound_attrs; attr->attr.name; attr++) {
3367                         ret = device_create_file(&wq_dev->dev, attr);
3368                         if (ret) {
3369                                 device_unregister(&wq_dev->dev);
3370                                 wq->wq_dev = NULL;
3371                                 return ret;
3372                         }
3373                 }
3374         }
3375
3376         dev_set_uevent_suppress(&wq_dev->dev, false);
3377         kobject_uevent(&wq_dev->dev.kobj, KOBJ_ADD);
3378         return 0;
3379 }
3380
3381 /**
3382  * workqueue_sysfs_unregister - undo workqueue_sysfs_register()
3383  * @wq: the workqueue to unregister
3384  *
3385  * If @wq is registered to sysfs by workqueue_sysfs_register(), unregister.
3386  */
3387 static void workqueue_sysfs_unregister(struct workqueue_struct *wq)
3388 {
3389         struct wq_device *wq_dev = wq->wq_dev;
3390
3391         if (!wq->wq_dev)
3392                 return;
3393
3394         wq->wq_dev = NULL;
3395         device_unregister(&wq_dev->dev);
3396 }
3397 #else   /* CONFIG_SYSFS */
3398 static void workqueue_sysfs_unregister(struct workqueue_struct *wq)     { }
3399 #endif  /* CONFIG_SYSFS */
3400
3401 /**
3402  * free_workqueue_attrs - free a workqueue_attrs
3403  * @attrs: workqueue_attrs to free
3404  *
3405  * Undo alloc_workqueue_attrs().
3406  */
3407 void free_workqueue_attrs(struct workqueue_attrs *attrs)
3408 {
3409         if (attrs) {
3410                 free_cpumask_var(attrs->cpumask);
3411                 kfree(attrs);
3412         }
3413 }
3414
3415 /**
3416  * alloc_workqueue_attrs - allocate a workqueue_attrs
3417  * @gfp_mask: allocation mask to use
3418  *
3419  * Allocate a new workqueue_attrs, initialize with default settings and
3420  * return it.  Returns NULL on failure.
3421  */
3422 struct workqueue_attrs *alloc_workqueue_attrs(gfp_t gfp_mask)
3423 {
3424         struct workqueue_attrs *attrs;
3425
3426         attrs = kzalloc(sizeof(*attrs), gfp_mask);
3427         if (!attrs)
3428                 goto fail;
3429         if (!alloc_cpumask_var(&attrs->cpumask, gfp_mask))
3430                 goto fail;
3431
3432         cpumask_copy(attrs->cpumask, cpu_possible_mask);
3433         return attrs;
3434 fail:
3435         free_workqueue_attrs(attrs);
3436         return NULL;
3437 }
3438
3439 static void copy_workqueue_attrs(struct workqueue_attrs *to,
3440                                  const struct workqueue_attrs *from)
3441 {
3442         to->nice = from->nice;
3443         cpumask_copy(to->cpumask, from->cpumask);
3444         /*
3445          * Unlike hash and equality test, this function doesn't ignore
3446          * ->no_numa as it is used for both pool and wq attrs.  Instead,
3447          * get_unbound_pool() explicitly clears ->no_numa after copying.
3448          */
3449         to->no_numa = from->no_numa;
3450 }
3451
3452 /* hash value of the content of @attr */
3453 static u32 wqattrs_hash(const struct workqueue_attrs *attrs)
3454 {
3455         u32 hash = 0;
3456
3457         hash = jhash_1word(attrs->nice, hash);
3458         hash = jhash(cpumask_bits(attrs->cpumask),
3459                      BITS_TO_LONGS(nr_cpumask_bits) * sizeof(long), hash);
3460         return hash;
3461 }
3462
3463 /* content equality test */
3464 static bool wqattrs_equal(const struct workqueue_attrs *a,
3465                           const struct workqueue_attrs *b)
3466 {
3467         if (a->nice != b->nice)
3468                 return false;
3469         if (!cpumask_equal(a->cpumask, b->cpumask))
3470                 return false;
3471         return true;
3472 }
3473
3474 /**
3475  * init_worker_pool - initialize a newly zalloc'd worker_pool
3476  * @pool: worker_pool to initialize
3477  *
3478  * Initiailize a newly zalloc'd @pool.  It also allocates @pool->attrs.
3479  * Returns 0 on success, -errno on failure.  Even on failure, all fields
3480  * inside @pool proper are initialized and put_unbound_pool() can be called
3481  * on @pool safely to release it.
3482  */
3483 static int init_worker_pool(struct worker_pool *pool)
3484 {
3485         spin_lock_init(&pool->lock);
3486         pool->id = -1;
3487         pool->cpu = -1;
3488         pool->node = NUMA_NO_NODE;
3489         pool->flags |= POOL_DISASSOCIATED;
3490         INIT_LIST_HEAD(&pool->worklist);
3491         INIT_LIST_HEAD(&pool->idle_list);
3492         hash_init(pool->busy_hash);
3493
3494         init_timer_deferrable(&pool->idle_timer);
3495         pool->idle_timer.function = idle_worker_timeout;
3496         pool->idle_timer.data = (unsigned long)pool;
3497
3498         setup_timer(&pool->mayday_timer, pool_mayday_timeout,
3499                     (unsigned long)pool);
3500
3501         mutex_init(&pool->manager_arb);
3502         mutex_init(&pool->manager_mutex);
3503         idr_init(&pool->worker_idr);
3504
3505         INIT_HLIST_NODE(&pool->hash_node);
3506         pool->refcnt = 1;
3507
3508         /* shouldn't fail above this point */
3509         pool->attrs = alloc_workqueue_attrs(GFP_KERNEL);
3510         if (!pool->attrs)
3511                 return -ENOMEM;
3512         return 0;
3513 }
3514
3515 static void rcu_free_pool(struct rcu_head *rcu)
3516 {
3517         struct worker_pool *pool = container_of(rcu, struct worker_pool, rcu);
3518
3519         idr_destroy(&pool->worker_idr);
3520         free_workqueue_attrs(pool->attrs);
3521         kfree(pool);
3522 }
3523
3524 /**
3525  * put_unbound_pool - put a worker_pool
3526  * @pool: worker_pool to put
3527  *
3528  * Put @pool.  If its refcnt reaches zero, it gets destroyed in sched-RCU
3529  * safe manner.  get_unbound_pool() calls this function on its failure path
3530  * and this function should be able to release pools which went through,
3531  * successfully or not, init_worker_pool().
3532  *
3533  * Should be called with wq_pool_mutex held.
3534  */
3535 static void put_unbound_pool(struct worker_pool *pool)
3536 {
3537         struct worker *worker;
3538
3539         lockdep_assert_held(&wq_pool_mutex);
3540
3541         if (--pool->refcnt)
3542                 return;
3543
3544         /* sanity checks */
3545         if (WARN_ON(!(pool->flags & POOL_DISASSOCIATED)) ||
3546             WARN_ON(!list_empty(&pool->worklist)))
3547                 return;
3548
3549         /* release id and unhash */
3550         if (pool->id >= 0)
3551                 idr_remove(&worker_pool_idr, pool->id);
3552         hash_del(&pool->hash_node);
3553
3554         /*
3555          * Become the manager and destroy all workers.  Grabbing
3556          * manager_arb prevents @pool's workers from blocking on
3557          * manager_mutex.
3558          */
3559         mutex_lock(&pool->manager_arb);
3560         mutex_lock(&pool->manager_mutex);
3561         spin_lock_irq(&pool->lock);
3562
3563         while ((worker = first_worker(pool)))
3564                 destroy_worker(worker);
3565         WARN_ON(pool->nr_workers || pool->nr_idle);
3566
3567         spin_unlock_irq(&pool->lock);
3568         mutex_unlock(&pool->manager_mutex);
3569         mutex_unlock(&pool->manager_arb);
3570
3571         /* shut down the timers */
3572         del_timer_sync(&pool->idle_timer);
3573         del_timer_sync(&pool->mayday_timer);
3574
3575         /* sched-RCU protected to allow dereferences from get_work_pool() */
3576         call_rcu_sched(&pool->rcu, rcu_free_pool);
3577 }
3578
3579 /**
3580  * get_unbound_pool - get a worker_pool with the specified attributes
3581  * @attrs: the attributes of the worker_pool to get
3582  *
3583  * Obtain a worker_pool which has the same attributes as @attrs, bump the
3584  * reference count and return it.  If there already is a matching
3585  * worker_pool, it will be used; otherwise, this function attempts to
3586  * create a new one.  On failure, returns NULL.
3587  *
3588  * Should be called with wq_pool_mutex held.
3589  */
3590 static struct worker_pool *get_unbound_pool(const struct workqueue_attrs *attrs)
3591 {
3592         u32 hash = wqattrs_hash(attrs);
3593         struct worker_pool *pool;
3594         int node;
3595
3596         lockdep_assert_held(&wq_pool_mutex);
3597
3598         /* do we already have a matching pool? */
3599         hash_for_each_possible(unbound_pool_hash, pool, hash_node, hash) {
3600                 if (wqattrs_equal(pool->attrs, attrs)) {
3601                         pool->refcnt++;
3602                         goto out_unlock;
3603                 }
3604         }
3605
3606         /* nope, create a new one */
3607         pool = kzalloc(sizeof(*pool), GFP_KERNEL);
3608         if (!pool || init_worker_pool(pool) < 0)
3609                 goto fail;
3610
3611         if (workqueue_freezing)
3612                 pool->flags |= POOL_FREEZING;
3613
3614         lockdep_set_subclass(&pool->lock, 1);   /* see put_pwq() */
3615         copy_workqueue_attrs(pool->attrs, attrs);
3616
3617         /*
3618          * no_numa isn't a worker_pool attribute, always clear it.  See
3619          * 'struct workqueue_attrs' comments for detail.
3620          */
3621         pool->attrs->no_numa = false;
3622
3623         /* if cpumask is contained inside a NUMA node, we belong to that node */
3624         if (wq_numa_enabled) {
3625                 for_each_node(node) {
3626                         if (cpumask_subset(pool->attrs->cpumask,
3627                                            wq_numa_possible_cpumask[node])) {
3628                                 pool->node = node;
3629                                 break;
3630                         }
3631                 }
3632         }
3633
3634         if (worker_pool_assign_id(pool) < 0)
3635                 goto fail;
3636
3637         /* create and start the initial worker */
3638         if (create_and_start_worker(pool) < 0)
3639                 goto fail;
3640
3641         /* install */
3642         hash_add(unbound_pool_hash, &pool->hash_node, hash);
3643 out_unlock:
3644         return pool;
3645 fail:
3646         if (pool)
3647                 put_unbound_pool(pool);
3648         return NULL;
3649 }
3650
3651 static void rcu_free_pwq(struct rcu_head *rcu)
3652 {
3653         kmem_cache_free(pwq_cache,
3654                         container_of(rcu, struct pool_workqueue, rcu));
3655 }
3656
3657 /*
3658  * Scheduled on system_wq by put_pwq() when an unbound pwq hits zero refcnt
3659  * and needs to be destroyed.
3660  */
3661 static void pwq_unbound_release_workfn(struct work_struct *work)
3662 {
3663         struct pool_workqueue *pwq = container_of(work, struct pool_workqueue,
3664                                                   unbound_release_work);
3665         struct workqueue_struct *wq = pwq->wq;
3666         struct worker_pool *pool = pwq->pool;
3667         bool is_last;
3668
3669         if (WARN_ON_ONCE(!(wq->flags & WQ_UNBOUND)))
3670                 return;
3671
3672         /*
3673          * Unlink @pwq.  Synchronization against wq->mutex isn't strictly
3674          * necessary on release but do it anyway.  It's easier to verify
3675          * and consistent with the linking path.
3676          */
3677         mutex_lock(&wq->mutex);
3678         list_del_rcu(&pwq->pwqs_node);
3679         is_last = list_empty(&wq->pwqs);
3680         mutex_unlock(&wq->mutex);
3681
3682         mutex_lock(&wq_pool_mutex);
3683         put_unbound_pool(pool);
3684         mutex_unlock(&wq_pool_mutex);
3685
3686         call_rcu_sched(&pwq->rcu, rcu_free_pwq);
3687
3688         /*
3689          * If we're the last pwq going away, @wq is already dead and no one
3690          * is gonna access it anymore.  Free it.
3691          */
3692         if (is_last) {
3693                 free_workqueue_attrs(wq->unbound_attrs);
3694                 kfree(wq);
3695         }
3696 }
3697
3698 /**
3699  * pwq_adjust_max_active - update a pwq's max_active to the current setting
3700  * @pwq: target pool_workqueue
3701  *
3702  * If @pwq isn't freezing, set @pwq->max_active to the associated
3703  * workqueue's saved_max_active and activate delayed work items
3704  * accordingly.  If @pwq is freezing, clear @pwq->max_active to zero.
3705  */
3706 static void pwq_adjust_max_active(struct pool_workqueue *pwq)
3707 {
3708         struct workqueue_struct *wq = pwq->wq;
3709         bool freezable = wq->flags & WQ_FREEZABLE;
3710
3711         /* for @wq->saved_max_active */
3712         lockdep_assert_held(&wq->mutex);
3713
3714         /* fast exit for non-freezable wqs */
3715         if (!freezable && pwq->max_active == wq->saved_max_active)
3716                 return;
3717
3718         spin_lock_irq(&pwq->pool->lock);
3719
3720         if (!freezable || !(pwq->pool->flags & POOL_FREEZING)) {
3721                 pwq->max_active = wq->saved_max_active;
3722
3723                 while (!list_empty(&pwq->delayed_works) &&
3724                        pwq->nr_active < pwq->max_active)
3725                         pwq_activate_first_delayed(pwq);
3726
3727                 /*
3728                  * Need to kick a worker after thawed or an unbound wq's
3729                  * max_active is bumped.  It's a slow path.  Do it always.
3730                  */
3731                 wake_up_worker(pwq->pool);
3732         } else {
3733                 pwq->max_active = 0;
3734         }
3735
3736         spin_unlock_irq(&pwq->pool->lock);
3737 }
3738
3739 /* initialize newly alloced @pwq which is associated with @wq and @pool */
3740 static void init_pwq(struct pool_workqueue *pwq, struct workqueue_struct *wq,
3741                      struct worker_pool *pool)
3742 {
3743         BUG_ON((unsigned long)pwq & WORK_STRUCT_FLAG_MASK);
3744
3745         memset(pwq, 0, sizeof(*pwq));
3746
3747         pwq->pool = pool;
3748         pwq->wq = wq;
3749         pwq->flush_color = -1;
3750         pwq->refcnt = 1;
3751         INIT_LIST_HEAD(&pwq->delayed_works);
3752         INIT_LIST_HEAD(&pwq->pwqs_node);
3753         INIT_LIST_HEAD(&pwq->mayday_node);
3754         INIT_WORK(&pwq->unbound_release_work, pwq_unbound_release_workfn);
3755 }
3756
3757 /* sync @pwq with the current state of its associated wq and link it */
3758 static void link_pwq(struct pool_workqueue *pwq)
3759 {
3760         struct workqueue_struct *wq = pwq->wq;
3761
3762         lockdep_assert_held(&wq->mutex);
3763
3764         /* may be called multiple times, ignore if already linked */
3765         if (!list_empty(&pwq->pwqs_node))
3766                 return;
3767
3768         /*
3769          * Set the matching work_color.  This is synchronized with
3770          * wq->mutex to avoid confusing flush_workqueue().
3771          */
3772         pwq->work_color = wq->work_color;
3773
3774         /* sync max_active to the current setting */
3775         pwq_adjust_max_active(pwq);
3776
3777         /* link in @pwq */
3778         list_add_rcu(&pwq->pwqs_node, &wq->pwqs);
3779 }
3780
3781 /* obtain a pool matching @attr and create a pwq associating the pool and @wq */
3782 static struct pool_workqueue *alloc_unbound_pwq(struct workqueue_struct *wq,
3783                                         const struct workqueue_attrs *attrs)
3784 {
3785         struct worker_pool *pool;
3786         struct pool_workqueue *pwq;
3787
3788         lockdep_assert_held(&wq_pool_mutex);
3789
3790         pool = get_unbound_pool(attrs);
3791         if (!pool)
3792                 return NULL;
3793
3794         pwq = kmem_cache_alloc_node(pwq_cache, GFP_KERNEL, pool->node);
3795         if (!pwq) {
3796                 put_unbound_pool(pool);
3797                 return NULL;
3798         }
3799
3800         init_pwq(pwq, wq, pool);
3801         return pwq;
3802 }
3803
3804 /* undo alloc_unbound_pwq(), used only in the error path */
3805 static void free_unbound_pwq(struct pool_workqueue *pwq)
3806 {
3807         lockdep_assert_held(&wq_pool_mutex);
3808
3809         if (pwq) {
3810                 put_unbound_pool(pwq->pool);
3811                 kmem_cache_free(pwq_cache, pwq);
3812         }
3813 }
3814
3815 /**
3816  * wq_calc_node_mask - calculate a wq_attrs' cpumask for the specified node
3817  * @attrs: the wq_attrs of interest
3818  * @node: the target NUMA node
3819  * @cpu_going_down: if >= 0, the CPU to consider as offline
3820  * @cpumask: outarg, the resulting cpumask
3821  *
3822  * Calculate the cpumask a workqueue with @attrs should use on @node.  If
3823  * @cpu_going_down is >= 0, that cpu is considered offline during
3824  * calculation.  The result is stored in @cpumask.  This function returns
3825  * %true if the resulting @cpumask is different from @attrs->cpumask,
3826  * %false if equal.
3827  *
3828  * If NUMA affinity is not enabled, @attrs->cpumask is always used.  If
3829  * enabled and @node has online CPUs requested by @attrs, the returned
3830  * cpumask is the intersection of the possible CPUs of @node and
3831  * @attrs->cpumask.
3832  *
3833  * The caller is responsible for ensuring that the cpumask of @node stays
3834  * stable.
3835  */
3836 static bool wq_calc_node_cpumask(const struct workqueue_attrs *attrs, int node,
3837                                  int cpu_going_down, cpumask_t *cpumask)
3838 {
3839         if (!wq_numa_enabled || attrs->no_numa)
3840                 goto use_dfl;
3841
3842         /* does @node have any online CPUs @attrs wants? */
3843         cpumask_and(cpumask, cpumask_of_node(node), attrs->cpumask);
3844         if (cpu_going_down >= 0)
3845                 cpumask_clear_cpu(cpu_going_down, cpumask);
3846
3847         if (cpumask_empty(cpumask))
3848                 goto use_dfl;
3849
3850         /* yeap, return possible CPUs in @node that @attrs wants */
3851         cpumask_and(cpumask, attrs->cpumask, wq_numa_possible_cpumask[node]);
3852         return !cpumask_equal(cpumask, attrs->cpumask);
3853
3854 use_dfl:
3855         cpumask_copy(cpumask, attrs->cpumask);
3856         return false;
3857 }
3858
3859 /* install @pwq into @wq's numa_pwq_tbl[] for @node and return the old pwq */
3860 static struct pool_workqueue *numa_pwq_tbl_install(struct workqueue_struct *wq,
3861                                                    int node,
3862                                                    struct pool_workqueue *pwq)
3863 {
3864         struct pool_workqueue *old_pwq;
3865
3866         lockdep_assert_held(&wq->mutex);
3867
3868         /* link_pwq() can handle duplicate calls */
3869         link_pwq(pwq);
3870
3871         old_pwq = rcu_access_pointer(wq->numa_pwq_tbl[node]);
3872         rcu_assign_pointer(wq->numa_pwq_tbl[node], pwq);
3873         return old_pwq;
3874 }
3875
3876 /**
3877  * apply_workqueue_attrs - apply new workqueue_attrs to an unbound workqueue
3878  * @wq: the target workqueue
3879  * @attrs: the workqueue_attrs to apply, allocated with alloc_workqueue_attrs()
3880  *
3881  * Apply @attrs to an unbound workqueue @wq.  Unless disabled, on NUMA
3882  * machines, this function maps a separate pwq to each NUMA node with
3883  * possibles CPUs in @attrs->cpumask so that work items are affine to the
3884  * NUMA node it was issued on.  Older pwqs are released as in-flight work
3885  * items finish.  Note that a work item which repeatedly requeues itself
3886  * back-to-back will stay on its current pwq.
3887  *
3888  * Performs GFP_KERNEL allocations.  Returns 0 on success and -errno on
3889  * failure.
3890  */
3891 int apply_workqueue_attrs(struct workqueue_struct *wq,
3892                           const struct workqueue_attrs *attrs)
3893 {
3894         struct workqueue_attrs *new_attrs, *tmp_attrs;
3895         struct pool_workqueue **pwq_tbl, *dfl_pwq;
3896         int node, ret;
3897
3898         /* only unbound workqueues can change attributes */
3899         if (WARN_ON(!(wq->flags & WQ_UNBOUND)))
3900                 return -EINVAL;
3901
3902         /* creating multiple pwqs breaks ordering guarantee */
3903         if (WARN_ON((wq->flags & __WQ_ORDERED) && !list_empty(&wq->pwqs)))
3904                 return -EINVAL;
3905
3906         pwq_tbl = kzalloc(wq_numa_tbl_len * sizeof(pwq_tbl[0]), GFP_KERNEL);
3907         new_attrs = alloc_workqueue_attrs(GFP_KERNEL);
3908         tmp_attrs = alloc_workqueue_attrs(GFP_KERNEL);
3909         if (!pwq_tbl || !new_attrs || !tmp_attrs)
3910                 goto enomem;
3911
3912         /* make a copy of @attrs and sanitize it */
3913         copy_workqueue_attrs(new_attrs, attrs);
3914         cpumask_and(new_attrs->cpumask, new_attrs->cpumask, cpu_possible_mask);
3915
3916         /*
3917          * We may create multiple pwqs with differing cpumasks.  Make a
3918          * copy of @new_attrs which will be modified and used to obtain
3919          * pools.
3920          */
3921         copy_workqueue_attrs(tmp_attrs, new_attrs);
3922
3923         /*
3924          * CPUs should stay stable across pwq creations and installations.
3925          * Pin CPUs, determine the target cpumask for each node and create
3926          * pwqs accordingly.
3927          */
3928         get_online_cpus();
3929
3930         mutex_lock(&wq_pool_mutex);
3931
3932         /*
3933          * If something goes wrong during CPU up/down, we'll fall back to
3934          * the default pwq covering whole @attrs->cpumask.  Always create
3935          * it even if we don't use it immediately.
3936          */
3937         dfl_pwq = alloc_unbound_pwq(wq, new_attrs);
3938         if (!dfl_pwq)
3939                 goto enomem_pwq;
3940
3941         for_each_node(node) {
3942                 if (wq_calc_node_cpumask(attrs, node, -1, tmp_attrs->cpumask)) {
3943                         pwq_tbl[node] = alloc_unbound_pwq(wq, tmp_attrs);
3944                         if (!pwq_tbl[node])
3945                                 goto enomem_pwq;
3946                 } else {
3947                         dfl_pwq->refcnt++;
3948                         pwq_tbl[node] = dfl_pwq;
3949                 }
3950         }
3951
3952         mutex_unlock(&wq_pool_mutex);
3953
3954         /* all pwqs have been created successfully, let's install'em */
3955         mutex_lock(&wq->mutex);
3956
3957         copy_workqueue_attrs(wq->unbound_attrs, new_attrs);
3958
3959         /* save the previous pwq and install the new one */
3960         for_each_node(node)
3961                 pwq_tbl[node] = numa_pwq_tbl_install(wq, node, pwq_tbl[node]);
3962
3963         /* @dfl_pwq might not have been used, ensure it's linked */
3964         link_pwq(dfl_pwq);
3965         swap(wq->dfl_pwq, dfl_pwq);
3966
3967         mutex_unlock(&wq->mutex);
3968
3969         /* put the old pwqs */
3970         for_each_node(node)
3971                 put_pwq_unlocked(pwq_tbl[node]);
3972         put_pwq_unlocked(dfl_pwq);
3973
3974         put_online_cpus();
3975         ret = 0;
3976         /* fall through */
3977 out_free:
3978         free_workqueue_attrs(tmp_attrs);
3979         free_workqueue_attrs(new_attrs);
3980         kfree(pwq_tbl);
3981         return ret;
3982
3983 enomem_pwq:
3984         free_unbound_pwq(dfl_pwq);
3985         for_each_node(node)
3986                 if (pwq_tbl && pwq_tbl[node] != dfl_pwq)
3987                         free_unbound_pwq(pwq_tbl[node]);
3988         mutex_unlock(&wq_pool_mutex);
3989         put_online_cpus();
3990 enomem:
3991         ret = -ENOMEM;
3992         goto out_free;
3993 }
3994
3995 /**
3996  * wq_update_unbound_numa - update NUMA affinity of a wq for CPU hot[un]plug
3997  * @wq: the target workqueue
3998  * @cpu: the CPU coming up or going down
3999  * @online: whether @cpu is coming up or going down
4000  *
4001  * This function is to be called from %CPU_DOWN_PREPARE, %CPU_ONLINE and
4002  * %CPU_DOWN_FAILED.  @cpu is being hot[un]plugged, update NUMA affinity of
4003  * @wq accordingly.
4004  *
4005  * If NUMA affinity can't be adjusted due to memory allocation failure, it
4006  * falls back to @wq->dfl_pwq which may not be optimal but is always
4007  * correct.
4008  *
4009  * Note that when the last allowed CPU of a NUMA node goes offline for a
4010  * workqueue with a cpumask spanning multiple nodes, the workers which were
4011  * already executing the work items for the workqueue will lose their CPU
4012  * affinity and may execute on any CPU.  This is similar to how per-cpu
4013  * workqueues behave on CPU_DOWN.  If a workqueue user wants strict
4014  * affinity, it's the user's responsibility to flush the work item from
4015  * CPU_DOWN_PREPARE.
4016  */
4017 static void wq_update_unbound_numa(struct workqueue_struct *wq, int cpu,
4018                                    bool online)
4019 {
4020         int node = cpu_to_node(cpu);
4021         int cpu_off = online ? -1 : cpu;
4022         struct pool_workqueue *old_pwq = NULL, *pwq;
4023         struct workqueue_attrs *target_attrs;
4024         cpumask_t *cpumask;
4025
4026         lockdep_assert_held(&wq_pool_mutex);
4027
4028         if (!wq_numa_enabled || !(wq->flags & WQ_UNBOUND))
4029                 return;
4030
4031         /*
4032          * We don't wanna alloc/free wq_attrs for each wq for each CPU.
4033          * Let's use a preallocated one.  The following buf is protected by
4034          * CPU hotplug exclusion.
4035          */
4036         target_attrs = wq_update_unbound_numa_attrs_buf;
4037         cpumask = target_attrs->cpumask;
4038
4039         mutex_lock(&wq->mutex);
4040         if (wq->unbound_attrs->no_numa)
4041                 goto out_unlock;
4042
4043         copy_workqueue_attrs(target_attrs, wq->unbound_attrs);
4044         pwq = unbound_pwq_by_node(wq, node);
4045
4046         /*
4047          * Let's determine what needs to be done.  If the target cpumask is
4048          * different from wq's, we need to compare it to @pwq's and create
4049          * a new one if they don't match.  If the target cpumask equals
4050          * wq's, the default pwq should be used.  If @pwq is already the
4051          * default one, nothing to do; otherwise, install the default one.
4052          */
4053         if (wq_calc_node_cpumask(wq->unbound_attrs, node, cpu_off, cpumask)) {
4054                 if (cpumask_equal(cpumask, pwq->pool->attrs->cpumask))
4055                         goto out_unlock;
4056         } else {
4057                 if (pwq == wq->dfl_pwq)
4058                         goto out_unlock;
4059                 else
4060                         goto use_dfl_pwq;
4061         }
4062
4063         mutex_unlock(&wq->mutex);
4064
4065         /* create a new pwq */
4066         pwq = alloc_unbound_pwq(wq, target_attrs);
4067         if (!pwq) {
4068                 pr_warning("workqueue: allocation failed while updating NUMA affinity of \"%s\"\n",
4069                            wq->name);
4070                 mutex_lock(&wq->mutex);
4071                 goto use_dfl_pwq;
4072         }
4073
4074         /*
4075          * Install the new pwq.  As this function is called only from CPU
4076          * hotplug callbacks and applying a new attrs is wrapped with
4077          * get/put_online_cpus(), @wq->unbound_attrs couldn't have changed
4078          * inbetween.
4079          */
4080         mutex_lock(&wq->mutex);
4081         old_pwq = numa_pwq_tbl_install(wq, node, pwq);
4082         goto out_unlock;
4083
4084 use_dfl_pwq:
4085         spin_lock_irq(&wq->dfl_pwq->pool->lock);
4086         get_pwq(wq->dfl_pwq);
4087         spin_unlock_irq(&wq->dfl_pwq->pool->lock);
4088         old_pwq = numa_pwq_tbl_install(wq, node, wq->dfl_pwq);
4089 out_unlock:
4090         mutex_unlock(&wq->mutex);
4091         put_pwq_unlocked(old_pwq);
4092 }
4093
4094 static int alloc_and_link_pwqs(struct workqueue_struct *wq)
4095 {
4096         bool highpri = wq->flags & WQ_HIGHPRI;
4097         int cpu, ret;
4098
4099         if (!(wq->flags & WQ_UNBOUND)) {
4100                 wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);
4101                 if (!wq->cpu_pwqs)
4102                         return -ENOMEM;
4103
4104                 for_each_possible_cpu(cpu) {
4105                         struct pool_workqueue *pwq =
4106                                 per_cpu_ptr(wq->cpu_pwqs, cpu);
4107                         struct worker_pool *cpu_pools =
4108                                 per_cpu(cpu_worker_pools, cpu);
4109
4110                         init_pwq(pwq, wq, &cpu_pools[highpri]);
4111
4112                         mutex_lock(&wq->mutex);
4113                         link_pwq(pwq);
4114                         mutex_unlock(&wq->mutex);
4115                 }
4116                 return 0;
4117         } else if (wq->flags & __WQ_ORDERED) {
4118                 ret = apply_workqueue_attrs(wq, ordered_wq_attrs[highpri]);
4119                 /* there should only be single pwq for ordering guarantee */
4120                 WARN(!ret && (wq->pwqs.next != &wq->dfl_pwq->pwqs_node ||
4121                               wq->pwqs.prev != &wq->dfl_pwq->pwqs_node),
4122                      "ordering guarantee broken for workqueue %s\n", wq->name);
4123                 return ret;
4124         } else {
4125                 return apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
4126         }
4127 }
4128
4129 static int wq_clamp_max_active(int max_active, unsigned int flags,
4130                                const char *name)
4131 {
4132         int lim = flags & WQ_UNBOUND ? WQ_UNBOUND_MAX_ACTIVE : WQ_MAX_ACTIVE;
4133
4134         if (max_active < 1 || max_active > lim)
4135                 pr_warn("workqueue: max_active %d requested for %s is out of range, clamping between %d and %d\n",
4136                         max_active, name, 1, lim);
4137
4138         return clamp_val(max_active, 1, lim);
4139 }
4140
4141 struct workqueue_struct *__alloc_workqueue_key(const char *fmt,
4142                                                unsigned int flags,
4143                                                int max_active,
4144                                                struct lock_class_key *key,
4145                                                const char *lock_name, ...)
4146 {
4147         size_t tbl_size = 0;
4148         va_list args;
4149         struct workqueue_struct *wq;
4150         struct pool_workqueue *pwq;
4151
4152         /* allocate wq and format name */
4153         if (flags & WQ_UNBOUND)
4154                 tbl_size = wq_numa_tbl_len * sizeof(wq->numa_pwq_tbl[0]);
4155
4156         wq = kzalloc(sizeof(*wq) + tbl_size, GFP_KERNEL);
4157         if (!wq)
4158                 return NULL;
4159
4160         if (flags & WQ_UNBOUND) {
4161                 wq->unbound_attrs = alloc_workqueue_attrs(GFP_KERNEL);
4162                 if (!wq->unbound_attrs)
4163                         goto err_free_wq;
4164         }
4165
4166         va_start(args, lock_name);
4167         vsnprintf(wq->name, sizeof(wq->name), fmt, args);
4168         va_end(args);
4169
4170         max_active = max_active ?: WQ_DFL_ACTIVE;
4171         max_active = wq_clamp_max_active(max_active, flags, wq->name);
4172
4173         /* init wq */
4174         wq->flags = flags;
4175         wq->saved_max_active = max_active;
4176         mutex_init(&wq->mutex);
4177         atomic_set(&wq->nr_pwqs_to_flush, 0);
4178         INIT_LIST_HEAD(&wq->pwqs);
4179         INIT_LIST_HEAD(&wq->flusher_queue);
4180         INIT_LIST_HEAD(&wq->flusher_overflow);
4181         INIT_LIST_HEAD(&wq->maydays);
4182
4183         lockdep_init_map(&wq->lockdep_map, lock_name, key, 0);
4184         INIT_LIST_HEAD(&wq->list);
4185
4186         if (alloc_and_link_pwqs(wq) < 0)
4187                 goto err_free_wq;
4188
4189         /*
4190          * Workqueues which may be used during memory reclaim should
4191          * have a rescuer to guarantee forward progress.
4192          */
4193         if (flags & WQ_MEM_RECLAIM) {
4194                 struct worker *rescuer;
4195
4196                 rescuer = alloc_worker();
4197                 if (!rescuer)
4198                         goto err_destroy;
4199
4200                 rescuer->rescue_wq = wq;
4201                 rescuer->task = kthread_create(rescuer_thread, rescuer, "%s",
4202                                                wq->name);
4203                 if (IS_ERR(rescuer->task)) {
4204                         kfree(rescuer);
4205                         goto err_destroy;
4206                 }
4207
4208                 wq->rescuer = rescuer;
4209                 rescuer->task->flags |= PF_NO_SETAFFINITY;
4210                 wake_up_process(rescuer->task);
4211         }
4212
4213         if ((wq->flags & WQ_SYSFS) && workqueue_sysfs_register(wq))
4214                 goto err_destroy;
4215
4216         /*
4217          * wq_pool_mutex protects global freeze state and workqueues list.
4218          * Grab it, adjust max_active and add the new @wq to workqueues
4219          * list.
4220          */
4221         mutex_lock(&wq_pool_mutex);
4222
4223         mutex_lock(&wq->mutex);
4224         for_each_pwq(pwq, wq)
4225                 pwq_adjust_max_active(pwq);
4226         mutex_unlock(&wq->mutex);
4227
4228         list_add(&wq->list, &workqueues);
4229
4230         mutex_unlock(&wq_pool_mutex);
4231
4232         return wq;
4233
4234 err_free_wq:
4235         free_workqueue_attrs(wq->unbound_attrs);
4236         kfree(wq);
4237         return NULL;
4238 err_destroy:
4239         destroy_workqueue(wq);
4240         return NULL;
4241 }
4242 EXPORT_SYMBOL_GPL(__alloc_workqueue_key);
4243
4244 /**
4245  * destroy_workqueue - safely terminate a workqueue
4246  * @wq: target workqueue
4247  *
4248  * Safely destroy a workqueue. All work currently pending will be done first.
4249  */
4250 void destroy_workqueue(struct workqueue_struct *wq)
4251 {
4252         struct pool_workqueue *pwq;
4253         int node;
4254
4255         /* drain it before proceeding with destruction */
4256         drain_workqueue(wq);
4257
4258         /* sanity checks */
4259         mutex_lock(&wq->mutex);
4260         for_each_pwq(pwq, wq) {
4261                 int i;
4262
4263                 for (i = 0; i < WORK_NR_COLORS; i++) {
4264                         if (WARN_ON(pwq->nr_in_flight[i])) {
4265                                 mutex_unlock(&wq->mutex);
4266                                 return;
4267                         }
4268                 }
4269
4270                 if (WARN_ON((pwq != wq->dfl_pwq) && (pwq->refcnt > 1)) ||
4271                     WARN_ON(pwq->nr_active) ||
4272                     WARN_ON(!list_empty(&pwq->delayed_works))) {
4273                         mutex_unlock(&wq->mutex);
4274                         return;
4275                 }
4276         }
4277         mutex_unlock(&wq->mutex);
4278
4279         /*
4280          * wq list is used to freeze wq, remove from list after
4281          * flushing is complete in case freeze races us.
4282          */
4283         mutex_lock(&wq_pool_mutex);
4284         list_del_init(&wq->list);
4285         mutex_unlock(&wq_pool_mutex);
4286
4287         workqueue_sysfs_unregister(wq);
4288
4289         if (wq->rescuer) {
4290                 kthread_stop(wq->rescuer->task);
4291                 kfree(wq->rescuer);
4292                 wq->rescuer = NULL;
4293         }
4294
4295         if (!(wq->flags & WQ_UNBOUND)) {
4296                 /*
4297                  * The base ref is never dropped on per-cpu pwqs.  Directly
4298                  * free the pwqs and wq.
4299                  */
4300                 free_percpu(wq->cpu_pwqs);
4301                 kfree(wq);
4302         } else {
4303                 /*
4304                  * We're the sole accessor of @wq at this point.  Directly
4305                  * access numa_pwq_tbl[] and dfl_pwq to put the base refs.
4306                  * @wq will be freed when the last pwq is released.
4307                  */
4308                 for_each_node(node) {
4309                         pwq = rcu_access_pointer(wq->numa_pwq_tbl[node]);
4310                         RCU_INIT_POINTER(wq->numa_pwq_tbl[node], NULL);
4311                         put_pwq_unlocked(pwq);
4312                 }
4313
4314                 /*
4315                  * Put dfl_pwq.  @wq may be freed any time after dfl_pwq is
4316                  * put.  Don't access it afterwards.
4317                  */
4318                 pwq = wq->dfl_pwq;
4319                 wq->dfl_pwq = NULL;
4320                 put_pwq_unlocked(pwq);
4321         }
4322 }
4323 EXPORT_SYMBOL_GPL(destroy_workqueue);
4324
4325 /**
4326  * workqueue_set_max_active - adjust max_active of a workqueue
4327  * @wq: target workqueue
4328  * @max_active: new max_active value.
4329  *
4330  * Set max_active of @wq to @max_active.
4331  *
4332  * CONTEXT:
4333  * Don't call from IRQ context.
4334  */
4335 void workqueue_set_max_active(struct workqueue_struct *wq, int max_active)
4336 {
4337         struct pool_workqueue *pwq;
4338
4339         /* disallow meddling with max_active for ordered workqueues */
4340         if (WARN_ON(wq->flags & __WQ_ORDERED))
4341                 return;
4342
4343         max_active = wq_clamp_max_active(max_active, wq->flags, wq->name);
4344
4345         mutex_lock(&wq->mutex);
4346
4347         wq->saved_max_active = max_active;
4348
4349         for_each_pwq(pwq, wq)
4350                 pwq_adjust_max_active(pwq);
4351
4352         mutex_unlock(&wq->mutex);
4353 }
4354 EXPORT_SYMBOL_GPL(workqueue_set_max_active);
4355
4356 /**
4357  * current_is_workqueue_rescuer - is %current workqueue rescuer?
4358  *
4359  * Determine whether %current is a workqueue rescuer.  Can be used from
4360  * work functions to determine whether it's being run off the rescuer task.
4361  */
4362 bool current_is_workqueue_rescuer(void)
4363 {
4364         struct worker *worker = current_wq_worker();
4365
4366         return worker && worker->rescue_wq;
4367 }
4368
4369 /**
4370  * workqueue_congested - test whether a workqueue is congested
4371  * @cpu: CPU in question
4372  * @wq: target workqueue
4373  *
4374  * Test whether @wq's cpu workqueue for @cpu is congested.  There is
4375  * no synchronization around this function and the test result is
4376  * unreliable and only useful as advisory hints or for debugging.
4377  *
4378  * If @cpu is WORK_CPU_UNBOUND, the test is performed on the local CPU.
4379  * Note that both per-cpu and unbound workqueues may be associated with
4380  * multiple pool_workqueues which have separate congested states.  A
4381  * workqueue being congested on one CPU doesn't mean the workqueue is also
4382  * contested on other CPUs / NUMA nodes.
4383  *
4384  * RETURNS:
4385  * %true if congested, %false otherwise.
4386  */
4387 bool workqueue_congested(int cpu, struct workqueue_struct *wq)
4388 {
4389         struct pool_workqueue *pwq;
4390         bool ret;
4391
4392         rcu_read_lock_sched();
4393
4394         if (cpu == WORK_CPU_UNBOUND)
4395                 cpu = smp_processor_id();
4396
4397         if (!(wq->flags & WQ_UNBOUND))
4398                 pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
4399         else
4400                 pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
4401
4402         ret = !list_empty(&pwq->delayed_works);
4403         rcu_read_unlock_sched();
4404
4405         return ret;
4406 }
4407 EXPORT_SYMBOL_GPL(workqueue_congested);
4408
4409 /**
4410  * work_busy - test whether a work is currently pending or running
4411  * @work: the work to be tested
4412  *
4413  * Test whether @work is currently pending or running.  There is no
4414  * synchronization around this function and the test result is
4415  * unreliable and only useful as advisory hints or for debugging.
4416  *
4417  * RETURNS:
4418  * OR'd bitmask of WORK_BUSY_* bits.
4419  */
4420 unsigned int work_busy(struct work_struct *work)
4421 {
4422         struct worker_pool *pool;
4423         unsigned long flags;
4424         unsigned int ret = 0;
4425
4426         if (work_pending(work))
4427                 ret |= WORK_BUSY_PENDING;
4428
4429         local_irq_save(flags);
4430         pool = get_work_pool(work);
4431         if (pool) {
4432                 spin_lock(&pool->lock);
4433                 if (find_worker_executing_work(pool, work))
4434                         ret |= WORK_BUSY_RUNNING;
4435                 spin_unlock(&pool->lock);
4436         }
4437         local_irq_restore(flags);
4438
4439         return ret;
4440 }
4441 EXPORT_SYMBOL_GPL(work_busy);
4442
4443 /**
4444  * set_worker_desc - set description for the current work item
4445  * @fmt: printf-style format string
4446  * @...: arguments for the format string
4447  *
4448  * This function can be called by a running work function to describe what
4449  * the work item is about.  If the worker task gets dumped, this
4450  * information will be printed out together to help debugging.  The
4451  * description can be at most WORKER_DESC_LEN including the trailing '\0'.
4452  */
4453 void set_worker_desc(const char *fmt, ...)
4454 {
4455         struct worker *worker = current_wq_worker();
4456         va_list args;
4457
4458         if (worker) {
4459                 va_start(args, fmt);
4460                 vsnprintf(worker->desc, sizeof(worker->desc), fmt, args);
4461                 va_end(args);
4462                 worker->desc_valid = true;
4463         }
4464 }
4465
4466 /**
4467  * print_worker_info - print out worker information and description
4468  * @log_lvl: the log level to use when printing
4469  * @task: target task
4470  *
4471  * If @task is a worker and currently executing a work item, print out the
4472  * name of the workqueue being serviced and worker description set with
4473  * set_worker_desc() by the currently executing work item.
4474  *
4475  * This function can be safely called on any task as long as the
4476  * task_struct itself is accessible.  While safe, this function isn't
4477  * synchronized and may print out mixups or garbages of limited length.
4478  */
4479 void print_worker_info(const char *log_lvl, struct task_struct *task)
4480 {
4481         work_func_t *fn = NULL;
4482         char name[WQ_NAME_LEN] = { };
4483         char desc[WORKER_DESC_LEN] = { };
4484         struct pool_workqueue *pwq = NULL;
4485         struct workqueue_struct *wq = NULL;
4486         bool desc_valid = false;
4487         struct worker *worker;
4488
4489         if (!(task->flags & PF_WQ_WORKER))
4490                 return;
4491
4492         /*
4493          * This function is called without any synchronization and @task
4494          * could be in any state.  Be careful with dereferences.
4495          */
4496         worker = probe_kthread_data(task);
4497
4498         /*
4499          * Carefully copy the associated workqueue's workfn and name.  Keep
4500          * the original last '\0' in case the original contains garbage.
4501          */
4502         probe_kernel_read(&fn, &worker->current_func, sizeof(fn));
4503         probe_kernel_read(&pwq, &worker->current_pwq, sizeof(pwq));
4504         probe_kernel_read(&wq, &pwq->wq, sizeof(wq));
4505         probe_kernel_read(name, wq->name, sizeof(name) - 1);
4506
4507         /* copy worker description */
4508         probe_kernel_read(&desc_valid, &worker->desc_valid, sizeof(desc_valid));
4509         if (desc_valid)
4510                 probe_kernel_read(desc, worker->desc, sizeof(desc) - 1);
4511
4512         if (fn || name[0] || desc[0]) {
4513                 printk("%sWorkqueue: %s %pf", log_lvl, name, fn);
4514                 if (desc[0])
4515                         pr_cont(" (%s)", desc);
4516                 pr_cont("\n");
4517         }
4518 }
4519
4520 /*
4521  * CPU hotplug.
4522  *
4523  * There are two challenges in supporting CPU hotplug.  Firstly, there
4524  * are a lot of assumptions on strong associations among work, pwq and
4525  * pool which make migrating pending and scheduled works very
4526  * difficult to implement without impacting hot paths.  Secondly,
4527  * worker pools serve mix of short, long and very long running works making
4528  * blocked draining impractical.
4529  *
4530  * This is solved by allowing the pools to be disassociated from the CPU
4531  * running as an unbound one and allowing it to be reattached later if the
4532  * cpu comes back online.
4533  */
4534
4535 static void wq_unbind_fn(struct work_struct *work)
4536 {
4537         int cpu = smp_processor_id();
4538         struct worker_pool *pool;
4539         struct worker *worker;
4540         int wi;
4541
4542         for_each_cpu_worker_pool(pool, cpu) {
4543                 WARN_ON_ONCE(cpu != smp_processor_id());
4544
4545                 mutex_lock(&pool->manager_mutex);
4546                 spin_lock_irq(&pool->lock);
4547
4548                 /*
4549                  * We've blocked all manager operations.  Make all workers
4550                  * unbound and set DISASSOCIATED.  Before this, all workers
4551                  * except for the ones which are still executing works from
4552                  * before the last CPU down must be on the cpu.  After
4553                  * this, they may become diasporas.
4554                  */
4555                 for_each_pool_worker(worker, wi, pool)
4556                         worker->flags |= WORKER_UNBOUND;
4557
4558                 pool->flags |= POOL_DISASSOCIATED;
4559
4560                 spin_unlock_irq(&pool->lock);
4561                 mutex_unlock(&pool->manager_mutex);
4562
4563                 /*
4564                  * Call schedule() so that we cross rq->lock and thus can
4565                  * guarantee sched callbacks see the %WORKER_UNBOUND flag.
4566                  * This is necessary as scheduler callbacks may be invoked
4567                  * from other cpus.
4568                  */
4569                 schedule();
4570
4571                 /*
4572                  * Sched callbacks are disabled now.  Zap nr_running.
4573                  * After this, nr_running stays zero and need_more_worker()
4574                  * and keep_working() are always true as long as the
4575                  * worklist is not empty.  This pool now behaves as an
4576                  * unbound (in terms of concurrency management) pool which
4577                  * are served by workers tied to the pool.
4578                  */
4579                 atomic_set(&pool->nr_running, 0);
4580
4581                 /*
4582                  * With concurrency management just turned off, a busy
4583                  * worker blocking could lead to lengthy stalls.  Kick off
4584                  * unbound chain execution of currently pending work items.
4585                  */
4586                 spin_lock_irq(&pool->lock);
4587                 wake_up_worker(pool);
4588                 spin_unlock_irq(&pool->lock);
4589         }
4590 }
4591
4592 /**
4593  * rebind_workers - rebind all workers of a pool to the associated CPU
4594  * @pool: pool of interest
4595  *
4596  * @pool->cpu is coming online.  Rebind all workers to the CPU.
4597  */
4598 static void rebind_workers(struct worker_pool *pool)
4599 {
4600         struct worker *worker;
4601         int wi;
4602
4603         lockdep_assert_held(&pool->manager_mutex);
4604
4605         /*
4606          * Restore CPU affinity of all workers.  As all idle workers should
4607          * be on the run-queue of the associated CPU before any local
4608          * wake-ups for concurrency management happen, restore CPU affinty
4609          * of all workers first and then clear UNBOUND.  As we're called
4610          * from CPU_ONLINE, the following shouldn't fail.
4611          */
4612         for_each_pool_worker(worker, wi, pool)
4613                 WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task,
4614                                                   pool->attrs->cpumask) < 0);
4615
4616         spin_lock_irq(&pool->lock);
4617
4618         for_each_pool_worker(worker, wi, pool) {
4619                 unsigned int worker_flags = worker->flags;
4620
4621                 /*
4622                  * A bound idle worker should actually be on the runqueue
4623                  * of the associated CPU for local wake-ups targeting it to
4624                  * work.  Kick all idle workers so that they migrate to the
4625                  * associated CPU.  Doing this in the same loop as
4626                  * replacing UNBOUND with REBOUND is safe as no worker will
4627                  * be bound before @pool->lock is released.
4628                  */
4629                 if (worker_flags & WORKER_IDLE)
4630                         wake_up_process(worker->task);
4631
4632                 /*
4633                  * We want to clear UNBOUND but can't directly call
4634                  * worker_clr_flags() or adjust nr_running.  Atomically
4635                  * replace UNBOUND with another NOT_RUNNING flag REBOUND.
4636                  * @worker will clear REBOUND using worker_clr_flags() when
4637                  * it initiates the next execution cycle thus restoring
4638                  * concurrency management.  Note that when or whether
4639                  * @worker clears REBOUND doesn't affect correctness.
4640                  *
4641                  * ACCESS_ONCE() is necessary because @worker->flags may be
4642                  * tested without holding any lock in
4643                  * wq_worker_waking_up().  Without it, NOT_RUNNING test may
4644                  * fail incorrectly leading to premature concurrency
4645                  * management operations.
4646                  */
4647                 WARN_ON_ONCE(!(worker_flags & WORKER_UNBOUND));
4648                 worker_flags |= WORKER_REBOUND;
4649                 worker_flags &= ~WORKER_UNBOUND;
4650                 ACCESS_ONCE(worker->flags) = worker_flags;
4651         }
4652
4653         spin_unlock_irq(&pool->lock);
4654 }
4655
4656 /**
4657  * restore_unbound_workers_cpumask - restore cpumask of unbound workers
4658  * @pool: unbound pool of interest
4659  * @cpu: the CPU which is coming up
4660  *
4661  * An unbound pool may end up with a cpumask which doesn't have any online
4662  * CPUs.  When a worker of such pool get scheduled, the scheduler resets
4663  * its cpus_allowed.  If @cpu is in @pool's cpumask which didn't have any
4664  * online CPU before, cpus_allowed of all its workers should be restored.
4665  */
4666 static void restore_unbound_workers_cpumask(struct worker_pool *pool, int cpu)
4667 {
4668         static cpumask_t cpumask;
4669         struct worker *worker;
4670         int wi;
4671
4672         lockdep_assert_held(&pool->manager_mutex);
4673
4674         /* is @cpu allowed for @pool? */
4675         if (!cpumask_test_cpu(cpu, pool->attrs->cpumask))
4676                 return;
4677
4678         /* is @cpu the only online CPU? */
4679         cpumask_and(&cpumask, pool->attrs->cpumask, cpu_online_mask);
4680         if (cpumask_weight(&cpumask) != 1)
4681                 return;
4682
4683         /* as we're called from CPU_ONLINE, the following shouldn't fail */
4684         for_each_pool_worker(worker, wi, pool)
4685                 WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task,
4686                                                   pool->attrs->cpumask) < 0);
4687 }
4688
4689 /*
4690  * Workqueues should be brought up before normal priority CPU notifiers.
4691  * This will be registered high priority CPU notifier.
4692  */
4693 static int __cpuinit workqueue_cpu_up_callback(struct notifier_block *nfb,
4694                                                unsigned long action,
4695                                                void *hcpu)
4696 {
4697         int cpu = (unsigned long)hcpu;
4698         struct worker_pool *pool;
4699         struct workqueue_struct *wq;
4700         int pi;
4701
4702         switch (action & ~CPU_TASKS_FROZEN) {
4703         case CPU_UP_PREPARE:
4704                 for_each_cpu_worker_pool(pool, cpu) {
4705                         if (pool->nr_workers)
4706                                 continue;
4707                         if (create_and_start_worker(pool) < 0)
4708                                 return NOTIFY_BAD;
4709                 }
4710                 break;
4711
4712         case CPU_DOWN_FAILED:
4713         case CPU_ONLINE:
4714                 mutex_lock(&wq_pool_mutex);
4715
4716                 for_each_pool(pool, pi) {
4717                         mutex_lock(&pool->manager_mutex);
4718
4719                         if (pool->cpu == cpu) {
4720                                 spin_lock_irq(&pool->lock);
4721                                 pool->flags &= ~POOL_DISASSOCIATED;
4722                                 spin_unlock_irq(&pool->lock);
4723
4724                                 rebind_workers(pool);
4725                         } else if (pool->cpu < 0) {
4726                                 restore_unbound_workers_cpumask(pool, cpu);
4727                         }
4728
4729                         mutex_unlock(&pool->manager_mutex);
4730                 }
4731
4732                 /* update NUMA affinity of unbound workqueues */
4733                 list_for_each_entry(wq, &workqueues, list)
4734                         wq_update_unbound_numa(wq, cpu, true);
4735
4736                 mutex_unlock(&wq_pool_mutex);
4737                 break;
4738         }
4739         return NOTIFY_OK;
4740 }
4741
4742 /*
4743  * Workqueues should be brought down after normal priority CPU notifiers.
4744  * This will be registered as low priority CPU notifier.
4745  */
4746 static int __cpuinit workqueue_cpu_down_callback(struct notifier_block *nfb,
4747                                                  unsigned long action,
4748                                                  void *hcpu)
4749 {
4750         int cpu = (unsigned long)hcpu;
4751         struct work_struct unbind_work;
4752         struct workqueue_struct *wq;
4753
4754         switch (action & ~CPU_TASKS_FROZEN) {
4755         case CPU_DOWN_PREPARE:
4756                 /* unbinding per-cpu workers should happen on the local CPU */
4757                 INIT_WORK_ONSTACK(&unbind_work, wq_unbind_fn);
4758                 queue_work_on(cpu, system_highpri_wq, &unbind_work);
4759
4760                 /* update NUMA affinity of unbound workqueues */
4761                 mutex_lock(&wq_pool_mutex);
4762                 list_for_each_entry(wq, &workqueues, list)
4763                         wq_update_unbound_numa(wq, cpu, false);
4764                 mutex_unlock(&wq_pool_mutex);
4765
4766                 /* wait for per-cpu unbinding to finish */
4767                 flush_work(&unbind_work);
4768                 break;
4769         }
4770         return NOTIFY_OK;
4771 }
4772
4773 #ifdef CONFIG_SMP
4774
4775 struct work_for_cpu {
4776         struct work_struct work;
4777         long (*fn)(void *);
4778         void *arg;
4779         long ret;
4780 };
4781
4782 static void work_for_cpu_fn(struct work_struct *work)
4783 {
4784         struct work_for_cpu *wfc = container_of(work, struct work_for_cpu, work);
4785
4786         wfc->ret = wfc->fn(wfc->arg);
4787 }
4788
4789 /**
4790  * work_on_cpu - run a function in user context on a particular cpu
4791  * @cpu: the cpu to run on
4792  * @fn: the function to run
4793  * @arg: the function arg
4794  *
4795  * This will return the value @fn returns.
4796  * It is up to the caller to ensure that the cpu doesn't go offline.
4797  * The caller must not hold any locks which would prevent @fn from completing.
4798  */
4799 long work_on_cpu(int cpu, long (*fn)(void *), void *arg)
4800 {
4801         struct work_for_cpu wfc = { .fn = fn, .arg = arg };
4802
4803         INIT_WORK_ONSTACK(&wfc.work, work_for_cpu_fn);
4804         schedule_work_on(cpu, &wfc.work);
4805         flush_work(&wfc.work);
4806         return wfc.ret;
4807 }
4808 EXPORT_SYMBOL_GPL(work_on_cpu);
4809 #endif /* CONFIG_SMP */
4810
4811 #ifdef CONFIG_FREEZER
4812
4813 /**
4814  * freeze_workqueues_begin - begin freezing workqueues
4815  *
4816  * Start freezing workqueues.  After this function returns, all freezable
4817  * workqueues will queue new works to their delayed_works list instead of
4818  * pool->worklist.
4819  *
4820  * CONTEXT:
4821  * Grabs and releases wq_pool_mutex, wq->mutex and pool->lock's.
4822  */
4823 void freeze_workqueues_begin(void)
4824 {
4825         struct worker_pool *pool;
4826         struct workqueue_struct *wq;
4827         struct pool_workqueue *pwq;
4828         int pi;
4829
4830         mutex_lock(&wq_pool_mutex);
4831
4832         WARN_ON_ONCE(workqueue_freezing);
4833         workqueue_freezing = true;
4834
4835         /* set FREEZING */
4836         for_each_pool(pool, pi) {
4837                 spin_lock_irq(&pool->lock);
4838                 WARN_ON_ONCE(pool->flags & POOL_FREEZING);
4839                 pool->flags |= POOL_FREEZING;
4840                 spin_unlock_irq(&pool->lock);
4841         }
4842
4843         list_for_each_entry(wq, &workqueues, list) {
4844                 mutex_lock(&wq->mutex);
4845                 for_each_pwq(pwq, wq)
4846                         pwq_adjust_max_active(pwq);
4847                 mutex_unlock(&wq->mutex);
4848         }
4849
4850         mutex_unlock(&wq_pool_mutex);
4851 }
4852
4853 /**
4854  * freeze_workqueues_busy - are freezable workqueues still busy?
4855  *
4856  * Check whether freezing is complete.  This function must be called
4857  * between freeze_workqueues_begin() and thaw_workqueues().
4858  *
4859  * CONTEXT:
4860  * Grabs and releases wq_pool_mutex.
4861  *
4862  * RETURNS:
4863  * %true if some freezable workqueues are still busy.  %false if freezing
4864  * is complete.
4865  */
4866 bool freeze_workqueues_busy(void)
4867 {
4868         bool busy = false;
4869         struct workqueue_struct *wq;
4870         struct pool_workqueue *pwq;
4871
4872         mutex_lock(&wq_pool_mutex);
4873
4874         WARN_ON_ONCE(!workqueue_freezing);
4875
4876         list_for_each_entry(wq, &workqueues, list) {
4877                 if (!(wq->flags & WQ_FREEZABLE))
4878                         continue;
4879                 /*
4880                  * nr_active is monotonically decreasing.  It's safe
4881                  * to peek without lock.
4882                  */
4883                 rcu_read_lock_sched();
4884                 for_each_pwq(pwq, wq) {
4885                         WARN_ON_ONCE(pwq->nr_active < 0);
4886                         if (pwq->nr_active) {
4887                                 busy = true;
4888                                 rcu_read_unlock_sched();
4889                                 goto out_unlock;
4890                         }
4891                 }
4892                 rcu_read_unlock_sched();
4893         }
4894 out_unlock:
4895         mutex_unlock(&wq_pool_mutex);
4896         return busy;
4897 }
4898
4899 /**
4900  * thaw_workqueues - thaw workqueues
4901  *
4902  * Thaw workqueues.  Normal queueing is restored and all collected
4903  * frozen works are transferred to their respective pool worklists.
4904  *
4905  * CONTEXT:
4906  * Grabs and releases wq_pool_mutex, wq->mutex and pool->lock's.
4907  */
4908 void thaw_workqueues(void)
4909 {
4910         struct workqueue_struct *wq;
4911         struct pool_workqueue *pwq;
4912         struct worker_pool *pool;
4913         int pi;
4914
4915         mutex_lock(&wq_pool_mutex);
4916
4917         if (!workqueue_freezing)
4918                 goto out_unlock;
4919
4920         /* clear FREEZING */
4921         for_each_pool(pool, pi) {
4922                 spin_lock_irq(&pool->lock);
4923                 WARN_ON_ONCE(!(pool->flags & POOL_FREEZING));
4924                 pool->flags &= ~POOL_FREEZING;
4925                 spin_unlock_irq(&pool->lock);
4926         }
4927
4928         /* restore max_active and repopulate worklist */
4929         list_for_each_entry(wq, &workqueues, list) {
4930                 mutex_lock(&wq->mutex);
4931                 for_each_pwq(pwq, wq)
4932                         pwq_adjust_max_active(pwq);
4933                 mutex_unlock(&wq->mutex);
4934         }
4935
4936         workqueue_freezing = false;
4937 out_unlock:
4938         mutex_unlock(&wq_pool_mutex);
4939 }
4940 #endif /* CONFIG_FREEZER */
4941
4942 static void __init wq_numa_init(void)
4943 {
4944         cpumask_var_t *tbl;
4945         int node, cpu;
4946
4947         /* determine NUMA pwq table len - highest node id + 1 */
4948         for_each_node(node)
4949                 wq_numa_tbl_len = max(wq_numa_tbl_len, node + 1);
4950
4951         if (num_possible_nodes() <= 1)
4952                 return;
4953
4954         if (wq_disable_numa) {
4955                 pr_info("workqueue: NUMA affinity support disabled\n");
4956                 return;
4957         }
4958
4959         wq_update_unbound_numa_attrs_buf = alloc_workqueue_attrs(GFP_KERNEL);
4960         BUG_ON(!wq_update_unbound_numa_attrs_buf);
4961
4962         /*
4963          * We want masks of possible CPUs of each node which isn't readily
4964          * available.  Build one from cpu_to_node() which should have been
4965          * fully initialized by now.
4966          */
4967         tbl = kzalloc(wq_numa_tbl_len * sizeof(tbl[0]), GFP_KERNEL);
4968         BUG_ON(!tbl);
4969
4970         for_each_node(node)
4971                 BUG_ON(!zalloc_cpumask_var_node(&tbl[node], GFP_KERNEL,
4972                                 node_online(node) ? node : NUMA_NO_NODE));
4973
4974         for_each_possible_cpu(cpu) {
4975                 node = cpu_to_node(cpu);
4976                 if (WARN_ON(node == NUMA_NO_NODE)) {
4977                         pr_warn("workqueue: NUMA node mapping not available for cpu%d, disabling NUMA support\n", cpu);
4978                         /* happens iff arch is bonkers, let's just proceed */
4979                         return;
4980                 }
4981                 cpumask_set_cpu(cpu, tbl[node]);
4982         }
4983
4984         wq_numa_possible_cpumask = tbl;
4985         wq_numa_enabled = true;
4986 }
4987
4988 static int __init init_workqueues(void)
4989 {
4990         int std_nice[NR_STD_WORKER_POOLS] = { 0, HIGHPRI_NICE_LEVEL };
4991         int i, cpu;
4992
4993         /* make sure we have enough bits for OFFQ pool ID */
4994         BUILD_BUG_ON((1LU << (BITS_PER_LONG - WORK_OFFQ_POOL_SHIFT)) <
4995                      WORK_CPU_END * NR_STD_WORKER_POOLS);
4996
4997         WARN_ON(__alignof__(struct pool_workqueue) < __alignof__(long long));
4998
4999         pwq_cache = KMEM_CACHE(pool_workqueue, SLAB_PANIC);
5000
5001         cpu_notifier(workqueue_cpu_up_callback, CPU_PRI_WORKQUEUE_UP);
5002         hotcpu_notifier(workqueue_cpu_down_callback, CPU_PRI_WORKQUEUE_DOWN);
5003
5004         wq_numa_init();
5005
5006         /* initialize CPU pools */
5007         for_each_possible_cpu(cpu) {
5008                 struct worker_pool *pool;
5009
5010                 i = 0;
5011                 for_each_cpu_worker_pool(pool, cpu) {
5012                         BUG_ON(init_worker_pool(pool));
5013                         pool->cpu = cpu;
5014                         cpumask_copy(pool->attrs->cpumask, cpumask_of(cpu));
5015                         pool->attrs->nice = std_nice[i++];
5016                         pool->node = cpu_to_node(cpu);
5017
5018                         /* alloc pool ID */
5019                         mutex_lock(&wq_pool_mutex);
5020                         BUG_ON(worker_pool_assign_id(pool));
5021                         mutex_unlock(&wq_pool_mutex);
5022                 }
5023         }
5024
5025         /* create the initial worker */
5026         for_each_online_cpu(cpu) {
5027                 struct worker_pool *pool;
5028
5029                 for_each_cpu_worker_pool(pool, cpu) {
5030                         pool->flags &= ~POOL_DISASSOCIATED;
5031                         BUG_ON(create_and_start_worker(pool) < 0);
5032                 }
5033         }
5034
5035         /* create default unbound and ordered wq attrs */
5036         for (i = 0; i < NR_STD_WORKER_POOLS; i++) {
5037                 struct workqueue_attrs *attrs;
5038
5039                 BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
5040                 attrs->nice = std_nice[i];
5041                 unbound_std_wq_attrs[i] = attrs;
5042
5043                 /*
5044                  * An ordered wq should have only one pwq as ordering is
5045                  * guaranteed by max_active which is enforced by pwqs.
5046                  * Turn off NUMA so that dfl_pwq is used for all nodes.
5047                  */
5048                 BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
5049                 attrs->nice = std_nice[i];
5050                 attrs->no_numa = true;
5051                 ordered_wq_attrs[i] = attrs;
5052         }
5053
5054         system_wq = alloc_workqueue("events", 0, 0);
5055         system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
5056         system_long_wq = alloc_workqueue("events_long", 0, 0);
5057         system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
5058                                             WQ_UNBOUND_MAX_ACTIVE);
5059         system_freezable_wq = alloc_workqueue("events_freezable",
5060                                               WQ_FREEZABLE, 0);
5061         BUG_ON(!system_wq || !system_highpri_wq || !system_long_wq ||
5062                !system_unbound_wq || !system_freezable_wq);
5063         return 0;
5064 }
5065 early_initcall(init_workqueues);