1// SPDX-License-Identifier: GPL-2.0 2/* 3 * Basic worker thread pool for io_uring 4 * 5 * Copyright (C) 2019 Jens Axboe 6 * 7 */ 8#include <linux/kernel.h> 9#include <linux/init.h> 10#include <linux/errno.h> 11#include <linux/sched/signal.h> 12#include <linux/percpu.h> 13#include <linux/slab.h> 14#include <linux/rculist_nulls.h> 15#include <linux/cpu.h> 16#include <linux/task_work.h> 17#include <linux/audit.h> 18#include <linux/mmu_context.h> 19#include <uapi/linux/io_uring.h> 20 21#include "io-wq.h" 22#include "slist.h" 23#include "io_uring.h" 24 25#define WORKER_IDLE_TIMEOUT (5 * HZ) 26 27enum { 28 IO_WORKER_F_UP = 1, /* up and active */ 29 IO_WORKER_F_RUNNING = 2, /* account as running */ 30 IO_WORKER_F_FREE = 4, /* worker on free list */ 31 IO_WORKER_F_BOUND = 8, /* is doing bounded work */ 32}; 33 34enum { 35 IO_WQ_BIT_EXIT = 0, /* wq exiting */ 36}; 37 38enum { 39 IO_ACCT_STALLED_BIT = 0, /* stalled on hash */ 40}; 41 42/* 43 * One for each thread in a wq pool 44 */ 45struct io_worker { 46 refcount_t ref; 47 unsigned flags; 48 struct hlist_nulls_node nulls_node; 49 struct list_head all_list; 50 struct task_struct *task; 51 struct io_wq *wq; 52 53 struct io_wq_work *cur_work; 54 struct io_wq_work *next_work; 55 raw_spinlock_t lock; 56 57 struct completion ref_done; 58 59 unsigned long create_state; 60 struct callback_head create_work; 61 int create_index; 62 63 union { 64 struct rcu_head rcu; 65 struct work_struct work; 66 }; 67}; 68 69#if BITS_PER_LONG == 64 70#define IO_WQ_HASH_ORDER 6 71#else 72#define IO_WQ_HASH_ORDER 5 73#endif 74 75#define IO_WQ_NR_HASH_BUCKETS (1u << IO_WQ_HASH_ORDER) 76 77struct io_wq_acct { 78 unsigned nr_workers; 79 unsigned max_workers; 80 int index; 81 atomic_t nr_running; 82 raw_spinlock_t lock; 83 struct io_wq_work_list work_list; 84 unsigned long flags; 85}; 86 87enum { 88 IO_WQ_ACCT_BOUND, 89 IO_WQ_ACCT_UNBOUND, 90 IO_WQ_ACCT_NR, 91}; 92 93/* 94 * Per io_wq state 95 */ 96struct io_wq { 97 unsigned long state; 98 99 free_work_fn *free_work; 100 io_wq_work_fn *do_work; 101 102 struct io_wq_hash *hash; 103 104 atomic_t worker_refs; 105 struct completion worker_done; 106 107 struct hlist_node cpuhp_node; 108 109 struct task_struct *task; 110 111 struct io_wq_acct acct[IO_WQ_ACCT_NR]; 112 113 /* lock protects access to elements below */ 114 raw_spinlock_t lock; 115 116 struct hlist_nulls_head free_list; 117 struct list_head all_list; 118 119 struct wait_queue_entry wait; 120 121 struct io_wq_work *hash_tail[IO_WQ_NR_HASH_BUCKETS]; 122 123 cpumask_var_t cpu_mask; 124}; 125 126static enum cpuhp_state io_wq_online; 127 128struct io_cb_cancel_data { 129 work_cancel_fn *fn; 130 void *data; 131 int nr_running; 132 int nr_pending; 133 bool cancel_all; 134}; 135 136static bool create_io_worker(struct io_wq *wq, int index); 137static void io_wq_dec_running(struct io_worker *worker); 138static bool io_acct_cancel_pending_work(struct io_wq *wq, 139 struct io_wq_acct *acct, 140 struct io_cb_cancel_data *match); 141static void create_worker_cb(struct callback_head *cb); 142static void io_wq_cancel_tw_create(struct io_wq *wq); 143 144static bool io_worker_get(struct io_worker *worker) 145{ 146 return refcount_inc_not_zero(&worker->ref); 147} 148 149static void io_worker_release(struct io_worker *worker) 150{ 151 if (refcount_dec_and_test(&worker->ref)) 152 complete(&worker->ref_done); 153} 154 155static inline struct io_wq_acct *io_get_acct(struct io_wq *wq, bool bound) 156{ 157 return &wq->acct[bound ? IO_WQ_ACCT_BOUND : IO_WQ_ACCT_UNBOUND]; 158} 159 160static inline struct io_wq_acct *io_work_get_acct(struct io_wq *wq, 161 struct io_wq_work *work) 162{ 163 return io_get_acct(wq, !(work->flags & IO_WQ_WORK_UNBOUND)); 164} 165 166static inline struct io_wq_acct *io_wq_get_acct(struct io_worker *worker) 167{ 168 return io_get_acct(worker->wq, worker->flags & IO_WORKER_F_BOUND); 169} 170 171static void io_worker_ref_put(struct io_wq *wq) 172{ 173 if (atomic_dec_and_test(&wq->worker_refs)) 174 complete(&wq->worker_done); 175} 176 177bool io_wq_worker_stopped(void) 178{ 179 struct io_worker *worker = current->worker_private; 180 181 if (WARN_ON_ONCE(!io_wq_current_is_worker())) 182 return true; 183 184 return test_bit(IO_WQ_BIT_EXIT, &worker->wq->state); 185} 186 187static void io_worker_cancel_cb(struct io_worker *worker) 188{ 189 struct io_wq_acct *acct = io_wq_get_acct(worker); 190 struct io_wq *wq = worker->wq; 191 192 atomic_dec(&acct->nr_running); 193 raw_spin_lock(&wq->lock); 194 acct->nr_workers--; 195 raw_spin_unlock(&wq->lock); 196 io_worker_ref_put(wq); 197 clear_bit_unlock(0, &worker->create_state); 198 io_worker_release(worker); 199} 200 201static bool io_task_worker_match(struct callback_head *cb, void *data) 202{ 203 struct io_worker *worker; 204 205 if (cb->func != create_worker_cb) 206 return false; 207 worker = container_of(cb, struct io_worker, create_work); 208 return worker == data; 209} 210 211static void io_worker_exit(struct io_worker *worker) 212{ 213 struct io_wq *wq = worker->wq; 214 215 while (1) { 216 struct callback_head *cb = task_work_cancel_match(wq->task, 217 io_task_worker_match, worker); 218 219 if (!cb) 220 break; 221 io_worker_cancel_cb(worker); 222 } 223 224 io_worker_release(worker); 225 wait_for_completion(&worker->ref_done); 226 227 raw_spin_lock(&wq->lock); 228 if (worker->flags & IO_WORKER_F_FREE) 229 hlist_nulls_del_rcu(&worker->nulls_node); 230 list_del_rcu(&worker->all_list); 231 raw_spin_unlock(&wq->lock); 232 io_wq_dec_running(worker); 233 /* 234 * this worker is a goner, clear ->worker_private to avoid any 235 * inc/dec running calls that could happen as part of exit from 236 * touching 'worker'. 237 */ 238 current->worker_private = NULL; 239 240 kfree_rcu(worker, rcu); 241 io_worker_ref_put(wq); 242 do_exit(0); 243} 244 245static inline bool __io_acct_run_queue(struct io_wq_acct *acct) 246{ 247 return !test_bit(IO_ACCT_STALLED_BIT, &acct->flags) && 248 !wq_list_empty(&acct->work_list); 249} 250 251/* 252 * If there's work to do, returns true with acct->lock acquired. If not, 253 * returns false with no lock held. 254 */ 255static inline bool io_acct_run_queue(struct io_wq_acct *acct) 256 __acquires(&acct->lock) 257{ 258 raw_spin_lock(&acct->lock); 259 if (__io_acct_run_queue(acct)) 260 return true; 261 262 raw_spin_unlock(&acct->lock); 263 return false; 264} 265 266/* 267 * Check head of free list for an available worker. If one isn't available, 268 * caller must create one. 269 */ 270static bool io_wq_activate_free_worker(struct io_wq *wq, 271 struct io_wq_acct *acct) 272 __must_hold(RCU) 273{ 274 struct hlist_nulls_node *n; 275 struct io_worker *worker; 276 277 /* 278 * Iterate free_list and see if we can find an idle worker to 279 * activate. If a given worker is on the free_list but in the process 280 * of exiting, keep trying. 281 */ 282 hlist_nulls_for_each_entry_rcu(worker, n, &wq->free_list, nulls_node) { 283 if (!io_worker_get(worker)) 284 continue; 285 if (io_wq_get_acct(worker) != acct) { 286 io_worker_release(worker); 287 continue; 288 } 289 /* 290 * If the worker is already running, it's either already 291 * starting work or finishing work. In either case, if it does 292 * to go sleep, we'll kick off a new task for this work anyway. 293 */ 294 wake_up_process(worker->task); 295 io_worker_release(worker); 296 return true; 297 } 298 299 return false; 300} 301 302/* 303 * We need a worker. If we find a free one, we're good. If not, and we're 304 * below the max number of workers, create one. 305 */ 306static bool io_wq_create_worker(struct io_wq *wq, struct io_wq_acct *acct) 307{ 308 /* 309 * Most likely an attempt to queue unbounded work on an io_wq that 310 * wasn't setup with any unbounded workers. 311 */ 312 if (unlikely(!acct->max_workers)) 313 pr_warn_once("io-wq is not configured for unbound workers"); 314 315 raw_spin_lock(&wq->lock); 316 if (acct->nr_workers >= acct->max_workers) { 317 raw_spin_unlock(&wq->lock); 318 return true; 319 } 320 acct->nr_workers++; 321 raw_spin_unlock(&wq->lock); 322 atomic_inc(&acct->nr_running); 323 atomic_inc(&wq->worker_refs); 324 return create_io_worker(wq, acct->index); 325} 326 327static void io_wq_inc_running(struct io_worker *worker) 328{ 329 struct io_wq_acct *acct = io_wq_get_acct(worker); 330 331 atomic_inc(&acct->nr_running); 332} 333 334static void create_worker_cb(struct callback_head *cb) 335{ 336 struct io_worker *worker; 337 struct io_wq *wq; 338 339 struct io_wq_acct *acct; 340 bool do_create = false; 341 342 worker = container_of(cb, struct io_worker, create_work); 343 wq = worker->wq; 344 acct = &wq->acct[worker->create_index]; 345 raw_spin_lock(&wq->lock); 346 347 if (acct->nr_workers < acct->max_workers) { 348 acct->nr_workers++; 349 do_create = true; 350 } 351 raw_spin_unlock(&wq->lock); 352 if (do_create) { 353 create_io_worker(wq, worker->create_index); 354 } else { 355 atomic_dec(&acct->nr_running); 356 io_worker_ref_put(wq); 357 } 358 clear_bit_unlock(0, &worker->create_state); 359 io_worker_release(worker); 360} 361 362static bool io_queue_worker_create(struct io_worker *worker, 363 struct io_wq_acct *acct, 364 task_work_func_t func) 365{ 366 struct io_wq *wq = worker->wq; 367 368 /* raced with exit, just ignore create call */ 369 if (test_bit(IO_WQ_BIT_EXIT, &wq->state)) 370 goto fail; 371 if (!io_worker_get(worker)) 372 goto fail; 373 /* 374 * create_state manages ownership of create_work/index. We should 375 * only need one entry per worker, as the worker going to sleep 376 * will trigger the condition, and waking will clear it once it 377 * runs the task_work. 378 */ 379 if (test_bit(0, &worker->create_state) || 380 test_and_set_bit_lock(0, &worker->create_state)) 381 goto fail_release; 382 383 atomic_inc(&wq->worker_refs); 384 init_task_work(&worker->create_work, func); 385 worker->create_index = acct->index; 386 if (!task_work_add(wq->task, &worker->create_work, TWA_SIGNAL)) { 387 /* 388 * EXIT may have been set after checking it above, check after 389 * adding the task_work and remove any creation item if it is 390 * now set. wq exit does that too, but we can have added this 391 * work item after we canceled in io_wq_exit_workers(). 392 */ 393 if (test_bit(IO_WQ_BIT_EXIT, &wq->state)) 394 io_wq_cancel_tw_create(wq); 395 io_worker_ref_put(wq); 396 return true; 397 } 398 io_worker_ref_put(wq); 399 clear_bit_unlock(0, &worker->create_state); 400fail_release: 401 io_worker_release(worker); 402fail: 403 atomic_dec(&acct->nr_running); 404 io_worker_ref_put(wq); 405 return false; 406} 407 408static void io_wq_dec_running(struct io_worker *worker) 409{ 410 struct io_wq_acct *acct = io_wq_get_acct(worker); 411 struct io_wq *wq = worker->wq; 412 413 if (!(worker->flags & IO_WORKER_F_UP)) 414 return; 415 416 if (!atomic_dec_and_test(&acct->nr_running)) 417 return; 418 if (!io_acct_run_queue(acct)) 419 return; 420 421 raw_spin_unlock(&acct->lock); 422 atomic_inc(&acct->nr_running); 423 atomic_inc(&wq->worker_refs); 424 io_queue_worker_create(worker, acct, create_worker_cb); 425} 426 427/* 428 * Worker will start processing some work. Move it to the busy list, if 429 * it's currently on the freelist 430 */ 431static void __io_worker_busy(struct io_wq *wq, struct io_worker *worker) 432{ 433 if (worker->flags & IO_WORKER_F_FREE) { 434 worker->flags &= ~IO_WORKER_F_FREE; 435 raw_spin_lock(&wq->lock); 436 hlist_nulls_del_init_rcu(&worker->nulls_node); 437 raw_spin_unlock(&wq->lock); 438 } 439} 440 441/* 442 * No work, worker going to sleep. Move to freelist. 443 */ 444static void __io_worker_idle(struct io_wq *wq, struct io_worker *worker) 445 __must_hold(wq->lock) 446{ 447 if (!(worker->flags & IO_WORKER_F_FREE)) { 448 worker->flags |= IO_WORKER_F_FREE; 449 hlist_nulls_add_head_rcu(&worker->nulls_node, &wq->free_list); 450 } 451} 452 453static inline unsigned int io_get_work_hash(struct io_wq_work *work) 454{ 455 return work->flags >> IO_WQ_HASH_SHIFT; 456} 457 458static bool io_wait_on_hash(struct io_wq *wq, unsigned int hash) 459{ 460 bool ret = false; 461 462 spin_lock_irq(&wq->hash->wait.lock); 463 if (list_empty(&wq->wait.entry)) { 464 __add_wait_queue(&wq->hash->wait, &wq->wait); 465 if (!test_bit(hash, &wq->hash->map)) { 466 __set_current_state(TASK_RUNNING); 467 list_del_init(&wq->wait.entry); 468 ret = true; 469 } 470 } 471 spin_unlock_irq(&wq->hash->wait.lock); 472 return ret; 473} 474 475static struct io_wq_work *io_get_next_work(struct io_wq_acct *acct, 476 struct io_worker *worker) 477 __must_hold(acct->lock) 478{ 479 struct io_wq_work_node *node, *prev; 480 struct io_wq_work *work, *tail; 481 unsigned int stall_hash = -1U; 482 struct io_wq *wq = worker->wq; 483 484 wq_list_for_each(node, prev, &acct->work_list) { 485 unsigned int hash; 486 487 work = container_of(node, struct io_wq_work, list); 488 489 /* not hashed, can run anytime */ 490 if (!io_wq_is_hashed(work)) { 491 wq_list_del(&acct->work_list, node, prev); 492 return work; 493 } 494 495 hash = io_get_work_hash(work); 496 /* all items with this hash lie in [work, tail] */ 497 tail = wq->hash_tail[hash]; 498 499 /* hashed, can run if not already running */ 500 if (!test_and_set_bit(hash, &wq->hash->map)) { 501 wq->hash_tail[hash] = NULL; 502 wq_list_cut(&acct->work_list, &tail->list, prev); 503 return work; 504 } 505 if (stall_hash == -1U) 506 stall_hash = hash; 507 /* fast forward to a next hash, for-each will fix up @prev */ 508 node = &tail->list; 509 } 510 511 if (stall_hash != -1U) { 512 bool unstalled; 513 514 /* 515 * Set this before dropping the lock to avoid racing with new 516 * work being added and clearing the stalled bit. 517 */ 518 set_bit(IO_ACCT_STALLED_BIT, &acct->flags); 519 raw_spin_unlock(&acct->lock); 520 unstalled = io_wait_on_hash(wq, stall_hash); 521 raw_spin_lock(&acct->lock); 522 if (unstalled) { 523 clear_bit(IO_ACCT_STALLED_BIT, &acct->flags); 524 if (wq_has_sleeper(&wq->hash->wait)) 525 wake_up(&wq->hash->wait); 526 } 527 } 528 529 return NULL; 530} 531 532static void io_assign_current_work(struct io_worker *worker, 533 struct io_wq_work *work) 534{ 535 if (work) { 536 io_run_task_work(); 537 cond_resched(); 538 } 539 540 raw_spin_lock(&worker->lock); 541 worker->cur_work = work; 542 worker->next_work = NULL; 543 raw_spin_unlock(&worker->lock); 544} 545 546/* 547 * Called with acct->lock held, drops it before returning 548 */ 549static void io_worker_handle_work(struct io_wq_acct *acct, 550 struct io_worker *worker) 551 __releases(&acct->lock) 552{ 553 struct io_wq *wq = worker->wq; 554 bool do_kill = test_bit(IO_WQ_BIT_EXIT, &wq->state); 555 556 do { 557 struct io_wq_work *work; 558 559 /* 560 * If we got some work, mark us as busy. If we didn't, but 561 * the list isn't empty, it means we stalled on hashed work. 562 * Mark us stalled so we don't keep looking for work when we 563 * can't make progress, any work completion or insertion will 564 * clear the stalled flag. 565 */ 566 work = io_get_next_work(acct, worker); 567 raw_spin_unlock(&acct->lock); 568 if (work) { 569 __io_worker_busy(wq, worker); 570 571 /* 572 * Make sure cancelation can find this, even before 573 * it becomes the active work. That avoids a window 574 * where the work has been removed from our general 575 * work list, but isn't yet discoverable as the 576 * current work item for this worker. 577 */ 578 raw_spin_lock(&worker->lock); 579 worker->next_work = work; 580 raw_spin_unlock(&worker->lock); 581 } else { 582 break; 583 } 584 io_assign_current_work(worker, work); 585 __set_current_state(TASK_RUNNING); 586 587 /* handle a whole dependent link */ 588 do { 589 struct io_wq_work *next_hashed, *linked; 590 unsigned int hash = io_get_work_hash(work); 591 592 next_hashed = wq_next_work(work); 593 594 if (unlikely(do_kill) && (work->flags & IO_WQ_WORK_UNBOUND)) 595 work->flags |= IO_WQ_WORK_CANCEL; 596 wq->do_work(work); 597 io_assign_current_work(worker, NULL); 598 599 linked = wq->free_work(work); 600 work = next_hashed; 601 if (!work && linked && !io_wq_is_hashed(linked)) { 602 work = linked; 603 linked = NULL; 604 } 605 io_assign_current_work(worker, work); 606 if (linked) 607 io_wq_enqueue(wq, linked); 608 609 if (hash != -1U && !next_hashed) { 610 /* serialize hash clear with wake_up() */ 611 spin_lock_irq(&wq->hash->wait.lock); 612 clear_bit(hash, &wq->hash->map); 613 clear_bit(IO_ACCT_STALLED_BIT, &acct->flags); 614 spin_unlock_irq(&wq->hash->wait.lock); 615 if (wq_has_sleeper(&wq->hash->wait)) 616 wake_up(&wq->hash->wait); 617 } 618 } while (work); 619 620 if (!__io_acct_run_queue(acct)) 621 break; 622 raw_spin_lock(&acct->lock); 623 } while (1); 624} 625 626static int io_wq_worker(void *data) 627{ 628 struct io_worker *worker = data; 629 struct io_wq_acct *acct = io_wq_get_acct(worker); 630 struct io_wq *wq = worker->wq; 631 bool exit_mask = false, last_timeout = false; 632 char buf[TASK_COMM_LEN]; 633 634 worker->flags |= (IO_WORKER_F_UP | IO_WORKER_F_RUNNING); 635 636 snprintf(buf, sizeof(buf), "iou-wrk-%d", wq->task->pid); 637 set_task_comm(current, buf); 638 639 while (!test_bit(IO_WQ_BIT_EXIT, &wq->state)) { 640 long ret; 641 642 set_current_state(TASK_INTERRUPTIBLE); 643 644 /* 645 * If we have work to do, io_acct_run_queue() returns with 646 * the acct->lock held. If not, it will drop it. 647 */ 648 while (io_acct_run_queue(acct)) 649 io_worker_handle_work(acct, worker); 650 651 raw_spin_lock(&wq->lock); 652 /* 653 * Last sleep timed out. Exit if we're not the last worker, 654 * or if someone modified our affinity. 655 */ 656 if (last_timeout && (exit_mask || acct->nr_workers > 1)) { 657 acct->nr_workers--; 658 raw_spin_unlock(&wq->lock); 659 __set_current_state(TASK_RUNNING); 660 break; 661 } 662 last_timeout = false; 663 __io_worker_idle(wq, worker); 664 raw_spin_unlock(&wq->lock); 665 if (io_run_task_work()) 666 continue; 667 ret = schedule_timeout(WORKER_IDLE_TIMEOUT); 668 if (signal_pending(current)) { 669 struct ksignal ksig; 670 671 if (!get_signal(&ksig)) 672 continue; 673 break; 674 } 675 if (!ret) { 676 last_timeout = true; 677 exit_mask = !cpumask_test_cpu(raw_smp_processor_id(), 678 wq->cpu_mask); 679 } 680 } 681 682 if (test_bit(IO_WQ_BIT_EXIT, &wq->state) && io_acct_run_queue(acct)) 683 io_worker_handle_work(acct, worker); 684 685 io_worker_exit(worker); 686 return 0; 687} 688 689/* 690 * Called when a worker is scheduled in. Mark us as currently running. 691 */ 692void io_wq_worker_running(struct task_struct *tsk) 693{ 694 struct io_worker *worker = tsk->worker_private; 695 696 if (!worker) 697 return; 698 if (!(worker->flags & IO_WORKER_F_UP)) 699 return; 700 if (worker->flags & IO_WORKER_F_RUNNING) 701 return; 702 worker->flags |= IO_WORKER_F_RUNNING; 703 io_wq_inc_running(worker); 704} 705 706/* 707 * Called when worker is going to sleep. If there are no workers currently 708 * running and we have work pending, wake up a free one or create a new one. 709 */ 710void io_wq_worker_sleeping(struct task_struct *tsk) 711{ 712 struct io_worker *worker = tsk->worker_private; 713 714 if (!worker) 715 return; 716 if (!(worker->flags & IO_WORKER_F_UP)) 717 return; 718 if (!(worker->flags & IO_WORKER_F_RUNNING)) 719 return; 720 721 worker->flags &= ~IO_WORKER_F_RUNNING; 722 io_wq_dec_running(worker); 723} 724 725static void io_init_new_worker(struct io_wq *wq, struct io_worker *worker, 726 struct task_struct *tsk) 727{ 728 tsk->worker_private = worker; 729 worker->task = tsk; 730 set_cpus_allowed_ptr(tsk, wq->cpu_mask); 731 732 raw_spin_lock(&wq->lock); 733 hlist_nulls_add_head_rcu(&worker->nulls_node, &wq->free_list); 734 list_add_tail_rcu(&worker->all_list, &wq->all_list); 735 worker->flags |= IO_WORKER_F_FREE; 736 raw_spin_unlock(&wq->lock); 737 wake_up_new_task(tsk); 738} 739 740static bool io_wq_work_match_all(struct io_wq_work *work, void *data) 741{ 742 return true; 743} 744 745static inline bool io_should_retry_thread(long err) 746{ 747 /* 748 * Prevent perpetual task_work retry, if the task (or its group) is 749 * exiting. 750 */ 751 if (fatal_signal_pending(current)) 752 return false; 753 754 switch (err) { 755 case -EAGAIN: 756 case -ERESTARTSYS: 757 case -ERESTARTNOINTR: 758 case -ERESTARTNOHAND: 759 return true; 760 default: 761 return false; 762 } 763} 764 765static void create_worker_cont(struct callback_head *cb) 766{ 767 struct io_worker *worker; 768 struct task_struct *tsk; 769 struct io_wq *wq; 770 771 worker = container_of(cb, struct io_worker, create_work); 772 clear_bit_unlock(0, &worker->create_state); 773 wq = worker->wq; 774 tsk = create_io_thread(io_wq_worker, worker, NUMA_NO_NODE); 775 if (!IS_ERR(tsk)) { 776 io_init_new_worker(wq, worker, tsk); 777 io_worker_release(worker); 778 return; 779 } else if (!io_should_retry_thread(PTR_ERR(tsk))) { 780 struct io_wq_acct *acct = io_wq_get_acct(worker); 781 782 atomic_dec(&acct->nr_running); 783 raw_spin_lock(&wq->lock); 784 acct->nr_workers--; 785 if (!acct->nr_workers) { 786 struct io_cb_cancel_data match = { 787 .fn = io_wq_work_match_all, 788 .cancel_all = true, 789 }; 790 791 raw_spin_unlock(&wq->lock); 792 while (io_acct_cancel_pending_work(wq, acct, &match)) 793 ; 794 } else { 795 raw_spin_unlock(&wq->lock); 796 } 797 io_worker_ref_put(wq); 798 kfree(worker); 799 return; 800 } 801 802 /* re-create attempts grab a new worker ref, drop the existing one */ 803 io_worker_release(worker); 804 schedule_work(&worker->work); 805} 806 807static void io_workqueue_create(struct work_struct *work) 808{ 809 struct io_worker *worker = container_of(work, struct io_worker, work); 810 struct io_wq_acct *acct = io_wq_get_acct(worker); 811 812 if (!io_queue_worker_create(worker, acct, create_worker_cont)) 813 kfree(worker); 814} 815 816static bool create_io_worker(struct io_wq *wq, int index) 817{ 818 struct io_wq_acct *acct = &wq->acct[index]; 819 struct io_worker *worker; 820 struct task_struct *tsk; 821 822 __set_current_state(TASK_RUNNING); 823 824 worker = kzalloc(sizeof(*worker), GFP_KERNEL); 825 if (!worker) { 826fail: 827 atomic_dec(&acct->nr_running); 828 raw_spin_lock(&wq->lock); 829 acct->nr_workers--; 830 raw_spin_unlock(&wq->lock); 831 io_worker_ref_put(wq); 832 return false; 833 } 834 835 refcount_set(&worker->ref, 1); 836 worker->wq = wq; 837 raw_spin_lock_init(&worker->lock); 838 init_completion(&worker->ref_done); 839 840 if (index == IO_WQ_ACCT_BOUND) 841 worker->flags |= IO_WORKER_F_BOUND; 842 843 tsk = create_io_thread(io_wq_worker, worker, NUMA_NO_NODE); 844 if (!IS_ERR(tsk)) { 845 io_init_new_worker(wq, worker, tsk); 846 } else if (!io_should_retry_thread(PTR_ERR(tsk))) { 847 kfree(worker); 848 goto fail; 849 } else { 850 INIT_WORK(&worker->work, io_workqueue_create); 851 schedule_work(&worker->work); 852 } 853 854 return true; 855} 856 857/* 858 * Iterate the passed in list and call the specific function for each 859 * worker that isn't exiting 860 */ 861static bool io_wq_for_each_worker(struct io_wq *wq, 862 bool (*func)(struct io_worker *, void *), 863 void *data) 864{ 865 struct io_worker *worker; 866 bool ret = false; 867 868 list_for_each_entry_rcu(worker, &wq->all_list, all_list) { 869 if (io_worker_get(worker)) { 870 /* no task if node is/was offline */ 871 if (worker->task) 872 ret = func(worker, data); 873 io_worker_release(worker); 874 if (ret) 875 break; 876 } 877 } 878 879 return ret; 880} 881 882static bool io_wq_worker_wake(struct io_worker *worker, void *data) 883{ 884 __set_notify_signal(worker->task); 885 wake_up_process(worker->task); 886 return false; 887} 888 889static void io_run_cancel(struct io_wq_work *work, struct io_wq *wq) 890{ 891 do { 892 work->flags |= IO_WQ_WORK_CANCEL; 893 wq->do_work(work); 894 work = wq->free_work(work); 895 } while (work); 896} 897 898static void io_wq_insert_work(struct io_wq *wq, struct io_wq_work *work) 899{ 900 struct io_wq_acct *acct = io_work_get_acct(wq, work); 901 unsigned int hash; 902 struct io_wq_work *tail; 903 904 if (!io_wq_is_hashed(work)) { 905append: 906 wq_list_add_tail(&work->list, &acct->work_list); 907 return; 908 } 909 910 hash = io_get_work_hash(work); 911 tail = wq->hash_tail[hash]; 912 wq->hash_tail[hash] = work; 913 if (!tail) 914 goto append; 915 916 wq_list_add_after(&work->list, &tail->list, &acct->work_list); 917} 918 919static bool io_wq_work_match_item(struct io_wq_work *work, void *data) 920{ 921 return work == data; 922} 923 924void io_wq_enqueue(struct io_wq *wq, struct io_wq_work *work) 925{ 926 struct io_wq_acct *acct = io_work_get_acct(wq, work); 927 struct io_cb_cancel_data match; 928 unsigned work_flags = work->flags; 929 bool do_create; 930 931 /* 932 * If io-wq is exiting for this task, or if the request has explicitly 933 * been marked as one that should not get executed, cancel it here. 934 */ 935 if (test_bit(IO_WQ_BIT_EXIT, &wq->state) || 936 (work->flags & IO_WQ_WORK_CANCEL)) { 937 io_run_cancel(work, wq); 938 return; 939 } 940 941 raw_spin_lock(&acct->lock); 942 io_wq_insert_work(wq, work); 943 clear_bit(IO_ACCT_STALLED_BIT, &acct->flags); 944 raw_spin_unlock(&acct->lock); 945 946 rcu_read_lock(); 947 do_create = !io_wq_activate_free_worker(wq, acct); 948 rcu_read_unlock(); 949 950 if (do_create && ((work_flags & IO_WQ_WORK_CONCURRENT) || 951 !atomic_read(&acct->nr_running))) { 952 bool did_create; 953 954 did_create = io_wq_create_worker(wq, acct); 955 if (likely(did_create)) 956 return; 957 958 raw_spin_lock(&wq->lock); 959 if (acct->nr_workers) { 960 raw_spin_unlock(&wq->lock); 961 return; 962 } 963 raw_spin_unlock(&wq->lock); 964 965 /* fatal condition, failed to create the first worker */ 966 match.fn = io_wq_work_match_item, 967 match.data = work, 968 match.cancel_all = false, 969 970 io_acct_cancel_pending_work(wq, acct, &match); 971 } 972} 973 974/* 975 * Work items that hash to the same value will not be done in parallel. 976 * Used to limit concurrent writes, generally hashed by inode. 977 */ 978void io_wq_hash_work(struct io_wq_work *work, void *val) 979{ 980 unsigned int bit; 981 982 bit = hash_ptr(val, IO_WQ_HASH_ORDER); 983 work->flags |= (IO_WQ_WORK_HASHED | (bit << IO_WQ_HASH_SHIFT)); 984} 985 986static bool __io_wq_worker_cancel(struct io_worker *worker, 987 struct io_cb_cancel_data *match, 988 struct io_wq_work *work) 989{ 990 if (work && match->fn(work, match->data)) { 991 work->flags |= IO_WQ_WORK_CANCEL; 992 __set_notify_signal(worker->task); 993 return true; 994 } 995 996 return false; 997} 998 999static bool io_wq_worker_cancel(struct io_worker *worker, void *data) 1000{ 1001 struct io_cb_cancel_data *match = data; 1002 1003 /* 1004 * Hold the lock to avoid ->cur_work going out of scope, caller 1005 * may dereference the passed in work. 1006 */ 1007 raw_spin_lock(&worker->lock); 1008 if (__io_wq_worker_cancel(worker, match, worker->cur_work) || 1009 __io_wq_worker_cancel(worker, match, worker->next_work)) 1010 match->nr_running++; 1011 raw_spin_unlock(&worker->lock); 1012 1013 return match->nr_running && !match->cancel_all; 1014} 1015 1016static inline void io_wq_remove_pending(struct io_wq *wq, 1017 struct io_wq_work *work, 1018 struct io_wq_work_node *prev) 1019{ 1020 struct io_wq_acct *acct = io_work_get_acct(wq, work); 1021 unsigned int hash = io_get_work_hash(work); 1022 struct io_wq_work *prev_work = NULL; 1023 1024 if (io_wq_is_hashed(work) && work == wq->hash_tail[hash]) { 1025 if (prev) 1026 prev_work = container_of(prev, struct io_wq_work, list); 1027 if (prev_work && io_get_work_hash(prev_work) == hash) 1028 wq->hash_tail[hash] = prev_work; 1029 else 1030 wq->hash_tail[hash] = NULL; 1031 } 1032 wq_list_del(&acct->work_list, &work->list, prev); 1033} 1034 1035static bool io_acct_cancel_pending_work(struct io_wq *wq, 1036 struct io_wq_acct *acct, 1037 struct io_cb_cancel_data *match) 1038{ 1039 struct io_wq_work_node *node, *prev; 1040 struct io_wq_work *work; 1041 1042 raw_spin_lock(&acct->lock); 1043 wq_list_for_each(node, prev, &acct->work_list) { 1044 work = container_of(node, struct io_wq_work, list); 1045 if (!match->fn(work, match->data)) 1046 continue; 1047 io_wq_remove_pending(wq, work, prev); 1048 raw_spin_unlock(&acct->lock); 1049 io_run_cancel(work, wq); 1050 match->nr_pending++; 1051 /* not safe to continue after unlock */ 1052 return true; 1053 } 1054 raw_spin_unlock(&acct->lock); 1055 1056 return false; 1057} 1058 1059static void io_wq_cancel_pending_work(struct io_wq *wq, 1060 struct io_cb_cancel_data *match) 1061{ 1062 int i; 1063retry: 1064 for (i = 0; i < IO_WQ_ACCT_NR; i++) { 1065 struct io_wq_acct *acct = io_get_acct(wq, i == 0); 1066 1067 if (io_acct_cancel_pending_work(wq, acct, match)) { 1068 if (match->cancel_all) 1069 goto retry; 1070 break; 1071 } 1072 } 1073} 1074 1075static void io_wq_cancel_running_work(struct io_wq *wq, 1076 struct io_cb_cancel_data *match) 1077{ 1078 rcu_read_lock(); 1079 io_wq_for_each_worker(wq, io_wq_worker_cancel, match); 1080 rcu_read_unlock(); 1081} 1082 1083enum io_wq_cancel io_wq_cancel_cb(struct io_wq *wq, work_cancel_fn *cancel, 1084 void *data, bool cancel_all) 1085{ 1086 struct io_cb_cancel_data match = { 1087 .fn = cancel, 1088 .data = data, 1089 .cancel_all = cancel_all, 1090 }; 1091 1092 /* 1093 * First check pending list, if we're lucky we can just remove it 1094 * from there. CANCEL_OK means that the work is returned as-new, 1095 * no completion will be posted for it. 1096 * 1097 * Then check if a free (going busy) or busy worker has the work 1098 * currently running. If we find it there, we'll return CANCEL_RUNNING 1099 * as an indication that we attempt to signal cancellation. The 1100 * completion will run normally in this case. 1101 * 1102 * Do both of these while holding the wq->lock, to ensure that 1103 * we'll find a work item regardless of state. 1104 */ 1105 io_wq_cancel_pending_work(wq, &match); 1106 if (match.nr_pending && !match.cancel_all) 1107 return IO_WQ_CANCEL_OK; 1108 1109 raw_spin_lock(&wq->lock); 1110 io_wq_cancel_running_work(wq, &match); 1111 raw_spin_unlock(&wq->lock); 1112 if (match.nr_running && !match.cancel_all) 1113 return IO_WQ_CANCEL_RUNNING; 1114 1115 if (match.nr_running) 1116 return IO_WQ_CANCEL_RUNNING; 1117 if (match.nr_pending) 1118 return IO_WQ_CANCEL_OK; 1119 return IO_WQ_CANCEL_NOTFOUND; 1120} 1121 1122static int io_wq_hash_wake(struct wait_queue_entry *wait, unsigned mode, 1123 int sync, void *key) 1124{ 1125 struct io_wq *wq = container_of(wait, struct io_wq, wait); 1126 int i; 1127 1128 list_del_init(&wait->entry); 1129 1130 rcu_read_lock(); 1131 for (i = 0; i < IO_WQ_ACCT_NR; i++) { 1132 struct io_wq_acct *acct = &wq->acct[i]; 1133 1134 if (test_and_clear_bit(IO_ACCT_STALLED_BIT, &acct->flags)) 1135 io_wq_activate_free_worker(wq, acct); 1136 } 1137 rcu_read_unlock(); 1138 return 1; 1139} 1140 1141struct io_wq *io_wq_create(unsigned bounded, struct io_wq_data *data) 1142{ 1143 int ret, i; 1144 struct io_wq *wq; 1145 1146 if (WARN_ON_ONCE(!data->free_work || !data->do_work)) 1147 return ERR_PTR(-EINVAL); 1148 if (WARN_ON_ONCE(!bounded)) 1149 return ERR_PTR(-EINVAL); 1150 1151 wq = kzalloc(sizeof(struct io_wq), GFP_KERNEL); 1152 if (!wq) 1153 return ERR_PTR(-ENOMEM); 1154 1155 refcount_inc(&data->hash->refs); 1156 wq->hash = data->hash; 1157 wq->free_work = data->free_work; 1158 wq->do_work = data->do_work; 1159 1160 ret = -ENOMEM; 1161 1162 if (!alloc_cpumask_var(&wq->cpu_mask, GFP_KERNEL)) 1163 goto err; 1164 cpumask_copy(wq->cpu_mask, cpu_possible_mask); 1165 wq->acct[IO_WQ_ACCT_BOUND].max_workers = bounded; 1166 wq->acct[IO_WQ_ACCT_UNBOUND].max_workers = 1167 task_rlimit(current, RLIMIT_NPROC); 1168 INIT_LIST_HEAD(&wq->wait.entry); 1169 wq->wait.func = io_wq_hash_wake; 1170 for (i = 0; i < IO_WQ_ACCT_NR; i++) { 1171 struct io_wq_acct *acct = &wq->acct[i]; 1172 1173 acct->index = i; 1174 atomic_set(&acct->nr_running, 0); 1175 INIT_WQ_LIST(&acct->work_list); 1176 raw_spin_lock_init(&acct->lock); 1177 } 1178 1179 raw_spin_lock_init(&wq->lock); 1180 INIT_HLIST_NULLS_HEAD(&wq->free_list, 0); 1181 INIT_LIST_HEAD(&wq->all_list); 1182 1183 wq->task = get_task_struct(data->task); 1184 atomic_set(&wq->worker_refs, 1); 1185 init_completion(&wq->worker_done); 1186 ret = cpuhp_state_add_instance_nocalls(io_wq_online, &wq->cpuhp_node); 1187 if (ret) 1188 goto err; 1189 1190 return wq; 1191err: 1192 io_wq_put_hash(data->hash); 1193 free_cpumask_var(wq->cpu_mask); 1194 kfree(wq); 1195 return ERR_PTR(ret); 1196} 1197 1198static bool io_task_work_match(struct callback_head *cb, void *data) 1199{ 1200 struct io_worker *worker; 1201 1202 if (cb->func != create_worker_cb && cb->func != create_worker_cont) 1203 return false; 1204 worker = container_of(cb, struct io_worker, create_work); 1205 return worker->wq == data; 1206} 1207 1208void io_wq_exit_start(struct io_wq *wq) 1209{ 1210 set_bit(IO_WQ_BIT_EXIT, &wq->state); 1211} 1212 1213static void io_wq_cancel_tw_create(struct io_wq *wq) 1214{ 1215 struct callback_head *cb; 1216 1217 while ((cb = task_work_cancel_match(wq->task, io_task_work_match, wq)) != NULL) { 1218 struct io_worker *worker; 1219 1220 worker = container_of(cb, struct io_worker, create_work); 1221 io_worker_cancel_cb(worker); 1222 /* 1223 * Only the worker continuation helper has worker allocated and 1224 * hence needs freeing. 1225 */ 1226 if (cb->func == create_worker_cont) 1227 kfree(worker); 1228 } 1229} 1230 1231static void io_wq_exit_workers(struct io_wq *wq) 1232{ 1233 if (!wq->task) 1234 return; 1235 1236 io_wq_cancel_tw_create(wq); 1237 1238 rcu_read_lock(); 1239 io_wq_for_each_worker(wq, io_wq_worker_wake, NULL); 1240 rcu_read_unlock(); 1241 io_worker_ref_put(wq); 1242 wait_for_completion(&wq->worker_done); 1243 1244 spin_lock_irq(&wq->hash->wait.lock); 1245 list_del_init(&wq->wait.entry); 1246 spin_unlock_irq(&wq->hash->wait.lock); 1247 1248 put_task_struct(wq->task); 1249 wq->task = NULL; 1250} 1251 1252static void io_wq_destroy(struct io_wq *wq) 1253{ 1254 struct io_cb_cancel_data match = { 1255 .fn = io_wq_work_match_all, 1256 .cancel_all = true, 1257 }; 1258 1259 cpuhp_state_remove_instance_nocalls(io_wq_online, &wq->cpuhp_node); 1260 io_wq_cancel_pending_work(wq, &match); 1261 free_cpumask_var(wq->cpu_mask); 1262 io_wq_put_hash(wq->hash); 1263 kfree(wq); 1264} 1265 1266void io_wq_put_and_exit(struct io_wq *wq) 1267{ 1268 WARN_ON_ONCE(!test_bit(IO_WQ_BIT_EXIT, &wq->state)); 1269 1270 io_wq_exit_workers(wq); 1271 io_wq_destroy(wq); 1272} 1273 1274struct online_data { 1275 unsigned int cpu; 1276 bool online; 1277}; 1278 1279static bool io_wq_worker_affinity(struct io_worker *worker, void *data) 1280{ 1281 struct online_data *od = data; 1282 1283 if (od->online) 1284 cpumask_set_cpu(od->cpu, worker->wq->cpu_mask); 1285 else 1286 cpumask_clear_cpu(od->cpu, worker->wq->cpu_mask); 1287 return false; 1288} 1289 1290static int __io_wq_cpu_online(struct io_wq *wq, unsigned int cpu, bool online) 1291{ 1292 struct online_data od = { 1293 .cpu = cpu, 1294 .online = online 1295 }; 1296 1297 rcu_read_lock(); 1298 io_wq_for_each_worker(wq, io_wq_worker_affinity, &od); 1299 rcu_read_unlock(); 1300 return 0; 1301} 1302 1303static int io_wq_cpu_online(unsigned int cpu, struct hlist_node *node) 1304{ 1305 struct io_wq *wq = hlist_entry_safe(node, struct io_wq, cpuhp_node); 1306 1307 return __io_wq_cpu_online(wq, cpu, true); 1308} 1309 1310static int io_wq_cpu_offline(unsigned int cpu, struct hlist_node *node) 1311{ 1312 struct io_wq *wq = hlist_entry_safe(node, struct io_wq, cpuhp_node); 1313 1314 return __io_wq_cpu_online(wq, cpu, false); 1315} 1316 1317int io_wq_cpu_affinity(struct io_uring_task *tctx, cpumask_var_t mask) 1318{ 1319 if (!tctx || !tctx->io_wq) 1320 return -EINVAL; 1321 1322 rcu_read_lock(); 1323 if (mask) 1324 cpumask_copy(tctx->io_wq->cpu_mask, mask); 1325 else 1326 cpumask_copy(tctx->io_wq->cpu_mask, cpu_possible_mask); 1327 rcu_read_unlock(); 1328 1329 return 0; 1330} 1331 1332/* 1333 * Set max number of unbounded workers, returns old value. If new_count is 0, 1334 * then just return the old value. 1335 */ 1336int io_wq_max_workers(struct io_wq *wq, int *new_count) 1337{ 1338 struct io_wq_acct *acct; 1339 int prev[IO_WQ_ACCT_NR]; 1340 int i; 1341 1342 BUILD_BUG_ON((int) IO_WQ_ACCT_BOUND != (int) IO_WQ_BOUND); 1343 BUILD_BUG_ON((int) IO_WQ_ACCT_UNBOUND != (int) IO_WQ_UNBOUND); 1344 BUILD_BUG_ON((int) IO_WQ_ACCT_NR != 2); 1345 1346 for (i = 0; i < IO_WQ_ACCT_NR; i++) { 1347 if (new_count[i] > task_rlimit(current, RLIMIT_NPROC)) 1348 new_count[i] = task_rlimit(current, RLIMIT_NPROC); 1349 } 1350 1351 for (i = 0; i < IO_WQ_ACCT_NR; i++) 1352 prev[i] = 0; 1353 1354 rcu_read_lock(); 1355 1356 raw_spin_lock(&wq->lock); 1357 for (i = 0; i < IO_WQ_ACCT_NR; i++) { 1358 acct = &wq->acct[i]; 1359 prev[i] = max_t(int, acct->max_workers, prev[i]); 1360 if (new_count[i]) 1361 acct->max_workers = new_count[i]; 1362 } 1363 raw_spin_unlock(&wq->lock); 1364 rcu_read_unlock(); 1365 1366 for (i = 0; i < IO_WQ_ACCT_NR; i++) 1367 new_count[i] = prev[i]; 1368 1369 return 0; 1370} 1371 1372static __init int io_wq_init(void) 1373{ 1374 int ret; 1375 1376 ret = cpuhp_setup_state_multi(CPUHP_AP_ONLINE_DYN, "io-wq/online", 1377 io_wq_cpu_online, io_wq_cpu_offline); 1378 if (ret < 0) 1379 return ret; 1380 io_wq_online = ret; 1381 return 0; 1382} 1383subsys_initcall(io_wq_init); 1384