io-wq.c 27.2 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
// SPDX-License-Identifier: GPL-2.0
/*
 * Basic worker thread pool for io_uring
 *
 * Copyright (C) 2019 Jens Axboe
 *
 */
#include <linux/kernel.h>
#include <linux/init.h>
#include <linux/errno.h>
#include <linux/sched/signal.h>
#include <linux/mm.h>
#include <linux/mmu_context.h>
#include <linux/sched/mm.h>
#include <linux/percpu.h>
#include <linux/slab.h>
#include <linux/kthread.h>
#include <linux/rculist_nulls.h>
19
#include <linux/uaccess.h>
20 21 22 23 24 25 26 27 28 29 30

#include "io-wq.h"

#define WORKER_IDLE_TIMEOUT	(5 * HZ)

enum {
	IO_WORKER_F_UP		= 1,	/* up and active */
	IO_WORKER_F_RUNNING	= 2,	/* account as running */
	IO_WORKER_F_FREE	= 4,	/* worker on free list */
	IO_WORKER_F_EXITING	= 8,	/* worker exiting */
	IO_WORKER_F_FIXED	= 16,	/* static idle worker */
31
	IO_WORKER_F_BOUND	= 32,	/* is doing bounded work */
32 33 34 35 36
};

enum {
	IO_WQ_BIT_EXIT		= 0,	/* wq exiting */
	IO_WQ_BIT_CANCEL	= 1,	/* cancel work on list */
37
	IO_WQ_BIT_ERROR		= 2,	/* error on setup */
38 39 40 41 42 43 44 45 46 47 48 49 50
};

enum {
	IO_WQE_FLAG_STALLED	= 1,	/* stalled on hash */
};

/*
 * One for each thread in a wqe pool
 */
struct io_worker {
	refcount_t ref;
	unsigned flags;
	struct hlist_nulls_node nulls_node;
51
	struct list_head all_list;
52 53
	struct task_struct *task;
	struct io_wqe *wqe;
54

55
	struct io_wq_work *cur_work;
56
	spinlock_t lock;
57 58 59

	struct rcu_head rcu;
	struct mm_struct *mm;
60 61
	const struct cred *cur_creds;
	const struct cred *saved_creds;
62
	struct files_struct *restore_files;
63 64 65 66 67 68 69 70
};

#if BITS_PER_LONG == 64
#define IO_WQ_HASH_ORDER	6
#else
#define IO_WQ_HASH_ORDER	5
#endif

71 72 73 74 75 76 77 78 79 80 81
struct io_wqe_acct {
	unsigned nr_workers;
	unsigned max_workers;
	atomic_t nr_running;
};

enum {
	IO_WQ_ACCT_BOUND,
	IO_WQ_ACCT_UNBOUND,
};

82 83 84 85 86 87
/*
 * Per-node worker thread pool
 */
struct io_wqe {
	struct {
		spinlock_t lock;
J
Jens Axboe 已提交
88
		struct io_wq_work_list work_list;
89 90 91 92 93
		unsigned long hash_map;
		unsigned flags;
	} ____cacheline_aligned_in_smp;

	int node;
94
	struct io_wqe_acct acct[2];
95

96
	struct hlist_nulls_head free_list;
97
	struct list_head all_list;
98 99 100 101 102 103 104 105 106 107 108

	struct io_wq *wq;
};

/*
 * Per io_wq state
  */
struct io_wq {
	struct io_wqe **wqes;
	unsigned long state;

109 110 111
	get_work_fn *get_work;
	put_work_fn *put_work;

112
	struct task_struct *manager;
113
	struct user_struct *user;
114 115
	refcount_t refs;
	struct completion done;
J
Jens Axboe 已提交
116 117

	refcount_t use_refs;
118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137
};

static bool io_worker_get(struct io_worker *worker)
{
	return refcount_inc_not_zero(&worker->ref);
}

static void io_worker_release(struct io_worker *worker)
{
	if (refcount_dec_and_test(&worker->ref))
		wake_up_process(worker->task);
}

/*
 * Note: drops the wqe->lock if returning true! The caller must re-acquire
 * the lock in that case. Some callers need to restart handling if this
 * happens, so we can't just re-acquire the lock on behalf of the caller.
 */
static bool __io_worker_unuse(struct io_wqe *wqe, struct io_worker *worker)
{
138 139
	bool dropped_lock = false;

140 141 142
	if (worker->saved_creds) {
		revert_creds(worker->saved_creds);
		worker->cur_creds = worker->saved_creds = NULL;
143 144
	}

145 146 147 148 149 150 151 152 153 154
	if (current->files != worker->restore_files) {
		__acquire(&wqe->lock);
		spin_unlock_irq(&wqe->lock);
		dropped_lock = true;

		task_lock(current);
		current->files = worker->restore_files;
		task_unlock(current);
	}

155 156 157 158 159
	/*
	 * If we have an active mm, we need to drop the wq lock before unusing
	 * it. If we do, return true and let the caller retry the idle loop.
	 */
	if (worker->mm) {
160 161 162 163 164
		if (!dropped_lock) {
			__acquire(&wqe->lock);
			spin_unlock_irq(&wqe->lock);
			dropped_lock = true;
		}
165 166 167 168 169 170 171
		__set_current_state(TASK_RUNNING);
		set_fs(KERNEL_DS);
		unuse_mm(worker->mm);
		mmput(worker->mm);
		worker->mm = NULL;
	}

172
	return dropped_lock;
173 174
}

