workqueue.c 125.7 KB
Newer Older
L
Linus Torvalds 已提交
1
/*
T
Tejun Heo 已提交
2
 * kernel/workqueue.c - generic async execution with shared worker pool
L
Linus Torvalds 已提交
3
 *
T
Tejun Heo 已提交
4
 * Copyright (C) 2002		Ingo Molnar
L
Linus Torvalds 已提交
5
 *
T
Tejun Heo 已提交
6 7 8 9 10
 *   Derived from the taskqueue/keventd code by:
 *     David Woodhouse <dwmw2@infradead.org>
 *     Andrew Morton
 *     Kai Petzke <wpp@marie.physik.tu-berlin.de>
 *     Theodore Ts'o <tytso@mit.edu>
L
Linus Torvalds 已提交
11
 *
T
Tejun Heo 已提交
12
 * Made to use alloc_percpu by Christoph Lameter.
L
Linus Torvalds 已提交
13
 *
T
Tejun Heo 已提交
14 15
 * Copyright (C) 2010		SUSE Linux Products GmbH
 * Copyright (C) 2010		Tejun Heo <tj@kernel.org>
16
 *
T
Tejun Heo 已提交
17 18 19 20 21 22 23
 * This is the generic async execution mechanism.  Work items as are
 * executed in process context.  The worker pool is shared and
 * automatically managed.  There is one worker pool for each CPU and
 * one extra for works which are better served by workers which are
 * not bound to any specific CPU.
 *
 * Please read Documentation/workqueue.txt for details.
L
Linus Torvalds 已提交
24 25
 */

26
#include <linux/export.h>
L
Linus Torvalds 已提交
27 28 29 30 31 32 33 34 35 36
#include <linux/kernel.h>
#include <linux/sched.h>
#include <linux/init.h>
#include <linux/signal.h>
#include <linux/completion.h>
#include <linux/workqueue.h>
#include <linux/slab.h>
#include <linux/cpu.h>
#include <linux/notifier.h>
#include <linux/kthread.h>
37
#include <linux/hardirq.h>
38
#include <linux/mempolicy.h>
39
#include <linux/freezer.h>
40 41
#include <linux/kallsyms.h>
#include <linux/debug_locks.h>
42
#include <linux/lockdep.h>
T
Tejun Heo 已提交
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

49
#include "workqueue_internal.h"
L
Linus Torvalds 已提交
50

T
Tejun Heo 已提交
51
enum {
52 53
	/*
	 * worker_pool flags
54
	 *
55
	 * A bound pool is either associated or disassociated with its CPU.
56 57 58 59 60 61
	 * While associated (!DISASSOCIATED), all workers are bound to the
	 * CPU and none has %WORKER_UNBOUND set and concurrency management
	 * is in effect.
	 *
	 * While DISASSOCIATED, the cpu may be offline and all workers have
	 * %WORKER_UNBOUND set and concurrency management disabled, and may
62
	 * be executing on any CPU.  The pool behaves as an unbound one.
63
	 *
64 65
	 * Note that DISASSOCIATED should be flipped only while holding
	 * manager_mutex to avoid changing binding state while
66
	 * create_worker() is in progress.
67
	 */
68
	POOL_MANAGE_WORKERS	= 1 << 0,	/* need to manage workers */
69
	POOL_DISASSOCIATED	= 1 << 2,	/* cpu can't serve workers */
70
	POOL_FREEZING		= 1 << 3,	/* freeze in progress */
71

T
Tejun Heo 已提交
72 73 74 75
	/* worker flags */
	WORKER_STARTED		= 1 << 0,	/* started */
	WORKER_DIE		= 1 << 1,	/* die die die */
	WORKER_IDLE		= 1 << 2,	/* is idle */
76
	WORKER_PREP		= 1 << 3,	/* preparing to run works */
77
	WORKER_CPU_INTENSIVE	= 1 << 6,	/* cpu intensive */
78
	WORKER_UNBOUND		= 1 << 7,	/* worker is unbound */
79
	WORKER_REBOUND		= 1 << 8,	/* worker was rebound */
80

81 82
	WORKER_NOT_RUNNING	= WORKER_PREP | WORKER_CPU_INTENSIVE |
				  WORKER_UNBOUND | WORKER_REBOUND,
83

84
	NR_STD_WORKER_POOLS	= 2,		/* # standard pools per cpu */
85

86
	UNBOUND_POOL_HASH_ORDER	= 6,		/* hashed by pool->attrs */
T
Tejun Heo 已提交
87
	BUSY_WORKER_HASH_ORDER	= 6,		/* 64 pointers */
88

89 90 91
	MAX_IDLE_WORKERS_RATIO	= 4,		/* 1/4 of busy can be idle */
	IDLE_WORKER_TIMEOUT	= 300 * HZ,	/* keep idle ones for 5 mins */

92 93 94
	MAYDAY_INITIAL_TIMEOUT  = HZ / 100 >= 2 ? HZ / 100 : 2,
						/* call for help after 10ms
						   (min two ticks) */
95 96 97 98 99 100 101 102
	MAYDAY_INTERVAL		= HZ / 10,	/* and then every 100ms */
	CREATE_COOLDOWN		= HZ,		/* time to breath after fail */

	/*
	 * Rescue workers are used only on emergencies and shared by
	 * all cpus.  Give -20.
	 */
	RESCUER_NICE_LEVEL	= -20,
103
	HIGHPRI_NICE_LEVEL	= -20,
104 105

	WQ_NAME_LEN		= 24,
T
Tejun Heo 已提交
106
};
L
Linus Torvalds 已提交
107 108

/*
T
Tejun Heo 已提交
109 110
 * Structure fields follow one of the following exclusion rules.
 *
111 112
 * I: Modifiable by initialization/destruction paths and read-only for
 *    everyone else.
T
Tejun Heo 已提交
113
 *
114 115 116
 * P: Preemption protected.  Disabling preemption is enough and should
 *    only be modified and accessed from the local cpu.
 *
117
 * L: pool->lock protected.  Access with pool->lock held.
T
Tejun Heo 已提交
118
 *
119 120 121 122
 * X: During normal operation, modification requires pool->lock and should
 *    be done only from local cpu.  Either disabling preemption on local
 *    cpu or grabbing pool->lock is enough for read access.  If
 *    POOL_DISASSOCIATED is set, it's identical to L.
123
 *
124 125 126
 * MG: pool->manager_mutex and pool->lock protected.  Writes require both
 *     locks.  Reads can happen under either lock.
 *
127
 * PL: wq_pool_mutex protected.
128
 *
129
 * PR: wq_pool_mutex protected for writes.  Sched-RCU protected for reads.
130
 *
131 132
 * WQ: wq->mutex protected.
 *
133
 * WR: wq->mutex protected for writes.  Sched-RCU protected for reads.
134 135
 *
 * MD: wq_mayday_lock protected.
L
Linus Torvalds 已提交
136 137
 */

138
/* struct worker is defined in workqueue_internal.h */
T
Tejun Heo 已提交
139

140
struct worker_pool {
141
	spinlock_t		lock;		/* the pool lock */
142
	int			cpu;		/* I: the associated cpu */
143
	int			node;		/* I: the associated node ID */
T
Tejun Heo 已提交
144
	int			id;		/* I: pool ID */
145
	unsigned int		flags;		/* X: flags */
146 147 148

	struct list_head	worklist;	/* L: list of pending works */
	int			nr_workers;	/* L: total number of workers */
149 150

	/* nr_idle includes the ones off idle_list for rebinding */
151 152 153 154 155 156
	int			nr_idle;	/* L: currently idle ones */

	struct list_head	idle_list;	/* X: list of idle workers */
	struct timer_list	idle_timer;	/* L: worker idle timeout */
	struct timer_list	mayday_timer;	/* L: SOS timer for workers */

157
	/* a workers is either on busy_hash or idle_list, or the manager */
158 159 160
	DECLARE_HASHTABLE(busy_hash, BUSY_WORKER_HASH_ORDER);
						/* L: hash of busy workers */

161
	/* see manage_workers() for details on the two manager mutexes */
162
	struct mutex		manager_arb;	/* manager arbitration */
163
	struct mutex		manager_mutex;	/* manager exclusion */
164
	struct idr		worker_idr;	/* MG: worker IDs and iteration */
165

T
Tejun Heo 已提交
166
	struct workqueue_attrs	*attrs;		/* I: worker attributes */
167 168
	struct hlist_node	hash_node;	/* PL: unbound_pool_hash node */
	int			refcnt;		/* PL: refcnt for unbound pools */
T
Tejun Heo 已提交
169

170 171 172 173 174 175
	/*
	 * The current concurrency level.  As it's likely to be accessed
	 * from other CPUs during try_to_wake_up(), put it in a separate
	 * cacheline.
	 */
	atomic_t		nr_running ____cacheline_aligned_in_smp;
176 177 178 179 180 181

	/*
	 * Destruction of pool is sched-RCU protected to allow dereferences
	 * from get_work_pool().
	 */
	struct rcu_head		rcu;
182 183
} ____cacheline_aligned_in_smp;

L
Linus Torvalds 已提交
184
/*
185 186 187 188
 * The per-pool workqueue.  While queued, the lower WORK_STRUCT_FLAG_BITS
 * of work_struct->data are used for flags and the remaining high bits
 * point to the pwq; thus, pwqs need to be aligned at two's power of the
 * number of flag bits.
L
Linus Torvalds 已提交
189
 */
190
struct pool_workqueue {
191
	struct worker_pool	*pool;		/* I: the associated pool */
T
Tejun Heo 已提交
192
	struct workqueue_struct *wq;		/* I: the owning workqueue */
193 194
	int			work_color;	/* L: current color */
	int			flush_color;	/* L: flushing color */
T
Tejun Heo 已提交
195
	int			refcnt;		/* L: reference count */
196 197
	int			nr_in_flight[WORK_NR_COLORS];
						/* L: nr of in_flight works */
198
	int			nr_active;	/* L: nr of active works */
199
	int			max_active;	/* L: max active works */
200
	struct list_head	delayed_works;	/* L: delayed works */
201
	struct list_head	pwqs_node;	/* WR: node on wq->pwqs */
202
	struct list_head	mayday_node;	/* MD: node on wq->maydays */
T
Tejun Heo 已提交
203 204 205 206 207

	/*
	 * Release of unbound pwq is punted to system_wq.  See put_pwq()
	 * and pwq_unbound_release_workfn() for details.  pool_workqueue
	 * itself is also sched-RCU protected so that the first pwq can be
208
	 * determined without grabbing wq->mutex.
T
Tejun Heo 已提交
209 210 211
	 */
	struct work_struct	unbound_release_work;
	struct rcu_head		rcu;
212
} __aligned(1 << WORK_STRUCT_FLAG_BITS);
L
Linus Torvalds 已提交
213

214 215 216 217
/*
 * Structure used to wait for workqueue flush.
 */
struct wq_flusher {
218 219
	struct list_head	list;		/* WQ: list of flushers */
	int			flush_color;	/* WQ: flush color waiting for */
220 221 222
	struct completion	done;		/* flush completion */
};

223 224
struct wq_device;

L
Linus Torvalds 已提交
225
/*
226 227
 * The externally visible workqueue.  It relays the issued work items to
 * the appropriate worker_pool through its pool_workqueues.
L
Linus Torvalds 已提交
228 229
 */
struct workqueue_struct {
230
	struct list_head	pwqs;		/* WR: all pwqs of this wq */
231
	struct list_head	list;		/* PL: list of all workqueues */
232

233 234 235
	struct mutex		mutex;		/* protects this wq */
	int			work_color;	/* WQ: current work color */
	int			flush_color;	/* WQ: current flush color */
236
	atomic_t		nr_pwqs_to_flush; /* flush in progress */
237 238 239
	struct wq_flusher	*first_flusher;	/* WQ: first flusher */
	struct list_head	flusher_queue;	/* WQ: flush waiters */
	struct list_head	flusher_overflow; /* WQ: flush overflow list */
240

241
	struct list_head	maydays;	/* MD: pwqs requesting rescue */
242 243
	struct worker		*rescuer;	/* I: rescue worker */

244
	int			nr_drainers;	/* WQ: drain in progress */
245
	int			saved_max_active; /* WQ: saved pwq max_active */
246

247 248
	struct workqueue_attrs	*unbound_attrs;	/* WQ: only for unbound wqs */

249 250 251
#ifdef CONFIG_SYSFS
	struct wq_device	*wq_dev;	/* I: for sysfs interface */
#endif
252
#ifdef CONFIG_LOCKDEP
T
Tejun Heo 已提交
253
	struct lockdep_map	lockdep_map;
254
#endif
255
	char			name[WQ_NAME_LEN]; /* I: workqueue name */
256 257 258 259

	/* hot fields used during command issue, aligned to cacheline */
	unsigned int		flags ____cacheline_aligned; /* WQ: WQ_* flags */
	struct pool_workqueue __percpu *cpu_pwqs; /* I: per-cpu pwqs */
260
	struct pool_workqueue __rcu *numa_pwq_tbl[]; /* FR: unbound pwqs indexed by node */
L
Linus Torvalds 已提交
261 262
};

263 264
static struct kmem_cache *pwq_cache;

265 266 267 268 269 270
static int wq_numa_tbl_len;		/* highest possible NUMA node id + 1 */
static cpumask_var_t *wq_numa_possible_cpumask;
					/* possible CPUs of each node */

static bool wq_numa_enabled;		/* unbound NUMA affinity enabled */

271
static DEFINE_MUTEX(wq_pool_mutex);	/* protects pools and workqueues list */
272
static DEFINE_SPINLOCK(wq_mayday_lock);	/* protects wq->maydays list */
273

274 275
static LIST_HEAD(workqueues);		/* PL: list of all workqueues */
static bool workqueue_freezing;		/* PL: have wqs started freezing? */
276 277 278 279 280

/* the per-cpu worker pools */
static DEFINE_PER_CPU_SHARED_ALIGNED(struct worker_pool [NR_STD_WORKER_POOLS],
				     cpu_worker_pools);

281
static DEFINE_IDR(worker_pool_idr);	/* PR: idr of all pools */
282

283
/* PL: hash of all unbound pools keyed by pool->attrs */
284 285
static DEFINE_HASHTABLE(unbound_pool_hash, UNBOUND_POOL_HASH_ORDER);

286
/* I: attributes used when instantiating standard unbound pools on demand */
287 288
static struct workqueue_attrs *unbound_std_wq_attrs[NR_STD_WORKER_POOLS];

289 290
struct workqueue_struct *system_wq __read_mostly;
EXPORT_SYMBOL_GPL(system_wq);
V
Valentin Ilie 已提交
291
struct workqueue_struct *system_highpri_wq __read_mostly;
292
EXPORT_SYMBOL_GPL(system_highpri_wq);
V
Valentin Ilie 已提交
293
struct workqueue_struct *system_long_wq __read_mostly;
294
EXPORT_SYMBOL_GPL(system_long_wq);
V
Valentin Ilie 已提交
295
struct workqueue_struct *system_unbound_wq __read_mostly;
296
EXPORT_SYMBOL_GPL(system_unbound_wq);
V
Valentin Ilie 已提交
297
struct workqueue_struct *system_freezable_wq __read_mostly;
298
EXPORT_SYMBOL_GPL(system_freezable_wq);
299

300 301 302 303
static int worker_thread(void *__worker);
static void copy_workqueue_attrs(struct workqueue_attrs *to,
				 const struct workqueue_attrs *from);

304 305 306
#define CREATE_TRACE_POINTS
#include <trace/events/workqueue.h>

307
#define assert_rcu_or_pool_mutex()					\
308
	rcu_lockdep_assert(rcu_read_lock_sched_held() ||		\
309 310
			   lockdep_is_held(&wq_pool_mutex),		\
			   "sched RCU or wq_pool_mutex should be held")
311

312
#define assert_rcu_or_wq_mutex(wq)					\
313
	rcu_lockdep_assert(rcu_read_lock_sched_held() ||		\
314
			   lockdep_is_held(&wq->mutex),			\
315
			   "sched RCU or wq->mutex should be held")
316

317 318
#ifdef CONFIG_LOCKDEP
#define assert_manager_or_pool_lock(pool)				\
319 320
	WARN_ONCE(debug_locks &&					\
		  !lockdep_is_held(&(pool)->manager_mutex) &&		\
321 322 323 324 325 326
		  !lockdep_is_held(&(pool)->lock),			\
		  "pool->manager_mutex or ->lock should be held")
#else
#define assert_manager_or_pool_lock(pool)	do { } while (0)
#endif

327 328 329
#define for_each_cpu_worker_pool(pool, cpu)				\
	for ((pool) = &per_cpu(cpu_worker_pools, cpu)[0];		\
	     (pool) < &per_cpu(cpu_worker_pools, cpu)[NR_STD_WORKER_POOLS]; \
330
	     (pool)++)
331

T
Tejun Heo 已提交
332 333 334
/**
 * for_each_pool - iterate through all worker_pools in the system
 * @pool: iteration cursor
335
 * @pi: integer used for iteration
336
 *
337 338 339
 * This must be called either with wq_pool_mutex held or sched RCU read
 * locked.  If the pool needs to be used beyond the locking in effect, the
 * caller is responsible for guaranteeing that the pool stays online.
340 341 342
 *
 * The if/else clause exists only for the lockdep assertion and can be
 * ignored.
T
Tejun Heo 已提交
343
 */
344 345
#define for_each_pool(pool, pi)						\
	idr_for_each_entry(&worker_pool_idr, pool, pi)			\
346
		if (({ assert_rcu_or_pool_mutex(); false; })) { }	\
347
		else
T
Tejun Heo 已提交
348

349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364
/**
 * for_each_pool_worker - iterate through all workers of a worker_pool
 * @worker: iteration cursor
 * @wi: integer used for iteration
 * @pool: worker_pool to iterate workers of
 *
 * This must be called with either @pool->manager_mutex or ->lock held.
 *
 * The if/else clause exists only for the lockdep assertion and can be
 * ignored.
 */
#define for_each_pool_worker(worker, wi, pool)				\
	idr_for_each_entry(&(pool)->worker_idr, (worker), (wi))		\
		if (({ assert_manager_or_pool_lock((pool)); false; })) { } \
		else

365 366 367 368
/**
 * for_each_pwq - iterate through all pool_workqueues of the specified workqueue
 * @pwq: iteration cursor
 * @wq: the target workqueue
369
 *
370
 * This must be called either with wq->mutex held or sched RCU read locked.
371 372
 * If the pwq needs to be used beyond the locking in effect, the caller is
 * responsible for guaranteeing that the pwq stays online.
373 374 375
 *
 * The if/else clause exists only for the lockdep assertion and can be
 * ignored.
376 377
 */
#define for_each_pwq(pwq, wq)						\
378
	list_for_each_entry_rcu((pwq), &(wq)->pwqs, pwqs_node)		\
379
		if (({ assert_rcu_or_wq_mutex(wq); false; })) { }	\
380
		else
381

382 383 384 385
#ifdef CONFIG_DEBUG_OBJECTS_WORK

static struct debug_obj_descr work_debug_descr;

386 387 388 389 390
static void *work_debug_hint(void *addr)
{
	return ((struct work_struct *) addr)->func;
}

391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425
/*
 * fixup_init is called when:
 * - an active object is initialized
 */
static int work_fixup_init(void *addr, enum debug_obj_state state)
{
	struct work_struct *work = addr;

	switch (state) {
	case ODEBUG_STATE_ACTIVE:
		cancel_work_sync(work);
		debug_object_init(work, &work_debug_descr);
		return 1;
	default:
		return 0;
	}
}

/*
 * fixup_activate is called when:
 * - an active object is activated
 * - an unknown object is activated (might be a statically initialized object)
 */
static int work_fixup_activate(void *addr, enum debug_obj_state state)
{
	struct work_struct *work = addr;

	switch (state) {

	case ODEBUG_STATE_NOTAVAILABLE:
		/*
		 * This is not really a fixup. The work struct was
		 * statically initialized. We just make sure that it
		 * is tracked in the object tracker.
		 */
426
		if (test_bit(WORK_STRUCT_STATIC_BIT, work_data_bits(work))) {
427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461
			debug_object_init(work, &work_debug_descr);
			debug_object_activate(work, &work_debug_descr);
			return 0;
		}
		WARN_ON_ONCE(1);
		return 0;

	case ODEBUG_STATE_ACTIVE:
		WARN_ON(1);

	default:
		return 0;
	}
}

/*
 * fixup_free is called when:
 * - an active object is freed
 */
static int work_fixup_free(void *addr, enum debug_obj_state state)
{
	struct work_struct *work = addr;

	switch (state) {
	case ODEBUG_STATE_ACTIVE:
		cancel_work_sync(work);
		debug_object_free(work, &work_debug_descr);
		return 1;
	default:
		return 0;
	}
}

static struct debug_obj_descr work_debug_descr = {
	.name		= "work_struct",
462
	.debug_hint	= work_debug_hint,
463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497
	.fixup_init	= work_fixup_init,
	.fixup_activate	= work_fixup_activate,
	.fixup_free	= work_fixup_free,
};

static inline void debug_work_activate(struct work_struct *work)
{
	debug_object_activate(work, &work_debug_descr);
}

static inline void debug_work_deactivate(struct work_struct *work)
{
	debug_object_deactivate(work, &work_debug_descr);
}

void __init_work(struct work_struct *work, int onstack)
{
	if (onstack)
		debug_object_init_on_stack(work, &work_debug_descr);
	else
		debug_object_init(work, &work_debug_descr);
}
EXPORT_SYMBOL_GPL(__init_work);

void destroy_work_on_stack(struct work_struct *work)
{
	debug_object_free(work, &work_debug_descr);
}
EXPORT_SYMBOL_GPL(destroy_work_on_stack);

#else
static inline void debug_work_activate(struct work_struct *work) { }
static inline void debug_work_deactivate(struct work_struct *work) { }
#endif

T
Tejun Heo 已提交
498 499 500 501 502
/* allocate ID and assign it to @pool */
static int worker_pool_assign_id(struct worker_pool *pool)
{
	int ret;

503
	lockdep_assert_held(&wq_pool_mutex);
504

505 506 507 508 509
	do {
		if (!idr_pre_get(&worker_pool_idr, GFP_KERNEL))
			return -ENOMEM;
		ret = idr_get_new(&worker_pool_idr, pool, &pool->id);
	} while (ret == -EAGAIN);
T
Tejun Heo 已提交
510

511
	return ret;
512 513
}

514 515 516 517
/**
 * first_pwq - return the first pool_workqueue of the specified workqueue
 * @wq: the target workqueue
 *
518
 * This must be called either with wq->mutex held or sched RCU read locked.
519 520
 * If the pwq needs to be used beyond the locking in effect, the caller is
 * responsible for guaranteeing that the pwq stays online.
521
 */
522
static struct pool_workqueue *first_pwq(struct workqueue_struct *wq)
523
{
524
	assert_rcu_or_wq_mutex(wq);
525 526
	return list_first_or_null_rcu(&wq->pwqs, struct pool_workqueue,
				      pwqs_node);
527 528
}

529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544
/**
 * unbound_pwq_by_node - return the unbound pool_workqueue for the given node
 * @wq: the target workqueue
 * @node: the node ID
 *
 * This must be called either with pwq_lock held or sched RCU read locked.
 * If the pwq needs to be used beyond the locking in effect, the caller is
 * responsible for guaranteeing that the pwq stays online.
 */
