fs/io-wq.c

   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/mm.h>
  13 #include <linux/mmu_context.h>
  14 #include <linux/sched/mm.h>
  15 #include <linux/percpu.h>
  16 #include <linux/slab.h>
  17 #include <linux/kthread.h>
  18 #include <linux/rculist_nulls.h>
  19
  20 #include "io-wq.h"
  21
  22 #define WORKER_IDLE_TIMEOUT     (5 * HZ)
  23
  24 enum {
  25         IO_WORKER_F_UP          = 1,    /* up and active */
  26         IO_WORKER_F_RUNNING     = 2,    /* account as running */
  27         IO_WORKER_F_FREE        = 4,    /* worker on free list */
  28         IO_WORKER_F_EXITING     = 8,    /* worker exiting */
  29         IO_WORKER_F_FIXED       = 16,   /* static idle worker */
  30         IO_WORKER_F_BOUND       = 32,   /* is doing bounded work */
  31 };
  32
  33 enum {
  34         IO_WQ_BIT_EXIT          = 0,    /* wq exiting */
  35         IO_WQ_BIT_CANCEL        = 1,    /* cancel work on list */
  36         IO_WQ_BIT_ERROR         = 2,    /* error on setup */
  37 };
  38
  39 enum {
  40         IO_WQE_FLAG_STALLED     = 1,    /* stalled on hash */
  41 };
  42
  43 /*
  44  * One for each thread in a wqe pool
  45  */
  46 struct io_worker {
  47         refcount_t ref;
  48         unsigned flags;
  49         struct hlist_nulls_node nulls_node;
  50         struct list_head all_list;
  51         struct task_struct *task;
  52         struct io_wqe *wqe;
  53
  54         struct io_wq_work *cur_work;
  55         spinlock_t lock;
  56
  57         struct rcu_head rcu;
  58         struct mm_struct *mm;
  59         const struct cred *cur_creds;
  60         const struct cred *saved_creds;
  61         struct files_struct *restore_files;
  62 };
  63
  64 #if BITS_PER_LONG == 64
  65 #define IO_WQ_HASH_ORDER        6
  66 #else
  67 #define IO_WQ_HASH_ORDER        5
  68 #endif
  69
  70 struct io_wqe_acct {
  71         unsigned nr_workers;
  72         unsigned max_workers;
  73         atomic_t nr_running;
  74 };
  75
  76 enum {
  77         IO_WQ_ACCT_BOUND,
  78         IO_WQ_ACCT_UNBOUND,
  79 };
  80
  81 /*
  82  * Per-node worker thread pool
  83  */
  84 struct io_wqe {
  85         struct {
  86                 spinlock_t lock;
  87                 struct io_wq_work_list work_list;
  88                 unsigned long hash_map;
  89                 unsigned flags;
  90         } ____cacheline_aligned_in_smp;
  91
  92         int node;
  93         struct io_wqe_acct acct[2];
  94
  95         struct hlist_nulls_head free_list;
  96         struct list_head all_list;
  97
  98         struct io_wq *wq;
  99 };
 100
 101 /*
 102  * Per io_wq state
 103   */
 104 struct io_wq {
 105         struct io_wqe **wqes;
 106         unsigned long state;
 107
 108         get_work_fn *get_work;
 109         put_work_fn *put_work;
 110
 111         struct task_struct *manager;
 112         struct user_struct *user;
 113         refcount_t refs;
 114         struct completion done;
 115
 116         refcount_t use_refs;
 117 };
 118
 119 static bool io_worker_get(struct io_worker *worker)
 120 {
 121         return refcount_inc_not_zero(&worker->ref);
 122 }
 123
 124 static void io_worker_release(struct io_worker *worker)
 125 {
 126         if (refcount_dec_and_test(&worker->ref))
 127                 wake_up_process(worker->task);
 128 }
 129
 130 /*
 131  * Note: drops the wqe->lock if returning true! The caller must re-acquire
 132  * the lock in that case. Some callers need to restart handling if this
 133  * happens, so we can't just re-acquire the lock on behalf of the caller.
 134  */
 135 static bool __io_worker_unuse(struct io_wqe *wqe, struct io_worker *worker)
 136 {
 137         bool dropped_lock = false;
 138
 139         if (worker->saved_creds) {
 140                 revert_creds(worker->saved_creds);
 141                 worker->cur_creds = worker->saved_creds = NULL;
 142         }
 143
 144         if (current->files != worker->restore_files) {
 145                 __acquire(&wqe->lock);
 146                 spin_unlock_irq(&wqe->lock);
 147                 dropped_lock = true;
 148
 149                 task_lock(current);
 150                 current->files = worker->restore_files;
 151                 task_unlock(current);
 152         }
 153
 154         /*
 155          * If we have an active mm, we need to drop the wq lock before unusing
 156          * it. If we do, return true and let the caller retry the idle loop.
 157          */
 158         if (worker->mm) {
 159                 if (!dropped_lock) {
 160                         __acquire(&wqe->lock);
 161                         spin_unlock_irq(&wqe->lock);
 162                         dropped_lock = true;
 163                 }
 164                 __set_current_state(TASK_RUNNING);
 165                 set_fs(KERNEL_DS);
 166                 unuse_mm(worker->mm);
 167                 mmput(worker->mm);
 168                 worker->mm = NULL;
 169         }
 170
 171         return dropped_lock;
 172 }
 173
 174 static inline struct io_wqe_acct *io_work_get_acct(struct io_wqe *wqe,
 175                                                    struct io_wq_work *work)
 176 {
 177         if (work->flags & IO_WQ_WORK_UNBOUND)
 178                 return &wqe->acct[IO_WQ_ACCT_UNBOUND];
 179
 180         return &wqe->acct[IO_WQ_ACCT_BOUND];
 181 }
 182
 183 static inline struct io_wqe_acct *io_wqe_get_acct(struct io_wqe *wqe,
 184                                                   struct io_worker *worker)
 185 {
 186         if (worker->flags & IO_WORKER_F_BOUND)
 187                 return &wqe->acct[IO_WQ_ACCT_BOUND];
 188
 189         return &wqe->acct[IO_WQ_ACCT_UNBOUND];
 190 }
 191
 192 static void io_worker_exit(struct io_worker *worker)
 193 {
 194         struct io_wqe *wqe = worker->wqe;
 195         struct io_wqe_acct *acct = io_wqe_get_acct(wqe, worker);
 196         unsigned nr_workers;
 197
 198         /*
 199          * If we're not at zero, someone else is holding a brief reference
 200          * to the worker. Wait for that to go away.
 201          */
 202         set_current_state(TASK_INTERRUPTIBLE);
 203         if (!refcount_dec_and_test(&worker->ref))
 204                 schedule();
 205         __set_current_state(TASK_RUNNING);
 206
 207         preempt_disable();
 208         current->flags &= ~PF_IO_WORKER;
 209         if (worker->flags & IO_WORKER_F_RUNNING)
 210                 atomic_dec(&acct->nr_running);
 211         if (!(worker->flags & IO_WORKER_F_BOUND))
 212                 atomic_dec(&wqe->wq->user->processes);
 213         worker->flags = 0;
 214         preempt_enable();
 215
 216         spin_lock_irq(&wqe->lock);
 217         hlist_nulls_del_rcu(&worker->nulls_node);
 218         list_del_rcu(&worker->all_list);
 219         if (__io_worker_unuse(wqe, worker)) {
 220                 __release(&wqe->lock);
 221                 spin_lock_irq(&wqe->lock);
 222         }
 223         acct->nr_workers--;
 224         nr_workers = wqe->acct[IO_WQ_ACCT_BOUND].nr_workers +
 225                         wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers;
 226         spin_unlock_irq(&wqe->lock);
 227
 228         /* all workers gone, wq exit can proceed */
 229         if (!nr_workers && refcount_dec_and_test(&wqe->wq->refs))
 230                 complete(&wqe->wq->done);
 231
 232         kfree_rcu(worker, rcu);
 233 }
 234
 235 static inline bool io_wqe_run_queue(struct io_wqe *wqe)
 236         __must_hold(wqe->lock)
 237 {
 238         if (!wq_list_empty(&wqe->work_list) &&
 239             !(wqe->flags & IO_WQE_FLAG_STALLED))
 240                 return true;
 241         return false;
 242 }
 243
 244 /*
 245  * Check head of free list for an available worker. If one isn't available,
 246  * caller must wake up the wq manager to create one.
 247  */
 248 static bool io_wqe_activate_free_worker(struct io_wqe *wqe)
 249         __must_hold(RCU)
 250 {
 251         struct hlist_nulls_node *n;
 252         struct io_worker *worker;
 253
 254         n = rcu_dereference(hlist_nulls_first_rcu(&wqe->free_list));
 255         if (is_a_nulls(n))
 256                 return false;
 257
 258         worker = hlist_nulls_entry(n, struct io_worker, nulls_node);
 259         if (io_worker_get(worker)) {
 260                 wake_up_process(worker->task);
 261                 io_worker_release(worker);
 262                 return true;
 263         }
 264
 265         return false;
 266 }
 267
 268 /*
 269  * We need a worker. If we find a free one, we're good. If not, and we're
 270  * below the max number of workers, wake up the manager to create one.
 271  */
 272 static void io_wqe_wake_worker(struct io_wqe *wqe, struct io_wqe_acct *acct)
 273 {
 274         bool ret;
 275
 276         /*
 277          * Most likely an attempt to queue unbounded work on an io_wq that
 278          * wasn't setup with any unbounded workers.
 279          */
 280         WARN_ON_ONCE(!acct->max_workers);
 281
 282         rcu_read_lock();
 283         ret = io_wqe_activate_free_worker(wqe);
 284         rcu_read_unlock();
 285
 286         if (!ret && acct->nr_workers < acct->max_workers)
 287                 wake_up_process(wqe->wq->manager);
 288 }
 289
 290 static void io_wqe_inc_running(struct io_wqe *wqe, struct io_worker *worker)
 291 {
 292         struct io_wqe_acct *acct = io_wqe_get_acct(wqe, worker);
 293
 294         atomic_inc(&acct->nr_running);
 295 }
 296
 297 static void io_wqe_dec_running(struct io_wqe *wqe, struct io_worker *worker)
 298         __must_hold(wqe->lock)
 299 {
 300         struct io_wqe_acct *acct = io_wqe_get_acct(wqe, worker);
 301
 302         if (atomic_dec_and_test(&acct->nr_running) && io_wqe_run_queue(wqe))
 303                 io_wqe_wake_worker(wqe, acct);
 304 }
 305
 306 static void io_worker_start(struct io_wqe *wqe, struct io_worker *worker)
 307 {
 308         allow_kernel_signal(SIGINT);
 309
 310         current->flags |= PF_IO_WORKER;
 311
 312         worker->flags |= (IO_WORKER_F_UP | IO_WORKER_F_RUNNING);
 313         worker->restore_files = current->files;
 314         io_wqe_inc_running(wqe, worker);
 315 }
 316
 317 /*
 318  * Worker will start processing some work. Move it to the busy list, if
 319  * it's currently on the freelist
 320  */
 321 static void __io_worker_busy(struct io_wqe *wqe, struct io_worker *worker,
 322                              struct io_wq_work *work)
 323         __must_hold(wqe->lock)
 324 {
 325         bool worker_bound, work_bound;
 326
 327         if (worker->flags & IO_WORKER_F_FREE) {
 328                 worker->flags &= ~IO_WORKER_F_FREE;
 329                 hlist_nulls_del_init_rcu(&worker->nulls_node);
 330         }
 331
 332         /*
 333          * If worker is moving from bound to unbound (or vice versa), then
 334          * ensure we update the running accounting.
 335          */
 336         worker_bound = (worker->flags & IO_WORKER_F_BOUND) != 0;
 337         work_bound = (work->flags & IO_WQ_WORK_UNBOUND) == 0;
 338         if (worker_bound != work_bound) {
 339                 io_wqe_dec_running(wqe, worker);
 340                 if (work_bound) {
 341                         worker->flags |= IO_WORKER_F_BOUND;
 342                         wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers--;
 343                         wqe->acct[IO_WQ_ACCT_BOUND].nr_workers++;
 344                         atomic_dec(&wqe->wq->user->processes);
 345                 } else {
 346                         worker->flags &= ~IO_WORKER_F_BOUND;
 347                         wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers++;
 348                         wqe->acct[IO_WQ_ACCT_BOUND].nr_workers--;
 349                         atomic_inc(&wqe->wq->user->processes);
 350                 }
 351                 io_wqe_inc_running(wqe, worker);
 352          }
 353 }
 354
 355 /*
 356  * No work, worker going to sleep. Move to freelist, and unuse mm if we
 357  * have one attached. Dropping the mm may potentially sleep, so we drop
 358  * the lock in that case and return success. Since the caller has to
 359  * retry the loop in that case (we changed task state), we don't regrab
 360  * the lock if we return success.
 361  */
 362 static bool __io_worker_idle(struct io_wqe *wqe, struct io_worker *worker)
 363         __must_hold(wqe->lock)
 364 {
 365         if (!(worker->flags & IO_WORKER_F_FREE)) {
 366                 worker->flags |= IO_WORKER_F_FREE;
 367                 hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);
 368         }
 369
 370         return __io_worker_unuse(wqe, worker);
 371 }
 372
 373 static struct io_wq_work *io_get_next_work(struct io_wqe *wqe, unsigned *hash)
 374         __must_hold(wqe->lock)
 375 {
 376         struct io_wq_work_node *node, *prev;
 377         struct io_wq_work *work;
 378
 379         wq_list_for_each(node, prev, &wqe->work_list) {
 380                 work = container_of(node, struct io_wq_work, list);
 381
 382                 /* not hashed, can run anytime */
 383                 if (!(work->flags & IO_WQ_WORK_HASHED)) {
 384                         wq_node_del(&wqe->work_list, node, prev);
 385                         return work;
 386                 }
 387
 388                 /* hashed, can run if not already running */
 389                 *hash = work->flags >> IO_WQ_HASH_SHIFT;
 390                 if (!(wqe->hash_map & BIT_ULL(*hash))) {
 391                         wqe->hash_map |= BIT_ULL(*hash);
 392                         wq_node_del(&wqe->work_list, node, prev);
 393                         return work;
 394                 }
 395         }
 396
 397         return NULL;
 398 }
 399
 400 static void io_wq_switch_mm(struct io_worker *worker, struct io_wq_work *work)
 401 {
 402         if (worker->mm) {
 403                 unuse_mm(worker->mm);
 404                 mmput(worker->mm);
 405                 worker->mm = NULL;
 406         }
 407         if (!work->mm) {
 408                 set_fs(KERNEL_DS);
 409                 return;
 410         }
 411         if (mmget_not_zero(work->mm)) {
 412                 use_mm(work->mm);
 413                 if (!worker->mm)
 414                         set_fs(USER_DS);
 415                 worker->mm = work->mm;
 416                 /* hang on to this mm */
 417                 work->mm = NULL;
 418                 return;
 419         }
 420
 421         /* failed grabbing mm, ensure work gets cancelled */
 422         work->flags |= IO_WQ_WORK_CANCEL;
 423 }
 424
 425 static void io_wq_switch_creds(struct io_worker *worker,
 426                                struct io_wq_work *work)
 427 {
 428         const struct cred *old_creds = override_creds(work->creds);
 429
 430         worker->cur_creds = work->creds;
 431         if (worker->saved_creds)
 432                 put_cred(old_creds); /* creds set by previous switch */
 433         else
 434                 worker->saved_creds = old_creds;
 435 }
 436
 437 static void io_worker_handle_work(struct io_worker *worker)
 438         __releases(wqe->lock)
 439 {
 440         struct io_wq_work *work, *old_work = NULL, *put_work = NULL;
 441         struct io_wqe *wqe = worker->wqe;
 442         struct io_wq *wq = wqe->wq;
 443
 444         do {
 445                 unsigned hash = -1U;
 446
 447                 /*
 448                  * If we got some work, mark us as busy. If we didn't, but
 449                  * the list isn't empty, it means we stalled on hashed work.
 450                  * Mark us stalled so we don't keep looking for work when we
 451                  * can't make progress, any work completion or insertion will
 452                  * clear the stalled flag.
 453                  */
 454                 work = io_get_next_work(wqe, &hash);
 455                 if (work)
 456                         __io_worker_busy(wqe, worker, work);
 457                 else if (!wq_list_empty(&wqe->work_list))
 458                         wqe->flags |= IO_WQE_FLAG_STALLED;
 459
 460                 spin_unlock_irq(&wqe->lock);
 461                 if (put_work && wq->put_work)
 462                         wq->put_work(old_work);
 463                 if (!work)
 464                         break;
 465 next:
 466                 /* flush any pending signals before assigning new work */
 467                 if (signal_pending(current))
 468                         flush_signals(current);
 469
 470                 cond_resched();
 471
 472                 spin_lock_irq(&worker->lock);
 473                 worker->cur_work = work;
 474                 spin_unlock_irq(&worker->lock);
 475
 476                 if (work->flags & IO_WQ_WORK_CB)
 477                         work->func(&work);
 478
 479                 if ((work->flags & IO_WQ_WORK_NEEDS_FILES) &&
 480                     current->files != work->files) {
 481                         task_lock(current);
 482                         current->files = work->files;
 483                         task_unlock(current);
 484                 }
 485                 if (work->mm != worker->mm)
 486                         io_wq_switch_mm(worker, work);
 487                 if (worker->cur_creds != work->creds)
 488                         io_wq_switch_creds(worker, work);
 489                 /*
 490                  * OK to set IO_WQ_WORK_CANCEL even for uncancellable work,
 491                  * the worker function will do the right thing.
 492                  */
 493                 if (test_bit(IO_WQ_BIT_CANCEL, &wq->state))
 494                         work->flags |= IO_WQ_WORK_CANCEL;
 495                 if (worker->mm)
 496                         work->flags |= IO_WQ_WORK_HAS_MM;
 497
 498                 if (wq->get_work && !(work->flags & IO_WQ_WORK_INTERNAL)) {
 499                         put_work = work;
 500                         wq->get_work(work);
 501                 }
 502
 503                 old_work = work;
 504                 work->func(&work);
 505
 506                 spin_lock_irq(&worker->lock);
 507                 worker->cur_work = NULL;
 508                 spin_unlock_irq(&worker->lock);
 509
 510                 spin_lock_irq(&wqe->lock);
 511
 512                 if (hash != -1U) {
 513                         wqe->hash_map &= ~BIT_ULL(hash);
 514                         wqe->flags &= ~IO_WQE_FLAG_STALLED;
 515                 }
 516                 if (work && work != old_work) {
 517                         spin_unlock_irq(&wqe->lock);
 518
 519                         if (put_work && wq->put_work) {
 520                                 wq->put_work(put_work);
 521                                 put_work = NULL;
 522                         }
 523
 524                         /* dependent work not hashed */
 525                         hash = -1U;
 526                         goto next;
 527                 }
 528         } while (1);
 529 }
 530
 531 static inline void io_worker_spin_for_work(struct io_wqe *wqe)
 532 {
 533         int i = 0;
 534
 535         while (++i < 1000) {
 536                 if (io_wqe_run_queue(wqe))
 537                         break;
 538                 if (need_resched())
 539                         break;
 540                 cpu_relax();
 541         }
 542 }
 543
 544 static int io_wqe_worker(void *data)
 545 {
 546         struct io_worker *worker = data;
 547         struct io_wqe *wqe = worker->wqe;
 548         struct io_wq *wq = wqe->wq;
 549         bool did_work;
 550
 551         io_worker_start(wqe, worker);
 552
 553         did_work = false;
 554         while (!test_bit(IO_WQ_BIT_EXIT, &wq->state)) {
 555                 set_current_state(TASK_INTERRUPTIBLE);
 556 loop:
 557                 if (did_work)
 558                         io_worker_spin_for_work(wqe);
 559                 spin_lock_irq(&wqe->lock);
 560                 if (io_wqe_run_queue(wqe)) {
 561                         __set_current_state(TASK_RUNNING);
 562                         io_worker_handle_work(worker);
 563                         did_work = true;
 564                         goto loop;
 565                 }
 566                 did_work = false;
 567                 /* drops the lock on success, retry */
 568                 if (__io_worker_idle(wqe, worker)) {
 569                         __release(&wqe->lock);
 570                         goto loop;
 571                 }
 572                 spin_unlock_irq(&wqe->lock);
 573                 if (signal_pending(current))
 574                         flush_signals(current);
 575                 if (schedule_timeout(WORKER_IDLE_TIMEOUT))
 576                         continue;
 577                 /* timed out, exit unless we're the fixed worker */
 578                 if (test_bit(IO_WQ_BIT_EXIT, &wq->state) ||
 579                     !(worker->flags & IO_WORKER_F_FIXED))
 580                         break;
 581         }
 582
 583         if (test_bit(IO_WQ_BIT_EXIT, &wq->state)) {
 584                 spin_lock_irq(&wqe->lock);
 585                 if (!wq_list_empty(&wqe->work_list))
 586                         io_worker_handle_work(worker);
 587                 else
 588                         spin_unlock_irq(&wqe->lock);
 589         }
 590
 591         io_worker_exit(worker);
 592         return 0;
 593 }
 594
 595 /*
 596  * Called when a worker is scheduled in. Mark us as currently running.
 597  */
 598 void io_wq_worker_running(struct task_struct *tsk)
 599 {
 600         struct io_worker *worker = kthread_data(tsk);
 601         struct io_wqe *wqe = worker->wqe;
 602
 603         if (!(worker->flags & IO_WORKER_F_UP))
 604                 return;
 605         if (worker->flags & IO_WORKER_F_RUNNING)
 606                 return;
 607         worker->flags |= IO_WORKER_F_RUNNING;
 608         io_wqe_inc_running(wqe, worker);
 609 }
 610
 611 /*
 612  * Called when worker is going to sleep. If there are no workers currently
 613  * running and we have work pending, wake up a free one or have the manager
 614  * set one up.
 615  */
 616 void io_wq_worker_sleeping(struct task_struct *tsk)
 617 {
 618         struct io_worker *worker = kthread_data(tsk);
 619         struct io_wqe *wqe = worker->wqe;
 620
 621         if (!(worker->flags & IO_WORKER_F_UP))
 622                 return;
 623         if (!(worker->flags & IO_WORKER_F_RUNNING))
 624                 return;
 625
 626         worker->flags &= ~IO_WORKER_F_RUNNING;
 627
 628         spin_lock_irq(&wqe->lock);
 629         io_wqe_dec_running(wqe, worker);
 630         spin_unlock_irq(&wqe->lock);
 631 }
 632
 633 static bool create_io_worker(struct io_wq *wq, struct io_wqe *wqe, int index)
 634 {
 635         struct io_wqe_acct *acct =&wqe->acct[index];
 636         struct io_worker *worker;
 637
 638         worker = kzalloc_node(sizeof(*worker), GFP_KERNEL, wqe->node);
 639         if (!worker)
 640                 return false;
 641
 642         refcount_set(&worker->ref, 1);
 643         worker->nulls_node.pprev = NULL;
 644         worker->wqe = wqe;
 645         spin_lock_init(&worker->lock);
 646
 647         worker->task = kthread_create_on_node(io_wqe_worker, worker, wqe->node,
 648                                 "io_wqe_worker-%d/%d", index, wqe->node);
 649         if (IS_ERR(worker->task)) {
 650                 kfree(worker);
 651                 return false;
 652         }
 653
 654         spin_lock_irq(&wqe->lock);
 655         hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);
 656         list_add_tail_rcu(&worker->all_list, &wqe->all_list);
 657         worker->flags |= IO_WORKER_F_FREE;
 658         if (index == IO_WQ_ACCT_BOUND)
 659                 worker->flags |= IO_WORKER_F_BOUND;
 660         if (!acct->nr_workers && (worker->flags & IO_WORKER_F_BOUND))
 661                 worker->flags |= IO_WORKER_F_FIXED;
 662         acct->nr_workers++;
 663         spin_unlock_irq(&wqe->lock);
 664
 665         if (index == IO_WQ_ACCT_UNBOUND)
 666                 atomic_inc(&wq->user->processes);
 667
 668         wake_up_process(worker->task);
 669         return true;
 670 }
 671
 672 static inline bool io_wqe_need_worker(struct io_wqe *wqe, int index)
 673         __must_hold(wqe->lock)
 674 {
 675         struct io_wqe_acct *acct = &wqe->acct[index];
 676
 677         /* if we have available workers or no work, no need */
 678         if (!hlist_nulls_empty(&wqe->free_list) || !io_wqe_run_queue(wqe))
 679                 return false;
 680         return acct->nr_workers < acct->max_workers;
 681 }
 682
 683 /*
 684  * Manager thread. Tasked with creating new workers, if we need them.
 685  */
 686 static int io_wq_manager(void *data)
 687 {
 688         struct io_wq *wq = data;
 689         int workers_to_create = num_possible_nodes();
 690         int node;
 691
 692         /* create fixed workers */
 693         refcount_set(&wq->refs, workers_to_create);
 694         for_each_node(node) {
 695                 if (!create_io_worker(wq, wq->wqes[node], IO_WQ_ACCT_BOUND))
 696                         goto err;
 697                 workers_to_create--;
 698         }
 699
 700         complete(&wq->done);
 701
 702         while (!kthread_should_stop()) {
 703                 for_each_node(node) {
 704                         struct io_wqe *wqe = wq->wqes[node];
 705                         bool fork_worker[2] = { false, false };
 706
 707                         spin_lock_irq(&wqe->lock);
 708                         if (io_wqe_need_worker(wqe, IO_WQ_ACCT_BOUND))
 709                                 fork_worker[IO_WQ_ACCT_BOUND] = true;
 710                         if (io_wqe_need_worker(wqe, IO_WQ_ACCT_UNBOUND))
 711                                 fork_worker[IO_WQ_ACCT_UNBOUND] = true;
 712                         spin_unlock_irq(&wqe->lock);
 713                         if (fork_worker[IO_WQ_ACCT_BOUND])
 714                                 create_io_worker(wq, wqe, IO_WQ_ACCT_BOUND);
 715                         if (fork_worker[IO_WQ_ACCT_UNBOUND])
 716                                 create_io_worker(wq, wqe, IO_WQ_ACCT_UNBOUND);
 717                 }
 718                 set_current_state(TASK_INTERRUPTIBLE);
 719                 schedule_timeout(HZ);
 720         }
 721
 722         return 0;
 723 err:
 724         set_bit(IO_WQ_BIT_ERROR, &wq->state);
 725         set_bit(IO_WQ_BIT_EXIT, &wq->state);
 726         if (refcount_sub_and_test(workers_to_create, &wq->refs))
 727                 complete(&wq->done);
 728         return 0;
 729 }
 730
 731 static bool io_wq_can_queue(struct io_wqe *wqe, struct io_wqe_acct *acct,
 732                             struct io_wq_work *work)
 733 {
 734         bool free_worker;
 735
 736         if (!(work->flags & IO_WQ_WORK_UNBOUND))
 737                 return true;
 738         if (atomic_read(&acct->nr_running))
 739                 return true;
 740
 741         rcu_read_lock();
 742         free_worker = !hlist_nulls_empty(&wqe->free_list);
 743         rcu_read_unlock();
 744         if (free_worker)
 745                 return true;
 746
 747         if (atomic_read(&wqe->wq->user->processes) >= acct->max_workers &&
 748             !(capable(CAP_SYS_RESOURCE) || capable(CAP_SYS_ADMIN)))
 749                 return false;
 750
 751         return true;
 752 }
 753
 754 static void io_wqe_enqueue(struct io_wqe *wqe, struct io_wq_work *work)
 755 {
 756         struct io_wqe_acct *acct = io_work_get_acct(wqe, work);
 757         int work_flags;
 758         unsigned long flags;
 759
 760         /*
 761          * Do early check to see if we need a new unbound worker, and if we do,
 762          * if we're allowed to do so. This isn't 100% accurate as there's a
 763          * gap between this check and incrementing the value, but that's OK.
 764          * It's close enough to not be an issue, fork() has the same delay.
 765          */
 766         if (unlikely(!io_wq_can_queue(wqe, acct, work))) {
 767                 work->flags |= IO_WQ_WORK_CANCEL;
 768                 work->func(&work);
 769                 return;
 770         }
 771
 772         work_flags = work->flags;
 773         spin_lock_irqsave(&wqe->lock, flags);
 774         wq_list_add_tail(&work->list, &wqe->work_list);
 775         wqe->flags &= ~IO_WQE_FLAG_STALLED;
 776         spin_unlock_irqrestore(&wqe->lock, flags);
 777
 778         if ((work_flags & IO_WQ_WORK_CONCURRENT) ||
 779             !atomic_read(&acct->nr_running))
 780                 io_wqe_wake_worker(wqe, acct);
 781 }
 782
 783 void io_wq_enqueue(struct io_wq *wq, struct io_wq_work *work)
 784 {
 785         struct io_wqe *wqe = wq->wqes[numa_node_id()];
 786
 787         io_wqe_enqueue(wqe, work);
 788 }
 789
 790 /*
 791  * Enqueue work, hashed by some key. Work items that hash to the same value
 792  * will not be done in parallel. Used to limit concurrent writes, generally
 793  * hashed by inode.
 794  */
 795 void io_wq_enqueue_hashed(struct io_wq *wq, struct io_wq_work *work, void *val)
 796 {
 797         struct io_wqe *wqe = wq->wqes[numa_node_id()];
 798         unsigned bit;
 799
 800
 801         bit = hash_ptr(val, IO_WQ_HASH_ORDER);
 802         work->flags |= (IO_WQ_WORK_HASHED | (bit << IO_WQ_HASH_SHIFT));
 803         io_wqe_enqueue(wqe, work);
 804 }
 805
 806 static bool io_wqe_worker_send_sig(struct io_worker *worker, void *data)
 807 {
 808         send_sig(SIGINT, worker->task, 1);
 809         return false;
 810 }
 811
 812 /*
 813  * Iterate the passed in list and call the specific function for each
 814  * worker that isn't exiting
 815  */
 816 static bool io_wq_for_each_worker(struct io_wqe *wqe,
 817                                   bool (*func)(struct io_worker *, void *),
 818                                   void *data)
 819 {
 820         struct io_worker *worker;
 821         bool ret = false;
 822
 823         list_for_each_entry_rcu(worker, &wqe->all_list, all_list) {
 824                 if (io_worker_get(worker)) {
 825                         ret = func(worker, data);
 826                         io_worker_release(worker);
 827                         if (ret)
 828                                 break;
 829                 }
 830         }
 831
 832         return ret;
 833 }
 834
 835 void io_wq_cancel_all(struct io_wq *wq)
 836 {
 837         int node;
 838
 839         set_bit(IO_WQ_BIT_CANCEL, &wq->state);
 840
 841         rcu_read_lock();
 842         for_each_node(node) {
 843                 struct io_wqe *wqe = wq->wqes[node];
 844
 845                 io_wq_for_each_worker(wqe, io_wqe_worker_send_sig, NULL);
 846         }
 847         rcu_read_unlock();
 848 }
 849
 850 struct io_cb_cancel_data {
 851         struct io_wqe *wqe;
 852         work_cancel_fn *cancel;
 853         void *caller_data;
 854 };
 855
 856 static bool io_work_cancel(struct io_worker *worker, void *cancel_data)
 857 {
 858         struct io_cb_cancel_data *data = cancel_data;
 859         unsigned long flags;
 860         bool ret = false;
 861
 862         /*
 863          * Hold the lock to avoid ->cur_work going out of scope, caller
 864          * may dereference the passed in work.
 865          */
 866         spin_lock_irqsave(&worker->lock, flags);
 867         if (worker->cur_work &&
 868             !(worker->cur_work->flags & IO_WQ_WORK_NO_CANCEL) &&
 869             data->cancel(worker->cur_work, data->caller_data)) {
 870                 send_sig(SIGINT, worker->task, 1);
 871                 ret = true;
 872         }
 873         spin_unlock_irqrestore(&worker->lock, flags);
 874
 875         return ret;
 876 }
 877
 878 static enum io_wq_cancel io_wqe_cancel_cb_work(struct io_wqe *wqe,
 879                                                work_cancel_fn *cancel,
 880                                                void *cancel_data)
 881 {
 882         struct io_cb_cancel_data data = {
 883                 .wqe = wqe,
 884                 .cancel = cancel,
 885                 .caller_data = cancel_data,
 886         };
 887         struct io_wq_work_node *node, *prev;
 888         struct io_wq_work *work;
 889         unsigned long flags;
 890         bool found = false;
 891
 892         spin_lock_irqsave(&wqe->lock, flags);
 893         wq_list_for_each(node, prev, &wqe->work_list) {
 894                 work = container_of(node, struct io_wq_work, list);
 895
 896                 if (cancel(work, cancel_data)) {
 897                         wq_node_del(&wqe->work_list, node, prev);
 898                         found = true;
 899                         break;
 900                 }
 901         }
 902         spin_unlock_irqrestore(&wqe->lock, flags);
 903
 904         if (found) {
 905                 work->flags |= IO_WQ_WORK_CANCEL;
 906                 work->func(&work);
 907                 return IO_WQ_CANCEL_OK;
 908         }
 909
 910         rcu_read_lock();
 911         found = io_wq_for_each_worker(wqe, io_work_cancel, &data);
 912         rcu_read_unlock();
 913         return found ? IO_WQ_CANCEL_RUNNING : IO_WQ_CANCEL_NOTFOUND;
 914 }
 915
 916 enum io_wq_cancel io_wq_cancel_cb(struct io_wq *wq, work_cancel_fn *cancel,
 917                                   void *data)
 918 {
 919         enum io_wq_cancel ret = IO_WQ_CANCEL_NOTFOUND;
 920         int node;
 921
 922         for_each_node(node) {
 923                 struct io_wqe *wqe = wq->wqes[node];
 924
 925                 ret = io_wqe_cancel_cb_work(wqe, cancel, data);
 926                 if (ret != IO_WQ_CANCEL_NOTFOUND)
 927                         break;
 928         }
 929
 930         return ret;
 931 }
 932
 933 static bool io_wq_worker_cancel(struct io_worker *worker, void *data)
 934 {
 935         struct io_wq_work *work = data;
 936         unsigned long flags;
 937         bool ret = false;
 938
 939         if (worker->cur_work != work)
 940                 return false;
 941
 942         spin_lock_irqsave(&worker->lock, flags);
 943         if (worker->cur_work == work &&
 944             !(worker->cur_work->flags & IO_WQ_WORK_NO_CANCEL)) {
 945                 send_sig(SIGINT, worker->task, 1);
 946                 ret = true;
 947         }
 948         spin_unlock_irqrestore(&worker->lock, flags);
 949
 950         return ret;
 951 }
 952
 953 static enum io_wq_cancel io_wqe_cancel_work(struct io_wqe *wqe,
 954                                             struct io_wq_work *cwork)
 955 {
 956         struct io_wq_work_node *node, *prev;
 957         struct io_wq_work *work;
 958         unsigned long flags;
 959         bool found = false;
 960
 961         cwork->flags |= IO_WQ_WORK_CANCEL;
 962
 963         /*
 964          * First check pending list, if we're lucky we can just remove it
 965          * from there. CANCEL_OK means that the work is returned as-new,
 966          * no completion will be posted for it.
 967          */
 968         spin_lock_irqsave(&wqe->lock, flags);
 969         wq_list_for_each(node, prev, &wqe->work_list) {
 970                 work = container_of(node, struct io_wq_work, list);
 971
 972                 if (work == cwork) {
 973                         wq_node_del(&wqe->work_list, node, prev);
 974                         found = true;
 975                         break;
 976                 }
 977         }
 978         spin_unlock_irqrestore(&wqe->lock, flags);
 979
 980         if (found) {
 981                 work->flags |= IO_WQ_WORK_CANCEL;
 982                 work->func(&work);
 983                 return IO_WQ_CANCEL_OK;
 984         }
 985
 986         /*
 987          * Now check if a free (going busy) or busy worker has the work
 988          * currently running. If we find it there, we'll return CANCEL_RUNNING
 989          * as an indication that we attempt to signal cancellation. The
 990          * completion will run normally in this case.
 991          */
 992         rcu_read_lock();
 993         found = io_wq_for_each_worker(wqe, io_wq_worker_cancel, cwork);
 994         rcu_read_unlock();
 995         return found ? IO_WQ_CANCEL_RUNNING : IO_WQ_CANCEL_NOTFOUND;
 996 }
 997
 998 enum io_wq_cancel io_wq_cancel_work(struct io_wq *wq, struct io_wq_work *cwork)
 999 {
1000         enum io_wq_cancel ret = IO_WQ_CANCEL_NOTFOUND;
1001         int node;
1002
1003         for_each_node(node) {
1004                 struct io_wqe *wqe = wq->wqes[node];
1005
1006                 ret = io_wqe_cancel_work(wqe, cwork);
1007                 if (ret != IO_WQ_CANCEL_NOTFOUND)
1008                         break;
1009         }
1010
1011         return ret;
1012 }
1013
1014 struct io_wq_flush_data {
1015         struct io_wq_work work;
1016         struct completion done;
1017 };
1018
1019 static void io_wq_flush_func(struct io_wq_work **workptr)
1020 {
1021         struct io_wq_work *work = *workptr;
1022         struct io_wq_flush_data *data;
1023
1024         data = container_of(work, struct io_wq_flush_data, work);
1025         complete(&data->done);
1026 }
1027
1028 /*
1029  * Doesn't wait for previously queued work to finish. When this completes,
1030  * it just means that previously queued work was started.
1031  */
1032 void io_wq_flush(struct io_wq *wq)
1033 {
1034         struct io_wq_flush_data data;
1035         int node;
1036
1037         for_each_node(node) {
1038                 struct io_wqe *wqe = wq->wqes[node];
1039
1040                 init_completion(&data.done);
1041                 INIT_IO_WORK(&data.work, io_wq_flush_func);
1042                 data.work.flags |= IO_WQ_WORK_INTERNAL;
1043                 io_wqe_enqueue(wqe, &data.work);
1044                 wait_for_completion(&data.done);
1045         }
1046 }
1047
1048 struct io_wq *io_wq_create(unsigned bounded, struct io_wq_data *data)
1049 {
1050         int ret = -ENOMEM, node;
1051         struct io_wq *wq;
1052
1053         wq = kzalloc(sizeof(*wq), GFP_KERNEL);
1054         if (!wq)
1055                 return ERR_PTR(-ENOMEM);
1056
1057         wq->wqes = kcalloc(nr_node_ids, sizeof(struct io_wqe *), GFP_KERNEL);
1058         if (!wq->wqes) {
1059                 kfree(wq);
1060                 return ERR_PTR(-ENOMEM);
1061         }
1062
1063         wq->get_work = data->get_work;
1064         wq->put_work = data->put_work;
1065
1066         /* caller must already hold a reference to this */
1067         wq->user = data->user;
1068
1069         for_each_node(node) {
1070                 struct io_wqe *wqe;
1071
1072                 wqe = kzalloc_node(sizeof(struct io_wqe), GFP_KERNEL, node);
1073                 if (!wqe)
1074                         goto err;
1075                 wq->wqes[node] = wqe;
1076                 wqe->node = node;
1077                 wqe->acct[IO_WQ_ACCT_BOUND].max_workers = bounded;
1078                 atomic_set(&wqe->acct[IO_WQ_ACCT_BOUND].nr_running, 0);
1079                 if (wq->user) {
1080                         wqe->acct[IO_WQ_ACCT_UNBOUND].max_workers =
1081                                         task_rlimit(current, RLIMIT_NPROC);
1082                 }
1083                 atomic_set(&wqe->acct[IO_WQ_ACCT_UNBOUND].nr_running, 0);
1084                 wqe->node = node;
1085                 wqe->wq = wq;
1086                 spin_lock_init(&wqe->lock);
1087                 INIT_WQ_LIST(&wqe->work_list);
1088                 INIT_HLIST_NULLS_HEAD(&wqe->free_list, 0);
1089                 INIT_LIST_HEAD(&wqe->all_list);
1090         }
1091
1092         init_completion(&wq->done);
1093
1094         wq->manager = kthread_create(io_wq_manager, wq, "io_wq_manager");
1095         if (!IS_ERR(wq->manager)) {
1096                 wake_up_process(wq->manager);
1097                 wait_for_completion(&wq->done);
1098                 if (test_bit(IO_WQ_BIT_ERROR, &wq->state)) {
1099                         ret = -ENOMEM;
1100                         goto err;
1101                 }
1102                 refcount_set(&wq->use_refs, 1);
1103                 reinit_completion(&wq->done);
1104                 return wq;
1105         }
1106
1107         ret = PTR_ERR(wq->manager);
1108         complete(&wq->done);
1109 err:
1110         for_each_node(node)
1111                 kfree(wq->wqes[node]);
1112         kfree(wq->wqes);
1113         kfree(wq);
1114         return ERR_PTR(ret);
1115 }
1116
1117 static bool io_wq_worker_wake(struct io_worker *worker, void *data)
1118 {
1119         wake_up_process(worker->task);
1120         return false;
1121 }
1122
1123 static void __io_wq_destroy(struct io_wq *wq)
1124 {
1125         int node;
1126
1127         set_bit(IO_WQ_BIT_EXIT, &wq->state);
1128         if (wq->manager)
1129                 kthread_stop(wq->manager);
1130
1131         rcu_read_lock();
1132         for_each_node(node)
1133                 io_wq_for_each_worker(wq->wqes[node], io_wq_worker_wake, NULL);
1134         rcu_read_unlock();
1135
1136         wait_for_completion(&wq->done);
1137
1138         for_each_node(node)
1139                 kfree(wq->wqes[node]);
1140         kfree(wq->wqes);
1141         kfree(wq);
1142 }
1143
1144 void io_wq_destroy(struct io_wq *wq)
1145 {
1146         if (refcount_dec_and_test(&wq->use_refs))
1147                 __io_wq_destroy(wq);
1148 }