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 27 enum { 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 34 enum { 35 IO_WQ_BIT_EXIT = 0, /* wq exiting */ 36 }; 37 38 enum { 39 IO_ACCT_STALLED_BIT = 0, /* stalled on hash */ 40 }; 41 42 /* 43 * One for each thread in a wq pool 44 */ 45 struct 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 77 struct 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 87 enum { 88 IO_WQ_ACCT_BOUND, 89 IO_WQ_ACCT_UNBOUND, 90 IO_WQ_ACCT_NR, 91 }; 92 93 /* 94 * Per io_wq state 95 */ 96 struct 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 126 static enum cpuhp_state io_wq_online; 127 128 struct 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 136 static bool create_io_worker(struct io_wq *wq, int index); 137 static void io_wq_dec_running(struct io_worker *worker); 138 static bool io_acct_cancel_pending_work(struct io_wq *wq, 139 struct io_wq_acct *acct, 140 struct io_cb_cancel_data *match); 141 static void create_worker_cb(struct callback_head *cb); 142 static void io_wq_cancel_tw_create(struct io_wq *wq); 143 144 static bool io_worker_get(struct io_worker *worker) 145 { 146 return refcount_inc_not_zero(&worker->ref); 147 } 148 149 static void io_worker_release(struct io_worker *worker) 150 { 151 if (refcount_dec_and_test(&worker->ref)) 152 complete(&worker->ref_done); 153 } 154 155 static 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 160 static 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 166 static 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 171 static 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 177 bool 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 187 static 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 201 static 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 211 static 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 245 static 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 */ 255 static 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 */ 270 static 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 */ 306 static 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 327 static 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 334 static 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 362 static 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); 400 fail_release: 401 io_worker_release(worker); 402 fail: 403 atomic_dec(&acct->nr_running); 404 io_worker_ref_put(wq); 405 return false; 406 } 407 408 static 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 */ 431 static 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 */ 444 static 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 453 static inline unsigned int io_get_work_hash(struct io_wq_work *work) 454 { 455 return work->flags >> IO_WQ_HASH_SHIFT; 456 } 457 458 static 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 475 static 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 532 static 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 */ 549 static 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 626 static 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 */ 692 void 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 */ 710 void 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 725 static 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 740 static bool io_wq_work_match_all(struct io_wq_work *work, void *data) 741 { 742 return true; 743 } 744 745 static 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 765 static 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 807 static 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 816 static 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) { 826 fail: 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 */ 861 static 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 882 static 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 889 static 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 898 static 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)) { 905 append: 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 919 static bool io_wq_work_match_item(struct io_wq_work *work, void *data) 920 { 921 return work == data; 922 } 923 924 void 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 */ 978 void 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 986 static 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 999 static 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 1016 static 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 1035 static 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 1059 static void io_wq_cancel_pending_work(struct io_wq *wq, 1060 struct io_cb_cancel_data *match) 1061 { 1062 int i; 1063 retry: 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 1075 static 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 1083 enum 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 1122 static 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 1141 struct 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 ret = cpuhp_state_add_instance_nocalls(io_wq_online, &wq->cpuhp_node); 1155 if (ret) 1156 goto err_wq; 1157 1158 refcount_inc(&data->hash->refs); 1159 wq->hash = data->hash; 1160 wq->free_work = data->free_work; 1161 wq->do_work = data->do_work; 1162 1163 ret = -ENOMEM; 1164 1165 if (!alloc_cpumask_var(&wq->cpu_mask, GFP_KERNEL)) 1166 goto err; 1167 cpumask_copy(wq->cpu_mask, cpu_possible_mask); 1168 wq->acct[IO_WQ_ACCT_BOUND].max_workers = bounded; 1169 wq->acct[IO_WQ_ACCT_UNBOUND].max_workers = 1170 task_rlimit(current, RLIMIT_NPROC); 1171 INIT_LIST_HEAD(&wq->wait.entry); 1172 wq->wait.func = io_wq_hash_wake; 1173 for (i = 0; i < IO_WQ_ACCT_NR; i++) { 1174 struct io_wq_acct *acct = &wq->acct[i]; 1175 1176 acct->index = i; 1177 atomic_set(&acct->nr_running, 0); 1178 INIT_WQ_LIST(&acct->work_list); 1179 raw_spin_lock_init(&acct->lock); 1180 } 1181 1182 raw_spin_lock_init(&wq->lock); 1183 INIT_HLIST_NULLS_HEAD(&wq->free_list, 0); 1184 INIT_LIST_HEAD(&wq->all_list); 1185 1186 wq->task = get_task_struct(data->task); 1187 atomic_set(&wq->worker_refs, 1); 1188 init_completion(&wq->worker_done); 1189 return wq; 1190 err: 1191 io_wq_put_hash(data->hash); 1192 cpuhp_state_remove_instance_nocalls(io_wq_online, &wq->cpuhp_node); 1193 1194 free_cpumask_var(wq->cpu_mask); 1195 err_wq: 1196 kfree(wq); 1197 return ERR_PTR(ret); 1198 } 1199 1200 static bool io_task_work_match(struct callback_head *cb, void *data) 1201 { 1202 struct io_worker *worker; 1203 1204 if (cb->func != create_worker_cb && cb->func != create_worker_cont) 1205 return false; 1206 worker = container_of(cb, struct io_worker, create_work); 1207 return worker->wq == data; 1208 } 1209 1210 void io_wq_exit_start(struct io_wq *wq) 1211 { 1212 set_bit(IO_WQ_BIT_EXIT, &wq->state); 1213 } 1214 1215 static void io_wq_cancel_tw_create(struct io_wq *wq) 1216 { 1217 struct callback_head *cb; 1218 1219 while ((cb = task_work_cancel_match(wq->task, io_task_work_match, wq)) != NULL) { 1220 struct io_worker *worker; 1221 1222 worker = container_of(cb, struct io_worker, create_work); 1223 io_worker_cancel_cb(worker); 1224 /* 1225 * Only the worker continuation helper has worker allocated and 1226 * hence needs freeing. 1227 */ 1228 if (cb->func == create_worker_cont) 1229 kfree(worker); 1230 } 1231 } 1232 1233 static void io_wq_exit_workers(struct io_wq *wq) 1234 { 1235 if (!wq->task) 1236 return; 1237 1238 io_wq_cancel_tw_create(wq); 1239 1240 rcu_read_lock(); 1241 io_wq_for_each_worker(wq, io_wq_worker_wake, NULL); 1242 rcu_read_unlock(); 1243 io_worker_ref_put(wq); 1244 wait_for_completion(&wq->worker_done); 1245 1246 spin_lock_irq(&wq->hash->wait.lock); 1247 list_del_init(&wq->wait.entry); 1248 spin_unlock_irq(&wq->hash->wait.lock); 1249 1250 put_task_struct(wq->task); 1251 wq->task = NULL; 1252 } 1253 1254 static void io_wq_destroy(struct io_wq *wq) 1255 { 1256 struct io_cb_cancel_data match = { 1257 .fn = io_wq_work_match_all, 1258 .cancel_all = true, 1259 }; 1260 1261 cpuhp_state_remove_instance_nocalls(io_wq_online, &wq->cpuhp_node); 1262 io_wq_cancel_pending_work(wq, &match); 1263 free_cpumask_var(wq->cpu_mask); 1264 io_wq_put_hash(wq->hash); 1265 kfree(wq); 1266 } 1267 1268 void io_wq_put_and_exit(struct io_wq *wq) 1269 { 1270 WARN_ON_ONCE(!test_bit(IO_WQ_BIT_EXIT, &wq->state)); 1271 1272 io_wq_exit_workers(wq); 1273 io_wq_destroy(wq); 1274 } 1275 1276 struct online_data { 1277 unsigned int cpu; 1278 bool online; 1279 }; 1280 1281 static bool io_wq_worker_affinity(struct io_worker *worker, void *data) 1282 { 1283 struct online_data *od = data; 1284 1285 if (od->online) 1286 cpumask_set_cpu(od->cpu, worker->wq->cpu_mask); 1287 else 1288 cpumask_clear_cpu(od->cpu, worker->wq->cpu_mask); 1289 return false; 1290 } 1291 1292 static int __io_wq_cpu_online(struct io_wq *wq, unsigned int cpu, bool online) 1293 { 1294 struct online_data od = { 1295 .cpu = cpu, 1296 .online = online 1297 }; 1298 1299 rcu_read_lock(); 1300 io_wq_for_each_worker(wq, io_wq_worker_affinity, &od); 1301 rcu_read_unlock(); 1302 return 0; 1303 } 1304 1305 static int io_wq_cpu_online(unsigned int cpu, struct hlist_node *node) 1306 { 1307 struct io_wq *wq = hlist_entry_safe(node, struct io_wq, cpuhp_node); 1308 1309 return __io_wq_cpu_online(wq, cpu, true); 1310 } 1311 1312 static int io_wq_cpu_offline(unsigned int cpu, struct hlist_node *node) 1313 { 1314 struct io_wq *wq = hlist_entry_safe(node, struct io_wq, cpuhp_node); 1315 1316 return __io_wq_cpu_online(wq, cpu, false); 1317 } 1318 1319 int io_wq_cpu_affinity(struct io_uring_task *tctx, cpumask_var_t mask) 1320 { 1321 if (!tctx || !tctx->io_wq) 1322 return -EINVAL; 1323 1324 rcu_read_lock(); 1325 if (mask) 1326 cpumask_copy(tctx->io_wq->cpu_mask, mask); 1327 else 1328 cpumask_copy(tctx->io_wq->cpu_mask, cpu_possible_mask); 1329 rcu_read_unlock(); 1330 1331 return 0; 1332 } 1333 1334 /* 1335 * Set max number of unbounded workers, returns old value. If new_count is 0, 1336 * then just return the old value. 1337 */ 1338 int io_wq_max_workers(struct io_wq *wq, int *new_count) 1339 { 1340 struct io_wq_acct *acct; 1341 int prev[IO_WQ_ACCT_NR]; 1342 int i; 1343 1344 BUILD_BUG_ON((int) IO_WQ_ACCT_BOUND != (int) IO_WQ_BOUND); 1345 BUILD_BUG_ON((int) IO_WQ_ACCT_UNBOUND != (int) IO_WQ_UNBOUND); 1346 BUILD_BUG_ON((int) IO_WQ_ACCT_NR != 2); 1347 1348 for (i = 0; i < IO_WQ_ACCT_NR; i++) { 1349 if (new_count[i] > task_rlimit(current, RLIMIT_NPROC)) 1350 new_count[i] = task_rlimit(current, RLIMIT_NPROC); 1351 } 1352 1353 for (i = 0; i < IO_WQ_ACCT_NR; i++) 1354 prev[i] = 0; 1355 1356 rcu_read_lock(); 1357 1358 raw_spin_lock(&wq->lock); 1359 for (i = 0; i < IO_WQ_ACCT_NR; i++) { 1360 acct = &wq->acct[i]; 1361 prev[i] = max_t(int, acct->max_workers, prev[i]); 1362 if (new_count[i]) 1363 acct->max_workers = new_count[i]; 1364 } 1365 raw_spin_unlock(&wq->lock); 1366 rcu_read_unlock(); 1367 1368 for (i = 0; i < IO_WQ_ACCT_NR; i++) 1369 new_count[i] = prev[i]; 1370 1371 return 0; 1372 } 1373 1374 static __init int io_wq_init(void) 1375 { 1376 int ret; 1377 1378 ret = cpuhp_setup_state_multi(CPUHP_AP_ONLINE_DYN, "io-wq/online", 1379 io_wq_cpu_online, io_wq_cpu_offline); 1380 if (ret < 0) 1381 return ret; 1382 io_wq_online = ret; 1383 return 0; 1384 } 1385 subsys_initcall(io_wq_init); 1386