static struct pool_workqueue *unbound_pwq_by_node(struct workqueue_struct *wq,
						  int node)
{
	assert_rcu_or_wq_mutex(wq);
	return rcu_dereference_raw(wq->numa_pwq_tbl[node]);
}

545 546 547 548 549 550 551 552 553 554 555 556 557 558 559
static unsigned int work_color_to_flags(int color)
{
	return color << WORK_STRUCT_COLOR_SHIFT;
}

static int get_work_color(struct work_struct *work)
{
	return (*work_data_bits(work) >> WORK_STRUCT_COLOR_SHIFT) &
		((1 << WORK_STRUCT_COLOR_BITS) - 1);
}

static int work_next_color(int color)
{
	return (color + 1) % WORK_NR_COLORS;
}
L
Linus Torvalds 已提交
560

561
/*
562 563
 * While queued, %WORK_STRUCT_PWQ is set and non flag bits of a work's data
 * contain the pointer to the queued pwq.  Once execution starts, the flag
564
 * is cleared and the high bits contain OFFQ flags and pool ID.
565
 *
566 567
 * set_work_pwq(), set_work_pool_and_clear_pending(), mark_work_canceling()
 * and clear_work_data() can be used to set the pwq, pool or clear
568 569
 * work->data.  These functions should only be called while the work is
 * owned - ie. while the PENDING bit is set.
570
 *
571
 * get_work_pool() and get_work_pwq() can be used to obtain the pool or pwq
572
 * corresponding to a work.  Pool is available once the work has been
573
 * queued anywhere after initialization until it is sync canceled.  pwq is
574
 * available only while the work item is queued.
575
 *
576 577 578 579
 * %WORK_OFFQ_CANCELING is used to mark a work item which is being
 * canceled.  While being canceled, a work item may have its PENDING set
 * but stay off timer and worklist for arbitrarily long and nobody should
 * try to steal the PENDING bit.
580
 */
581 582
static inline void set_work_data(struct work_struct *work, unsigned long data,
				 unsigned long flags)
583
{
584
	WARN_ON_ONCE(!work_pending(work));
585 586
	atomic_long_set(&work->data, data | flags | work_static(work));
}
587

588
static void set_work_pwq(struct work_struct *work, struct pool_workqueue *pwq,
589 590
			 unsigned long extra_flags)
{
591 592
	set_work_data(work, (unsigned long)pwq,
		      WORK_STRUCT_PENDING | WORK_STRUCT_PWQ | extra_flags);
593 594
}

595 596 597 598 599 600 601
static void set_work_pool_and_keep_pending(struct work_struct *work,
					   int pool_id)
{
	set_work_data(work, (unsigned long)pool_id << WORK_OFFQ_POOL_SHIFT,
		      WORK_STRUCT_PENDING);
}

602 603
static void set_work_pool_and_clear_pending(struct work_struct *work,
					    int pool_id)
604
{
605 606 607 608 609 610 611
	/*
	 * The following wmb is paired with the implied mb in
	 * test_and_set_bit(PENDING) and ensures all updates to @work made
	 * here are visible to and precede any updates by the next PENDING
	 * owner.
	 */
	smp_wmb();
612
	set_work_data(work, (unsigned long)pool_id << WORK_OFFQ_POOL_SHIFT, 0);
613
}
614

615
static void clear_work_data(struct work_struct *work)
L
Linus Torvalds 已提交
616
{
617 618
	smp_wmb();	/* see set_work_pool_and_clear_pending() */
	set_work_data(work, WORK_STRUCT_NO_POOL, 0);
L
Linus Torvalds 已提交
619 620
}

621
static struct pool_workqueue *get_work_pwq(struct work_struct *work)
622
{
623
	unsigned long data = atomic_long_read(&work->data);
624

625
	if (data & WORK_STRUCT_PWQ)
626 627 628
		return (void *)(data & WORK_STRUCT_WQ_DATA_MASK);
	else
		return NULL;
629 630
}

631 632 633 634 635
/**
 * get_work_pool - return the worker_pool a given work was associated with
 * @work: the work item of interest
 *
 * Return the worker_pool @work was last associated with.  %NULL if none.
636
 *
637 638 639
 * Pools are created and destroyed under wq_pool_mutex, and allows read
 * access under sched-RCU read lock.  As such, this function should be
 * called under wq_pool_mutex or with preemption disabled.
640 641 642 643 644
 *
 * All fields of the returned pool are accessible as long as the above
 * mentioned locking is in effect.  If the returned pool needs to be used
 * beyond the critical section, the caller is responsible for ensuring the
 * returned pool is and stays online.
645 646
 */
static struct worker_pool *get_work_pool(struct work_struct *work)
647
{
648
	unsigned long data = atomic_long_read(&work->data);
649
	int pool_id;
650

651
	assert_rcu_or_pool_mutex();
652

653 654
	if (data & WORK_STRUCT_PWQ)
		return ((struct pool_workqueue *)
655
			(data & WORK_STRUCT_WQ_DATA_MASK))->pool;
656

657 658
	pool_id = data >> WORK_OFFQ_POOL_SHIFT;
	if (pool_id == WORK_OFFQ_POOL_NONE)
659 660
		return NULL;

661
	return idr_find(&worker_pool_idr, pool_id);
662 663 664 665 666 667 668 669 670 671 672
}

/**
 * get_work_pool_id - return the worker pool ID a given work is associated with
 * @work: the work item of interest
 *
 * Return the worker_pool ID @work was last associated with.
 * %WORK_OFFQ_POOL_NONE if none.
 */
static int get_work_pool_id(struct work_struct *work)
{
673 674
	unsigned long data = atomic_long_read(&work->data);

675 676
	if (data & WORK_STRUCT_PWQ)
		return ((struct pool_workqueue *)
677
			(data & WORK_STRUCT_WQ_DATA_MASK))->pool->id;
678

679
	return data >> WORK_OFFQ_POOL_SHIFT;
680 681
}

682 683
static void mark_work_canceling(struct work_struct *work)
{
684
	unsigned long pool_id = get_work_pool_id(work);
685

686 687
	pool_id <<= WORK_OFFQ_POOL_SHIFT;
	set_work_data(work, pool_id | WORK_OFFQ_CANCELING, WORK_STRUCT_PENDING);
688 689 690 691 692 693
}

static bool work_is_canceling(struct work_struct *work)
{
	unsigned long data = atomic_long_read(&work->data);

694
	return !(data & WORK_STRUCT_PWQ) && (data & WORK_OFFQ_CANCELING);
695 696
}

697
/*
698 699
 * Policy functions.  These define the policies on how the global worker
 * pools are managed.  Unless noted otherwise, these functions assume that
700
 * they're being called with pool->lock held.
701 702
 */

703
static bool __need_more_worker(struct worker_pool *pool)
704
{
705
	return !atomic_read(&pool->nr_running);
706 707
}

708
/*
709 710
 * Need to wake up a worker?  Called from anything but currently
 * running workers.
711 712
 *
 * Note that, because unbound workers never contribute to nr_running, this
713
 * function will always return %true for unbound pools as long as the
714
 * worklist isn't empty.
715
 */
716
static bool need_more_worker(struct worker_pool *pool)
717
{
718
	return !list_empty(&pool->worklist) && __need_more_worker(pool);
719
}
720

721
/* Can I start working?  Called from busy but !running workers. */
722
static bool may_start_working(struct worker_pool *pool)
723
{
724
	return pool->nr_idle;
725 726 727
}

/* Do I need to keep working?  Called from currently running workers. */
728
static bool keep_working(struct worker_pool *pool)
729
{
730 731
	return !list_empty(&pool->worklist) &&
		atomic_read(&pool->nr_running) <= 1;
732 733 734
}

/* Do we need a new worker?  Called from manager. */
735
static bool need_to_create_worker(struct worker_pool *pool)
736
{
737
	return need_more_worker(pool) && !may_start_working(pool);
738
}
739

740
/* Do I need to be the manager? */
741
static bool need_to_manage_workers(struct worker_pool *pool)
742
{
743
	return need_to_create_worker(pool) ||
744
		(pool->flags & POOL_MANAGE_WORKERS);
745 746 747
}

/* Do we have too many workers and should some go away? */
748
static bool too_many_workers(struct worker_pool *pool)
749
{
750
	bool managing = mutex_is_locked(&pool->manager_arb);
751 752
	int nr_idle = pool->nr_idle + managing; /* manager is considered idle */
	int nr_busy = pool->nr_workers - nr_idle;
753

754 755 756 757 758 759 760
	/*
	 * nr_idle and idle_list may disagree if idle rebinding is in
	 * progress.  Never return %true if idle_list is empty.
	 */
	if (list_empty(&pool->idle_list))
		return false;

761
	return nr_idle > 2 && (nr_idle - 2) * MAX_IDLE_WORKERS_RATIO >= nr_busy;
762 763
}

764
/*
765 766 767
 * Wake up functions.
 */

768
/* Return the first worker.  Safe with preemption disabled */
769
static struct worker *first_worker(struct worker_pool *pool)
770
{
771
	if (unlikely(list_empty(&pool->idle_list)))
772 773
		return NULL;

774
	return list_first_entry(&pool->idle_list, struct worker, entry);
775 776 777 778
}

/**
 * wake_up_worker - wake up an idle worker
779
 * @pool: worker pool to wake worker from
780
 *
781
 * Wake up the first idle worker of @pool.
782 783
 *
 * CONTEXT:
784
 * spin_lock_irq(pool->lock).
785
 */
786
static void wake_up_worker(struct worker_pool *pool)
787
{
788
	struct worker *worker = first_worker(pool);
789 790 791 792 793

	if (likely(worker))
		wake_up_process(worker->task);
}

794
/**
795 796 797 798 799 800 801 802 803 804
 * wq_worker_waking_up - a worker is waking up
 * @task: task waking up
 * @cpu: CPU @task is waking up to
 *
 * This function is called during try_to_wake_up() when a worker is
 * being awoken.
 *
 * CONTEXT:
 * spin_lock_irq(rq->lock)
 */
805
void wq_worker_waking_up(struct task_struct *task, int cpu)
806 807 808
{
	struct worker *worker = kthread_data(task);

809
	if (!(worker->flags & WORKER_NOT_RUNNING)) {
810
		WARN_ON_ONCE(worker->pool->cpu != cpu);
811
		atomic_inc(&worker->pool->nr_running);
812
	}
813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829
}

/**
 * wq_worker_sleeping - a worker is going to sleep
 * @task: task going to sleep
 * @cpu: CPU in question, must be the current CPU number
 *
 * This function is called during schedule() when a busy worker is
 * going to sleep.  Worker on the same cpu can be woken up by
 * returning pointer to its task.
 *
 * CONTEXT:
 * spin_lock_irq(rq->lock)
 *
 * RETURNS:
 * Worker task on @cpu to wake up, %NULL if none.
 */
830
struct task_struct *wq_worker_sleeping(struct task_struct *task, int cpu)
831 832
{
	struct worker *worker = kthread_data(task), *to_wakeup = NULL;
833
	struct worker_pool *pool;
834

835 836 837 838 839
	/*
	 * Rescuers, which may not have all the fields set up like normal
	 * workers, also reach here, let's not access anything before
	 * checking NOT_RUNNING.
	 */
840
	if (worker->flags & WORKER_NOT_RUNNING)
841 842
		return NULL;

843 844
	pool = worker->pool;

845
	/* this can only happen on the local cpu */
846 847
	if (WARN_ON_ONCE(cpu != raw_smp_processor_id()))
		return NULL;
848 849 850 851 852 853

	/*
	 * The counterpart of the following dec_and_test, implied mb,
	 * worklist not empty test sequence is in insert_work().
	 * Please read comment there.
	 *
854 855 856
	 * NOT_RUNNING is clear.  This means that we're bound to and
	 * running on the local cpu w/ rq lock held and preemption
	 * disabled, which in turn means that none else could be
857
	 * manipulating idle_list, so dereferencing idle_list without pool
858
	 * lock is safe.
859
	 */
860 861
	if (atomic_dec_and_test(&pool->nr_running) &&
	    !list_empty(&pool->worklist))
862
		to_wakeup = first_worker(pool);
863 864 865 866 867
	return to_wakeup ? to_wakeup->task : NULL;
}

/**
 * worker_set_flags - set worker flags and adjust nr_running accordingly
868
 * @worker: self
869 870 871
 * @flags: flags to set
 * @wakeup: wakeup an idle worker if necessary
 *
872 873 874
 * Set @flags in @worker->flags and adjust nr_running accordingly.  If
 * nr_running becomes zero and @wakeup is %true, an idle worker is
 * woken up.
875
 *
876
 * CONTEXT:
877
 * spin_lock_irq(pool->lock)
878 879 880 881
 */
static inline void worker_set_flags(struct worker *worker, unsigned int flags,
				    bool wakeup)
{
882
	struct worker_pool *pool = worker->pool;
883

884 885
	WARN_ON_ONCE(worker->task != current);

886 887 888 889 890 891 892 893
	/*
	 * If transitioning into NOT_RUNNING, adjust nr_running and
	 * wake up an idle worker as necessary if requested by
	 * @wakeup.
	 */
	if ((flags & WORKER_NOT_RUNNING) &&
	    !(worker->flags & WORKER_NOT_RUNNING)) {
		if (wakeup) {
894
			if (atomic_dec_and_test(&pool->nr_running) &&
895
			    !list_empty(&pool->worklist))
896
				wake_up_worker(pool);
897
		} else
898
			atomic_dec(&pool->nr_running);
899 900
	}

901 902 903 904
	worker->flags |= flags;
}

/**
905
 * worker_clr_flags - clear worker flags and adjust nr_running accordingly
906
 * @worker: self
907 908
 * @flags: flags to clear
 *
909
 * Clear @flags in @worker->flags and adjust nr_running accordingly.
910
 *
911
 * CONTEXT:
912
 * spin_lock_irq(pool->lock)
913 914 915
 */
static inline void worker_clr_flags(struct worker *worker, unsigned int flags)
{
916
	struct worker_pool *pool = worker->pool;
917 918
	unsigned int oflags = worker->flags;

919 920
	WARN_ON_ONCE(worker->task != current);

921
	worker->flags &= ~flags;
922

923 924 925 926 927
	/*
	 * If transitioning out of NOT_RUNNING, increment nr_running.  Note
	 * that the nested NOT_RUNNING is not a noop.  NOT_RUNNING is mask
	 * of multiple flags, not a single flag.
	 */
928 929
	if ((flags & WORKER_NOT_RUNNING) && (oflags & WORKER_NOT_RUNNING))
		if (!(worker->flags & WORKER_NOT_RUNNING))
930
			atomic_inc(&pool->nr_running);
931 932
}

933 934
/**
 * find_worker_executing_work - find worker which is executing a work
935
 * @pool: pool of interest
936 937
 * @work: work to find worker for
 *
938 939
 * Find a worker which is executing @work on @pool by searching
 * @pool->busy_hash which is keyed by the address of @work.  For a worker
940 941 942 943 944 945 946 947 948 949 950 951
 * to match, its current execution should match the address of @work and
 * its work function.  This is to avoid unwanted dependency between
 * unrelated work executions through a work item being recycled while still
 * being executed.
 *
 * This is a bit tricky.  A work item may be freed once its execution
 * starts and nothing prevents the freed area from being recycled for
 * another work item.  If the same work item address ends up being reused
 * before the original execution finishes, workqueue will identify the
 * recycled work item as currently executing and make it wait until the
 * current execution finishes, introducing an unwanted dependency.
 *
952 953 954 955 956 957
 * This function checks the work item address and work function to avoid
 * false positives.  Note that this isn't complete as one may construct a
 * work function which can introduce dependency onto itself through a
 * recycled work item.  Well, if somebody wants to shoot oneself in the
 * foot that badly, there's only so much we can do, and if such deadlock
 * actually occurs, it should be easy to locate the culprit work function.
958 959
 *
 * CONTEXT:
960
 * spin_lock_irq(pool->lock).
961 962 963 964
 *
 * RETURNS:
 * Pointer to worker which is executing @work if found, NULL
 * otherwise.
965
 */
966
static struct worker *find_worker_executing_work(struct worker_pool *pool,
967
						 struct work_struct *work)
968
{
969 970
	struct worker *worker;

971
	hash_for_each_possible(pool->busy_hash, worker, hentry,
972 973 974
			       (unsigned long)work)
		if (worker->current_work == work &&
		    worker->current_func == work->func)
975 976 977
			return worker;

	return NULL;
978 979
}

980 981 982 983 984 985 986 987 988 989 990 991 992 993 994
/**
 * move_linked_works - move linked works to a list
 * @work: start of series of works to be scheduled
 * @head: target list to append @work to
 * @nextp: out paramter for nested worklist walking
 *
 * Schedule linked works starting from @work to @head.  Work series to
 * be scheduled starts at @work and includes any consecutive work with
 * WORK_STRUCT_LINKED set in its predecessor.
 *
 * If @nextp is not NULL, it's updated to point to the next work of
 * the last scheduled work.  This allows move_linked_works() to be
 * nested inside outer list_for_each_entry_safe().
 *
 * CONTEXT:
995
 * spin_lock_irq(pool->lock).
996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020
 */
static void move_linked_works(struct work_struct *work, struct list_head *head,
			      struct work_struct **nextp)
{
	struct work_struct *n;

	/*
	 * Linked worklist will always end before the end of the list,
	 * use NULL for list head.
	 */
	list_for_each_entry_safe_from(work, n, NULL, entry) {
		list_move_tail(&work->entry, head);
		if (!(*work_data_bits(work) & WORK_STRUCT_LINKED))
			break;
	}

	/*
	 * If we're already inside safe list traversal and have moved
	 * multiple works to the scheduled queue, the next position
	 * needs to be updated.
	 */
	if (nextp)
		*nextp = n;
}

T
Tejun Heo 已提交
1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059
/**
 * get_pwq - get an extra reference on the specified pool_workqueue
 * @pwq: pool_workqueue to get
 *
 * Obtain an extra reference on @pwq.  The caller should guarantee that
 * @pwq has positive refcnt and be holding the matching pool->lock.
 */
static void get_pwq(struct pool_workqueue *pwq)
{
	lockdep_assert_held(&pwq->pool->lock);
	WARN_ON_ONCE(pwq->refcnt <= 0);
	pwq->refcnt++;
}

/**
 * put_pwq - put a pool_workqueue reference
 * @pwq: pool_workqueue to put
 *
 * Drop a reference of @pwq.  If its refcnt reaches zero, schedule its
 * destruction.  The caller should be holding the matching pool->lock.
 */
static void put_pwq(struct pool_workqueue *pwq)
{
	lockdep_assert_held(&pwq->pool->lock);
	if (likely(--pwq->refcnt))
		return;
	if (WARN_ON_ONCE(!(pwq->wq->flags & WQ_UNBOUND)))
		return;
	/*
	 * @pwq can't be released under pool->lock, bounce to
	 * pwq_unbound_release_workfn().  This never recurses on the same
	 * pool->lock as this path is taken only for unbound workqueues and
	 * the release work item is scheduled on a per-cpu workqueue.  To
	 * avoid lockdep warning, unbound pool->locks are given lockdep
	 * subclass of 1 in get_unbound_pool().
	 */
	schedule_work(&pwq->unbound_release_work);
}

1060
static void pwq_activate_delayed_work(struct work_struct *work)
1061
{
1062
	struct pool_workqueue *pwq = get_work_pwq(work);
1063 1064

	trace_workqueue_activate_work(work);
1065
	move_linked_works(work, &pwq->pool->worklist, NULL);
1066
	__clear_bit(WORK_STRUCT_DELAYED_BIT, work_data_bits(work));
1067
	pwq->nr_active++;
1068 1069
}

1070
static void pwq_activate_first_delayed(struct pool_workqueue *pwq)
1071
{
1072
	struct work_struct *work = list_first_entry(&pwq->delayed_works,
1073 1074
						    struct work_struct, entry);

1075
	pwq_activate_delayed_work(work);
1076 1077
}

1078
/**
1079 1080
 * pwq_dec_nr_in_flight - decrement pwq's nr_in_flight
 * @pwq: pwq of interest
1081 1082 1083
 * @color: color of work which left the queue
 *
 * A work either has completed or is removed from pending queue,
1084
 * decrement nr_in_flight of its pwq and handle workqueue flushing.
1085 1086
 *
 * CONTEXT:
1087
 * spin_lock_irq(pool->lock).
1088
 */
1089
static void pwq_dec_nr_in_flight(struct pool_workqueue *pwq, int color)
1090
{
T
Tejun Heo 已提交
1091
	/* uncolored work items don't participate in flushing or nr_active */
1092
	if (color == WORK_NO_COLOR)
T
Tejun Heo 已提交
1093
		goto out_put;
1094

1095
	pwq->nr_in_flight[color]--;
1096

1097 1098
	pwq->nr_active--;
	if (!list_empty(&pwq->delayed_works)) {
1099
		/* one down, submit a delayed one */
1100 1101
		if (pwq->nr_active < pwq->max_active)
			pwq_activate_first_delayed(pwq);
1102 1103 1104
	}

	/* is flush in progress and are we at the flushing tip? */
1105
	if (likely(pwq->flush_color != color))
T
Tejun Heo 已提交
1106
		goto out_put;
1107 1108

	/* are there still in-flight works? */
1109
	if (pwq->nr_in_flight[color])
T
Tejun Heo 已提交
1110
		goto out_put;
1111

1112 1113
	/* this pwq is done, clear flush_color */
	pwq->flush_color = -1;
1114 1115

	/*
1116
	 * If this was the last pwq, wake up the first flusher.  It
1117 1118
	 * will handle the rest.
	 */
1119 1120
	if (atomic_dec_and_test(&pwq->wq->nr_pwqs_to_flush))
		complete(&pwq->wq->first_flusher->done);
T
Tejun Heo 已提交
1121 1122
out_put:
	put_pwq(pwq);
1123 1124
}

1125
/**
1126
 * try_to_grab_pending - steal work item from worklist and disable irq
1127 1128
 * @work: work item to steal
 * @is_dwork: @work is a delayed_work
1129
 * @flags: place to store irq state
1130 1131 1132 1133 1134 1135 1136
 *
 * Try to grab PENDING bit of @work.  This function can handle @work in any
 * stable state - idle, on timer or on worklist.  Return values are
 *
 *  1		if @work was pending and we successfully stole PENDING
 *  0		if @work was idle and we claimed PENDING
 *  -EAGAIN	if PENDING couldn't be grabbed at the moment, safe to busy-retry
1137 1138
 *  -ENOENT	if someone else is canceling @work, this state may persist
 *		for arbitrarily long
1139
 *
1140
 * On >= 0 return, the caller owns @work's PENDING bit.  To avoid getting
1141 1142 1143
 * interrupted while holding PENDING and @work off queue, irq must be
 * disabled on entry.  This, combined with delayed_work->timer being
 * irqsafe, ensures that we return -EAGAIN for finite short period of time.
1144 1145 1146 1147
 *
 * On successful return, >= 0, irq is disabled and the caller is
 * responsible for releasing it using local_irq_restore(*@flags).
 *
1148
 * This function is safe to call from any context including IRQ handler.
1149
 */
