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