175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192
static inline struct io_wqe_acct *io_work_get_acct(struct io_wqe *wqe,
						   struct io_wq_work *work)
{
	if (work->flags & IO_WQ_WORK_UNBOUND)
		return &wqe->acct[IO_WQ_ACCT_UNBOUND];

	return &wqe->acct[IO_WQ_ACCT_BOUND];
}

static inline struct io_wqe_acct *io_wqe_get_acct(struct io_wqe *wqe,
						  struct io_worker *worker)
{
	if (worker->flags & IO_WORKER_F_BOUND)
		return &wqe->acct[IO_WQ_ACCT_BOUND];

	return &wqe->acct[IO_WQ_ACCT_UNBOUND];
}

193 194 195
static void io_worker_exit(struct io_worker *worker)
{
	struct io_wqe *wqe = worker->wqe;
196 197
	struct io_wqe_acct *acct = io_wqe_get_acct(wqe, worker);
	unsigned nr_workers;
198 199 200 201 202 203 204 205 206 207 208 209 210

	/*
	 * If we're not at zero, someone else is holding a brief reference
	 * to the worker. Wait for that to go away.
	 */
	set_current_state(TASK_INTERRUPTIBLE);
	if (!refcount_dec_and_test(&worker->ref))
		schedule();
	__set_current_state(TASK_RUNNING);

	preempt_disable();
	current->flags &= ~PF_IO_WORKER;
	if (worker->flags & IO_WORKER_F_RUNNING)
211 212 213
		atomic_dec(&acct->nr_running);
	if (!(worker->flags & IO_WORKER_F_BOUND))
		atomic_dec(&wqe->wq->user->processes);
214 215 216 217 218
	worker->flags = 0;
	preempt_enable();

	spin_lock_irq(&wqe->lock);
	hlist_nulls_del_rcu(&worker->nulls_node);
219
	list_del_rcu(&worker->all_list);
220 221 222 223
	if (__io_worker_unuse(wqe, worker)) {
		__release(&wqe->lock);
		spin_lock_irq(&wqe->lock);
	}
224 225 226
	acct->nr_workers--;
	nr_workers = wqe->acct[IO_WQ_ACCT_BOUND].nr_workers +
			wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers;
227 228 229
	spin_unlock_irq(&wqe->lock);

	/* all workers gone, wq exit can proceed */
230
	if (!nr_workers && refcount_dec_and_test(&wqe->wq->refs))
231 232
		complete(&wqe->wq->done);

233
	kfree_rcu(worker, rcu);
234 235
}

236 237 238
static inline bool io_wqe_run_queue(struct io_wqe *wqe)
	__must_hold(wqe->lock)
{
J
Jens Axboe 已提交
239 240
	if (!wq_list_empty(&wqe->work_list) &&
	    !(wqe->flags & IO_WQE_FLAG_STALLED))
241 242 243 244 245 246 247 248 249 250 251 252 253 254
		return true;
	return false;
}

/*
 * Check head of free list for an available worker. If one isn't available,
 * caller must wake up the wq manager to create one.
 */
static bool io_wqe_activate_free_worker(struct io_wqe *wqe)
	__must_hold(RCU)
{
	struct hlist_nulls_node *n;
	struct io_worker *worker;

255
	n = rcu_dereference(hlist_nulls_first_rcu(&wqe->free_list));
256 257 258 259 260
	if (is_a_nulls(n))
		return false;

	worker = hlist_nulls_entry(n, struct io_worker, nulls_node);
	if (io_worker_get(worker)) {
261
		wake_up_process(worker->task);
262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306
		io_worker_release(worker);
		return true;
	}

	return false;
}

/*
 * We need a worker. If we find a free one, we're good. If not, and we're
 * below the max number of workers, wake up the manager to create one.
 */
static void io_wqe_wake_worker(struct io_wqe *wqe, struct io_wqe_acct *acct)
{
	bool ret;

	/*
	 * Most likely an attempt to queue unbounded work on an io_wq that
	 * wasn't setup with any unbounded workers.
	 */
	WARN_ON_ONCE(!acct->max_workers);

	rcu_read_lock();
	ret = io_wqe_activate_free_worker(wqe);
	rcu_read_unlock();

	if (!ret && acct->nr_workers < acct->max_workers)
		wake_up_process(wqe->wq->manager);
}

static void io_wqe_inc_running(struct io_wqe *wqe, struct io_worker *worker)
{
	struct io_wqe_acct *acct = io_wqe_get_acct(wqe, worker);

	atomic_inc(&acct->nr_running);
}

static void io_wqe_dec_running(struct io_wqe *wqe, struct io_worker *worker)
	__must_hold(wqe->lock)
{
	struct io_wqe_acct *acct = io_wqe_get_acct(wqe, worker);

	if (atomic_dec_and_test(&acct->nr_running) && io_wqe_run_queue(wqe))
		io_wqe_wake_worker(wqe, acct);
}

307 308 309 310 311 312 313
static void io_worker_start(struct io_wqe *wqe, struct io_worker *worker)
{
	allow_kernel_signal(SIGINT);

	current->flags |= PF_IO_WORKER;

	worker->flags |= (IO_WORKER_F_UP | IO_WORKER_F_RUNNING);
314
	worker->restore_files = current->files;
315
	io_wqe_inc_running(wqe, worker);
316 317 318 319 320 321 322 323 324 325
}

