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