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