/*
 * Worker will start processing some work. Move it to the busy list, if
 * it's currently on the freelist
 */
static void __io_worker_busy(struct io_wqe *wqe, struct io_worker *worker,
			     struct io_wq_work *work)
	__must_hold(wqe->lock)
{
326 327
	bool worker_bound, work_bound;

328 329 330 331
	if (worker->flags & IO_WORKER_F_FREE) {
		worker->flags &= ~IO_WORKER_F_FREE;
		hlist_nulls_del_init_rcu(&worker->nulls_node);
	}
332 333 334 335 336

	/*
	 * If worker is moving from bound to unbound (or vice versa), then
	 * ensure we update the running accounting.
	 */
337 338 339
	worker_bound = (worker->flags & IO_WORKER_F_BOUND) != 0;
	work_bound = (work->flags & IO_WQ_WORK_UNBOUND) == 0;
	if (worker_bound != work_bound) {
340 341 342 343 344 345 346 347 348 349 350 351 352 353
		io_wqe_dec_running(wqe, worker);
		if (work_bound) {
			worker->flags |= IO_WORKER_F_BOUND;
			wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers--;
			wqe->acct[IO_WQ_ACCT_BOUND].nr_workers++;
			atomic_dec(&wqe->wq->user->processes);
		} else {
			worker->flags &= ~IO_WORKER_F_BOUND;
			wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers++;
			wqe->acct[IO_WQ_ACCT_BOUND].nr_workers--;
			atomic_inc(&wqe->wq->user->processes);
		}
		io_wqe_inc_running(wqe, worker);
	 }
354 355 356 357 358 359 360 361 362 363 364 365 366 367
}

/*
 * No work, worker going to sleep. Move to freelist, and unuse mm if we
 * have one attached. Dropping the mm may potentially sleep, so we drop
 * the lock in that case and return success. Since the caller has to
 * retry the loop in that case (we changed task state), we don't regrab
 * the lock if we return success.
 */
static bool __io_worker_idle(struct io_wqe *wqe, struct io_worker *worker)
	__must_hold(wqe->lock)
{
	if (!(worker->flags & IO_WORKER_F_FREE)) {
		worker->flags |= IO_WORKER_F_FREE;
368
		hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);
369 370 371 372 373 374 375 376
	}

	return __io_worker_unuse(wqe, worker);
}

static struct io_wq_work *io_get_next_work(struct io_wqe *wqe, unsigned *hash)
	__must_hold(wqe->lock)
{
J
Jens Axboe 已提交
377
	struct io_wq_work_node *node, *prev;
378 379
	struct io_wq_work *work;

J
Jens Axboe 已提交
380 381 382
	wq_list_for_each(node, prev, &wqe->work_list) {
		work = container_of(node, struct io_wq_work, list);

383 384
		/* not hashed, can run anytime */
		if (!(work->flags & IO_WQ_WORK_HASHED)) {
J
Jens Axboe 已提交
385
			wq_node_del(&wqe->work_list, node, prev);
386 387 388 389 390 391 392
			return work;
		}

		/* hashed, can run if not already running */
		*hash = work->flags >> IO_WQ_HASH_SHIFT;
		if (!(wqe->hash_map & BIT_ULL(*hash))) {
			wqe->hash_map |= BIT_ULL(*hash);
J
Jens Axboe 已提交
393
			wq_node_del(&wqe->work_list, node, prev);
394 395 396 397 398 399 400
			return work;
		}
	}

	return NULL;
}

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 426 427 428 429 430 431 432 433 434 435 436 437
static void io_wq_switch_mm(struct io_worker *worker, struct io_wq_work *work)
{
	if (worker->mm) {
		unuse_mm(worker->mm);
		mmput(worker->mm);
		worker->mm = NULL;
	}
	if (!work->mm) {
		set_fs(KERNEL_DS);
		return;
	}
	if (mmget_not_zero(work->mm)) {
		use_mm(work->mm);
		if (!worker->mm)
			set_fs(USER_DS);
		worker->mm = work->mm;
		/* hang on to this mm */
		work->mm = NULL;
		return;
	}

	/* failed grabbing mm, ensure work gets cancelled */
	work->flags |= IO_WQ_WORK_CANCEL;
}

static void io_wq_switch_creds(struct io_worker *worker,
			       struct io_wq_work *work)
{
	const struct cred *old_creds = override_creds(work->creds);

	worker->cur_creds = work->creds;
	if (worker->saved_creds)
		put_cred(old_creds); /* creds set by previous switch */
	else
		worker->saved_creds = old_creds;
}

