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