1150 1151
static int try_to_grab_pending(struct work_struct *work, bool is_dwork,
			       unsigned long *flags)
1152
{
1153
	struct worker_pool *pool;
1154
	struct pool_workqueue *pwq;
1155

1156 1157
	local_irq_save(*flags);

1158 1159 1160 1161
	/* try to steal the timer if it exists */
	if (is_dwork) {
		struct delayed_work *dwork = to_delayed_work(work);

1162 1163 1164 1165 1166
		/*
		 * dwork->timer is irqsafe.  If del_timer() fails, it's
		 * guaranteed that the timer is not queued anywhere and not
		 * running on the local CPU.
		 */
1167 1168 1169 1170 1171
		if (likely(del_timer(&dwork->timer)))
			return 1;
	}

	/* try to claim PENDING the normal way */
1172 1173 1174 1175 1176 1177 1178
	if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work)))
		return 0;

	/*
	 * The queueing is in progress, or it is already queued. Try to
	 * steal it from ->worklist without clearing WORK_STRUCT_PENDING.
	 */
1179 1180
	pool = get_work_pool(work);
	if (!pool)
1181
		goto fail;
1182

1183
	spin_lock(&pool->lock);
1184
	/*
1185 1186 1187 1188 1189
	 * work->data is guaranteed to point to pwq only while the work
	 * item is queued on pwq->wq, and both updating work->data to point
	 * to pwq on queueing and to pool on dequeueing are done under
	 * pwq->pool->lock.  This in turn guarantees that, if work->data
	 * points to pwq which is associated with a locked pool, the work
1190 1191
	 * item is currently queued on that pool.
	 */
1192 1193
	pwq = get_work_pwq(work);
	if (pwq && pwq->pool == pool) {
1194 1195 1196 1197 1198
		debug_work_deactivate(work);

		/*
		 * A delayed work item cannot be grabbed directly because
		 * it might have linked NO_COLOR work items which, if left
1199
		 * on the delayed_list, will confuse pwq->nr_active
1200 1201 1202 1203
		 * management later on and cause stall.  Make sure the work
		 * item is activated before grabbing.
		 */
		if (*work_data_bits(work) & WORK_STRUCT_DELAYED)
1204
			pwq_activate_delayed_work(work);
1205 1206

		list_del_init(&work->entry);
1207
		pwq_dec_nr_in_flight(get_work_pwq(work), get_work_color(work));
1208

1209
		/* work->data points to pwq iff queued, point to pool */
1210 1211 1212 1213
		set_work_pool_and_keep_pending(work, pool->id);

		spin_unlock(&pool->lock);
		return 1;
1214
	}
1215
	spin_unlock(&pool->lock);
1216 1217 1218 1219 1220
fail:
	local_irq_restore(*flags);
	if (work_is_canceling(work))
		return -ENOENT;
	cpu_relax();
1221
	return -EAGAIN;
1222 1223
}

T
Tejun Heo 已提交
1224
/**
1225
 * insert_work - insert a work into a pool
1226
 * @pwq: pwq @work belongs to
T
Tejun Heo 已提交
1227 1228 1229 1230
 * @work: work to insert
 * @head: insertion point
 * @extra_flags: extra WORK_STRUCT_* flags to set
 *
1231
 * Insert @work which belongs to @pwq after @head.  @extra_flags is or'd to
1232
 * work_struct flags.
T
Tejun Heo 已提交
1233 1234
 *
 * CONTEXT:
1235
 * spin_lock_irq(pool->lock).
T
Tejun Heo 已提交
1236
 */
1237 1238
static void insert_work(struct pool_workqueue *pwq, struct work_struct *work,
			struct list_head *head, unsigned int extra_flags)
O
Oleg Nesterov 已提交
1239
{
1240
	struct worker_pool *pool = pwq->pool;
1241

T
Tejun Heo 已提交
1242
	/* we own @work, set data and link */
1243
	set_work_pwq(work, pwq, extra_flags);
1244
	list_add_tail(&work->entry, head);
T
Tejun Heo 已提交
1245
	get_pwq(pwq);
1246 1247

	/*
1248 1249 1250
	 * Ensure either wq_worker_sleeping() sees the above
	 * list_add_tail() or we see zero nr_running to avoid workers lying
	 * around lazily while there are works to be processed.
1251 1252 1253
	 */
	smp_mb();

1254 1255
	if (__need_more_worker(pool))
		wake_up_worker(pool);
O
Oleg Nesterov 已提交
1256 1257
}

1258 1259
/*
 * Test whether @work is being queued from another work executing on the
1260
 * same workqueue.
1261 1262 1263
 */
static bool is_chained_work(struct workqueue_struct *wq)
{
1264 1265 1266 1267 1268 1269 1270
	struct worker *worker;

	worker = current_wq_worker();
	/*
	 * Return %true iff I'm a worker execuing a work item on @wq.  If
	 * I'm @worker, it's safe to dereference it without locking.
	 */
1271
	return worker && worker->current_pwq->wq == wq;
1272 1273
}

1274
static void __queue_work(int cpu, struct workqueue_struct *wq,
L
Linus Torvalds 已提交
1275 1276
			 struct work_struct *work)
{
1277
	struct pool_workqueue *pwq;
1278
	struct worker_pool *last_pool;
1279
	struct list_head *worklist;
1280
	unsigned int work_flags;
1281
	unsigned int req_cpu = cpu;
1282 1283 1284 1285 1286 1287 1288 1289

	/*
	 * While a work item is PENDING && off queue, a task trying to
	 * steal the PENDING will busy-loop waiting for it to either get
	 * queued or lose PENDING.  Grabbing PENDING and queueing should
	 * happen with IRQ disabled.
	 */
	WARN_ON_ONCE(!irqs_disabled());
L
Linus Torvalds 已提交
1290

1291
	debug_work_activate(work);
1292

1293
	/* if dying, only works from the same workqueue are allowed */
1294
	if (unlikely(wq->flags & __WQ_DRAINING) &&
1295
	    WARN_ON_ONCE(!is_chained_work(wq)))
1296
		return;
1297
retry:
1298 1299 1300
	if (req_cpu == WORK_CPU_UNBOUND)
		cpu = raw_smp_processor_id();

1301
	/* pwq which will be used unless @work is executing elsewhere */
1302
	if (!(wq->flags & WQ_UNBOUND))
1303
		pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
1304 1305
	else
		pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
1306

1307 1308 1309 1310 1311 1312 1313 1314
	/*
	 * If @work was previously on a different pool, it might still be
	 * running there, in which case the work needs to be queued on that
	 * pool to guarantee non-reentrancy.
	 */
	last_pool = get_work_pool(work);
	if (last_pool && last_pool != pwq->pool) {
		struct worker *worker;
1315

1316
		spin_lock(&last_pool->lock);
1317

1318
		worker = find_worker_executing_work(last_pool, work);
1319

1320 1321
		if (worker && worker->current_pwq->wq == wq) {
			pwq = worker->current_pwq;
1322
		} else {
1323 1324
			/* meh... not running there, queue here */
			spin_unlock(&last_pool->lock);
1325
			spin_lock(&pwq->pool->lock);
1326
		}
1327
	} else {
1328
		spin_lock(&pwq->pool->lock);
1329 1330
	}

1331 1332 1333 1334
	/*
	 * pwq is determined and locked.  For unbound pools, we could have
	 * raced with pwq release and it could already be dead.  If its
	 * refcnt is zero, repeat pwq selection.  Note that pwqs never die
1335 1336
	 * without another pwq replacing it in the numa_pwq_tbl or while
	 * work items are executing on it, so the retrying is guaranteed to
1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349
	 * make forward-progress.
	 */
	if (unlikely(!pwq->refcnt)) {
		if (wq->flags & WQ_UNBOUND) {
			spin_unlock(&pwq->pool->lock);
			cpu_relax();
			goto retry;
		}
		/* oops */
		WARN_ONCE(true, "workqueue: per-cpu pwq for %s on cpu%d has 0 refcnt",
			  wq->name, cpu);
	}

1350 1351
	/* pwq determined, queue */
	trace_workqueue_queue_work(req_cpu, pwq, work);
1352

1353
	if (WARN_ON(!list_empty(&work->entry))) {
1354
		spin_unlock(&pwq->pool->lock);
1355 1356
		return;
	}
1357

1358 1359
	pwq->nr_in_flight[pwq->work_color]++;
	work_flags = work_color_to_flags(pwq->work_color);
1360

1361
	if (likely(pwq->nr_active < pwq->max_active)) {
1362
		trace_workqueue_activate_work(work);
1363 1364
		pwq->nr_active++;
		worklist = &pwq->pool->worklist;
1365 1366
	} else {
		work_flags |= WORK_STRUCT_DELAYED;
1367
		worklist = &pwq->delayed_works;
1368
	}
1369

1370
	insert_work(pwq, work, worklist, work_flags);
1371

1372
	spin_unlock(&pwq->pool->lock);
L
Linus Torvalds 已提交
1373 1374
}

1375
/**
1376 1377
 * queue_work_on - queue work on specific cpu
 * @cpu: CPU number to execute work on
1378 1379 1380
 * @wq: workqueue to use
 * @work: work to queue
 *
1381
 * Returns %false if @work was already on a queue, %true otherwise.
L
Linus Torvalds 已提交
1382
 *
1383 1384
 * We queue the work to a specific CPU, the caller must ensure it
 * can't go away.
L
Linus Torvalds 已提交
1385
 */
1386 1387
bool queue_work_on(int cpu, struct workqueue_struct *wq,
		   struct work_struct *work)
L
Linus Torvalds 已提交
1388
{
1389
	bool ret = false;
1390
	unsigned long flags;
1391

1392
	local_irq_save(flags);
1393

1394
	if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
T
Tejun Heo 已提交
1395
		__queue_work(cpu, wq, work);
1396
		ret = true;
1397
	}
1398

1399
	local_irq_restore(flags);
L
Linus Torvalds 已提交
1400 1401
	return ret;
}
1402
EXPORT_SYMBOL_GPL(queue_work_on);
L
Linus Torvalds 已提交
1403

1404
void delayed_work_timer_fn(unsigned long __data)
L
Linus Torvalds 已提交
1405
{
1406
	struct delayed_work *dwork = (struct delayed_work *)__data;
L
Linus Torvalds 已提交
1407

1408
	/* should have been called from irqsafe timer with irq already off */
1409
	__queue_work(dwork->cpu, dwork->wq, &dwork->work);
L
Linus Torvalds 已提交
1410
}
1411
EXPORT_SYMBOL(delayed_work_timer_fn);
L
Linus Torvalds 已提交
1412

1413 1414
static void __queue_delayed_work(int cpu, struct workqueue_struct *wq,
				struct delayed_work *dwork, unsigned long delay)
L
Linus Torvalds 已提交
1415
{
1416 1417 1418 1419 1420
	struct timer_list *timer = &dwork->timer;
	struct work_struct *work = &dwork->work;

	WARN_ON_ONCE(timer->function != delayed_work_timer_fn ||
		     timer->data != (unsigned long)dwork);
1421 1422
	WARN_ON_ONCE(timer_pending(timer));
	WARN_ON_ONCE(!list_empty(&work->entry));
1423

1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434
	/*
	 * If @delay is 0, queue @dwork->work immediately.  This is for
	 * both optimization and correctness.  The earliest @timer can
	 * expire is on the closest next tick and delayed_work users depend
	 * on that there's no such delay when @delay is 0.
	 */
	if (!delay) {
		__queue_work(cpu, wq, &dwork->work);
		return;
	}

1435
	timer_stats_timer_set_start_info(&dwork->timer);
L
Linus Torvalds 已提交
1436

1437
	dwork->wq = wq;
1438
	dwork->cpu = cpu;
1439 1440 1441 1442 1443 1444
	timer->expires = jiffies + delay;

	if (unlikely(cpu != WORK_CPU_UNBOUND))
		add_timer_on(timer, cpu);
	else
		add_timer(timer);
L
Linus Torvalds 已提交
1445 1446
}

1447 1448 1449 1450
/**
 * queue_delayed_work_on - queue work on specific CPU after delay
 * @cpu: CPU number to execute work on
 * @wq: workqueue to use
1451
 * @dwork: work to queue
1452 1453
 * @delay: number of jiffies to wait before queueing
 *
1454 1455 1456
 * Returns %false if @work was already on a queue, %true otherwise.  If
 * @delay is zero and @dwork is idle, it will be scheduled for immediate
 * execution.
1457
 */
1458 1459
bool queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
			   struct delayed_work *dwork, unsigned long delay)
1460
{
1461
	struct work_struct *work = &dwork->work;
1462
	bool ret = false;
1463
	unsigned long flags;
1464

1465 1466
	/* read the comment in __queue_work() */
	local_irq_save(flags);
1467

1468
	if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
1469
		__queue_delayed_work(cpu, wq, dwork, delay);
1470
		ret = true;
1471
	}
1472

1473
	local_irq_restore(flags);
1474 1475
	return ret;
}
1476
EXPORT_SYMBOL_GPL(queue_delayed_work_on);
1477

1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492
/**
 * mod_delayed_work_on - modify delay of or queue a delayed work on specific CPU
 * @cpu: CPU number to execute work on
 * @wq: workqueue to use
 * @dwork: work to queue
 * @delay: number of jiffies to wait before queueing
 *
 * If @dwork is idle, equivalent to queue_delayed_work_on(); otherwise,
 * modify @dwork's timer so that it expires after @delay.  If @delay is
 * zero, @work is guaranteed to be scheduled immediately regardless of its
 * current state.
 *
 * Returns %false if @dwork was idle and queued, %true if @dwork was
 * pending and its timer was modified.
 *
1493
 * This function is safe to call from any context including IRQ handler.
1494 1495 1496 1497 1498 1499 1500
 * See try_to_grab_pending() for details.
 */
bool mod_delayed_work_on(int cpu, struct workqueue_struct *wq,
			 struct delayed_work *dwork, unsigned long delay)
{
	unsigned long flags;
	int ret;
1501

1502 1503 1504
	do {
		ret = try_to_grab_pending(&dwork->work, true, &flags);
	} while (unlikely(ret == -EAGAIN));
1505

1506 1507 1508
	if (likely(ret >= 0)) {
		__queue_delayed_work(cpu, wq, dwork, delay);
		local_irq_restore(flags);
1509
	}
1510 1511

	/* -ENOENT from try_to_grab_pending() becomes %true */
1512 1513
	return ret;
}
1514 1515
EXPORT_SYMBOL_GPL(mod_delayed_work_on);

T
Tejun Heo 已提交
1516 1517 1518 1519 1520 1521 1522 1523
/**
 * worker_enter_idle - enter idle state
 * @worker: worker which is entering idle state
 *
 * @worker is entering idle state.  Update stats and idle timer if
 * necessary.
 *
 * LOCKING:
1524
 * spin_lock_irq(pool->lock).
T
Tejun Heo 已提交
1525 1526
 */
static void worker_enter_idle(struct worker *worker)
L
Linus Torvalds 已提交
1527
{
1528
	struct worker_pool *pool = worker->pool;
T
Tejun Heo 已提交
1529

1530 1531 1532 1533
	if (WARN_ON_ONCE(worker->flags & WORKER_IDLE) ||
	    WARN_ON_ONCE(!list_empty(&worker->entry) &&
			 (worker->hentry.next || worker->hentry.pprev)))
		return;
T
Tejun Heo 已提交
1534

1535 1536
	/* can't use worker_set_flags(), also called from start_worker() */
	worker->flags |= WORKER_IDLE;
1537
	pool->nr_idle++;
1538
	worker->last_active = jiffies;
T
Tejun Heo 已提交
1539 1540

	/* idle_list is LIFO */
1541
	list_add(&worker->entry, &pool->idle_list);
1542

1543 1544
	if (too_many_workers(pool) && !timer_pending(&pool->idle_timer))
		mod_timer(&pool->idle_timer, jiffies + IDLE_WORKER_TIMEOUT);
1545

1546
	/*
1547
	 * Sanity check nr_running.  Because wq_unbind_fn() releases
1548
	 * pool->lock between setting %WORKER_UNBOUND and zapping
1549 1550
	 * nr_running, the warning may trigger spuriously.  Check iff
	 * unbind is not in progress.
1551
	 */
1552
	WARN_ON_ONCE(!(pool->flags & POOL_DISASSOCIATED) &&
1553
		     pool->nr_workers == pool->nr_idle &&
1554
		     atomic_read(&pool->nr_running));
T
Tejun Heo 已提交
1555 1556 1557 1558 1559 1560 1561 1562 1563
}

/**
 * worker_leave_idle - leave idle state
 * @worker: worker which is leaving idle state
 *
 * @worker is leaving idle state.  Update stats.
 *
 * LOCKING:
1564
 * spin_lock_irq(pool->lock).
T
Tejun Heo 已提交
1565 1566 1567
 */
static void worker_leave_idle(struct worker *worker)
{
1568
	struct worker_pool *pool = worker->pool;
T
Tejun Heo 已提交
1569

1570 1571
	if (WARN_ON_ONCE(!(worker->flags & WORKER_IDLE)))
		return;
1572
	worker_clr_flags(worker, WORKER_IDLE);
1573
	pool->nr_idle--;
T
Tejun Heo 已提交
1574 1575 1576
	list_del_init(&worker->entry);
}

1577
/**
1578 1579 1580 1581
 * worker_maybe_bind_and_lock - try to bind %current to worker_pool and lock it
 * @pool: target worker_pool
 *
 * Bind %current to the cpu of @pool if it is associated and lock @pool.
1582 1583 1584 1585 1586 1587
 *
 * Works which are scheduled while the cpu is online must at least be
 * scheduled to a worker which is bound to the cpu so that if they are
 * flushed from cpu callbacks while cpu is going down, they are
 * guaranteed to execute on the cpu.
 *
1588
 * This function is to be used by unbound workers and rescuers to bind
1589 1590 1591
 * themselves to the target cpu and may race with cpu going down or
 * coming online.  kthread_bind() can't be used because it may put the
 * worker to already dead cpu and set_cpus_allowed_ptr() can't be used
1592
 * verbatim as it's best effort and blocking and pool may be
1593 1594
 * [dis]associated in the meantime.
 *
1595
 * This function tries set_cpus_allowed() and locks pool and verifies the
1596
 * binding against %POOL_DISASSOCIATED which is set during
1597 1598 1599
 * %CPU_DOWN_PREPARE and cleared during %CPU_ONLINE, so if the worker
 * enters idle state or fetches works without dropping lock, it can
 * guarantee the scheduling requirement described in the first paragraph.
1600 1601
 *
 * CONTEXT:
1602
 * Might sleep.  Called without any lock but returns with pool->lock
1603 1604 1605
 * held.
 *
 * RETURNS:
1606
 * %true if the associated pool is online (@worker is successfully
1607 1608
 * bound), %false if offline.
 */
1609
static bool worker_maybe_bind_and_lock(struct worker_pool *pool)
1610
__acquires(&pool->lock)
1611 1612
{
	while (true) {
1613
		/*
1614 1615 1616
		 * The following call may fail, succeed or succeed
		 * without actually migrating the task to the cpu if
		 * it races with cpu hotunplug operation.  Verify
1617
		 * against POOL_DISASSOCIATED.
1618
		 */
1619
		if (!(pool->flags & POOL_DISASSOCIATED))
T
Tejun Heo 已提交
1620
			set_cpus_allowed_ptr(current, pool->attrs->cpumask);
1621

1622
		spin_lock_irq(&pool->lock);
1623
		if (pool->flags & POOL_DISASSOCIATED)
1624
			return false;
1625
		if (task_cpu(current) == pool->cpu &&
T
Tejun Heo 已提交
1626
		    cpumask_equal(&current->cpus_allowed, pool->attrs->cpumask))
1627
			return true;
1628
		spin_unlock_irq(&pool->lock);
1629

1630 1631 1632 1633 1634 1635
		/*
		 * We've raced with CPU hot[un]plug.  Give it a breather
		 * and retry migration.  cond_resched() is required here;
		 * otherwise, we might deadlock against cpu_stop trying to
		 * bring down the CPU on non-preemptive kernel.
		 */
1636
		cpu_relax();
1637
		cond_resched();
1638 1639 1640
	}
}

T
Tejun Heo 已提交
1641 1642 1643 1644 1645
static struct worker *alloc_worker(void)
{
	struct worker *worker;

	worker = kzalloc(sizeof(*worker), GFP_KERNEL);
T
Tejun Heo 已提交
1646 1647
	if (worker) {
		INIT_LIST_HEAD(&worker->entry);
1648
		INIT_LIST_HEAD(&worker->scheduled);
1649 1650
		/* on creation a worker is in !idle && prep state */
		worker->flags = WORKER_PREP;
T
Tejun Heo 已提交
1651
	}
T
Tejun Heo 已提交
1652 1653 1654 1655 1656
	return worker;
}

/**
 * create_worker - create a new workqueue worker
1657
 * @pool: pool the new worker will belong to
T
Tejun Heo 已提交
1658
 *
1659
 * Create a new worker which is bound to @pool.  The returned worker
T
Tejun Heo 已提交
1660 1661 1662 1663 1664 1665 1666 1667 1668
 * can be started by calling start_worker() or destroyed using
 * destroy_worker().
 *
 * CONTEXT:
 * Might sleep.  Does GFP_KERNEL allocations.
 *
 * RETURNS:
 * Pointer to the newly created worker.
 */
1669
static struct worker *create_worker(struct worker_pool *pool)
T
Tejun Heo 已提交
1670 1671
{
	struct worker *worker = NULL;
1672
	int id = -1;
1673
	char id_buf[16];
T
Tejun Heo 已提交
1674

1675 1676
	lockdep_assert_held(&pool->manager_mutex);

1677 1678 1679 1680 1681
	/*
	 * ID is needed to determine kthread name.  Allocate ID first
	 * without installing the pointer.
	 */
	idr_preload(GFP_KERNEL);
1682
	spin_lock_irq(&pool->lock);
1683 1684 1685

	id = idr_alloc(&pool->worker_idr, NULL, 0, 0, GFP_NOWAIT);

1686
	spin_unlock_irq(&pool->lock);
1687 1688 1689
	idr_preload_end();
	if (id < 0)
		goto fail;
T
Tejun Heo 已提交
1690 1691 1692 1693 1694

	worker = alloc_worker();
	if (!worker)
		goto fail;

1695
	worker->pool = pool;
T
Tejun Heo 已提交
1696 1697
	worker->id = id;

1698
	if (pool->cpu >= 0)
1699 1700
		snprintf(id_buf, sizeof(id_buf), "%d:%d%s", pool->cpu, id,
			 pool->attrs->nice < 0  ? "H" : "");
1701
	else
1702 1703
		snprintf(id_buf, sizeof(id_buf), "u%d:%d", pool->id, id);

1704
	worker->task = kthread_create_on_node(worker_thread, worker, pool->node,
1705
					      "kworker/%s", id_buf);
T
Tejun Heo 已提交
1706 1707 1708
	if (IS_ERR(worker->task))
		goto fail;

1709 1710 1711 1712
	/*
	 * set_cpus_allowed_ptr() will fail if the cpumask doesn't have any
	 * online CPUs.  It'll be re-applied when any of the CPUs come up.
	 */
T
Tejun Heo 已提交
1713 1714
	set_user_nice(worker->task, pool->attrs->nice);
	set_cpus_allowed_ptr(worker->task, pool->attrs->cpumask);
1715

1716 1717
	/* prevent userland from meddling with cpumask of workqueue workers */
	worker->task->flags |= PF_NO_SETAFFINITY;
T
Tejun Heo 已提交
1718 1719 1720 1721 1722 1723 1724

	/*
	 * The caller is responsible for ensuring %POOL_DISASSOCIATED
	 * remains stable across this function.  See the comments above the
	 * flag definition for details.
	 */
	if (pool->flags & POOL_DISASSOCIATED)
1725
		worker->flags |= WORKER_UNBOUND;
T
Tejun Heo 已提交
1726

1727 1728 1729 1730 1731
	/* successful, commit the pointer to idr */
	spin_lock_irq(&pool->lock);
	idr_replace(&pool->worker_idr, worker, worker->id);
	spin_unlock_irq(&pool->lock);

T
Tejun Heo 已提交
1732
	return worker;
1733

T
Tejun Heo 已提交
1734 1735
fail:
	if (id >= 0) {
1736
		spin_lock_irq(&pool->lock);
1737
		idr_remove(&pool->worker_idr, id);
1738
		spin_unlock_irq(&pool->lock);
T
Tejun Heo 已提交
1739 1740 1741 1742 1743 1744 1745 1746 1747
	}
	kfree(worker);
	return NULL;
}