438 439 440
static void io_worker_handle_work(struct io_worker *worker)
	__releases(wqe->lock)
{
441
	struct io_wq_work *work, *old_work = NULL, *put_work = NULL;
442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457
	struct io_wqe *wqe = worker->wqe;
	struct io_wq *wq = wqe->wq;

	do {
		unsigned hash = -1U;

		/*
		 * If we got some work, mark us as busy. If we didn't, but
		 * the list isn't empty, it means we stalled on hashed work.
		 * Mark us stalled so we don't keep looking for work when we
		 * can't make progress, any work completion or insertion will
		 * clear the stalled flag.
		 */
		work = io_get_next_work(wqe, &hash);
		if (work)
			__io_worker_busy(wqe, worker, work);
J
Jens Axboe 已提交
458
		else if (!wq_list_empty(&wqe->work_list))
459 460 461
			wqe->flags |= IO_WQE_FLAG_STALLED;

		spin_unlock_irq(&wqe->lock);
462 463
		if (put_work && wq->put_work)
			wq->put_work(old_work);
464 465 466
		if (!work)
			break;
next:
467 468 469 470
		/* flush any pending signals before assigning new work */
		if (signal_pending(current))
			flush_signals(current);

471 472
		cond_resched();

473 474 475 476
		spin_lock_irq(&worker->lock);
		worker->cur_work = work;
		spin_unlock_irq(&worker->lock);

477 478 479
		if (work->flags & IO_WQ_WORK_CB)
			work->func(&work);

480 481 482 483 484 485
		if ((work->flags & IO_WQ_WORK_NEEDS_FILES) &&
		    current->files != work->files) {
			task_lock(current);
			current->files = work->files;
			task_unlock(current);
		}
486 487 488 489
		if (work->mm != worker->mm)
			io_wq_switch_mm(worker, work);
		if (worker->cur_creds != work->creds)
			io_wq_switch_creds(worker, work);
490 491 492 493
		/*
		 * OK to set IO_WQ_WORK_CANCEL even for uncancellable work,
		 * the worker function will do the right thing.
		 */
494 495 496 497 498
		if (test_bit(IO_WQ_BIT_CANCEL, &wq->state))
			work->flags |= IO_WQ_WORK_CANCEL;
		if (worker->mm)
			work->flags |= IO_WQ_WORK_HAS_MM;

499 500 501 502 503
		if (wq->get_work && !(work->flags & IO_WQ_WORK_INTERNAL)) {
			put_work = work;
			wq->get_work(work);
		}

504 505 506
		old_work = work;
		work->func(&work);

507
		spin_lock_irq(&worker->lock);
508
		worker->cur_work = NULL;
509 510 511 512
		spin_unlock_irq(&worker->lock);

		spin_lock_irq(&wqe->lock);

513 514 515 516 517 518
		if (hash != -1U) {
			wqe->hash_map &= ~BIT_ULL(hash);
			wqe->flags &= ~IO_WQE_FLAG_STALLED;
		}
		if (work && work != old_work) {
			spin_unlock_irq(&wqe->lock);
519 520 521 522 523 524

			if (put_work && wq->put_work) {
				wq->put_work(put_work);
				put_work = NULL;
			}

525 526 527 528 529 530 531
			/* dependent work not hashed */
			hash = -1U;
			goto next;
		}
	} while (1);
}

532 533 534 535 536 537 538 539 540 541 542 543 544
static inline void io_worker_spin_for_work(struct io_wqe *wqe)
{
	int i = 0;

	while (++i < 1000) {
		if (io_wqe_run_queue(wqe))
			break;
		if (need_resched())
			break;
		cpu_relax();
	}
}

545 546 547 548 549
static int io_wqe_worker(void *data)
{
	struct io_worker *worker = data;
	struct io_wqe *wqe = worker->wqe;
	struct io_wq *wq = wqe->wq;
550
	bool did_work;
551 552 553

	io_worker_start(wqe, worker);

554
	did_work = false;
555
	while (!test_bit(IO_WQ_BIT_EXIT, &wq->state)) {
556
		set_current_state(TASK_INTERRUPTIBLE);
557 558 559
loop:
		if (did_work)
			io_worker_spin_for_work(wqe);
560 561 562 563
		spin_lock_irq(&wqe->lock);
		if (io_wqe_run_queue(wqe)) {
			__set_current_state(TASK_RUNNING);
			io_worker_handle_work(worker);
564 565
			did_work = true;
			goto loop;
566
		}
567
		did_work = false;
568 569 570
		/* drops the lock on success, retry */
		if (__io_worker_idle(wqe, worker)) {
			__release(&wqe->lock);
571
			goto loop;
572 573 574 575 576 577 578 579 580 581 582 583 584 585
		}
		spin_unlock_irq(&wqe->lock);
		if (signal_pending(current))
			flush_signals(current);
		if (schedule_timeout(WORKER_IDLE_TIMEOUT))
			continue;
		/* timed out, exit unless we're the fixed worker */
		if (test_bit(IO_WQ_BIT_EXIT, &wq->state) ||
		    !(worker->flags & IO_WORKER_F_FIXED))
			break;
	}

	if (test_bit(IO_WQ_BIT_EXIT, &wq->state)) {
		spin_lock_irq(&wqe->lock);
J
Jens Axboe 已提交
586
		if (!wq_list_empty(&wqe->work_list))
587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608
			io_worker_handle_work(worker);
		else
			spin_unlock_irq(&wqe->lock);
	}

	io_worker_exit(worker);
	return 0;
}

/*
 * Called when a worker is scheduled in. Mark us as currently running.
 */
void io_wq_worker_running(struct task_struct *tsk)
{
	struct io_worker *worker = kthread_data(tsk);
	struct io_wqe *wqe = worker->wqe;

	if (!(worker->flags & IO_WORKER_F_UP))
		return;
	if (worker->flags & IO_WORKER_F_RUNNING)
		return;
	worker->flags |= IO_WORKER_F_RUNNING;
609
	io_wqe_inc_running(wqe, worker);
610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629
}

