cgroup: fix to break the while loop in cgroup_attach_task() correctly
[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         /*
2192          * The following prevents a kworker from hogging CPU on !PREEMPT
2193          * kernels, where a requeueing work item waiting for something to
2194          * happen could deadlock with stop_machine as such work item could
2195          * indefinitely requeue itself while all other CPUs are trapped in
2196          * stop_machine.
2197          */
2198         cond_resched();
2199
2200         spin_lock_irq(&pool->lock);
2201
2202         /* clear cpu intensive status */
2203         if (unlikely(cpu_intensive))
2204                 worker_clr_flags(worker, WORKER_CPU_INTENSIVE);
2205
2206         /* we're done with it, release */
2207         hash_del(&worker->hentry);
2208         worker->current_work = NULL;
2209         worker->current_func = NULL;
2210         worker->current_pwq = NULL;
2211         worker->desc_valid = false;
2212         pwq_dec_nr_in_flight(pwq, work_color);
2213 }
2214
2215 /**
2216  * process_scheduled_works - process scheduled works
2217  * @worker: self
2218  *
2219  * Process all scheduled works.  Please note that the scheduled list
2220  * may change while processing a work, so this function repeatedly
2221  * fetches a work from the top and executes it.
2222  *
2223  * CONTEXT:
2224  * spin_lock_irq(pool->lock) which may be released and regrabbed
2225  * multiple times.
2226  */
2227 static void process_scheduled_works(struct worker *worker)
2228 {
2229         while (!list_empty(&worker->scheduled)) {
2230                 struct work_struct *work = list_first_entry(&worker->scheduled,
2231                                                 struct work_struct, entry);
2232                 process_one_work(worker, work);
2233         }
2234 }
2235
2236 /**
2237  * worker_thread - the worker thread function
2238  * @__worker: self
2239  *
2240  * The worker thread function.  All workers belong to a worker_pool -
2241  * either a per-cpu one or dynamic unbound one.  These workers process all
2242  * work items regardless of their specific target workqueue.  The only
2243  * exception is work items which belong to workqueues with a rescuer which
2244  * will be explained in rescuer_thread().
2245  */
2246 static int worker_thread(void *__worker)
2247 {
2248         struct worker *worker = __worker;
2249         struct worker_pool *pool = worker->pool;
2250
2251         /* tell the scheduler that this is a workqueue worker */
2252         worker->task->flags |= PF_WQ_WORKER;
2253 woke_up:
2254         spin_lock_irq(&pool->lock);
2255
2256         /* am I supposed to die? */
2257         if (unlikely(worker->flags & WORKER_DIE)) {
2258                 spin_unlock_irq(&pool->lock);
2259                 WARN_ON_ONCE(!list_empty(&worker->entry));
2260                 worker->task->flags &= ~PF_WQ_WORKER;
2261                 return 0;
2262         }
2263
2264         worker_leave_idle(worker);
2265 recheck:
2266         /* no more worker necessary? */
2267         if (!need_more_worker(pool))
2268                 goto sleep;
2269
2270         /* do we need to manage? */
2271         if (unlikely(!may_start_working(pool)) && manage_workers(worker))
2272                 goto recheck;
2273
2274         /*
2275          * ->scheduled list can only be filled while a worker is
2276          * preparing to process a work or actually processing it.
2277          * Make sure nobody diddled with it while I was sleeping.
2278          */
2279         WARN_ON_ONCE(!list_empty(&worker->scheduled));
2280
2281         /*
2282          * Finish PREP stage.  We're guaranteed to have at least one idle
2283          * worker or that someone else has already assumed the manager
2284          * role.  This is where @worker starts participating in concurrency
2285          * management if applicable and concurrency management is restored
2286          * after being rebound.  See rebind_workers() for details.
2287          */
2288         worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);
2289
2290         do {
2291                 struct work_struct *work =
2292                         list_first_entry(&pool->worklist,
2293                                          struct work_struct, entry);
2294
2295                 if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
2296                         /* optimization path, not strictly necessary */
2297                         process_one_work(worker, work);
2298                         if (unlikely(!list_empty(&worker->scheduled)))
2299                                 process_scheduled_works(worker);
2300                 } else {
2301                         move_linked_works(work, &worker->scheduled, NULL);
2302                         process_scheduled_works(worker);
2303                 }
2304         } while (keep_working(pool));
2305
2306         worker_set_flags(worker, WORKER_PREP, false);
2307 sleep:
2308         if (unlikely(need_to_manage_workers(pool)) && manage_workers(worker))
2309                 goto recheck;
2310
2311         /*
2312          * pool->lock is held and there's no work to process and no need to
2313          * manage, sleep.  Workers are woken up only while holding
2314          * pool->lock or from local cpu, so setting the current state
2315          * before releasing pool->lock is enough to prevent losing any
2316          * event.
2317          */
2318         worker_enter_idle(worker);
2319         __set_current_state(TASK_INTERRUPTIBLE);
2320         spin_unlock_irq(&pool->lock);
2321         schedule();
2322         goto woke_up;
2323 }
2324
2325 /**
2326  * rescuer_thread - the rescuer thread function
2327  * @__rescuer: self
2328  *
2329  * Workqueue rescuer thread function.  There's one rescuer for each
2330  * workqueue which has WQ_MEM_RECLAIM set.
2331  *
2332  * Regular work processing on a pool may block trying to create a new
2333  * worker which uses GFP_KERNEL allocation which has slight chance of
2334  * developing into deadlock if some works currently on the same queue
2335  * need to be processed to satisfy the GFP_KERNEL allocation.  This is
2336  * the problem rescuer solves.
2337  *
2338  * When such condition is possible, the pool summons rescuers of all
2339  * workqueues which have works queued on the pool and let them process
2340  * those works so that forward progress can be guaranteed.
2341  *
2342  * This should happen rarely.
2343  */
2344 static int rescuer_thread(void *__rescuer)
2345 {
2346         struct worker *rescuer = __rescuer;
2347         struct workqueue_struct *wq = rescuer->rescue_wq;
2348         struct list_head *scheduled = &rescuer->scheduled;
2349
2350         set_user_nice(current, RESCUER_NICE_LEVEL);
2351
2352         /*
2353          * Mark rescuer as worker too.  As WORKER_PREP is never cleared, it
2354          * doesn't participate in concurrency management.
2355          */
2356         rescuer->task->flags |= PF_WQ_WORKER;
2357 repeat:
2358         set_current_state(TASK_INTERRUPTIBLE);
2359
2360         if (kthread_should_stop()) {
2361                 __set_current_state(TASK_RUNNING);
2362                 rescuer->task->flags &= ~PF_WQ_WORKER;
2363                 return 0;
2364         }
2365
2366         /* see whether any pwq is asking for help */
2367         spin_lock_irq(&wq_mayday_lock);
2368
2369         while (!list_empty(&wq->maydays)) {
2370                 struct pool_workqueue *pwq = list_first_entry(&wq->maydays,
2371                                         struct pool_workqueue, mayday_node);
2372                 struct worker_pool *pool = pwq->pool;
2373                 struct work_struct *work, *n;
2374
2375                 __set_current_state(TASK_RUNNING);
2376                 list_del_init(&pwq->mayday_node);
2377
2378                 spin_unlock_irq(&wq_mayday_lock);
2379
2380                 /* migrate to the target cpu if possible */
2381                 worker_maybe_bind_and_lock(pool);
2382                 rescuer->pool = pool;
2383
2384                 /*
2385                  * Slurp in all works issued via this workqueue and
2386                  * process'em.
2387                  */
2388                 WARN_ON_ONCE(!list_empty(&rescuer->scheduled));
2389                 list_for_each_entry_safe(work, n, &pool->worklist, entry)
2390                         if (get_work_pwq(work) == pwq)
2391                                 move_linked_works(work, scheduled, &n);
2392
2393                 process_scheduled_works(rescuer);
2394
2395                 /*
2396                  * Leave this pool.  If keep_working() is %true, notify a
2397                  * regular worker; otherwise, we end up with 0 concurrency
2398                  * and stalling the execution.
2399                  */
2400                 if (keep_working(pool))
2401                         wake_up_worker(pool);
2402
2403                 rescuer->pool = NULL;
2404                 spin_unlock(&pool->lock);
2405                 spin_lock(&wq_mayday_lock);
2406         }
2407
2408         spin_unlock_irq(&wq_mayday_lock);
2409
2410         /* rescuers should never participate in concurrency management */
2411         WARN_ON_ONCE(!(rescuer->flags & WORKER_NOT_RUNNING));
2412         schedule();
2413         goto repeat;
2414 }
2415
2416 struct wq_barrier {
2417         struct work_struct      work;
2418         struct completion       done;
2419 };
2420
2421 static void wq_barrier_func(struct work_struct *work)
2422 {
2423         struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
2424         complete(&barr->done);
2425 }
2426
2427 /**
2428  * insert_wq_barrier - insert a barrier work
2429  * @pwq: pwq to insert barrier into
2430  * @barr: wq_barrier to insert
2431  * @target: target work to attach @barr to
2432  * @worker: worker currently executing @target, NULL if @target is not executing
2433  *
2434  * @barr is linked to @target such that @barr is completed only after
2435  * @target finishes execution.  Please note that the ordering
2436  * guarantee is observed only with respect to @target and on the local
2437  * cpu.
2438  *
2439  * Currently, a queued barrier can't be canceled.  This is because
2440  * try_to_grab_pending() can't determine whether the work to be
2441  * grabbed is at the head of the queue and thus can't clear LINKED
2442  * flag of the previous work while there must be a valid next work
2443  * after a work with LINKED flag set.
2444  *
2445  * Note that when @worker is non-NULL, @target may be modified
2446  * underneath us, so we can't reliably determine pwq from @target.
2447  *
2448  * CONTEXT:
2449  * spin_lock_irq(pool->lock).
2450  */
2451 static void insert_wq_barrier(struct pool_workqueue *pwq,
2452                               struct wq_barrier *barr,
2453                               struct work_struct *target, struct worker *worker)
2454 {
2455         struct list_head *head;
2456         unsigned int linked = 0;
2457
2458         /*
2459          * debugobject calls are safe here even with pool->lock locked
2460          * as we know for sure that this will not trigger any of the
2461          * checks and call back into the fixup functions where we
2462          * might deadlock.
2463          */
2464         INIT_WORK_ONSTACK(&barr->work, wq_barrier_func);
2465         __set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(&barr->work));
2466         init_completion(&barr->done);
2467
2468         /*
2469          * If @target is currently being executed, schedule the
2470          * barrier to the worker; otherwise, put it after @target.
2471          */
2472         if (worker)
2473                 head = worker->scheduled.next;
2474         else {
2475                 unsigned long *bits = work_data_bits(target);
2476
2477                 head = target->entry.next;
2478                 /* there can already be other linked works, inherit and set */
2479                 linked = *bits & WORK_STRUCT_LINKED;
2480                 __set_bit(WORK_STRUCT_LINKED_BIT, bits);
2481         }
2482
2483         debug_work_activate(&barr->work);
2484         insert_work(pwq, &barr->work, head,
2485                     work_color_to_flags(WORK_NO_COLOR) | linked);
2486 }
2487
2488 /**
2489  * flush_workqueue_prep_pwqs - prepare pwqs for workqueue flushing
2490  * @wq: workqueue being flushed
2491  * @flush_color: new flush color, < 0 for no-op
2492  * @work_color: new work color, < 0 for no-op
2493  *
2494  * Prepare pwqs for workqueue flushing.
2495  *
2496  * If @flush_color is non-negative, flush_color on all pwqs should be
2497  * -1.  If no pwq has in-flight commands at the specified color, all
2498  * pwq->flush_color's stay at -1 and %false is returned.  If any pwq
2499  * has in flight commands, its pwq->flush_color is set to
2500  * @flush_color, @wq->nr_pwqs_to_flush is updated accordingly, pwq
2501  * wakeup logic is armed and %true is returned.
2502  *
2503  * The caller should have initialized @wq->first_flusher prior to
2504  * calling this function with non-negative @flush_color.  If
2505  * @flush_color is negative, no flush color update is done and %false
2506  * is returned.
2507  *
2508  * If @work_color is non-negative, all pwqs should have the same
2509  * work_color which is previous to @work_color and all will be
2510  * advanced to @work_color.
2511  *
2512  * CONTEXT:
2513  * mutex_lock(wq->mutex).
2514  *
2515  * RETURNS:
2516  * %true if @flush_color >= 0 and there's something to flush.  %false
2517  * otherwise.
2518  */
2519 static bool flush_workqueue_prep_pwqs(struct workqueue_struct *wq,
2520                                       int flush_color, int work_color)
2521 {
2522         bool wait = false;
2523         struct pool_workqueue *pwq;
2524
2525         if (flush_color >= 0) {
2526                 WARN_ON_ONCE(atomic_read(&wq->nr_pwqs_to_flush));
2527                 atomic_set(&wq->nr_pwqs_to_flush, 1);
2528         }
2529
2530         for_each_pwq(pwq, wq) {
2531                 struct worker_pool *pool = pwq->pool;
2532
2533                 spin_lock_irq(&pool->lock);
2534
2535                 if (flush_color >= 0) {
2536                         WARN_ON_ONCE(pwq->flush_color != -1);
2537
2538                         if (pwq->nr_in_flight[flush_color]) {
2539                                 pwq->flush_color = flush_color;
2540                                 atomic_inc(&wq->nr_pwqs_to_flush);
2541                                 wait = true;
2542                         }
2543                 }
2544
2545                 if (work_color >= 0) {
2546                         WARN_ON_ONCE(work_color != work_next_color(pwq->work_color));
2547                         pwq->work_color = work_color;
2548                 }
2549
2550                 spin_unlock_irq(&pool->lock);
2551         }
2552
2553         if (flush_color >= 0 && atomic_dec_and_test(&wq->nr_pwqs_to_flush))
2554                 complete(&wq->first_flusher->done);
2555
2556         return wait;
2557 }
2558
2559 /**
2560  * flush_workqueue - ensure that any scheduled work has run to completion.
2561  * @wq: workqueue to flush
2562  *
2563  * This function sleeps until all work items which were queued on entry
2564  * have finished execution, but it is not livelocked by new incoming ones.
2565  */
2566 void flush_workqueue(struct workqueue_struct *wq)
2567 {
2568         struct wq_flusher this_flusher = {
2569                 .list = LIST_HEAD_INIT(this_flusher.list),
2570                 .flush_color = -1,
2571                 .done = COMPLETION_INITIALIZER_ONSTACK(this_flusher.done),
2572         };
2573         int next_color;
2574
2575         lock_map_acquire(&wq->lockdep_map);
2576         lock_map_release(&wq->lockdep_map);
2577
2578         mutex_lock(&wq->mutex);
2579
2580         /*
2581          * Start-to-wait phase
2582          */
2583         next_color = work_next_color(wq->work_color);
2584
2585         if (next_color != wq->flush_color) {
2586                 /*
2587                  * Color space is not full.  The current work_color
2588                  * becomes our flush_color and work_color is advanced
2589                  * by one.
2590                  */
2591                 WARN_ON_ONCE(!list_empty(&wq->flusher_overflow));
2592                 this_flusher.flush_color = wq->work_color;
2593                 wq->work_color = next_color;
2594
2595                 if (!wq->first_flusher) {
2596                         /* no flush in progress, become the first flusher */
2597                         WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);
2598
2599                         wq->first_flusher = &this_flusher;
2600
2601                         if (!flush_workqueue_prep_pwqs(wq, wq->flush_color,
2602                                                        wq->work_color)) {
2603                                 /* nothing to flush, done */
2604                                 wq->flush_color = next_color;
2605                                 wq->first_flusher = NULL;
2606                                 goto out_unlock;
2607                         }
2608                 } else {
2609                         /* wait in queue */
2610                         WARN_ON_ONCE(wq->flush_color == this_flusher.flush_color);
2611                         list_add_tail(&this_flusher.list, &wq->flusher_queue);
2612                         flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
2613                 }
2614         } else {
2615                 /*
2616                  * Oops, color space is full, wait on overflow queue.
2617                  * The next flush completion will assign us
2618                  * flush_color and transfer to flusher_queue.
2619                  */
2620                 list_add_tail(&this_flusher.list, &wq->flusher_overflow);
2621         }
2622
2623         mutex_unlock(&wq->mutex);
2624
2625         wait_for_completion(&this_flusher.done);
2626
2627         /*
2628          * Wake-up-and-cascade phase
2629          *
2630          * First flushers are responsible for cascading flushes and
2631          * handling overflow.  Non-first flushers can simply return.
2632          */
2633         if (wq->first_flusher != &this_flusher)
2634                 return;
2635
2636         mutex_lock(&wq->mutex);
2637
2638         /* we might have raced, check again with mutex held */
2639         if (wq->first_flusher != &this_flusher)
2640                 goto out_unlock;
2641
2642         wq->first_flusher = NULL;
2643
2644         WARN_ON_ONCE(!list_empty(&this_flusher.list));
2645         WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);
2646
2647         while (true) {
2648                 struct wq_flusher *next, *tmp;
2649
2650                 /* complete all the flushers sharing the current flush color */
2651                 list_for_each_entry_safe(next, tmp, &wq->flusher_queue, list) {
2652                         if (next->flush_color != wq->flush_color)
2653                                 break;
2654                         list_del_init(&next->list);
2655                         complete(&next->done);
2656                 }
2657
2658                 WARN_ON_ONCE(!list_empty(&wq->flusher_overflow) &&
2659                              wq->flush_color != work_next_color(wq->work_color));
2660
2661                 /* this flush_color is finished, advance by one */
2662                 wq->flush_color = work_next_color(wq->flush_color);
2663
2664                 /* one color has been freed, handle overflow queue */
2665                 if (!list_empty(&wq->flusher_overflow)) {
2666                         /*
2667                          * Assign the same color to all overflowed
2668                          * flushers, advance work_color and append to
2669                          * flusher_queue.  This is the start-to-wait
2670                          * phase for these overflowed flushers.
2671                          */
2672                         list_for_each_entry(tmp, &wq->flusher_overflow, list)
2673                                 tmp->flush_color = wq->work_color;
2674
2675                         wq->work_color = work_next_color(wq->work_color);
2676
2677                         list_splice_tail_init(&wq->flusher_overflow,
2678                                               &wq->flusher_queue);
2679                         flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
2680                 }
2681
2682                 if (list_empty(&wq->flusher_queue)) {
2683                         WARN_ON_ONCE(wq->flush_color != wq->work_color);
2684                         break;
2685                 }
2686
2687                 /*
2688                  * Need to flush more colors.  Make the next flusher
2689                  * the new first flusher and arm pwqs.
2690                  */
2691                 WARN_ON_ONCE(wq->flush_color == wq->work_color);
2692                 WARN_ON_ONCE(wq->flush_color != next->flush_color);
2693
2694                 list_del_init(&next->list);
2695                 wq->first_flusher = next;
2696
2697                 if (flush_workqueue_prep_pwqs(wq, wq->flush_color, -1))
2698                         break;
2699
2700                 /*
2701                  * Meh... this color is already done, clear first
2702                  * flusher and repeat cascading.
2703                  */
2704                 wq->first_flusher = NULL;
2705         }
2706
2707 out_unlock:
2708         mutex_unlock(&wq->mutex);
2709 }
2710 EXPORT_SYMBOL_GPL(flush_workqueue);
2711
2712 /**
2713  * drain_workqueue - drain a workqueue
2714  * @wq: workqueue to drain
2715  *
2716  * Wait until the workqueue becomes empty.  While draining is in progress,
2717  * only chain queueing is allowed.  IOW, only currently pending or running
2718  * work items on @wq can queue further work items on it.  @wq is flushed
2719  * repeatedly until it becomes empty.  The number of flushing is detemined
2720  * by the depth of chaining and should be relatively short.  Whine if it
2721  * takes too long.
2722  */
2723 void drain_workqueue(struct workqueue_struct *wq)
2724 {
2725         unsigned int flush_cnt = 0;
2726         struct pool_workqueue *pwq;
2727
2728         /*
2729          * __queue_work() needs to test whether there are drainers, is much
2730          * hotter than drain_workqueue() and already looks at @wq->flags.
2731          * Use __WQ_DRAINING so that queue doesn't have to check nr_drainers.
2732          */
2733         mutex_lock(&wq->mutex);
2734         if (!wq->nr_drainers++)
2735                 wq->flags |= __WQ_DRAINING;
2736         mutex_unlock(&wq->mutex);
2737 reflush:
2738         flush_workqueue(wq);
2739
2740         mutex_lock(&wq->mutex);
2741
2742         for_each_pwq(pwq, wq) {
2743                 bool drained;
2744
2745                 spin_lock_irq(&pwq->pool->lock);
2746                 drained = !pwq->nr_active && list_empty(&pwq->delayed_works);
2747                 spin_unlock_irq(&pwq->pool->lock);
2748
2749                 if (drained)
2750                         continue;
2751
2752                 if (++flush_cnt == 10 ||
2753                     (flush_cnt % 100 == 0 && flush_cnt <= 1000))
2754                         pr_warn("workqueue %s: drain_workqueue() isn't complete after %u tries\n",
2755                                 wq->name, flush_cnt);
2756
2757                 mutex_unlock(&wq->mutex);
2758                 goto reflush;
2759         }
2760
2761         if (!--wq->nr_drainers)
2762                 wq->flags &= ~__WQ_DRAINING;
2763         mutex_unlock(&wq->mutex);
2764 }
2765 EXPORT_SYMBOL_GPL(drain_workqueue);
2766
2767 static bool start_flush_work(struct work_struct *work, struct wq_barrier *barr)
2768 {
2769         struct worker *worker = NULL;
2770         struct worker_pool *pool;
2771         struct pool_workqueue *pwq;
2772
2773         might_sleep();
2774
2775         local_irq_disable();
2776         pool = get_work_pool(work);
2777         if (!pool) {
2778                 local_irq_enable();
2779                 return false;
2780         }
2781
2782         spin_lock(&pool->lock);
2783         /* see the comment in try_to_grab_pending() with the same code */
2784         pwq = get_work_pwq(work);
2785         if (pwq) {
2786                 if (unlikely(pwq->pool != pool))
2787                         goto already_gone;
2788         } else {
2789                 worker = find_worker_executing_work(pool, work);
2790                 if (!worker)
2791                         goto already_gone;
2792                 pwq = worker->current_pwq;
2793         }
2794
2795         insert_wq_barrier(pwq, barr, work, worker);
2796         spin_unlock_irq(&pool->lock);
2797
2798         /*
2799          * If @max_active is 1 or rescuer is in use, flushing another work
2800          * item on the same workqueue may lead to deadlock.  Make sure the
2801          * flusher is not running on the same workqueue by verifying write
2802          * access.
2803          */
2804         if (pwq->wq->saved_max_active == 1 || pwq->wq->rescuer)
2805                 lock_map_acquire(&pwq->wq->lockdep_map);
2806         else
2807                 lock_map_acquire_read(&pwq->wq->lockdep_map);
2808         lock_map_release(&pwq->wq->lockdep_map);
2809
2810         return true;
2811 already_gone:
2812         spin_unlock_irq(&pool->lock);
2813         return false;
2814 }
2815
2816 /**
2817  * flush_work - wait for a work to finish executing the last queueing instance
2818  * @work: the work to flush
2819  *
2820  * Wait until @work has finished execution.  @work is guaranteed to be idle
2821  * on return if it hasn't been requeued since flush started.
2822  *
2823  * RETURNS:
2824  * %true if flush_work() waited for the work to finish execution,
2825  * %false if it was already idle.
2826  */
2827 bool flush_work(struct work_struct *work)
2828 {
2829         struct wq_barrier barr;
2830
2831         lock_map_acquire(&work->lockdep_map);
2832         lock_map_release(&work->lockdep_map);
2833
2834         if (start_flush_work(work, &barr)) {
2835                 wait_for_completion(&barr.done);
2836                 destroy_work_on_stack(&barr.work);
2837                 return true;
2838         } else {
2839                 return false;
2840         }
2841 }
2842 EXPORT_SYMBOL_GPL(flush_work);
2843
2844 static bool __cancel_work_timer(struct work_struct *work, bool is_dwork)
2845 {
2846         unsigned long flags;
2847         int ret;
2848
2849         do {
2850                 ret = try_to_grab_pending(work, is_dwork, &flags);
2851                 /*
2852                  * If someone else is canceling, wait for the same event it
2853                  * would be waiting for before retrying.
2854                  */
2855                 if (unlikely(ret == -ENOENT))
2856                         flush_work(work);
2857         } while (unlikely(ret < 0));
2858
2859         /* tell other tasks trying to grab @work to back off */
2860         mark_work_canceling(work);
2861         local_irq_restore(flags);
2862
2863         flush_work(work);
2864         clear_work_data(work);
2865         return ret;
2866 }
2867
2868 /**
2869  * cancel_work_sync - cancel a work and wait for it to finish
2870  * @work: the work to cancel
2871  *
2872  * Cancel @work and wait for its execution to finish.  This function
2873  * can be used even if the work re-queues itself or migrates to
2874  * another workqueue.  On return from this function, @work is
2875  * guaranteed to be not pending or executing on any CPU.
2876  *
2877  * cancel_work_sync(&delayed_work->work) must not be used for
2878  * delayed_work's.  Use cancel_delayed_work_sync() instead.
2879  *
2880  * The caller must ensure that the workqueue on which @work was last
2881  * queued can't be destroyed before this function returns.
2882  *
2883  * RETURNS:
2884  * %true if @work was pending, %false otherwise.
2885  */
2886 bool cancel_work_sync(struct work_struct *work)
2887 {
2888         return __cancel_work_timer(work, false);
2889 }
2890 EXPORT_SYMBOL_GPL(cancel_work_sync);
2891
2892 /**
2893  * flush_delayed_work - wait for a dwork to finish executing the last queueing
2894  * @dwork: the delayed work to flush
2895  *
2896  * Delayed timer is cancelled and the pending work is queued for
2897  * immediate execution.  Like flush_work(), this function only
2898  * considers the last queueing instance of @dwork.
2899  *
2900  * RETURNS:
2901  * %true if flush_work() waited for the work to finish execution,
2902  * %false if it was already idle.
2903  */
2904 bool flush_delayed_work(struct delayed_work *dwork)
2905 {
2906         local_irq_disable();
2907         if (del_timer_sync(&dwork->timer))
2908                 __queue_work(dwork->cpu, dwork->wq, &dwork->work);
2909         local_irq_enable();
2910         return flush_work(&dwork->work);
2911 }
2912 EXPORT_SYMBOL(flush_delayed_work);
2913
2914 /**
2915  * cancel_delayed_work - cancel a delayed work
2916  * @dwork: delayed_work to cancel
2917  *
2918  * Kill off a pending delayed_work.  Returns %true if @dwork was pending
2919  * and canceled; %false if wasn't pending.  Note that the work callback
2920  * function may still be running on return, unless it returns %true and the
2921  * work doesn't re-arm itself.  Explicitly flush or use
2922  * cancel_delayed_work_sync() to wait on it.
2923  *
2924  * This function is safe to call from any context including IRQ handler.
2925  */
2926 bool cancel_delayed_work(struct delayed_work *dwork)
2927 {
2928         unsigned long flags;
2929         int ret;
2930
2931         do {
2932                 ret = try_to_grab_pending(&dwork->work, true, &flags);
2933         } while (unlikely(ret == -EAGAIN));
2934
2935         if (unlikely(ret < 0))
2936                 return false;
2937
2938         set_work_pool_and_clear_pending(&dwork->work,
2939                                         get_work_pool_id(&dwork->work));
2940         local_irq_restore(flags);
2941         return ret;
2942 }
2943 EXPORT_SYMBOL(cancel_delayed_work);
2944
2945 /**
2946  * cancel_delayed_work_sync - cancel a delayed work and wait for it to finish
2947  * @dwork: the delayed work cancel
2948  *
2949  * This is cancel_work_sync() for delayed works.
2950  *
2951  * RETURNS:
2952  * %true if @dwork was pending, %false otherwise.
2953  */
2954 bool cancel_delayed_work_sync(struct delayed_work *dwork)
2955 {
2956         return __cancel_work_timer(&dwork->work, true);
2957 }
2958 EXPORT_SYMBOL(cancel_delayed_work_sync);
2959
2960 /**
2961  * schedule_on_each_cpu - execute a function synchronously on each online CPU
2962  * @func: the function to call
2963  *
2964  * schedule_on_each_cpu() executes @func on each online CPU using the
2965  * system workqueue and blocks until all CPUs have completed.
2966  * schedule_on_each_cpu() is very slow.
2967  *
2968  * RETURNS:
2969  * 0 on success, -errno on failure.
2970  */
2971 int schedule_on_each_cpu(work_func_t func)
2972 {
2973         int cpu;
2974         struct work_struct __percpu *works;
2975
2976         works = alloc_percpu(struct work_struct);
2977         if (!works)
2978                 return -ENOMEM;
2979
2980         get_online_cpus();
2981
2982         for_each_online_cpu(cpu) {
2983                 struct work_struct *work = per_cpu_ptr(works, cpu);
2984
2985                 INIT_WORK(work, func);
2986                 schedule_work_on(cpu, work);
2987         }
2988
2989         for_each_online_cpu(cpu)
2990                 flush_work(per_cpu_ptr(works, cpu));
2991
2992         put_online_cpus();
2993         free_percpu(works);
2994         return 0;
2995 }
2996
2997 /**
2998  * flush_scheduled_work - ensure that any scheduled work has run to completion.
2999  *
3000  * Forces execution of the kernel-global workqueue and blocks until its
3001  * completion.
3002  *
3003  * Think twice before calling this function!  It's very easy to get into
3004  * trouble if you don't take great care.  Either of the following situations
3005  * will lead to deadlock:
3006  *
3007  *      One of the work items currently on the workqueue needs to acquire
3008  *      a lock held by your code or its caller.
3009  *
3010  *      Your code is running in the context of a work routine.
3011  *
3012  * They will be detected by lockdep when they occur, but the first might not
3013  * occur very often.  It depends on what work items are on the workqueue and
3014  * what locks they need, which you have no control over.
3015  *
3016  * In most situations flushing the entire workqueue is overkill; you merely
3017  * need to know that a particular work item isn't queued and isn't running.
3018  * In such cases you should use cancel_delayed_work_sync() or
3019  * cancel_work_sync() instead.
3020  */
3021 void flush_scheduled_work(void)
3022 {
3023         flush_workqueue(system_wq);
3024 }
3025 EXPORT_SYMBOL(flush_scheduled_work);
3026
3027 /**
3028  * execute_in_process_context - reliably execute the routine with user context
3029  * @fn:         the function to execute
3030  * @ew:         guaranteed storage for the execute work structure (must
3031  *              be available when the work executes)
3032  *
3033  * Executes the function immediately if process context is available,
3034  * otherwise schedules the function for delayed execution.
3035  *
3036  * Returns:     0 - function was executed
3037  *              1 - function was scheduled for execution
3038  */
3039 int execute_in_process_context(work_func_t fn, struct execute_work *ew)
3040 {
3041         if (!in_interrupt()) {
3042                 fn(&ew->work);
3043                 return 0;
3044         }
3045
3046         INIT_WORK(&ew->work, fn);
3047         schedule_work(&ew->work);
3048
3049         return 1;
3050 }
3051 EXPORT_SYMBOL_GPL(execute_in_process_context);
3052
3053 #ifdef CONFIG_SYSFS
3054 /*
3055  * Workqueues with WQ_SYSFS flag set is visible to userland via
3056  * /sys/bus/workqueue/devices/WQ_NAME.  All visible workqueues have the
3057  * following attributes.
3058  *
3059  *  per_cpu     RO bool : whether the workqueue is per-cpu or unbound
3060  *  max_active  RW int  : maximum number of in-flight work items
3061  *
3062  * Unbound workqueues have the following extra attributes.
3063  *
3064  *  id          RO int  : the associated pool ID
3065  *  nice        RW int  : nice value of the workers
3066  *  cpumask     RW mask : bitmask of allowed CPUs for the workers
3067  */
3068 struct wq_device {
3069         struct workqueue_struct         *wq;
3070         struct device                   dev;
3071 };
3072
3073 static struct workqueue_struct *dev_to_wq(struct device *dev)
3074 {
3075         struct wq_device *wq_dev = container_of(dev, struct wq_device, dev);
3076
3077         return wq_dev->wq;
3078 }
3079
3080 static ssize_t wq_per_cpu_show(struct device *dev,
3081                                struct device_attribute *attr, char *buf)
3082 {
3083         struct workqueue_struct *wq = dev_to_wq(dev);
3084
3085         return scnprintf(buf, PAGE_SIZE, "%d\n", (bool)!(wq->flags & WQ_UNBOUND));
3086 }
3087
3088 static ssize_t wq_max_active_show(struct device *dev,
3089                                   struct device_attribute *attr, char *buf)
3090 {
3091         struct workqueue_struct *wq = dev_to_wq(dev);
3092
3093         return scnprintf(buf, PAGE_SIZE, "%d\n", wq->saved_max_active);
3094 }
3095
3096 static ssize_t wq_max_active_store(struct device *dev,
3097                                    struct device_attribute *attr,
3098                                    const char *buf, size_t count)
3099 {
3100         struct workqueue_struct *wq = dev_to_wq(dev);
3101         int val;
3102
3103         if (sscanf(buf, "%d", &val) != 1 || val <= 0)
3104                 return -EINVAL;
3105
3106         workqueue_set_max_active(wq, val);
3107         return count;
3108 }
3109
3110 static struct device_attribute wq_sysfs_attrs[] = {
3111         __ATTR(per_cpu, 0444, wq_per_cpu_show, NULL),
3112         __ATTR(max_active, 0644, wq_max_active_show, wq_max_active_store),
3113         __ATTR_NULL,
3114 };
3115
3116 static ssize_t wq_pool_ids_show(struct device *dev,
3117                                 struct device_attribute *attr, char *buf)
3118 {
3119         struct workqueue_struct *wq = dev_to_wq(dev);
3120         const char *delim = "";
3121         int node, written = 0;
3122
3123         rcu_read_lock_sched();
3124         for_each_node(node) {
3125                 written += scnprintf(buf + written, PAGE_SIZE - written,
3126                                      "%s%d:%d", delim, node,
3127                                      unbound_pwq_by_node(wq, node)->pool->id);
3128                 delim = " ";
3129         }
3130         written += scnprintf(buf + written, PAGE_SIZE - written, "\n");
3131         rcu_read_unlock_sched();
3132
3133         return written;
3134 }
3135
3136 static ssize_t wq_nice_show(struct device *dev, struct device_attribute *attr,
3137                             char *buf)
3138 {
3139         struct workqueue_struct *wq = dev_to_wq(dev);
3140         int written;
3141
3142         mutex_lock(&wq->mutex);
3143         written = scnprintf(buf, PAGE_SIZE, "%d\n", wq->unbound_attrs->nice);
3144         mutex_unlock(&wq->mutex);
3145
3146         return written;
3147 }
3148
3149 /* prepare workqueue_attrs for sysfs store operations */
3150 static struct workqueue_attrs *wq_sysfs_prep_attrs(struct workqueue_struct *wq)
3151 {
3152         struct workqueue_attrs *attrs;
3153
3154         attrs = alloc_workqueue_attrs(GFP_KERNEL);
3155         if (!attrs)
3156                 return NULL;
3157
3158         mutex_lock(&wq->mutex);
3159         copy_workqueue_attrs(attrs, wq->unbound_attrs);
3160         mutex_unlock(&wq->mutex);
3161         return attrs;
3162 }
3163
3164 static ssize_t wq_nice_store(struct device *dev, struct device_attribute *attr,
3165                              const char *buf, size_t count)
3166 {
3167         struct workqueue_struct *wq = dev_to_wq(dev);
3168         struct workqueue_attrs *attrs;
3169         int ret;
3170
3171         attrs = wq_sysfs_prep_attrs(wq);
3172         if (!attrs)
3173                 return -ENOMEM;
3174
3175         if (sscanf(buf, "%d", &attrs->nice) == 1 &&
3176             attrs->nice >= -20 && attrs->nice <= 19)
3177                 ret = apply_workqueue_attrs(wq, attrs);
3178         else
3179                 ret = -EINVAL;
3180
3181         free_workqueue_attrs(attrs);
3182         return ret ?: count;
3183 }
3184
3185 static ssize_t wq_cpumask_show(struct device *dev,
3186                                struct device_attribute *attr, char *buf)
3187 {
3188         struct workqueue_struct *wq = dev_to_wq(dev);
3189         int written;
3190
3191         mutex_lock(&wq->mutex);
3192         written = cpumask_scnprintf(buf, PAGE_SIZE, wq->unbound_attrs->cpumask);
3193         mutex_unlock(&wq->mutex);
3194
3195         written += scnprintf(buf + written, PAGE_SIZE - written, "\n");
3196         return written;
3197 }
3198
3199 static ssize_t wq_cpumask_store(struct device *dev,
3200                                 struct device_attribute *attr,
3201                                 const char *buf, size_t count)
3202 {
3203         struct workqueue_struct *wq = dev_to_wq(dev);
3204         struct workqueue_attrs *attrs;
3205         int ret;
3206
3207         attrs = wq_sysfs_prep_attrs(wq);
3208         if (!attrs)
3209                 return -ENOMEM;
3210
3211         ret = cpumask_parse(buf, attrs->cpumask);
3212         if (!ret)
3213                 ret = apply_workqueue_attrs(wq, attrs);
3214
3215         free_workqueue_attrs(attrs);
3216         return ret ?: count;
3217 }
3218
3219 static ssize_t wq_numa_show(struct device *dev, struct device_attribute *attr,
3220                             char *buf)
3221 {
3222         struct workqueue_struct *wq = dev_to_wq(dev);
3223         int written;
3224
3225         mutex_lock(&wq->mutex);
3226         written = scnprintf(buf, PAGE_SIZE, "%d\n",
3227                             !wq->unbound_attrs->no_numa);
3228         mutex_unlock(&wq->mutex);
3229
3230         return written;
3231 }
3232
3233 static ssize_t wq_numa_store(struct device *dev, struct device_attribute *attr,
3234                              const char *buf, size_t count)
3235 {
3236         struct workqueue_struct *wq = dev_to_wq(dev);
3237         struct workqueue_attrs *attrs;
3238         int v, ret;
3239
3240         attrs = wq_sysfs_prep_attrs(wq);
3241         if (!attrs)
3242                 return -ENOMEM;
3243
3244         ret = -EINVAL;
3245         if (sscanf(buf, "%d", &v) == 1) {
3246                 attrs->no_numa = !v;
3247                 ret = apply_workqueue_attrs(wq, attrs);
3248         }
3249
3250         free_workqueue_attrs(attrs);
3251         return ret ?: count;
3252 }
3253
3254 static struct device_attribute wq_sysfs_unbound_attrs[] = {
3255         __ATTR(pool_ids, 0444, wq_pool_ids_show, NULL),
3256         __ATTR(nice, 0644, wq_nice_show, wq_nice_store),
3257         __ATTR(cpumask, 0644, wq_cpumask_show, wq_cpumask_store),
3258         __ATTR(numa, 0644, wq_numa_show, wq_numa_store),
3259         __ATTR_NULL,
3260 };
3261
3262 static struct bus_type wq_subsys = {
3263         .name                           = "workqueue",
3264         .dev_attrs                      = wq_sysfs_attrs,
3265 };
3266
3267 static int __init wq_sysfs_init(void)
3268 {
3269         return subsys_virtual_register(&wq_subsys, NULL);
3270 }
3271 core_initcall(wq_sysfs_init);
3272
3273 static void wq_device_release(struct device *dev)
3274 {
3275         struct wq_device *wq_dev = container_of(dev, struct wq_device, dev);
3276
3277         kfree(wq_dev);
3278 }
3279
3280 /**
3281  * workqueue_sysfs_register - make a workqueue visible in sysfs
3282  * @wq: the workqueue to register
3283  *
3284  * Expose @wq in sysfs under /sys/bus/workqueue/devices.
3285  * alloc_workqueue*() automatically calls this function if WQ_SYSFS is set
3286  * which is the preferred method.
3287  *
3288  * Workqueue user should use this function directly iff it wants to apply
3289  * workqueue_attrs before making the workqueue visible in sysfs; otherwise,
3290  * apply_workqueue_attrs() may race against userland updating the
3291  * attributes.
3292  *
3293  * Returns 0 on success, -errno on failure.
3294  */
3295 int workqueue_sysfs_register(struct workqueue_struct *wq)
3296 {
3297         struct wq_device *wq_dev;
3298         int ret;
3299
3300         /*
3301          * Adjusting max_active or creating new pwqs by applyting
3302          * attributes breaks ordering guarantee.  Disallow exposing ordered
3303          * workqueues.
3304          */
3305         if (WARN_ON(wq->flags & __WQ_ORDERED))
3306                 return -EINVAL;
3307
3308         wq->wq_dev = wq_dev = kzalloc(sizeof(*wq_dev), GFP_KERNEL);
3309         if (!wq_dev)
3310                 return -ENOMEM;
3311
3312         wq_dev->wq = wq;
3313         wq_dev->dev.bus = &wq_subsys;
3314         wq_dev->dev.init_name = wq->name;
3315         wq_dev->dev.release = wq_device_release;
3316
3317         /*
3318          * unbound_attrs are created separately.  Suppress uevent until
3319          * everything is ready.
3320          */
3321         dev_set_uevent_suppress(&wq_dev->dev, true);
3322
3323         ret = device_register(&wq_dev->dev);
3324         if (ret) {
3325                 kfree(wq_dev);
3326                 wq->wq_dev = NULL;
3327                 return ret;
3328         }
3329
3330         if (wq->flags & WQ_UNBOUND) {
3331                 struct device_attribute *attr;
3332
3333                 for (attr = wq_sysfs_unbound_attrs; attr->attr.name; attr++) {
3334                         ret = device_create_file(&wq_dev->dev, attr);
3335                         if (ret) {
3336                                 device_unregister(&wq_dev->dev);
3337                                 wq->wq_dev = NULL;
3338                                 return ret;
3339                         }
3340                 }
3341         }
3342
3343         kobject_uevent(&wq_dev->dev.kobj, KOBJ_ADD);
3344         return 0;
3345 }
3346
3347 /**
3348  * workqueue_sysfs_unregister - undo workqueue_sysfs_register()
3349  * @wq: the workqueue to unregister
3350  *
3351  * If @wq is registered to sysfs by workqueue_sysfs_register(), unregister.
3352  */
3353 static void workqueue_sysfs_unregister(struct workqueue_struct *wq)
3354 {
3355         struct wq_device *wq_dev = wq->wq_dev;
3356
3357         if (!wq->wq_dev)
3358                 return;
3359
3360         wq->wq_dev = NULL;
3361         device_unregister(&wq_dev->dev);
3362 }
3363 #else   /* CONFIG_SYSFS */
3364 static void workqueue_sysfs_unregister(struct workqueue_struct *wq)     { }
3365 #endif  /* CONFIG_SYSFS */
3366
3367 /**
3368  * free_workqueue_attrs - free a workqueue_attrs
3369  * @attrs: workqueue_attrs to free
3370  *
3371  * Undo alloc_workqueue_attrs().
3372  */
3373 void free_workqueue_attrs(struct workqueue_attrs *attrs)
3374 {
3375         if (attrs) {
3376                 free_cpumask_var(attrs->cpumask);
3377                 kfree(attrs);
3378         }
3379 }
3380
3381 /**
3382  * alloc_workqueue_attrs - allocate a workqueue_attrs
3383  * @gfp_mask: allocation mask to use
3384  *
3385  * Allocate a new workqueue_attrs, initialize with default settings and
3386  * return it.  Returns NULL on failure.
3387  */
3388 struct workqueue_attrs *alloc_workqueue_attrs(gfp_t gfp_mask)
3389 {
3390         struct workqueue_attrs *attrs;
3391
3392         attrs = kzalloc(sizeof(*attrs), gfp_mask);
3393         if (!attrs)
3394                 goto fail;
3395         if (!alloc_cpumask_var(&attrs->cpumask, gfp_mask))
3396                 goto fail;
3397
3398         cpumask_copy(attrs->cpumask, cpu_possible_mask);
3399         return attrs;
3400 fail:
3401         free_workqueue_attrs(attrs);
3402         return NULL;
3403 }
3404
3405 static void copy_workqueue_attrs(struct workqueue_attrs *to,
3406                                  const struct workqueue_attrs *from)
3407 {
3408         to->nice = from->nice;
3409         cpumask_copy(to->cpumask, from->cpumask);
3410         /*
3411          * Unlike hash and equality test, this function doesn't ignore
3412          * ->no_numa as it is used for both pool and wq attrs.  Instead,
3413          * get_unbound_pool() explicitly clears ->no_numa after copying.
3414          */
3415         to->no_numa = from->no_numa;
3416 }
3417
3418 /* hash value of the content of @attr */
3419 static u32 wqattrs_hash(const struct workqueue_attrs *attrs)
3420 {
3421         u32 hash = 0;
3422
3423         hash = jhash_1word(attrs->nice, hash);
3424         hash = jhash(cpumask_bits(attrs->cpumask),
3425                      BITS_TO_LONGS(nr_cpumask_bits) * sizeof(long), hash);
3426         return hash;
3427 }
3428
3429 /* content equality test */
3430 static bool wqattrs_equal(const struct workqueue_attrs *a,
3431                           const struct workqueue_attrs *b)
3432 {
3433         if (a->nice != b->nice)
3434                 return false;
3435         if (!cpumask_equal(a->cpumask, b->cpumask))
3436                 return false;
3437         return true;
3438 }
3439
3440 /**
3441  * init_worker_pool - initialize a newly zalloc'd worker_pool
3442  * @pool: worker_pool to initialize
3443  *
3444  * Initiailize a newly zalloc'd @pool.  It also allocates @pool->attrs.
3445  * Returns 0 on success, -errno on failure.  Even on failure, all fields
3446  * inside @pool proper are initialized and put_unbound_pool() can be called
3447  * on @pool safely to release it.
3448  */
3449 static int init_worker_pool(struct worker_pool *pool)
3450 {
3451         spin_lock_init(&pool->lock);
3452         pool->id = -1;
3453         pool->cpu = -1;
3454         pool->node = NUMA_NO_NODE;
3455         pool->flags |= POOL_DISASSOCIATED;
3456         INIT_LIST_HEAD(&pool->worklist);
3457         INIT_LIST_HEAD(&pool->idle_list);
3458         hash_init(pool->busy_hash);
3459
3460         init_timer_deferrable(&pool->idle_timer);
3461         pool->idle_timer.function = idle_worker_timeout;
3462         pool->idle_timer.data = (unsigned long)pool;
3463
3464         setup_timer(&pool->mayday_timer, pool_mayday_timeout,
3465                     (unsigned long)pool);
3466
3467         mutex_init(&pool->manager_arb);
3468         mutex_init(&pool->manager_mutex);
3469         idr_init(&pool->worker_idr);
3470
3471         INIT_HLIST_NODE(&pool->hash_node);
3472         pool->refcnt = 1;
3473
3474         /* shouldn't fail above this point */
3475         pool->attrs = alloc_workqueue_attrs(GFP_KERNEL);
3476         if (!pool->attrs)
3477                 return -ENOMEM;
3478         return 0;
3479 }
3480
3481 static void rcu_free_pool(struct rcu_head *rcu)
3482 {
3483         struct worker_pool *pool = container_of(rcu, struct worker_pool, rcu);
3484
3485         idr_destroy(&pool->worker_idr);
3486         free_workqueue_attrs(pool->attrs);
3487         kfree(pool);
3488 }
3489
3490 /**
3491  * put_unbound_pool - put a worker_pool
3492  * @pool: worker_pool to put
3493  *
3494  * Put @pool.  If its refcnt reaches zero, it gets destroyed in sched-RCU
3495  * safe manner.  get_unbound_pool() calls this function on its failure path
3496  * and this function should be able to release pools which went through,
3497  * successfully or not, init_worker_pool().
3498  *
3499  * Should be called with wq_pool_mutex held.
3500  */
3501 static void put_unbound_pool(struct worker_pool *pool)
3502 {
3503         struct worker *worker;
3504
3505         lockdep_assert_held(&wq_pool_mutex);
3506
3507         if (--pool->refcnt)
3508                 return;
3509
3510         /* sanity checks */
3511         if (WARN_ON(!(pool->flags & POOL_DISASSOCIATED)) ||
3512             WARN_ON(!list_empty(&pool->worklist)))
3513                 return;
3514
3515         /* release id and unhash */
3516         if (pool->id >= 0)
3517                 idr_remove(&worker_pool_idr, pool->id);
3518         hash_del(&pool->hash_node);
3519
3520         /*
3521          * Become the manager and destroy all workers.  Grabbing
3522          * manager_arb prevents @pool's workers from blocking on
3523          * manager_mutex.
3524          */
3525         mutex_lock(&pool->manager_arb);
3526         mutex_lock(&pool->manager_mutex);
3527         spin_lock_irq(&pool->lock);
3528
3529         while ((worker = first_worker(pool)))
3530                 destroy_worker(worker);
3531         WARN_ON(pool->nr_workers || pool->nr_idle);
3532
3533         spin_unlock_irq(&pool->lock);
3534         mutex_unlock(&pool->manager_mutex);
3535         mutex_unlock(&pool->manager_arb);
3536
3537         /* shut down the timers */
3538         del_timer_sync(&pool->idle_timer);
3539         del_timer_sync(&pool->mayday_timer);
3540
3541         /* sched-RCU protected to allow dereferences from get_work_pool() */
3542         call_rcu_sched(&pool->rcu, rcu_free_pool);
3543 }
3544
3545 /**
3546  * get_unbound_pool - get a worker_pool with the specified attributes
3547  * @attrs: the attributes of the worker_pool to get
3548  *
3549  * Obtain a worker_pool which has the same attributes as @attrs, bump the
3550  * reference count and return it.  If there already is a matching
3551  * worker_pool, it will be used; otherwise, this function attempts to
3552  * create a new one.  On failure, returns NULL.
3553  *
3554  * Should be called with wq_pool_mutex held.
3555  */
3556 static struct worker_pool *get_unbound_pool(const struct workqueue_attrs *attrs)
3557 {
3558         u32 hash = wqattrs_hash(attrs);
3559         struct worker_pool *pool;
3560         int node;
3561
3562         lockdep_assert_held(&wq_pool_mutex);
3563
3564         /* do we already have a matching pool? */
3565         hash_for_each_possible(unbound_pool_hash, pool, hash_node, hash) {
3566                 if (wqattrs_equal(pool->attrs, attrs)) {
3567                         pool->refcnt++;
3568                         goto out_unlock;
3569                 }
3570         }
3571
3572         /* nope, create a new one */
3573         pool = kzalloc(sizeof(*pool), GFP_KERNEL);
3574         if (!pool || init_worker_pool(pool) < 0)
3575                 goto fail;
3576
3577         if (workqueue_freezing)
3578                 pool->flags |= POOL_FREEZING;
3579
3580         lockdep_set_subclass(&pool->lock, 1);   /* see put_pwq() */
3581         copy_workqueue_attrs(pool->attrs, attrs);
3582
3583         /*
3584          * no_numa isn't a worker_pool attribute, always clear it.  See
3585          * 'struct workqueue_attrs' comments for detail.
3586          */
3587         pool->attrs->no_numa = false;
3588
3589         /* if cpumask is contained inside a NUMA node, we belong to that node */
3590         if (wq_numa_enabled) {
3591                 for_each_node(node) {
3592                         if (cpumask_subset(pool->attrs->cpumask,
3593                                            wq_numa_possible_cpumask[node])) {
3594                                 pool->node = node;
3595                                 break;
3596                         }
3597                 }
3598         }
3599
3600         if (worker_pool_assign_id(pool) < 0)
3601                 goto fail;
3602
3603         /* create and start the initial worker */
3604         if (create_and_start_worker(pool) < 0)
3605                 goto fail;
3606
3607         /* install */
3608         hash_add(unbound_pool_hash, &pool->hash_node, hash);
3609 out_unlock:
3610         return pool;
3611 fail:
3612         if (pool)
3613                 put_unbound_pool(pool);
3614         return NULL;
3615 }
3616
3617 static void rcu_free_pwq(struct rcu_head *rcu)
3618 {
3619         kmem_cache_free(pwq_cache,
3620                         container_of(rcu, struct pool_workqueue, rcu));
3621 }
3622
3623 /*
3624  * Scheduled on system_wq by put_pwq() when an unbound pwq hits zero refcnt
3625  * and needs to be destroyed.
3626  */
3627 static void pwq_unbound_release_workfn(struct work_struct *work)
3628 {
3629         struct pool_workqueue *pwq = container_of(work, struct pool_workqueue,
3630                                                   unbound_release_work);
3631         struct workqueue_struct *wq = pwq->wq;
3632         struct worker_pool *pool = pwq->pool;
3633         bool is_last;
3634
3635         if (WARN_ON_ONCE(!(wq->flags & WQ_UNBOUND)))
3636                 return;
3637
3638         /*
3639          * Unlink @pwq.  Synchronization against wq->mutex isn't strictly
3640          * necessary on release but do it anyway.  It's easier to verify
3641          * and consistent with the linking path.
3642          */
3643         mutex_lock(&wq->mutex);
3644         list_del_rcu(&pwq->pwqs_node);
3645         is_last = list_empty(&wq->pwqs);
3646         mutex_unlock(&wq->mutex);
3647
3648         mutex_lock(&wq_pool_mutex);
3649         put_unbound_pool(pool);
3650         mutex_unlock(&wq_pool_mutex);
3651
3652         call_rcu_sched(&pwq->rcu, rcu_free_pwq);
3653
3654         /*
3655          * If we're the last pwq going away, @wq is already dead and no one
3656          * is gonna access it anymore.  Free it.
3657          */
3658         if (is_last) {
3659                 free_workqueue_attrs(wq->unbound_attrs);
3660                 kfree(wq);
3661         }
3662 }
3663
3664 /**
3665  * pwq_adjust_max_active - update a pwq's max_active to the current setting
3666  * @pwq: target pool_workqueue
3667  *
3668  * If @pwq isn't freezing, set @pwq->max_active to the associated
3669  * workqueue's saved_max_active and activate delayed work items
3670  * accordingly.  If @pwq is freezing, clear @pwq->max_active to zero.
3671  */
3672 static void pwq_adjust_max_active(struct pool_workqueue *pwq)
3673 {
3674         struct workqueue_struct *wq = pwq->wq;
3675         bool freezable = wq->flags & WQ_FREEZABLE;
3676
3677         /* for @wq->saved_max_active */
3678         lockdep_assert_held(&wq->mutex);
3679
3680         /* fast exit for non-freezable wqs */
3681         if (!freezable && pwq->max_active == wq->saved_max_active)
3682                 return;
3683
3684         spin_lock_irq(&pwq->pool->lock);
3685
3686         if (!freezable || !(pwq->pool->flags & POOL_FREEZING)) {
3687                 pwq->max_active = wq->saved_max_active;
3688
3689                 while (!list_empty(&pwq->delayed_works) &&
3690                        pwq->nr_active < pwq->max_active)
3691                         pwq_activate_first_delayed(pwq);
3692
3693                 /*
3694                  * Need to kick a worker after thawed or an unbound wq's
3695                  * max_active is bumped.  It's a slow path.  Do it always.
3696                  */
3697                 wake_up_worker(pwq->pool);
3698         } else {
3699                 pwq->max_active = 0;
3700         }
3701
3702         spin_unlock_irq(&pwq->pool->lock);
3703 }
3704
3705 /* initialize newly alloced @pwq which is associated with @wq and @pool */
3706 static void init_pwq(struct pool_workqueue *pwq, struct workqueue_struct *wq,
3707                      struct worker_pool *pool)
3708 {
3709         BUG_ON((unsigned long)pwq & WORK_STRUCT_FLAG_MASK);
3710
3711         memset(pwq, 0, sizeof(*pwq));
3712
3713         pwq->pool = pool;
3714         pwq->wq = wq;
3715         pwq->flush_color = -1;
3716         pwq->refcnt = 1;
3717         INIT_LIST_HEAD(&pwq->delayed_works);
3718         INIT_LIST_HEAD(&pwq->pwqs_node);
3719         INIT_LIST_HEAD(&pwq->mayday_node);
3720         INIT_WORK(&pwq->unbound_release_work, pwq_unbound_release_workfn);
3721 }
3722
3723 /* sync @pwq with the current state of its associated wq and link it */
3724 static void link_pwq(struct pool_workqueue *pwq)
3725 {
3726         struct workqueue_struct *wq = pwq->wq;
3727
3728         lockdep_assert_held(&wq->mutex);
3729
3730         /* may be called multiple times, ignore if already linked */
3731         if (!list_empty(&pwq->pwqs_node))
3732                 return;
3733
3734         /*
3735          * Set the matching work_color.  This is synchronized with
3736          * wq->mutex to avoid confusing flush_workqueue().
3737          */
3738         pwq->work_color = wq->work_color;
3739
3740         /* sync max_active to the current setting */
3741         pwq_adjust_max_active(pwq);
3742
3743         /* link in @pwq */
3744         list_add_rcu(&pwq->pwqs_node, &wq->pwqs);
3745 }
3746
3747 /* obtain a pool matching @attr and create a pwq associating the pool and @wq */
3748 static struct pool_workqueue *alloc_unbound_pwq(struct workqueue_struct *wq,
3749                                         const struct workqueue_attrs *attrs)
3750 {
3751         struct worker_pool *pool;
3752         struct pool_workqueue *pwq;
3753
3754         lockdep_assert_held(&wq_pool_mutex);
3755
3756         pool = get_unbound_pool(attrs);
3757         if (!pool)
3758                 return NULL;
3759
3760         pwq = kmem_cache_alloc_node(pwq_cache, GFP_KERNEL, pool->node);
3761         if (!pwq) {
3762                 put_unbound_pool(pool);
3763                 return NULL;
3764         }
3765
3766         init_pwq(pwq, wq, pool);
3767         return pwq;
3768 }
3769
3770 /* undo alloc_unbound_pwq(), used only in the error path */
3771 static void free_unbound_pwq(struct pool_workqueue *pwq)
3772 {
3773         lockdep_assert_held(&wq_pool_mutex);
3774
3775         if (pwq) {
3776                 put_unbound_pool(pwq->pool);
3777                 kmem_cache_free(pwq_cache, pwq);
3778         }
3779 }
3780
3781 /**
3782  * wq_calc_node_mask - calculate a wq_attrs' cpumask for the specified node
3783  * @attrs: the wq_attrs of interest
3784  * @node: the target NUMA node
3785  * @cpu_going_down: if >= 0, the CPU to consider as offline
3786  * @cpumask: outarg, the resulting cpumask
3787  *
3788  * Calculate the cpumask a workqueue with @attrs should use on @node.  If
3789  * @cpu_going_down is >= 0, that cpu is considered offline during
3790  * calculation.  The result is stored in @cpumask.  This function returns
3791  * %true if the resulting @cpumask is different from @attrs->cpumask,
3792  * %false if equal.
3793  *
3794  * If NUMA affinity is not enabled, @attrs->cpumask is always used.  If
3795  * enabled and @node has online CPUs requested by @attrs, the returned
3796  * cpumask is the intersection of the possible CPUs of @node and
3797  * @attrs->cpumask.
3798  *
3799  * The caller is responsible for ensuring that the cpumask of @node stays
3800  * stable.
3801  */
3802 static bool wq_calc_node_cpumask(const struct workqueue_attrs *attrs, int node,
3803                                  int cpu_going_down, cpumask_t *cpumask)
3804 {
3805         if (!wq_numa_enabled || attrs->no_numa)
3806                 goto use_dfl;
3807
3808         /* does @node have any online CPUs @attrs wants? */
3809         cpumask_and(cpumask, cpumask_of_node(node), attrs->cpumask);
3810         if (cpu_going_down >= 0)
3811                 cpumask_clear_cpu(cpu_going_down, cpumask);
3812
3813         if (cpumask_empty(cpumask))
3814                 goto use_dfl;
3815
3816         /* yeap, return possible CPUs in @node that @attrs wants */
3817         cpumask_and(cpumask, attrs->cpumask, wq_numa_possible_cpumask[node]);
3818         return !cpumask_equal(cpumask, attrs->cpumask);
3819
3820 use_dfl:
3821         cpumask_copy(cpumask, attrs->cpumask);
3822         return false;
3823 }
3824
3825 /* install @pwq into @wq's numa_pwq_tbl[] for @node and return the old pwq */
3826 static struct pool_workqueue *numa_pwq_tbl_install(struct workqueue_struct *wq,
3827                                                    int node,
3828                                                    struct pool_workqueue *pwq)
3829 {
3830         struct pool_workqueue *old_pwq;
3831
3832         lockdep_assert_held(&wq->mutex);
3833
3834         /* link_pwq() can handle duplicate calls */
3835         link_pwq(pwq);
3836
3837         old_pwq = rcu_access_pointer(wq->numa_pwq_tbl[node]);
3838         rcu_assign_pointer(wq->numa_pwq_tbl[node], pwq);
3839         return old_pwq;
3840 }
3841
3842 /**
3843  * apply_workqueue_attrs - apply new workqueue_attrs to an unbound workqueue
3844  * @wq: the target workqueue
3845  * @attrs: the workqueue_attrs to apply, allocated with alloc_workqueue_attrs()
3846  *
3847  * Apply @attrs to an unbound workqueue @wq.  Unless disabled, on NUMA
3848  * machines, this function maps a separate pwq to each NUMA node with
3849  * possibles CPUs in @attrs->cpumask so that work items are affine to the
3850  * NUMA node it was issued on.  Older pwqs are released as in-flight work
3851  * items finish.  Note that a work item which repeatedly requeues itself
3852  * back-to-back will stay on its current pwq.
3853  *
3854  * Performs GFP_KERNEL allocations.  Returns 0 on success and -errno on
3855  * failure.
3856  */
3857 int apply_workqueue_attrs(struct workqueue_struct *wq,
3858                           const struct workqueue_attrs *attrs)
3859 {
3860         struct workqueue_attrs *new_attrs, *tmp_attrs;
3861         struct pool_workqueue **pwq_tbl, *dfl_pwq;
3862         int node, ret;
3863
3864         /* only unbound workqueues can change attributes */
3865         if (WARN_ON(!(wq->flags & WQ_UNBOUND)))
3866                 return -EINVAL;
3867
3868         /* creating multiple pwqs breaks ordering guarantee */
3869         if (WARN_ON((wq->flags & __WQ_ORDERED) && !list_empty(&wq->pwqs)))
3870                 return -EINVAL;
3871
3872         pwq_tbl = kzalloc(wq_numa_tbl_len * sizeof(pwq_tbl[0]), GFP_KERNEL);
3873         new_attrs = alloc_workqueue_attrs(GFP_KERNEL);
3874         tmp_attrs = alloc_workqueue_attrs(GFP_KERNEL);
3875         if (!pwq_tbl || !new_attrs || !tmp_attrs)
3876                 goto enomem;
3877
3878         /* make a copy of @attrs and sanitize it */
3879         copy_workqueue_attrs(new_attrs, attrs);
3880         cpumask_and(new_attrs->cpumask, new_attrs->cpumask, cpu_possible_mask);
3881
3882         /*
3883          * We may create multiple pwqs with differing cpumasks.  Make a
3884          * copy of @new_attrs which will be modified and used to obtain
3885          * pools.
3886          */
3887         copy_workqueue_attrs(tmp_attrs, new_attrs);
3888
3889         /*
3890          * CPUs should stay stable across pwq creations and installations.
3891          * Pin CPUs, determine the target cpumask for each node and create
3892          * pwqs accordingly.
3893          */
3894         get_online_cpus();
3895
3896         mutex_lock(&wq_pool_mutex);
3897
3898         /*
3899          * If something goes wrong during CPU up/down, we'll fall back to
3900          * the default pwq covering whole @attrs->cpumask.  Always create
3901          * it even if we don't use it immediately.
3902          */
3903         dfl_pwq = alloc_unbound_pwq(wq, new_attrs);
3904         if (!dfl_pwq)
3905                 goto enomem_pwq;
3906
3907         for_each_node(node) {
3908                 if (wq_calc_node_cpumask(attrs, node, -1, tmp_attrs->cpumask)) {
3909                         pwq_tbl[node] = alloc_unbound_pwq(wq, tmp_attrs);
3910                         if (!pwq_tbl[node])
3911                                 goto enomem_pwq;
3912                 } else {
3913                         dfl_pwq->refcnt++;
3914                         pwq_tbl[node] = dfl_pwq;
3915                 }
3916         }
3917
3918         mutex_unlock(&wq_pool_mutex);
3919
3920         /* all pwqs have been created successfully, let's install'em */
3921         mutex_lock(&wq->mutex);
3922
3923         copy_workqueue_attrs(wq->unbound_attrs, new_attrs);
3924
3925         /* save the previous pwq and install the new one */
3926         for_each_node(node)
3927                 pwq_tbl[node] = numa_pwq_tbl_install(wq, node, pwq_tbl[node]);
3928
3929         /* @dfl_pwq might not have been used, ensure it's linked */
3930         link_pwq(dfl_pwq);
3931         swap(wq->dfl_pwq, dfl_pwq);
3932
3933         mutex_unlock(&wq->mutex);
3934
3935         /* put the old pwqs */
3936         for_each_node(node)
3937                 put_pwq_unlocked(pwq_tbl[node]);
3938         put_pwq_unlocked(dfl_pwq);
3939
3940         put_online_cpus();
3941         ret = 0;
3942         /* fall through */
3943 out_free:
3944         free_workqueue_attrs(tmp_attrs);
3945         free_workqueue_attrs(new_attrs);
3946         kfree(pwq_tbl);
3947         return ret;
3948
3949 enomem_pwq:
3950         free_unbound_pwq(dfl_pwq);
3951         for_each_node(node)
3952                 if (pwq_tbl && pwq_tbl[node] != dfl_pwq)
3953                         free_unbound_pwq(pwq_tbl[node]);
3954         mutex_unlock(&wq_pool_mutex);
3955         put_online_cpus();
3956 enomem:
3957         ret = -ENOMEM;
3958         goto out_free;
3959 }
3960
3961 /**
3962  * wq_update_unbound_numa - update NUMA affinity of a wq for CPU hot[un]plug
3963  * @wq: the target workqueue
3964  * @cpu: the CPU coming up or going down
3965  * @online: whether @cpu is coming up or going down
3966  *
3967  * This function is to be called from %CPU_DOWN_PREPARE, %CPU_ONLINE and
3968  * %CPU_DOWN_FAILED.  @cpu is being hot[un]plugged, update NUMA affinity of
3969  * @wq accordingly.
3970  *
3971  * If NUMA affinity can't be adjusted due to memory allocation failure, it
3972  * falls back to @wq->dfl_pwq which may not be optimal but is always
3973  * correct.
3974  *
3975  * Note that when the last allowed CPU of a NUMA node goes offline for a
3976  * workqueue with a cpumask spanning multiple nodes, the workers which were
3977  * already executing the work items for the workqueue will lose their CPU
3978  * affinity and may execute on any CPU.  This is similar to how per-cpu
3979  * workqueues behave on CPU_DOWN.  If a workqueue user wants strict
3980  * affinity, it's the user's responsibility to flush the work item from
3981  * CPU_DOWN_PREPARE.
3982  */
3983 static void wq_update_unbound_numa(struct workqueue_struct *wq, int cpu,
3984                                    bool online)
3985 {
3986         int node = cpu_to_node(cpu);
3987         int cpu_off = online ? -1 : cpu;
3988         struct pool_workqueue *old_pwq = NULL, *pwq;
3989         struct workqueue_attrs *target_attrs;
3990         cpumask_t *cpumask;
3991
3992         lockdep_assert_held(&wq_pool_mutex);
3993
3994         if (!wq_numa_enabled || !(wq->flags & WQ_UNBOUND))
3995                 return;
3996
3997         /*
3998          * We don't wanna alloc/free wq_attrs for each wq for each CPU.
3999          * Let's use a preallocated one.  The following buf is protected by
4000          * CPU hotplug exclusion.
4001          */
4002         target_attrs = wq_update_unbound_numa_attrs_buf;
4003         cpumask = target_attrs->cpumask;
4004
4005         mutex_lock(&wq->mutex);
4006         if (wq->unbound_attrs->no_numa)
4007                 goto out_unlock;
4008
4009         copy_workqueue_attrs(target_attrs, wq->unbound_attrs);
4010         pwq = unbound_pwq_by_node(wq, node);
4011
4012         /*
4013          * Let's determine what needs to be done.  If the target cpumask is
4014          * different from wq's, we need to compare it to @pwq's and create
4015          * a new one if they don't match.  If the target cpumask equals
4016          * wq's, the default pwq should be used.  If @pwq is already the
4017          * default one, nothing to do; otherwise, install the default one.
4018          */
4019         if (wq_calc_node_cpumask(wq->unbound_attrs, node, cpu_off, cpumask)) {
4020                 if (cpumask_equal(cpumask, pwq->pool->attrs->cpumask))
4021                         goto out_unlock;
4022         } else {
4023                 if (pwq == wq->dfl_pwq)
4024                         goto out_unlock;
4025                 else
4026                         goto use_dfl_pwq;
4027         }
4028
4029         mutex_unlock(&wq->mutex);
4030
4031         /* create a new pwq */
4032         pwq = alloc_unbound_pwq(wq, target_attrs);
4033         if (!pwq) {
4034                 pr_warning("workqueue: allocation failed while updating NUMA affinity of \"%s\"\n",
4035                            wq->name);
4036                 goto out_unlock;
4037         }
4038
4039         /*
4040          * Install the new pwq.  As this function is called only from CPU
4041          * hotplug callbacks and applying a new attrs is wrapped with
4042          * get/put_online_cpus(), @wq->unbound_attrs couldn't have changed
4043          * inbetween.
4044          */
4045         mutex_lock(&wq->mutex);
4046         old_pwq = numa_pwq_tbl_install(wq, node, pwq);
4047         goto out_unlock;
4048
4049 use_dfl_pwq:
4050         spin_lock_irq(&wq->dfl_pwq->pool->lock);
4051         get_pwq(wq->dfl_pwq);
4052         spin_unlock_irq(&wq->dfl_pwq->pool->lock);
4053         old_pwq = numa_pwq_tbl_install(wq, node, wq->dfl_pwq);
4054 out_unlock:
4055         mutex_unlock(&wq->mutex);
4056         put_pwq_unlocked(old_pwq);
4057 }
4058
4059 static int alloc_and_link_pwqs(struct workqueue_struct *wq)
4060 {
4061         bool highpri = wq->flags & WQ_HIGHPRI;
4062         int cpu;
4063
4064         if (!(wq->flags & WQ_UNBOUND)) {
4065                 wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);
4066                 if (!wq->cpu_pwqs)
4067                         return -ENOMEM;
4068
4069                 for_each_possible_cpu(cpu) {
4070                         struct pool_workqueue *pwq =
4071                                 per_cpu_ptr(wq->cpu_pwqs, cpu);
4072                         struct worker_pool *cpu_pools =
4073                                 per_cpu(cpu_worker_pools, cpu);
4074
4075                         init_pwq(pwq, wq, &cpu_pools[highpri]);
4076
4077                         mutex_lock(&wq->mutex);
4078                         link_pwq(pwq);
4079                         mutex_unlock(&wq->mutex);
4080                 }
4081                 return 0;
4082         } else {
4083                 return apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
4084         }
4085 }
4086
4087 static int wq_clamp_max_active(int max_active, unsigned int flags,
4088                                const char *name)
4089 {
4090         int lim = flags & WQ_UNBOUND ? WQ_UNBOUND_MAX_ACTIVE : WQ_MAX_ACTIVE;
4091
4092         if (max_active < 1 || max_active > lim)
4093                 pr_warn("workqueue: max_active %d requested for %s is out of range, clamping between %d and %d\n",
4094                         max_active, name, 1, lim);
4095
4096         return clamp_val(max_active, 1, lim);
4097 }
4098
4099 struct workqueue_struct *__alloc_workqueue_key(const char *fmt,
4100                                                unsigned int flags,
4101                                                int max_active,
4102                                                struct lock_class_key *key,
4103                                                const char *lock_name, ...)
4104 {
4105         size_t tbl_size = 0;
4106         va_list args;
4107         struct workqueue_struct *wq;
4108         struct pool_workqueue *pwq;
4109
4110         /* allocate wq and format name */
4111         if (flags & WQ_UNBOUND)
4112                 tbl_size = wq_numa_tbl_len * sizeof(wq->numa_pwq_tbl[0]);
4113
4114         wq = kzalloc(sizeof(*wq) + tbl_size, GFP_KERNEL);
4115         if (!wq)
4116                 return NULL;
4117
4118         if (flags & WQ_UNBOUND) {
4119                 wq->unbound_attrs = alloc_workqueue_attrs(GFP_KERNEL);
4120                 if (!wq->unbound_attrs)
4121                         goto err_free_wq;
4122         }
4123
4124         va_start(args, lock_name);
4125         vsnprintf(wq->name, sizeof(wq->name), fmt, args);
4126         va_end(args);
4127
4128         max_active = max_active ?: WQ_DFL_ACTIVE;
4129         max_active = wq_clamp_max_active(max_active, flags, wq->name);
4130
4131         /* init wq */
4132         wq->flags = flags;
4133         wq->saved_max_active = max_active;
4134         mutex_init(&wq->mutex);
4135         atomic_set(&wq->nr_pwqs_to_flush, 0);
4136         INIT_LIST_HEAD(&wq->pwqs);
4137         INIT_LIST_HEAD(&wq->flusher_queue);
4138         INIT_LIST_HEAD(&wq->flusher_overflow);
4139         INIT_LIST_HEAD(&wq->maydays);
4140
4141         lockdep_init_map(&wq->lockdep_map, lock_name, key, 0);
4142         INIT_LIST_HEAD(&wq->list);
4143
4144         if (alloc_and_link_pwqs(wq) < 0)
4145                 goto err_free_wq;
4146
4147         /*
4148          * Workqueues which may be used during memory reclaim should
4149          * have a rescuer to guarantee forward progress.
4150          */
4151         if (flags & WQ_MEM_RECLAIM) {
4152                 struct worker *rescuer;
4153
4154                 rescuer = alloc_worker();
4155                 if (!rescuer)
4156                         goto err_destroy;
4157
4158                 rescuer->rescue_wq = wq;
4159                 rescuer->task = kthread_create(rescuer_thread, rescuer, "%s",
4160                                                wq->name);
4161                 if (IS_ERR(rescuer->task)) {
4162                         kfree(rescuer);
4163                         goto err_destroy;
4164                 }
4165
4166                 wq->rescuer = rescuer;
4167                 rescuer->task->flags |= PF_NO_SETAFFINITY;
4168                 wake_up_process(rescuer->task);
4169         }
4170
4171         if ((wq->flags & WQ_SYSFS) && workqueue_sysfs_register(wq))
4172                 goto err_destroy;
4173
4174         /*
4175          * wq_pool_mutex protects global freeze state and workqueues list.
4176          * Grab it, adjust max_active and add the new @wq to workqueues
4177          * list.
4178          */
4179         mutex_lock(&wq_pool_mutex);
4180
4181         mutex_lock(&wq->mutex);
4182         for_each_pwq(pwq, wq)
4183                 pwq_adjust_max_active(pwq);
4184         mutex_unlock(&wq->mutex);
4185
4186         list_add(&wq->list, &workqueues);
4187
4188         mutex_unlock(&wq_pool_mutex);
4189
4190         return wq;
4191
4192 err_free_wq:
4193         free_workqueue_attrs(wq->unbound_attrs);
4194         kfree(wq);
4195         return NULL;
4196 err_destroy:
4197         destroy_workqueue(wq);
4198         return NULL;
4199 }
4200 EXPORT_SYMBOL_GPL(__alloc_workqueue_key);
4201
4202 /**
4203  * destroy_workqueue - safely terminate a workqueue
4204  * @wq: target workqueue
4205  *
4206  * Safely destroy a workqueue. All work currently pending will be done first.
4207  */
4208 void destroy_workqueue(struct workqueue_struct *wq)
4209 {
4210         struct pool_workqueue *pwq;
4211         int node;
4212
4213         /* drain it before proceeding with destruction */
4214         drain_workqueue(wq);
4215
4216         /* sanity checks */
4217         mutex_lock(&wq->mutex);
4218         for_each_pwq(pwq, wq) {
4219                 int i;
4220
4221                 for (i = 0; i < WORK_NR_COLORS; i++) {
4222                         if (WARN_ON(pwq->nr_in_flight[i])) {
4223                                 mutex_unlock(&wq->mutex);
4224                                 return;
4225                         }
4226                 }
4227
4228                 if (WARN_ON((pwq != wq->dfl_pwq) && (pwq->refcnt > 1)) ||
4229                     WARN_ON(pwq->nr_active) ||
4230                     WARN_ON(!list_empty(&pwq->delayed_works))) {
4231                         mutex_unlock(&wq->mutex);
4232                         return;
4233                 }
4234         }
4235         mutex_unlock(&wq->mutex);
4236
4237         /*
4238          * wq list is used to freeze wq, remove from list after
4239          * flushing is complete in case freeze races us.
4240          */
4241         mutex_lock(&wq_pool_mutex);
4242         list_del_init(&wq->list);
4243         mutex_unlock(&wq_pool_mutex);
4244
4245         workqueue_sysfs_unregister(wq);
4246
4247         if (wq->rescuer) {
4248                 kthread_stop(wq->rescuer->task);
4249                 kfree(wq->rescuer);
4250                 wq->rescuer = NULL;
4251         }
4252
4253         if (!(wq->flags & WQ_UNBOUND)) {
4254                 /*
4255                  * The base ref is never dropped on per-cpu pwqs.  Directly
4256                  * free the pwqs and wq.
4257                  */
4258                 free_percpu(wq->cpu_pwqs);
4259                 kfree(wq);
4260         } else {
4261                 /*
4262                  * We're the sole accessor of @wq at this point.  Directly
4263                  * access numa_pwq_tbl[] and dfl_pwq to put the base refs.
4264                  * @wq will be freed when the last pwq is released.
4265                  */
4266                 for_each_node(node) {
4267                         pwq = rcu_access_pointer(wq->numa_pwq_tbl[node]);
4268                         RCU_INIT_POINTER(wq->numa_pwq_tbl[node], NULL);
4269                         put_pwq_unlocked(pwq);
4270                 }
4271
4272                 /*
4273                  * Put dfl_pwq.  @wq may be freed any time after dfl_pwq is
4274                  * put.  Don't access it afterwards.
4275                  */
4276                 pwq = wq->dfl_pwq;
4277                 wq->dfl_pwq = NULL;
4278                 put_pwq_unlocked(pwq);
4279         }
4280 }
4281 EXPORT_SYMBOL_GPL(destroy_workqueue);
4282
4283 /**
4284  * workqueue_set_max_active - adjust max_active of a workqueue
4285  * @wq: target workqueue
4286  * @max_active: new max_active value.
4287  *
4288  * Set max_active of @wq to @max_active.
4289  *
4290  * CONTEXT:
4291  * Don't call from IRQ context.
4292  */
4293 void workqueue_set_max_active(struct workqueue_struct *wq, int max_active)
4294 {
4295         struct pool_workqueue *pwq;
4296
4297         /* disallow meddling with max_active for ordered workqueues */
4298         if (WARN_ON(wq->flags & __WQ_ORDERED))
4299                 return;
4300
4301         max_active = wq_clamp_max_active(max_active, wq->flags, wq->name);
4302
4303         mutex_lock(&wq->mutex);
4304
4305         wq->saved_max_active = max_active;
4306
4307         for_each_pwq(pwq, wq)
4308                 pwq_adjust_max_active(pwq);
4309
4310         mutex_unlock(&wq->mutex);
4311 }
4312 EXPORT_SYMBOL_GPL(workqueue_set_max_active);
4313
4314 /**
4315  * current_is_workqueue_rescuer - is %current workqueue rescuer?
4316  *
4317  * Determine whether %current is a workqueue rescuer.  Can be used from
4318  * work functions to determine whether it's being run off the rescuer task.
4319  */
4320 bool current_is_workqueue_rescuer(void)
4321 {
4322         struct worker *worker = current_wq_worker();
4323
4324         return worker && worker->rescue_wq;
4325 }
4326
4327 /**
4328  * workqueue_congested - test whether a workqueue is congested
4329  * @cpu: CPU in question
4330  * @wq: target workqueue
4331  *
4332  * Test whether @wq's cpu workqueue for @cpu is congested.  There is
4333  * no synchronization around this function and the test result is
4334  * unreliable and only useful as advisory hints or for debugging.
4335  *
4336  * If @cpu is WORK_CPU_UNBOUND, the test is performed on the local CPU.
4337  * Note that both per-cpu and unbound workqueues may be associated with
4338  * multiple pool_workqueues which have separate congested states.  A
4339  * workqueue being congested on one CPU doesn't mean the workqueue is also
4340  * contested on other CPUs / NUMA nodes.
4341  *
4342  * RETURNS:
4343  * %true if congested, %false otherwise.
4344  */
4345 bool workqueue_congested(int cpu, struct workqueue_struct *wq)
4346 {
4347         struct pool_workqueue *pwq;
4348         bool ret;
4349
4350         rcu_read_lock_sched();
4351
4352         if (cpu == WORK_CPU_UNBOUND)
4353                 cpu = smp_processor_id();
4354
4355         if (!(wq->flags & WQ_UNBOUND))
4356                 pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
4357         else
4358                 pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
4359
4360         ret = !list_empty(&pwq->delayed_works);
4361         rcu_read_unlock_sched();
4362
4363         return ret;
4364 }
4365 EXPORT_SYMBOL_GPL(workqueue_congested);
4366
4367 /**
4368  * work_busy - test whether a work is currently pending or running
4369  * @work: the work to be tested
4370  *
4371  * Test whether @work is currently pending or running.  There is no
4372  * synchronization around this function and the test result is
4373  * unreliable and only useful as advisory hints or for debugging.
4374  *
4375  * RETURNS:
4376  * OR'd bitmask of WORK_BUSY_* bits.
4377  */
4378 unsigned int work_busy(struct work_struct *work)
4379 {
4380         struct worker_pool *pool;
4381         unsigned long flags;
4382         unsigned int ret = 0;
4383
4384         if (work_pending(work))
4385                 ret |= WORK_BUSY_PENDING;
4386
4387         local_irq_save(flags);
4388         pool = get_work_pool(work);
4389         if (pool) {
4390                 spin_lock(&pool->lock);
4391                 if (find_worker_executing_work(pool, work))
4392                         ret |= WORK_BUSY_RUNNING;
4393                 spin_unlock(&pool->lock);
4394         }
4395         local_irq_restore(flags);
4396
4397         return ret;
4398 }
4399 EXPORT_SYMBOL_GPL(work_busy);
4400
4401 /**
4402  * set_worker_desc - set description for the current work item
4403  * @fmt: printf-style format string
4404  * @...: arguments for the format string
4405  *
4406  * This function can be called by a running work function to describe what
4407  * the work item is about.  If the worker task gets dumped, this
4408  * information will be printed out together to help debugging.  The
4409  * description can be at most WORKER_DESC_LEN including the trailing '\0'.
4410  */
4411 void set_worker_desc(const char *fmt, ...)
4412 {
4413         struct worker *worker = current_wq_worker();
4414         va_list args;
4415
4416         if (worker) {
4417                 va_start(args, fmt);
4418                 vsnprintf(worker->desc, sizeof(worker->desc), fmt, args);
4419                 va_end(args);
4420                 worker->desc_valid = true;
4421         }
4422 }
4423
4424 /**
4425  * print_worker_info - print out worker information and description
4426  * @log_lvl: the log level to use when printing
4427  * @task: target task
4428  *
4429  * If @task is a worker and currently executing a work item, print out the
4430  * name of the workqueue being serviced and worker description set with
4431  * set_worker_desc() by the currently executing work item.
4432  *
4433  * This function can be safely called on any task as long as the
4434  * task_struct itself is accessible.  While safe, this function isn't
4435  * synchronized and may print out mixups or garbages of limited length.
4436  */
4437 void print_worker_info(const char *log_lvl, struct task_struct *task)
4438 {
4439         work_func_t *fn = NULL;
4440         char name[WQ_NAME_LEN] = { };
4441         char desc[WORKER_DESC_LEN] = { };
4442         struct pool_workqueue *pwq = NULL;
4443         struct workqueue_struct *wq = NULL;
4444         bool desc_valid = false;
4445         struct worker *worker;
4446
4447         if (!(task->flags & PF_WQ_WORKER))
4448                 return;
4449
4450         /*
4451          * This function is called without any synchronization and @task
4452          * could be in any state.  Be careful with dereferences.
4453          */
4454         worker = probe_kthread_data(task);
4455
4456         /*
4457          * Carefully copy the associated workqueue's workfn and name.  Keep
4458          * the original last '\0' in case the original contains garbage.
4459          */
4460         probe_kernel_read(&fn, &worker->current_func, sizeof(fn));
4461         probe_kernel_read(&pwq, &worker->current_pwq, sizeof(pwq));
4462         probe_kernel_read(&wq, &pwq->wq, sizeof(wq));
4463         probe_kernel_read(name, wq->name, sizeof(name) - 1);
4464
4465         /* copy worker description */
4466         probe_kernel_read(&desc_valid, &worker->desc_valid, sizeof(desc_valid));
4467         if (desc_valid)
4468                 probe_kernel_read(desc, worker->desc, sizeof(desc) - 1);
4469
4470         if (fn || name[0] || desc[0]) {
4471                 printk("%sWorkqueue: %s %pf", log_lvl, name, fn);
4472                 if (desc[0])
4473                         pr_cont(" (%s)", desc);
4474                 pr_cont("\n");
4475         }
4476 }
4477
4478 /*
4479  * CPU hotplug.
4480  *
4481  * There are two challenges in supporting CPU hotplug.  Firstly, there
4482  * are a lot of assumptions on strong associations among work, pwq and
4483  * pool which make migrating pending and scheduled works very
4484  * difficult to implement without impacting hot paths.  Secondly,
4485  * worker pools serve mix of short, long and very long running works making
4486  * blocked draining impractical.
4487  *
4488  * This is solved by allowing the pools to be disassociated from the CPU
4489  * running as an unbound one and allowing it to be reattached later if the
4490  * cpu comes back online.
4491  */
4492
4493 static void wq_unbind_fn(struct work_struct *work)
4494 {
4495         int cpu = smp_processor_id();
4496         struct worker_pool *pool;
4497         struct worker *worker;
4498         int wi;
4499
4500         for_each_cpu_worker_pool(pool, cpu) {
4501                 WARN_ON_ONCE(cpu != smp_processor_id());
4502
4503                 mutex_lock(&pool->manager_mutex);
4504                 spin_lock_irq(&pool->lock);
4505
4506                 /*
4507                  * We've blocked all manager operations.  Make all workers
4508                  * unbound and set DISASSOCIATED.  Before this, all workers
4509                  * except for the ones which are still executing works from
4510                  * before the last CPU down must be on the cpu.  After
4511                  * this, they may become diasporas.
4512                  */
4513                 for_each_pool_worker(worker, wi, pool)
4514                         worker->flags |= WORKER_UNBOUND;
4515
4516                 pool->flags |= POOL_DISASSOCIATED;
4517
4518                 spin_unlock_irq(&pool->lock);
4519                 mutex_unlock(&pool->manager_mutex);
4520
4521                 /*
4522                  * Call schedule() so that we cross rq->lock and thus can
4523                  * guarantee sched callbacks see the %WORKER_UNBOUND flag.
4524                  * This is necessary as scheduler callbacks may be invoked
4525                  * from other cpus.
4526                  */
4527                 schedule();
4528
4529                 /*
4530                  * Sched callbacks are disabled now.  Zap nr_running.
4531                  * After this, nr_running stays zero and need_more_worker()
4532                  * and keep_working() are always true as long as the
4533                  * worklist is not empty.  This pool now behaves as an
4534                  * unbound (in terms of concurrency management) pool which
4535                  * are served by workers tied to the pool.
4536                  */
4537                 atomic_set(&pool->nr_running, 0);
4538
4539                 /*
4540                  * With concurrency management just turned off, a busy
4541                  * worker blocking could lead to lengthy stalls.  Kick off
4542                  * unbound chain execution of currently pending work items.
4543                  */
4544                 spin_lock_irq(&pool->lock);
4545                 wake_up_worker(pool);
4546                 spin_unlock_irq(&pool->lock);
4547         }
4548 }
4549
4550 /**
4551  * rebind_workers - rebind all workers of a pool to the associated CPU
4552  * @pool: pool of interest
4553  *
4554  * @pool->cpu is coming online.  Rebind all workers to the CPU.
4555  */
4556 static void rebind_workers(struct worker_pool *pool)
4557 {
4558         struct worker *worker;
4559         int wi;
4560
4561         lockdep_assert_held(&pool->manager_mutex);
4562
4563         /*
4564          * Restore CPU affinity of all workers.  As all idle workers should
4565          * be on the run-queue of the associated CPU before any local
4566          * wake-ups for concurrency management happen, restore CPU affinty
4567          * of all workers first and then clear UNBOUND.  As we're called
4568          * from CPU_ONLINE, the following shouldn't fail.
4569          */
4570         for_each_pool_worker(worker, wi, pool)
4571                 WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task,
4572                                                   pool->attrs->cpumask) < 0);
4573
4574         spin_lock_irq(&pool->lock);
4575
4576         for_each_pool_worker(worker, wi, pool) {
4577                 unsigned int worker_flags = worker->flags;
4578
4579                 /*
4580                  * A bound idle worker should actually be on the runqueue
4581                  * of the associated CPU for local wake-ups targeting it to
4582                  * work.  Kick all idle workers so that they migrate to the
4583                  * associated CPU.  Doing this in the same loop as
4584                  * replacing UNBOUND with REBOUND is safe as no worker will
4585                  * be bound before @pool->lock is released.
4586                  */
4587                 if (worker_flags & WORKER_IDLE)
4588                         wake_up_process(worker->task);
4589
4590                 /*
4591                  * We want to clear UNBOUND but can't directly call
4592                  * worker_clr_flags() or adjust nr_running.  Atomically
4593                  * replace UNBOUND with another NOT_RUNNING flag REBOUND.
4594                  * @worker will clear REBOUND using worker_clr_flags() when
4595                  * it initiates the next execution cycle thus restoring
4596                  * concurrency management.  Note that when or whether
4597                  * @worker clears REBOUND doesn't affect correctness.
4598                  *
4599                  * ACCESS_ONCE() is necessary because @worker->flags may be
4600                  * tested without holding any lock in
4601                  * wq_worker_waking_up().  Without it, NOT_RUNNING test may
4602                  * fail incorrectly leading to premature concurrency
4603                  * management operations.
4604                  */
4605                 WARN_ON_ONCE(!(worker_flags & WORKER_UNBOUND));
4606                 worker_flags |= WORKER_REBOUND;
4607                 worker_flags &= ~WORKER_UNBOUND;
4608                 ACCESS_ONCE(worker->flags) = worker_flags;
4609         }
4610
4611         spin_unlock_irq(&pool->lock);
4612 }
4613
4614 /**
4615  * restore_unbound_workers_cpumask - restore cpumask of unbound workers
4616  * @pool: unbound pool of interest
4617  * @cpu: the CPU which is coming up
4618  *
4619  * An unbound pool may end up with a cpumask which doesn't have any online
4620  * CPUs.  When a worker of such pool get scheduled, the scheduler resets
4621  * its cpus_allowed.  If @cpu is in @pool's cpumask which didn't have any
4622  * online CPU before, cpus_allowed of all its workers should be restored.
4623  */
4624 static void restore_unbound_workers_cpumask(struct worker_pool *pool, int cpu)
4625 {
4626         static cpumask_t cpumask;
4627         struct worker *worker;
4628         int wi;
4629
4630         lockdep_assert_held(&pool->manager_mutex);
4631
4632         /* is @cpu allowed for @pool? */
4633         if (!cpumask_test_cpu(cpu, pool->attrs->cpumask))
4634                 return;
4635
4636         /* is @cpu the only online CPU? */
4637         cpumask_and(&cpumask, pool->attrs->cpumask, cpu_online_mask);
4638         if (cpumask_weight(&cpumask) != 1)
4639                 return;
4640
4641         /* as we're called from CPU_ONLINE, the following shouldn't fail */
4642         for_each_pool_worker(worker, wi, pool)
4643                 WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task,
4644                                                   pool->attrs->cpumask) < 0);
4645 }
4646
4647 /*
4648  * Workqueues should be brought up before normal priority CPU notifiers.
4649  * This will be registered high priority CPU notifier.
4650  */
4651 static int __cpuinit workqueue_cpu_up_callback(struct notifier_block *nfb,
4652                                                unsigned long action,
4653                                                void *hcpu)
4654 {
4655         int cpu = (unsigned long)hcpu;
4656         struct worker_pool *pool;
4657         struct workqueue_struct *wq;
4658         int pi;
4659
4660         switch (action & ~CPU_TASKS_FROZEN) {
4661         case CPU_UP_PREPARE:
4662                 for_each_cpu_worker_pool(pool, cpu) {
4663                         if (pool->nr_workers)
4664                                 continue;
4665                         if (create_and_start_worker(pool) < 0)
4666                                 return NOTIFY_BAD;
4667                 }
4668                 break;
4669
4670         case CPU_DOWN_FAILED:
4671         case CPU_ONLINE:
4672                 mutex_lock(&wq_pool_mutex);
4673
4674                 for_each_pool(pool, pi) {
4675                         mutex_lock(&pool->manager_mutex);
4676
4677                         if (pool->cpu == cpu) {
4678                                 spin_lock_irq(&pool->lock);
4679                                 pool->flags &= ~POOL_DISASSOCIATED;
4680                                 spin_unlock_irq(&pool->lock);
4681
4682                                 rebind_workers(pool);
4683                         } else if (pool->cpu < 0) {
4684                                 restore_unbound_workers_cpumask(pool, cpu);
4685                         }
4686
4687                         mutex_unlock(&pool->manager_mutex);
4688                 }
4689
4690                 /* update NUMA affinity of unbound workqueues */
4691                 list_for_each_entry(wq, &workqueues, list)
4692                         wq_update_unbound_numa(wq, cpu, true);
4693
4694                 mutex_unlock(&wq_pool_mutex);
4695                 break;
4696         }
4697         return NOTIFY_OK;
4698 }
4699
4700 /*
4701  * Workqueues should be brought down after normal priority CPU notifiers.
4702  * This will be registered as low priority CPU notifier.
4703  */
4704 static int __cpuinit workqueue_cpu_down_callback(struct notifier_block *nfb,
4705                                                  unsigned long action,
4706                                                  void *hcpu)
4707 {
4708         int cpu = (unsigned long)hcpu;
4709         struct work_struct unbind_work;
4710         struct workqueue_struct *wq;
4711
4712         switch (action & ~CPU_TASKS_FROZEN) {
4713         case CPU_DOWN_PREPARE:
4714                 /* unbinding per-cpu workers should happen on the local CPU */
4715                 INIT_WORK_ONSTACK(&unbind_work, wq_unbind_fn);
4716                 queue_work_on(cpu, system_highpri_wq, &unbind_work);
4717
4718                 /* update NUMA affinity of unbound workqueues */
4719                 mutex_lock(&wq_pool_mutex);
4720                 list_for_each_entry(wq, &workqueues, list)
4721                         wq_update_unbound_numa(wq, cpu, false);
4722                 mutex_unlock(&wq_pool_mutex);
4723
4724                 /* wait for per-cpu unbinding to finish */
4725                 flush_work(&unbind_work);
4726                 break;
4727         }
4728         return NOTIFY_OK;
4729 }
4730
4731 #ifdef CONFIG_SMP
4732
4733 struct work_for_cpu {
4734         struct work_struct work;
4735         long (*fn)(void *);
4736         void *arg;
4737         long ret;
4738 };
4739
4740 static void work_for_cpu_fn(struct work_struct *work)
4741 {
4742         struct work_for_cpu *wfc = container_of(work, struct work_for_cpu, work);
4743
4744         wfc->ret = wfc->fn(wfc->arg);
4745 }
4746
4747 /**
4748  * work_on_cpu - run a function in user context on a particular cpu
4749  * @cpu: the cpu to run on
4750  * @fn: the function to run
4751  * @arg: the function arg
4752  *
4753  * This will return the value @fn returns.
4754  * It is up to the caller to ensure that the cpu doesn't go offline.
4755  * The caller must not hold any locks which would prevent @fn from completing.
4756  */
4757 long work_on_cpu(int cpu, long (*fn)(void *), void *arg)
4758 {
4759         struct work_for_cpu wfc = { .fn = fn, .arg = arg };
4760
4761         INIT_WORK_ONSTACK(&wfc.work, work_for_cpu_fn);
4762         schedule_work_on(cpu, &wfc.work);
4763         flush_work(&wfc.work);
4764         return wfc.ret;
4765 }
4766 EXPORT_SYMBOL_GPL(work_on_cpu);
4767 #endif /* CONFIG_SMP */
4768
4769 #ifdef CONFIG_FREEZER
4770
4771 /**
4772  * freeze_workqueues_begin - begin freezing workqueues
4773  *
4774  * Start freezing workqueues.  After this function returns, all freezable
4775  * workqueues will queue new works to their delayed_works list instead of
4776  * pool->worklist.
4777  *
4778  * CONTEXT:
4779  * Grabs and releases wq_pool_mutex, wq->mutex and pool->lock's.
4780  */
4781 void freeze_workqueues_begin(void)
4782 {
4783         struct worker_pool *pool;
4784         struct workqueue_struct *wq;
4785         struct pool_workqueue *pwq;
4786         int pi;
4787
4788         mutex_lock(&wq_pool_mutex);
4789
4790         WARN_ON_ONCE(workqueue_freezing);
4791         workqueue_freezing = true;
4792
4793         /* set FREEZING */
4794         for_each_pool(pool, pi) {
4795                 spin_lock_irq(&pool->lock);
4796                 WARN_ON_ONCE(pool->flags & POOL_FREEZING);
4797                 pool->flags |= POOL_FREEZING;
4798                 spin_unlock_irq(&pool->lock);
4799         }
4800
4801         list_for_each_entry(wq, &workqueues, list) {
4802                 mutex_lock(&wq->mutex);
4803                 for_each_pwq(pwq, wq)
4804                         pwq_adjust_max_active(pwq);
4805                 mutex_unlock(&wq->mutex);
4806         }
4807
4808         mutex_unlock(&wq_pool_mutex);
4809 }
4810
4811 /**
4812  * freeze_workqueues_busy - are freezable workqueues still busy?
4813  *
4814  * Check whether freezing is complete.  This function must be called
4815  * between freeze_workqueues_begin() and thaw_workqueues().
4816  *
4817  * CONTEXT:
4818  * Grabs and releases wq_pool_mutex.
4819  *
4820  * RETURNS:
4821  * %true if some freezable workqueues are still busy.  %false if freezing
4822  * is complete.
4823  */
4824 bool freeze_workqueues_busy(void)
4825 {
4826         bool busy = false;
4827         struct workqueue_struct *wq;
4828         struct pool_workqueue *pwq;
4829
4830         mutex_lock(&wq_pool_mutex);
4831
4832         WARN_ON_ONCE(!workqueue_freezing);
4833
4834         list_for_each_entry(wq, &workqueues, list) {
4835                 if (!(wq->flags & WQ_FREEZABLE))
4836                         continue;
4837                 /*
4838                  * nr_active is monotonically decreasing.  It's safe
4839                  * to peek without lock.
4840                  */
4841                 rcu_read_lock_sched();
4842                 for_each_pwq(pwq, wq) {
4843                         WARN_ON_ONCE(pwq->nr_active < 0);
4844                         if (pwq->nr_active) {
4845                                 busy = true;
4846                                 rcu_read_unlock_sched();
4847                                 goto out_unlock;
4848                         }
4849                 }
4850                 rcu_read_unlock_sched();
4851         }
4852 out_unlock:
4853         mutex_unlock(&wq_pool_mutex);
4854         return busy;
4855 }
4856
4857 /**
4858  * thaw_workqueues - thaw workqueues
4859  *
4860  * Thaw workqueues.  Normal queueing is restored and all collected
4861  * frozen works are transferred to their respective pool worklists.
4862  *
4863  * CONTEXT:
4864  * Grabs and releases wq_pool_mutex, wq->mutex and pool->lock's.
4865  */
4866 void thaw_workqueues(void)
4867 {
4868         struct workqueue_struct *wq;
4869         struct pool_workqueue *pwq;
4870         struct worker_pool *pool;
4871         int pi;
4872
4873         mutex_lock(&wq_pool_mutex);
4874
4875         if (!workqueue_freezing)
4876                 goto out_unlock;
4877
4878         /* clear FREEZING */
4879         for_each_pool(pool, pi) {
4880                 spin_lock_irq(&pool->lock);
4881                 WARN_ON_ONCE(!(pool->flags & POOL_FREEZING));
4882                 pool->flags &= ~POOL_FREEZING;
4883                 spin_unlock_irq(&pool->lock);
4884         }
4885
4886         /* restore max_active and repopulate worklist */
4887         list_for_each_entry(wq, &workqueues, list) {
4888                 mutex_lock(&wq->mutex);
4889                 for_each_pwq(pwq, wq)
4890                         pwq_adjust_max_active(pwq);
4891                 mutex_unlock(&wq->mutex);
4892         }
4893
4894         workqueue_freezing = false;
4895 out_unlock:
4896         mutex_unlock(&wq_pool_mutex);
4897 }
4898 #endif /* CONFIG_FREEZER */
4899
4900 static void __init wq_numa_init(void)
4901 {
4902         cpumask_var_t *tbl;
4903         int node, cpu;
4904
4905         /* determine NUMA pwq table len - highest node id + 1 */
4906         for_each_node(node)
4907                 wq_numa_tbl_len = max(wq_numa_tbl_len, node + 1);
4908
4909         if (num_possible_nodes() <= 1)
4910                 return;
4911
4912         if (wq_disable_numa) {
4913                 pr_info("workqueue: NUMA affinity support disabled\n");
4914                 return;
4915         }
4916
4917         wq_update_unbound_numa_attrs_buf = alloc_workqueue_attrs(GFP_KERNEL);
4918         BUG_ON(!wq_update_unbound_numa_attrs_buf);
4919
4920         /*
4921          * We want masks of possible CPUs of each node which isn't readily
4922          * available.  Build one from cpu_to_node() which should have been
4923          * fully initialized by now.
4924          */
4925         tbl = kzalloc(wq_numa_tbl_len * sizeof(tbl[0]), GFP_KERNEL);
4926         BUG_ON(!tbl);
4927
4928         for_each_node(node)
4929                 BUG_ON(!alloc_cpumask_var_node(&tbl[node], GFP_KERNEL,
4930                                 node_online(node) ? node : NUMA_NO_NODE));
4931
4932         for_each_possible_cpu(cpu) {
4933                 node = cpu_to_node(cpu);
4934                 if (WARN_ON(node == NUMA_NO_NODE)) {
4935                         pr_warn("workqueue: NUMA node mapping not available for cpu%d, disabling NUMA support\n", cpu);
4936                         /* happens iff arch is bonkers, let's just proceed */
4937                         return;
4938                 }
4939                 cpumask_set_cpu(cpu, tbl[node]);
4940         }
4941
4942         wq_numa_possible_cpumask = tbl;
4943         wq_numa_enabled = true;
4944 }
4945
4946 static int __init init_workqueues(void)
4947 {
4948         int std_nice[NR_STD_WORKER_POOLS] = { 0, HIGHPRI_NICE_LEVEL };
4949         int i, cpu;
4950
4951         /* make sure we have enough bits for OFFQ pool ID */
4952         BUILD_BUG_ON((1LU << (BITS_PER_LONG - WORK_OFFQ_POOL_SHIFT)) <
4953                      WORK_CPU_END * NR_STD_WORKER_POOLS);
4954
4955         WARN_ON(__alignof__(struct pool_workqueue) < __alignof__(long long));
4956
4957         pwq_cache = KMEM_CACHE(pool_workqueue, SLAB_PANIC);
4958
4959         cpu_notifier(workqueue_cpu_up_callback, CPU_PRI_WORKQUEUE_UP);
4960         hotcpu_notifier(workqueue_cpu_down_callback, CPU_PRI_WORKQUEUE_DOWN);
4961
4962         wq_numa_init();
4963
4964         /* initialize CPU pools */
4965         for_each_possible_cpu(cpu) {
4966                 struct worker_pool *pool;
4967
4968                 i = 0;
4969                 for_each_cpu_worker_pool(pool, cpu) {
4970                         BUG_ON(init_worker_pool(pool));
4971                         pool->cpu = cpu;
4972                         cpumask_copy(pool->attrs->cpumask, cpumask_of(cpu));
4973                         pool->attrs->nice = std_nice[i++];
4974                         pool->node = cpu_to_node(cpu);
4975
4976                         /* alloc pool ID */
4977                         mutex_lock(&wq_pool_mutex);
4978                         BUG_ON(worker_pool_assign_id(pool));
4979                         mutex_unlock(&wq_pool_mutex);
4980                 }
4981         }
4982
4983         /* create the initial worker */
4984         for_each_online_cpu(cpu) {
4985                 struct worker_pool *pool;
4986
4987                 for_each_cpu_worker_pool(pool, cpu) {
4988                         pool->flags &= ~POOL_DISASSOCIATED;
4989                         BUG_ON(create_and_start_worker(pool) < 0);
4990                 }
4991         }
4992
4993         /* create default unbound wq attrs */
4994         for (i = 0; i < NR_STD_WORKER_POOLS; i++) {
4995                 struct workqueue_attrs *attrs;
4996
4997                 BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
4998                 attrs->nice = std_nice[i];
4999                 unbound_std_wq_attrs[i] = attrs;
5000         }
5001
5002         system_wq = alloc_workqueue("events", 0, 0);
5003         system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
5004         system_long_wq = alloc_workqueue("events_long", 0, 0);
5005         system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
5006                                             WQ_UNBOUND_MAX_ACTIVE);
5007         system_freezable_wq = alloc_workqueue("events_freezable",
5008                                               WQ_FREEZABLE, 0);
5009         BUG_ON(!system_wq || !system_highpri_wq || !system_long_wq ||
5010                !system_unbound_wq || !system_freezable_wq);
5011         return 0;
5012 }
5013 early_initcall(init_workqueues);