/**
 * start_worker - start a newly created worker
 * @worker: worker to start
 *
1748
 * Make the pool aware of @worker and start it.
T
Tejun Heo 已提交
1749 1750
 *
 * CONTEXT:
1751
 * spin_lock_irq(pool->lock).
T
Tejun Heo 已提交
1752 1753 1754
 */
static void start_worker(struct worker *worker)
{
1755
	worker->flags |= WORKER_STARTED;
1756
	worker->pool->nr_workers++;
T
Tejun Heo 已提交
1757
	worker_enter_idle(worker);
T
Tejun Heo 已提交
1758 1759 1760
	wake_up_process(worker->task);
}

1761 1762 1763 1764
/**
 * create_and_start_worker - create and start a worker for a pool
 * @pool: the target pool
 *
1765
 * Grab the managership of @pool and create and start a new worker for it.
1766 1767 1768 1769 1770
 */
static int create_and_start_worker(struct worker_pool *pool)
{
	struct worker *worker;

1771 1772
	mutex_lock(&pool->manager_mutex);

1773 1774 1775 1776 1777 1778 1779
	worker = create_worker(pool);
	if (worker) {
		spin_lock_irq(&pool->lock);
		start_worker(worker);
		spin_unlock_irq(&pool->lock);
	}

1780 1781
	mutex_unlock(&pool->manager_mutex);

1782 1783 1784
	return worker ? 0 : -ENOMEM;
}

T
Tejun Heo 已提交
1785 1786 1787 1788
/**
 * destroy_worker - destroy a workqueue worker
 * @worker: worker to be destroyed
 *
1789
 * Destroy @worker and adjust @pool stats accordingly.
T
Tejun Heo 已提交
1790 1791
 *
 * CONTEXT:
1792
 * spin_lock_irq(pool->lock) which is released and regrabbed.
T
Tejun Heo 已提交
1793 1794 1795
 */
static void destroy_worker(struct worker *worker)
{
1796
	struct worker_pool *pool = worker->pool;
T
Tejun Heo 已提交
1797

1798 1799 1800
	lockdep_assert_held(&pool->manager_mutex);
	lockdep_assert_held(&pool->lock);

T
Tejun Heo 已提交
1801
	/* sanity check frenzy */
1802 1803 1804
	if (WARN_ON(worker->current_work) ||
	    WARN_ON(!list_empty(&worker->scheduled)))
		return;
T
Tejun Heo 已提交
1805

T
Tejun Heo 已提交
1806
	if (worker->flags & WORKER_STARTED)
1807
		pool->nr_workers--;
T
Tejun Heo 已提交
1808
	if (worker->flags & WORKER_IDLE)
1809
		pool->nr_idle--;
T
Tejun Heo 已提交
1810 1811

	list_del_init(&worker->entry);
1812
	worker->flags |= WORKER_DIE;
T
Tejun Heo 已提交
1813

1814 1815
	idr_remove(&pool->worker_idr, worker->id);

1816
	spin_unlock_irq(&pool->lock);
T
Tejun Heo 已提交
1817

T
Tejun Heo 已提交
1818 1819 1820
	kthread_stop(worker->task);
	kfree(worker);

1821
	spin_lock_irq(&pool->lock);
T
Tejun Heo 已提交
1822 1823
}

1824
static void idle_worker_timeout(unsigned long __pool)
1825
{
1826
	struct worker_pool *pool = (void *)__pool;
1827

1828
	spin_lock_irq(&pool->lock);
1829

1830
	if (too_many_workers(pool)) {
1831 1832 1833 1834
		struct worker *worker;
		unsigned long expires;

		/* idle_list is kept in LIFO order, check the last one */
1835
		worker = list_entry(pool->idle_list.prev, struct worker, entry);
1836 1837 1838
		expires = worker->last_active + IDLE_WORKER_TIMEOUT;

		if (time_before(jiffies, expires))
1839
			mod_timer(&pool->idle_timer, expires);
1840 1841
		else {
			/* it's been idle for too long, wake up manager */
1842
			pool->flags |= POOL_MANAGE_WORKERS;
1843
			wake_up_worker(pool);
1844
		}
1845 1846
	}

1847
	spin_unlock_irq(&pool->lock);
1848
}
1849

1850
static void send_mayday(struct work_struct *work)
1851
{
1852 1853
	struct pool_workqueue *pwq = get_work_pwq(work);
	struct workqueue_struct *wq = pwq->wq;
1854

1855
	lockdep_assert_held(&wq_mayday_lock);
1856

1857
	if (!wq->rescuer)
1858
		return;
1859 1860

	/* mayday mayday mayday */
1861 1862
	if (list_empty(&pwq->mayday_node)) {
		list_add_tail(&pwq->mayday_node, &wq->maydays);
1863
		wake_up_process(wq->rescuer->task);
1864
	}
1865 1866
}

1867
static void pool_mayday_timeout(unsigned long __pool)
1868
{
1869
	struct worker_pool *pool = (void *)__pool;
1870 1871
	struct work_struct *work;

1872
	spin_lock_irq(&wq_mayday_lock);		/* for wq->maydays */
1873
	spin_lock(&pool->lock);
1874

1875
	if (need_to_create_worker(pool)) {
1876 1877 1878 1879 1880 1881
		/*
		 * We've been trying to create a new worker but
		 * haven't been successful.  We might be hitting an
		 * allocation deadlock.  Send distress signals to
		 * rescuers.
		 */
1882
		list_for_each_entry(work, &pool->worklist, entry)
1883
			send_mayday(work);
L
Linus Torvalds 已提交
1884
	}
1885

1886
	spin_unlock(&pool->lock);
1887
	spin_unlock_irq(&wq_mayday_lock);
1888

1889
	mod_timer(&pool->mayday_timer, jiffies + MAYDAY_INTERVAL);
L
Linus Torvalds 已提交
1890 1891
}

1892 1893
/**
 * maybe_create_worker - create a new worker if necessary
1894
 * @pool: pool to create a new worker for
1895
 *
1896
 * Create a new worker for @pool if necessary.  @pool is guaranteed to
1897 1898
 * have at least one idle worker on return from this function.  If
 * creating a new worker takes longer than MAYDAY_INTERVAL, mayday is
1899
 * sent to all rescuers with works scheduled on @pool to resolve
1900 1901
 * possible allocation deadlock.
 *
1902 1903
 * On return, need_to_create_worker() is guaranteed to be %false and
 * may_start_working() %true.
1904 1905
 *
 * LOCKING:
1906
 * spin_lock_irq(pool->lock) which may be released and regrabbed
1907 1908 1909 1910
 * multiple times.  Does GFP_KERNEL allocations.  Called only from
 * manager.
 *
 * RETURNS:
1911
 * %false if no action was taken and pool->lock stayed locked, %true
1912 1913
 * otherwise.
 */
1914
static bool maybe_create_worker(struct worker_pool *pool)
1915 1916
__releases(&pool->lock)
__acquires(&pool->lock)
L
Linus Torvalds 已提交
1917
{
1918
	if (!need_to_create_worker(pool))
1919 1920
		return false;
restart:
1921
	spin_unlock_irq(&pool->lock);
1922

1923
	/* if we don't make progress in MAYDAY_INITIAL_TIMEOUT, call for help */
1924
	mod_timer(&pool->mayday_timer, jiffies + MAYDAY_INITIAL_TIMEOUT);
1925 1926 1927 1928

	while (true) {
		struct worker *worker;

1929
		worker = create_worker(pool);
1930
		if (worker) {
1931
			del_timer_sync(&pool->mayday_timer);
1932
			spin_lock_irq(&pool->lock);
1933
			start_worker(worker);
1934 1935
			if (WARN_ON_ONCE(need_to_create_worker(pool)))
				goto restart;
1936 1937 1938
			return true;
		}

1939
		if (!need_to_create_worker(pool))
1940
			break;
L
Linus Torvalds 已提交
1941

1942 1943
		__set_current_state(TASK_INTERRUPTIBLE);
		schedule_timeout(CREATE_COOLDOWN);
1944

1945
		if (!need_to_create_worker(pool))
1946 1947 1948
			break;
	}

1949
	del_timer_sync(&pool->mayday_timer);
1950
	spin_lock_irq(&pool->lock);
1951
	if (need_to_create_worker(pool))
1952 1953 1954 1955 1956 1957
		goto restart;
	return true;
}

/**
 * maybe_destroy_worker - destroy workers which have been idle for a while
1958
 * @pool: pool to destroy workers for
1959
 *
1960
 * Destroy @pool workers which have been idle for longer than
1961 1962 1963
 * IDLE_WORKER_TIMEOUT.
 *
 * LOCKING:
1964
 * spin_lock_irq(pool->lock) which may be released and regrabbed
1965 1966 1967
 * multiple times.  Called only from manager.
 *
 * RETURNS:
1968
 * %false if no action was taken and pool->lock stayed locked, %true
1969 1970
 * otherwise.
 */
1971
static bool maybe_destroy_workers(struct worker_pool *pool)
1972 1973
{
	bool ret = false;
L
Linus Torvalds 已提交
1974

1975
	while (too_many_workers(pool)) {
1976 1977
		struct worker *worker;
		unsigned long expires;
1978

1979
		worker = list_entry(pool->idle_list.prev, struct worker, entry);
1980
		expires = worker->last_active + IDLE_WORKER_TIMEOUT;
1981

1982
		if (time_before(jiffies, expires)) {
1983
			mod_timer(&pool->idle_timer, expires);
1984
			break;
1985
		}
L
Linus Torvalds 已提交
1986

1987 1988
		destroy_worker(worker);
		ret = true;
L
Linus Torvalds 已提交
1989
	}
1990

1991
	return ret;
1992 1993
}

1994
/**
1995 1996
 * manage_workers - manage worker pool
 * @worker: self
1997
 *
1998
 * Assume the manager role and manage the worker pool @worker belongs
1999
 * to.  At any given time, there can be only zero or one manager per
2000
 * pool.  The exclusion is handled automatically by this function.
2001 2002 2003 2004
 *
 * The caller can safely start processing works on false return.  On
 * true return, it's guaranteed that need_to_create_worker() is false
 * and may_start_working() is true.
2005 2006
 *
 * CONTEXT:
2007
 * spin_lock_irq(pool->lock) which may be released and regrabbed
2008 2009 2010
 * multiple times.  Does GFP_KERNEL allocations.
 *
 * RETURNS:
2011 2012
 * spin_lock_irq(pool->lock) which may be released and regrabbed
 * multiple times.  Does GFP_KERNEL allocations.
2013
 */
2014
static bool manage_workers(struct worker *worker)
2015
{
2016
	struct worker_pool *pool = worker->pool;
2017
	bool ret = false;
2018

2019 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039
	/*
	 * Managership is governed by two mutexes - manager_arb and
	 * manager_mutex.  manager_arb handles arbitration of manager role.
	 * Anyone who successfully grabs manager_arb wins the arbitration
	 * and becomes the manager.  mutex_trylock() on pool->manager_arb
	 * failure while holding pool->lock reliably indicates that someone
	 * else is managing the pool and the worker which failed trylock
	 * can proceed to executing work items.  This means that anyone
	 * grabbing manager_arb is responsible for actually performing
	 * manager duties.  If manager_arb is grabbed and released without
	 * actual management, the pool may stall indefinitely.
	 *
	 * manager_mutex is used for exclusion of actual management
	 * operations.  The holder of manager_mutex can be sure that none
	 * of management operations, including creation and destruction of
	 * workers, won't take place until the mutex is released.  Because
	 * manager_mutex doesn't interfere with manager role arbitration,
	 * it is guaranteed that the pool's management, while may be
	 * delayed, won't be disturbed by someone else grabbing
	 * manager_mutex.
	 */
2040
	if (!mutex_trylock(&pool->manager_arb))
2041
		return ret;
2042

2043
	/*
2044 2045
	 * With manager arbitration won, manager_mutex would be free in
	 * most cases.  trylock first without dropping @pool->lock.
2046
	 */
2047
	if (unlikely(!mutex_trylock(&pool->manager_mutex))) {
2048
		spin_unlock_irq(&pool->lock);
2049
		mutex_lock(&pool->manager_mutex);
2050 2051
		ret = true;
	}
2052

2053
	pool->flags &= ~POOL_MANAGE_WORKERS;
2054 2055

	/*
2056 2057
	 * Destroy and then create so that may_start_working() is true
	 * on return.
2058
	 */
2059 2060
	ret |= maybe_destroy_workers(pool);
	ret |= maybe_create_worker(pool);
2061

2062
	mutex_unlock(&pool->manager_mutex);
2063
	mutex_unlock(&pool->manager_arb);
2064
	return ret;
2065 2066
}

2067 2068
/**
 * process_one_work - process single work
T
Tejun Heo 已提交
2069
 * @worker: self
2070 2071 2072 2073 2074 2075 2076 2077 2078
 * @work: work to process
 *
 * Process @work.  This function contains all the logics necessary to
 * process a single work including synchronization against and
 * interaction with other workers on the same cpu, queueing and
 * flushing.  As long as context requirement is met, any worker can
 * call this function to process a work.
 *
 * CONTEXT:
2079
 * spin_lock_irq(pool->lock) which is released and regrabbed.
2080
 */
T
Tejun Heo 已提交
2081
static void process_one_work(struct worker *worker, struct work_struct *work)
2082 2083
__releases(&pool->lock)
__acquires(&pool->lock)
2084
{
2085
	struct pool_workqueue *pwq = get_work_pwq(work);
2086
	struct worker_pool *pool = worker->pool;
2087
	bool cpu_intensive = pwq->wq->flags & WQ_CPU_INTENSIVE;
2088
	int work_color;
2089
	struct worker *collision;
2090 2091 2092 2093 2094 2095 2096 2097
#ifdef CONFIG_LOCKDEP
	/*
	 * It is permissible to free the struct work_struct from
	 * inside the function that is called from it, this we need to
	 * take into account for lockdep too.  To avoid bogus "held
	 * lock freed" warnings as well as problems when looking into
	 * work->lockdep_map, make a copy and use that here.
	 */
2098 2099 2100
	struct lockdep_map lockdep_map;

	lockdep_copy_map(&lockdep_map, &work->lockdep_map);
2101
#endif
2102 2103 2104
	/*
	 * Ensure we're on the correct CPU.  DISASSOCIATED test is
	 * necessary to avoid spurious warnings from rescuers servicing the
2105
	 * unbound or a disassociated pool.
2106
	 */
2107
	WARN_ON_ONCE(!(worker->flags & WORKER_UNBOUND) &&
2108
		     !(pool->flags & POOL_DISASSOCIATED) &&
2109
		     raw_smp_processor_id() != pool->cpu);
2110

2111 2112 2113 2114 2115 2116
	/*
	 * A single work shouldn't be executed concurrently by
	 * multiple workers on a single cpu.  Check whether anyone is
	 * already processing the work.  If so, defer the work to the
	 * currently executing one.
	 */
2117
	collision = find_worker_executing_work(pool, work);
2118 2119 2120 2121 2122
	if (unlikely(collision)) {
		move_linked_works(work, &collision->scheduled, NULL);
		return;
	}

2123
	/* claim and dequeue */
2124
	debug_work_deactivate(work);
2125
	hash_add(pool->busy_hash, &worker->hentry, (unsigned long)work);
T
Tejun Heo 已提交
2126
	worker->current_work = work;
2127
	worker->current_func = work->func;
2128
	worker->current_pwq = pwq;
2129
	work_color = get_work_color(work);
2130

2131 2132
	list_del_init(&work->entry);

2133 2134 2135 2136 2137 2138 2139
	/*
	 * CPU intensive works don't participate in concurrency
	 * management.  They're the scheduler's responsibility.
	 */
	if (unlikely(cpu_intensive))
		worker_set_flags(worker, WORKER_CPU_INTENSIVE, true);

2140
	/*
2141
	 * Unbound pool isn't concurrency managed and work items should be
2142 2143
	 * executed ASAP.  Wake up another worker if necessary.
	 */
2144 2145
	if ((worker->flags & WORKER_UNBOUND) && need_more_worker(pool))
		wake_up_worker(pool);
2146

2147
	/*
2148
	 * Record the last pool and clear PENDING which should be the last
2149
	 * update to @work.  Also, do this inside @pool->lock so that
2150 2151
	 * PENDING and queued state changes happen together while IRQ is
	 * disabled.
2152
	 */
2153
	set_work_pool_and_clear_pending(work, pool->id);
2154

2155
	spin_unlock_irq(&pool->lock);
2156

2157
	lock_map_acquire_read(&pwq->wq->lockdep_map);
2158
	lock_map_acquire(&lockdep_map);
2159
	trace_workqueue_execute_start(work);
2160
	worker->current_func(work);
2161 2162 2163 2164 2165
	/*
	 * While we must be careful to not use "work" after this, the trace
	 * point will only record its address.
	 */
	trace_workqueue_execute_end(work);
2166
	lock_map_release(&lockdep_map);
2167
	lock_map_release(&pwq->wq->lockdep_map);
2168 2169

	if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
V
Valentin Ilie 已提交
2170 2171
		pr_err("BUG: workqueue leaked lock or atomic: %s/0x%08x/%d\n"
		       "     last function: %pf\n",
2172 2173
		       current->comm, preempt_count(), task_pid_nr(current),
		       worker->current_func);
2174 2175 2176 2177
		debug_show_held_locks(current);
		dump_stack();
	}

2178
	spin_lock_irq(&pool->lock);
2179

2180 2181 2182 2183
	/* clear cpu intensive status */
	if (unlikely(cpu_intensive))
		worker_clr_flags(worker, WORKER_CPU_INTENSIVE);

2184
	/* we're done with it, release */
2185
	hash_del(&worker->hentry);
T
Tejun Heo 已提交
2186
	worker->current_work = NULL;
2187
	worker->current_func = NULL;
2188 2189
	worker->current_pwq = NULL;
	pwq_dec_nr_in_flight(pwq, work_color);
2190 2191
}

2192 2193 2194 2195 2196 2197 2198 2199 2200
/**
 * process_scheduled_works - process scheduled works
 * @worker: self
 *
 * Process all scheduled works.  Please note that the scheduled list
 * may change while processing a work, so this function repeatedly
 * fetches a work from the top and executes it.
 *
 * CONTEXT:
2201
 * spin_lock_irq(pool->lock) which may be released and regrabbed
2202 2203 2204
 * multiple times.
 */
static void process_scheduled_works(struct worker *worker)
L
Linus Torvalds 已提交
2205
{
2206 2207
	while (!list_empty(&worker->scheduled)) {
		struct work_struct *work = list_first_entry(&worker->scheduled,
L
Linus Torvalds 已提交
2208
						struct work_struct, entry);
T
Tejun Heo 已提交
2209
		process_one_work(worker, work);
L
Linus Torvalds 已提交
2210 2211 2212
	}
}

T
Tejun Heo 已提交
2213 2214
/**
 * worker_thread - the worker thread function
T
Tejun Heo 已提交
2215
 * @__worker: self
T
Tejun Heo 已提交
2216
 *
2217 2218 2219 2220 2221
 * The worker thread function.  All workers belong to a worker_pool -
 * either a per-cpu one or dynamic unbound one.  These workers process all
 * work items regardless of their specific target workqueue.  The only
 * exception is work items which belong to workqueues with a rescuer which
 * will be explained in rescuer_thread().
T
Tejun Heo 已提交
2222
 */
T
Tejun Heo 已提交
2223
static int worker_thread(void *__worker)
L
Linus Torvalds 已提交
2224
{
T
Tejun Heo 已提交
2225
	struct worker *worker = __worker;
2226
	struct worker_pool *pool = worker->pool;
L
Linus Torvalds 已提交
2227

2228 2229
	/* tell the scheduler that this is a workqueue worker */
	worker->task->flags |= PF_WQ_WORKER;
T
Tejun Heo 已提交
2230
woke_up:
2231
	spin_lock_irq(&pool->lock);
L
Linus Torvalds 已提交
2232

2233 2234
	/* am I supposed to die? */
	if (unlikely(worker->flags & WORKER_DIE)) {
2235
		spin_unlock_irq(&pool->lock);
2236 2237 2238
		WARN_ON_ONCE(!list_empty(&worker->entry));
		worker->task->flags &= ~PF_WQ_WORKER;
		return 0;
T
Tejun Heo 已提交
2239
	}
2240

T
Tejun Heo 已提交
2241
	worker_leave_idle(worker);
2242
recheck:
2243
	/* no more worker necessary? */
2244
	if (!need_more_worker(pool))
2245 2246 2247
		goto sleep;

	/* do we need to manage? */
2248
	if (unlikely(!may_start_working(pool)) && manage_workers(worker))
2249 2250
		goto recheck;

T
Tejun Heo 已提交
2251 2252 2253 2254 2255
	/*
	 * ->scheduled list can only be filled while a worker is
	 * preparing to process a work or actually processing it.
	 * Make sure nobody diddled with it while I was sleeping.
	 */
2256
	WARN_ON_ONCE(!list_empty(&worker->scheduled));
T
Tejun Heo 已提交
2257

2258
	/*
2259 2260 2261 2262 2263
	 * Finish PREP stage.  We're guaranteed to have at least one idle
	 * worker or that someone else has already assumed the manager
	 * role.  This is where @worker starts participating in concurrency
	 * management if applicable and concurrency management is restored
	 * after being rebound.  See rebind_workers() for details.
2264
	 */
2265
	worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);
2266 2267

	do {
T
Tejun Heo 已提交
2268
		struct work_struct *work =
2269
			list_first_entry(&pool->worklist,
T
Tejun Heo 已提交
2270 2271 2272 2273 2274 2275
					 struct work_struct, entry);

		if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
			/* optimization path, not strictly necessary */
			process_one_work(worker, work);
			if (unlikely(!list_empty(&worker->scheduled)))
2276
				process_scheduled_works(worker);
T
Tejun Heo 已提交
2277 2278 2279
		} else {
			move_linked_works(work, &worker->scheduled, NULL);
			process_scheduled_works(worker);
2280
		}