/*
 * Called when worker is going to sleep. If there are no workers currently
 * running and we have work pending, wake up a free one or have the manager
 * set one up.
 */
void io_wq_worker_sleeping(struct task_struct *tsk)
{
	struct io_worker *worker = kthread_data(tsk);
	struct io_wqe *wqe = worker->wqe;

	if (!(worker->flags & IO_WORKER_F_UP))
		return;
	if (!(worker->flags & IO_WORKER_F_RUNNING))
		return;

	worker->flags &= ~IO_WORKER_F_RUNNING;

	spin_lock_irq(&wqe->lock);
630
	io_wqe_dec_running(wqe, worker);
631 632 633
	spin_unlock_irq(&wqe->lock);
}

634
static bool create_io_worker(struct io_wq *wq, struct io_wqe *wqe, int index)
635
{
636
	struct io_wqe_acct *acct =&wqe->acct[index];
637 638
	struct io_worker *worker;

639
	worker = kzalloc_node(sizeof(*worker), GFP_KERNEL, wqe->node);
640
	if (!worker)
641
		return false;
642 643 644 645

	refcount_set(&worker->ref, 1);
	worker->nulls_node.pprev = NULL;
	worker->wqe = wqe;
646
	spin_lock_init(&worker->lock);
647 648

	worker->task = kthread_create_on_node(io_wqe_worker, worker, wqe->node,
649
				"io_wqe_worker-%d/%d", index, wqe->node);
650 651
	if (IS_ERR(worker->task)) {
		kfree(worker);
652
		return false;
653 654 655
	}

	spin_lock_irq(&wqe->lock);
656
	hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);
657
	list_add_tail_rcu(&worker->all_list, &wqe->all_list);
658
	worker->flags |= IO_WORKER_F_FREE;
659 660 661
	if (index == IO_WQ_ACCT_BOUND)
		worker->flags |= IO_WORKER_F_BOUND;
	if (!acct->nr_workers && (worker->flags & IO_WORKER_F_BOUND))
662
		worker->flags |= IO_WORKER_F_FIXED;
663
	acct->nr_workers++;
664 665
	spin_unlock_irq(&wqe->lock);

666 667 668
	if (index == IO_WQ_ACCT_UNBOUND)
		atomic_inc(&wq->user->processes);

669
	wake_up_process(worker->task);
670
	return true;
671 672
}

673
static inline bool io_wqe_need_worker(struct io_wqe *wqe, int index)
674 675
	__must_hold(wqe->lock)
{
676
	struct io_wqe_acct *acct = &wqe->acct[index];
677

678
	/* if we have available workers or no work, no need */
679
	if (!hlist_nulls_empty(&wqe->free_list) || !io_wqe_run_queue(wqe))
680 681
		return false;
	return acct->nr_workers < acct->max_workers;
682 683 684 685 686 687 688 689
}

/*
 * Manager thread. Tasked with creating new workers, if we need them.
 */
static int io_wq_manager(void *data)
{
	struct io_wq *wq = data;
J
Jann Horn 已提交
690 691
	int workers_to_create = num_possible_nodes();
	int node;
692

693
	/* create fixed workers */
J
Jann Horn 已提交
694 695 696 697 698
	refcount_set(&wq->refs, workers_to_create);
	for_each_node(node) {
		if (!create_io_worker(wq, wq->wqes[node], IO_WQ_ACCT_BOUND))
			goto err;
		workers_to_create--;
699
	}
700

701 702 703
	complete(&wq->done);

	while (!kthread_should_stop()) {
J
Jann Horn 已提交
704 705
		for_each_node(node) {
			struct io_wqe *wqe = wq->wqes[node];
706
			bool fork_worker[2] = { false, false };
707 708

			spin_lock_irq(&wqe->lock);
709 710 711 712
			if (io_wqe_need_worker(wqe, IO_WQ_ACCT_BOUND))
				fork_worker[IO_WQ_ACCT_BOUND] = true;
			if (io_wqe_need_worker(wqe, IO_WQ_ACCT_UNBOUND))
				fork_worker[IO_WQ_ACCT_UNBOUND] = true;
713
			spin_unlock_irq(&wqe->lock);
714 715 716 717
			if (fork_worker[IO_WQ_ACCT_BOUND])
				create_io_worker(wq, wqe, IO_WQ_ACCT_BOUND);
			if (fork_worker[IO_WQ_ACCT_UNBOUND])
				create_io_worker(wq, wqe, IO_WQ_ACCT_UNBOUND);
718 719 720 721 722 723
		}
		set_current_state(TASK_INTERRUPTIBLE);
		schedule_timeout(HZ);
	}

	return 0;
724 725 726
err:
	set_bit(IO_WQ_BIT_ERROR, &wq->state);
	set_bit(IO_WQ_BIT_EXIT, &wq->state);
J
Jann Horn 已提交
727
	if (refcount_sub_and_test(workers_to_create, &wq->refs))
728 729
		complete(&wq->done);
	return 0;
730 731
}

732 733 734 735 736 737 738 739 740 741 742
static bool io_wq_can_queue(struct io_wqe *wqe, struct io_wqe_acct *acct,
			    struct io_wq_work *work)
{
	bool free_worker;

	if (!(work->flags & IO_WQ_WORK_UNBOUND))
		return true;
	if (atomic_read(&acct->nr_running))
		return true;

