Merge drm-misc-next-fixes-2019-05-20 into drm-misc-fixes
[linux-2.6-microblaze.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 are two worker pools for each CPU (one for
20  * normal work items and the other for high priority ones) and some extra
21  * pools for workqueues which are not bound to any specific CPU - the
22  * number of these backing pools is dynamic.
23  *
24  * Please read Documentation/core-api/workqueue.rst for details.
25  */
26
27 #include <linux/export.h>
28 #include <linux/kernel.h>
29 #include <linux/sched.h>
30 #include <linux/init.h>
31 #include <linux/signal.h>
32 #include <linux/completion.h>
33 #include <linux/workqueue.h>
34 #include <linux/slab.h>
35 #include <linux/cpu.h>
36 #include <linux/notifier.h>
37 #include <linux/kthread.h>
38 #include <linux/hardirq.h>
39 #include <linux/mempolicy.h>
40 #include <linux/freezer.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 #include <linux/sched/isolation.h>
51 #include <linux/nmi.h>
52
53 #include "workqueue_internal.h"
54
55 enum {
56         /*
57          * worker_pool flags
58          *
59          * A bound pool is either associated or disassociated with its CPU.
60          * While associated (!DISASSOCIATED), all workers are bound to the
61          * CPU and none has %WORKER_UNBOUND set and concurrency management
62          * is in effect.
63          *
64          * While DISASSOCIATED, the cpu may be offline and all workers have
65          * %WORKER_UNBOUND set and concurrency management disabled, and may
66          * be executing on any CPU.  The pool behaves as an unbound one.
67          *
68          * Note that DISASSOCIATED should be flipped only while holding
69          * wq_pool_attach_mutex to avoid changing binding state while
70          * worker_attach_to_pool() is in progress.
71          */
72         POOL_MANAGER_ACTIVE     = 1 << 0,       /* being managed */
73         POOL_DISASSOCIATED      = 1 << 2,       /* cpu can't serve workers */
74
75         /* worker flags */
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 MIN_NICE.
103          */
104         RESCUER_NICE_LEVEL      = MIN_NICE,
105         HIGHPRI_NICE_LEVEL      = MIN_NICE,
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  * A: wq_pool_attach_mutex protected.
127  *
128  * PL: wq_pool_mutex protected.
129  *
130  * PR: wq_pool_mutex protected for writes.  RCU protected for reads.
131  *
132  * PW: wq_pool_mutex and wq->mutex protected for writes.  Either for reads.
133  *
134  * PWR: wq_pool_mutex and wq->mutex protected for writes.  Either or
135  *      RCU for reads.
136  *
137  * WQ: wq->mutex protected.
138  *
139  * WR: wq->mutex protected for writes.  RCU protected for reads.
140  *
141  * MD: wq_mayday_lock protected.
142  */
143
144 /* struct worker is defined in workqueue_internal.h */
145
146 struct worker_pool {
147         spinlock_t              lock;           /* the pool lock */
148         int                     cpu;            /* I: the associated cpu */
149         int                     node;           /* I: the associated node ID */
150         int                     id;             /* I: pool ID */
151         unsigned int            flags;          /* X: flags */
152
153         unsigned long           watchdog_ts;    /* L: watchdog timestamp */
154
155         struct list_head        worklist;       /* L: list of pending works */
156
157         int                     nr_workers;     /* L: total number of workers */
158         int                     nr_idle;        /* L: currently idle workers */
159
160         struct list_head        idle_list;      /* X: list of idle workers */
161         struct timer_list       idle_timer;     /* L: worker idle timeout */
162         struct timer_list       mayday_timer;   /* L: SOS timer for workers */
163
164         /* a workers is either on busy_hash or idle_list, or the manager */
165         DECLARE_HASHTABLE(busy_hash, BUSY_WORKER_HASH_ORDER);
166                                                 /* L: hash of busy workers */
167
168         struct worker           *manager;       /* L: purely informational */
169         struct list_head        workers;        /* A: attached workers */
170         struct completion       *detach_completion; /* all workers detached */
171
172         struct ida              worker_ida;     /* worker IDs for task name */
173
174         struct workqueue_attrs  *attrs;         /* I: worker attributes */
175         struct hlist_node       hash_node;      /* PL: unbound_pool_hash node */
176         int                     refcnt;         /* PL: refcnt for unbound pools */
177
178         /*
179          * The current concurrency level.  As it's likely to be accessed
180          * from other CPUs during try_to_wake_up(), put it in a separate
181          * cacheline.
182          */
183         atomic_t                nr_running ____cacheline_aligned_in_smp;
184
185         /*
186          * Destruction of pool is RCU protected to allow dereferences
187          * from get_work_pool().
188          */
189         struct rcu_head         rcu;
190 } ____cacheline_aligned_in_smp;
191
192 /*
193  * The per-pool workqueue.  While queued, the lower WORK_STRUCT_FLAG_BITS
194  * of work_struct->data are used for flags and the remaining high bits
195  * point to the pwq; thus, pwqs need to be aligned at two's power of the
196  * number of flag bits.
197  */
198 struct pool_workqueue {
199         struct worker_pool      *pool;          /* I: the associated pool */
200         struct workqueue_struct *wq;            /* I: the owning workqueue */
201         int                     work_color;     /* L: current color */
202         int                     flush_color;    /* L: flushing color */
203         int                     refcnt;         /* L: reference count */
204         int                     nr_in_flight[WORK_NR_COLORS];
205                                                 /* L: nr of in_flight works */
206         int                     nr_active;      /* L: nr of active works */
207         int                     max_active;     /* L: max active works */
208         struct list_head        delayed_works;  /* L: delayed works */
209         struct list_head        pwqs_node;      /* WR: node on wq->pwqs */
210         struct list_head        mayday_node;    /* MD: node on wq->maydays */
211
212         /*
213          * Release of unbound pwq is punted to system_wq.  See put_pwq()
214          * and pwq_unbound_release_workfn() for details.  pool_workqueue
215          * itself is also RCU protected so that the first pwq can be
216          * determined without grabbing wq->mutex.
217          */
218         struct work_struct      unbound_release_work;
219         struct rcu_head         rcu;
220 } __aligned(1 << WORK_STRUCT_FLAG_BITS);
221
222 /*
223  * Structure used to wait for workqueue flush.
224  */
225 struct wq_flusher {
226         struct list_head        list;           /* WQ: list of flushers */
227         int                     flush_color;    /* WQ: flush color waiting for */
228         struct completion       done;           /* flush completion */
229 };
230
231 struct wq_device;
232
233 /*
234  * The externally visible workqueue.  It relays the issued work items to
235  * the appropriate worker_pool through its pool_workqueues.
236  */
237 struct workqueue_struct {
238         struct list_head        pwqs;           /* WR: all pwqs of this wq */
239         struct list_head        list;           /* PR: list of all workqueues */
240
241         struct mutex            mutex;          /* protects this wq */
242         int                     work_color;     /* WQ: current work color */
243         int                     flush_color;    /* WQ: current flush color */
244         atomic_t                nr_pwqs_to_flush; /* flush in progress */
245         struct wq_flusher       *first_flusher; /* WQ: first flusher */
246         struct list_head        flusher_queue;  /* WQ: flush waiters */
247         struct list_head        flusher_overflow; /* WQ: flush overflow list */
248
249         struct list_head        maydays;        /* MD: pwqs requesting rescue */
250         struct worker           *rescuer;       /* I: rescue worker */
251
252         int                     nr_drainers;    /* WQ: drain in progress */
253         int                     saved_max_active; /* WQ: saved pwq max_active */
254
255         struct workqueue_attrs  *unbound_attrs; /* PW: only for unbound wqs */
256         struct pool_workqueue   *dfl_pwq;       /* PW: only for unbound wqs */
257
258 #ifdef CONFIG_SYSFS
259         struct wq_device        *wq_dev;        /* I: for sysfs interface */
260 #endif
261 #ifdef CONFIG_LOCKDEP
262         char                    *lock_name;
263         struct lock_class_key   key;
264         struct lockdep_map      lockdep_map;
265 #endif
266         char                    name[WQ_NAME_LEN]; /* I: workqueue name */
267
268         /*
269          * Destruction of workqueue_struct is RCU protected to allow walking
270          * the workqueues list without grabbing wq_pool_mutex.
271          * This is used to dump all workqueues from sysrq.
272          */
273         struct rcu_head         rcu;
274
275         /* hot fields used during command issue, aligned to cacheline */
276         unsigned int            flags ____cacheline_aligned; /* WQ: WQ_* flags */
277         struct pool_workqueue __percpu *cpu_pwqs; /* I: per-cpu pwqs */
278         struct pool_workqueue __rcu *numa_pwq_tbl[]; /* PWR: unbound pwqs indexed by node */
279 };
280
281 static struct kmem_cache *pwq_cache;
282
283 static cpumask_var_t *wq_numa_possible_cpumask;
284                                         /* possible CPUs of each node */
285
286 static bool wq_disable_numa;
287 module_param_named(disable_numa, wq_disable_numa, bool, 0444);
288
289 /* see the comment above the definition of WQ_POWER_EFFICIENT */
290 static bool wq_power_efficient = IS_ENABLED(CONFIG_WQ_POWER_EFFICIENT_DEFAULT);
291 module_param_named(power_efficient, wq_power_efficient, bool, 0444);
292
293 static bool wq_online;                  /* can kworkers be created yet? */
294
295 static bool wq_numa_enabled;            /* unbound NUMA affinity enabled */
296
297 /* buf for wq_update_unbound_numa_attrs(), protected by CPU hotplug exclusion */
298 static struct workqueue_attrs *wq_update_unbound_numa_attrs_buf;
299
300 static DEFINE_MUTEX(wq_pool_mutex);     /* protects pools and workqueues list */
301 static DEFINE_MUTEX(wq_pool_attach_mutex); /* protects worker attach/detach */
302 static DEFINE_SPINLOCK(wq_mayday_lock); /* protects wq->maydays list */
303 static DECLARE_WAIT_QUEUE_HEAD(wq_manager_wait); /* wait for manager to go away */
304
305 static LIST_HEAD(workqueues);           /* PR: list of all workqueues */
306 static bool workqueue_freezing;         /* PL: have wqs started freezing? */
307
308 /* PL: allowable cpus for unbound wqs and work items */
309 static cpumask_var_t wq_unbound_cpumask;
310
311 /* CPU where unbound work was last round robin scheduled from this CPU */
312 static DEFINE_PER_CPU(int, wq_rr_cpu_last);
313
314 /*
315  * Local execution of unbound work items is no longer guaranteed.  The
316  * following always forces round-robin CPU selection on unbound work items
317  * to uncover usages which depend on it.
318  */
319 #ifdef CONFIG_DEBUG_WQ_FORCE_RR_CPU
320 static bool wq_debug_force_rr_cpu = true;
321 #else
322 static bool wq_debug_force_rr_cpu = false;
323 #endif
324 module_param_named(debug_force_rr_cpu, wq_debug_force_rr_cpu, bool, 0644);
325
326 /* the per-cpu worker pools */
327 static DEFINE_PER_CPU_SHARED_ALIGNED(struct worker_pool [NR_STD_WORKER_POOLS], cpu_worker_pools);
328
329 static DEFINE_IDR(worker_pool_idr);     /* PR: idr of all pools */
330
331 /* PL: hash of all unbound pools keyed by pool->attrs */
332 static DEFINE_HASHTABLE(unbound_pool_hash, UNBOUND_POOL_HASH_ORDER);
333
334 /* I: attributes used when instantiating standard unbound pools on demand */
335 static struct workqueue_attrs *unbound_std_wq_attrs[NR_STD_WORKER_POOLS];
336
337 /* I: attributes used when instantiating ordered pools on demand */
338 static struct workqueue_attrs *ordered_wq_attrs[NR_STD_WORKER_POOLS];
339
340 struct workqueue_struct *system_wq __read_mostly;
341 EXPORT_SYMBOL(system_wq);
342 struct workqueue_struct *system_highpri_wq __read_mostly;
343 EXPORT_SYMBOL_GPL(system_highpri_wq);
344 struct workqueue_struct *system_long_wq __read_mostly;
345 EXPORT_SYMBOL_GPL(system_long_wq);
346 struct workqueue_struct *system_unbound_wq __read_mostly;
347 EXPORT_SYMBOL_GPL(system_unbound_wq);
348 struct workqueue_struct *system_freezable_wq __read_mostly;
349 EXPORT_SYMBOL_GPL(system_freezable_wq);
350 struct workqueue_struct *system_power_efficient_wq __read_mostly;
351 EXPORT_SYMBOL_GPL(system_power_efficient_wq);
352 struct workqueue_struct *system_freezable_power_efficient_wq __read_mostly;
353 EXPORT_SYMBOL_GPL(system_freezable_power_efficient_wq);
354
355 static int worker_thread(void *__worker);
356 static void workqueue_sysfs_unregister(struct workqueue_struct *wq);
357
358 #define CREATE_TRACE_POINTS
359 #include <trace/events/workqueue.h>
360
361 #define assert_rcu_or_pool_mutex()                                      \
362         RCU_LOCKDEP_WARN(!rcu_read_lock_held() &&                       \
363                          !lockdep_is_held(&wq_pool_mutex),              \
364                          "RCU or wq_pool_mutex should be held")
365
366 #define assert_rcu_or_wq_mutex(wq)                                      \
367         RCU_LOCKDEP_WARN(!rcu_read_lock_held() &&                       \
368                          !lockdep_is_held(&wq->mutex),                  \
369                          "RCU or wq->mutex should be held")
370
371 #define assert_rcu_or_wq_mutex_or_pool_mutex(wq)                        \
372         RCU_LOCKDEP_WARN(!rcu_read_lock_held() &&                       \
373                          !lockdep_is_held(&wq->mutex) &&                \
374                          !lockdep_is_held(&wq_pool_mutex),              \
375                          "RCU, wq->mutex or wq_pool_mutex should be held")
376
377 #define for_each_cpu_worker_pool(pool, cpu)                             \
378         for ((pool) = &per_cpu(cpu_worker_pools, cpu)[0];               \
379              (pool) < &per_cpu(cpu_worker_pools, cpu)[NR_STD_WORKER_POOLS]; \
380              (pool)++)
381
382 /**
383  * for_each_pool - iterate through all worker_pools in the system
384  * @pool: iteration cursor
385  * @pi: integer used for iteration
386  *
387  * This must be called either with wq_pool_mutex held or RCU read
388  * locked.  If the pool needs to be used beyond the locking in effect, the
389  * caller is responsible for guaranteeing that the pool stays online.
390  *
391  * The if/else clause exists only for the lockdep assertion and can be
392  * ignored.
393  */
394 #define for_each_pool(pool, pi)                                         \
395         idr_for_each_entry(&worker_pool_idr, pool, pi)                  \
396                 if (({ assert_rcu_or_pool_mutex(); false; })) { }       \
397                 else
398
399 /**
400  * for_each_pool_worker - iterate through all workers of a worker_pool
401  * @worker: iteration cursor
402  * @pool: worker_pool to iterate workers of
403  *
404  * This must be called with wq_pool_attach_mutex.
405  *
406  * The if/else clause exists only for the lockdep assertion and can be
407  * ignored.
408  */
409 #define for_each_pool_worker(worker, pool)                              \
410         list_for_each_entry((worker), &(pool)->workers, node)           \
411                 if (({ lockdep_assert_held(&wq_pool_attach_mutex); false; })) { } \
412                 else
413
414 /**
415  * for_each_pwq - iterate through all pool_workqueues of the specified workqueue
416  * @pwq: iteration cursor
417  * @wq: the target workqueue
418  *
419  * This must be called either with wq->mutex held or RCU read locked.
420  * If the pwq needs to be used beyond the locking in effect, the caller is
421  * responsible for guaranteeing that the pwq stays online.
422  *
423  * The if/else clause exists only for the lockdep assertion and can be
424  * ignored.
425  */
426 #define for_each_pwq(pwq, wq)                                           \
427         list_for_each_entry_rcu((pwq), &(wq)->pwqs, pwqs_node)          \
428                 if (({ assert_rcu_or_wq_mutex(wq); false; })) { }       \
429                 else
430
431 #ifdef CONFIG_DEBUG_OBJECTS_WORK
432
433 static struct debug_obj_descr work_debug_descr;
434
435 static void *work_debug_hint(void *addr)
436 {
437         return ((struct work_struct *) addr)->func;
438 }
439
440 static bool work_is_static_object(void *addr)
441 {
442         struct work_struct *work = addr;
443
444         return test_bit(WORK_STRUCT_STATIC_BIT, work_data_bits(work));
445 }
446
447 /*
448  * fixup_init is called when:
449  * - an active object is initialized
450  */
451 static bool work_fixup_init(void *addr, enum debug_obj_state state)
452 {
453         struct work_struct *work = addr;
454
455         switch (state) {
456         case ODEBUG_STATE_ACTIVE:
457                 cancel_work_sync(work);
458                 debug_object_init(work, &work_debug_descr);
459                 return true;
460         default:
461                 return false;
462         }
463 }
464
465 /*
466  * fixup_free is called when:
467  * - an active object is freed
468  */
469 static bool work_fixup_free(void *addr, enum debug_obj_state state)
470 {
471         struct work_struct *work = addr;
472
473         switch (state) {
474         case ODEBUG_STATE_ACTIVE:
475                 cancel_work_sync(work);
476                 debug_object_free(work, &work_debug_descr);
477                 return true;
478         default:
479                 return false;
480         }
481 }
482
483 static struct debug_obj_descr work_debug_descr = {
484         .name           = "work_struct",
485         .debug_hint     = work_debug_hint,
486         .is_static_object = work_is_static_object,
487         .fixup_init     = work_fixup_init,
488         .fixup_free     = work_fixup_free,
489 };
490
491 static inline void debug_work_activate(struct work_struct *work)
492 {
493         debug_object_activate(work, &work_debug_descr);
494 }
495
496 static inline void debug_work_deactivate(struct work_struct *work)
497 {
498         debug_object_deactivate(work, &work_debug_descr);
499 }
500
501 void __init_work(struct work_struct *work, int onstack)
502 {
503         if (onstack)
504                 debug_object_init_on_stack(work, &work_debug_descr);
505         else
506                 debug_object_init(work, &work_debug_descr);
507 }
508 EXPORT_SYMBOL_GPL(__init_work);
509
510 void destroy_work_on_stack(struct work_struct *work)
511 {
512         debug_object_free(work, &work_debug_descr);
513 }
514 EXPORT_SYMBOL_GPL(destroy_work_on_stack);
515
516 void destroy_delayed_work_on_stack(struct delayed_work *work)
517 {
518         destroy_timer_on_stack(&work->timer);
519         debug_object_free(&work->work, &work_debug_descr);
520 }
521 EXPORT_SYMBOL_GPL(destroy_delayed_work_on_stack);
522
523 #else
524 static inline void debug_work_activate(struct work_struct *work) { }
525 static inline void debug_work_deactivate(struct work_struct *work) { }
526 #endif
527
528 /**
529  * worker_pool_assign_id - allocate ID and assing it to @pool
530  * @pool: the pool pointer of interest
531  *
532  * Returns 0 if ID in [0, WORK_OFFQ_POOL_NONE) is allocated and assigned
533  * successfully, -errno on failure.
534  */
535 static int worker_pool_assign_id(struct worker_pool *pool)
536 {
537         int ret;
538
539         lockdep_assert_held(&wq_pool_mutex);
540
541         ret = idr_alloc(&worker_pool_idr, pool, 0, WORK_OFFQ_POOL_NONE,
542                         GFP_KERNEL);
543         if (ret >= 0) {
544                 pool->id = ret;
545                 return 0;
546         }
547         return ret;
548 }
549
550 /**
551  * unbound_pwq_by_node - return the unbound pool_workqueue for the given node
552  * @wq: the target workqueue
553  * @node: the node ID
554  *
555  * This must be called with any of wq_pool_mutex, wq->mutex or RCU
556  * read locked.
557  * If the pwq needs to be used beyond the locking in effect, the caller is
558  * responsible for guaranteeing that the pwq stays online.
559  *
560  * Return: The unbound pool_workqueue for @node.
561  */
562 static struct pool_workqueue *unbound_pwq_by_node(struct workqueue_struct *wq,
563                                                   int node)
564 {
565         assert_rcu_or_wq_mutex_or_pool_mutex(wq);
566
567         /*
568          * XXX: @node can be NUMA_NO_NODE if CPU goes offline while a
569          * delayed item is pending.  The plan is to keep CPU -> NODE
570          * mapping valid and stable across CPU on/offlines.  Once that
571          * happens, this workaround can be removed.
572          */
573         if (unlikely(node == NUMA_NO_NODE))
574                 return wq->dfl_pwq;
575
576         return rcu_dereference_raw(wq->numa_pwq_tbl[node]);
577 }
578
579 static unsigned int work_color_to_flags(int color)
580 {
581         return color << WORK_STRUCT_COLOR_SHIFT;
582 }
583
584 static int get_work_color(struct work_struct *work)
585 {
586         return (*work_data_bits(work) >> WORK_STRUCT_COLOR_SHIFT) &
587                 ((1 << WORK_STRUCT_COLOR_BITS) - 1);
588 }
589
590 static int work_next_color(int color)
591 {
592         return (color + 1) % WORK_NR_COLORS;
593 }
594
595 /*
596  * While queued, %WORK_STRUCT_PWQ is set and non flag bits of a work's data
597  * contain the pointer to the queued pwq.  Once execution starts, the flag
598  * is cleared and the high bits contain OFFQ flags and pool ID.
599  *
600  * set_work_pwq(), set_work_pool_and_clear_pending(), mark_work_canceling()
601  * and clear_work_data() can be used to set the pwq, pool or clear
602  * work->data.  These functions should only be called while the work is
603  * owned - ie. while the PENDING bit is set.
604  *
605  * get_work_pool() and get_work_pwq() can be used to obtain the pool or pwq
606  * corresponding to a work.  Pool is available once the work has been
607  * queued anywhere after initialization until it is sync canceled.  pwq is
608  * available only while the work item is queued.
609  *
610  * %WORK_OFFQ_CANCELING is used to mark a work item which is being
611  * canceled.  While being canceled, a work item may have its PENDING set
612  * but stay off timer and worklist for arbitrarily long and nobody should
613  * try to steal the PENDING bit.
614  */
615 static inline void set_work_data(struct work_struct *work, unsigned long data,
616                                  unsigned long flags)
617 {
618         WARN_ON_ONCE(!work_pending(work));
619         atomic_long_set(&work->data, data | flags | work_static(work));
620 }
621
622 static void set_work_pwq(struct work_struct *work, struct pool_workqueue *pwq,
623                          unsigned long extra_flags)
624 {
625         set_work_data(work, (unsigned long)pwq,
626                       WORK_STRUCT_PENDING | WORK_STRUCT_PWQ | extra_flags);
627 }
628
629 static void set_work_pool_and_keep_pending(struct work_struct *work,
630                                            int pool_id)
631 {
632         set_work_data(work, (unsigned long)pool_id << WORK_OFFQ_POOL_SHIFT,
633                       WORK_STRUCT_PENDING);
634 }
635
636 static void set_work_pool_and_clear_pending(struct work_struct *work,
637                                             int pool_id)
638 {
639         /*
640          * The following wmb is paired with the implied mb in
641          * test_and_set_bit(PENDING) and ensures all updates to @work made
642          * here are visible to and precede any updates by the next PENDING
643          * owner.
644          */
645         smp_wmb();
646         set_work_data(work, (unsigned long)pool_id << WORK_OFFQ_POOL_SHIFT, 0);
647         /*
648          * The following mb guarantees that previous clear of a PENDING bit
649          * will not be reordered with any speculative LOADS or STORES from
650          * work->current_func, which is executed afterwards.  This possible
651          * reordering can lead to a missed execution on attempt to queue
652          * the same @work.  E.g. consider this case:
653          *
654          *   CPU#0                         CPU#1
655          *   ----------------------------  --------------------------------
656          *
657          * 1  STORE event_indicated
658          * 2  queue_work_on() {
659          * 3    test_and_set_bit(PENDING)
660          * 4 }                             set_..._and_clear_pending() {
661          * 5                                 set_work_data() # clear bit
662          * 6                                 smp_mb()
663          * 7                               work->current_func() {
664          * 8                                  LOAD event_indicated
665          *                                 }
666          *
667          * Without an explicit full barrier speculative LOAD on line 8 can
668          * be executed before CPU#0 does STORE on line 1.  If that happens,
669          * CPU#0 observes the PENDING bit is still set and new execution of
670          * a @work is not queued in a hope, that CPU#1 will eventually
671          * finish the queued @work.  Meanwhile CPU#1 does not see
672          * event_indicated is set, because speculative LOAD was executed
673          * before actual STORE.
674          */
675         smp_mb();
676 }
677
678 static void clear_work_data(struct work_struct *work)
679 {
680         smp_wmb();      /* see set_work_pool_and_clear_pending() */
681         set_work_data(work, WORK_STRUCT_NO_POOL, 0);
682 }
683
684 static struct pool_workqueue *get_work_pwq(struct work_struct *work)
685 {
686         unsigned long data = atomic_long_read(&work->data);
687
688         if (data & WORK_STRUCT_PWQ)
689                 return (void *)(data & WORK_STRUCT_WQ_DATA_MASK);
690         else
691                 return NULL;
692 }
693
694 /**
695  * get_work_pool - return the worker_pool a given work was associated with
696  * @work: the work item of interest
697  *
698  * Pools are created and destroyed under wq_pool_mutex, and allows read
699  * access under RCU read lock.  As such, this function should be
700  * called under wq_pool_mutex or inside of a rcu_read_lock() region.
701  *
702  * All fields of the returned pool are accessible as long as the above
703  * mentioned locking is in effect.  If the returned pool needs to be used
704  * beyond the critical section, the caller is responsible for ensuring the
705  * returned pool is and stays online.
706  *
707  * Return: The worker_pool @work was last associated with.  %NULL if none.
708  */
709 static struct worker_pool *get_work_pool(struct work_struct *work)
710 {
711         unsigned long data = atomic_long_read(&work->data);
712         int pool_id;
713
714         assert_rcu_or_pool_mutex();
715
716         if (data & WORK_STRUCT_PWQ)
717                 return ((struct pool_workqueue *)
718                         (data & WORK_STRUCT_WQ_DATA_MASK))->pool;
719
720         pool_id = data >> WORK_OFFQ_POOL_SHIFT;
721         if (pool_id == WORK_OFFQ_POOL_NONE)
722                 return NULL;
723
724         return idr_find(&worker_pool_idr, pool_id);
725 }
726
727 /**
728  * get_work_pool_id - return the worker pool ID a given work is associated with
729  * @work: the work item of interest
730  *
731  * Return: The worker_pool ID @work was last associated with.
732  * %WORK_OFFQ_POOL_NONE if none.
733  */
734 static int get_work_pool_id(struct work_struct *work)
735 {
736         unsigned long data = atomic_long_read(&work->data);
737
738         if (data & WORK_STRUCT_PWQ)
739                 return ((struct pool_workqueue *)
740                         (data & WORK_STRUCT_WQ_DATA_MASK))->pool->id;
741
742         return data >> WORK_OFFQ_POOL_SHIFT;
743 }
744
745 static void mark_work_canceling(struct work_struct *work)
746 {
747         unsigned long pool_id = get_work_pool_id(work);
748
749         pool_id <<= WORK_OFFQ_POOL_SHIFT;
750         set_work_data(work, pool_id | WORK_OFFQ_CANCELING, WORK_STRUCT_PENDING);
751 }
752
753 static bool work_is_canceling(struct work_struct *work)
754 {
755         unsigned long data = atomic_long_read(&work->data);
756
757         return !(data & WORK_STRUCT_PWQ) && (data & WORK_OFFQ_CANCELING);
758 }
759
760 /*
761  * Policy functions.  These define the policies on how the global worker
762  * pools are managed.  Unless noted otherwise, these functions assume that
763  * they're being called with pool->lock held.
764  */
765
766 static bool __need_more_worker(struct worker_pool *pool)
767 {
768         return !atomic_read(&pool->nr_running);
769 }
770
771 /*
772  * Need to wake up a worker?  Called from anything but currently
773  * running workers.
774  *
775  * Note that, because unbound workers never contribute to nr_running, this
776  * function will always return %true for unbound pools as long as the
777  * worklist isn't empty.
778  */
779 static bool need_more_worker(struct worker_pool *pool)
780 {
781         return !list_empty(&pool->worklist) && __need_more_worker(pool);
782 }
783
784 /* Can I start working?  Called from busy but !running workers. */
785 static bool may_start_working(struct worker_pool *pool)
786 {
787         return pool->nr_idle;
788 }
789
790 /* Do I need to keep working?  Called from currently running workers. */
791 static bool keep_working(struct worker_pool *pool)
792 {
793         return !list_empty(&pool->worklist) &&
794                 atomic_read(&pool->nr_running) <= 1;
795 }
796
797 /* Do we need a new worker?  Called from manager. */
798 static bool need_to_create_worker(struct worker_pool *pool)
799 {
800         return need_more_worker(pool) && !may_start_working(pool);
801 }
802
803 /* Do we have too many workers and should some go away? */
804 static bool too_many_workers(struct worker_pool *pool)
805 {
806         bool managing = pool->flags & POOL_MANAGER_ACTIVE;
807         int nr_idle = pool->nr_idle + managing; /* manager is considered idle */
808         int nr_busy = pool->nr_workers - nr_idle;
809
810         return nr_idle > 2 && (nr_idle - 2) * MAX_IDLE_WORKERS_RATIO >= nr_busy;
811 }
812
813 /*
814  * Wake up functions.
815  */
816
817 /* Return the first idle worker.  Safe with preemption disabled */
818 static struct worker *first_idle_worker(struct worker_pool *pool)
819 {
820         if (unlikely(list_empty(&pool->idle_list)))
821                 return NULL;
822
823         return list_first_entry(&pool->idle_list, struct worker, entry);
824 }
825
826 /**
827  * wake_up_worker - wake up an idle worker
828  * @pool: worker pool to wake worker from
829  *
830  * Wake up the first idle worker of @pool.
831  *
832  * CONTEXT:
833  * spin_lock_irq(pool->lock).
834  */
835 static void wake_up_worker(struct worker_pool *pool)
836 {
837         struct worker *worker = first_idle_worker(pool);
838
839         if (likely(worker))
840                 wake_up_process(worker->task);
841 }
842
843 /**
844  * wq_worker_running - a worker is running again
845  * @task: task waking up
846  *
847  * This function is called when a worker returns from schedule()
848  */
849 void wq_worker_running(struct task_struct *task)
850 {
851         struct worker *worker = kthread_data(task);
852
853         if (!worker->sleeping)
854                 return;
855         if (!(worker->flags & WORKER_NOT_RUNNING))
856                 atomic_inc(&worker->pool->nr_running);
857         worker->sleeping = 0;
858 }
859
860 /**
861  * wq_worker_sleeping - a worker is going to sleep
862  * @task: task going to sleep
863  *
864  * This function is called from schedule() when a busy worker is
865  * going to sleep.
866  */
867 void wq_worker_sleeping(struct task_struct *task)
868 {
869         struct worker *next, *worker = kthread_data(task);
870         struct worker_pool *pool;
871
872         /*
873          * Rescuers, which may not have all the fields set up like normal
874          * workers, also reach here, let's not access anything before
875          * checking NOT_RUNNING.
876          */
877         if (worker->flags & WORKER_NOT_RUNNING)
878                 return;
879
880         pool = worker->pool;
881
882         if (WARN_ON_ONCE(worker->sleeping))
883                 return;
884
885         worker->sleeping = 1;
886         spin_lock_irq(&pool->lock);
887
888         /*
889          * The counterpart of the following dec_and_test, implied mb,
890          * worklist not empty test sequence is in insert_work().
891          * Please read comment there.
892          *
893          * NOT_RUNNING is clear.  This means that we're bound to and
894          * running on the local cpu w/ rq lock held and preemption
895          * disabled, which in turn means that none else could be
896          * manipulating idle_list, so dereferencing idle_list without pool
897          * lock is safe.
898          */
899         if (atomic_dec_and_test(&pool->nr_running) &&
900             !list_empty(&pool->worklist)) {
901                 next = first_idle_worker(pool);
902                 if (next)
903                         wake_up_process(next->task);
904         }
905         spin_unlock_irq(&pool->lock);
906 }
907
908 /**
909  * wq_worker_last_func - retrieve worker's last work function
910  * @task: Task to retrieve last work function of.
911  *
912  * Determine the last function a worker executed. This is called from
913  * the scheduler to get a worker's last known identity.
914  *
915  * CONTEXT:
916  * spin_lock_irq(rq->lock)
917  *
918  * This function is called during schedule() when a kworker is going
919  * to sleep. It's used by psi to identify aggregation workers during
920  * dequeuing, to allow periodic aggregation to shut-off when that
921  * worker is the last task in the system or cgroup to go to sleep.
922  *
923  * As this function doesn't involve any workqueue-related locking, it
924  * only returns stable values when called from inside the scheduler's
925  * queuing and dequeuing paths, when @task, which must be a kworker,
926  * is guaranteed to not be processing any works.
927  *
928  * Return:
929  * The last work function %current executed as a worker, NULL if it
930  * hasn't executed any work yet.
931  */
932 work_func_t wq_worker_last_func(struct task_struct *task)
933 {
934         struct worker *worker = kthread_data(task);
935
936         return worker->last_func;
937 }
938
939 /**
940  * worker_set_flags - set worker flags and adjust nr_running accordingly
941  * @worker: self
942  * @flags: flags to set
943  *
944  * Set @flags in @worker->flags and adjust nr_running accordingly.
945  *
946  * CONTEXT:
947  * spin_lock_irq(pool->lock)
948  */
949 static inline void worker_set_flags(struct worker *worker, unsigned int flags)
950 {
951         struct worker_pool *pool = worker->pool;
952
953         WARN_ON_ONCE(worker->task != current);
954
955         /* If transitioning into NOT_RUNNING, adjust nr_running. */
956         if ((flags & WORKER_NOT_RUNNING) &&
957             !(worker->flags & WORKER_NOT_RUNNING)) {
958                 atomic_dec(&pool->nr_running);
959         }
960
961         worker->flags |= flags;
962 }
963
964 /**
965  * worker_clr_flags - clear worker flags and adjust nr_running accordingly
966  * @worker: self
967  * @flags: flags to clear
968  *
969  * Clear @flags in @worker->flags and adjust nr_running accordingly.
970  *
971  * CONTEXT:
972  * spin_lock_irq(pool->lock)
973  */
974 static inline void worker_clr_flags(struct worker *worker, unsigned int flags)
975 {
976         struct worker_pool *pool = worker->pool;
977         unsigned int oflags = worker->flags;
978
979         WARN_ON_ONCE(worker->task != current);
980
981         worker->flags &= ~flags;
982
983         /*
984          * If transitioning out of NOT_RUNNING, increment nr_running.  Note
985          * that the nested NOT_RUNNING is not a noop.  NOT_RUNNING is mask
986          * of multiple flags, not a single flag.
987          */
988         if ((flags & WORKER_NOT_RUNNING) && (oflags & WORKER_NOT_RUNNING))
989                 if (!(worker->flags & WORKER_NOT_RUNNING))
990                         atomic_inc(&pool->nr_running);
991 }
992
993 /**
994  * find_worker_executing_work - find worker which is executing a work
995  * @pool: pool of interest
996  * @work: work to find worker for
997  *
998  * Find a worker which is executing @work on @pool by searching
999  * @pool->busy_hash which is keyed by the address of @work.  For a worker
1000  * to match, its current execution should match the address of @work and
1001  * its work function.  This is to avoid unwanted dependency between
1002  * unrelated work executions through a work item being recycled while still
1003  * being executed.
1004  *
1005  * This is a bit tricky.  A work item may be freed once its execution
1006  * starts and nothing prevents the freed area from being recycled for
1007  * another work item.  If the same work item address ends up being reused
1008  * before the original execution finishes, workqueue will identify the
1009  * recycled work item as currently executing and make it wait until the
1010  * current execution finishes, introducing an unwanted dependency.
1011  *
1012  * This function checks the work item address and work function to avoid
1013  * false positives.  Note that this isn't complete as one may construct a
1014  * work function which can introduce dependency onto itself through a
1015  * recycled work item.  Well, if somebody wants to shoot oneself in the
1016  * foot that badly, there's only so much we can do, and if such deadlock
1017  * actually occurs, it should be easy to locate the culprit work function.
1018  *
1019  * CONTEXT:
1020  * spin_lock_irq(pool->lock).
1021  *
1022  * Return:
1023  * Pointer to worker which is executing @work if found, %NULL
1024  * otherwise.
1025  */
1026 static struct worker *find_worker_executing_work(struct worker_pool *pool,
1027                                                  struct work_struct *work)
1028 {
1029         struct worker *worker;
1030
1031         hash_for_each_possible(pool->busy_hash, worker, hentry,
1032                                (unsigned long)work)
1033                 if (worker->current_work == work &&
1034                     worker->current_func == work->func)
1035                         return worker;
1036
1037         return NULL;
1038 }
1039
1040 /**
1041  * move_linked_works - move linked works to a list
1042  * @work: start of series of works to be scheduled
1043  * @head: target list to append @work to
1044  * @nextp: out parameter for nested worklist walking
1045  *
1046  * Schedule linked works starting from @work to @head.  Work series to
1047  * be scheduled starts at @work and includes any consecutive work with
1048  * WORK_STRUCT_LINKED set in its predecessor.
1049  *
1050  * If @nextp is not NULL, it's updated to point to the next work of
1051  * the last scheduled work.  This allows move_linked_works() to be
1052  * nested inside outer list_for_each_entry_safe().
1053  *
1054  * CONTEXT:
1055  * spin_lock_irq(pool->lock).
1056  */
1057 static void move_linked_works(struct work_struct *work, struct list_head *head,
1058                               struct work_struct **nextp)
1059 {
1060         struct work_struct *n;
1061
1062         /*
1063          * Linked worklist will always end before the end of the list,
1064          * use NULL for list head.
1065          */
1066         list_for_each_entry_safe_from(work, n, NULL, entry) {
1067                 list_move_tail(&work->entry, head);
1068                 if (!(*work_data_bits(work) & WORK_STRUCT_LINKED))
1069                         break;
1070         }
1071
1072         /*
1073          * If we're already inside safe list traversal and have moved
1074          * multiple works to the scheduled queue, the next position
1075          * needs to be updated.
1076          */
1077         if (nextp)
1078                 *nextp = n;
1079 }
1080
1081 /**
1082  * get_pwq - get an extra reference on the specified pool_workqueue
1083  * @pwq: pool_workqueue to get
1084  *
1085  * Obtain an extra reference on @pwq.  The caller should guarantee that
1086  * @pwq has positive refcnt and be holding the matching pool->lock.
1087  */
1088 static void get_pwq(struct pool_workqueue *pwq)
1089 {
1090         lockdep_assert_held(&pwq->pool->lock);
1091         WARN_ON_ONCE(pwq->refcnt <= 0);
1092         pwq->refcnt++;
1093 }
1094
1095 /**
1096  * put_pwq - put a pool_workqueue reference
1097  * @pwq: pool_workqueue to put
1098  *
1099  * Drop a reference of @pwq.  If its refcnt reaches zero, schedule its
1100  * destruction.  The caller should be holding the matching pool->lock.
1101  */
1102 static void put_pwq(struct pool_workqueue *pwq)
1103 {
1104         lockdep_assert_held(&pwq->pool->lock);
1105         if (likely(--pwq->refcnt))
1106                 return;
1107         if (WARN_ON_ONCE(!(pwq->wq->flags & WQ_UNBOUND)))
1108                 return;
1109         /*
1110          * @pwq can't be released under pool->lock, bounce to
1111          * pwq_unbound_release_workfn().  This never recurses on the same
1112          * pool->lock as this path is taken only for unbound workqueues and
1113          * the release work item is scheduled on a per-cpu workqueue.  To
1114          * avoid lockdep warning, unbound pool->locks are given lockdep
1115          * subclass of 1 in get_unbound_pool().
1116          */
1117         schedule_work(&pwq->unbound_release_work);
1118 }
1119
1120 /**
1121  * put_pwq_unlocked - put_pwq() with surrounding pool lock/unlock
1122  * @pwq: pool_workqueue to put (can be %NULL)
1123  *
1124  * put_pwq() with locking.  This function also allows %NULL @pwq.
1125  */
1126 static void put_pwq_unlocked(struct pool_workqueue *pwq)
1127 {
1128         if (pwq) {
1129                 /*
1130                  * As both pwqs and pools are RCU protected, the
1131                  * following lock operations are safe.
1132                  */
1133                 spin_lock_irq(&pwq->pool->lock);
1134                 put_pwq(pwq);
1135                 spin_unlock_irq(&pwq->pool->lock);
1136         }
1137 }
1138
1139 static void pwq_activate_delayed_work(struct work_struct *work)
1140 {
1141         struct pool_workqueue *pwq = get_work_pwq(work);
1142
1143         trace_workqueue_activate_work(work);
1144         if (list_empty(&pwq->pool->worklist))
1145                 pwq->pool->watchdog_ts = jiffies;
1146         move_linked_works(work, &pwq->pool->worklist, NULL);
1147         __clear_bit(WORK_STRUCT_DELAYED_BIT, work_data_bits(work));
1148         pwq->nr_active++;
1149 }
1150
1151 static void pwq_activate_first_delayed(struct pool_workqueue *pwq)
1152 {
1153         struct work_struct *work = list_first_entry(&pwq->delayed_works,
1154                                                     struct work_struct, entry);
1155
1156         pwq_activate_delayed_work(work);
1157 }
1158
1159 /**
1160  * pwq_dec_nr_in_flight - decrement pwq's nr_in_flight
1161  * @pwq: pwq of interest
1162  * @color: color of work which left the queue
1163  *
1164  * A work either has completed or is removed from pending queue,
1165  * decrement nr_in_flight of its pwq and handle workqueue flushing.
1166  *
1167  * CONTEXT:
1168  * spin_lock_irq(pool->lock).
1169  */
1170 static void pwq_dec_nr_in_flight(struct pool_workqueue *pwq, int color)
1171 {
1172         /* uncolored work items don't participate in flushing or nr_active */
1173         if (color == WORK_NO_COLOR)
1174                 goto out_put;
1175
1176         pwq->nr_in_flight[color]--;
1177
1178         pwq->nr_active--;
1179         if (!list_empty(&pwq->delayed_works)) {
1180                 /* one down, submit a delayed one */
1181                 if (pwq->nr_active < pwq->max_active)
1182                         pwq_activate_first_delayed(pwq);
1183         }
1184
1185         /* is flush in progress and are we at the flushing tip? */
1186         if (likely(pwq->flush_color != color))
1187                 goto out_put;
1188
1189         /* are there still in-flight works? */
1190         if (pwq->nr_in_flight[color])
1191                 goto out_put;
1192
1193         /* this pwq is done, clear flush_color */
1194         pwq->flush_color = -1;
1195
1196         /*
1197          * If this was the last pwq, wake up the first flusher.  It
1198          * will handle the rest.
1199          */
1200         if (atomic_dec_and_test(&pwq->wq->nr_pwqs_to_flush))
1201                 complete(&pwq->wq->first_flusher->done);
1202 out_put:
1203         put_pwq(pwq);
1204 }
1205
1206 /**
1207  * try_to_grab_pending - steal work item from worklist and disable irq
1208  * @work: work item to steal
1209  * @is_dwork: @work is a delayed_work
1210  * @flags: place to store irq state
1211  *
1212  * Try to grab PENDING bit of @work.  This function can handle @work in any
1213  * stable state - idle, on timer or on worklist.
1214  *
1215  * Return:
1216  *  1           if @work was pending and we successfully stole PENDING
1217  *  0           if @work was idle and we claimed PENDING
1218  *  -EAGAIN     if PENDING couldn't be grabbed at the moment, safe to busy-retry
1219  *  -ENOENT     if someone else is canceling @work, this state may persist
1220  *              for arbitrarily long
1221  *
1222  * Note:
1223  * On >= 0 return, the caller owns @work's PENDING bit.  To avoid getting
1224  * interrupted while holding PENDING and @work off queue, irq must be
1225  * disabled on entry.  This, combined with delayed_work->timer being
1226  * irqsafe, ensures that we return -EAGAIN for finite short period of time.
1227  *
1228  * On successful return, >= 0, irq is disabled and the caller is
1229  * responsible for releasing it using local_irq_restore(*@flags).
1230  *
1231  * This function is safe to call from any context including IRQ handler.
1232  */
1233 static int try_to_grab_pending(struct work_struct *work, bool is_dwork,
1234                                unsigned long *flags)
1235 {
1236         struct worker_pool *pool;
1237         struct pool_workqueue *pwq;
1238
1239         local_irq_save(*flags);
1240
1241         /* try to steal the timer if it exists */
1242         if (is_dwork) {
1243                 struct delayed_work *dwork = to_delayed_work(work);
1244
1245                 /*
1246                  * dwork->timer is irqsafe.  If del_timer() fails, it's
1247                  * guaranteed that the timer is not queued anywhere and not
1248                  * running on the local CPU.
1249                  */
1250                 if (likely(del_timer(&dwork->timer)))
1251                         return 1;
1252         }
1253
1254         /* try to claim PENDING the normal way */
1255         if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work)))
1256                 return 0;
1257
1258         rcu_read_lock();
1259         /*
1260          * The queueing is in progress, or it is already queued. Try to
1261          * steal it from ->worklist without clearing WORK_STRUCT_PENDING.
1262          */
1263         pool = get_work_pool(work);
1264         if (!pool)
1265                 goto fail;
1266
1267         spin_lock(&pool->lock);
1268         /*
1269          * work->data is guaranteed to point to pwq only while the work
1270          * item is queued on pwq->wq, and both updating work->data to point
1271          * to pwq on queueing and to pool on dequeueing are done under
1272          * pwq->pool->lock.  This in turn guarantees that, if work->data
1273          * points to pwq which is associated with a locked pool, the work
1274          * item is currently queued on that pool.
1275          */
1276         pwq = get_work_pwq(work);
1277         if (pwq && pwq->pool == pool) {
1278                 debug_work_deactivate(work);
1279
1280                 /*
1281                  * A delayed work item cannot be grabbed directly because
1282                  * it might have linked NO_COLOR work items which, if left
1283                  * on the delayed_list, will confuse pwq->nr_active
1284                  * management later on and cause stall.  Make sure the work
1285                  * item is activated before grabbing.
1286                  */
1287                 if (*work_data_bits(work) & WORK_STRUCT_DELAYED)
1288                         pwq_activate_delayed_work(work);
1289
1290                 list_del_init(&work->entry);
1291                 pwq_dec_nr_in_flight(pwq, get_work_color(work));
1292
1293                 /* work->data points to pwq iff queued, point to pool */
1294                 set_work_pool_and_keep_pending(work, pool->id);
1295
1296                 spin_unlock(&pool->lock);
1297                 rcu_read_unlock();
1298                 return 1;
1299         }
1300         spin_unlock(&pool->lock);
1301 fail:
1302         rcu_read_unlock();
1303         local_irq_restore(*flags);
1304         if (work_is_canceling(work))
1305                 return -ENOENT;
1306         cpu_relax();
1307         return -EAGAIN;
1308 }
1309
1310 /**
1311  * insert_work - insert a work into a pool
1312  * @pwq: pwq @work belongs to
1313  * @work: work to insert
1314  * @head: insertion point
1315  * @extra_flags: extra WORK_STRUCT_* flags to set
1316  *
1317  * Insert @work which belongs to @pwq after @head.  @extra_flags is or'd to
1318  * work_struct flags.
1319  *
1320  * CONTEXT:
1321  * spin_lock_irq(pool->lock).
1322  */
1323 static void insert_work(struct pool_workqueue *pwq, struct work_struct *work,
1324                         struct list_head *head, unsigned int extra_flags)
1325 {
1326         struct worker_pool *pool = pwq->pool;
1327
1328         /* we own @work, set data and link */
1329         set_work_pwq(work, pwq, extra_flags);
1330         list_add_tail(&work->entry, head);
1331         get_pwq(pwq);
1332
1333         /*
1334          * Ensure either wq_worker_sleeping() sees the above
1335          * list_add_tail() or we see zero nr_running to avoid workers lying
1336          * around lazily while there are works to be processed.
1337          */
1338         smp_mb();
1339
1340         if (__need_more_worker(pool))
1341                 wake_up_worker(pool);
1342 }
1343
1344 /*
1345  * Test whether @work is being queued from another work executing on the
1346  * same workqueue.
1347  */
1348 static bool is_chained_work(struct workqueue_struct *wq)
1349 {
1350         struct worker *worker;
1351
1352         worker = current_wq_worker();
1353         /*
1354          * Return %true iff I'm a worker executing a work item on @wq.  If
1355          * I'm @worker, it's safe to dereference it without locking.
1356          */
1357         return worker && worker->current_pwq->wq == wq;
1358 }
1359
1360 /*
1361  * When queueing an unbound work item to a wq, prefer local CPU if allowed
1362  * by wq_unbound_cpumask.  Otherwise, round robin among the allowed ones to
1363  * avoid perturbing sensitive tasks.
1364  */
1365 static int wq_select_unbound_cpu(int cpu)
1366 {
1367         static bool printed_dbg_warning;
1368         int new_cpu;
1369
1370         if (likely(!wq_debug_force_rr_cpu)) {
1371                 if (cpumask_test_cpu(cpu, wq_unbound_cpumask))
1372                         return cpu;
1373         } else if (!printed_dbg_warning) {
1374                 pr_warn("workqueue: round-robin CPU selection forced, expect performance impact\n");
1375                 printed_dbg_warning = true;
1376         }
1377
1378         if (cpumask_empty(wq_unbound_cpumask))
1379                 return cpu;
1380
1381         new_cpu = __this_cpu_read(wq_rr_cpu_last);
1382         new_cpu = cpumask_next_and(new_cpu, wq_unbound_cpumask, cpu_online_mask);
1383         if (unlikely(new_cpu >= nr_cpu_ids)) {
1384                 new_cpu = cpumask_first_and(wq_unbound_cpumask, cpu_online_mask);
1385                 if (unlikely(new_cpu >= nr_cpu_ids))
1386                         return cpu;
1387         }
1388         __this_cpu_write(wq_rr_cpu_last, new_cpu);
1389
1390         return new_cpu;
1391 }
1392
1393 static void __queue_work(int cpu, struct workqueue_struct *wq,
1394                          struct work_struct *work)
1395 {
1396         struct pool_workqueue *pwq;
1397         struct worker_pool *last_pool;
1398         struct list_head *worklist;
1399         unsigned int work_flags;
1400         unsigned int req_cpu = cpu;
1401
1402         /*
1403          * While a work item is PENDING && off queue, a task trying to
1404          * steal the PENDING will busy-loop waiting for it to either get
1405          * queued or lose PENDING.  Grabbing PENDING and queueing should
1406          * happen with IRQ disabled.
1407          */
1408         lockdep_assert_irqs_disabled();
1409
1410         debug_work_activate(work);
1411
1412         /* if draining, only works from the same workqueue are allowed */
1413         if (unlikely(wq->flags & __WQ_DRAINING) &&
1414             WARN_ON_ONCE(!is_chained_work(wq)))
1415                 return;
1416         rcu_read_lock();
1417 retry:
1418         if (req_cpu == WORK_CPU_UNBOUND)
1419                 cpu = wq_select_unbound_cpu(raw_smp_processor_id());
1420
1421         /* pwq which will be used unless @work is executing elsewhere */
1422         if (!(wq->flags & WQ_UNBOUND))
1423                 pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
1424         else
1425                 pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
1426
1427         /*
1428          * If @work was previously on a different pool, it might still be
1429          * running there, in which case the work needs to be queued on that
1430          * pool to guarantee non-reentrancy.
1431          */
1432         last_pool = get_work_pool(work);
1433         if (last_pool && last_pool != pwq->pool) {
1434                 struct worker *worker;
1435
1436                 spin_lock(&last_pool->lock);
1437
1438                 worker = find_worker_executing_work(last_pool, work);
1439
1440                 if (worker && worker->current_pwq->wq == wq) {
1441                         pwq = worker->current_pwq;
1442                 } else {
1443                         /* meh... not running there, queue here */
1444                         spin_unlock(&last_pool->lock);
1445                         spin_lock(&pwq->pool->lock);
1446                 }
1447         } else {
1448                 spin_lock(&pwq->pool->lock);
1449         }
1450
1451         /*
1452          * pwq is determined and locked.  For unbound pools, we could have
1453          * raced with pwq release and it could already be dead.  If its
1454          * refcnt is zero, repeat pwq selection.  Note that pwqs never die
1455          * without another pwq replacing it in the numa_pwq_tbl or while
1456          * work items are executing on it, so the retrying is guaranteed to
1457          * make forward-progress.
1458          */
1459         if (unlikely(!pwq->refcnt)) {
1460                 if (wq->flags & WQ_UNBOUND) {
1461                         spin_unlock(&pwq->pool->lock);
1462                         cpu_relax();
1463                         goto retry;
1464                 }
1465                 /* oops */
1466                 WARN_ONCE(true, "workqueue: per-cpu pwq for %s on cpu%d has 0 refcnt",
1467                           wq->name, cpu);
1468         }
1469
1470         /* pwq determined, queue */
1471         trace_workqueue_queue_work(req_cpu, pwq, work);
1472
1473         if (WARN_ON(!list_empty(&work->entry)))
1474                 goto out;
1475
1476         pwq->nr_in_flight[pwq->work_color]++;
1477         work_flags = work_color_to_flags(pwq->work_color);
1478
1479         if (likely(pwq->nr_active < pwq->max_active)) {
1480                 trace_workqueue_activate_work(work);
1481                 pwq->nr_active++;
1482                 worklist = &pwq->pool->worklist;
1483                 if (list_empty(worklist))
1484                         pwq->pool->watchdog_ts = jiffies;
1485         } else {
1486                 work_flags |= WORK_STRUCT_DELAYED;
1487                 worklist = &pwq->delayed_works;
1488         }
1489
1490         insert_work(pwq, work, worklist, work_flags);
1491
1492 out:
1493         spin_unlock(&pwq->pool->lock);
1494         rcu_read_unlock();
1495 }
1496
1497 /**
1498  * queue_work_on - queue work on specific cpu
1499  * @cpu: CPU number to execute work on
1500  * @wq: workqueue to use
1501  * @work: work to queue
1502  *
1503  * We queue the work to a specific CPU, the caller must ensure it
1504  * can't go away.
1505  *
1506  * Return: %false if @work was already on a queue, %true otherwise.
1507  */
1508 bool queue_work_on(int cpu, struct workqueue_struct *wq,
1509                    struct work_struct *work)
1510 {
1511         bool ret = false;
1512         unsigned long flags;
1513
1514         local_irq_save(flags);
1515
1516         if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
1517                 __queue_work(cpu, wq, work);
1518                 ret = true;
1519         }
1520
1521         local_irq_restore(flags);
1522         return ret;
1523 }
1524 EXPORT_SYMBOL(queue_work_on);
1525
1526 /**
1527  * workqueue_select_cpu_near - Select a CPU based on NUMA node
1528  * @node: NUMA node ID that we want to select a CPU from
1529  *
1530  * This function will attempt to find a "random" cpu available on a given
1531  * node. If there are no CPUs available on the given node it will return
1532  * WORK_CPU_UNBOUND indicating that we should just schedule to any
1533  * available CPU if we need to schedule this work.
1534  */
1535 static int workqueue_select_cpu_near(int node)
1536 {
1537         int cpu;
1538
1539         /* No point in doing this if NUMA isn't enabled for workqueues */
1540         if (!wq_numa_enabled)
1541                 return WORK_CPU_UNBOUND;
1542
1543         /* Delay binding to CPU if node is not valid or online */
1544         if (node < 0 || node >= MAX_NUMNODES || !node_online(node))
1545                 return WORK_CPU_UNBOUND;
1546
1547         /* Use local node/cpu if we are already there */
1548         cpu = raw_smp_processor_id();
1549         if (node == cpu_to_node(cpu))
1550                 return cpu;
1551
1552         /* Use "random" otherwise know as "first" online CPU of node */
1553         cpu = cpumask_any_and(cpumask_of_node(node), cpu_online_mask);
1554
1555         /* If CPU is valid return that, otherwise just defer */
1556         return cpu < nr_cpu_ids ? cpu : WORK_CPU_UNBOUND;
1557 }
1558
1559 /**
1560  * queue_work_node - queue work on a "random" cpu for a given NUMA node
1561  * @node: NUMA node that we are targeting the work for
1562  * @wq: workqueue to use
1563  * @work: work to queue
1564  *
1565  * We queue the work to a "random" CPU within a given NUMA node. The basic
1566  * idea here is to provide a way to somehow associate work with a given
1567  * NUMA node.
1568  *
1569  * This function will only make a best effort attempt at getting this onto
1570  * the right NUMA node. If no node is requested or the requested node is
1571  * offline then we just fall back to standard queue_work behavior.
1572  *
1573  * Currently the "random" CPU ends up being the first available CPU in the
1574  * intersection of cpu_online_mask and the cpumask of the node, unless we
1575  * are running on the node. In that case we just use the current CPU.
1576  *
1577  * Return: %false if @work was already on a queue, %true otherwise.
1578  */
1579 bool queue_work_node(int node, struct workqueue_struct *wq,
1580                      struct work_struct *work)
1581 {
1582         unsigned long flags;
1583         bool ret = false;
1584
1585         /*
1586          * This current implementation is specific to unbound workqueues.
1587          * Specifically we only return the first available CPU for a given
1588          * node instead of cycling through individual CPUs within the node.
1589          *
1590          * If this is used with a per-cpu workqueue then the logic in
1591          * workqueue_select_cpu_near would need to be updated to allow for
1592          * some round robin type logic.
1593          */
1594         WARN_ON_ONCE(!(wq->flags & WQ_UNBOUND));
1595
1596         local_irq_save(flags);
1597
1598         if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
1599                 int cpu = workqueue_select_cpu_near(node);
1600
1601                 __queue_work(cpu, wq, work);
1602                 ret = true;
1603         }
1604
1605         local_irq_restore(flags);
1606         return ret;
1607 }
1608 EXPORT_SYMBOL_GPL(queue_work_node);
1609
1610 void delayed_work_timer_fn(struct timer_list *t)
1611 {
1612         struct delayed_work *dwork = from_timer(dwork, t, timer);
1613
1614         /* should have been called from irqsafe timer with irq already off */
1615         __queue_work(dwork->cpu, dwork->wq, &dwork->work);
1616 }
1617 EXPORT_SYMBOL(delayed_work_timer_fn);
1618
1619 static void __queue_delayed_work(int cpu, struct workqueue_struct *wq,
1620                                 struct delayed_work *dwork, unsigned long delay)
1621 {
1622         struct timer_list *timer = &dwork->timer;
1623         struct work_struct *work = &dwork->work;
1624
1625         WARN_ON_ONCE(!wq);
1626         WARN_ON_ONCE(timer->function != delayed_work_timer_fn);
1627         WARN_ON_ONCE(timer_pending(timer));
1628         WARN_ON_ONCE(!list_empty(&work->entry));
1629
1630         /*
1631          * If @delay is 0, queue @dwork->work immediately.  This is for
1632          * both optimization and correctness.  The earliest @timer can
1633          * expire is on the closest next tick and delayed_work users depend
1634          * on that there's no such delay when @delay is 0.
1635          */
1636         if (!delay) {
1637                 __queue_work(cpu, wq, &dwork->work);
1638                 return;
1639         }
1640
1641         dwork->wq = wq;
1642         dwork->cpu = cpu;
1643         timer->expires = jiffies + delay;
1644
1645         if (unlikely(cpu != WORK_CPU_UNBOUND))
1646                 add_timer_on(timer, cpu);
1647         else
1648                 add_timer(timer);
1649 }
1650
1651 /**
1652  * queue_delayed_work_on - queue work on specific CPU after delay
1653  * @cpu: CPU number to execute work on
1654  * @wq: workqueue to use
1655  * @dwork: work to queue
1656  * @delay: number of jiffies to wait before queueing
1657  *
1658  * Return: %false if @work was already on a queue, %true otherwise.  If
1659  * @delay is zero and @dwork is idle, it will be scheduled for immediate
1660  * execution.
1661  */
1662 bool queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
1663                            struct delayed_work *dwork, unsigned long delay)
1664 {
1665         struct work_struct *work = &dwork->work;
1666         bool ret = false;
1667         unsigned long flags;
1668
1669         /* read the comment in __queue_work() */
1670         local_irq_save(flags);
1671
1672         if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
1673                 __queue_delayed_work(cpu, wq, dwork, delay);
1674                 ret = true;
1675         }
1676
1677         local_irq_restore(flags);
1678         return ret;
1679 }
1680 EXPORT_SYMBOL(queue_delayed_work_on);
1681
1682 /**
1683  * mod_delayed_work_on - modify delay of or queue a delayed work on specific CPU
1684  * @cpu: CPU number to execute work on
1685  * @wq: workqueue to use
1686  * @dwork: work to queue
1687  * @delay: number of jiffies to wait before queueing
1688  *
1689  * If @dwork is idle, equivalent to queue_delayed_work_on(); otherwise,
1690  * modify @dwork's timer so that it expires after @delay.  If @delay is
1691  * zero, @work is guaranteed to be scheduled immediately regardless of its
1692  * current state.
1693  *
1694  * Return: %false if @dwork was idle and queued, %true if @dwork was
1695  * pending and its timer was modified.
1696  *
1697  * This function is safe to call from any context including IRQ handler.
1698  * See try_to_grab_pending() for details.
1699  */
1700 bool mod_delayed_work_on(int cpu, struct workqueue_struct *wq,
1701                          struct delayed_work *dwork, unsigned long delay)
1702 {
1703         unsigned long flags;
1704         int ret;
1705
1706         do {
1707                 ret = try_to_grab_pending(&dwork->work, true, &flags);
1708         } while (unlikely(ret == -EAGAIN));
1709
1710         if (likely(ret >= 0)) {
1711                 __queue_delayed_work(cpu, wq, dwork, delay);
1712                 local_irq_restore(flags);
1713         }
1714
1715         /* -ENOENT from try_to_grab_pending() becomes %true */
1716         return ret;
1717 }
1718 EXPORT_SYMBOL_GPL(mod_delayed_work_on);
1719
1720 static void rcu_work_rcufn(struct rcu_head *rcu)
1721 {
1722         struct rcu_work *rwork = container_of(rcu, struct rcu_work, rcu);
1723
1724         /* read the comment in __queue_work() */
1725         local_irq_disable();
1726         __queue_work(WORK_CPU_UNBOUND, rwork->wq, &rwork->work);
1727         local_irq_enable();
1728 }
1729
1730 /**
1731  * queue_rcu_work - queue work after a RCU grace period
1732  * @wq: workqueue to use
1733  * @rwork: work to queue
1734  *
1735  * Return: %false if @rwork was already pending, %true otherwise.  Note
1736  * that a full RCU grace period is guaranteed only after a %true return.
1737  * While @rwork is guaranteed to be executed after a %false return, the
1738  * execution may happen before a full RCU grace period has passed.
1739  */
1740 bool queue_rcu_work(struct workqueue_struct *wq, struct rcu_work *rwork)
1741 {
1742         struct work_struct *work = &rwork->work;
1743
1744         if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
1745                 rwork->wq = wq;
1746                 call_rcu(&rwork->rcu, rcu_work_rcufn);
1747                 return true;
1748         }
1749
1750         return false;
1751 }
1752 EXPORT_SYMBOL(queue_rcu_work);
1753
1754 /**
1755  * worker_enter_idle - enter idle state
1756  * @worker: worker which is entering idle state
1757  *
1758  * @worker is entering idle state.  Update stats and idle timer if
1759  * necessary.
1760  *
1761  * LOCKING:
1762  * spin_lock_irq(pool->lock).
1763  */
1764 static void worker_enter_idle(struct worker *worker)
1765 {
1766         struct worker_pool *pool = worker->pool;
1767
1768         if (WARN_ON_ONCE(worker->flags & WORKER_IDLE) ||
1769             WARN_ON_ONCE(!list_empty(&worker->entry) &&
1770                          (worker->hentry.next || worker->hentry.pprev)))
1771                 return;
1772
1773         /* can't use worker_set_flags(), also called from create_worker() */
1774         worker->flags |= WORKER_IDLE;
1775         pool->nr_idle++;
1776         worker->last_active = jiffies;
1777
1778         /* idle_list is LIFO */
1779         list_add(&worker->entry, &pool->idle_list);
1780
1781         if (too_many_workers(pool) && !timer_pending(&pool->idle_timer))
1782                 mod_timer(&pool->idle_timer, jiffies + IDLE_WORKER_TIMEOUT);
1783
1784         /*
1785          * Sanity check nr_running.  Because unbind_workers() releases
1786          * pool->lock between setting %WORKER_UNBOUND and zapping
1787          * nr_running, the warning may trigger spuriously.  Check iff
1788          * unbind is not in progress.
1789          */
1790         WARN_ON_ONCE(!(pool->flags & POOL_DISASSOCIATED) &&
1791                      pool->nr_workers == pool->nr_idle &&
1792                      atomic_read(&pool->nr_running));
1793 }
1794
1795 /**
1796  * worker_leave_idle - leave idle state
1797  * @worker: worker which is leaving idle state
1798  *
1799  * @worker is leaving idle state.  Update stats.
1800  *
1801  * LOCKING:
1802  * spin_lock_irq(pool->lock).
1803  */
1804 static void worker_leave_idle(struct worker *worker)
1805 {
1806         struct worker_pool *pool = worker->pool;
1807
1808         if (WARN_ON_ONCE(!(worker->flags & WORKER_IDLE)))
1809                 return;
1810         worker_clr_flags(worker, WORKER_IDLE);
1811         pool->nr_idle--;
1812         list_del_init(&worker->entry);
1813 }
1814
1815 static struct worker *alloc_worker(int node)
1816 {
1817         struct worker *worker;
1818
1819         worker = kzalloc_node(sizeof(*worker), GFP_KERNEL, node);
1820         if (worker) {
1821                 INIT_LIST_HEAD(&worker->entry);
1822                 INIT_LIST_HEAD(&worker->scheduled);
1823                 INIT_LIST_HEAD(&worker->node);
1824                 /* on creation a worker is in !idle && prep state */
1825                 worker->flags = WORKER_PREP;
1826         }
1827         return worker;
1828 }
1829
1830 /**
1831  * worker_attach_to_pool() - attach a worker to a pool
1832  * @worker: worker to be attached
1833  * @pool: the target pool
1834  *
1835  * Attach @worker to @pool.  Once attached, the %WORKER_UNBOUND flag and
1836  * cpu-binding of @worker are kept coordinated with the pool across
1837  * cpu-[un]hotplugs.
1838  */
1839 static void worker_attach_to_pool(struct worker *worker,
1840                                    struct worker_pool *pool)
1841 {
1842         mutex_lock(&wq_pool_attach_mutex);
1843
1844         /*
1845          * set_cpus_allowed_ptr() will fail if the cpumask doesn't have any
1846          * online CPUs.  It'll be re-applied when any of the CPUs come up.
1847          */
1848         set_cpus_allowed_ptr(worker->task, pool->attrs->cpumask);
1849
1850         /*
1851          * The wq_pool_attach_mutex ensures %POOL_DISASSOCIATED remains
1852          * stable across this function.  See the comments above the flag
1853          * definition for details.
1854          */
1855         if (pool->flags & POOL_DISASSOCIATED)
1856                 worker->flags |= WORKER_UNBOUND;
1857
1858         list_add_tail(&worker->node, &pool->workers);
1859         worker->pool = pool;
1860
1861         mutex_unlock(&wq_pool_attach_mutex);
1862 }
1863
1864 /**
1865  * worker_detach_from_pool() - detach a worker from its pool
1866  * @worker: worker which is attached to its pool
1867  *
1868  * Undo the attaching which had been done in worker_attach_to_pool().  The
1869  * caller worker shouldn't access to the pool after detached except it has
1870  * other reference to the pool.
1871  */
1872 static void worker_detach_from_pool(struct worker *worker)
1873 {
1874         struct worker_pool *pool = worker->pool;
1875         struct completion *detach_completion = NULL;
1876
1877         mutex_lock(&wq_pool_attach_mutex);
1878
1879         list_del(&worker->node);
1880         worker->pool = NULL;
1881
1882         if (list_empty(&pool->workers))
1883                 detach_completion = pool->detach_completion;
1884         mutex_unlock(&wq_pool_attach_mutex);
1885
1886         /* clear leftover flags without pool->lock after it is detached */
1887         worker->flags &= ~(WORKER_UNBOUND | WORKER_REBOUND);
1888
1889         if (detach_completion)
1890                 complete(detach_completion);
1891 }
1892
1893 /**
1894  * create_worker - create a new workqueue worker
1895  * @pool: pool the new worker will belong to
1896  *
1897  * Create and start a new worker which is attached to @pool.
1898  *
1899  * CONTEXT:
1900  * Might sleep.  Does GFP_KERNEL allocations.
1901  *
1902  * Return:
1903  * Pointer to the newly created worker.
1904  */
1905 static struct worker *create_worker(struct worker_pool *pool)
1906 {
1907         struct worker *worker = NULL;
1908         int id = -1;
1909         char id_buf[16];
1910
1911         /* ID is needed to determine kthread name */
1912         id = ida_simple_get(&pool->worker_ida, 0, 0, GFP_KERNEL);
1913         if (id < 0)
1914                 goto fail;
1915
1916         worker = alloc_worker(pool->node);
1917         if (!worker)
1918                 goto fail;
1919
1920         worker->id = id;
1921
1922         if (pool->cpu >= 0)
1923                 snprintf(id_buf, sizeof(id_buf), "%d:%d%s", pool->cpu, id,
1924                          pool->attrs->nice < 0  ? "H" : "");
1925         else
1926                 snprintf(id_buf, sizeof(id_buf), "u%d:%d", pool->id, id);
1927
1928         worker->task = kthread_create_on_node(worker_thread, worker, pool->node,
1929                                               "kworker/%s", id_buf);
1930         if (IS_ERR(worker->task))
1931                 goto fail;
1932
1933         set_user_nice(worker->task, pool->attrs->nice);
1934         kthread_bind_mask(worker->task, pool->attrs->cpumask);
1935
1936         /* successful, attach the worker to the pool */
1937         worker_attach_to_pool(worker, pool);
1938
1939         /* start the newly created worker */
1940         spin_lock_irq(&pool->lock);
1941         worker->pool->nr_workers++;
1942         worker_enter_idle(worker);
1943         wake_up_process(worker->task);
1944         spin_unlock_irq(&pool->lock);
1945
1946         return worker;
1947
1948 fail:
1949         if (id >= 0)
1950                 ida_simple_remove(&pool->worker_ida, id);
1951         kfree(worker);
1952         return NULL;
1953 }
1954
1955 /**
1956  * destroy_worker - destroy a workqueue worker
1957  * @worker: worker to be destroyed
1958  *
1959  * Destroy @worker and adjust @pool stats accordingly.  The worker should
1960  * be idle.
1961  *
1962  * CONTEXT:
1963  * spin_lock_irq(pool->lock).
1964  */
1965 static void destroy_worker(struct worker *worker)
1966 {
1967         struct worker_pool *pool = worker->pool;
1968
1969         lockdep_assert_held(&pool->lock);
1970
1971         /* sanity check frenzy */
1972         if (WARN_ON(worker->current_work) ||
1973             WARN_ON(!list_empty(&worker->scheduled)) ||
1974             WARN_ON(!(worker->flags & WORKER_IDLE)))
1975                 return;
1976
1977         pool->nr_workers--;
1978         pool->nr_idle--;
1979
1980         list_del_init(&worker->entry);
1981         worker->flags |= WORKER_DIE;
1982         wake_up_process(worker->task);
1983 }
1984
1985 static void idle_worker_timeout(struct timer_list *t)
1986 {
1987         struct worker_pool *pool = from_timer(pool, t, idle_timer);
1988
1989         spin_lock_irq(&pool->lock);
1990
1991         while (too_many_workers(pool)) {
1992                 struct worker *worker;
1993                 unsigned long expires;
1994
1995                 /* idle_list is kept in LIFO order, check the last one */
1996                 worker = list_entry(pool->idle_list.prev, struct worker, entry);
1997                 expires = worker->last_active + IDLE_WORKER_TIMEOUT;
1998
1999                 if (time_before(jiffies, expires)) {
2000                         mod_timer(&pool->idle_timer, expires);
2001                         break;
2002                 }
2003
2004                 destroy_worker(worker);
2005         }
2006
2007         spin_unlock_irq(&pool->lock);
2008 }
2009
2010 static void send_mayday(struct work_struct *work)
2011 {
2012         struct pool_workqueue *pwq = get_work_pwq(work);
2013         struct workqueue_struct *wq = pwq->wq;
2014
2015         lockdep_assert_held(&wq_mayday_lock);
2016
2017         if (!wq->rescuer)
2018                 return;
2019
2020         /* mayday mayday mayday */
2021         if (list_empty(&pwq->mayday_node)) {
2022                 /*
2023                  * If @pwq is for an unbound wq, its base ref may be put at
2024                  * any time due to an attribute change.  Pin @pwq until the
2025                  * rescuer is done with it.
2026                  */
2027                 get_pwq(pwq);
2028                 list_add_tail(&pwq->mayday_node, &wq->maydays);
2029                 wake_up_process(wq->rescuer->task);
2030         }
2031 }
2032
2033 static void pool_mayday_timeout(struct timer_list *t)
2034 {
2035         struct worker_pool *pool = from_timer(pool, t, mayday_timer);
2036         struct work_struct *work;
2037
2038         spin_lock_irq(&pool->lock);
2039         spin_lock(&wq_mayday_lock);             /* for wq->maydays */
2040
2041         if (need_to_create_worker(pool)) {
2042                 /*
2043                  * We've been trying to create a new worker but
2044                  * haven't been successful.  We might be hitting an
2045                  * allocation deadlock.  Send distress signals to
2046                  * rescuers.
2047                  */
2048                 list_for_each_entry(work, &pool->worklist, entry)
2049                         send_mayday(work);
2050         }
2051
2052         spin_unlock(&wq_mayday_lock);
2053         spin_unlock_irq(&pool->lock);
2054
2055         mod_timer(&pool->mayday_timer, jiffies + MAYDAY_INTERVAL);
2056 }
2057
2058 /**
2059  * maybe_create_worker - create a new worker if necessary
2060  * @pool: pool to create a new worker for
2061  *
2062  * Create a new worker for @pool if necessary.  @pool is guaranteed to
2063  * have at least one idle worker on return from this function.  If
2064  * creating a new worker takes longer than MAYDAY_INTERVAL, mayday is
2065  * sent to all rescuers with works scheduled on @pool to resolve
2066  * possible allocation deadlock.
2067  *
2068  * On return, need_to_create_worker() is guaranteed to be %false and
2069  * may_start_working() %true.
2070  *
2071  * LOCKING:
2072  * spin_lock_irq(pool->lock) which may be released and regrabbed
2073  * multiple times.  Does GFP_KERNEL allocations.  Called only from
2074  * manager.
2075  */
2076 static void maybe_create_worker(struct worker_pool *pool)
2077 __releases(&pool->lock)
2078 __acquires(&pool->lock)
2079 {
2080 restart:
2081         spin_unlock_irq(&pool->lock);
2082
2083         /* if we don't make progress in MAYDAY_INITIAL_TIMEOUT, call for help */
2084         mod_timer(&pool->mayday_timer, jiffies + MAYDAY_INITIAL_TIMEOUT);
2085
2086         while (true) {
2087                 if (create_worker(pool) || !need_to_create_worker(pool))
2088                         break;
2089
2090                 schedule_timeout_interruptible(CREATE_COOLDOWN);
2091
2092                 if (!need_to_create_worker(pool))
2093                         break;
2094         }
2095
2096         del_timer_sync(&pool->mayday_timer);
2097         spin_lock_irq(&pool->lock);
2098         /*
2099          * This is necessary even after a new worker was just successfully
2100          * created as @pool->lock was dropped and the new worker might have
2101          * already become busy.
2102          */
2103         if (need_to_create_worker(pool))
2104                 goto restart;
2105 }
2106
2107 /**
2108  * manage_workers - manage worker pool
2109  * @worker: self
2110  *
2111  * Assume the manager role and manage the worker pool @worker belongs
2112  * to.  At any given time, there can be only zero or one manager per
2113  * pool.  The exclusion is handled automatically by this function.
2114  *
2115  * The caller can safely start processing works on false return.  On
2116  * true return, it's guaranteed that need_to_create_worker() is false
2117  * and may_start_working() is true.
2118  *
2119  * CONTEXT:
2120  * spin_lock_irq(pool->lock) which may be released and regrabbed
2121  * multiple times.  Does GFP_KERNEL allocations.
2122  *
2123  * Return:
2124  * %false if the pool doesn't need management and the caller can safely
2125  * start processing works, %true if management function was performed and
2126  * the conditions that the caller verified before calling the function may
2127  * no longer be true.
2128  */
2129 static bool manage_workers(struct worker *worker)
2130 {
2131         struct worker_pool *pool = worker->pool;
2132
2133         if (pool->flags & POOL_MANAGER_ACTIVE)
2134                 return false;
2135
2136         pool->flags |= POOL_MANAGER_ACTIVE;
2137         pool->manager = worker;
2138
2139         maybe_create_worker(pool);
2140
2141         pool->manager = NULL;
2142         pool->flags &= ~POOL_MANAGER_ACTIVE;
2143         wake_up(&wq_manager_wait);
2144         return true;
2145 }
2146
2147 /**
2148  * process_one_work - process single work
2149  * @worker: self
2150  * @work: work to process
2151  *
2152  * Process @work.  This function contains all the logics necessary to
2153  * process a single work including synchronization against and
2154  * interaction with other workers on the same cpu, queueing and
2155  * flushing.  As long as context requirement is met, any worker can
2156  * call this function to process a work.
2157  *
2158  * CONTEXT:
2159  * spin_lock_irq(pool->lock) which is released and regrabbed.
2160  */
2161 static void process_one_work(struct worker *worker, struct work_struct *work)
2162 __releases(&pool->lock)
2163 __acquires(&pool->lock)
2164 {
2165         struct pool_workqueue *pwq = get_work_pwq(work);
2166         struct worker_pool *pool = worker->pool;
2167         bool cpu_intensive = pwq->wq->flags & WQ_CPU_INTENSIVE;
2168         int work_color;
2169         struct worker *collision;
2170 #ifdef CONFIG_LOCKDEP
2171         /*
2172          * It is permissible to free the struct work_struct from
2173          * inside the function that is called from it, this we need to
2174          * take into account for lockdep too.  To avoid bogus "held
2175          * lock freed" warnings as well as problems when looking into
2176          * work->lockdep_map, make a copy and use that here.
2177          */
2178         struct lockdep_map lockdep_map;
2179
2180         lockdep_copy_map(&lockdep_map, &work->lockdep_map);
2181 #endif
2182         /* ensure we're on the correct CPU */
2183         WARN_ON_ONCE(!(pool->flags & POOL_DISASSOCIATED) &&
2184                      raw_smp_processor_id() != pool->cpu);
2185
2186         /*
2187          * A single work shouldn't be executed concurrently by
2188          * multiple workers on a single cpu.  Check whether anyone is
2189          * already processing the work.  If so, defer the work to the
2190          * currently executing one.
2191          */
2192         collision = find_worker_executing_work(pool, work);
2193         if (unlikely(collision)) {
2194                 move_linked_works(work, &collision->scheduled, NULL);
2195                 return;
2196         }
2197
2198         /* claim and dequeue */
2199         debug_work_deactivate(work);
2200         hash_add(pool->busy_hash, &worker->hentry, (unsigned long)work);
2201         worker->current_work = work;
2202         worker->current_func = work->func;
2203         worker->current_pwq = pwq;
2204         work_color = get_work_color(work);
2205
2206         /*
2207          * Record wq name for cmdline and debug reporting, may get
2208          * overridden through set_worker_desc().
2209          */
2210         strscpy(worker->desc, pwq->wq->name, WORKER_DESC_LEN);
2211
2212         list_del_init(&work->entry);
2213
2214         /*
2215          * CPU intensive works don't participate in concurrency management.
2216          * They're the scheduler's responsibility.  This takes @worker out
2217          * of concurrency management and the next code block will chain
2218          * execution of the pending work items.
2219          */
2220         if (unlikely(cpu_intensive))
2221                 worker_set_flags(worker, WORKER_CPU_INTENSIVE);
2222
2223         /*
2224          * Wake up another worker if necessary.  The condition is always
2225          * false for normal per-cpu workers since nr_running would always
2226          * be >= 1 at this point.  This is used to chain execution of the
2227          * pending work items for WORKER_NOT_RUNNING workers such as the
2228          * UNBOUND and CPU_INTENSIVE ones.
2229          */
2230         if (need_more_worker(pool))
2231                 wake_up_worker(pool);
2232
2233         /*
2234          * Record the last pool and clear PENDING which should be the last
2235          * update to @work.  Also, do this inside @pool->lock so that
2236          * PENDING and queued state changes happen together while IRQ is
2237          * disabled.
2238          */
2239         set_work_pool_and_clear_pending(work, pool->id);
2240
2241         spin_unlock_irq(&pool->lock);
2242
2243         lock_map_acquire(&pwq->wq->lockdep_map);
2244         lock_map_acquire(&lockdep_map);
2245         /*
2246          * Strictly speaking we should mark the invariant state without holding
2247          * any locks, that is, before these two lock_map_acquire()'s.
2248          *
2249          * However, that would result in:
2250          *
2251          *   A(W1)
2252          *   WFC(C)
2253          *              A(W1)
2254          *              C(C)
2255          *
2256          * Which would create W1->C->W1 dependencies, even though there is no
2257          * actual deadlock possible. There are two solutions, using a
2258          * read-recursive acquire on the work(queue) 'locks', but this will then
2259          * hit the lockdep limitation on recursive locks, or simply discard
2260          * these locks.
2261          *
2262          * AFAICT there is no possible deadlock scenario between the
2263          * flush_work() and complete() primitives (except for single-threaded
2264          * workqueues), so hiding them isn't a problem.
2265          */
2266         lockdep_invariant_state(true);
2267         trace_workqueue_execute_start(work);
2268         worker->current_func(work);
2269         /*
2270          * While we must be careful to not use "work" after this, the trace
2271          * point will only record its address.
2272          */
2273         trace_workqueue_execute_end(work);
2274         lock_map_release(&lockdep_map);
2275         lock_map_release(&pwq->wq->lockdep_map);
2276
2277         if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
2278                 pr_err("BUG: workqueue leaked lock or atomic: %s/0x%08x/%d\n"
2279                        "     last function: %ps\n",
2280                        current->comm, preempt_count(), task_pid_nr(current),
2281                        worker->current_func);
2282                 debug_show_held_locks(current);
2283                 dump_stack();
2284         }
2285
2286         /*
2287          * The following prevents a kworker from hogging CPU on !PREEMPT
2288          * kernels, where a requeueing work item waiting for something to
2289          * happen could deadlock with stop_machine as such work item could
2290          * indefinitely requeue itself while all other CPUs are trapped in
2291          * stop_machine. At the same time, report a quiescent RCU state so
2292          * the same condition doesn't freeze RCU.
2293          */
2294         cond_resched();
2295
2296         spin_lock_irq(&pool->lock);
2297
2298         /* clear cpu intensive status */
2299         if (unlikely(cpu_intensive))
2300                 worker_clr_flags(worker, WORKER_CPU_INTENSIVE);
2301
2302         /* tag the worker for identification in schedule() */
2303         worker->last_func = worker->current_func;
2304
2305         /* we're done with it, release */
2306         hash_del(&worker->hentry);
2307         worker->current_work = NULL;
2308         worker->current_func = NULL;
2309         worker->current_pwq = NULL;
2310         pwq_dec_nr_in_flight(pwq, work_color);
2311 }
2312
2313 /**
2314  * process_scheduled_works - process scheduled works
2315  * @worker: self
2316  *
2317  * Process all scheduled works.  Please note that the scheduled list
2318  * may change while processing a work, so this function repeatedly
2319  * fetches a work from the top and executes it.
2320  *
2321  * CONTEXT:
2322  * spin_lock_irq(pool->lock) which may be released and regrabbed
2323  * multiple times.
2324  */
2325 static void process_scheduled_works(struct worker *worker)
2326 {
2327         while (!list_empty(&worker->scheduled)) {
2328                 struct work_struct *work = list_first_entry(&worker->scheduled,
2329                                                 struct work_struct, entry);
2330                 process_one_work(worker, work);
2331         }
2332 }
2333
2334 static void set_pf_worker(bool val)
2335 {
2336         mutex_lock(&wq_pool_attach_mutex);
2337         if (val)
2338                 current->flags |= PF_WQ_WORKER;
2339         else
2340                 current->flags &= ~PF_WQ_WORKER;
2341         mutex_unlock(&wq_pool_attach_mutex);
2342 }
2343
2344 /**
2345  * worker_thread - the worker thread function
2346  * @__worker: self
2347  *
2348  * The worker thread function.  All workers belong to a worker_pool -
2349  * either a per-cpu one or dynamic unbound one.  These workers process all
2350  * work items regardless of their specific target workqueue.  The only
2351  * exception is work items which belong to workqueues with a rescuer which
2352  * will be explained in rescuer_thread().
2353  *
2354  * Return: 0
2355  */
2356 static int worker_thread(void *__worker)
2357 {
2358         struct worker *worker = __worker;
2359         struct worker_pool *pool = worker->pool;
2360
2361         /* tell the scheduler that this is a workqueue worker */
2362         set_pf_worker(true);
2363 woke_up:
2364         spin_lock_irq(&pool->lock);
2365
2366         /* am I supposed to die? */
2367         if (unlikely(worker->flags & WORKER_DIE)) {
2368                 spin_unlock_irq(&pool->lock);
2369                 WARN_ON_ONCE(!list_empty(&worker->entry));
2370                 set_pf_worker(false);
2371
2372                 set_task_comm(worker->task, "kworker/dying");
2373                 ida_simple_remove(&pool->worker_ida, worker->id);
2374                 worker_detach_from_pool(worker);
2375                 kfree(worker);
2376                 return 0;
2377         }
2378
2379         worker_leave_idle(worker);
2380 recheck:
2381         /* no more worker necessary? */
2382         if (!need_more_worker(pool))
2383                 goto sleep;
2384
2385         /* do we need to manage? */
2386         if (unlikely(!may_start_working(pool)) && manage_workers(worker))
2387                 goto recheck;
2388
2389         /*
2390          * ->scheduled list can only be filled while a worker is
2391          * preparing to process a work or actually processing it.
2392          * Make sure nobody diddled with it while I was sleeping.
2393          */
2394         WARN_ON_ONCE(!list_empty(&worker->scheduled));
2395
2396         /*
2397          * Finish PREP stage.  We're guaranteed to have at least one idle
2398          * worker or that someone else has already assumed the manager
2399          * role.  This is where @worker starts participating in concurrency
2400          * management if applicable and concurrency management is restored
2401          * after being rebound.  See rebind_workers() for details.
2402          */
2403         worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);
2404
2405         do {
2406                 struct work_struct *work =
2407                         list_first_entry(&pool->worklist,
2408                                          struct work_struct, entry);
2409
2410                 pool->watchdog_ts = jiffies;
2411
2412                 if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
2413                         /* optimization path, not strictly necessary */
2414                         process_one_work(worker, work);
2415                         if (unlikely(!list_empty(&worker->scheduled)))
2416                                 process_scheduled_works(worker);
2417                 } else {
2418                         move_linked_works(work, &worker->scheduled, NULL);
2419                         process_scheduled_works(worker);
2420                 }
2421         } while (keep_working(pool));
2422
2423         worker_set_flags(worker, WORKER_PREP);
2424 sleep:
2425         /*
2426          * pool->lock is held and there's no work to process and no need to
2427          * manage, sleep.  Workers are woken up only while holding
2428          * pool->lock or from local cpu, so setting the current state
2429          * before releasing pool->lock is enough to prevent losing any
2430          * event.
2431          */
2432         worker_enter_idle(worker);
2433         __set_current_state(TASK_IDLE);
2434         spin_unlock_irq(&pool->lock);
2435         schedule();
2436         goto woke_up;
2437 }
2438
2439 /**
2440  * rescuer_thread - the rescuer thread function
2441  * @__rescuer: self
2442  *
2443  * Workqueue rescuer thread function.  There's one rescuer for each
2444  * workqueue which has WQ_MEM_RECLAIM set.
2445  *
2446  * Regular work processing on a pool may block trying to create a new
2447  * worker which uses GFP_KERNEL allocation which has slight chance of
2448  * developing into deadlock if some works currently on the same queue
2449  * need to be processed to satisfy the GFP_KERNEL allocation.  This is
2450  * the problem rescuer solves.
2451  *
2452  * When such condition is possible, the pool summons rescuers of all
2453  * workqueues which have works queued on the pool and let them process
2454  * those works so that forward progress can be guaranteed.
2455  *
2456  * This should happen rarely.
2457  *
2458  * Return: 0
2459  */
2460 static int rescuer_thread(void *__rescuer)
2461 {
2462         struct worker *rescuer = __rescuer;
2463         struct workqueue_struct *wq = rescuer->rescue_wq;
2464         struct list_head *scheduled = &rescuer->scheduled;
2465         bool should_stop;
2466
2467         set_user_nice(current, RESCUER_NICE_LEVEL);
2468
2469         /*
2470          * Mark rescuer as worker too.  As WORKER_PREP is never cleared, it
2471          * doesn't participate in concurrency management.
2472          */
2473         set_pf_worker(true);
2474 repeat:
2475         set_current_state(TASK_IDLE);
2476
2477         /*
2478          * By the time the rescuer is requested to stop, the workqueue
2479          * shouldn't have any work pending, but @wq->maydays may still have
2480          * pwq(s) queued.  This can happen by non-rescuer workers consuming
2481          * all the work items before the rescuer got to them.  Go through
2482          * @wq->maydays processing before acting on should_stop so that the
2483          * list is always empty on exit.
2484          */
2485         should_stop = kthread_should_stop();
2486
2487         /* see whether any pwq is asking for help */
2488         spin_lock_irq(&wq_mayday_lock);
2489
2490         while (!list_empty(&wq->maydays)) {
2491                 struct pool_workqueue *pwq = list_first_entry(&wq->maydays,
2492                                         struct pool_workqueue, mayday_node);
2493                 struct worker_pool *pool = pwq->pool;
2494                 struct work_struct *work, *n;
2495                 bool first = true;
2496
2497                 __set_current_state(TASK_RUNNING);
2498                 list_del_init(&pwq->mayday_node);
2499
2500                 spin_unlock_irq(&wq_mayday_lock);
2501
2502                 worker_attach_to_pool(rescuer, pool);
2503
2504                 spin_lock_irq(&pool->lock);
2505
2506                 /*
2507                  * Slurp in all works issued via this workqueue and
2508                  * process'em.
2509                  */
2510                 WARN_ON_ONCE(!list_empty(scheduled));
2511                 list_for_each_entry_safe(work, n, &pool->worklist, entry) {
2512                         if (get_work_pwq(work) == pwq) {
2513                                 if (first)
2514                                         pool->watchdog_ts = jiffies;
2515                                 move_linked_works(work, scheduled, &n);
2516                         }
2517                         first = false;
2518                 }
2519
2520                 if (!list_empty(scheduled)) {
2521                         process_scheduled_works(rescuer);
2522
2523                         /*
2524                          * The above execution of rescued work items could
2525                          * have created more to rescue through
2526                          * pwq_activate_first_delayed() or chained
2527                          * queueing.  Let's put @pwq back on mayday list so
2528                          * that such back-to-back work items, which may be
2529                          * being used to relieve memory pressure, don't
2530                          * incur MAYDAY_INTERVAL delay inbetween.
2531                          */
2532                         if (need_to_create_worker(pool)) {
2533                                 spin_lock(&wq_mayday_lock);
2534                                 get_pwq(pwq);
2535                                 list_move_tail(&pwq->mayday_node, &wq->maydays);
2536                                 spin_unlock(&wq_mayday_lock);
2537                         }
2538                 }
2539
2540                 /*
2541                  * Put the reference grabbed by send_mayday().  @pool won't
2542                  * go away while we're still attached to it.
2543                  */
2544                 put_pwq(pwq);
2545
2546                 /*
2547                  * Leave this pool.  If need_more_worker() is %true, notify a
2548                  * regular worker; otherwise, we end up with 0 concurrency
2549                  * and stalling the execution.
2550                  */
2551                 if (need_more_worker(pool))
2552                         wake_up_worker(pool);
2553
2554                 spin_unlock_irq(&pool->lock);
2555
2556                 worker_detach_from_pool(rescuer);
2557
2558                 spin_lock_irq(&wq_mayday_lock);
2559         }
2560
2561         spin_unlock_irq(&wq_mayday_lock);
2562
2563         if (should_stop) {
2564                 __set_current_state(TASK_RUNNING);
2565                 set_pf_worker(false);
2566                 return 0;
2567         }
2568
2569         /* rescuers should never participate in concurrency management */
2570         WARN_ON_ONCE(!(rescuer->flags & WORKER_NOT_RUNNING));
2571         schedule();
2572         goto repeat;
2573 }
2574
2575 /**
2576  * check_flush_dependency - check for flush dependency sanity
2577  * @target_wq: workqueue being flushed
2578  * @target_work: work item being flushed (NULL for workqueue flushes)
2579  *
2580  * %current is trying to flush the whole @target_wq or @target_work on it.
2581  * If @target_wq doesn't have %WQ_MEM_RECLAIM, verify that %current is not
2582  * reclaiming memory or running on a workqueue which doesn't have
2583  * %WQ_MEM_RECLAIM as that can break forward-progress guarantee leading to
2584  * a deadlock.
2585  */
2586 static void check_flush_dependency(struct workqueue_struct *target_wq,
2587                                    struct work_struct *target_work)
2588 {
2589         work_func_t target_func = target_work ? target_work->func : NULL;
2590         struct worker *worker;
2591
2592         if (target_wq->flags & WQ_MEM_RECLAIM)
2593                 return;
2594
2595         worker = current_wq_worker();
2596
2597         WARN_ONCE(current->flags & PF_MEMALLOC,
2598                   "workqueue: PF_MEMALLOC task %d(%s) is flushing !WQ_MEM_RECLAIM %s:%ps",
2599                   current->pid, current->comm, target_wq->name, target_func);
2600         WARN_ONCE(worker && ((worker->current_pwq->wq->flags &
2601                               (WQ_MEM_RECLAIM | __WQ_LEGACY)) == WQ_MEM_RECLAIM),
2602                   "workqueue: WQ_MEM_RECLAIM %s:%ps is flushing !WQ_MEM_RECLAIM %s:%ps",
2603                   worker->current_pwq->wq->name, worker->current_func,
2604                   target_wq->name, target_func);
2605 }
2606
2607 struct wq_barrier {
2608         struct work_struct      work;
2609         struct completion       done;
2610         struct task_struct      *task;  /* purely informational */
2611 };
2612
2613 static void wq_barrier_func(struct work_struct *work)
2614 {
2615         struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
2616         complete(&barr->done);
2617 }
2618
2619 /**
2620  * insert_wq_barrier - insert a barrier work
2621  * @pwq: pwq to insert barrier into
2622  * @barr: wq_barrier to insert
2623  * @target: target work to attach @barr to
2624  * @worker: worker currently executing @target, NULL if @target is not executing
2625  *
2626  * @barr is linked to @target such that @barr is completed only after
2627  * @target finishes execution.  Please note that the ordering
2628  * guarantee is observed only with respect to @target and on the local
2629  * cpu.
2630  *
2631  * Currently, a queued barrier can't be canceled.  This is because
2632  * try_to_grab_pending() can't determine whether the work to be
2633  * grabbed is at the head of the queue and thus can't clear LINKED
2634  * flag of the previous work while there must be a valid next work
2635  * after a work with LINKED flag set.
2636  *
2637  * Note that when @worker is non-NULL, @target may be modified
2638  * underneath us, so we can't reliably determine pwq from @target.
2639  *
2640  * CONTEXT:
2641  * spin_lock_irq(pool->lock).
2642  */
2643 static void insert_wq_barrier(struct pool_workqueue *pwq,
2644                               struct wq_barrier *barr,
2645                               struct work_struct *target, struct worker *worker)
2646 {
2647         struct list_head *head;
2648         unsigned int linked = 0;
2649
2650         /*
2651          * debugobject calls are safe here even with pool->lock locked
2652          * as we know for sure that this will not trigger any of the
2653          * checks and call back into the fixup functions where we
2654          * might deadlock.
2655          */
2656         INIT_WORK_ONSTACK(&barr->work, wq_barrier_func);
2657         __set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(&barr->work));
2658
2659         init_completion_map(&barr->done, &target->lockdep_map);
2660
2661         barr->task = current;
2662
2663         /*
2664          * If @target is currently being executed, schedule the
2665          * barrier to the worker; otherwise, put it after @target.
2666          */
2667         if (worker)
2668                 head = worker->scheduled.next;
2669         else {
2670                 unsigned long *bits = work_data_bits(target);
2671
2672                 head = target->entry.next;
2673                 /* there can already be other linked works, inherit and set */
2674                 linked = *bits & WORK_STRUCT_LINKED;
2675                 __set_bit(WORK_STRUCT_LINKED_BIT, bits);
2676         }
2677
2678         debug_work_activate(&barr->work);
2679         insert_work(pwq, &barr->work, head,
2680                     work_color_to_flags(WORK_NO_COLOR) | linked);
2681 }
2682
2683 /**
2684  * flush_workqueue_prep_pwqs - prepare pwqs for workqueue flushing
2685  * @wq: workqueue being flushed
2686  * @flush_color: new flush color, < 0 for no-op
2687  * @work_color: new work color, < 0 for no-op
2688  *
2689  * Prepare pwqs for workqueue flushing.
2690  *
2691  * If @flush_color is non-negative, flush_color on all pwqs should be
2692  * -1.  If no pwq has in-flight commands at the specified color, all
2693  * pwq->flush_color's stay at -1 and %false is returned.  If any pwq
2694  * has in flight commands, its pwq->flush_color is set to
2695  * @flush_color, @wq->nr_pwqs_to_flush is updated accordingly, pwq
2696  * wakeup logic is armed and %true is returned.
2697  *
2698  * The caller should have initialized @wq->first_flusher prior to
2699  * calling this function with non-negative @flush_color.  If
2700  * @flush_color is negative, no flush color update is done and %false
2701  * is returned.
2702  *
2703  * If @work_color is non-negative, all pwqs should have the same
2704  * work_color which is previous to @work_color and all will be
2705  * advanced to @work_color.
2706  *
2707  * CONTEXT:
2708  * mutex_lock(wq->mutex).
2709  *
2710  * Return:
2711  * %true if @flush_color >= 0 and there's something to flush.  %false
2712  * otherwise.
2713  */
2714 static bool flush_workqueue_prep_pwqs(struct workqueue_struct *wq,
2715                                       int flush_color, int work_color)
2716 {
2717         bool wait = false;
2718         struct pool_workqueue *pwq;
2719
2720         if (flush_color >= 0) {
2721                 WARN_ON_ONCE(atomic_read(&wq->nr_pwqs_to_flush));
2722                 atomic_set(&wq->nr_pwqs_to_flush, 1);
2723         }
2724
2725         for_each_pwq(pwq, wq) {
2726                 struct worker_pool *pool = pwq->pool;
2727
2728                 spin_lock_irq(&pool->lock);
2729
2730                 if (flush_color >= 0) {
2731                         WARN_ON_ONCE(pwq->flush_color != -1);
2732
2733                         if (pwq->nr_in_flight[flush_color]) {
2734                                 pwq->flush_color = flush_color;
2735                                 atomic_inc(&wq->nr_pwqs_to_flush);
2736                                 wait = true;
2737                         }
2738                 }
2739
2740                 if (work_color >= 0) {
2741                         WARN_ON_ONCE(work_color != work_next_color(pwq->work_color));
2742                         pwq->work_color = work_color;
2743                 }
2744
2745                 spin_unlock_irq(&pool->lock);
2746         }
2747
2748         if (flush_color >= 0 && atomic_dec_and_test(&wq->nr_pwqs_to_flush))
2749                 complete(&wq->first_flusher->done);
2750
2751         return wait;
2752 }
2753
2754 /**
2755  * flush_workqueue - ensure that any scheduled work has run to completion.
2756  * @wq: workqueue to flush
2757  *
2758  * This function sleeps until all work items which were queued on entry
2759  * have finished execution, but it is not livelocked by new incoming ones.
2760  */
2761 void flush_workqueue(struct workqueue_struct *wq)
2762 {
2763         struct wq_flusher this_flusher = {
2764                 .list = LIST_HEAD_INIT(this_flusher.list),
2765                 .flush_color = -1,
2766                 .done = COMPLETION_INITIALIZER_ONSTACK_MAP(this_flusher.done, wq->lockdep_map),
2767         };
2768         int next_color;
2769
2770         if (WARN_ON(!wq_online))
2771                 return;
2772
2773         lock_map_acquire(&wq->lockdep_map);
2774         lock_map_release(&wq->lockdep_map);
2775
2776         mutex_lock(&wq->mutex);
2777
2778         /*
2779          * Start-to-wait phase
2780          */
2781         next_color = work_next_color(wq->work_color);
2782
2783         if (next_color != wq->flush_color) {
2784                 /*
2785                  * Color space is not full.  The current work_color
2786                  * becomes our flush_color and work_color is advanced
2787                  * by one.
2788                  */
2789                 WARN_ON_ONCE(!list_empty(&wq->flusher_overflow));
2790                 this_flusher.flush_color = wq->work_color;
2791                 wq->work_color = next_color;
2792
2793                 if (!wq->first_flusher) {
2794                         /* no flush in progress, become the first flusher */
2795                         WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);
2796
2797                         wq->first_flusher = &this_flusher;
2798
2799                         if (!flush_workqueue_prep_pwqs(wq, wq->flush_color,
2800                                                        wq->work_color)) {
2801                                 /* nothing to flush, done */
2802                                 wq->flush_color = next_color;
2803                                 wq->first_flusher = NULL;
2804                                 goto out_unlock;
2805                         }
2806                 } else {
2807                         /* wait in queue */
2808                         WARN_ON_ONCE(wq->flush_color == this_flusher.flush_color);
2809                         list_add_tail(&this_flusher.list, &wq->flusher_queue);
2810                         flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
2811                 }
2812         } else {
2813                 /*
2814                  * Oops, color space is full, wait on overflow queue.
2815                  * The next flush completion will assign us
2816                  * flush_color and transfer to flusher_queue.
2817                  */
2818                 list_add_tail(&this_flusher.list, &wq->flusher_overflow);
2819         }
2820
2821         check_flush_dependency(wq, NULL);
2822
2823         mutex_unlock(&wq->mutex);
2824
2825         wait_for_completion(&this_flusher.done);
2826
2827         /*
2828          * Wake-up-and-cascade phase
2829          *
2830          * First flushers are responsible for cascading flushes and
2831          * handling overflow.  Non-first flushers can simply return.
2832          */
2833         if (wq->first_flusher != &this_flusher)
2834                 return;
2835
2836         mutex_lock(&wq->mutex);
2837
2838         /* we might have raced, check again with mutex held */
2839         if (wq->first_flusher != &this_flusher)
2840                 goto out_unlock;
2841
2842         wq->first_flusher = NULL;
2843
2844         WARN_ON_ONCE(!list_empty(&this_flusher.list));
2845         WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);
2846
2847         while (true) {
2848                 struct wq_flusher *next, *tmp;
2849
2850                 /* complete all the flushers sharing the current flush color */
2851                 list_for_each_entry_safe(next, tmp, &wq->flusher_queue, list) {
2852                         if (next->flush_color != wq->flush_color)
2853                                 break;
2854                         list_del_init(&next->list);
2855                         complete(&next->done);
2856                 }
2857
2858                 WARN_ON_ONCE(!list_empty(&wq->flusher_overflow) &&
2859                              wq->flush_color != work_next_color(wq->work_color));
2860
2861                 /* this flush_color is finished, advance by one */
2862                 wq->flush_color = work_next_color(wq->flush_color);
2863
2864                 /* one color has been freed, handle overflow queue */
2865                 if (!list_empty(&wq->flusher_overflow)) {
2866                         /*
2867                          * Assign the same color to all overflowed
2868                          * flushers, advance work_color and append to
2869                          * flusher_queue.  This is the start-to-wait
2870                          * phase for these overflowed flushers.
2871                          */
2872                         list_for_each_entry(tmp, &wq->flusher_overflow, list)
2873                                 tmp->flush_color = wq->work_color;
2874
2875                         wq->work_color = work_next_color(wq->work_color);
2876
2877                         list_splice_tail_init(&wq->flusher_overflow,
2878                                               &wq->flusher_queue);
2879                         flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
2880                 }
2881
2882                 if (list_empty(&wq->flusher_queue)) {
2883                         WARN_ON_ONCE(wq->flush_color != wq->work_color);
2884                         break;
2885                 }
2886
2887                 /*
2888                  * Need to flush more colors.  Make the next flusher
2889                  * the new first flusher and arm pwqs.
2890                  */
2891                 WARN_ON_ONCE(wq->flush_color == wq->work_color);
2892                 WARN_ON_ONCE(wq->flush_color != next->flush_color);
2893
2894                 list_del_init(&next->list);
2895                 wq->first_flusher = next;
2896
2897                 if (flush_workqueue_prep_pwqs(wq, wq->flush_color, -1))
2898                         break;
2899
2900                 /*
2901                  * Meh... this color is already done, clear first
2902                  * flusher and repeat cascading.
2903                  */
2904                 wq->first_flusher = NULL;
2905         }
2906
2907 out_unlock:
2908         mutex_unlock(&wq->mutex);
2909 }
2910 EXPORT_SYMBOL(flush_workqueue);
2911
2912 /**
2913  * drain_workqueue - drain a workqueue
2914  * @wq: workqueue to drain
2915  *
2916  * Wait until the workqueue becomes empty.  While draining is in progress,
2917  * only chain queueing is allowed.  IOW, only currently pending or running
2918  * work items on @wq can queue further work items on it.  @wq is flushed
2919  * repeatedly until it becomes empty.  The number of flushing is determined
2920  * by the depth of chaining and should be relatively short.  Whine if it
2921  * takes too long.
2922  */
2923 void drain_workqueue(struct workqueue_struct *wq)
2924 {
2925         unsigned int flush_cnt = 0;
2926         struct pool_workqueue *pwq;
2927
2928         /*
2929          * __queue_work() needs to test whether there are drainers, is much
2930          * hotter than drain_workqueue() and already looks at @wq->flags.
2931          * Use __WQ_DRAINING so that queue doesn't have to check nr_drainers.
2932          */
2933         mutex_lock(&wq->mutex);
2934         if (!wq->nr_drainers++)
2935                 wq->flags |= __WQ_DRAINING;
2936         mutex_unlock(&wq->mutex);
2937 reflush:
2938         flush_workqueue(wq);
2939
2940         mutex_lock(&wq->mutex);
2941
2942         for_each_pwq(pwq, wq) {
2943                 bool drained;
2944
2945                 spin_lock_irq(&pwq->pool->lock);
2946                 drained = !pwq->nr_active && list_empty(&pwq->delayed_works);
2947                 spin_unlock_irq(&pwq->pool->lock);
2948
2949                 if (drained)
2950                         continue;
2951
2952                 if (++flush_cnt == 10 ||
2953                     (flush_cnt % 100 == 0 && flush_cnt <= 1000))
2954                         pr_warn("workqueue %s: drain_workqueue() isn't complete after %u tries\n",
2955                                 wq->name, flush_cnt);
2956
2957                 mutex_unlock(&wq->mutex);
2958                 goto reflush;
2959         }
2960
2961         if (!--wq->nr_drainers)
2962                 wq->flags &= ~__WQ_DRAINING;
2963         mutex_unlock(&wq->mutex);
2964 }
2965 EXPORT_SYMBOL_GPL(drain_workqueue);
2966
2967 static bool start_flush_work(struct work_struct *work, struct wq_barrier *barr,
2968                              bool from_cancel)
2969 {
2970         struct worker *worker = NULL;
2971         struct worker_pool *pool;
2972         struct pool_workqueue *pwq;
2973
2974         might_sleep();
2975
2976         rcu_read_lock();
2977         pool = get_work_pool(work);
2978         if (!pool) {
2979                 rcu_read_unlock();
2980                 return false;
2981         }
2982
2983         spin_lock_irq(&pool->lock);
2984         /* see the comment in try_to_grab_pending() with the same code */
2985         pwq = get_work_pwq(work);
2986         if (pwq) {
2987                 if (unlikely(pwq->pool != pool))
2988                         goto already_gone;
2989         } else {
2990                 worker = find_worker_executing_work(pool, work);
2991                 if (!worker)
2992                         goto already_gone;
2993                 pwq = worker->current_pwq;
2994         }
2995
2996         check_flush_dependency(pwq->wq, work);
2997
2998         insert_wq_barrier(pwq, barr, work, worker);
2999         spin_unlock_irq(&pool->lock);
3000
3001         /*
3002          * Force a lock recursion deadlock when using flush_work() inside a
3003          * single-threaded or rescuer equipped workqueue.
3004          *
3005          * For single threaded workqueues the deadlock happens when the work
3006          * is after the work issuing the flush_work(). For rescuer equipped
3007          * workqueues the deadlock happens when the rescuer stalls, blocking
3008          * forward progress.
3009          */
3010         if (!from_cancel &&
3011             (pwq->wq->saved_max_active == 1 || pwq->wq->rescuer)) {
3012                 lock_map_acquire(&pwq->wq->lockdep_map);
3013                 lock_map_release(&pwq->wq->lockdep_map);
3014         }
3015         rcu_read_unlock();
3016         return true;
3017 already_gone:
3018         spin_unlock_irq(&pool->lock);
3019         rcu_read_unlock();
3020         return false;
3021 }
3022
3023 static bool __flush_work(struct work_struct *work, bool from_cancel)
3024 {
3025         struct wq_barrier barr;
3026
3027         if (WARN_ON(!wq_online))
3028                 return false;
3029
3030         if (WARN_ON(!work->func))
3031                 return false;
3032
3033         if (!from_cancel) {
3034                 lock_map_acquire(&work->lockdep_map);
3035                 lock_map_release(&work->lockdep_map);
3036         }
3037
3038         if (start_flush_work(work, &barr, from_cancel)) {
3039                 wait_for_completion(&barr.done);
3040                 destroy_work_on_stack(&barr.work);
3041                 return true;
3042         } else {
3043                 return false;
3044         }
3045 }
3046
3047 /**
3048  * flush_work - wait for a work to finish executing the last queueing instance
3049  * @work: the work to flush
3050  *
3051  * Wait until @work has finished execution.  @work is guaranteed to be idle
3052  * on return if it hasn't been requeued since flush started.
3053  *
3054  * Return:
3055  * %true if flush_work() waited for the work to finish execution,
3056  * %false if it was already idle.
3057  */
3058 bool flush_work(struct work_struct *work)
3059 {
3060         return __flush_work(work, false);
3061 }
3062 EXPORT_SYMBOL_GPL(flush_work);
3063
3064 struct cwt_wait {
3065         wait_queue_entry_t              wait;
3066         struct work_struct      *work;
3067 };
3068
3069 static int cwt_wakefn(wait_queue_entry_t *wait, unsigned mode, int sync, void *key)
3070 {
3071         struct cwt_wait *cwait = container_of(wait, struct cwt_wait, wait);
3072
3073         if (cwait->work != key)
3074                 return 0;
3075         return autoremove_wake_function(wait, mode, sync, key);
3076 }
3077
3078 static bool __cancel_work_timer(struct work_struct *work, bool is_dwork)
3079 {
3080         static DECLARE_WAIT_QUEUE_HEAD(cancel_waitq);
3081         unsigned long flags;
3082         int ret;
3083
3084         do {
3085                 ret = try_to_grab_pending(work, is_dwork, &flags);
3086                 /*
3087                  * If someone else is already canceling, wait for it to
3088                  * finish.  flush_work() doesn't work for PREEMPT_NONE
3089                  * because we may get scheduled between @work's completion
3090                  * and the other canceling task resuming and clearing
3091                  * CANCELING - flush_work() will return false immediately
3092                  * as @work is no longer busy, try_to_grab_pending() will
3093                  * return -ENOENT as @work is still being canceled and the
3094                  * other canceling task won't be able to clear CANCELING as
3095                  * we're hogging the CPU.
3096                  *
3097                  * Let's wait for completion using a waitqueue.  As this
3098                  * may lead to the thundering herd problem, use a custom
3099                  * wake function which matches @work along with exclusive
3100                  * wait and wakeup.
3101                  */
3102                 if (unlikely(ret == -ENOENT)) {
3103                         struct cwt_wait cwait;
3104
3105                         init_wait(&cwait.wait);
3106                         cwait.wait.func = cwt_wakefn;
3107                         cwait.work = work;
3108
3109                         prepare_to_wait_exclusive(&cancel_waitq, &cwait.wait,
3110                                                   TASK_UNINTERRUPTIBLE);
3111                         if (work_is_canceling(work))
3112                                 schedule();
3113                         finish_wait(&cancel_waitq, &cwait.wait);
3114                 }
3115         } while (unlikely(ret < 0));
3116
3117         /* tell other tasks trying to grab @work to back off */
3118         mark_work_canceling(work);
3119         local_irq_restore(flags);
3120
3121         /*
3122          * This allows canceling during early boot.  We know that @work
3123          * isn't executing.
3124          */
3125         if (wq_online)
3126                 __flush_work(work, true);
3127
3128         clear_work_data(work);
3129
3130         /*
3131          * Paired with prepare_to_wait() above so that either
3132          * waitqueue_active() is visible here or !work_is_canceling() is
3133          * visible there.
3134          */
3135         smp_mb();
3136         if (waitqueue_active(&cancel_waitq))
3137                 __wake_up(&cancel_waitq, TASK_NORMAL, 1, work);
3138
3139         return ret;
3140 }
3141
3142 /**
3143  * cancel_work_sync - cancel a work and wait for it to finish
3144  * @work: the work to cancel
3145  *
3146  * Cancel @work and wait for its execution to finish.  This function
3147  * can be used even if the work re-queues itself or migrates to
3148  * another workqueue.  On return from this function, @work is
3149  * guaranteed to be not pending or executing on any CPU.
3150  *
3151  * cancel_work_sync(&delayed_work->work) must not be used for
3152  * delayed_work's.  Use cancel_delayed_work_sync() instead.
3153  *
3154  * The caller must ensure that the workqueue on which @work was last
3155  * queued can't be destroyed before this function returns.
3156  *
3157  * Return:
3158  * %true if @work was pending, %false otherwise.
3159  */
3160 bool cancel_work_sync(struct work_struct *work)
3161 {
3162         return __cancel_work_timer(work, false);
3163 }
3164 EXPORT_SYMBOL_GPL(cancel_work_sync);
3165
3166 /**
3167  * flush_delayed_work - wait for a dwork to finish executing the last queueing
3168  * @dwork: the delayed work to flush
3169  *
3170  * Delayed timer is cancelled and the pending work is queued for
3171  * immediate execution.  Like flush_work(), this function only
3172  * considers the last queueing instance of @dwork.
3173  *
3174  * Return:
3175  * %true if flush_work() waited for the work to finish execution,
3176  * %false if it was already idle.
3177  */
3178 bool flush_delayed_work(struct delayed_work *dwork)
3179 {
3180         local_irq_disable();
3181         if (del_timer_sync(&dwork->timer))
3182                 __queue_work(dwork->cpu, dwork->wq, &dwork->work);
3183         local_irq_enable();
3184         return flush_work(&dwork->work);
3185 }
3186 EXPORT_SYMBOL(flush_delayed_work);
3187
3188 /**
3189  * flush_rcu_work - wait for a rwork to finish executing the last queueing
3190  * @rwork: the rcu work to flush
3191  *
3192  * Return:
3193  * %true if flush_rcu_work() waited for the work to finish execution,
3194  * %false if it was already idle.
3195  */
3196 bool flush_rcu_work(struct rcu_work *rwork)
3197 {
3198         if (test_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(&rwork->work))) {
3199                 rcu_barrier();
3200                 flush_work(&rwork->work);
3201                 return true;
3202         } else {
3203                 return flush_work(&rwork->work);
3204         }
3205 }
3206 EXPORT_SYMBOL(flush_rcu_work);
3207
3208 static bool __cancel_work(struct work_struct *work, bool is_dwork)
3209 {
3210         unsigned long flags;
3211         int ret;
3212
3213         do {
3214                 ret = try_to_grab_pending(work, is_dwork, &flags);
3215         } while (unlikely(ret == -EAGAIN));
3216
3217         if (unlikely(ret < 0))
3218                 return false;
3219
3220         set_work_pool_and_clear_pending(work, get_work_pool_id(work));
3221         local_irq_restore(flags);
3222         return ret;
3223 }
3224
3225 /**
3226  * cancel_delayed_work - cancel a delayed work
3227  * @dwork: delayed_work to cancel
3228  *
3229  * Kill off a pending delayed_work.
3230  *
3231  * Return: %true if @dwork was pending and canceled; %false if it wasn't
3232  * pending.
3233  *
3234  * Note:
3235  * The work callback function may still be running on return, unless
3236  * it returns %true and the work doesn't re-arm itself.  Explicitly flush or
3237  * use cancel_delayed_work_sync() to wait on it.
3238  *
3239  * This function is safe to call from any context including IRQ handler.
3240  */
3241 bool cancel_delayed_work(struct delayed_work *dwork)
3242 {
3243         return __cancel_work(&dwork->work, true);
3244 }
3245 EXPORT_SYMBOL(cancel_delayed_work);
3246
3247 /**
3248  * cancel_delayed_work_sync - cancel a delayed work and wait for it to finish
3249  * @dwork: the delayed work cancel
3250  *
3251  * This is cancel_work_sync() for delayed works.
3252  *
3253  * Return:
3254  * %true if @dwork was pending, %false otherwise.
3255  */
3256 bool cancel_delayed_work_sync(struct delayed_work *dwork)
3257 {
3258         return __cancel_work_timer(&dwork->work, true);
3259 }
3260 EXPORT_SYMBOL(cancel_delayed_work_sync);
3261
3262 /**
3263  * schedule_on_each_cpu - execute a function synchronously on each online CPU
3264  * @func: the function to call
3265  *
3266  * schedule_on_each_cpu() executes @func on each online CPU using the
3267  * system workqueue and blocks until all CPUs have completed.
3268  * schedule_on_each_cpu() is very slow.
3269  *
3270  * Return:
3271  * 0 on success, -errno on failure.
3272  */
3273 int schedule_on_each_cpu(work_func_t func)
3274 {
3275         int cpu;
3276         struct work_struct __percpu *works;
3277
3278         works = alloc_percpu(struct work_struct);
3279         if (!works)
3280                 return -ENOMEM;
3281
3282         get_online_cpus();
3283
3284         for_each_online_cpu(cpu) {
3285                 struct work_struct *work = per_cpu_ptr(works, cpu);
3286
3287                 INIT_WORK(work, func);
3288                 schedule_work_on(cpu, work);
3289         }
3290
3291         for_each_online_cpu(cpu)
3292                 flush_work(per_cpu_ptr(works, cpu));
3293
3294         put_online_cpus();
3295         free_percpu(works);
3296         return 0;
3297 }
3298
3299 /**
3300  * execute_in_process_context - reliably execute the routine with user context
3301  * @fn:         the function to execute
3302  * @ew:         guaranteed storage for the execute work structure (must
3303  *              be available when the work executes)
3304  *
3305  * Executes the function immediately if process context is available,
3306  * otherwise schedules the function for delayed execution.
3307  *
3308  * Return:      0 - function was executed
3309  *              1 - function was scheduled for execution
3310  */
3311 int execute_in_process_context(work_func_t fn, struct execute_work *ew)
3312 {
3313         if (!in_interrupt()) {
3314                 fn(&ew->work);
3315                 return 0;
3316         }
3317
3318         INIT_WORK(&ew->work, fn);
3319         schedule_work(&ew->work);
3320
3321         return 1;
3322 }
3323 EXPORT_SYMBOL_GPL(execute_in_process_context);
3324
3325 /**
3326  * free_workqueue_attrs - free a workqueue_attrs
3327  * @attrs: workqueue_attrs to free
3328  *
3329  * Undo alloc_workqueue_attrs().
3330  */
3331 void free_workqueue_attrs(struct workqueue_attrs *attrs)
3332 {
3333         if (attrs) {
3334                 free_cpumask_var(attrs->cpumask);
3335                 kfree(attrs);
3336         }
3337 }
3338
3339 /**
3340  * alloc_workqueue_attrs - allocate a workqueue_attrs
3341  * @gfp_mask: allocation mask to use
3342  *
3343  * Allocate a new workqueue_attrs, initialize with default settings and
3344  * return it.
3345  *
3346  * Return: The allocated new workqueue_attr on success. %NULL on failure.
3347  */
3348 struct workqueue_attrs *alloc_workqueue_attrs(gfp_t gfp_mask)
3349 {
3350         struct workqueue_attrs *attrs;
3351
3352         attrs = kzalloc(sizeof(*attrs), gfp_mask);
3353         if (!attrs)
3354                 goto fail;
3355         if (!alloc_cpumask_var(&attrs->cpumask, gfp_mask))
3356                 goto fail;
3357
3358         cpumask_copy(attrs->cpumask, cpu_possible_mask);
3359         return attrs;
3360 fail:
3361         free_workqueue_attrs(attrs);
3362         return NULL;
3363 }
3364
3365 static void copy_workqueue_attrs(struct workqueue_attrs *to,
3366                                  const struct workqueue_attrs *from)
3367 {
3368         to->nice = from->nice;
3369         cpumask_copy(to->cpumask, from->cpumask);
3370         /*
3371          * Unlike hash and equality test, this function doesn't ignore
3372          * ->no_numa as it is used for both pool and wq attrs.  Instead,
3373          * get_unbound_pool() explicitly clears ->no_numa after copying.
3374          */
3375         to->no_numa = from->no_numa;
3376 }
3377
3378 /* hash value of the content of @attr */
3379 static u32 wqattrs_hash(const struct workqueue_attrs *attrs)
3380 {
3381         u32 hash = 0;
3382
3383         hash = jhash_1word(attrs->nice, hash);
3384         hash = jhash(cpumask_bits(attrs->cpumask),
3385                      BITS_TO_LONGS(nr_cpumask_bits) * sizeof(long), hash);
3386         return hash;
3387 }
3388
3389 /* content equality test */
3390 static bool wqattrs_equal(const struct workqueue_attrs *a,
3391                           const struct workqueue_attrs *b)
3392 {
3393         if (a->nice != b->nice)
3394                 return false;
3395         if (!cpumask_equal(a->cpumask, b->cpumask))
3396                 return false;
3397         return true;
3398 }
3399
3400 /**
3401  * init_worker_pool - initialize a newly zalloc'd worker_pool
3402  * @pool: worker_pool to initialize
3403  *
3404  * Initialize a newly zalloc'd @pool.  It also allocates @pool->attrs.
3405  *
3406  * Return: 0 on success, -errno on failure.  Even on failure, all fields
3407  * inside @pool proper are initialized and put_unbound_pool() can be called
3408  * on @pool safely to release it.
3409  */
3410 static int init_worker_pool(struct worker_pool *pool)
3411 {
3412         spin_lock_init(&pool->lock);
3413         pool->id = -1;
3414         pool->cpu = -1;
3415         pool->node = NUMA_NO_NODE;
3416         pool->flags |= POOL_DISASSOCIATED;
3417         pool->watchdog_ts = jiffies;
3418         INIT_LIST_HEAD(&pool->worklist);
3419         INIT_LIST_HEAD(&pool->idle_list);
3420         hash_init(pool->busy_hash);
3421
3422         timer_setup(&pool->idle_timer, idle_worker_timeout, TIMER_DEFERRABLE);
3423
3424         timer_setup(&pool->mayday_timer, pool_mayday_timeout, 0);
3425
3426         INIT_LIST_HEAD(&pool->workers);
3427
3428         ida_init(&pool->worker_ida);
3429         INIT_HLIST_NODE(&pool->hash_node);
3430         pool->refcnt = 1;
3431
3432         /* shouldn't fail above this point */
3433         pool->attrs = alloc_workqueue_attrs(GFP_KERNEL);
3434         if (!pool->attrs)
3435                 return -ENOMEM;
3436         return 0;
3437 }
3438
3439 #ifdef CONFIG_LOCKDEP
3440 static void wq_init_lockdep(struct workqueue_struct *wq)
3441 {
3442         char *lock_name;
3443
3444         lockdep_register_key(&wq->key);
3445         lock_name = kasprintf(GFP_KERNEL, "%s%s", "(wq_completion)", wq->name);
3446         if (!lock_name)
3447                 lock_name = wq->name;
3448
3449         wq->lock_name = lock_name;
3450         lockdep_init_map(&wq->lockdep_map, lock_name, &wq->key, 0);
3451 }
3452
3453 static void wq_unregister_lockdep(struct workqueue_struct *wq)
3454 {
3455         lockdep_unregister_key(&wq->key);
3456 }
3457
3458 static void wq_free_lockdep(struct workqueue_struct *wq)
3459 {
3460         if (wq->lock_name != wq->name)
3461                 kfree(wq->lock_name);
3462 }
3463 #else
3464 static void wq_init_lockdep(struct workqueue_struct *wq)
3465 {
3466 }
3467
3468 static void wq_unregister_lockdep(struct workqueue_struct *wq)
3469 {
3470 }
3471
3472 static void wq_free_lockdep(struct workqueue_struct *wq)
3473 {
3474 }
3475 #endif
3476
3477 static void rcu_free_wq(struct rcu_head *rcu)
3478 {
3479         struct workqueue_struct *wq =
3480                 container_of(rcu, struct workqueue_struct, rcu);
3481
3482         wq_free_lockdep(wq);
3483
3484         if (!(wq->flags & WQ_UNBOUND))
3485                 free_percpu(wq->cpu_pwqs);
3486         else
3487                 free_workqueue_attrs(wq->unbound_attrs);
3488
3489         kfree(wq->rescuer);
3490         kfree(wq);
3491 }
3492
3493 static void rcu_free_pool(struct rcu_head *rcu)
3494 {
3495         struct worker_pool *pool = container_of(rcu, struct worker_pool, rcu);
3496
3497         ida_destroy(&pool->worker_ida);
3498         free_workqueue_attrs(pool->attrs);
3499         kfree(pool);
3500 }
3501
3502 /**
3503  * put_unbound_pool - put a worker_pool
3504  * @pool: worker_pool to put
3505  *
3506  * Put @pool.  If its refcnt reaches zero, it gets destroyed in RCU
3507  * safe manner.  get_unbound_pool() calls this function on its failure path
3508  * and this function should be able to release pools which went through,
3509  * successfully or not, init_worker_pool().
3510  *
3511  * Should be called with wq_pool_mutex held.
3512  */
3513 static void put_unbound_pool(struct worker_pool *pool)
3514 {
3515         DECLARE_COMPLETION_ONSTACK(detach_completion);
3516         struct worker *worker;
3517
3518         lockdep_assert_held(&wq_pool_mutex);
3519
3520         if (--pool->refcnt)
3521                 return;
3522
3523         /* sanity checks */
3524         if (WARN_ON(!(pool->cpu < 0)) ||
3525             WARN_ON(!list_empty(&pool->worklist)))
3526                 return;
3527
3528         /* release id and unhash */
3529         if (pool->id >= 0)
3530                 idr_remove(&worker_pool_idr, pool->id);
3531         hash_del(&pool->hash_node);
3532
3533         /*
3534          * Become the manager and destroy all workers.  This prevents
3535          * @pool's workers from blocking on attach_mutex.  We're the last
3536          * manager and @pool gets freed with the flag set.
3537          */
3538         spin_lock_irq(&pool->lock);
3539         wait_event_lock_irq(wq_manager_wait,
3540                             !(pool->flags & POOL_MANAGER_ACTIVE), pool->lock);
3541         pool->flags |= POOL_MANAGER_ACTIVE;
3542
3543         while ((worker = first_idle_worker(pool)))
3544                 destroy_worker(worker);
3545         WARN_ON(pool->nr_workers || pool->nr_idle);
3546         spin_unlock_irq(&pool->lock);
3547
3548         mutex_lock(&wq_pool_attach_mutex);
3549         if (!list_empty(&pool->workers))
3550                 pool->detach_completion = &detach_completion;
3551         mutex_unlock(&wq_pool_attach_mutex);
3552
3553         if (pool->detach_completion)
3554                 wait_for_completion(pool->detach_completion);
3555
3556         /* shut down the timers */
3557         del_timer_sync(&pool->idle_timer);
3558         del_timer_sync(&pool->mayday_timer);
3559
3560         /* RCU protected to allow dereferences from get_work_pool() */
3561         call_rcu(&pool->rcu, rcu_free_pool);
3562 }
3563
3564 /**
3565  * get_unbound_pool - get a worker_pool with the specified attributes
3566  * @attrs: the attributes of the worker_pool to get
3567  *
3568  * Obtain a worker_pool which has the same attributes as @attrs, bump the
3569  * reference count and return it.  If there already is a matching
3570  * worker_pool, it will be used; otherwise, this function attempts to
3571  * create a new one.
3572  *
3573  * Should be called with wq_pool_mutex held.
3574  *
3575  * Return: On success, a worker_pool with the same attributes as @attrs.
3576  * On failure, %NULL.
3577  */
3578 static struct worker_pool *get_unbound_pool(const struct workqueue_attrs *attrs)
3579 {
3580         u32 hash = wqattrs_hash(attrs);
3581         struct worker_pool *pool;
3582         int node;
3583         int target_node = NUMA_NO_NODE;
3584
3585         lockdep_assert_held(&wq_pool_mutex);
3586
3587         /* do we already have a matching pool? */
3588         hash_for_each_possible(unbound_pool_hash, pool, hash_node, hash) {
3589                 if (wqattrs_equal(pool->attrs, attrs)) {
3590                         pool->refcnt++;
3591                         return pool;
3592                 }
3593         }
3594
3595         /* if cpumask is contained inside a NUMA node, we belong to that node */
3596         if (wq_numa_enabled) {
3597                 for_each_node(node) {
3598                         if (cpumask_subset(attrs->cpumask,
3599                                            wq_numa_possible_cpumask[node])) {
3600                                 target_node = node;
3601                                 break;
3602                         }
3603                 }
3604         }
3605
3606         /* nope, create a new one */
3607         pool = kzalloc_node(sizeof(*pool), GFP_KERNEL, target_node);
3608         if (!pool || init_worker_pool(pool) < 0)
3609                 goto fail;
3610
3611         lockdep_set_subclass(&pool->lock, 1);   /* see put_pwq() */
3612         copy_workqueue_attrs(pool->attrs, attrs);
3613         pool->node = target_node;
3614
3615         /*
3616          * no_numa isn't a worker_pool attribute, always clear it.  See
3617          * 'struct workqueue_attrs' comments for detail.
3618          */
3619         pool->attrs->no_numa = false;
3620
3621         if (worker_pool_assign_id(pool) < 0)
3622                 goto fail;
3623
3624         /* create and start the initial worker */
3625         if (wq_online && !create_worker(pool))
3626                 goto fail;
3627
3628         /* install */
3629         hash_add(unbound_pool_hash, &pool->hash_node, hash);
3630
3631         return pool;
3632 fail:
3633         if (pool)
3634                 put_unbound_pool(pool);
3635         return NULL;
3636 }
3637
3638 static void rcu_free_pwq(struct rcu_head *rcu)
3639 {
3640         kmem_cache_free(pwq_cache,
3641                         container_of(rcu, struct pool_workqueue, rcu));
3642 }
3643
3644 /*
3645  * Scheduled on system_wq by put_pwq() when an unbound pwq hits zero refcnt
3646  * and needs to be destroyed.
3647  */
3648 static void pwq_unbound_release_workfn(struct work_struct *work)
3649 {
3650         struct pool_workqueue *pwq = container_of(work, struct pool_workqueue,
3651                                                   unbound_release_work);
3652         struct workqueue_struct *wq = pwq->wq;
3653         struct worker_pool *pool = pwq->pool;
3654         bool is_last;
3655
3656         if (WARN_ON_ONCE(!(wq->flags & WQ_UNBOUND)))
3657                 return;
3658
3659         mutex_lock(&wq->mutex);
3660         list_del_rcu(&pwq->pwqs_node);
3661         is_last = list_empty(&wq->pwqs);
3662         mutex_unlock(&wq->mutex);
3663
3664         mutex_lock(&wq_pool_mutex);
3665         put_unbound_pool(pool);
3666         mutex_unlock(&wq_pool_mutex);
3667
3668         call_rcu(&pwq->rcu, rcu_free_pwq);
3669
3670         /*
3671          * If we're the last pwq going away, @wq is already dead and no one
3672          * is gonna access it anymore.  Schedule RCU free.
3673          */
3674         if (is_last) {
3675                 wq_unregister_lockdep(wq);
3676                 call_rcu(&wq->rcu, rcu_free_wq);
3677         }
3678 }
3679
3680 /**
3681  * pwq_adjust_max_active - update a pwq's max_active to the current setting
3682  * @pwq: target pool_workqueue
3683  *
3684  * If @pwq isn't freezing, set @pwq->max_active to the associated
3685  * workqueue's saved_max_active and activate delayed work items
3686  * accordingly.  If @pwq is freezing, clear @pwq->max_active to zero.
3687  */
3688 static void pwq_adjust_max_active(struct pool_workqueue *pwq)
3689 {
3690         struct workqueue_struct *wq = pwq->wq;
3691         bool freezable = wq->flags & WQ_FREEZABLE;
3692         unsigned long flags;
3693
3694         /* for @wq->saved_max_active */
3695         lockdep_assert_held(&wq->mutex);
3696
3697         /* fast exit for non-freezable wqs */
3698         if (!freezable && pwq->max_active == wq->saved_max_active)
3699                 return;
3700
3701         /* this function can be called during early boot w/ irq disabled */
3702         spin_lock_irqsave(&pwq->pool->lock, flags);
3703
3704         /*
3705          * During [un]freezing, the caller is responsible for ensuring that
3706          * this function is called at least once after @workqueue_freezing
3707          * is updated and visible.
3708          */
3709         if (!freezable || !workqueue_freezing) {
3710                 pwq->max_active = wq->saved_max_active;
3711
3712                 while (!list_empty(&pwq->delayed_works) &&
3713                        pwq->nr_active < pwq->max_active)
3714                         pwq_activate_first_delayed(pwq);
3715
3716                 /*
3717                  * Need to kick a worker after thawed or an unbound wq's
3718                  * max_active is bumped.  It's a slow path.  Do it always.
3719                  */
3720                 wake_up_worker(pwq->pool);
3721         } else {
3722                 pwq->max_active = 0;
3723         }
3724
3725         spin_unlock_irqrestore(&pwq->pool->lock, flags);
3726 }
3727
3728 /* initialize newly alloced @pwq which is associated with @wq and @pool */
3729 static void init_pwq(struct pool_workqueue *pwq, struct workqueue_struct *wq,
3730                      struct worker_pool *pool)
3731 {
3732         BUG_ON((unsigned long)pwq & WORK_STRUCT_FLAG_MASK);
3733
3734         memset(pwq, 0, sizeof(*pwq));
3735
3736         pwq->pool = pool;
3737         pwq->wq = wq;
3738         pwq->flush_color = -1;
3739         pwq->refcnt = 1;
3740         INIT_LIST_HEAD(&pwq->delayed_works);
3741         INIT_LIST_HEAD(&pwq->pwqs_node);
3742         INIT_LIST_HEAD(&pwq->mayday_node);
3743         INIT_WORK(&pwq->unbound_release_work, pwq_unbound_release_workfn);
3744 }
3745
3746 /* sync @pwq with the current state of its associated wq and link it */
3747 static void link_pwq(struct pool_workqueue *pwq)
3748 {
3749         struct workqueue_struct *wq = pwq->wq;
3750
3751         lockdep_assert_held(&wq->mutex);
3752
3753         /* may be called multiple times, ignore if already linked */
3754         if (!list_empty(&pwq->pwqs_node))
3755                 return;
3756
3757         /* set the matching work_color */
3758         pwq->work_color = wq->work_color;
3759
3760         /* sync max_active to the current setting */
3761         pwq_adjust_max_active(pwq);
3762
3763         /* link in @pwq */
3764         list_add_rcu(&pwq->pwqs_node, &wq->pwqs);
3765 }
3766
3767 /* obtain a pool matching @attr and create a pwq associating the pool and @wq */
3768 static struct pool_workqueue *alloc_unbound_pwq(struct workqueue_struct *wq,
3769                                         const struct workqueue_attrs *attrs)
3770 {
3771         struct worker_pool *pool;
3772         struct pool_workqueue *pwq;
3773
3774         lockdep_assert_held(&wq_pool_mutex);
3775
3776         pool = get_unbound_pool(attrs);
3777         if (!pool)
3778                 return NULL;
3779
3780         pwq = kmem_cache_alloc_node(pwq_cache, GFP_KERNEL, pool->node);
3781         if (!pwq) {
3782                 put_unbound_pool(pool);
3783                 return NULL;
3784         }
3785
3786         init_pwq(pwq, wq, pool);
3787         return pwq;
3788 }
3789
3790 /**
3791  * wq_calc_node_cpumask - calculate a wq_attrs' cpumask for the specified node
3792  * @attrs: the wq_attrs of the default pwq of the target workqueue
3793  * @node: the target NUMA node
3794  * @cpu_going_down: if >= 0, the CPU to consider as offline
3795  * @cpumask: outarg, the resulting cpumask
3796  *
3797  * Calculate the cpumask a workqueue with @attrs should use on @node.  If
3798  * @cpu_going_down is >= 0, that cpu is considered offline during
3799  * calculation.  The result is stored in @cpumask.
3800  *
3801  * If NUMA affinity is not enabled, @attrs->cpumask is always used.  If
3802  * enabled and @node has online CPUs requested by @attrs, the returned
3803  * cpumask is the intersection of the possible CPUs of @node and
3804  * @attrs->cpumask.
3805  *
3806  * The caller is responsible for ensuring that the cpumask of @node stays
3807  * stable.
3808  *
3809  * Return: %true if the resulting @cpumask is different from @attrs->cpumask,
3810  * %false if equal.
3811  */
3812 static bool wq_calc_node_cpumask(const struct workqueue_attrs *attrs, int node,
3813                                  int cpu_going_down, cpumask_t *cpumask)
3814 {
3815         if (!wq_numa_enabled || attrs->no_numa)
3816                 goto use_dfl;
3817
3818         /* does @node have any online CPUs @attrs wants? */
3819         cpumask_and(cpumask, cpumask_of_node(node), attrs->cpumask);
3820         if (cpu_going_down >= 0)
3821                 cpumask_clear_cpu(cpu_going_down, cpumask);
3822
3823         if (cpumask_empty(cpumask))
3824                 goto use_dfl;
3825
3826         /* yeap, return possible CPUs in @node that @attrs wants */
3827         cpumask_and(cpumask, attrs->cpumask, wq_numa_possible_cpumask[node]);
3828
3829         if (cpumask_empty(cpumask)) {
3830                 pr_warn_once("WARNING: workqueue cpumask: online intersect > "
3831                                 "possible intersect\n");
3832                 return false;
3833         }
3834
3835         return !cpumask_equal(cpumask, attrs->cpumask);
3836
3837 use_dfl:
3838         cpumask_copy(cpumask, attrs->cpumask);
3839         return false;
3840 }
3841
3842 /* install @pwq into @wq's numa_pwq_tbl[] for @node and return the old pwq */
3843 static struct pool_workqueue *numa_pwq_tbl_install(struct workqueue_struct *wq,
3844                                                    int node,
3845                                                    struct pool_workqueue *pwq)
3846 {
3847         struct pool_workqueue *old_pwq;
3848
3849         lockdep_assert_held(&wq_pool_mutex);
3850         lockdep_assert_held(&wq->mutex);
3851
3852         /* link_pwq() can handle duplicate calls */
3853         link_pwq(pwq);
3854
3855         old_pwq = rcu_access_pointer(wq->numa_pwq_tbl[node]);
3856         rcu_assign_pointer(wq->numa_pwq_tbl[node], pwq);
3857         return old_pwq;
3858 }
3859
3860 /* context to store the prepared attrs & pwqs before applying */
3861 struct apply_wqattrs_ctx {
3862         struct workqueue_struct *wq;            /* target workqueue */
3863         struct workqueue_attrs  *attrs;         /* attrs to apply */
3864         struct list_head        list;           /* queued for batching commit */
3865         struct pool_workqueue   *dfl_pwq;
3866         struct pool_workqueue   *pwq_tbl[];
3867 };
3868
3869 /* free the resources after success or abort */
3870 static void apply_wqattrs_cleanup(struct apply_wqattrs_ctx *ctx)
3871 {
3872         if (ctx) {
3873                 int node;
3874
3875                 for_each_node(node)
3876                         put_pwq_unlocked(ctx->pwq_tbl[node]);
3877                 put_pwq_unlocked(ctx->dfl_pwq);
3878
3879                 free_workqueue_attrs(ctx->attrs);
3880
3881                 kfree(ctx);
3882         }
3883 }
3884
3885 /* allocate the attrs and pwqs for later installation */
3886 static struct apply_wqattrs_ctx *
3887 apply_wqattrs_prepare(struct workqueue_struct *wq,
3888                       const struct workqueue_attrs *attrs)
3889 {
3890         struct apply_wqattrs_ctx *ctx;
3891         struct workqueue_attrs *new_attrs, *tmp_attrs;
3892         int node;
3893
3894         lockdep_assert_held(&wq_pool_mutex);
3895
3896         ctx = kzalloc(struct_size(ctx, pwq_tbl, nr_node_ids), GFP_KERNEL);
3897
3898         new_attrs = alloc_workqueue_attrs(GFP_KERNEL);
3899         tmp_attrs = alloc_workqueue_attrs(GFP_KERNEL);
3900         if (!ctx || !new_attrs || !tmp_attrs)
3901                 goto out_free;
3902
3903         /*
3904          * Calculate the attrs of the default pwq.
3905          * If the user configured cpumask doesn't overlap with the
3906          * wq_unbound_cpumask, we fallback to the wq_unbound_cpumask.
3907          */
3908         copy_workqueue_attrs(new_attrs, attrs);
3909         cpumask_and(new_attrs->cpumask, new_attrs->cpumask, wq_unbound_cpumask);
3910         if (unlikely(cpumask_empty(new_attrs->cpumask)))
3911                 cpumask_copy(new_attrs->cpumask, wq_unbound_cpumask);
3912
3913         /*
3914          * We may create multiple pwqs with differing cpumasks.  Make a
3915          * copy of @new_attrs which will be modified and used to obtain
3916          * pools.
3917          */
3918         copy_workqueue_attrs(tmp_attrs, new_attrs);
3919
3920         /*
3921          * If something goes wrong during CPU up/down, we'll fall back to
3922          * the default pwq covering whole @attrs->cpumask.  Always create
3923          * it even if we don't use it immediately.
3924          */
3925         ctx->dfl_pwq = alloc_unbound_pwq(wq, new_attrs);
3926         if (!ctx->dfl_pwq)
3927                 goto out_free;
3928
3929         for_each_node(node) {
3930                 if (wq_calc_node_cpumask(new_attrs, node, -1, tmp_attrs->cpumask)) {
3931                         ctx->pwq_tbl[node] = alloc_unbound_pwq(wq, tmp_attrs);
3932                         if (!ctx->pwq_tbl[node])
3933                                 goto out_free;
3934                 } else {
3935                         ctx->dfl_pwq->refcnt++;
3936                         ctx->pwq_tbl[node] = ctx->dfl_pwq;
3937                 }
3938         }
3939
3940         /* save the user configured attrs and sanitize it. */
3941         copy_workqueue_attrs(new_attrs, attrs);
3942         cpumask_and(new_attrs->cpumask, new_attrs->cpumask, cpu_possible_mask);
3943         ctx->attrs = new_attrs;
3944
3945         ctx->wq = wq;
3946         free_workqueue_attrs(tmp_attrs);
3947         return ctx;
3948
3949 out_free:
3950         free_workqueue_attrs(tmp_attrs);
3951         free_workqueue_attrs(new_attrs);
3952         apply_wqattrs_cleanup(ctx);
3953         return NULL;
3954 }
3955
3956 /* set attrs and install prepared pwqs, @ctx points to old pwqs on return */
3957 static void apply_wqattrs_commit(struct apply_wqattrs_ctx *ctx)
3958 {
3959         int node;
3960
3961         /* all pwqs have been created successfully, let's install'em */
3962         mutex_lock(&ctx->wq->mutex);
3963
3964         copy_workqueue_attrs(ctx->wq->unbound_attrs, ctx->attrs);
3965
3966         /* save the previous pwq and install the new one */
3967         for_each_node(node)
3968                 ctx->pwq_tbl[node] = numa_pwq_tbl_install(ctx->wq, node,
3969                                                           ctx->pwq_tbl[node]);
3970
3971         /* @dfl_pwq might not have been used, ensure it's linked */
3972         link_pwq(ctx->dfl_pwq);
3973         swap(ctx->wq->dfl_pwq, ctx->dfl_pwq);
3974
3975         mutex_unlock(&ctx->wq->mutex);
3976 }
3977
3978 static void apply_wqattrs_lock(void)
3979 {
3980         /* CPUs should stay stable across pwq creations and installations */
3981         get_online_cpus();
3982         mutex_lock(&wq_pool_mutex);
3983 }
3984
3985 static void apply_wqattrs_unlock(void)
3986 {
3987         mutex_unlock(&wq_pool_mutex);
3988         put_online_cpus();
3989 }
3990
3991 static int apply_workqueue_attrs_locked(struct workqueue_struct *wq,
3992                                         const struct workqueue_attrs *attrs)
3993 {
3994         struct apply_wqattrs_ctx *ctx;
3995
3996         /* only unbound workqueues can change attributes */
3997         if (WARN_ON(!(wq->flags & WQ_UNBOUND)))
3998                 return -EINVAL;
3999
4000         /* creating multiple pwqs breaks ordering guarantee */
4001         if (!list_empty(&wq->pwqs)) {
4002                 if (WARN_ON(wq->flags & __WQ_ORDERED_EXPLICIT))
4003                         return -EINVAL;
4004
4005                 wq->flags &= ~__WQ_ORDERED;
4006         }
4007
4008         ctx = apply_wqattrs_prepare(wq, attrs);
4009         if (!ctx)
4010                 return -ENOMEM;
4011
4012         /* the ctx has been prepared successfully, let's commit it */
4013         apply_wqattrs_commit(ctx);
4014         apply_wqattrs_cleanup(ctx);
4015
4016         return 0;
4017 }
4018
4019 /**
4020  * apply_workqueue_attrs - apply new workqueue_attrs to an unbound workqueue
4021  * @wq: the target workqueue
4022  * @attrs: the workqueue_attrs to apply, allocated with alloc_workqueue_attrs()
4023  *
4024  * Apply @attrs to an unbound workqueue @wq.  Unless disabled, on NUMA
4025  * machines, this function maps a separate pwq to each NUMA node with
4026  * possibles CPUs in @attrs->cpumask so that work items are affine to the
4027  * NUMA node it was issued on.  Older pwqs are released as in-flight work
4028  * items finish.  Note that a work item which repeatedly requeues itself
4029  * back-to-back will stay on its current pwq.
4030  *
4031  * Performs GFP_KERNEL allocations.
4032  *
4033  * Return: 0 on success and -errno on failure.
4034  */
4035 int apply_workqueue_attrs(struct workqueue_struct *wq,
4036                           const struct workqueue_attrs *attrs)
4037 {
4038         int ret;
4039
4040         apply_wqattrs_lock();
4041         ret = apply_workqueue_attrs_locked(wq, attrs);
4042         apply_wqattrs_unlock();
4043
4044         return ret;
4045 }
4046 EXPORT_SYMBOL_GPL(apply_workqueue_attrs);
4047
4048 /**
4049  * wq_update_unbound_numa - update NUMA affinity of a wq for CPU hot[un]plug
4050  * @wq: the target workqueue
4051  * @cpu: the CPU coming up or going down
4052  * @online: whether @cpu is coming up or going down
4053  *
4054  * This function is to be called from %CPU_DOWN_PREPARE, %CPU_ONLINE and
4055  * %CPU_DOWN_FAILED.  @cpu is being hot[un]plugged, update NUMA affinity of
4056  * @wq accordingly.
4057  *
4058  * If NUMA affinity can't be adjusted due to memory allocation failure, it
4059  * falls back to @wq->dfl_pwq which may not be optimal but is always
4060  * correct.
4061  *
4062  * Note that when the last allowed CPU of a NUMA node goes offline for a
4063  * workqueue with a cpumask spanning multiple nodes, the workers which were
4064  * already executing the work items for the workqueue will lose their CPU
4065  * affinity and may execute on any CPU.  This is similar to how per-cpu
4066  * workqueues behave on CPU_DOWN.  If a workqueue user wants strict
4067  * affinity, it's the user's responsibility to flush the work item from
4068  * CPU_DOWN_PREPARE.
4069  */
4070 static void wq_update_unbound_numa(struct workqueue_struct *wq, int cpu,
4071                                    bool online)
4072 {
4073         int node = cpu_to_node(cpu);
4074         int cpu_off = online ? -1 : cpu;
4075         struct pool_workqueue *old_pwq = NULL, *pwq;
4076         struct workqueue_attrs *target_attrs;
4077         cpumask_t *cpumask;
4078
4079         lockdep_assert_held(&wq_pool_mutex);
4080
4081         if (!wq_numa_enabled || !(wq->flags & WQ_UNBOUND) ||
4082             wq->unbound_attrs->no_numa)
4083                 return;
4084
4085         /*
4086          * We don't wanna alloc/free wq_attrs for each wq for each CPU.
4087          * Let's use a preallocated one.  The following buf is protected by
4088          * CPU hotplug exclusion.
4089          */
4090         target_attrs = wq_update_unbound_numa_attrs_buf;
4091         cpumask = target_attrs->cpumask;
4092
4093         copy_workqueue_attrs(target_attrs, wq->unbound_attrs);
4094         pwq = unbound_pwq_by_node(wq, node);
4095
4096         /*
4097          * Let's determine what needs to be done.  If the target cpumask is
4098          * different from the default pwq's, we need to compare it to @pwq's
4099          * and create a new one if they don't match.  If the target cpumask
4100          * equals the default pwq's, the default pwq should be used.
4101          */
4102         if (wq_calc_node_cpumask(wq->dfl_pwq->pool->attrs, node, cpu_off, cpumask)) {
4103                 if (cpumask_equal(cpumask, pwq->pool->attrs->cpumask))
4104                         return;
4105         } else {
4106                 goto use_dfl_pwq;
4107         }
4108
4109         /* create a new pwq */
4110         pwq = alloc_unbound_pwq(wq, target_attrs);
4111         if (!pwq) {
4112                 pr_warn("workqueue: allocation failed while updating NUMA affinity of \"%s\"\n",
4113                         wq->name);
4114                 goto use_dfl_pwq;
4115         }
4116
4117         /* Install the new pwq. */
4118         mutex_lock(&wq->mutex);
4119         old_pwq = numa_pwq_tbl_install(wq, node, pwq);
4120         goto out_unlock;
4121
4122 use_dfl_pwq:
4123         mutex_lock(&wq->mutex);
4124         spin_lock_irq(&wq->dfl_pwq->pool->lock);
4125         get_pwq(wq->dfl_pwq);
4126         spin_unlock_irq(&wq->dfl_pwq->pool->lock);
4127         old_pwq = numa_pwq_tbl_install(wq, node, wq->dfl_pwq);
4128 out_unlock:
4129         mutex_unlock(&wq->mutex);
4130         put_pwq_unlocked(old_pwq);
4131 }
4132
4133 static int alloc_and_link_pwqs(struct workqueue_struct *wq)
4134 {
4135         bool highpri = wq->flags & WQ_HIGHPRI;
4136         int cpu, ret;
4137
4138         if (!(wq->flags & WQ_UNBOUND)) {
4139                 wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);
4140                 if (!wq->cpu_pwqs)
4141                         return -ENOMEM;
4142
4143                 for_each_possible_cpu(cpu) {
4144                         struct pool_workqueue *pwq =
4145                                 per_cpu_ptr(wq->cpu_pwqs, cpu);
4146                         struct worker_pool *cpu_pools =
4147                                 per_cpu(cpu_worker_pools, cpu);
4148
4149                         init_pwq(pwq, wq, &cpu_pools[highpri]);
4150
4151                         mutex_lock(&wq->mutex);
4152                         link_pwq(pwq);
4153                         mutex_unlock(&wq->mutex);
4154                 }
4155                 return 0;
4156         } else if (wq->flags & __WQ_ORDERED) {
4157                 ret = apply_workqueue_attrs(wq, ordered_wq_attrs[highpri]);
4158                 /* there should only be single pwq for ordering guarantee */
4159                 WARN(!ret && (wq->pwqs.next != &wq->dfl_pwq->pwqs_node ||
4160                               wq->pwqs.prev != &wq->dfl_pwq->pwqs_node),
4161                      "ordering guarantee broken for workqueue %s\n", wq->name);
4162                 return ret;
4163         } else {
4164                 return apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
4165         }
4166 }
4167
4168 static int wq_clamp_max_active(int max_active, unsigned int flags,
4169                                const char *name)
4170 {
4171         int lim = flags & WQ_UNBOUND ? WQ_UNBOUND_MAX_ACTIVE : WQ_MAX_ACTIVE;
4172
4173         if (max_active < 1 || max_active > lim)
4174                 pr_warn("workqueue: max_active %d requested for %s is out of range, clamping between %d and %d\n",
4175                         max_active, name, 1, lim);
4176
4177         return clamp_val(max_active, 1, lim);
4178 }
4179
4180 /*
4181  * Workqueues which may be used during memory reclaim should have a rescuer
4182  * to guarantee forward progress.
4183  */
4184 static int init_rescuer(struct workqueue_struct *wq)
4185 {
4186         struct worker *rescuer;
4187         int ret;
4188
4189         if (!(wq->flags & WQ_MEM_RECLAIM))
4190                 return 0;
4191
4192         rescuer = alloc_worker(NUMA_NO_NODE);
4193         if (!rescuer)
4194                 return -ENOMEM;
4195
4196         rescuer->rescue_wq = wq;
4197         rescuer->task = kthread_create(rescuer_thread, rescuer, "%s", wq->name);
4198         ret = PTR_ERR_OR_ZERO(rescuer->task);
4199         if (ret) {
4200                 kfree(rescuer);
4201                 return ret;
4202         }
4203
4204         wq->rescuer = rescuer;
4205         kthread_bind_mask(rescuer->task, cpu_possible_mask);
4206         wake_up_process(rescuer->task);
4207
4208         return 0;
4209 }
4210
4211 __printf(1, 4)
4212 struct workqueue_struct *alloc_workqueue(const char *fmt,
4213                                          unsigned int flags,
4214                                          int max_active, ...)
4215 {
4216         size_t tbl_size = 0;
4217         va_list args;
4218         struct workqueue_struct *wq;
4219         struct pool_workqueue *pwq;
4220
4221         /*
4222          * Unbound && max_active == 1 used to imply ordered, which is no
4223          * longer the case on NUMA machines due to per-node pools.  While
4224          * alloc_ordered_workqueue() is the right way to create an ordered
4225          * workqueue, keep the previous behavior to avoid subtle breakages
4226          * on NUMA.
4227          */
4228         if ((flags & WQ_UNBOUND) && max_active == 1)
4229                 flags |= __WQ_ORDERED;
4230
4231         /* see the comment above the definition of WQ_POWER_EFFICIENT */
4232         if ((flags & WQ_POWER_EFFICIENT) && wq_power_efficient)
4233                 flags |= WQ_UNBOUND;
4234
4235         /* allocate wq and format name */
4236         if (flags & WQ_UNBOUND)
4237                 tbl_size = nr_node_ids * sizeof(wq->numa_pwq_tbl[0]);
4238
4239         wq = kzalloc(sizeof(*wq) + tbl_size, GFP_KERNEL);
4240         if (!wq)
4241                 return NULL;
4242
4243         if (flags & WQ_UNBOUND) {
4244                 wq->unbound_attrs = alloc_workqueue_attrs(GFP_KERNEL);
4245                 if (!wq->unbound_attrs)
4246                         goto err_free_wq;
4247         }
4248
4249         va_start(args, max_active);
4250         vsnprintf(wq->name, sizeof(wq->name), fmt, args);
4251         va_end(args);
4252
4253         max_active = max_active ?: WQ_DFL_ACTIVE;
4254         max_active = wq_clamp_max_active(max_active, flags, wq->name);
4255
4256         /* init wq */
4257         wq->flags = flags;
4258         wq->saved_max_active = max_active;
4259         mutex_init(&wq->mutex);
4260         atomic_set(&wq->nr_pwqs_to_flush, 0);
4261         INIT_LIST_HEAD(&wq->pwqs);
4262         INIT_LIST_HEAD(&wq->flusher_queue);
4263         INIT_LIST_HEAD(&wq->flusher_overflow);
4264         INIT_LIST_HEAD(&wq->maydays);
4265
4266         wq_init_lockdep(wq);
4267         INIT_LIST_HEAD(&wq->list);
4268
4269         if (alloc_and_link_pwqs(wq) < 0)
4270                 goto err_unreg_lockdep;
4271
4272         if (wq_online && init_rescuer(wq) < 0)
4273                 goto err_destroy;
4274
4275         if ((wq->flags & WQ_SYSFS) && workqueue_sysfs_register(wq))
4276                 goto err_destroy;
4277
4278         /*
4279          * wq_pool_mutex protects global freeze state and workqueues list.
4280          * Grab it, adjust max_active and add the new @wq to workqueues
4281          * list.
4282          */
4283         mutex_lock(&wq_pool_mutex);
4284
4285         mutex_lock(&wq->mutex);
4286         for_each_pwq(pwq, wq)
4287                 pwq_adjust_max_active(pwq);
4288         mutex_unlock(&wq->mutex);
4289
4290         list_add_tail_rcu(&wq->list, &workqueues);
4291
4292         mutex_unlock(&wq_pool_mutex);
4293
4294         return wq;
4295
4296 err_unreg_lockdep:
4297         wq_unregister_lockdep(wq);
4298         wq_free_lockdep(wq);
4299 err_free_wq:
4300         free_workqueue_attrs(wq->unbound_attrs);
4301         kfree(wq);
4302         return NULL;
4303 err_destroy:
4304         destroy_workqueue(wq);
4305         return NULL;
4306 }
4307 EXPORT_SYMBOL_GPL(alloc_workqueue);
4308
4309 /**
4310  * destroy_workqueue - safely terminate a workqueue
4311  * @wq: target workqueue
4312  *
4313  * Safely destroy a workqueue. All work currently pending will be done first.
4314  */
4315 void destroy_workqueue(struct workqueue_struct *wq)
4316 {
4317         struct pool_workqueue *pwq;
4318         int node;
4319
4320         /* drain it before proceeding with destruction */
4321         drain_workqueue(wq);
4322
4323         /* sanity checks */
4324         mutex_lock(&wq->mutex);
4325         for_each_pwq(pwq, wq) {
4326                 int i;
4327
4328                 for (i = 0; i < WORK_NR_COLORS; i++) {
4329                         if (WARN_ON(pwq->nr_in_flight[i])) {
4330                                 mutex_unlock(&wq->mutex);
4331                                 show_workqueue_state();
4332                                 return;
4333                         }
4334                 }
4335
4336                 if (WARN_ON((pwq != wq->dfl_pwq) && (pwq->refcnt > 1)) ||
4337                     WARN_ON(pwq->nr_active) ||
4338                     WARN_ON(!list_empty(&pwq->delayed_works))) {
4339                         mutex_unlock(&wq->mutex);
4340                         show_workqueue_state();
4341                         return;
4342                 }
4343         }
4344         mutex_unlock(&wq->mutex);
4345
4346         /*
4347          * wq list is used to freeze wq, remove from list after
4348          * flushing is complete in case freeze races us.
4349          */
4350         mutex_lock(&wq_pool_mutex);
4351         list_del_rcu(&wq->list);
4352         mutex_unlock(&wq_pool_mutex);
4353
4354         workqueue_sysfs_unregister(wq);
4355
4356         if (wq->rescuer)
4357                 kthread_stop(wq->rescuer->task);
4358
4359         if (!(wq->flags & WQ_UNBOUND)) {
4360                 wq_unregister_lockdep(wq);
4361                 /*
4362                  * The base ref is never dropped on per-cpu pwqs.  Directly
4363                  * schedule RCU free.
4364                  */
4365                 call_rcu(&wq->rcu, rcu_free_wq);
4366         } else {
4367                 /*
4368                  * We're the sole accessor of @wq at this point.  Directly
4369                  * access numa_pwq_tbl[] and dfl_pwq to put the base refs.
4370                  * @wq will be freed when the last pwq is released.
4371                  */
4372                 for_each_node(node) {
4373                         pwq = rcu_access_pointer(wq->numa_pwq_tbl[node]);
4374                         RCU_INIT_POINTER(wq->numa_pwq_tbl[node], NULL);
4375                         put_pwq_unlocked(pwq);
4376                 }
4377
4378                 /*
4379                  * Put dfl_pwq.  @wq may be freed any time after dfl_pwq is
4380                  * put.  Don't access it afterwards.
4381                  */
4382                 pwq = wq->dfl_pwq;
4383                 wq->dfl_pwq = NULL;
4384                 put_pwq_unlocked(pwq);
4385         }
4386 }
4387 EXPORT_SYMBOL_GPL(destroy_workqueue);
4388
4389 /**
4390  * workqueue_set_max_active - adjust max_active of a workqueue
4391  * @wq: target workqueue
4392  * @max_active: new max_active value.
4393  *
4394  * Set max_active of @wq to @max_active.
4395  *
4396  * CONTEXT:
4397  * Don't call from IRQ context.
4398  */
4399 void workqueue_set_max_active(struct workqueue_struct *wq, int max_active)
4400 {
4401         struct pool_workqueue *pwq;
4402
4403         /* disallow meddling with max_active for ordered workqueues */
4404         if (WARN_ON(wq->flags & __WQ_ORDERED_EXPLICIT))
4405                 return;
4406
4407         max_active = wq_clamp_max_active(max_active, wq->flags, wq->name);
4408
4409         mutex_lock(&wq->mutex);
4410
4411         wq->flags &= ~__WQ_ORDERED;
4412         wq->saved_max_active = max_active;
4413
4414         for_each_pwq(pwq, wq)
4415                 pwq_adjust_max_active(pwq);
4416
4417         mutex_unlock(&wq->mutex);
4418 }
4419 EXPORT_SYMBOL_GPL(workqueue_set_max_active);
4420
4421 /**
4422  * current_work - retrieve %current task's work struct
4423  *
4424  * Determine if %current task is a workqueue worker and what it's working on.
4425  * Useful to find out the context that the %current task is running in.
4426  *
4427  * Return: work struct if %current task is a workqueue worker, %NULL otherwise.
4428  */
4429 struct work_struct *current_work(void)
4430 {
4431         struct worker *worker = current_wq_worker();
4432
4433         return worker ? worker->current_work : NULL;
4434 }
4435 EXPORT_SYMBOL(current_work);
4436
4437 /**
4438  * current_is_workqueue_rescuer - is %current workqueue rescuer?
4439  *
4440  * Determine whether %current is a workqueue rescuer.  Can be used from
4441  * work functions to determine whether it's being run off the rescuer task.
4442  *
4443  * Return: %true if %current is a workqueue rescuer. %false otherwise.
4444  */
4445 bool current_is_workqueue_rescuer(void)
4446 {
4447         struct worker *worker = current_wq_worker();
4448
4449         return worker && worker->rescue_wq;
4450 }
4451
4452 /**
4453  * workqueue_congested - test whether a workqueue is congested
4454  * @cpu: CPU in question
4455  * @wq: target workqueue
4456  *
4457  * Test whether @wq's cpu workqueue for @cpu is congested.  There is
4458  * no synchronization around this function and the test result is
4459  * unreliable and only useful as advisory hints or for debugging.
4460  *
4461  * If @cpu is WORK_CPU_UNBOUND, the test is performed on the local CPU.
4462  * Note that both per-cpu and unbound workqueues may be associated with
4463  * multiple pool_workqueues which have separate congested states.  A
4464  * workqueue being congested on one CPU doesn't mean the workqueue is also
4465  * contested on other CPUs / NUMA nodes.
4466  *
4467  * Return:
4468  * %true if congested, %false otherwise.
4469  */
4470 bool workqueue_congested(int cpu, struct workqueue_struct *wq)
4471 {
4472         struct pool_workqueue *pwq;
4473         bool ret;
4474
4475         rcu_read_lock();
4476         preempt_disable();
4477
4478         if (cpu == WORK_CPU_UNBOUND)
4479                 cpu = smp_processor_id();
4480
4481         if (!(wq->flags & WQ_UNBOUND))
4482                 pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
4483         else
4484                 pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
4485
4486         ret = !list_empty(&pwq->delayed_works);
4487         preempt_enable();
4488         rcu_read_unlock();
4489
4490         return ret;
4491 }
4492 EXPORT_SYMBOL_GPL(workqueue_congested);
4493
4494 /**
4495  * work_busy - test whether a work is currently pending or running
4496  * @work: the work to be tested
4497  *
4498  * Test whether @work is currently pending or running.  There is no
4499  * synchronization around this function and the test result is
4500  * unreliable and only useful as advisory hints or for debugging.
4501  *
4502  * Return:
4503  * OR'd bitmask of WORK_BUSY_* bits.
4504  */
4505 unsigned int work_busy(struct work_struct *work)
4506 {
4507         struct worker_pool *pool;
4508         unsigned long flags;
4509         unsigned int ret = 0;
4510
4511         if (work_pending(work))
4512                 ret |= WORK_BUSY_PENDING;
4513
4514         rcu_read_lock();
4515         pool = get_work_pool(work);
4516         if (pool) {
4517                 spin_lock_irqsave(&pool->lock, flags);
4518                 if (find_worker_executing_work(pool, work))
4519                         ret |= WORK_BUSY_RUNNING;
4520                 spin_unlock_irqrestore(&pool->lock, flags);
4521         }
4522         rcu_read_unlock();
4523
4524         return ret;
4525 }
4526 EXPORT_SYMBOL_GPL(work_busy);
4527
4528 /**
4529  * set_worker_desc - set description for the current work item
4530  * @fmt: printf-style format string
4531  * @...: arguments for the format string
4532  *
4533  * This function can be called by a running work function to describe what
4534  * the work item is about.  If the worker task gets dumped, this
4535  * information will be printed out together to help debugging.  The
4536  * description can be at most WORKER_DESC_LEN including the trailing '\0'.
4537  */
4538 void set_worker_desc(const char *fmt, ...)
4539 {
4540         struct worker *worker = current_wq_worker();
4541         va_list args;
4542
4543         if (worker) {
4544                 va_start(args, fmt);
4545                 vsnprintf(worker->desc, sizeof(worker->desc), fmt, args);
4546                 va_end(args);
4547         }
4548 }
4549 EXPORT_SYMBOL_GPL(set_worker_desc);
4550
4551 /**
4552  * print_worker_info - print out worker information and description
4553  * @log_lvl: the log level to use when printing
4554  * @task: target task
4555  *
4556  * If @task is a worker and currently executing a work item, print out the
4557  * name of the workqueue being serviced and worker description set with
4558  * set_worker_desc() by the currently executing work item.
4559  *
4560  * This function can be safely called on any task as long as the
4561  * task_struct itself is accessible.  While safe, this function isn't
4562  * synchronized and may print out mixups or garbages of limited length.
4563  */
4564 void print_worker_info(const char *log_lvl, struct task_struct *task)
4565 {
4566         work_func_t *fn = NULL;
4567         char name[WQ_NAME_LEN] = { };
4568         char desc[WORKER_DESC_LEN] = { };
4569         struct pool_workqueue *pwq = NULL;
4570         struct workqueue_struct *wq = NULL;
4571         struct worker *worker;
4572
4573         if (!(task->flags & PF_WQ_WORKER))
4574                 return;
4575
4576         /*
4577          * This function is called without any synchronization and @task
4578          * could be in any state.  Be careful with dereferences.
4579          */
4580         worker = kthread_probe_data(task);
4581
4582         /*
4583          * Carefully copy the associated workqueue's workfn, name and desc.
4584          * Keep the original last '\0' in case the original is garbage.
4585          */
4586         probe_kernel_read(&fn, &worker->current_func, sizeof(fn));
4587         probe_kernel_read(&pwq, &worker->current_pwq, sizeof(pwq));
4588         probe_kernel_read(&wq, &pwq->wq, sizeof(wq));
4589         probe_kernel_read(name, wq->name, sizeof(name) - 1);
4590         probe_kernel_read(desc, worker->desc, sizeof(desc) - 1);
4591
4592         if (fn || name[0] || desc[0]) {
4593                 printk("%sWorkqueue: %s %ps", log_lvl, name, fn);
4594                 if (strcmp(name, desc))
4595                         pr_cont(" (%s)", desc);
4596                 pr_cont("\n");
4597         }
4598 }
4599
4600 static void pr_cont_pool_info(struct worker_pool *pool)
4601 {
4602         pr_cont(" cpus=%*pbl", nr_cpumask_bits, pool->attrs->cpumask);
4603         if (pool->node != NUMA_NO_NODE)
4604                 pr_cont(" node=%d", pool->node);
4605         pr_cont(" flags=0x%x nice=%d", pool->flags, pool->attrs->nice);
4606 }
4607
4608 static void pr_cont_work(bool comma, struct work_struct *work)
4609 {
4610         if (work->func == wq_barrier_func) {
4611                 struct wq_barrier *barr;
4612
4613                 barr = container_of(work, struct wq_barrier, work);
4614
4615                 pr_cont("%s BAR(%d)", comma ? "," : "",
4616                         task_pid_nr(barr->task));
4617         } else {
4618                 pr_cont("%s %ps", comma ? "," : "", work->func);
4619         }
4620 }
4621
4622 static void show_pwq(struct pool_workqueue *pwq)
4623 {
4624         struct worker_pool *pool = pwq->pool;
4625         struct work_struct *work;
4626         struct worker *worker;
4627         bool has_in_flight = false, has_pending = false;
4628         int bkt;
4629
4630         pr_info("  pwq %d:", pool->id);
4631         pr_cont_pool_info(pool);
4632
4633         pr_cont(" active=%d/%d%s\n", pwq->nr_active, pwq->max_active,
4634                 !list_empty(&pwq->mayday_node) ? " MAYDAY" : "");
4635
4636         hash_for_each(pool->busy_hash, bkt, worker, hentry) {
4637                 if (worker->current_pwq == pwq) {
4638                         has_in_flight = true;
4639                         break;
4640                 }
4641         }
4642         if (has_in_flight) {
4643                 bool comma = false;
4644
4645                 pr_info("    in-flight:");
4646                 hash_for_each(pool->busy_hash, bkt, worker, hentry) {
4647                         if (worker->current_pwq != pwq)
4648                                 continue;
4649
4650                         pr_cont("%s %d%s:%ps", comma ? "," : "",
4651                                 task_pid_nr(worker->task),
4652                                 worker == pwq->wq->rescuer ? "(RESCUER)" : "",
4653                                 worker->current_func);
4654                         list_for_each_entry(work, &worker->scheduled, entry)
4655                                 pr_cont_work(false, work);
4656                         comma = true;
4657                 }
4658                 pr_cont("\n");
4659         }
4660
4661         list_for_each_entry(work, &pool->worklist, entry) {
4662                 if (get_work_pwq(work) == pwq) {
4663                         has_pending = true;
4664                         break;
4665                 }
4666         }
4667         if (has_pending) {
4668                 bool comma = false;
4669
4670                 pr_info("    pending:");
4671                 list_for_each_entry(work, &pool->worklist, entry) {
4672                         if (get_work_pwq(work) != pwq)
4673                                 continue;
4674
4675                         pr_cont_work(comma, work);
4676                         comma = !(*work_data_bits(work) & WORK_STRUCT_LINKED);
4677                 }
4678                 pr_cont("\n");
4679         }
4680
4681         if (!list_empty(&pwq->delayed_works)) {
4682                 bool comma = false;
4683
4684                 pr_info("    delayed:");
4685                 list_for_each_entry(work, &pwq->delayed_works, entry) {
4686                         pr_cont_work(comma, work);
4687                         comma = !(*work_data_bits(work) & WORK_STRUCT_LINKED);
4688                 }
4689                 pr_cont("\n");
4690         }
4691 }
4692
4693 /**
4694  * show_workqueue_state - dump workqueue state
4695  *
4696  * Called from a sysrq handler or try_to_freeze_tasks() and prints out
4697  * all busy workqueues and pools.
4698  */
4699 void show_workqueue_state(void)
4700 {
4701         struct workqueue_struct *wq;
4702         struct worker_pool *pool;
4703         unsigned long flags;
4704         int pi;
4705
4706         rcu_read_lock();
4707
4708         pr_info("Showing busy workqueues and worker pools:\n");
4709
4710         list_for_each_entry_rcu(wq, &workqueues, list) {
4711                 struct pool_workqueue *pwq;
4712                 bool idle = true;
4713
4714                 for_each_pwq(pwq, wq) {
4715                         if (pwq->nr_active || !list_empty(&pwq->delayed_works)) {
4716                                 idle = false;
4717                                 break;
4718                         }
4719                 }
4720                 if (idle)
4721                         continue;
4722
4723                 pr_info("workqueue %s: flags=0x%x\n", wq->name, wq->flags);
4724
4725                 for_each_pwq(pwq, wq) {
4726                         spin_lock_irqsave(&pwq->pool->lock, flags);
4727                         if (pwq->nr_active || !list_empty(&pwq->delayed_works))
4728                                 show_pwq(pwq);
4729                         spin_unlock_irqrestore(&pwq->pool->lock, flags);
4730                         /*
4731                          * We could be printing a lot from atomic context, e.g.
4732                          * sysrq-t -> show_workqueue_state(). Avoid triggering
4733                          * hard lockup.
4734                          */
4735                         touch_nmi_watchdog();
4736                 }
4737         }
4738
4739         for_each_pool(pool, pi) {
4740                 struct worker *worker;
4741                 bool first = true;
4742
4743                 spin_lock_irqsave(&pool->lock, flags);
4744                 if (pool->nr_workers == pool->nr_idle)
4745                         goto next_pool;
4746
4747                 pr_info("pool %d:", pool->id);
4748                 pr_cont_pool_info(pool);
4749                 pr_cont(" hung=%us workers=%d",
4750                         jiffies_to_msecs(jiffies - pool->watchdog_ts) / 1000,
4751                         pool->nr_workers);
4752                 if (pool->manager)
4753                         pr_cont(" manager: %d",
4754                                 task_pid_nr(pool->manager->task));
4755                 list_for_each_entry(worker, &pool->idle_list, entry) {
4756                         pr_cont(" %s%d", first ? "idle: " : "",
4757                                 task_pid_nr(worker->task));
4758                         first = false;
4759                 }
4760                 pr_cont("\n");
4761         next_pool:
4762                 spin_unlock_irqrestore(&pool->lock, flags);
4763                 /*
4764                  * We could be printing a lot from atomic context, e.g.
4765                  * sysrq-t -> show_workqueue_state(). Avoid triggering
4766                  * hard lockup.
4767                  */
4768                 touch_nmi_watchdog();
4769         }
4770
4771         rcu_read_unlock();
4772 }
4773
4774 /* used to show worker information through /proc/PID/{comm,stat,status} */
4775 void wq_worker_comm(char *buf, size_t size, struct task_struct *task)
4776 {
4777         int off;
4778
4779         /* always show the actual comm */
4780         off = strscpy(buf, task->comm, size);
4781         if (off < 0)
4782                 return;
4783
4784         /* stabilize PF_WQ_WORKER and worker pool association */
4785         mutex_lock(&wq_pool_attach_mutex);
4786
4787         if (task->flags & PF_WQ_WORKER) {
4788                 struct worker *worker = kthread_data(task);
4789                 struct worker_pool *pool = worker->pool;
4790
4791                 if (pool) {
4792                         spin_lock_irq(&pool->lock);
4793                         /*
4794                          * ->desc tracks information (wq name or
4795                          * set_worker_desc()) for the latest execution.  If
4796                          * current, prepend '+', otherwise '-'.
4797                          */
4798                         if (worker->desc[0] != '\0') {
4799                                 if (worker->current_work)
4800                                         scnprintf(buf + off, size - off, "+%s",
4801                                                   worker->desc);
4802                                 else
4803                                         scnprintf(buf + off, size - off, "-%s",
4804                                                   worker->desc);
4805                         }
4806                         spin_unlock_irq(&pool->lock);
4807                 }
4808         }
4809
4810         mutex_unlock(&wq_pool_attach_mutex);
4811 }
4812
4813 #ifdef CONFIG_SMP
4814
4815 /*
4816  * CPU hotplug.
4817  *
4818  * There are two challenges in supporting CPU hotplug.  Firstly, there
4819  * are a lot of assumptions on strong associations among work, pwq and
4820  * pool which make migrating pending and scheduled works very
4821  * difficult to implement without impacting hot paths.  Secondly,
4822  * worker pools serve mix of short, long and very long running works making
4823  * blocked draining impractical.
4824  *
4825  * This is solved by allowing the pools to be disassociated from the CPU
4826  * running as an unbound one and allowing it to be reattached later if the
4827  * cpu comes back online.
4828  */
4829
4830 static void unbind_workers(int cpu)
4831 {
4832         struct worker_pool *pool;
4833         struct worker *worker;
4834
4835         for_each_cpu_worker_pool(pool, cpu) {
4836                 mutex_lock(&wq_pool_attach_mutex);
4837                 spin_lock_irq(&pool->lock);
4838
4839                 /*
4840                  * We've blocked all attach/detach operations. Make all workers
4841                  * unbound and set DISASSOCIATED.  Before this, all workers
4842                  * except for the ones which are still executing works from
4843                  * before the last CPU down must be on the cpu.  After
4844                  * this, they may become diasporas.
4845                  */
4846                 for_each_pool_worker(worker, pool)
4847                         worker->flags |= WORKER_UNBOUND;
4848
4849                 pool->flags |= POOL_DISASSOCIATED;
4850
4851                 spin_unlock_irq(&pool->lock);
4852                 mutex_unlock(&wq_pool_attach_mutex);
4853
4854                 /*
4855                  * Call schedule() so that we cross rq->lock and thus can
4856                  * guarantee sched callbacks see the %WORKER_UNBOUND flag.
4857                  * This is necessary as scheduler callbacks may be invoked
4858                  * from other cpus.
4859                  */
4860                 schedule();
4861
4862                 /*
4863                  * Sched callbacks are disabled now.  Zap nr_running.
4864                  * After this, nr_running stays zero and need_more_worker()
4865                  * and keep_working() are always true as long as the
4866                  * worklist is not empty.  This pool now behaves as an
4867                  * unbound (in terms of concurrency management) pool which
4868                  * are served by workers tied to the pool.
4869                  */
4870                 atomic_set(&pool->nr_running, 0);
4871
4872                 /*
4873                  * With concurrency management just turned off, a busy
4874                  * worker blocking could lead to lengthy stalls.  Kick off
4875                  * unbound chain execution of currently pending work items.
4876                  */
4877                 spin_lock_irq(&pool->lock);
4878                 wake_up_worker(pool);
4879                 spin_unlock_irq(&pool->lock);
4880         }
4881 }
4882
4883 /**
4884  * rebind_workers - rebind all workers of a pool to the associated CPU
4885  * @pool: pool of interest
4886  *
4887  * @pool->cpu is coming online.  Rebind all workers to the CPU.
4888  */
4889 static void rebind_workers(struct worker_pool *pool)
4890 {
4891         struct worker *worker;
4892
4893         lockdep_assert_held(&wq_pool_attach_mutex);
4894
4895         /*
4896          * Restore CPU affinity of all workers.  As all idle workers should
4897          * be on the run-queue of the associated CPU before any local
4898          * wake-ups for concurrency management happen, restore CPU affinity
4899          * of all workers first and then clear UNBOUND.  As we're called
4900          * from CPU_ONLINE, the following shouldn't fail.
4901          */
4902         for_each_pool_worker(worker, pool)
4903                 WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task,
4904                                                   pool->attrs->cpumask) < 0);
4905
4906         spin_lock_irq(&pool->lock);
4907
4908         pool->flags &= ~POOL_DISASSOCIATED;
4909
4910         for_each_pool_worker(worker, pool) {
4911                 unsigned int worker_flags = worker->flags;
4912
4913                 /*
4914                  * A bound idle worker should actually be on the runqueue
4915                  * of the associated CPU for local wake-ups targeting it to
4916                  * work.  Kick all idle workers so that they migrate to the
4917                  * associated CPU.  Doing this in the same loop as
4918                  * replacing UNBOUND with REBOUND is safe as no worker will
4919                  * be bound before @pool->lock is released.
4920                  */
4921                 if (worker_flags & WORKER_IDLE)
4922                         wake_up_process(worker->task);
4923
4924                 /*
4925                  * We want to clear UNBOUND but can't directly call
4926                  * worker_clr_flags() or adjust nr_running.  Atomically
4927                  * replace UNBOUND with another NOT_RUNNING flag REBOUND.
4928                  * @worker will clear REBOUND using worker_clr_flags() when
4929                  * it initiates the next execution cycle thus restoring
4930                  * concurrency management.  Note that when or whether
4931                  * @worker clears REBOUND doesn't affect correctness.
4932                  *
4933                  * WRITE_ONCE() is necessary because @worker->flags may be
4934                  * tested without holding any lock in
4935                  * wq_worker_running().  Without it, NOT_RUNNING test may
4936                  * fail incorrectly leading to premature concurrency
4937                  * management operations.
4938                  */
4939                 WARN_ON_ONCE(!(worker_flags & WORKER_UNBOUND));
4940                 worker_flags |= WORKER_REBOUND;
4941                 worker_flags &= ~WORKER_UNBOUND;
4942                 WRITE_ONCE(worker->flags, worker_flags);
4943         }
4944
4945         spin_unlock_irq(&pool->lock);
4946 }
4947
4948 /**
4949  * restore_unbound_workers_cpumask - restore cpumask of unbound workers
4950  * @pool: unbound pool of interest
4951  * @cpu: the CPU which is coming up
4952  *
4953  * An unbound pool may end up with a cpumask which doesn't have any online
4954  * CPUs.  When a worker of such pool get scheduled, the scheduler resets
4955  * its cpus_allowed.  If @cpu is in @pool's cpumask which didn't have any
4956  * online CPU before, cpus_allowed of all its workers should be restored.
4957  */
4958 static void restore_unbound_workers_cpumask(struct worker_pool *pool, int cpu)
4959 {
4960         static cpumask_t cpumask;
4961         struct worker *worker;
4962
4963         lockdep_assert_held(&wq_pool_attach_mutex);
4964
4965         /* is @cpu allowed for @pool? */
4966         if (!cpumask_test_cpu(cpu, pool->attrs->cpumask))
4967                 return;
4968
4969         cpumask_and(&cpumask, pool->attrs->cpumask, cpu_online_mask);
4970
4971         /* as we're called from CPU_ONLINE, the following shouldn't fail */
4972         for_each_pool_worker(worker, pool)
4973                 WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task, &cpumask) < 0);
4974 }
4975
4976 int workqueue_prepare_cpu(unsigned int cpu)
4977 {
4978         struct worker_pool *pool;
4979
4980         for_each_cpu_worker_pool(pool, cpu) {
4981                 if (pool->nr_workers)
4982                         continue;
4983                 if (!create_worker(pool))
4984                         return -ENOMEM;
4985         }
4986         return 0;
4987 }
4988
4989 int workqueue_online_cpu(unsigned int cpu)
4990 {
4991         struct worker_pool *pool;
4992         struct workqueue_struct *wq;
4993         int pi;
4994
4995         mutex_lock(&wq_pool_mutex);
4996
4997         for_each_pool(pool, pi) {
4998                 mutex_lock(&wq_pool_attach_mutex);
4999
5000                 if (pool->cpu == cpu)
5001                         rebind_workers(pool);
5002                 else if (pool->cpu < 0)
5003                         restore_unbound_workers_cpumask(pool, cpu);
5004
5005                 mutex_unlock(&wq_pool_attach_mutex);
5006         }
5007
5008         /* update NUMA affinity of unbound workqueues */
5009         list_for_each_entry(wq, &workqueues, list)
5010                 wq_update_unbound_numa(wq, cpu, true);
5011
5012         mutex_unlock(&wq_pool_mutex);
5013         return 0;
5014 }
5015
5016 int workqueue_offline_cpu(unsigned int cpu)
5017 {
5018         struct workqueue_struct *wq;
5019
5020         /* unbinding per-cpu workers should happen on the local CPU */
5021         if (WARN_ON(cpu != smp_processor_id()))
5022                 return -1;
5023
5024         unbind_workers(cpu);
5025
5026         /* update NUMA affinity of unbound workqueues */
5027         mutex_lock(&wq_pool_mutex);
5028         list_for_each_entry(wq, &workqueues, list)
5029                 wq_update_unbound_numa(wq, cpu, false);
5030         mutex_unlock(&wq_pool_mutex);
5031
5032         return 0;
5033 }
5034
5035 struct work_for_cpu {
5036         struct work_struct work;
5037         long (*fn)(void *);
5038         void *arg;
5039         long ret;
5040 };
5041
5042 static void work_for_cpu_fn(struct work_struct *work)
5043 {
5044         struct work_for_cpu *wfc = container_of(work, struct work_for_cpu, work);
5045
5046         wfc->ret = wfc->fn(wfc->arg);
5047 }
5048
5049 /**
5050  * work_on_cpu - run a function in thread context on a particular cpu
5051  * @cpu: the cpu to run on
5052  * @fn: the function to run
5053  * @arg: the function arg
5054  *
5055  * It is up to the caller to ensure that the cpu doesn't go offline.
5056  * The caller must not hold any locks which would prevent @fn from completing.
5057  *
5058  * Return: The value @fn returns.
5059  */
5060 long work_on_cpu(int cpu, long (*fn)(void *), void *arg)
5061 {
5062         struct work_for_cpu wfc = { .fn = fn, .arg = arg };
5063
5064         INIT_WORK_ONSTACK(&wfc.work, work_for_cpu_fn);
5065         schedule_work_on(cpu, &wfc.work);
5066         flush_work(&wfc.work);
5067         destroy_work_on_stack(&wfc.work);
5068         return wfc.ret;
5069 }
5070 EXPORT_SYMBOL_GPL(work_on_cpu);
5071
5072 /**
5073  * work_on_cpu_safe - run a function in thread context on a particular cpu
5074  * @cpu: the cpu to run on
5075  * @fn:  the function to run
5076  * @arg: the function argument
5077  *
5078  * Disables CPU hotplug and calls work_on_cpu(). The caller must not hold
5079  * any locks which would prevent @fn from completing.
5080  *
5081  * Return: The value @fn returns.
5082  */
5083 long work_on_cpu_safe(int cpu, long (*fn)(void *), void *arg)
5084 {
5085         long ret = -ENODEV;
5086
5087         get_online_cpus();
5088         if (cpu_online(cpu))
5089                 ret = work_on_cpu(cpu, fn, arg);
5090         put_online_cpus();
5091         return ret;
5092 }
5093 EXPORT_SYMBOL_GPL(work_on_cpu_safe);
5094 #endif /* CONFIG_SMP */
5095
5096 #ifdef CONFIG_FREEZER
5097
5098 /**
5099  * freeze_workqueues_begin - begin freezing workqueues
5100  *
5101  * Start freezing workqueues.  After this function returns, all freezable
5102  * workqueues will queue new works to their delayed_works list instead of
5103  * pool->worklist.
5104  *
5105  * CONTEXT:
5106  * Grabs and releases wq_pool_mutex, wq->mutex and pool->lock's.
5107  */
5108 void freeze_workqueues_begin(void)
5109 {
5110         struct workqueue_struct *wq;
5111         struct pool_workqueue *pwq;
5112
5113         mutex_lock(&wq_pool_mutex);
5114
5115         WARN_ON_ONCE(workqueue_freezing);
5116         workqueue_freezing = true;
5117
5118         list_for_each_entry(wq, &workqueues, list) {
5119                 mutex_lock(&wq->mutex);
5120                 for_each_pwq(pwq, wq)
5121                         pwq_adjust_max_active(pwq);
5122                 mutex_unlock(&wq->mutex);
5123         }
5124
5125         mutex_unlock(&wq_pool_mutex);
5126 }
5127
5128 /**
5129  * freeze_workqueues_busy - are freezable workqueues still busy?
5130  *
5131  * Check whether freezing is complete.  This function must be called
5132  * between freeze_workqueues_begin() and thaw_workqueues().
5133  *
5134  * CONTEXT:
5135  * Grabs and releases wq_pool_mutex.
5136  *
5137  * Return:
5138  * %true if some freezable workqueues are still busy.  %false if freezing
5139  * is complete.
5140  */
5141 bool freeze_workqueues_busy(void)
5142 {
5143         bool busy = false;
5144         struct workqueue_struct *wq;
5145         struct pool_workqueue *pwq;
5146
5147         mutex_lock(&wq_pool_mutex);
5148
5149         WARN_ON_ONCE(!workqueue_freezing);
5150
5151         list_for_each_entry(wq, &workqueues, list) {
5152                 if (!(wq->flags & WQ_FREEZABLE))
5153                         continue;
5154                 /*
5155                  * nr_active is monotonically decreasing.  It's safe
5156                  * to peek without lock.
5157                  */
5158                 rcu_read_lock();
5159                 for_each_pwq(pwq, wq) {
5160                         WARN_ON_ONCE(pwq->nr_active < 0);
5161                         if (pwq->nr_active) {
5162                                 busy = true;
5163                                 rcu_read_unlock();
5164                                 goto out_unlock;
5165                         }
5166                 }
5167                 rcu_read_unlock();
5168         }
5169 out_unlock:
5170         mutex_unlock(&wq_pool_mutex);
5171         return busy;
5172 }
5173
5174 /**
5175  * thaw_workqueues - thaw workqueues
5176  *
5177  * Thaw workqueues.  Normal queueing is restored and all collected
5178  * frozen works are transferred to their respective pool worklists.
5179  *
5180  * CONTEXT:
5181  * Grabs and releases wq_pool_mutex, wq->mutex and pool->lock's.
5182  */
5183 void thaw_workqueues(void)
5184 {
5185         struct workqueue_struct *wq;
5186         struct pool_workqueue *pwq;
5187
5188         mutex_lock(&wq_pool_mutex);
5189
5190         if (!workqueue_freezing)
5191                 goto out_unlock;
5192
5193         workqueue_freezing = false;
5194
5195         /* restore max_active and repopulate worklist */
5196         list_for_each_entry(wq, &workqueues, list) {
5197                 mutex_lock(&wq->mutex);
5198                 for_each_pwq(pwq, wq)
5199                         pwq_adjust_max_active(pwq);
5200                 mutex_unlock(&wq->mutex);
5201         }
5202
5203 out_unlock:
5204         mutex_unlock(&wq_pool_mutex);
5205 }
5206 #endif /* CONFIG_FREEZER */
5207
5208 static int workqueue_apply_unbound_cpumask(void)
5209 {
5210         LIST_HEAD(ctxs);
5211         int ret = 0;
5212         struct workqueue_struct *wq;
5213         struct apply_wqattrs_ctx *ctx, *n;
5214
5215         lockdep_assert_held(&wq_pool_mutex);
5216
5217         list_for_each_entry(wq, &workqueues, list) {
5218                 if (!(wq->flags & WQ_UNBOUND))
5219                         continue;
5220                 /* creating multiple pwqs breaks ordering guarantee */
5221                 if (wq->flags & __WQ_ORDERED)
5222                         continue;
5223
5224                 ctx = apply_wqattrs_prepare(wq, wq->unbound_attrs);
5225                 if (!ctx) {
5226                         ret = -ENOMEM;
5227                         break;
5228                 }
5229
5230                 list_add_tail(&ctx->list, &ctxs);
5231         }
5232
5233         list_for_each_entry_safe(ctx, n, &ctxs, list) {
5234                 if (!ret)
5235                         apply_wqattrs_commit(ctx);
5236                 apply_wqattrs_cleanup(ctx);
5237         }
5238
5239         return ret;
5240 }
5241
5242 /**
5243  *  workqueue_set_unbound_cpumask - Set the low-level unbound cpumask
5244  *  @cpumask: the cpumask to set
5245  *
5246  *  The low-level workqueues cpumask is a global cpumask that limits
5247  *  the affinity of all unbound workqueues.  This function check the @cpumask
5248  *  and apply it to all unbound workqueues and updates all pwqs of them.
5249  *
5250  *  Retun:      0       - Success
5251  *              -EINVAL - Invalid @cpumask
5252  *              -ENOMEM - Failed to allocate memory for attrs or pwqs.
5253  */
5254 int workqueue_set_unbound_cpumask(cpumask_var_t cpumask)
5255 {
5256         int ret = -EINVAL;
5257         cpumask_var_t saved_cpumask;
5258
5259         if (!zalloc_cpumask_var(&saved_cpumask, GFP_KERNEL))
5260                 return -ENOMEM;
5261
5262         /*
5263          * Not excluding isolated cpus on purpose.
5264          * If the user wishes to include them, we allow that.
5265          */
5266         cpumask_and(cpumask, cpumask, cpu_possible_mask);
5267         if (!cpumask_empty(cpumask)) {
5268                 apply_wqattrs_lock();
5269
5270                 /* save the old wq_unbound_cpumask. */
5271                 cpumask_copy(saved_cpumask, wq_unbound_cpumask);
5272
5273                 /* update wq_unbound_cpumask at first and apply it to wqs. */
5274                 cpumask_copy(wq_unbound_cpumask, cpumask);
5275                 ret = workqueue_apply_unbound_cpumask();
5276
5277                 /* restore the wq_unbound_cpumask when failed. */
5278                 if (ret < 0)
5279                         cpumask_copy(wq_unbound_cpumask, saved_cpumask);
5280
5281                 apply_wqattrs_unlock();
5282         }
5283
5284         free_cpumask_var(saved_cpumask);
5285         return ret;
5286 }
5287
5288 #ifdef CONFIG_SYSFS
5289 /*
5290  * Workqueues with WQ_SYSFS flag set is visible to userland via
5291  * /sys/bus/workqueue/devices/WQ_NAME.  All visible workqueues have the
5292  * following attributes.
5293  *
5294  *  per_cpu     RO bool : whether the workqueue is per-cpu or unbound
5295  *  max_active  RW int  : maximum number of in-flight work items
5296  *
5297  * Unbound workqueues have the following extra attributes.
5298  *
5299  *  pool_ids    RO int  : the associated pool IDs for each node
5300  *  nice        RW int  : nice value of the workers
5301  *  cpumask     RW mask : bitmask of allowed CPUs for the workers
5302  *  numa        RW bool : whether enable NUMA affinity
5303  */
5304 struct wq_device {
5305         struct workqueue_struct         *wq;
5306         struct device                   dev;
5307 };
5308
5309 static struct workqueue_struct *dev_to_wq(struct device *dev)
5310 {
5311         struct wq_device *wq_dev = container_of(dev, struct wq_device, dev);
5312
5313         return wq_dev->wq;
5314 }
5315
5316 static ssize_t per_cpu_show(struct device *dev, struct device_attribute *attr,
5317                             char *buf)
5318 {
5319         struct workqueue_struct *wq = dev_to_wq(dev);
5320
5321         return scnprintf(buf, PAGE_SIZE, "%d\n", (bool)!(wq->flags & WQ_UNBOUND));
5322 }
5323 static DEVICE_ATTR_RO(per_cpu);
5324
5325 static ssize_t max_active_show(struct device *dev,
5326                                struct device_attribute *attr, char *buf)
5327 {
5328         struct workqueue_struct *wq = dev_to_wq(dev);
5329
5330         return scnprintf(buf, PAGE_SIZE, "%d\n", wq->saved_max_active);
5331 }
5332
5333 static ssize_t max_active_store(struct device *dev,
5334                                 struct device_attribute *attr, const char *buf,
5335                                 size_t count)
5336 {
5337         struct workqueue_struct *wq = dev_to_wq(dev);
5338         int val;
5339
5340         if (sscanf(buf, "%d", &val) != 1 || val <= 0)
5341                 return -EINVAL;
5342
5343         workqueue_set_max_active(wq, val);
5344         return count;
5345 }
5346 static DEVICE_ATTR_RW(max_active);
5347
5348 static struct attribute *wq_sysfs_attrs[] = {
5349         &dev_attr_per_cpu.attr,
5350         &dev_attr_max_active.attr,
5351         NULL,
5352 };
5353 ATTRIBUTE_GROUPS(wq_sysfs);
5354
5355 static ssize_t wq_pool_ids_show(struct device *dev,
5356                                 struct device_attribute *attr, char *buf)
5357 {
5358         struct workqueue_struct *wq = dev_to_wq(dev);
5359         const char *delim = "";
5360         int node, written = 0;
5361
5362         get_online_cpus();
5363         rcu_read_lock();
5364         for_each_node(node) {
5365                 written += scnprintf(buf + written, PAGE_SIZE - written,
5366                                      "%s%d:%d", delim, node,
5367                                      unbound_pwq_by_node(wq, node)->pool->id);
5368                 delim = " ";
5369         }
5370         written += scnprintf(buf + written, PAGE_SIZE - written, "\n");
5371         rcu_read_unlock();
5372         put_online_cpus();
5373
5374         return written;
5375 }
5376
5377 static ssize_t wq_nice_show(struct device *dev, struct device_attribute *attr,
5378                             char *buf)
5379 {
5380         struct workqueue_struct *wq = dev_to_wq(dev);
5381         int written;
5382
5383         mutex_lock(&wq->mutex);
5384         written = scnprintf(buf, PAGE_SIZE, "%d\n", wq->unbound_attrs->nice);
5385         mutex_unlock(&wq->mutex);
5386
5387         return written;
5388 }
5389
5390 /* prepare workqueue_attrs for sysfs store operations */
5391 static struct workqueue_attrs *wq_sysfs_prep_attrs(struct workqueue_struct *wq)
5392 {
5393         struct workqueue_attrs *attrs;
5394
5395         lockdep_assert_held(&wq_pool_mutex);
5396
5397         attrs = alloc_workqueue_attrs(GFP_KERNEL);
5398         if (!attrs)
5399                 return NULL;
5400
5401         copy_workqueue_attrs(attrs, wq->unbound_attrs);
5402         return attrs;
5403 }
5404
5405 static ssize_t wq_nice_store(struct device *dev, struct device_attribute *attr,
5406                              const char *buf, size_t count)
5407 {
5408         struct workqueue_struct *wq = dev_to_wq(dev);
5409         struct workqueue_attrs *attrs;
5410         int ret = -ENOMEM;
5411
5412         apply_wqattrs_lock();
5413
5414         attrs = wq_sysfs_prep_attrs(wq);
5415         if (!attrs)
5416                 goto out_unlock;
5417
5418         if (sscanf(buf, "%d", &attrs->nice) == 1 &&
5419             attrs->nice >= MIN_NICE && attrs->nice <= MAX_NICE)
5420                 ret = apply_workqueue_attrs_locked(wq, attrs);
5421         else
5422                 ret = -EINVAL;
5423
5424 out_unlock:
5425         apply_wqattrs_unlock();
5426         free_workqueue_attrs(attrs);
5427         return ret ?: count;
5428 }
5429
5430 static ssize_t wq_cpumask_show(struct device *dev,
5431                                struct device_attribute *attr, char *buf)
5432 {
5433         struct workqueue_struct *wq = dev_to_wq(dev);
5434         int written;
5435
5436         mutex_lock(&wq->mutex);
5437         written = scnprintf(buf, PAGE_SIZE, "%*pb\n",
5438                             cpumask_pr_args(wq->unbound_attrs->cpumask));
5439         mutex_unlock(&wq->mutex);
5440         return written;
5441 }
5442
5443 static ssize_t wq_cpumask_store(struct device *dev,
5444                                 struct device_attribute *attr,
5445                                 const char *buf, size_t count)
5446 {
5447         struct workqueue_struct *wq = dev_to_wq(dev);
5448         struct workqueue_attrs *attrs;
5449         int ret = -ENOMEM;
5450
5451         apply_wqattrs_lock();
5452
5453         attrs = wq_sysfs_prep_attrs(wq);
5454         if (!attrs)
5455                 goto out_unlock;
5456
5457         ret = cpumask_parse(buf, attrs->cpumask);
5458         if (!ret)
5459                 ret = apply_workqueue_attrs_locked(wq, attrs);
5460
5461 out_unlock:
5462         apply_wqattrs_unlock();
5463         free_workqueue_attrs(attrs);
5464         return ret ?: count;
5465 }
5466
5467 static ssize_t wq_numa_show(struct device *dev, struct device_attribute *attr,
5468                             char *buf)
5469 {
5470         struct workqueue_struct *wq = dev_to_wq(dev);
5471         int written;
5472
5473         mutex_lock(&wq->mutex);
5474         written = scnprintf(buf, PAGE_SIZE, "%d\n",
5475                             !wq->unbound_attrs->no_numa);
5476         mutex_unlock(&wq->mutex);
5477
5478         return written;
5479 }
5480
5481 static ssize_t wq_numa_store(struct device *dev, struct device_attribute *attr,
5482                              const char *buf, size_t count)
5483 {
5484         struct workqueue_struct *wq = dev_to_wq(dev);
5485         struct workqueue_attrs *attrs;
5486         int v, ret = -ENOMEM;
5487
5488         apply_wqattrs_lock();
5489
5490         attrs = wq_sysfs_prep_attrs(wq);
5491         if (!attrs)
5492                 goto out_unlock;
5493
5494         ret = -EINVAL;
5495         if (sscanf(buf, "%d", &v) == 1) {
5496                 attrs->no_numa = !v;
5497                 ret = apply_workqueue_attrs_locked(wq, attrs);
5498         }
5499
5500 out_unlock:
5501         apply_wqattrs_unlock();
5502         free_workqueue_attrs(attrs);
5503         return ret ?: count;
5504 }
5505
5506 static struct device_attribute wq_sysfs_unbound_attrs[] = {
5507         __ATTR(pool_ids, 0444, wq_pool_ids_show, NULL),
5508         __ATTR(nice, 0644, wq_nice_show, wq_nice_store),
5509         __ATTR(cpumask, 0644, wq_cpumask_show, wq_cpumask_store),
5510         __ATTR(numa, 0644, wq_numa_show, wq_numa_store),
5511         __ATTR_NULL,
5512 };
5513
5514 static struct bus_type wq_subsys = {
5515         .name                           = "workqueue",
5516         .dev_groups                     = wq_sysfs_groups,
5517 };
5518
5519 static ssize_t wq_unbound_cpumask_show(struct device *dev,
5520                 struct device_attribute *attr, char *buf)
5521 {
5522         int written;
5523
5524         mutex_lock(&wq_pool_mutex);
5525         written = scnprintf(buf, PAGE_SIZE, "%*pb\n",
5526                             cpumask_pr_args(wq_unbound_cpumask));
5527         mutex_unlock(&wq_pool_mutex);
5528
5529         return written;
5530 }
5531
5532 static ssize_t wq_unbound_cpumask_store(struct device *dev,
5533                 struct device_attribute *attr, const char *buf, size_t count)
5534 {
5535         cpumask_var_t cpumask;
5536         int ret;
5537
5538         if (!zalloc_cpumask_var(&cpumask, GFP_KERNEL))
5539                 return -ENOMEM;
5540
5541         ret = cpumask_parse(buf, cpumask);
5542         if (!ret)
5543                 ret = workqueue_set_unbound_cpumask(cpumask);
5544
5545         free_cpumask_var(cpumask);
5546         return ret ? ret : count;
5547 }
5548
5549 static struct device_attribute wq_sysfs_cpumask_attr =
5550         __ATTR(cpumask, 0644, wq_unbound_cpumask_show,
5551                wq_unbound_cpumask_store);
5552
5553 static int __init wq_sysfs_init(void)
5554 {
5555         int err;
5556
5557         err = subsys_virtual_register(&wq_subsys, NULL);
5558         if (err)
5559                 return err;
5560
5561         return device_create_file(wq_subsys.dev_root, &wq_sysfs_cpumask_attr);
5562 }
5563 core_initcall(wq_sysfs_init);
5564
5565 static void wq_device_release(struct device *dev)
5566 {
5567         struct wq_device *wq_dev = container_of(dev, struct wq_device, dev);
5568
5569         kfree(wq_dev);
5570 }
5571
5572 /**
5573  * workqueue_sysfs_register - make a workqueue visible in sysfs
5574  * @wq: the workqueue to register
5575  *
5576  * Expose @wq in sysfs under /sys/bus/workqueue/devices.
5577  * alloc_workqueue*() automatically calls this function if WQ_SYSFS is set
5578  * which is the preferred method.
5579  *
5580  * Workqueue user should use this function directly iff it wants to apply
5581  * workqueue_attrs before making the workqueue visible in sysfs; otherwise,
5582  * apply_workqueue_attrs() may race against userland updating the
5583  * attributes.
5584  *
5585  * Return: 0 on success, -errno on failure.
5586  */
5587 int workqueue_sysfs_register(struct workqueue_struct *wq)
5588 {
5589         struct wq_device *wq_dev;
5590         int ret;
5591
5592         /*
5593          * Adjusting max_active or creating new pwqs by applying
5594          * attributes breaks ordering guarantee.  Disallow exposing ordered
5595          * workqueues.
5596          */
5597         if (WARN_ON(wq->flags & __WQ_ORDERED_EXPLICIT))
5598                 return -EINVAL;
5599
5600         wq->wq_dev = wq_dev = kzalloc(sizeof(*wq_dev), GFP_KERNEL);
5601         if (!wq_dev)
5602                 return -ENOMEM;
5603
5604         wq_dev->wq = wq;
5605         wq_dev->dev.bus = &wq_subsys;
5606         wq_dev->dev.release = wq_device_release;
5607         dev_set_name(&wq_dev->dev, "%s", wq->name);
5608
5609         /*
5610          * unbound_attrs are created separately.  Suppress uevent until
5611          * everything is ready.
5612          */
5613         dev_set_uevent_suppress(&wq_dev->dev, true);
5614
5615         ret = device_register(&wq_dev->dev);
5616         if (ret) {
5617                 put_device(&wq_dev->dev);
5618                 wq->wq_dev = NULL;
5619                 return ret;
5620         }
5621
5622         if (wq->flags & WQ_UNBOUND) {
5623                 struct device_attribute *attr;
5624
5625                 for (attr = wq_sysfs_unbound_attrs; attr->attr.name; attr++) {
5626                         ret = device_create_file(&wq_dev->dev, attr);
5627                         if (ret) {
5628                                 device_unregister(&wq_dev->dev);
5629                                 wq->wq_dev = NULL;
5630                                 return ret;
5631                         }
5632                 }
5633         }
5634
5635         dev_set_uevent_suppress(&wq_dev->dev, false);
5636         kobject_uevent(&wq_dev->dev.kobj, KOBJ_ADD);
5637         return 0;
5638 }
5639
5640 /**
5641  * workqueue_sysfs_unregister - undo workqueue_sysfs_register()
5642  * @wq: the workqueue to unregister
5643  *
5644  * If @wq is registered to sysfs by workqueue_sysfs_register(), unregister.
5645  */
5646 static void workqueue_sysfs_unregister(struct workqueue_struct *wq)
5647 {
5648         struct wq_device *wq_dev = wq->wq_dev;
5649
5650         if (!wq->wq_dev)
5651                 return;
5652
5653         wq->wq_dev = NULL;
5654         device_unregister(&wq_dev->dev);
5655 }
5656 #else   /* CONFIG_SYSFS */
5657 static void workqueue_sysfs_unregister(struct workqueue_struct *wq)     { }
5658 #endif  /* CONFIG_SYSFS */
5659
5660 /*
5661  * Workqueue watchdog.
5662  *
5663  * Stall may be caused by various bugs - missing WQ_MEM_RECLAIM, illegal
5664  * flush dependency, a concurrency managed work item which stays RUNNING
5665  * indefinitely.  Workqueue stalls can be very difficult to debug as the
5666  * usual warning mechanisms don't trigger and internal workqueue state is
5667  * largely opaque.
5668  *
5669  * Workqueue watchdog monitors all worker pools periodically and dumps
5670  * state if some pools failed to make forward progress for a while where
5671  * forward progress is defined as the first item on ->worklist changing.
5672  *
5673  * This mechanism is controlled through the kernel parameter
5674  * "workqueue.watchdog_thresh" which can be updated at runtime through the
5675  * corresponding sysfs parameter file.
5676  */
5677 #ifdef CONFIG_WQ_WATCHDOG
5678
5679 static unsigned long wq_watchdog_thresh = 30;
5680 static struct timer_list wq_watchdog_timer;
5681
5682 static unsigned long wq_watchdog_touched = INITIAL_JIFFIES;
5683 static DEFINE_PER_CPU(unsigned long, wq_watchdog_touched_cpu) = INITIAL_JIFFIES;
5684
5685 static void wq_watchdog_reset_touched(void)
5686 {
5687         int cpu;
5688
5689         wq_watchdog_touched = jiffies;
5690         for_each_possible_cpu(cpu)
5691                 per_cpu(wq_watchdog_touched_cpu, cpu) = jiffies;
5692 }
5693
5694 static void wq_watchdog_timer_fn(struct timer_list *unused)
5695 {
5696         unsigned long thresh = READ_ONCE(wq_watchdog_thresh) * HZ;
5697         bool lockup_detected = false;
5698         struct worker_pool *pool;
5699         int pi;
5700
5701         if (!thresh)
5702                 return;
5703
5704         rcu_read_lock();
5705
5706         for_each_pool(pool, pi) {
5707                 unsigned long pool_ts, touched, ts;
5708
5709                 if (list_empty(&pool->worklist))
5710                         continue;
5711
5712                 /* get the latest of pool and touched timestamps */
5713                 pool_ts = READ_ONCE(pool->watchdog_ts);
5714                 touched = READ_ONCE(wq_watchdog_touched);
5715
5716                 if (time_after(pool_ts, touched))
5717                         ts = pool_ts;
5718                 else
5719                         ts = touched;
5720
5721                 if (pool->cpu >= 0) {
5722                         unsigned long cpu_touched =
5723                                 READ_ONCE(per_cpu(wq_watchdog_touched_cpu,
5724                                                   pool->cpu));
5725                         if (time_after(cpu_touched, ts))
5726                                 ts = cpu_touched;
5727                 }
5728
5729                 /* did we stall? */
5730                 if (time_after(jiffies, ts + thresh)) {
5731                         lockup_detected = true;
5732                         pr_emerg("BUG: workqueue lockup - pool");
5733                         pr_cont_pool_info(pool);
5734                         pr_cont(" stuck for %us!\n",
5735                                 jiffies_to_msecs(jiffies - pool_ts) / 1000);
5736                 }
5737         }
5738
5739         rcu_read_unlock();
5740
5741         if (lockup_detected)
5742                 show_workqueue_state();
5743
5744         wq_watchdog_reset_touched();
5745         mod_timer(&wq_watchdog_timer, jiffies + thresh);
5746 }
5747
5748 notrace void wq_watchdog_touch(int cpu)
5749 {
5750         if (cpu >= 0)
5751                 per_cpu(wq_watchdog_touched_cpu, cpu) = jiffies;
5752         else
5753                 wq_watchdog_touched = jiffies;
5754 }
5755
5756 static void wq_watchdog_set_thresh(unsigned long thresh)
5757 {
5758         wq_watchdog_thresh = 0;
5759         del_timer_sync(&wq_watchdog_timer);
5760
5761         if (thresh) {
5762                 wq_watchdog_thresh = thresh;
5763                 wq_watchdog_reset_touched();
5764                 mod_timer(&wq_watchdog_timer, jiffies + thresh * HZ);
5765         }
5766 }
5767
5768 static int wq_watchdog_param_set_thresh(const char *val,
5769                                         const struct kernel_param *kp)
5770 {
5771         unsigned long thresh;
5772         int ret;
5773
5774         ret = kstrtoul(val, 0, &thresh);
5775         if (ret)
5776                 return ret;
5777
5778         if (system_wq)
5779                 wq_watchdog_set_thresh(thresh);
5780         else
5781                 wq_watchdog_thresh = thresh;
5782
5783         return 0;
5784 }
5785
5786 static const struct kernel_param_ops wq_watchdog_thresh_ops = {
5787         .set    = wq_watchdog_param_set_thresh,
5788         .get    = param_get_ulong,
5789 };
5790
5791 module_param_cb(watchdog_thresh, &wq_watchdog_thresh_ops, &wq_watchdog_thresh,
5792                 0644);
5793
5794 static void wq_watchdog_init(void)
5795 {
5796         timer_setup(&wq_watchdog_timer, wq_watchdog_timer_fn, TIMER_DEFERRABLE);
5797         wq_watchdog_set_thresh(wq_watchdog_thresh);
5798 }
5799
5800 #else   /* CONFIG_WQ_WATCHDOG */
5801
5802 static inline void wq_watchdog_init(void) { }
5803
5804 #endif  /* CONFIG_WQ_WATCHDOG */
5805
5806 static void __init wq_numa_init(void)
5807 {
5808         cpumask_var_t *tbl;
5809         int node, cpu;
5810
5811         if (num_possible_nodes() <= 1)
5812                 return;
5813
5814         if (wq_disable_numa) {
5815                 pr_info("workqueue: NUMA affinity support disabled\n");
5816                 return;
5817         }
5818
5819         wq_update_unbound_numa_attrs_buf = alloc_workqueue_attrs(GFP_KERNEL);
5820         BUG_ON(!wq_update_unbound_numa_attrs_buf);
5821
5822         /*
5823          * We want masks of possible CPUs of each node which isn't readily
5824          * available.  Build one from cpu_to_node() which should have been
5825          * fully initialized by now.
5826          */
5827         tbl = kcalloc(nr_node_ids, sizeof(tbl[0]), GFP_KERNEL);
5828         BUG_ON(!tbl);
5829
5830         for_each_node(node)
5831                 BUG_ON(!zalloc_cpumask_var_node(&tbl[node], GFP_KERNEL,
5832                                 node_online(node) ? node : NUMA_NO_NODE));
5833
5834         for_each_possible_cpu(cpu) {
5835                 node = cpu_to_node(cpu);
5836                 if (WARN_ON(node == NUMA_NO_NODE)) {
5837                         pr_warn("workqueue: NUMA node mapping not available for cpu%d, disabling NUMA support\n", cpu);
5838                         /* happens iff arch is bonkers, let's just proceed */
5839                         return;
5840                 }
5841                 cpumask_set_cpu(cpu, tbl[node]);
5842         }
5843
5844         wq_numa_possible_cpumask = tbl;
5845         wq_numa_enabled = true;
5846 }
5847
5848 /**
5849  * workqueue_init_early - early init for workqueue subsystem
5850  *
5851  * This is the first half of two-staged workqueue subsystem initialization
5852  * and invoked as soon as the bare basics - memory allocation, cpumasks and
5853  * idr are up.  It sets up all the data structures and system workqueues
5854  * and allows early boot code to create workqueues and queue/cancel work
5855  * items.  Actual work item execution starts only after kthreads can be
5856  * created and scheduled right before early initcalls.
5857  */
5858 int __init workqueue_init_early(void)
5859 {
5860         int std_nice[NR_STD_WORKER_POOLS] = { 0, HIGHPRI_NICE_LEVEL };
5861         int hk_flags = HK_FLAG_DOMAIN | HK_FLAG_WQ;
5862         int i, cpu;
5863
5864         WARN_ON(__alignof__(struct pool_workqueue) < __alignof__(long long));
5865
5866         BUG_ON(!alloc_cpumask_var(&wq_unbound_cpumask, GFP_KERNEL));
5867         cpumask_copy(wq_unbound_cpumask, housekeeping_cpumask(hk_flags));
5868
5869         pwq_cache = KMEM_CACHE(pool_workqueue, SLAB_PANIC);
5870
5871         /* initialize CPU pools */
5872         for_each_possible_cpu(cpu) {
5873                 struct worker_pool *pool;
5874
5875                 i = 0;
5876                 for_each_cpu_worker_pool(pool, cpu) {
5877                         BUG_ON(init_worker_pool(pool));
5878                         pool->cpu = cpu;
5879                         cpumask_copy(pool->attrs->cpumask, cpumask_of(cpu));
5880                         pool->attrs->nice = std_nice[i++];
5881                         pool->node = cpu_to_node(cpu);
5882
5883                         /* alloc pool ID */
5884                         mutex_lock(&wq_pool_mutex);
5885                         BUG_ON(worker_pool_assign_id(pool));
5886                         mutex_unlock(&wq_pool_mutex);
5887                 }
5888         }
5889
5890         /* create default unbound and ordered wq attrs */
5891         for (i = 0; i < NR_STD_WORKER_POOLS; i++) {
5892                 struct workqueue_attrs *attrs;
5893
5894                 BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
5895                 attrs->nice = std_nice[i];
5896                 unbound_std_wq_attrs[i] = attrs;
5897
5898                 /*
5899                  * An ordered wq should have only one pwq as ordering is
5900                  * guaranteed by max_active which is enforced by pwqs.
5901                  * Turn off NUMA so that dfl_pwq is used for all nodes.
5902                  */
5903                 BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
5904                 attrs->nice = std_nice[i];
5905                 attrs->no_numa = true;
5906                 ordered_wq_attrs[i] = attrs;
5907         }
5908
5909         system_wq = alloc_workqueue("events", 0, 0);
5910         system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
5911         system_long_wq = alloc_workqueue("events_long", 0, 0);
5912         system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
5913                                             WQ_UNBOUND_MAX_ACTIVE);
5914         system_freezable_wq = alloc_workqueue("events_freezable",
5915                                               WQ_FREEZABLE, 0);
5916         system_power_efficient_wq = alloc_workqueue("events_power_efficient",
5917                                               WQ_POWER_EFFICIENT, 0);
5918         system_freezable_power_efficient_wq = alloc_workqueue("events_freezable_power_efficient",
5919                                               WQ_FREEZABLE | WQ_POWER_EFFICIENT,
5920                                               0);
5921         BUG_ON(!system_wq || !system_highpri_wq || !system_long_wq ||
5922                !system_unbound_wq || !system_freezable_wq ||
5923                !system_power_efficient_wq ||
5924                !system_freezable_power_efficient_wq);
5925
5926         return 0;
5927 }
5928
5929 /**
5930  * workqueue_init - bring workqueue subsystem fully online
5931  *
5932  * This is the latter half of two-staged workqueue subsystem initialization
5933  * and invoked as soon as kthreads can be created and scheduled.
5934  * Workqueues have been created and work items queued on them, but there
5935  * are no kworkers executing the work items yet.  Populate the worker pools
5936  * with the initial workers and enable future kworker creations.
5937  */
5938 int __init workqueue_init(void)
5939 {
5940         struct workqueue_struct *wq;
5941         struct worker_pool *pool;
5942         int cpu, bkt;
5943
5944         /*
5945          * It'd be simpler to initialize NUMA in workqueue_init_early() but
5946          * CPU to node mapping may not be available that early on some
5947          * archs such as power and arm64.  As per-cpu pools created
5948          * previously could be missing node hint and unbound pools NUMA
5949          * affinity, fix them up.
5950          *
5951          * Also, while iterating workqueues, create rescuers if requested.
5952          */
5953         wq_numa_init();
5954
5955         mutex_lock(&wq_pool_mutex);
5956
5957         for_each_possible_cpu(cpu) {
5958                 for_each_cpu_worker_pool(pool, cpu) {
5959                         pool->node = cpu_to_node(cpu);
5960                 }
5961         }
5962
5963         list_for_each_entry(wq, &workqueues, list) {
5964                 wq_update_unbound_numa(wq, smp_processor_id(), true);
5965                 WARN(init_rescuer(wq),
5966                      "workqueue: failed to create early rescuer for %s",
5967                      wq->name);
5968         }
5969
5970         mutex_unlock(&wq_pool_mutex);
5971
5972         /* create the initial workers */
5973         for_each_online_cpu(cpu) {
5974                 for_each_cpu_worker_pool(pool, cpu) {
5975                         pool->flags &= ~POOL_DISASSOCIATED;
5976                         BUG_ON(!create_worker(pool));
5977                 }
5978         }
5979
5980         hash_for_each(unbound_pool_hash, bkt, pool, hash_node)
5981                 BUG_ON(!create_worker(pool));
5982
5983         wq_online = true;
5984         wq_watchdog_init();
5985
5986         return 0;
5987 }