2281
	} while (keep_working(pool));
2282 2283

	worker_set_flags(worker, WORKER_PREP, false);
2284
sleep:
2285
	if (unlikely(need_to_manage_workers(pool)) && manage_workers(worker))
2286
		goto recheck;
2287

T
Tejun Heo 已提交
2288
	/*
2289 2290 2291 2292 2293
	 * pool->lock is held and there's no work to process and no need to
	 * manage, sleep.  Workers are woken up only while holding
	 * pool->lock or from local cpu, so setting the current state
	 * before releasing pool->lock is enough to prevent losing any
	 * event.
T
Tejun Heo 已提交
2294 2295 2296
	 */
	worker_enter_idle(worker);
	__set_current_state(TASK_INTERRUPTIBLE);
2297
	spin_unlock_irq(&pool->lock);
T
Tejun Heo 已提交
2298 2299
	schedule();
	goto woke_up;
L
Linus Torvalds 已提交
2300 2301
}

2302 2303
/**
 * rescuer_thread - the rescuer thread function
2304
 * @__rescuer: self
2305 2306
 *
 * Workqueue rescuer thread function.  There's one rescuer for each
2307
 * workqueue which has WQ_MEM_RECLAIM set.
2308
 *
2309
 * Regular work processing on a pool may block trying to create a new
2310 2311 2312 2313 2314
 * worker which uses GFP_KERNEL allocation which has slight chance of
 * developing into deadlock if some works currently on the same queue
 * need to be processed to satisfy the GFP_KERNEL allocation.  This is
 * the problem rescuer solves.
 *
2315 2316
 * When such condition is possible, the pool summons rescuers of all
 * workqueues which have works queued on the pool and let them process
2317 2318 2319 2320
 * those works so that forward progress can be guaranteed.
 *
 * This should happen rarely.
 */
2321
static int rescuer_thread(void *__rescuer)
2322
{
2323 2324
	struct worker *rescuer = __rescuer;
	struct workqueue_struct *wq = rescuer->rescue_wq;
2325 2326 2327
	struct list_head *scheduled = &rescuer->scheduled;

	set_user_nice(current, RESCUER_NICE_LEVEL);
2328 2329 2330 2331 2332 2333

	/*
	 * Mark rescuer as worker too.  As WORKER_PREP is never cleared, it
	 * doesn't participate in concurrency management.
	 */
	rescuer->task->flags |= PF_WQ_WORKER;
2334 2335 2336
repeat:
	set_current_state(TASK_INTERRUPTIBLE);

2337 2338
	if (kthread_should_stop()) {
		__set_current_state(TASK_RUNNING);
2339
		rescuer->task->flags &= ~PF_WQ_WORKER;
2340
		return 0;
2341
	}
2342

2343
	/* see whether any pwq is asking for help */
2344
	spin_lock_irq(&wq_mayday_lock);
2345 2346 2347 2348

	while (!list_empty(&wq->maydays)) {
		struct pool_workqueue *pwq = list_first_entry(&wq->maydays,
					struct pool_workqueue, mayday_node);
2349
		struct worker_pool *pool = pwq->pool;
2350 2351 2352
		struct work_struct *work, *n;

		__set_current_state(TASK_RUNNING);
2353 2354
		list_del_init(&pwq->mayday_node);

2355
		spin_unlock_irq(&wq_mayday_lock);
2356 2357

		/* migrate to the target cpu if possible */
2358
		worker_maybe_bind_and_lock(pool);
2359
		rescuer->pool = pool;
2360 2361 2362 2363 2364

		/*
		 * Slurp in all works issued via this workqueue and
		 * process'em.
		 */
2365
		WARN_ON_ONCE(!list_empty(&rescuer->scheduled));
2366
		list_for_each_entry_safe(work, n, &pool->worklist, entry)
2367
			if (get_work_pwq(work) == pwq)
2368 2369 2370
				move_linked_works(work, scheduled, &n);

		process_scheduled_works(rescuer);
2371 2372

		/*
2373
		 * Leave this pool.  If keep_working() is %true, notify a
2374 2375 2376
		 * regular worker; otherwise, we end up with 0 concurrency
		 * and stalling the execution.
		 */
2377 2378
		if (keep_working(pool))
			wake_up_worker(pool);
2379

2380
		rescuer->pool = NULL;
2381
		spin_unlock(&pool->lock);
2382
		spin_lock(&wq_mayday_lock);
2383 2384
	}

2385
	spin_unlock_irq(&wq_mayday_lock);
2386

2387 2388
	/* rescuers should never participate in concurrency management */
	WARN_ON_ONCE(!(rescuer->flags & WORKER_NOT_RUNNING));
2389 2390
	schedule();
	goto repeat;
L
Linus Torvalds 已提交
2391 2392
}

O
Oleg Nesterov 已提交
2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403
struct wq_barrier {
	struct work_struct	work;
	struct completion	done;
};

static void wq_barrier_func(struct work_struct *work)
{
	struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
	complete(&barr->done);
}

T
Tejun Heo 已提交
2404 2405
/**
 * insert_wq_barrier - insert a barrier work
2406
 * @pwq: pwq to insert barrier into
T
Tejun Heo 已提交
2407
 * @barr: wq_barrier to insert
2408 2409
 * @target: target work to attach @barr to
 * @worker: worker currently executing @target, NULL if @target is not executing
T
Tejun Heo 已提交
2410
 *
2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422
 * @barr is linked to @target such that @barr is completed only after
 * @target finishes execution.  Please note that the ordering
 * guarantee is observed only with respect to @target and on the local
 * cpu.
 *
 * Currently, a queued barrier can't be canceled.  This is because
 * try_to_grab_pending() can't determine whether the work to be
 * grabbed is at the head of the queue and thus can't clear LINKED
 * flag of the previous work while there must be a valid next work
 * after a work with LINKED flag set.
 *
 * Note that when @worker is non-NULL, @target may be modified
2423
 * underneath us, so we can't reliably determine pwq from @target.
T
Tejun Heo 已提交
2424 2425
 *
 * CONTEXT:
2426
 * spin_lock_irq(pool->lock).
T
Tejun Heo 已提交
2427
 */
2428
static void insert_wq_barrier(struct pool_workqueue *pwq,
2429 2430
			      struct wq_barrier *barr,
			      struct work_struct *target, struct worker *worker)
O
Oleg Nesterov 已提交
2431
{
2432 2433 2434
	struct list_head *head;
	unsigned int linked = 0;

2435
	/*
2436
	 * debugobject calls are safe here even with pool->lock locked
2437 2438 2439 2440
	 * as we know for sure that this will not trigger any of the
	 * checks and call back into the fixup functions where we
	 * might deadlock.
	 */
A
Andrew Morton 已提交
2441
	INIT_WORK_ONSTACK(&barr->work, wq_barrier_func);
2442
	__set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(&barr->work));
O
Oleg Nesterov 已提交
2443
	init_completion(&barr->done);
2444

2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459
	/*
	 * If @target is currently being executed, schedule the
	 * barrier to the worker; otherwise, put it after @target.
	 */
	if (worker)
		head = worker->scheduled.next;
	else {
		unsigned long *bits = work_data_bits(target);

		head = target->entry.next;
		/* there can already be other linked works, inherit and set */
		linked = *bits & WORK_STRUCT_LINKED;
		__set_bit(WORK_STRUCT_LINKED_BIT, bits);
	}

2460
	debug_work_activate(&barr->work);
2461
	insert_work(pwq, &barr->work, head,
2462
		    work_color_to_flags(WORK_NO_COLOR) | linked);
O
Oleg Nesterov 已提交
2463 2464
}

2465
/**
2466
 * flush_workqueue_prep_pwqs - prepare pwqs for workqueue flushing
2467 2468 2469 2470
 * @wq: workqueue being flushed
 * @flush_color: new flush color, < 0 for no-op
 * @work_color: new work color, < 0 for no-op
 *
2471
 * Prepare pwqs for workqueue flushing.
2472
 *
2473 2474 2475 2476 2477
 * If @flush_color is non-negative, flush_color on all pwqs should be
 * -1.  If no pwq has in-flight commands at the specified color, all
 * pwq->flush_color's stay at -1 and %false is returned.  If any pwq
 * has in flight commands, its pwq->flush_color is set to
 * @flush_color, @wq->nr_pwqs_to_flush is updated accordingly, pwq
2478 2479 2480 2481 2482 2483 2484
 * wakeup logic is armed and %true is returned.
 *
 * The caller should have initialized @wq->first_flusher prior to
 * calling this function with non-negative @flush_color.  If
 * @flush_color is negative, no flush color update is done and %false
 * is returned.
 *
2485
 * If @work_color is non-negative, all pwqs should have the same
2486 2487 2488 2489
 * work_color which is previous to @work_color and all will be
 * advanced to @work_color.
 *
 * CONTEXT:
2490
 * mutex_lock(wq->mutex).
2491 2492 2493 2494 2495
 *
 * RETURNS:
 * %true if @flush_color >= 0 and there's something to flush.  %false
 * otherwise.
 */
2496
static bool flush_workqueue_prep_pwqs(struct workqueue_struct *wq,
2497
				      int flush_color, int work_color)
L
Linus Torvalds 已提交
2498
{
2499
	bool wait = false;
2500
	struct pool_workqueue *pwq;
L
Linus Torvalds 已提交
2501

2502
	if (flush_color >= 0) {
2503
		WARN_ON_ONCE(atomic_read(&wq->nr_pwqs_to_flush));
2504
		atomic_set(&wq->nr_pwqs_to_flush, 1);
L
Linus Torvalds 已提交
2505
	}
2506

2507
	for_each_pwq(pwq, wq) {
2508
		struct worker_pool *pool = pwq->pool;
O
Oleg Nesterov 已提交
2509

2510
		spin_lock_irq(&pool->lock);
2511

2512
		if (flush_color >= 0) {
2513
			WARN_ON_ONCE(pwq->flush_color != -1);
O
Oleg Nesterov 已提交
2514

2515 2516 2517
			if (pwq->nr_in_flight[flush_color]) {
				pwq->flush_color = flush_color;
				atomic_inc(&wq->nr_pwqs_to_flush);
2518 2519 2520
				wait = true;
			}
		}
L
Linus Torvalds 已提交
2521

2522
		if (work_color >= 0) {
2523
			WARN_ON_ONCE(work_color != work_next_color(pwq->work_color));
2524
			pwq->work_color = work_color;
2525
		}
L
Linus Torvalds 已提交
2526

2527
		spin_unlock_irq(&pool->lock);
L
Linus Torvalds 已提交
2528
	}
2529

2530
	if (flush_color >= 0 && atomic_dec_and_test(&wq->nr_pwqs_to_flush))
2531
		complete(&wq->first_flusher->done);
2532

2533
	return wait;
L
Linus Torvalds 已提交
2534 2535
}

2536
/**
L
Linus Torvalds 已提交
2537
 * flush_workqueue - ensure that any scheduled work has run to completion.
2538
 * @wq: workqueue to flush
L
Linus Torvalds 已提交
2539
 *
2540 2541
 * This function sleeps until all work items which were queued on entry
 * have finished execution, but it is not livelocked by new incoming ones.
L
Linus Torvalds 已提交
2542
 */
2543
void flush_workqueue(struct workqueue_struct *wq)
L
Linus Torvalds 已提交
2544
{
2545 2546 2547 2548 2549 2550
	struct wq_flusher this_flusher = {
		.list = LIST_HEAD_INIT(this_flusher.list),
		.flush_color = -1,
		.done = COMPLETION_INITIALIZER_ONSTACK(this_flusher.done),
	};
	int next_color;
L
Linus Torvalds 已提交
2551

2552 2553
	lock_map_acquire(&wq->lockdep_map);
	lock_map_release(&wq->lockdep_map);
2554

2555
	mutex_lock(&wq->mutex);
2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567

	/*
	 * Start-to-wait phase
	 */
	next_color = work_next_color(wq->work_color);

	if (next_color != wq->flush_color) {
		/*
		 * Color space is not full.  The current work_color
		 * becomes our flush_color and work_color is advanced
		 * by one.
		 */
2568
		WARN_ON_ONCE(!list_empty(&wq->flusher_overflow));
2569 2570 2571 2572 2573
		this_flusher.flush_color = wq->work_color;
		wq->work_color = next_color;

		if (!wq->first_flusher) {
			/* no flush in progress, become the first flusher */
2574
			WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);
2575 2576 2577

			wq->first_flusher = &this_flusher;

2578
			if (!flush_workqueue_prep_pwqs(wq, wq->flush_color,
2579 2580 2581 2582 2583 2584 2585 2586
						       wq->work_color)) {
				/* nothing to flush, done */
				wq->flush_color = next_color;
				wq->first_flusher = NULL;
				goto out_unlock;
			}
		} else {
			/* wait in queue */
2587
			WARN_ON_ONCE(wq->flush_color == this_flusher.flush_color);
2588
			list_add_tail(&this_flusher.list, &wq->flusher_queue);
2589
			flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
2590 2591 2592 2593 2594 2595 2596 2597 2598 2599
		}
	} else {
		/*
		 * Oops, color space is full, wait on overflow queue.
		 * The next flush completion will assign us
		 * flush_color and transfer to flusher_queue.
		 */
		list_add_tail(&this_flusher.list, &wq->flusher_overflow);
	}

2600
	mutex_unlock(&wq->mutex);
2601 2602 2603 2604 2605 2606 2607 2608 2609 2610 2611 2612

	wait_for_completion(&this_flusher.done);

	/*
	 * Wake-up-and-cascade phase
	 *
	 * First flushers are responsible for cascading flushes and
	 * handling overflow.  Non-first flushers can simply return.
	 */
	if (wq->first_flusher != &this_flusher)
		return;

2613
	mutex_lock(&wq->mutex);
2614

2615 2616 2617 2618
	/* we might have raced, check again with mutex held */
	if (wq->first_flusher != &this_flusher)
		goto out_unlock;

2619 2620
	wq->first_flusher = NULL;

2621 2622
	WARN_ON_ONCE(!list_empty(&this_flusher.list));
	WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);
2623 2624 2625 2626 2627 2628 2629 2630 2631 2632 2633 2634

	while (true) {
		struct wq_flusher *next, *tmp;

		/* complete all the flushers sharing the current flush color */
		list_for_each_entry_safe(next, tmp, &wq->flusher_queue, list) {
			if (next->flush_color != wq->flush_color)
				break;
			list_del_init(&next->list);
			complete(&next->done);
		}

2635 2636
		WARN_ON_ONCE(!list_empty(&wq->flusher_overflow) &&
			     wq->flush_color != work_next_color(wq->work_color));
2637 2638 2639 2640 2641 2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655

		/* this flush_color is finished, advance by one */
		wq->flush_color = work_next_color(wq->flush_color);

		/* one color has been freed, handle overflow queue */
		if (!list_empty(&wq->flusher_overflow)) {
			/*
			 * Assign the same color to all overflowed
			 * flushers, advance work_color and append to
			 * flusher_queue.  This is the start-to-wait
			 * phase for these overflowed flushers.
			 */
			list_for_each_entry(tmp, &wq->flusher_overflow, list)
				tmp->flush_color = wq->work_color;

			wq->work_color = work_next_color(wq->work_color);

			list_splice_tail_init(&wq->flusher_overflow,
					      &wq->flusher_queue);
2656
			flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
2657 2658 2659
		}

		if (list_empty(&wq->flusher_queue)) {
2660
			WARN_ON_ONCE(wq->flush_color != wq->work_color);
2661 2662 2663 2664 2665
			break;
		}

		/*
		 * Need to flush more colors.  Make the next flusher
2666
		 * the new first flusher and arm pwqs.
2667
		 */
2668 2669
		WARN_ON_ONCE(wq->flush_color == wq->work_color);
		WARN_ON_ONCE(wq->flush_color != next->flush_color);
2670 2671 2672 2673

		list_del_init(&next->list);
		wq->first_flusher = next;

2674
		if (flush_workqueue_prep_pwqs(wq, wq->flush_color, -1))
2675 2676 2677 2678 2679 2680 2681 2682 2683 2684
			break;

		/*
		 * Meh... this color is already done, clear first
		 * flusher and repeat cascading.
		 */
		wq->first_flusher = NULL;
	}

out_unlock:
2685
	mutex_unlock(&wq->mutex);
L
Linus Torvalds 已提交
2686
}
2687
EXPORT_SYMBOL_GPL(flush_workqueue);
L
Linus Torvalds 已提交
2688

2689 2690 2691 2692 2693 2694 2695 2696 2697 2698 2699 2700 2701 2702
/**
 * drain_workqueue - drain a workqueue
 * @wq: workqueue to drain
 *
 * Wait until the workqueue becomes empty.  While draining is in progress,
 * only chain queueing is allowed.  IOW, only currently pending or running
 * work items on @wq can queue further work items on it.  @wq is flushed
 * repeatedly until it becomes empty.  The number of flushing is detemined
 * by the depth of chaining and should be relatively short.  Whine if it
 * takes too long.
 */
void drain_workqueue(struct workqueue_struct *wq)
{
	unsigned int flush_cnt = 0;
2703
	struct pool_workqueue *pwq;
2704 2705 2706 2707

	/*
	 * __queue_work() needs to test whether there are drainers, is much
	 * hotter than drain_workqueue() and already looks at @wq->flags.
2708
	 * Use __WQ_DRAINING so that queue doesn't have to check nr_drainers.
2709
	 */
2710
	mutex_lock(&wq->mutex);
2711
	if (!wq->nr_drainers++)
2712
		wq->flags |= __WQ_DRAINING;
2713
	mutex_unlock(&wq->mutex);
2714 2715 2716
reflush:
	flush_workqueue(wq);

2717
	mutex_lock(&wq->mutex);
2718

2719
	for_each_pwq(pwq, wq) {
2720
		bool drained;
2721

2722
		spin_lock_irq(&pwq->pool->lock);
2723
		drained = !pwq->nr_active && list_empty(&pwq->delayed_works);
2724
		spin_unlock_irq(&pwq->pool->lock);
2725 2726

		if (drained)
2727 2728 2729 2730
			continue;

		if (++flush_cnt == 10 ||
		    (flush_cnt % 100 == 0 && flush_cnt <= 1000))
2731
			pr_warn("workqueue %s: drain_workqueue() isn't complete after %u tries\n",
V
Valentin Ilie 已提交
2732
				wq->name, flush_cnt);
2733

2734
		mutex_unlock(&wq->mutex);
2735 2736 2737 2738
		goto reflush;
	}

	if (!--wq->nr_drainers)
2739
		wq->flags &= ~__WQ_DRAINING;
2740
	mutex_unlock(&wq->mutex);
2741 2742 2743
}
EXPORT_SYMBOL_GPL(drain_workqueue);

2744
static bool start_flush_work(struct work_struct *work, struct wq_barrier *barr)
2745
{
2746
	struct worker *worker = NULL;
2747
	struct worker_pool *pool;
2748
	struct pool_workqueue *pwq;
2749 2750

	might_sleep();
2751 2752

	local_irq_disable();
2753
	pool = get_work_pool(work);
2754 2755
	if (!pool) {
		local_irq_enable();
2756
		return false;
2757
	}
2758

2759
	spin_lock(&pool->lock);
2760
	/* see the comment in try_to_grab_pending() with the same code */
2761 2762 2763
	pwq = get_work_pwq(work);
	if (pwq) {
		if (unlikely(pwq->pool != pool))
T
Tejun Heo 已提交
2764
			goto already_gone;
2765
	} else {
2766
		worker = find_worker_executing_work(pool, work);
2767
		if (!worker)
T
Tejun Heo 已提交
2768
			goto already_gone;
2769
		pwq = worker->current_pwq;
2770
	}
2771

2772
	insert_wq_barrier(pwq, barr, work, worker);
2773
	spin_unlock_irq(&pool->lock);
2774

2775 2776 2777 2778 2779 2780
	/*
	 * If @max_active is 1 or rescuer is in use, flushing another work
	 * item on the same workqueue may lead to deadlock.  Make sure the
	 * flusher is not running on the same workqueue by verifying write
	 * access.
	 */
2781
	if (pwq->wq->saved_max_active == 1 || pwq->wq->rescuer)
2782
		lock_map_acquire(&pwq->wq->lockdep_map);
2783
	else
2784 2785
		lock_map_acquire_read(&pwq->wq->lockdep_map);
	lock_map_release(&pwq->wq->lockdep_map);
2786

2787
	return true;
T
Tejun Heo 已提交
2788
already_gone:
2789
	spin_unlock_irq(&pool->lock);
2790
	return false;
2791
}
2792 2793 2794 2795 2796

/**
 * flush_work - wait for a work to finish executing the last queueing instance
 * @work: the work to flush
 *
2797 2798
 * Wait until @work has finished execution.  @work is guaranteed to be idle
 * on return if it hasn't been requeued since flush started.
2799 2800 2801 2802 2803 2804 2805 2806 2807
 *
 * RETURNS:
 * %true if flush_work() waited for the work to finish execution,
 * %false if it was already idle.
 */
bool flush_work(struct work_struct *work)
{
	struct wq_barrier barr;

2808 2809 2810
	lock_map_acquire(&work->lockdep_map);
	lock_map_release(&work->lockdep_map);

2811
	if (start_flush_work(work, &barr)) {
2812 2813 2814
		wait_for_completion(&barr.done);
		destroy_work_on_stack(&barr.work);
		return true;
2815
	} else {
2816
		return false;
2817 2818
	}
}
2819
EXPORT_SYMBOL_GPL(flush_work);
2820

2821
static bool __cancel_work_timer(struct work_struct *work, bool is_dwork)
2822
{
2823
	unsigned long flags;
2824 2825 2826
	int ret;

	do {
2827 2828 2829 2830 2831 2832
		ret = try_to_grab_pending(work, is_dwork, &flags);
		/*
		 * If someone else is canceling, wait for the same event it
		 * would be waiting for before retrying.
		 */
		if (unlikely(ret == -ENOENT))
2833
			flush_work(work);
2834 2835
	} while (unlikely(ret < 0));

2836 2837 2838 2839
	/* tell other tasks trying to grab @work to back off */
	mark_work_canceling(work);
	local_irq_restore(flags);

2840
	flush_work(work);
2841
	clear_work_data(work);
2842 2843 2844
	return ret;
}

2845
/**
2846 2847
 * cancel_work_sync - cancel a work and wait for it to finish
 * @work: the work to cancel
2848
 *
2849 2850 2851 2852
 * Cancel @work and wait for its execution to finish.  This function
 * can be used even if the work re-queues itself or migrates to
 * another workqueue.  On return from this function, @work is
 * guaranteed to be not pending or executing on any CPU.
2853
 *
2854 2855
 * cancel_work_sync(&delayed_work->work) must not be used for
 * delayed_work's.  Use cancel_delayed_work_sync() instead.
2856
 *
2857
 * The caller must ensure that the workqueue on which @work was last
2858
 * queued can't be destroyed before this function returns.
2859 2860 2861
 *
 * RETURNS:
 * %true if @work was pending, %false otherwise.
2862
 */