	rcu_read_lock();
743
	free_worker = !hlist_nulls_empty(&wqe->free_list);
744 745 746 747 748 749 750 751 752 753 754
	rcu_read_unlock();
	if (free_worker)
		return true;

	if (atomic_read(&wqe->wq->user->processes) >= acct->max_workers &&
	    !(capable(CAP_SYS_RESOURCE) || capable(CAP_SYS_ADMIN)))
		return false;

	return true;
}

755 756
static void io_wqe_enqueue(struct io_wqe *wqe, struct io_wq_work *work)
{
757
	struct io_wqe_acct *acct = io_work_get_acct(wqe, work);
758
	int work_flags;
759 760
	unsigned long flags;

761 762 763 764 765 766 767 768 769 770 771 772
	/*
	 * Do early check to see if we need a new unbound worker, and if we do,
	 * if we're allowed to do so. This isn't 100% accurate as there's a
	 * gap between this check and incrementing the value, but that's OK.
	 * It's close enough to not be an issue, fork() has the same delay.
	 */
	if (unlikely(!io_wq_can_queue(wqe, acct, work))) {
		work->flags |= IO_WQ_WORK_CANCEL;
		work->func(&work);
		return;
	}

773
	work_flags = work->flags;
774
	spin_lock_irqsave(&wqe->lock, flags);
J
Jens Axboe 已提交
775
	wq_list_add_tail(&work->list, &wqe->work_list);
776 777 778
	wqe->flags &= ~IO_WQE_FLAG_STALLED;
	spin_unlock_irqrestore(&wqe->lock, flags);

779 780
	if ((work_flags & IO_WQ_WORK_CONCURRENT) ||
	    !atomic_read(&acct->nr_running))
781
		io_wqe_wake_worker(wqe, acct);
782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823
}

void io_wq_enqueue(struct io_wq *wq, struct io_wq_work *work)
{
	struct io_wqe *wqe = wq->wqes[numa_node_id()];

	io_wqe_enqueue(wqe, work);
}

/*
 * Enqueue work, hashed by some key. Work items that hash to the same value
 * will not be done in parallel. Used to limit concurrent writes, generally
 * hashed by inode.
 */
void io_wq_enqueue_hashed(struct io_wq *wq, struct io_wq_work *work, void *val)
{
	struct io_wqe *wqe = wq->wqes[numa_node_id()];
	unsigned bit;


	bit = hash_ptr(val, IO_WQ_HASH_ORDER);
	work->flags |= (IO_WQ_WORK_HASHED | (bit << IO_WQ_HASH_SHIFT));
	io_wqe_enqueue(wqe, work);
}

static bool io_wqe_worker_send_sig(struct io_worker *worker, void *data)
{
	send_sig(SIGINT, worker->task, 1);
	return false;
}

/*
 * Iterate the passed in list and call the specific function for each
 * worker that isn't exiting
 */
static bool io_wq_for_each_worker(struct io_wqe *wqe,
				  bool (*func)(struct io_worker *, void *),
				  void *data)
{
	struct io_worker *worker;
	bool ret = false;

824
	list_for_each_entry_rcu(worker, &wqe->all_list, all_list) {
825 826 827 828 829 830 831
		if (io_worker_get(worker)) {
			ret = func(worker, data);
			io_worker_release(worker);
			if (ret)
				break;
		}
	}
832

833 834 835 836 837
	return ret;
}

void io_wq_cancel_all(struct io_wq *wq)
{
J
Jann Horn 已提交
838
	int node;
839 840 841 842

	set_bit(IO_WQ_BIT_CANCEL, &wq->state);

	rcu_read_lock();
J
Jann Horn 已提交
843 844
	for_each_node(node) {
		struct io_wqe *wqe = wq->wqes[node];
845

846
		io_wq_for_each_worker(wqe, io_wqe_worker_send_sig, NULL);
847 848 849 850
	}
	rcu_read_unlock();
}

851 852 853 854 855 856 857 858 859
struct io_cb_cancel_data {
	struct io_wqe *wqe;
	work_cancel_fn *cancel;
	void *caller_data;
};

static bool io_work_cancel(struct io_worker *worker, void *cancel_data)
{
	struct io_cb_cancel_data *data = cancel_data;
860
	unsigned long flags;
861 862 863 864
	bool ret = false;

	/*
	 * Hold the lock to avoid ->cur_work going out of scope, caller
865
	 * may dereference the passed in work.
866
	 */
867
	spin_lock_irqsave(&worker->lock, flags);
868
	if (worker->cur_work &&
869
	    !(worker->cur_work->flags & IO_WQ_WORK_NO_CANCEL) &&
870 871 872 873
	    data->cancel(worker->cur_work, data->caller_data)) {
		send_sig(SIGINT, worker->task, 1);
		ret = true;
	}
874
	spin_unlock_irqrestore(&worker->lock, flags);
875 876 877 878 879 880 881 882 883 884 885 886 887

	return ret;
}