2863
bool cancel_work_sync(struct work_struct *work)
2864
{
2865
	return __cancel_work_timer(work, false);
O
Oleg Nesterov 已提交
2866
}
2867
EXPORT_SYMBOL_GPL(cancel_work_sync);
O
Oleg Nesterov 已提交
2868

2869
/**
2870 2871
 * flush_delayed_work - wait for a dwork to finish executing the last queueing
 * @dwork: the delayed work to flush
2872
 *
2873 2874 2875
 * Delayed timer is cancelled and the pending work is queued for
 * immediate execution.  Like flush_work(), this function only
 * considers the last queueing instance of @dwork.
2876
 *
2877 2878 2879
 * RETURNS:
 * %true if flush_work() waited for the work to finish execution,
 * %false if it was already idle.
2880
 */
2881 2882
bool flush_delayed_work(struct delayed_work *dwork)
{
2883
	local_irq_disable();
2884
	if (del_timer_sync(&dwork->timer))
2885
		__queue_work(dwork->cpu, dwork->wq, &dwork->work);
2886
	local_irq_enable();
2887 2888 2889 2890
	return flush_work(&dwork->work);
}
EXPORT_SYMBOL(flush_delayed_work);

2891
/**
2892 2893
 * cancel_delayed_work - cancel a delayed work
 * @dwork: delayed_work to cancel
2894
 *
2895 2896 2897 2898 2899
 * Kill off a pending delayed_work.  Returns %true if @dwork was pending
 * and canceled; %false if wasn't pending.  Note that the work callback
 * function may still be running on return, unless it returns %true and the
 * work doesn't re-arm itself.  Explicitly flush or use
 * cancel_delayed_work_sync() to wait on it.
2900
 *
2901
 * This function is safe to call from any context including IRQ handler.
2902
 */
2903
bool cancel_delayed_work(struct delayed_work *dwork)
2904
{
2905 2906 2907 2908 2909 2910 2911 2912 2913 2914
	unsigned long flags;
	int ret;

	do {
		ret = try_to_grab_pending(&dwork->work, true, &flags);
	} while (unlikely(ret == -EAGAIN));

	if (unlikely(ret < 0))
		return false;

2915 2916
	set_work_pool_and_clear_pending(&dwork->work,
					get_work_pool_id(&dwork->work));
2917
	local_irq_restore(flags);
2918
	return ret;
2919
}
2920
EXPORT_SYMBOL(cancel_delayed_work);
2921

2922 2923 2924 2925 2926 2927 2928 2929 2930 2931
/**
 * cancel_delayed_work_sync - cancel a delayed work and wait for it to finish
 * @dwork: the delayed work cancel
 *
 * This is cancel_work_sync() for delayed works.
 *
 * RETURNS:
 * %true if @dwork was pending, %false otherwise.
 */
bool cancel_delayed_work_sync(struct delayed_work *dwork)
2932
{
2933
	return __cancel_work_timer(&dwork->work, true);
2934
}
2935
EXPORT_SYMBOL(cancel_delayed_work_sync);
L
Linus Torvalds 已提交
2936

2937
/**
2938
 * schedule_on_each_cpu - execute a function synchronously on each online CPU
2939 2940
 * @func: the function to call
 *
2941 2942
 * schedule_on_each_cpu() executes @func on each online CPU using the
 * system workqueue and blocks until all CPUs have completed.
2943
 * schedule_on_each_cpu() is very slow.
2944 2945 2946
 *
 * RETURNS:
 * 0 on success, -errno on failure.
2947
 */
2948
int schedule_on_each_cpu(work_func_t func)
2949 2950
{
	int cpu;
2951
	struct work_struct __percpu *works;
2952

2953 2954
	works = alloc_percpu(struct work_struct);
	if (!works)
2955
		return -ENOMEM;
2956

2957 2958
	get_online_cpus();

2959
	for_each_online_cpu(cpu) {
2960 2961 2962
		struct work_struct *work = per_cpu_ptr(works, cpu);

		INIT_WORK(work, func);
2963
		schedule_work_on(cpu, work);
2964
	}
2965 2966 2967 2968

	for_each_online_cpu(cpu)
		flush_work(per_cpu_ptr(works, cpu));

2969
	put_online_cpus();
2970
	free_percpu(works);
2971 2972 2973
	return 0;
}

2974 2975 2976 2977 2978 2979 2980 2981 2982 2983 2984 2985 2986 2987 2988 2989 2990 2991 2992 2993 2994 2995 2996 2997
/**
 * flush_scheduled_work - ensure that any scheduled work has run to completion.
 *
 * Forces execution of the kernel-global workqueue and blocks until its
 * completion.
 *
 * Think twice before calling this function!  It's very easy to get into
 * trouble if you don't take great care.  Either of the following situations
 * will lead to deadlock:
 *
 *	One of the work items currently on the workqueue needs to acquire
 *	a lock held by your code or its caller.
 *
 *	Your code is running in the context of a work routine.
 *
 * They will be detected by lockdep when they occur, but the first might not
 * occur very often.  It depends on what work items are on the workqueue and
 * what locks they need, which you have no control over.
 *
 * In most situations flushing the entire workqueue is overkill; you merely
 * need to know that a particular work item isn't queued and isn't running.
 * In such cases you should use cancel_delayed_work_sync() or
 * cancel_work_sync() instead.
 */
L
Linus Torvalds 已提交
2998 2999
void flush_scheduled_work(void)
{
3000
	flush_workqueue(system_wq);
L
Linus Torvalds 已提交
3001
}
3002
EXPORT_SYMBOL(flush_scheduled_work);
L
Linus Torvalds 已提交
3003

3004 3005 3006 3007 3008 3009 3010 3011 3012 3013 3014 3015
/**
 * execute_in_process_context - reliably execute the routine with user context
 * @fn:		the function to execute
 * @ew:		guaranteed storage for the execute work structure (must
 *		be available when the work executes)
 *
 * Executes the function immediately if process context is available,
 * otherwise schedules the function for delayed execution.
 *
 * Returns:	0 - function was executed
 *		1 - function was scheduled for execution
 */
3016
int execute_in_process_context(work_func_t fn, struct execute_work *ew)
3017 3018
{
	if (!in_interrupt()) {
3019
		fn(&ew->work);
3020 3021 3022
		return 0;
	}

3023
	INIT_WORK(&ew->work, fn);
3024 3025 3026 3027 3028 3029
	schedule_work(&ew->work);

	return 1;
}
EXPORT_SYMBOL_GPL(execute_in_process_context);

3030 3031 3032 3033 3034 3035 3036 3037 3038 3039 3040 3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052 3053 3054 3055 3056 3057 3058 3059 3060 3061 3062 3063 3064 3065 3066 3067 3068 3069 3070 3071 3072 3073 3074 3075 3076 3077 3078 3079 3080 3081 3082 3083 3084 3085 3086 3087 3088 3089 3090 3091 3092 3093 3094 3095 3096 3097 3098 3099 3100 3101 3102 3103 3104 3105 3106 3107 3108 3109 3110 3111 3112 3113
#ifdef CONFIG_SYSFS
/*
 * Workqueues with WQ_SYSFS flag set is visible to userland via
 * /sys/bus/workqueue/devices/WQ_NAME.  All visible workqueues have the
 * following attributes.
 *
 *  per_cpu	RO bool	: whether the workqueue is per-cpu or unbound
 *  max_active	RW int	: maximum number of in-flight work items
 *
 * Unbound workqueues have the following extra attributes.
 *
 *  id		RO int	: the associated pool ID
 *  nice	RW int	: nice value of the workers
 *  cpumask	RW mask	: bitmask of allowed CPUs for the workers
 */
struct wq_device {
	struct workqueue_struct		*wq;
	struct device			dev;
};

static struct workqueue_struct *dev_to_wq(struct device *dev)
{
	struct wq_device *wq_dev = container_of(dev, struct wq_device, dev);

	return wq_dev->wq;
}

static ssize_t wq_per_cpu_show(struct device *dev,
			       struct device_attribute *attr, char *buf)
{
	struct workqueue_struct *wq = dev_to_wq(dev);

	return scnprintf(buf, PAGE_SIZE, "%d\n", (bool)!(wq->flags & WQ_UNBOUND));
}

static ssize_t wq_max_active_show(struct device *dev,
				  struct device_attribute *attr, char *buf)
{
	struct workqueue_struct *wq = dev_to_wq(dev);

	return scnprintf(buf, PAGE_SIZE, "%d\n", wq->saved_max_active);
}

static ssize_t wq_max_active_store(struct device *dev,
				   struct device_attribute *attr,
				   const char *buf, size_t count)
{
	struct workqueue_struct *wq = dev_to_wq(dev);
	int val;

	if (sscanf(buf, "%d", &val) != 1 || val <= 0)
		return -EINVAL;

	workqueue_set_max_active(wq, val);
	return count;
}

static struct device_attribute wq_sysfs_attrs[] = {
	__ATTR(per_cpu, 0444, wq_per_cpu_show, NULL),
	__ATTR(max_active, 0644, wq_max_active_show, wq_max_active_store),
	__ATTR_NULL,
};

static ssize_t wq_pool_id_show(struct device *dev,
			       struct device_attribute *attr, char *buf)
{
	struct workqueue_struct *wq = dev_to_wq(dev);
	struct worker_pool *pool;
	int written;

	rcu_read_lock_sched();
	pool = first_pwq(wq)->pool;
	written = scnprintf(buf, PAGE_SIZE, "%d\n", pool->id);
	rcu_read_unlock_sched();

	return written;
}

static ssize_t wq_nice_show(struct device *dev, struct device_attribute *attr,
			    char *buf)
{
	struct workqueue_struct *wq = dev_to_wq(dev);
	int written;

3114 3115 3116
	mutex_lock(&wq->mutex);
	written = scnprintf(buf, PAGE_SIZE, "%d\n", wq->unbound_attrs->nice);
	mutex_unlock(&wq->mutex);
3117 3118 3119 3120 3121 3122 3123 3124 3125 3126 3127 3128 3129

	return written;
}

/* prepare workqueue_attrs for sysfs store operations */
static struct workqueue_attrs *wq_sysfs_prep_attrs(struct workqueue_struct *wq)
{
	struct workqueue_attrs *attrs;

	attrs = alloc_workqueue_attrs(GFP_KERNEL);
	if (!attrs)
		return NULL;

3130 3131 3132
	mutex_lock(&wq->mutex);
	copy_workqueue_attrs(attrs, wq->unbound_attrs);
	mutex_unlock(&wq->mutex);
3133 3134 3135 3136 3137 3138 3139 3140 3141 3142 3143 3144 3145 3146 3147 3148 3149 3150 3151 3152 3153 3154 3155 3156 3157 3158 3159 3160 3161 3162
	return attrs;
}

static ssize_t wq_nice_store(struct device *dev, struct device_attribute *attr,
			     const char *buf, size_t count)
{
	struct workqueue_struct *wq = dev_to_wq(dev);
	struct workqueue_attrs *attrs;
	int ret;

	attrs = wq_sysfs_prep_attrs(wq);
	if (!attrs)
		return -ENOMEM;

	if (sscanf(buf, "%d", &attrs->nice) == 1 &&
	    attrs->nice >= -20 && attrs->nice <= 19)
		ret = apply_workqueue_attrs(wq, attrs);
	else
		ret = -EINVAL;

	free_workqueue_attrs(attrs);
	return ret ?: count;
}

static ssize_t wq_cpumask_show(struct device *dev,
			       struct device_attribute *attr, char *buf)
{
	struct workqueue_struct *wq = dev_to_wq(dev);
	int written;

3163 3164 3165
	mutex_lock(&wq->mutex);
	written = cpumask_scnprintf(buf, PAGE_SIZE, wq->unbound_attrs->cpumask);
	mutex_unlock(&wq->mutex);
3166 3167 3168 3169 3170 3171 3172 3173 3174 3175 3176 3177 3178 3179 3180 3181 3182 3183 3184 3185 3186 3187 3188 3189 3190 3191 3192 3193 3194 3195 3196 3197 3198 3199 3200 3201 3202 3203 3204 3205 3206 3207 3208 3209 3210 3211 3212 3213 3214 3215 3216 3217 3218 3219 3220 3221 3222 3223 3224 3225 3226 3227 3228 3229 3230 3231 3232 3233 3234 3235 3236 3237 3238 3239 3240 3241 3242 3243 3244 3245 3246 3247 3248 3249 3250 3251 3252 3253 3254 3255 3256 3257 3258 3259 3260 3261 3262 3263 3264 3265 3266 3267 3268 3269 3270 3271 3272 3273 3274 3275 3276 3277 3278 3279 3280 3281 3282 3283 3284 3285 3286 3287 3288 3289 3290 3291 3292 3293 3294 3295 3296 3297 3298 3299 3300 3301 3302

	written += scnprintf(buf + written, PAGE_SIZE - written, "\n");
	return written;
}

static ssize_t wq_cpumask_store(struct device *dev,
				struct device_attribute *attr,
				const char *buf, size_t count)
{
	struct workqueue_struct *wq = dev_to_wq(dev);
	struct workqueue_attrs *attrs;
	int ret;

	attrs = wq_sysfs_prep_attrs(wq);
	if (!attrs)
		return -ENOMEM;

	ret = cpumask_parse(buf, attrs->cpumask);
	if (!ret)
		ret = apply_workqueue_attrs(wq, attrs);

	free_workqueue_attrs(attrs);
	return ret ?: count;
}

static struct device_attribute wq_sysfs_unbound_attrs[] = {
	__ATTR(pool_id, 0444, wq_pool_id_show, NULL),
	__ATTR(nice, 0644, wq_nice_show, wq_nice_store),
	__ATTR(cpumask, 0644, wq_cpumask_show, wq_cpumask_store),
	__ATTR_NULL,
};

static struct bus_type wq_subsys = {
	.name				= "workqueue",
	.dev_attrs			= wq_sysfs_attrs,
};

static int __init wq_sysfs_init(void)
{
	return subsys_virtual_register(&wq_subsys, NULL);
}
core_initcall(wq_sysfs_init);

static void wq_device_release(struct device *dev)
{
	struct wq_device *wq_dev = container_of(dev, struct wq_device, dev);

	kfree(wq_dev);
}

/**
 * workqueue_sysfs_register - make a workqueue visible in sysfs
 * @wq: the workqueue to register
 *
 * Expose @wq in sysfs under /sys/bus/workqueue/devices.
 * alloc_workqueue*() automatically calls this function if WQ_SYSFS is set
 * which is the preferred method.
 *
 * Workqueue user should use this function directly iff it wants to apply
 * workqueue_attrs before making the workqueue visible in sysfs; otherwise,
 * apply_workqueue_attrs() may race against userland updating the
 * attributes.
 *
 * Returns 0 on success, -errno on failure.
 */
int workqueue_sysfs_register(struct workqueue_struct *wq)
{
	struct wq_device *wq_dev;
	int ret;

	/*
	 * Adjusting max_active or creating new pwqs by applyting
	 * attributes breaks ordering guarantee.  Disallow exposing ordered
	 * workqueues.
	 */
	if (WARN_ON(wq->flags & __WQ_ORDERED))
		return -EINVAL;

	wq->wq_dev = wq_dev = kzalloc(sizeof(*wq_dev), GFP_KERNEL);
	if (!wq_dev)
		return -ENOMEM;

	wq_dev->wq = wq;
	wq_dev->dev.bus = &wq_subsys;
	wq_dev->dev.init_name = wq->name;
	wq_dev->dev.release = wq_device_release;

	/*
	 * unbound_attrs are created separately.  Suppress uevent until
	 * everything is ready.
	 */
	dev_set_uevent_suppress(&wq_dev->dev, true);

	ret = device_register(&wq_dev->dev);
	if (ret) {
		kfree(wq_dev);
		wq->wq_dev = NULL;
		return ret;
	}

	if (wq->flags & WQ_UNBOUND) {
		struct device_attribute *attr;

		for (attr = wq_sysfs_unbound_attrs; attr->attr.name; attr++) {
			ret = device_create_file(&wq_dev->dev, attr);
			if (ret) {
				device_unregister(&wq_dev->dev);
				wq->wq_dev = NULL;
				return ret;
			}
		}
	}

	kobject_uevent(&wq_dev->dev.kobj, KOBJ_ADD);
	return 0;
}

/**
 * workqueue_sysfs_unregister - undo workqueue_sysfs_register()
 * @wq: the workqueue to unregister
 *
 * If @wq is registered to sysfs by workqueue_sysfs_register(), unregister.
 */
static void workqueue_sysfs_unregister(struct workqueue_struct *wq)
{
	struct wq_device *wq_dev = wq->wq_dev;

	if (!wq->wq_dev)
		return;

	wq->wq_dev = NULL;
	device_unregister(&wq_dev->dev);
}
#else	/* CONFIG_SYSFS */
static void workqueue_sysfs_unregister(struct workqueue_struct *wq)	{ }
#endif	/* CONFIG_SYSFS */

T
Tejun Heo 已提交
3303 3304 3305 3306 3307 3308 3309 3310 3311 3312 3313 3314 3315 3316 3317 3318 3319 3320 3321 3322 3323 3324 3325 3326 3327 3328 3329 3330 3331 3332 3333
/**
 * free_workqueue_attrs - free a workqueue_attrs
 * @attrs: workqueue_attrs to free
 *
 * Undo alloc_workqueue_attrs().
 */
void free_workqueue_attrs(struct workqueue_attrs *attrs)
{
	if (attrs) {
		free_cpumask_var(attrs->cpumask);
		kfree(attrs);
	}
}

/**
 * alloc_workqueue_attrs - allocate a workqueue_attrs
 * @gfp_mask: allocation mask to use
 *
 * Allocate a new workqueue_attrs, initialize with default settings and
 * return it.  Returns NULL on failure.
 */
struct workqueue_attrs *alloc_workqueue_attrs(gfp_t gfp_mask)
{
	struct workqueue_attrs *attrs;

	attrs = kzalloc(sizeof(*attrs), gfp_mask);
	if (!attrs)
		goto fail;
	if (!alloc_cpumask_var(&attrs->cpumask, gfp_mask))
		goto fail;

3334
	cpumask_copy(attrs->cpumask, cpu_possible_mask);
T
Tejun Heo 已提交
3335 3336 3337 3338 3339 3340
	return attrs;
fail:
	free_workqueue_attrs(attrs);
	return NULL;
}

3341 3342 3343 3344 3345 3346 3347 3348 3349 3350 3351 3352 3353
static void copy_workqueue_attrs(struct workqueue_attrs *to,
				 const struct workqueue_attrs *from)
{
	to->nice = from->nice;
	cpumask_copy(to->cpumask, from->cpumask);
}

/* hash value of the content of @attr */
static u32 wqattrs_hash(const struct workqueue_attrs *attrs)
{
	u32 hash = 0;

	hash = jhash_1word(attrs->nice, hash);
3354 3355
	hash = jhash(cpumask_bits(attrs->cpumask),
		     BITS_TO_LONGS(nr_cpumask_bits) * sizeof(long), hash);
3356 3357 3358 3359 3360 3361 3362 3363 3364 3365 3366 3367 3368 3369
	return hash;
}

/* content equality test */
static bool wqattrs_equal(const struct workqueue_attrs *a,
			  const struct workqueue_attrs *b)
{
	if (a->nice != b->nice)
		return false;
	if (!cpumask_equal(a->cpumask, b->cpumask))
		return false;
	return true;
}

T
Tejun Heo 已提交
3370 3371 3372 3373 3374
/**
 * init_worker_pool - initialize a newly zalloc'd worker_pool
 * @pool: worker_pool to initialize
 *
 * Initiailize a newly zalloc'd @pool.  It also allocates @pool->attrs.
3375 3376 3377
 * Returns 0 on success, -errno on failure.  Even on failure, all fields
 * inside @pool proper are initialized and put_unbound_pool() can be called
 * on @pool safely to release it.
T
Tejun Heo 已提交
3378 3379
 */
static int init_worker_pool(struct worker_pool *pool)
3380 3381
{
	spin_lock_init(&pool->lock);
3382 3383
	pool->id = -1;
	pool->cpu = -1;
3384
	pool->node = NUMA_NO_NODE;
3385 3386 3387 3388 3389 3390 3391 3392 3393 3394 3395 3396 3397
	pool->flags |= POOL_DISASSOCIATED;
	INIT_LIST_HEAD(&pool->worklist);
	INIT_LIST_HEAD(&pool->idle_list);
	hash_init(pool->busy_hash);

	init_timer_deferrable(&pool->idle_timer);
	pool->idle_timer.function = idle_worker_timeout;
	pool->idle_timer.data = (unsigned long)pool;

	setup_timer(&pool->mayday_timer, pool_mayday_timeout,
		    (unsigned long)pool);

	mutex_init(&pool->manager_arb);
3398
	mutex_init(&pool->manager_mutex);
3399
	idr_init(&pool->worker_idr);
T
Tejun Heo 已提交
3400

3401 3402 3403 3404
	INIT_HLIST_NODE(&pool->hash_node);
	pool->refcnt = 1;

	/* shouldn't fail above this point */
T
Tejun Heo 已提交
3405 3406 3407 3408
	pool->attrs = alloc_workqueue_attrs(GFP_KERNEL);
	if (!pool->attrs)
		return -ENOMEM;
	return 0;
3409 3410
}

3411 3412 3413 3414
static void rcu_free_pool(struct rcu_head *rcu)
{
	struct worker_pool *pool = container_of(rcu, struct worker_pool, rcu);

3415
	idr_destroy(&pool->worker_idr);
3416 3417 3418 3419 3420 3421 3422 3423 3424
	free_workqueue_attrs(pool->attrs);
	kfree(pool);
}

/**
 * put_unbound_pool - put a worker_pool
 * @pool: worker_pool to put
 *
 * Put @pool.  If its refcnt reaches zero, it gets destroyed in sched-RCU
3425 3426 3427
 * safe manner.  get_unbound_pool() calls this function on its failure path
 * and this function should be able to release pools which went through,
 * successfully or not, init_worker_pool().
3428 3429
 *
 * Should be called with wq_pool_mutex held.
3430 3431 3432 3433 3434
 */
static void put_unbound_pool(struct worker_pool *pool)
{
	struct worker *worker;

3435 3436 3437
	lockdep_assert_held(&wq_pool_mutex);

	if (--pool->refcnt)
3438 3439 3440 3441
		return;

	/* sanity checks */
	if (WARN_ON(!(pool->flags & POOL_DISASSOCIATED)) ||
3442
	    WARN_ON(!list_empty(&pool->worklist)))
3443 3444 3445 3446 3447 3448 3449
		return;

	/* release id and unhash */
	if (pool->id >= 0)
		idr_remove(&worker_pool_idr, pool->id);
	hash_del(&pool->hash_node);

3450 3451 3452 3453 3454
	/*
	 * Become the manager and destroy all workers.  Grabbing
	 * manager_arb prevents @pool's workers from blocking on
	 * manager_mutex.
	 */
3455
	mutex_lock(&pool->manager_arb);
3456
	mutex_lock(&pool->manager_mutex);
3457 3458 3459 3460 3461 3462 3463
	spin_lock_irq(&pool->lock);

	while ((worker = first_worker(pool)))
		destroy_worker(worker);
	WARN_ON(pool->nr_workers || pool->nr_idle);

	spin_unlock_irq(&pool->lock);
3464
	mutex_unlock(&pool->manager_mutex);
3465 3466 3467 3468 3469 3470 3471 3472 3473 3474 3475 3476 3477 3478 3479 3480 3481 3482
	mutex_unlock(&pool->manager_arb);

	/* shut down the timers */
	del_timer_sync(&pool->idle_timer);
	del_timer_sync(&pool->mayday_timer);

	/* sched-RCU protected to allow dereferences from get_work_pool() */
	call_rcu_sched(&pool->rcu, rcu_free_pool);
}