static enum io_wq_cancel io_wqe_cancel_cb_work(struct io_wqe *wqe,
					       work_cancel_fn *cancel,
					       void *cancel_data)
{
	struct io_cb_cancel_data data = {
		.wqe = wqe,
		.cancel = cancel,
		.caller_data = cancel_data,
	};
J
Jens Axboe 已提交
888
	struct io_wq_work_node *node, *prev;
889
	struct io_wq_work *work;
890
	unsigned long flags;
891 892
	bool found = false;

893
	spin_lock_irqsave(&wqe->lock, flags);
J
Jens Axboe 已提交
894 895 896
	wq_list_for_each(node, prev, &wqe->work_list) {
		work = container_of(node, struct io_wq_work, list);

897
		if (cancel(work, cancel_data)) {
J
Jens Axboe 已提交
898
			wq_node_del(&wqe->work_list, node, prev);
899 900 901 902
			found = true;
			break;
		}
	}
903
	spin_unlock_irqrestore(&wqe->lock, flags);
904 905 906 907 908 909 910 911

	if (found) {
		work->flags |= IO_WQ_WORK_CANCEL;
		work->func(&work);
		return IO_WQ_CANCEL_OK;
	}

	rcu_read_lock();
912
	found = io_wq_for_each_worker(wqe, io_work_cancel, &data);
913 914 915 916 917 918 919 920
	rcu_read_unlock();
	return found ? IO_WQ_CANCEL_RUNNING : IO_WQ_CANCEL_NOTFOUND;
}

enum io_wq_cancel io_wq_cancel_cb(struct io_wq *wq, work_cancel_fn *cancel,
				  void *data)
{
	enum io_wq_cancel ret = IO_WQ_CANCEL_NOTFOUND;
J
Jann Horn 已提交
921
	int node;
922

J
Jann Horn 已提交
923 924
	for_each_node(node) {
		struct io_wqe *wqe = wq->wqes[node];
925 926 927 928 929 930 931 932 933

		ret = io_wqe_cancel_cb_work(wqe, cancel, data);
		if (ret != IO_WQ_CANCEL_NOTFOUND)
			break;
	}

	return ret;
}

934 935 936
static bool io_wq_worker_cancel(struct io_worker *worker, void *data)
{
	struct io_wq_work *work = data;
937 938
	unsigned long flags;
	bool ret = false;
939

940 941 942 943
	if (worker->cur_work != work)
		return false;

	spin_lock_irqsave(&worker->lock, flags);
944 945
	if (worker->cur_work == work &&
	    !(worker->cur_work->flags & IO_WQ_WORK_NO_CANCEL)) {
946
		send_sig(SIGINT, worker->task, 1);
947
		ret = true;
948
	}
949
	spin_unlock_irqrestore(&worker->lock, flags);
950

951
	return ret;
952 953 954 955 956
}

static enum io_wq_cancel io_wqe_cancel_work(struct io_wqe *wqe,
					    struct io_wq_work *cwork)
{
J
Jens Axboe 已提交
957
	struct io_wq_work_node *node, *prev;
958
	struct io_wq_work *work;
959
	unsigned long flags;
960 961 962 963 964 965 966 967 968
	bool found = false;

	cwork->flags |= IO_WQ_WORK_CANCEL;

	/*
	 * First check pending list, if we're lucky we can just remove it
	 * from there. CANCEL_OK means that the work is returned as-new,
	 * no completion will be posted for it.
	 */
969
	spin_lock_irqsave(&wqe->lock, flags);
J
Jens Axboe 已提交
970 971 972
	wq_list_for_each(node, prev, &wqe->work_list) {
		work = container_of(node, struct io_wq_work, list);

973
		if (work == cwork) {
J
Jens Axboe 已提交
974
			wq_node_del(&wqe->work_list, node, prev);
975 976 977 978
			found = true;
			break;
		}
	}
979
	spin_unlock_irqrestore(&wqe->lock, flags);
980 981 982 983 984 985 986 987 988 989

	if (found) {
		work->flags |= IO_WQ_WORK_CANCEL;
		work->func(&work);
		return IO_WQ_CANCEL_OK;
	}

	/*
	 * Now check if a free (going busy) or busy worker has the work
	 * currently running. If we find it there, we'll return CANCEL_RUNNING
990
	 * as an indication that we attempt to signal cancellation. The
991 992 993
	 * completion will run normally in this case.
	 */
	rcu_read_lock();
994
	found = io_wq_for_each_worker(wqe, io_wq_worker_cancel, cwork);
995 996 997 998 999 1000 1001
	rcu_read_unlock();
	return found ? IO_WQ_CANCEL_RUNNING : IO_WQ_CANCEL_NOTFOUND;
}

enum io_wq_cancel io_wq_cancel_work(struct io_wq *wq, struct io_wq_work *cwork)
{
	enum io_wq_cancel ret = IO_WQ_CANCEL_NOTFOUND;
J
Jann Horn 已提交
1002
	int node;
1003

J
Jann Horn 已提交
1004 1005
	for_each_node(node) {
		struct io_wqe *wqe = wq->wqes[node];
1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035

		ret = io_wqe_cancel_work(wqe, cwork);
		if (ret != IO_WQ_CANCEL_NOTFOUND)
			break;
	}

	return ret;
}

struct io_wq_flush_data {
	struct io_wq_work work;
	struct completion done;
};

static void io_wq_flush_func(struct io_wq_work **workptr)
{
	struct io_wq_work *work = *workptr;
	struct io_wq_flush_data *data;

	data = container_of(work, struct io_wq_flush_data, work);
	complete(&data->done);
}

/*
 * Doesn't wait for previously queued work to finish. When this completes,
 * it just means that previously queued work was started.
 */
void io_wq_flush(struct io_wq *wq)
{
	struct io_wq_flush_data data;
J
Jann Horn 已提交
1036
	int node;
1037

J
Jann Horn 已提交
1038 1039
	for_each_node(node) {
		struct io_wqe *wqe = wq->wqes[node];
1040 1041 1042

		init_completion(&data.done);
		INIT_IO_WORK(&data.work, io_wq_flush_func);
1043
		data.work.flags |= IO_WQ_WORK_INTERNAL;
1044 1045 1046 1047 1048
		io_wqe_enqueue(wqe, &data.work);
		wait_for_completion(&data.done);
	}
}

1049
struct io_wq *io_wq_create(unsigned bounded, struct io_wq_data *data)
1050
{
J
Jann Horn 已提交
1051
	int ret = -ENOMEM, node;
1052 1053
	struct io_wq *wq;

1054
	wq = kzalloc(sizeof(*wq), GFP_KERNEL);
1055 1056 1057
	if (!wq)
		return ERR_PTR(-ENOMEM);

J
Jann Horn 已提交
1058
	wq->wqes = kcalloc(nr_node_ids, sizeof(struct io_wqe *), GFP_KERNEL);
1059 1060 1061 1062 1063
	if (!wq->wqes) {
		kfree(wq);
		return ERR_PTR(-ENOMEM);
	}

1064 1065
	wq->get_work = data->get_work;
	wq->put_work = data->put_work;
1066

1067
	/* caller must already hold a reference to this */
1068
	wq->user = data->user;
1069

J
Jann Horn 已提交
1070
	for_each_node(node) {
1071 1072
		struct io_wqe *wqe;

1073
		wqe = kzalloc_node(sizeof(struct io_wqe), GFP_KERNEL, node);
1074
		if (!wqe)
J
Jann Horn 已提交
1075 1076
			goto err;
		wq->wqes[node] = wqe;
1077
		wqe->node = node;
1078 1079
		wqe->acct[IO_WQ_ACCT_BOUND].max_workers = bounded;
		atomic_set(&wqe->acct[IO_WQ_ACCT_BOUND].nr_running, 0);
1080
		if (wq->user) {
1081 1082 1083 1084
			wqe->acct[IO_WQ_ACCT_UNBOUND].max_workers =
					task_rlimit(current, RLIMIT_NPROC);
		}
		atomic_set(&wqe->acct[IO_WQ_ACCT_UNBOUND].nr_running, 0);
1085 1086 1087
		wqe->node = node;
		wqe->wq = wq;
		spin_lock_init(&wqe->lock);
J
Jens Axboe 已提交
1088
		INIT_WQ_LIST(&wqe->work_list);
1089
		INIT_HLIST_NULLS_HEAD(&wqe->free_list, 0);
1090
		INIT_LIST_HEAD(&wqe->all_list);
1091 1092 1093 1094 1095 1096 1097
	}

	init_completion(&wq->done);

	wq->manager = kthread_create(io_wq_manager, wq, "io_wq_manager");
	if (!IS_ERR(wq->manager)) {
		wake_up_process(wq->manager);
1098 1099 1100 1101 1102
		wait_for_completion(&wq->done);
		if (test_bit(IO_WQ_BIT_ERROR, &wq->state)) {
			ret = -ENOMEM;
			goto err;
		}
J
Jens Axboe 已提交
1103
		refcount_set(&wq->use_refs, 1);
1104
		reinit_completion(&wq->done);
1105 1106 1107 1108 1109
		return wq;
	}

	ret = PTR_ERR(wq->manager);
	complete(&wq->done);
1110
err:
J
Jann Horn 已提交
1111 1112
	for_each_node(node)
		kfree(wq->wqes[node]);
1113 1114
	kfree(wq->wqes);
	kfree(wq);
1115 1116 1117
	return ERR_PTR(ret);
}

1118 1119 1120 1121 1122 1123 1124 1125
bool io_wq_get(struct io_wq *wq, struct io_wq_data *data)
{
	if (data->get_work != wq->get_work || data->put_work != wq->put_work)
		return false;

	return refcount_inc_not_zero(&wq->use_refs);
}

1126 1127 1128 1129 1130 1131
static bool io_wq_worker_wake(struct io_worker *worker, void *data)
{
	wake_up_process(worker->task);
	return false;
}

J
Jens Axboe 已提交
1132
static void __io_wq_destroy(struct io_wq *wq)
1133
{
J
Jann Horn 已提交
1134
	int node;
1135

1136 1137
	set_bit(IO_WQ_BIT_EXIT, &wq->state);
	if (wq->manager)
1138 1139 1140
		kthread_stop(wq->manager);

	rcu_read_lock();
J
Jann Horn 已提交
1141 1142
	for_each_node(node)
		io_wq_for_each_worker(wq->wqes[node], io_wq_worker_wake, NULL);
1143 1144 1145 1146
	rcu_read_unlock();

	wait_for_completion(&wq->done);

J
Jann Horn 已提交
1147 1148
	for_each_node(node)
		kfree(wq->wqes[node]);
1149 1150 1151
	kfree(wq->wqes);
	kfree(wq);
}
J
Jens Axboe 已提交
1152 1153 1154 1155 1156 1157

void io_wq_destroy(struct io_wq *wq)
{
	if (refcount_dec_and_test(&wq->use_refs))
		__io_wq_destroy(wq);
}