/**
 * get_unbound_pool - get a worker_pool with the specified attributes
 * @attrs: the attributes of the worker_pool to get
 *
 * Obtain a worker_pool which has the same attributes as @attrs, bump the
 * reference count and return it.  If there already is a matching
 * worker_pool, it will be used; otherwise, this function attempts to
 * create a new one.  On failure, returns NULL.
3483 3484
 *
 * Should be called with wq_pool_mutex held.
3485 3486 3487 3488 3489
 */
static struct worker_pool *get_unbound_pool(const struct workqueue_attrs *attrs)
{
	u32 hash = wqattrs_hash(attrs);
	struct worker_pool *pool;
3490
	int node;
3491

3492
	lockdep_assert_held(&wq_pool_mutex);
3493 3494 3495 3496 3497 3498 3499 3500 3501 3502 3503 3504 3505 3506

	/* do we already have a matching pool? */
	hash_for_each_possible(unbound_pool_hash, pool, hash_node, hash) {
		if (wqattrs_equal(pool->attrs, attrs)) {
			pool->refcnt++;
			goto out_unlock;
		}
	}

	/* nope, create a new one */
	pool = kzalloc(sizeof(*pool), GFP_KERNEL);
	if (!pool || init_worker_pool(pool) < 0)
		goto fail;

3507 3508 3509
	if (workqueue_freezing)
		pool->flags |= POOL_FREEZING;

T
Tejun Heo 已提交
3510
	lockdep_set_subclass(&pool->lock, 1);	/* see put_pwq() */
3511 3512
	copy_workqueue_attrs(pool->attrs, attrs);

3513 3514 3515 3516 3517 3518 3519 3520 3521 3522 3523
	/* if cpumask is contained inside a NUMA node, we belong to that node */
	if (wq_numa_enabled) {
		for_each_node(node) {
			if (cpumask_subset(pool->attrs->cpumask,
					   wq_numa_possible_cpumask[node])) {
				pool->node = node;
				break;
			}
		}
	}

3524 3525 3526 3527
	if (worker_pool_assign_id(pool) < 0)
		goto fail;

	/* create and start the initial worker */
3528
	if (create_and_start_worker(pool) < 0)
3529 3530 3531 3532 3533 3534 3535 3536 3537 3538 3539 3540
		goto fail;

	/* install */
	hash_add(unbound_pool_hash, &pool->hash_node, hash);
out_unlock:
	return pool;
fail:
	if (pool)
		put_unbound_pool(pool);
	return NULL;
}

T
Tejun Heo 已提交
3541 3542 3543 3544 3545 3546 3547 3548 3549 3550 3551 3552 3553 3554 3555 3556
static void rcu_free_pwq(struct rcu_head *rcu)
{
	kmem_cache_free(pwq_cache,
			container_of(rcu, struct pool_workqueue, rcu));
}

/*
 * Scheduled on system_wq by put_pwq() when an unbound pwq hits zero refcnt
 * and needs to be destroyed.
 */
static void pwq_unbound_release_workfn(struct work_struct *work)
{
	struct pool_workqueue *pwq = container_of(work, struct pool_workqueue,
						  unbound_release_work);
	struct workqueue_struct *wq = pwq->wq;
	struct worker_pool *pool = pwq->pool;
3557
	bool is_last;
T
Tejun Heo 已提交
3558 3559 3560 3561

	if (WARN_ON_ONCE(!(wq->flags & WQ_UNBOUND)))
		return;

3562
	/*
3563
	 * Unlink @pwq.  Synchronization against wq->mutex isn't strictly
3564 3565 3566
	 * necessary on release but do it anyway.  It's easier to verify
	 * and consistent with the linking path.
	 */
3567
	mutex_lock(&wq->mutex);
T
Tejun Heo 已提交
3568
	list_del_rcu(&pwq->pwqs_node);
3569
	is_last = list_empty(&wq->pwqs);
3570
	mutex_unlock(&wq->mutex);
T
Tejun Heo 已提交
3571

3572
	mutex_lock(&wq_pool_mutex);
T
Tejun Heo 已提交
3573
	put_unbound_pool(pool);
3574 3575
	mutex_unlock(&wq_pool_mutex);

T
Tejun Heo 已提交
3576 3577 3578 3579 3580 3581
	call_rcu_sched(&pwq->rcu, rcu_free_pwq);

	/*
	 * If we're the last pwq going away, @wq is already dead and no one
	 * is gonna access it anymore.  Free it.
	 */
3582 3583
	if (is_last) {
		free_workqueue_attrs(wq->unbound_attrs);
T
Tejun Heo 已提交
3584
		kfree(wq);
3585
	}
T
Tejun Heo 已提交
3586 3587
}

3588
/**
3589
 * pwq_adjust_max_active - update a pwq's max_active to the current setting
3590 3591
 * @pwq: target pool_workqueue
 *
3592 3593 3594
 * If @pwq isn't freezing, set @pwq->max_active to the associated
 * workqueue's saved_max_active and activate delayed work items
 * accordingly.  If @pwq is freezing, clear @pwq->max_active to zero.
3595
 */
3596
static void pwq_adjust_max_active(struct pool_workqueue *pwq)
3597
{
3598 3599 3600 3601
	struct workqueue_struct *wq = pwq->wq;
	bool freezable = wq->flags & WQ_FREEZABLE;

	/* for @wq->saved_max_active */
3602
	lockdep_assert_held(&wq->mutex);
3603 3604 3605 3606 3607

	/* fast exit for non-freezable wqs */
	if (!freezable && pwq->max_active == wq->saved_max_active)
		return;

3608
	spin_lock_irq(&pwq->pool->lock);
3609 3610 3611

	if (!freezable || !(pwq->pool->flags & POOL_FREEZING)) {
		pwq->max_active = wq->saved_max_active;
3612

3613 3614 3615
		while (!list_empty(&pwq->delayed_works) &&
		       pwq->nr_active < pwq->max_active)
			pwq_activate_first_delayed(pwq);
3616 3617 3618 3619 3620 3621

		/*
		 * Need to kick a worker after thawed or an unbound wq's
		 * max_active is bumped.  It's a slow path.  Do it always.
		 */
		wake_up_worker(pwq->pool);
3622 3623 3624 3625
	} else {
		pwq->max_active = 0;
	}

3626
	spin_unlock_irq(&pwq->pool->lock);
3627 3628
}

3629
/* initialize newly alloced @pwq which is associated with @wq and @pool */
3630 3631
static void init_pwq(struct pool_workqueue *pwq, struct workqueue_struct *wq,
		     struct worker_pool *pool)
3632 3633 3634
{
	BUG_ON((unsigned long)pwq & WORK_STRUCT_FLAG_MASK);

3635 3636
	memset(pwq, 0, sizeof(*pwq));

3637 3638 3639
	pwq->pool = pool;
	pwq->wq = wq;
	pwq->flush_color = -1;
T
Tejun Heo 已提交
3640
	pwq->refcnt = 1;
3641 3642
	INIT_LIST_HEAD(&pwq->delayed_works);
	INIT_LIST_HEAD(&pwq->mayday_node);
T
Tejun Heo 已提交
3643
	INIT_WORK(&pwq->unbound_release_work, pwq_unbound_release_workfn);
3644
}
3645

3646 3647 3648 3649 3650 3651 3652
/* sync @pwq with the current state of its associated wq and link it */
static void link_pwq(struct pool_workqueue *pwq,
		     struct pool_workqueue **p_last_pwq)
{
	struct workqueue_struct *wq = pwq->wq;

	lockdep_assert_held(&wq->mutex);
3653

3654 3655
	/*
	 * Set the matching work_color.  This is synchronized with
3656
	 * wq->mutex to avoid confusing flush_workqueue().
3657
	 */
3658 3659
	if (p_last_pwq)
		*p_last_pwq = first_pwq(wq);
3660
	pwq->work_color = wq->work_color;
3661 3662 3663 3664 3665

	/* sync max_active to the current setting */
	pwq_adjust_max_active(pwq);

	/* link in @pwq */
3666
	list_add_rcu(&pwq->pwqs_node, &wq->pwqs);
3667
}
3668

3669 3670 3671 3672 3673 3674 3675 3676 3677 3678 3679 3680 3681
/* obtain a pool matching @attr and create a pwq associating the pool and @wq */
static struct pool_workqueue *alloc_unbound_pwq(struct workqueue_struct *wq,
					const struct workqueue_attrs *attrs)
{
	struct worker_pool *pool;
	struct pool_workqueue *pwq;

	lockdep_assert_held(&wq_pool_mutex);

	pool = get_unbound_pool(attrs);
	if (!pool)
		return NULL;

3682
	pwq = kmem_cache_alloc_node(pwq_cache, GFP_KERNEL, pool->node);
3683 3684 3685
	if (!pwq) {
		put_unbound_pool(pool);
		return NULL;
3686
	}
3687

3688 3689
	init_pwq(pwq, wq, pool);
	return pwq;
3690 3691
}

3692 3693 3694 3695 3696 3697 3698 3699 3700 3701 3702 3703 3704 3705 3706 3707 3708
/**
 * apply_workqueue_attrs - apply new workqueue_attrs to an unbound workqueue
 * @wq: the target workqueue
 * @attrs: the workqueue_attrs to apply, allocated with alloc_workqueue_attrs()
 *
 * Apply @attrs to an unbound workqueue @wq.  If @attrs doesn't match the
 * current attributes, a new pwq is created and made the first pwq which
 * will serve all new work items.  Older pwqs are released as in-flight
 * work items finish.  Note that a work item which repeatedly requeues
 * itself back-to-back will stay on its current pwq.
 *
 * Performs GFP_KERNEL allocations.  Returns 0 on success and -errno on
 * failure.
 */
int apply_workqueue_attrs(struct workqueue_struct *wq,
			  const struct workqueue_attrs *attrs)
{
3709
	struct workqueue_attrs *new_attrs;
3710 3711
	struct pool_workqueue *pwq, *last_pwq;
	int node, ret;
3712

3713
	/* only unbound workqueues can change attributes */
3714 3715 3716
	if (WARN_ON(!(wq->flags & WQ_UNBOUND)))
		return -EINVAL;

3717 3718 3719 3720
	/* creating multiple pwqs breaks ordering guarantee */
	if (WARN_ON((wq->flags & __WQ_ORDERED) && !list_empty(&wq->pwqs)))
		return -EINVAL;

3721 3722 3723 3724 3725 3726 3727 3728
	/* make a copy of @attrs and sanitize it */
	new_attrs = alloc_workqueue_attrs(GFP_KERNEL);
	if (!new_attrs)
		goto enomem;

	copy_workqueue_attrs(new_attrs, attrs);
	cpumask_and(new_attrs->cpumask, new_attrs->cpumask, cpu_possible_mask);

3729
	mutex_lock(&wq_pool_mutex);
3730 3731 3732
	pwq = alloc_unbound_pwq(wq, new_attrs);
	mutex_unlock(&wq_pool_mutex);
	if (!pwq)
3733
		goto enomem;
3734

3735
	mutex_lock(&wq->mutex);
3736

3737 3738 3739 3740 3741 3742 3743
	link_pwq(pwq, &last_pwq);

	copy_workqueue_attrs(wq->unbound_attrs, new_attrs);
	for_each_node(node)
		rcu_assign_pointer(wq->numa_pwq_tbl[node], pwq);

	mutex_unlock(&wq->mutex);
3744 3745 3746 3747 3748 3749 3750

	if (last_pwq) {
		spin_lock_irq(&last_pwq->pool->lock);
		put_pwq(last_pwq);
		spin_unlock_irq(&last_pwq->pool->lock);
	}

3751 3752 3753 3754 3755
	ret = 0;
	/* fall through */
out_free:
	free_workqueue_attrs(new_attrs);
	return ret;
3756 3757

enomem:
3758 3759
	ret = -ENOMEM;
	goto out_free;
3760 3761
}

3762
static int alloc_and_link_pwqs(struct workqueue_struct *wq)
T
Tejun Heo 已提交
3763
{
3764
	bool highpri = wq->flags & WQ_HIGHPRI;
3765 3766 3767
	int cpu;

	if (!(wq->flags & WQ_UNBOUND)) {
3768 3769
		wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);
		if (!wq->cpu_pwqs)
3770 3771 3772
			return -ENOMEM;

		for_each_possible_cpu(cpu) {
3773 3774
			struct pool_workqueue *pwq =
				per_cpu_ptr(wq->cpu_pwqs, cpu);
3775
			struct worker_pool *cpu_pools =
3776
				per_cpu(cpu_worker_pools, cpu);
3777

3778 3779 3780 3781 3782
			init_pwq(pwq, wq, &cpu_pools[highpri]);

			mutex_lock(&wq->mutex);
			link_pwq(pwq, NULL);
			mutex_unlock(&wq->mutex);
3783
		}
3784
		return 0;
3785
	} else {
3786
		return apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
3787
	}
T
Tejun Heo 已提交
3788 3789
}

3790 3791
static int wq_clamp_max_active(int max_active, unsigned int flags,
			       const char *name)
3792
{
3793 3794 3795
	int lim = flags & WQ_UNBOUND ? WQ_UNBOUND_MAX_ACTIVE : WQ_MAX_ACTIVE;

	if (max_active < 1 || max_active > lim)
V
Valentin Ilie 已提交
3796 3797
		pr_warn("workqueue: max_active %d requested for %s is out of range, clamping between %d and %d\n",
			max_active, name, 1, lim);
3798

3799
	return clamp_val(max_active, 1, lim);
3800 3801
}

3802
struct workqueue_struct *__alloc_workqueue_key(const char *fmt,
3803 3804 3805
					       unsigned int flags,
					       int max_active,
					       struct lock_class_key *key,
3806
					       const char *lock_name, ...)
L
Linus Torvalds 已提交
3807
{
3808
	size_t tbl_size = 0;
3809
	va_list args;
L
Linus Torvalds 已提交
3810
	struct workqueue_struct *wq;
3811
	struct pool_workqueue *pwq;
3812

3813
	/* allocate wq and format name */
3814 3815 3816 3817
	if (flags & WQ_UNBOUND)
		tbl_size = wq_numa_tbl_len * sizeof(wq->numa_pwq_tbl[0]);

	wq = kzalloc(sizeof(*wq) + tbl_size, GFP_KERNEL);
3818
	if (!wq)
3819
		return NULL;
3820

3821 3822 3823 3824 3825 3826
	if (flags & WQ_UNBOUND) {
		wq->unbound_attrs = alloc_workqueue_attrs(GFP_KERNEL);
		if (!wq->unbound_attrs)
			goto err_free_wq;
	}

3827 3828
	va_start(args, lock_name);
	vsnprintf(wq->name, sizeof(wq->name), fmt, args);
3829
	va_end(args);
L
Linus Torvalds 已提交
3830

3831
	max_active = max_active ?: WQ_DFL_ACTIVE;
3832
	max_active = wq_clamp_max_active(max_active, flags, wq->name);
3833

3834
	/* init wq */
3835
	wq->flags = flags;
3836
	wq->saved_max_active = max_active;
3837
	mutex_init(&wq->mutex);
3838
	atomic_set(&wq->nr_pwqs_to_flush, 0);
3839
	INIT_LIST_HEAD(&wq->pwqs);
3840 3841
	INIT_LIST_HEAD(&wq->flusher_queue);
	INIT_LIST_HEAD(&wq->flusher_overflow);
3842
	INIT_LIST_HEAD(&wq->maydays);
3843

3844
	lockdep_init_map(&wq->lockdep_map, lock_name, key, 0);
3845
	INIT_LIST_HEAD(&wq->list);
3846

3847
	if (alloc_and_link_pwqs(wq) < 0)
3848
		goto err_free_wq;
T
Tejun Heo 已提交
3849

3850 3851 3852 3853 3854
	/*
	 * Workqueues which may be used during memory reclaim should
	 * have a rescuer to guarantee forward progress.
	 */
	if (flags & WQ_MEM_RECLAIM) {
3855 3856
		struct worker *rescuer;

3857
		rescuer = alloc_worker();
3858
		if (!rescuer)
3859
			goto err_destroy;
3860

3861 3862
		rescuer->rescue_wq = wq;
		rescuer->task = kthread_create(rescuer_thread, rescuer, "%s",
3863
					       wq->name);
3864 3865 3866 3867
		if (IS_ERR(rescuer->task)) {
			kfree(rescuer);
			goto err_destroy;
		}
3868

3869
		wq->rescuer = rescuer;
3870
		rescuer->task->flags |= PF_NO_SETAFFINITY;
3871
		wake_up_process(rescuer->task);
3872 3873
	}

3874 3875 3876
	if ((wq->flags & WQ_SYSFS) && workqueue_sysfs_register(wq))
		goto err_destroy;

3877
	/*
3878 3879 3880
	 * wq_pool_mutex protects global freeze state and workqueues list.
	 * Grab it, adjust max_active and add the new @wq to workqueues
	 * list.
3881
	 */
3882
	mutex_lock(&wq_pool_mutex);
3883

3884
	mutex_lock(&wq->mutex);
3885 3886
	for_each_pwq(pwq, wq)
		pwq_adjust_max_active(pwq);
3887
	mutex_unlock(&wq->mutex);
3888

T
Tejun Heo 已提交
3889
	list_add(&wq->list, &workqueues);
3890

3891
	mutex_unlock(&wq_pool_mutex);
T
Tejun Heo 已提交
3892

3893
	return wq;
3894 3895

err_free_wq:
3896
	free_workqueue_attrs(wq->unbound_attrs);
3897 3898 3899 3900
	kfree(wq);
	return NULL;
err_destroy:
	destroy_workqueue(wq);
T
Tejun Heo 已提交
3901
	return NULL;
3902
}
3903
EXPORT_SYMBOL_GPL(__alloc_workqueue_key);
L
Linus Torvalds 已提交
3904

3905 3906 3907 3908 3909 3910 3911 3912
/**
 * destroy_workqueue - safely terminate a workqueue
 * @wq: target workqueue
 *
 * Safely destroy a workqueue. All work currently pending will be done first.
 */
void destroy_workqueue(struct workqueue_struct *wq)
{
3913
	struct pool_workqueue *pwq;
3914

3915 3916
	/* drain it before proceeding with destruction */
	drain_workqueue(wq);
3917

3918
	/* sanity checks */
3919
	mutex_lock(&wq->mutex);
3920
	for_each_pwq(pwq, wq) {
3921 3922
		int i;

3923 3924
		for (i = 0; i < WORK_NR_COLORS; i++) {
			if (WARN_ON(pwq->nr_in_flight[i])) {
3925
				mutex_unlock(&wq->mutex);
3926
				return;
3927 3928 3929
			}
		}

T
Tejun Heo 已提交
3930 3931
		if (WARN_ON(pwq->refcnt > 1) ||
		    WARN_ON(pwq->nr_active) ||
3932
		    WARN_ON(!list_empty(&pwq->delayed_works))) {
3933
			mutex_unlock(&wq->mutex);
3934
			return;
3935
		}
3936
	}
3937
	mutex_unlock(&wq->mutex);
3938

3939 3940 3941 3942
	/*
	 * wq list is used to freeze wq, remove from list after
	 * flushing is complete in case freeze races us.
	 */
3943
	mutex_lock(&wq_pool_mutex);
3944
	list_del_init(&wq->list);
3945
	mutex_unlock(&wq_pool_mutex);
3946

3947 3948
	workqueue_sysfs_unregister(wq);

3949
	if (wq->rescuer) {
3950
		kthread_stop(wq->rescuer->task);
3951
		kfree(wq->rescuer);
3952
		wq->rescuer = NULL;
3953 3954
	}

T
Tejun Heo 已提交
3955 3956 3957 3958 3959 3960 3961 3962 3963 3964 3965 3966 3967 3968 3969
	if (!(wq->flags & WQ_UNBOUND)) {
		/*
		 * The base ref is never dropped on per-cpu pwqs.  Directly
		 * free the pwqs and wq.
		 */
		free_percpu(wq->cpu_pwqs);
		kfree(wq);
	} else {
		/*
		 * We're the sole accessor of @wq at this point.  Directly
		 * access the first pwq and put the base ref.  As both pwqs
		 * and pools are sched-RCU protected, the lock operations
		 * are safe.  @wq will be freed when the last pwq is
		 * released.
		 */
3970 3971
		pwq = list_first_entry(&wq->pwqs, struct pool_workqueue,
				       pwqs_node);
T
Tejun Heo 已提交
3972 3973 3974
		spin_lock_irq(&pwq->pool->lock);
		put_pwq(pwq);
		spin_unlock_irq(&pwq->pool->lock);
3975
	}
3976 3977 3978
}
EXPORT_SYMBOL_GPL(destroy_workqueue);

3979 3980 3981 3982 3983 3984 3985 3986 3987 3988 3989 3990
/**
 * workqueue_set_max_active - adjust max_active of a workqueue
 * @wq: target workqueue
 * @max_active: new max_active value.
 *
 * Set max_active of @wq to @max_active.
 *
 * CONTEXT:
 * Don't call from IRQ context.
 */
void workqueue_set_max_active(struct workqueue_struct *wq, int max_active)
{
3991
	struct pool_workqueue *pwq;
3992

3993 3994 3995 3996
	/* disallow meddling with max_active for ordered workqueues */
	if (WARN_ON(wq->flags & __WQ_ORDERED))
		return;

3997
	max_active = wq_clamp_max_active(max_active, wq->flags, wq->name);
3998

3999
	mutex_lock(&wq->mutex);
4000 4001 4002

	wq->saved_max_active = max_active;

4003 4004
	for_each_pwq(pwq, wq)
		pwq_adjust_max_active(pwq);
4005

4006
	mutex_unlock(&wq->mutex);
4007
}
4008
EXPORT_SYMBOL_GPL(workqueue_set_max_active);
4009

4010 4011 4012 4013 4014 4015 4016 4017 4018 4019
/**
 * current_is_workqueue_rescuer - is %current workqueue rescuer?
 *
 * Determine whether %current is a workqueue rescuer.  Can be used from
 * work functions to determine whether it's being run off the rescuer task.
 */
bool current_is_workqueue_rescuer(void)
{
	struct worker *worker = current_wq_worker();

4020
	return worker && worker->rescue_wq;
4021 4022
}

4023
/**
4024 4025 4026
 * workqueue_congested - test whether a workqueue is congested
 * @cpu: CPU in question
 * @wq: target workqueue
4027
 *
4028 4029 4030
 * Test whether @wq's cpu workqueue for @cpu is congested.  There is
 * no synchronization around this function and the test result is
 * unreliable and only useful as advisory hints or for debugging.
4031
 *
4032 4033
 * RETURNS:
 * %true if congested, %false otherwise.
4034
 */
4035
bool workqueue_congested(int cpu, struct workqueue_struct *wq)
L
Linus Torvalds 已提交
4036
{
4037
	struct pool_workqueue *pwq;
4038 4039
	bool ret;

4040
	rcu_read_lock_sched();
4041 4042 4043 4044

	if (!(wq->flags & WQ_UNBOUND))
		pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
	else
4045
		pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
4046

4047
	ret = !list_empty(&pwq->delayed_works);
4048
	rcu_read_unlock_sched();
4049 4050

	return ret;
L
Linus Torvalds 已提交
4051
}
4052
EXPORT_SYMBOL_GPL(workqueue_congested);
L
Linus Torvalds 已提交
4053

4054 4055 4056 4057 4058 4059 4060 4061 4062 4063 4064 4065
/**
 * work_busy - test whether a work is currently pending or running
 * @work: the work to be tested
 *
 * Test whether @work is currently pending or running.  There is no
 * synchronization around this function and the test result is
 * unreliable and only useful as advisory hints or for debugging.
 *
 * RETURNS:
 * OR'd bitmask of WORK_BUSY_* bits.
 */
unsigned int work_busy(struct work_struct *work)
L
Linus Torvalds 已提交
4066
{
4067
	struct worker_pool *pool;
4068 4069
	unsigned long flags;
	unsigned int ret = 0;
L
Linus Torvalds 已提交
4070

4071 4072
	if (work_pending(work))
		ret |= WORK_BUSY_PENDING;
L
Linus Torvalds 已提交
4073

4074 4075
	local_irq_save(flags);
	pool = get_work_pool(work);
4076
	if (pool) {
4077
		spin_lock(&pool->lock);
4078 4079
		if (find_worker_executing_work(pool, work))
			ret |= WORK_BUSY_RUNNING;
4080
		spin_unlock(&pool->lock);
4081
	}
4082
	local_irq_restore(flags);
L
Linus Torvalds 已提交
4083

4084
	return ret;
L
Linus Torvalds 已提交
4085
}
4086
EXPORT_SYMBOL_GPL(work_busy);
L
Linus Torvalds 已提交
4087

4088 4089 4090
/*
 * CPU hotplug.
 *
4091
 * There are two challenges in supporting CPU hotplug.  Firstly, there
4092
 * are a lot of assumptions on strong associations among work, pwq and
4093
 * pool which make migrating pending and scheduled works very
4094
 * difficult to implement without impacting hot paths.  Secondly,
4095
 * worker pools serve mix of short, long and very long running works making
4096 4097
 * blocked draining impractical.
 *
4098
 * This is solved by allowing the pools to be disassociated from the CPU
4099 4100
 * running as an unbound one and allowing it to be reattached later if the
 * cpu comes back online.
4101
 */
L
Linus Torvalds 已提交
4102

4103
static void wq_unbind_fn(struct work_struct *work)
4104
{
4105
	int cpu = smp_processor_id();
4106
	struct worker_pool *pool;
4107
	struct worker *worker;
4108
	int wi;
4109

4110
	for_each_cpu_worker_pool(pool, cpu) {
4111
		WARN_ON_ONCE(cpu != smp_processor_id());
4112

4113
		mutex_lock(&pool->manager_mutex);
4114
		spin_lock_irq(&pool->lock);
4115

4116
		/*
4117
		 * We've blocked all manager operations.  Make all workers
4118 4119 4120 4121 4122
		 * unbound and set DISASSOCIATED.  Before this, all workers
		 * except for the ones which are still executing works from
		 * before the last CPU down must be on the cpu.  After
		 * this, they may become diasporas.
		 */
4123
		for_each_pool_worker(worker, wi, pool)
4124
			worker->flags |= WORKER_UNBOUND;
4125

4126
		pool->flags |= POOL_DISASSOCIATED;
4127

4128
		spin_unlock_irq(&pool->lock);
4129
		mutex_unlock(&pool->manager_mutex);
4130
	}
4131

4132
	/*
4133
	 * Call schedule() so that we cross rq->lock and thus can guarantee
4134 4135
	 * sched callbacks see the %WORKER_UNBOUND flag.  This is necessary
	 * as scheduler callbacks may be invoked from other cpus.
4136 4137
	 */
	schedule();
4138

4139
	/*
4140 4141
	 * Sched callbacks are disabled now.  Zap nr_running.  After this,
	 * nr_running stays zero and need_more_worker() and keep_working()
4142 4143 4144
	 * are always true as long as the worklist is not empty.  Pools on
	 * @cpu now behave as unbound (in terms of concurrency management)
	 * pools which are served by workers tied to the CPU.
4145 4146 4147 4148
	 *
	 * On return from this function, the current worker would trigger
	 * unbound chain execution of pending work items if other workers
	 * didn't already.
4149
	 */
4150
	for_each_cpu_worker_pool(pool, cpu)
4151
		atomic_set(&pool->nr_running, 0);
4152 4153
}

T
Tejun Heo 已提交
4154 4155 4156 4157
/**
 * rebind_workers - rebind all workers of a pool to the associated CPU
 * @pool: pool of interest
 *
4158
 * @pool->cpu is coming online.  Rebind all workers to the CPU.
T
Tejun Heo 已提交
4159 4160 4161
 */
static void rebind_workers(struct worker_pool *pool)
{
4162 4163
	struct worker *worker;
	int wi;
T
Tejun Heo 已提交
4164 4165 4166

	lockdep_assert_held(&pool->manager_mutex);

4167 4168 4169 4170 4171 4172 4173 4174 4175 4176
	/*
	 * Restore CPU affinity of all workers.  As all idle workers should
	 * be on the run-queue of the associated CPU before any local
	 * wake-ups for concurrency management happen, restore CPU affinty
	 * of all workers first and then clear UNBOUND.  As we're called
	 * from CPU_ONLINE, the following shouldn't fail.
	 */
	for_each_pool_worker(worker, wi, pool)
		WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task,
						  pool->attrs->cpumask) < 0);
T
Tejun Heo 已提交
4177

4178
	spin_lock_irq(&pool->lock);
T
Tejun Heo 已提交
4179

4180 4181
	for_each_pool_worker(worker, wi, pool) {
		unsigned int worker_flags = worker->flags;
T
Tejun Heo 已提交
4182 4183

		/*
4184 4185 4186 4187 4188 4189
		 * A bound idle worker should actually be on the runqueue
		 * of the associated CPU for local wake-ups targeting it to
		 * work.  Kick all idle workers so that they migrate to the
		 * associated CPU.  Doing this in the same loop as
		 * replacing UNBOUND with REBOUND is safe as no worker will
		 * be bound before @pool->lock is released.
T
Tejun Heo 已提交
4190
		 */
4191 4192
		if (worker_flags & WORKER_IDLE)
			wake_up_process(worker->task);
T
Tejun Heo 已提交
4193

4194 4195 4196 4197 4198 4199 4200 4201 4202 4203 4204 4205 4206 4207 4208 4209 4210 4211 4212
		/*
		 * We want to clear UNBOUND but can't directly call
		 * worker_clr_flags() or adjust nr_running.  Atomically
		 * replace UNBOUND with another NOT_RUNNING flag REBOUND.
		 * @worker will clear REBOUND using worker_clr_flags() when
		 * it initiates the next execution cycle thus restoring
		 * concurrency management.  Note that when or whether
		 * @worker clears REBOUND doesn't affect correctness.
		 *
		 * ACCESS_ONCE() is necessary because @worker->flags may be
		 * tested without holding any lock in
		 * wq_worker_waking_up().  Without it, NOT_RUNNING test may
		 * fail incorrectly leading to premature concurrency
		 * management operations.
		 */
		WARN_ON_ONCE(!(worker_flags & WORKER_UNBOUND));
		worker_flags |= WORKER_REBOUND;
		worker_flags &= ~WORKER_UNBOUND;
		ACCESS_ONCE(worker->flags) = worker_flags;
T
Tejun Heo 已提交
4213
	}
4214 4215

	spin_unlock_irq(&pool->lock);
T
Tejun Heo 已提交
4216 4217
}

4218 4219 4220 4221 4222 4223 4224 4225 4226 4227 4228 4229 4230 4231 4232 4233 4234 4235 4236 4237 4238 4239 4240 4241 4242 4243 4244 4245 4246 4247 4248 4249 4250
/**
 * restore_unbound_workers_cpumask - restore cpumask of unbound workers
 * @pool: unbound pool of interest
 * @cpu: the CPU which is coming up
 *
 * An unbound pool may end up with a cpumask which doesn't have any online
 * CPUs.  When a worker of such pool get scheduled, the scheduler resets
 * its cpus_allowed.  If @cpu is in @pool's cpumask which didn't have any
 * online CPU before, cpus_allowed of all its workers should be restored.
 */
static void restore_unbound_workers_cpumask(struct worker_pool *pool, int cpu)
{
	static cpumask_t cpumask;
	struct worker *worker;
	int wi;

	lockdep_assert_held(&pool->manager_mutex);

	/* is @cpu allowed for @pool? */
	if (!cpumask_test_cpu(cpu, pool->attrs->cpumask))
		return;

	/* is @cpu the only online CPU? */
	cpumask_and(&cpumask, pool->attrs->cpumask, cpu_online_mask);
	if (cpumask_weight(&cpumask) != 1)
		return;

	/* as we're called from CPU_ONLINE, the following shouldn't fail */
	for_each_pool_worker(worker, wi, pool)
		WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task,
						  pool->attrs->cpumask) < 0);
}

T
Tejun Heo 已提交
4251 4252 4253 4254
/*
 * Workqueues should be brought up before normal priority CPU notifiers.
 * This will be registered high priority CPU notifier.
 */
4255
static int __cpuinit workqueue_cpu_up_callback(struct notifier_block *nfb,
T
Tejun Heo 已提交
4256 4257
					       unsigned long action,
					       void *hcpu)
4258
{
4259
	int cpu = (unsigned long)hcpu;
4260
	struct worker_pool *pool;
4261
	int pi;
4262

T
Tejun Heo 已提交
4263
	switch (action & ~CPU_TASKS_FROZEN) {
4264
	case CPU_UP_PREPARE:
4265
		for_each_cpu_worker_pool(pool, cpu) {
4266 4267
			if (pool->nr_workers)
				continue;
4268
			if (create_and_start_worker(pool) < 0)
4269
				return NOTIFY_BAD;
4270
		}
T
Tejun Heo 已提交
4271
		break;
4272

4273 4274
	case CPU_DOWN_FAILED:
	case CPU_ONLINE:
4275
		mutex_lock(&wq_pool_mutex);
4276 4277

		for_each_pool(pool, pi) {
4278
			mutex_lock(&pool->manager_mutex);
4279

4280 4281 4282 4283
			if (pool->cpu == cpu) {
				spin_lock_irq(&pool->lock);
				pool->flags &= ~POOL_DISASSOCIATED;
				spin_unlock_irq(&pool->lock);
4284

4285 4286 4287 4288
				rebind_workers(pool);
			} else if (pool->cpu < 0) {
				restore_unbound_workers_cpumask(pool, cpu);
			}
4289

4290
			mutex_unlock(&pool->manager_mutex);
4291
		}
4292

4293
		mutex_unlock(&wq_pool_mutex);
4294
		break;
4295
	}
4296 4297 4298 4299 4300 4301 4302
	return NOTIFY_OK;
}

/*
 * Workqueues should be brought down after normal priority CPU notifiers.
 * This will be registered as low priority CPU notifier.
 */
4303
static int __cpuinit workqueue_cpu_down_callback(struct notifier_block *nfb,
4304 4305 4306
						 unsigned long action,
						 void *hcpu)
{
4307
	int cpu = (unsigned long)hcpu;
T
Tejun Heo 已提交
4308 4309
	struct work_struct unbind_work;

4310 4311
	switch (action & ~CPU_TASKS_FROZEN) {
	case CPU_DOWN_PREPARE:
T
Tejun Heo 已提交
4312
		/* unbinding should happen on the local CPU */
4313
		INIT_WORK_ONSTACK(&unbind_work, wq_unbind_fn);
4314
		queue_work_on(cpu, system_highpri_wq, &unbind_work);
T
Tejun Heo 已提交
4315 4316
		flush_work(&unbind_work);
		break;
4317 4318 4319 4320
	}
	return NOTIFY_OK;
}

4321
#ifdef CONFIG_SMP
4322

4323
struct work_for_cpu {
4324
	struct work_struct work;
4325 4326 4327 4328 4329
	long (*fn)(void *);
	void *arg;
	long ret;
};

4330
static void work_for_cpu_fn(struct work_struct *work)
4331
{
4332 4333
	struct work_for_cpu *wfc = container_of(work, struct work_for_cpu, work);

4334 4335 4336 4337 4338 4339 4340 4341 4342
	wfc->ret = wfc->fn(wfc->arg);
}

/**
 * work_on_cpu - run a function in user context on a particular cpu
 * @cpu: the cpu to run on
 * @fn: the function to run
 * @arg: the function arg
 *
4343 4344
 * This will return the value @fn returns.
 * It is up to the caller to ensure that the cpu doesn't go offline.
4345
 * The caller must not hold any locks which would prevent @fn from completing.
4346
 */
4347
long work_on_cpu(int cpu, long (*fn)(void *), void *arg)
4348
{
4349
	struct work_for_cpu wfc = { .fn = fn, .arg = arg };
4350

4351 4352 4353
	INIT_WORK_ONSTACK(&wfc.work, work_for_cpu_fn);
	schedule_work_on(cpu, &wfc.work);
	flush_work(&wfc.work);
4354 4355 4356 4357 4358
	return wfc.ret;
}
EXPORT_SYMBOL_GPL(work_on_cpu);
#endif /* CONFIG_SMP */

4359 4360 4361 4362 4363
#ifdef CONFIG_FREEZER

/**
 * freeze_workqueues_begin - begin freezing workqueues
 *
4364
 * Start freezing workqueues.  After this function returns, all freezable
4365
 * workqueues will queue new works to their delayed_works list instead of
4366
 * pool->worklist.
4367 4368
 *
 * CONTEXT:
4369
 * Grabs and releases wq_pool_mutex, wq->mutex and pool->lock's.
4370 4371 4372
 */
void freeze_workqueues_begin(void)
{
T
Tejun Heo 已提交
4373
	struct worker_pool *pool;
4374 4375
	struct workqueue_struct *wq;
	struct pool_workqueue *pwq;
4376
	int pi;
4377

4378
	mutex_lock(&wq_pool_mutex);
4379

4380
	WARN_ON_ONCE(workqueue_freezing);
4381 4382
	workqueue_freezing = true;

4383
	/* set FREEZING */
4384
	for_each_pool(pool, pi) {
4385
		spin_lock_irq(&pool->lock);
T
Tejun Heo 已提交
4386 4387
		WARN_ON_ONCE(pool->flags & POOL_FREEZING);
		pool->flags |= POOL_FREEZING;
4388
		spin_unlock_irq(&pool->lock);
4389
	}
4390

4391
	list_for_each_entry(wq, &workqueues, list) {
4392
		mutex_lock(&wq->mutex);
4393 4394
		for_each_pwq(pwq, wq)
			pwq_adjust_max_active(pwq);
4395
		mutex_unlock(&wq->mutex);
4396
	}
4397

4398
	mutex_unlock(&wq_pool_mutex);
4399 4400 4401
}

/**
4402
 * freeze_workqueues_busy - are freezable workqueues still busy?
4403 4404 4405 4406 4407
 *
 * Check whether freezing is complete.  This function must be called
 * between freeze_workqueues_begin() and thaw_workqueues().
 *
 * CONTEXT:
4408
 * Grabs and releases wq_pool_mutex.
4409 4410
 *
 * RETURNS:
4411 4412
 * %true if some freezable workqueues are still busy.  %false if freezing
 * is complete.
4413 4414 4415 4416
 */
bool freeze_workqueues_busy(void)
{
	bool busy = false;
4417 4418
	struct workqueue_struct *wq;
	struct pool_workqueue *pwq;
4419

4420
	mutex_lock(&wq_pool_mutex);
4421

4422
	WARN_ON_ONCE(!workqueue_freezing);
4423

4424 4425 4426
	list_for_each_entry(wq, &workqueues, list) {
		if (!(wq->flags & WQ_FREEZABLE))
			continue;
4427 4428 4429 4430
		/*
		 * nr_active is monotonically decreasing.  It's safe
		 * to peek without lock.
		 */
4431
		rcu_read_lock_sched();
4432
		for_each_pwq(pwq, wq) {
4433
			WARN_ON_ONCE(pwq->nr_active < 0);
4434
			if (pwq->nr_active) {
4435
				busy = true;
4436
				rcu_read_unlock_sched();
4437 4438 4439
				goto out_unlock;
			}
		}
4440
		rcu_read_unlock_sched();
4441 4442
	}
out_unlock:
4443
	mutex_unlock(&wq_pool_mutex);
4444 4445 4446 4447 4448 4449 4450
	return busy;
}

/**
 * thaw_workqueues - thaw workqueues
 *
 * Thaw workqueues.  Normal queueing is restored and all collected
4451
 * frozen works are transferred to their respective pool worklists.
4452 4453
 *
 * CONTEXT:
4454
 * Grabs and releases wq_pool_mutex, wq->mutex and pool->lock's.
4455 4456 4457
 */
void thaw_workqueues(void)
{
4458 4459 4460
	struct workqueue_struct *wq;
	struct pool_workqueue *pwq;
	struct worker_pool *pool;
4461
	int pi;
4462

4463
	mutex_lock(&wq_pool_mutex);
4464 4465 4466 4467

	if (!workqueue_freezing)
		goto out_unlock;

4468
	/* clear FREEZING */
4469
	for_each_pool(pool, pi) {
4470
		spin_lock_irq(&pool->lock);
4471 4472
		WARN_ON_ONCE(!(pool->flags & POOL_FREEZING));
		pool->flags &= ~POOL_FREEZING;
4473
		spin_unlock_irq(&pool->lock);
4474
	}
4475

4476 4477
	/* restore max_active and repopulate worklist */
	list_for_each_entry(wq, &workqueues, list) {
4478
		mutex_lock(&wq->mutex);
4479 4480
		for_each_pwq(pwq, wq)
			pwq_adjust_max_active(pwq);
4481
		mutex_unlock(&wq->mutex);
4482 4483 4484 4485
	}

	workqueue_freezing = false;
out_unlock:
4486
	mutex_unlock(&wq_pool_mutex);
4487 4488 4489
}
#endif /* CONFIG_FREEZER */

4490 4491 4492 4493 4494 4495 4496 4497 4498 4499 4500 4501 4502 4503 4504 4505 4506 4507 4508 4509 4510 4511 4512 4513 4514 4515 4516 4517 4518 4519 4520 4521 4522 4523 4524 4525 4526
static void __init wq_numa_init(void)
{
	cpumask_var_t *tbl;
	int node, cpu;

	/* determine NUMA pwq table len - highest node id + 1 */
	for_each_node(node)
		wq_numa_tbl_len = max(wq_numa_tbl_len, node + 1);

	if (num_possible_nodes() <= 1)
		return;

	/*
	 * We want masks of possible CPUs of each node which isn't readily
	 * available.  Build one from cpu_to_node() which should have been
	 * fully initialized by now.
	 */
	tbl = kzalloc(wq_numa_tbl_len * sizeof(tbl[0]), GFP_KERNEL);
	BUG_ON(!tbl);

	for_each_node(node)
		BUG_ON(!alloc_cpumask_var_node(&tbl[node], GFP_KERNEL, node));

	for_each_possible_cpu(cpu) {
		node = cpu_to_node(cpu);
		if (WARN_ON(node == NUMA_NO_NODE)) {
			pr_warn("workqueue: NUMA node mapping not available for cpu%d, disabling NUMA support\n", cpu);
			/* happens iff arch is bonkers, let's just proceed */
			return;
		}
		cpumask_set_cpu(cpu, tbl[node]);
	}

	wq_numa_possible_cpumask = tbl;
	wq_numa_enabled = true;
}

4527
static int __init init_workqueues(void)
L
Linus Torvalds 已提交
4528
{
T
Tejun Heo 已提交
4529 4530
	int std_nice[NR_STD_WORKER_POOLS] = { 0, HIGHPRI_NICE_LEVEL };
	int i, cpu;
T
Tejun Heo 已提交
4531

4532 4533
	/* make sure we have enough bits for OFFQ pool ID */
	BUILD_BUG_ON((1LU << (BITS_PER_LONG - WORK_OFFQ_POOL_SHIFT)) <
4534
		     WORK_CPU_END * NR_STD_WORKER_POOLS);
4535

4536 4537 4538 4539
	WARN_ON(__alignof__(struct pool_workqueue) < __alignof__(long long));

	pwq_cache = KMEM_CACHE(pool_workqueue, SLAB_PANIC);

4540
	cpu_notifier(workqueue_cpu_up_callback, CPU_PRI_WORKQUEUE_UP);
4541
	hotcpu_notifier(workqueue_cpu_down_callback, CPU_PRI_WORKQUEUE_DOWN);
4542

4543 4544
	wq_numa_init();

4545
	/* initialize CPU pools */
4546
	for_each_possible_cpu(cpu) {
4547
		struct worker_pool *pool;
4548

T
Tejun Heo 已提交
4549
		i = 0;
4550
		for_each_cpu_worker_pool(pool, cpu) {
T
Tejun Heo 已提交
4551
			BUG_ON(init_worker_pool(pool));
4552
			pool->cpu = cpu;
4553
			cpumask_copy(pool->attrs->cpumask, cpumask_of(cpu));
T
Tejun Heo 已提交
4554
			pool->attrs->nice = std_nice[i++];
4555
			pool->node = cpu_to_node(cpu);
T
Tejun Heo 已提交
4556

T
Tejun Heo 已提交
4557
			/* alloc pool ID */
4558
			mutex_lock(&wq_pool_mutex);
T
Tejun Heo 已提交
4559
			BUG_ON(worker_pool_assign_id(pool));
4560
			mutex_unlock(&wq_pool_mutex);
4561
		}
4562 4563
	}

4564
	/* create the initial worker */
4565
	for_each_online_cpu(cpu) {
4566
		struct worker_pool *pool;
4567

4568
		for_each_cpu_worker_pool(pool, cpu) {
4569
			pool->flags &= ~POOL_DISASSOCIATED;
4570
			BUG_ON(create_and_start_worker(pool) < 0);
4571
		}
4572 4573
	}

4574 4575 4576 4577 4578 4579 4580 4581 4582
	/* create default unbound wq attrs */
	for (i = 0; i < NR_STD_WORKER_POOLS; i++) {
		struct workqueue_attrs *attrs;

		BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
		attrs->nice = std_nice[i];
		unbound_std_wq_attrs[i] = attrs;
	}

4583
	system_wq = alloc_workqueue("events", 0, 0);
4584
	system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
4585
	system_long_wq = alloc_workqueue("events_long", 0, 0);
4586 4587
	system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
					    WQ_UNBOUND_MAX_ACTIVE);
4588 4589
	system_freezable_wq = alloc_workqueue("events_freezable",
					      WQ_FREEZABLE, 0);
4590
	BUG_ON(!system_wq || !system_highpri_wq || !system_long_wq ||
4591
	       !system_unbound_wq || !system_freezable_wq);
4592
	return 0;
L
Linus Torvalds 已提交
4593
}
4594
early_initcall(init_workqueues);