io_uring: move to using create_io_thread()
[linux-2.6-microblaze.git] / fs / io-wq.c
index 3a506f1..d7cfe8f 100644 (file)
@@ -16,6 +16,7 @@
 #include <linux/rculist_nulls.h>
 #include <linux/cpu.h>
 #include <linux/tracehook.h>
+#include <linux/freezer.h>
 
 #include "../kernel/sched/sched.h"
 #include "io-wq.h"
@@ -32,7 +33,6 @@ enum {
 
 enum {
        IO_WQ_BIT_EXIT          = 0,    /* wq exiting */
-       IO_WQ_BIT_ERROR         = 1,    /* error on setup */
 };
 
 enum {
@@ -53,8 +53,7 @@ struct io_worker {
        struct io_wq_work *cur_work;
        spinlock_t lock;
 
-       const struct cred *cur_creds;
-       const struct cred *saved_creds;
+       struct completion ref_done;
 
        struct rcu_head rcu;
 };
@@ -85,7 +84,6 @@ struct io_wqe {
        struct {
                raw_spinlock_t lock;
                struct io_wq_work_list work_list;
-               unsigned long hash_map;
                unsigned flags;
        } ____cacheline_aligned_in_smp;
 
@@ -95,6 +93,8 @@ struct io_wqe {
        struct hlist_nulls_head free_list;
        struct list_head all_list;
 
+       struct wait_queue_entry wait;
+
        struct io_wq *wq;
        struct io_wq_work *hash_tail[IO_WQ_NR_HASH_BUCKETS];
 };
@@ -111,8 +111,14 @@ struct io_wq {
 
        struct task_struct *manager;
        struct user_struct *user;
+
+       struct io_wq_hash *hash;
+
        refcount_t refs;
-       struct completion done;
+       struct completion exited;
+
+       atomic_t worker_refs;
+       struct completion worker_done;
 
        struct hlist_node cpuhp_node;
 
@@ -121,6 +127,17 @@ struct io_wq {
 
 static enum cpuhp_state io_wq_online;
 
+struct io_cb_cancel_data {
+       work_cancel_fn *fn;
+       void *data;
+       int nr_running;
+       int nr_pending;
+       bool cancel_all;
+};
+
+static void io_wqe_cancel_pending_work(struct io_wqe *wqe,
+                                      struct io_cb_cancel_data *match);
+
 static bool io_worker_get(struct io_worker *worker)
 {
        return refcount_inc_not_zero(&worker->ref);
@@ -129,7 +146,7 @@ static bool io_worker_get(struct io_worker *worker)
 static void io_worker_release(struct io_worker *worker)
 {
        if (refcount_dec_and_test(&worker->ref))
-               wake_up_process(worker->task);
+               complete(&worker->ref_done);
 }
 
 static inline struct io_wqe_acct *io_work_get_acct(struct io_wqe *wqe,
@@ -157,14 +174,9 @@ static void io_worker_exit(struct io_worker *worker)
        struct io_wqe_acct *acct = io_wqe_get_acct(worker);
        unsigned flags;
 
-       /*
-        * If we're not at zero, someone else is holding a brief reference
-        * to the worker. Wait for that to go away.
-        */
-       set_current_state(TASK_INTERRUPTIBLE);
-       if (!refcount_dec_and_test(&worker->ref))
-               schedule();
-       __set_current_state(TASK_RUNNING);
+       if (refcount_dec_and_test(&worker->ref))
+               complete(&worker->ref_done);
+       wait_for_completion(&worker->ref_done);
 
        preempt_disable();
        current->flags &= ~PF_IO_WORKER;
@@ -172,16 +184,9 @@ static void io_worker_exit(struct io_worker *worker)
        worker->flags = 0;
        if (flags & IO_WORKER_F_RUNNING)
                atomic_dec(&acct->nr_running);
-       if (!(flags & IO_WORKER_F_BOUND))
-               atomic_dec(&wqe->wq->user->processes);
        worker->flags = 0;
        preempt_enable();
 
-       if (worker->saved_creds) {
-               revert_creds(worker->saved_creds);
-               worker->cur_creds = worker->saved_creds = NULL;
-       }
-
        raw_spin_lock_irq(&wqe->lock);
        if (flags & IO_WORKER_F_FREE)
                hlist_nulls_del_rcu(&worker->nulls_node);
@@ -190,8 +195,9 @@ static void io_worker_exit(struct io_worker *worker)
        raw_spin_unlock_irq(&wqe->lock);
 
        kfree_rcu(worker, rcu);
-       if (refcount_dec_and_test(&wqe->wq->refs))
-               complete(&wqe->wq->done);
+       if (atomic_dec_and_test(&wqe->wq->worker_refs))
+               complete(&wqe->wq->worker_done);
+       do_exit(0);
 }
 
 static inline bool io_wqe_run_queue(struct io_wqe *wqe)
@@ -266,12 +272,6 @@ static void io_wqe_dec_running(struct io_worker *worker)
                io_wqe_wake_worker(wqe, acct);
 }
 
-static void io_worker_start(struct io_worker *worker)
-{
-       worker->flags |= (IO_WORKER_F_UP | IO_WORKER_F_RUNNING);
-       io_wqe_inc_running(worker);
-}
-
 /*
  * Worker will start processing some work. Move it to the busy list, if
  * it's currently on the freelist
@@ -299,12 +299,10 @@ static void __io_worker_busy(struct io_wqe *wqe, struct io_worker *worker,
                        worker->flags |= IO_WORKER_F_BOUND;
                        wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers--;
                        wqe->acct[IO_WQ_ACCT_BOUND].nr_workers++;
-                       atomic_dec(&wqe->wq->user->processes);
                } else {
                        worker->flags &= ~IO_WORKER_F_BOUND;
                        wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers++;
                        wqe->acct[IO_WQ_ACCT_BOUND].nr_workers--;
-                       atomic_inc(&wqe->wq->user->processes);
                }
                io_wqe_inc_running(worker);
         }
@@ -324,10 +322,6 @@ static void __io_worker_idle(struct io_wqe *wqe, struct io_worker *worker)
                worker->flags |= IO_WORKER_F_FREE;
                hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);
        }
-       if (worker->saved_creds) {
-               revert_creds(worker->saved_creds);
-               worker->cur_creds = worker->saved_creds = NULL;
-       }
 }
 
 static inline unsigned int io_get_work_hash(struct io_wq_work *work)
@@ -335,14 +329,31 @@ static inline unsigned int io_get_work_hash(struct io_wq_work *work)
        return work->flags >> IO_WQ_HASH_SHIFT;
 }
 
+static void io_wait_on_hash(struct io_wqe *wqe, unsigned int hash)
+{
+       struct io_wq *wq = wqe->wq;
+
+       spin_lock(&wq->hash->wait.lock);
+       if (list_empty(&wqe->wait.entry)) {
+               __add_wait_queue(&wq->hash->wait, &wqe->wait);
+               if (!test_bit(hash, &wq->hash->map)) {
+                       __set_current_state(TASK_RUNNING);
+                       list_del_init(&wqe->wait.entry);
+               }
+       }
+       spin_unlock(&wq->hash->wait.lock);
+}
+
 static struct io_wq_work *io_get_next_work(struct io_wqe *wqe)
        __must_hold(wqe->lock)
 {
        struct io_wq_work_node *node, *prev;
        struct io_wq_work *work, *tail;
-       unsigned int hash;
+       unsigned int stall_hash = -1U;
 
        wq_list_for_each(node, prev, &wqe->work_list) {
+               unsigned int hash;
+
                work = container_of(node, struct io_wq_work, list);
 
                /* not hashed, can run anytime */
@@ -351,16 +362,26 @@ static struct io_wq_work *io_get_next_work(struct io_wqe *wqe)
                        return work;
                }
 
-               /* hashed, can run if not already running */
                hash = io_get_work_hash(work);
-               if (!(wqe->hash_map & BIT(hash))) {
-                       wqe->hash_map |= BIT(hash);
-                       /* all items with this hash lie in [work, tail] */
-                       tail = wqe->hash_tail[hash];
+               /* all items with this hash lie in [work, tail] */
+               tail = wqe->hash_tail[hash];
+
+               /* hashed, can run if not already running */
+               if (!test_and_set_bit(hash, &wqe->wq->hash->map)) {
                        wqe->hash_tail[hash] = NULL;
                        wq_list_cut(&wqe->work_list, &tail->list, prev);
                        return work;
                }
+               if (stall_hash == -1U)
+                       stall_hash = hash;
+               /* fast forward to a next hash, for-each will fix up @prev */
+               node = &tail->list;
+       }
+
+       if (stall_hash != -1U) {
+               raw_spin_unlock(&wqe->lock);
+               io_wait_on_hash(wqe, stall_hash);
+               raw_spin_lock(&wqe->lock);
        }
 
        return NULL;
@@ -375,18 +396,6 @@ static void io_flush_signals(void)
        }
 }
 
-static void io_wq_switch_creds(struct io_worker *worker,
-                              struct io_wq_work *work)
-{
-       const struct cred *old_creds = override_creds(work->creds);
-
-       worker->cur_creds = work->creds;
-       if (worker->saved_creds)
-               put_cred(old_creds); /* creds set by previous switch */
-       else
-               worker->saved_creds = old_creds;
-}
-
 static void io_assign_current_work(struct io_worker *worker,
                                   struct io_wq_work *work)
 {
@@ -428,6 +437,7 @@ get_next:
                if (!work)
                        break;
                io_assign_current_work(worker, work);
+               __set_current_state(TASK_RUNNING);
 
                /* handle a whole dependent link */
                do {
@@ -435,8 +445,6 @@ get_next:
                        unsigned int hash = io_get_work_hash(work);
 
                        next_hashed = wq_next_work(work);
-                       if (work->creds && worker->cur_creds != work->creds)
-                               io_wq_switch_creds(worker, work);
                        wq->do_work(work);
                        io_assign_current_work(worker, NULL);
 
@@ -451,8 +459,10 @@ get_next:
                                io_wqe_enqueue(wqe, linked);
 
                        if (hash != -1U && !next_hashed) {
+                               clear_bit(hash, &wq->hash->map);
+                               if (wq_has_sleeper(&wq->hash->wait))
+                                       wake_up(&wq->hash->wait);
                                raw_spin_lock_irq(&wqe->lock);
-                               wqe->hash_map &= ~BIT_ULL(hash);
                                wqe->flags &= ~IO_WQE_FLAG_STALLED;
                                /* skip unnecessary unlock-lock wqe->lock */
                                if (!work)
@@ -470,15 +480,19 @@ static int io_wqe_worker(void *data)
        struct io_worker *worker = data;
        struct io_wqe *wqe = worker->wqe;
        struct io_wq *wq = wqe->wq;
+       char buf[TASK_COMM_LEN];
 
-       io_worker_start(worker);
+       worker->flags |= (IO_WORKER_F_UP | IO_WORKER_F_RUNNING);
+       io_wqe_inc_running(worker);
+
+       sprintf(buf, "iou-wrk-%d", wq->task_pid);
+       set_task_comm(current, buf);
 
        while (!test_bit(IO_WQ_BIT_EXIT, &wq->state)) {
                set_current_state(TASK_INTERRUPTIBLE);
 loop:
                raw_spin_lock_irq(&wqe->lock);
                if (io_wqe_run_queue(wqe)) {
-                       __set_current_state(TASK_RUNNING);
                        io_worker_handle_work(worker);
                        goto loop;
                }
@@ -547,70 +561,13 @@ void io_wq_worker_sleeping(struct task_struct *tsk)
        raw_spin_unlock_irq(&worker->wqe->lock);
 }
 
-static int task_thread(void *data, int index)
-{
-       struct io_worker *worker = data;
-       struct io_wqe *wqe = worker->wqe;
-       struct io_wqe_acct *acct = &wqe->acct[index];
-       struct io_wq *wq = wqe->wq;
-       char buf[TASK_COMM_LEN];
-
-       sprintf(buf, "iou-wrk-%d", wq->task_pid);
-       set_task_comm(current, buf);
-
-       current->pf_io_worker = worker;
-       worker->task = current;
-
-       set_cpus_allowed_ptr(current, cpumask_of_node(wqe->node));
-       current->flags |= PF_NO_SETAFFINITY;
-
-       raw_spin_lock_irq(&wqe->lock);
-       hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);
-       list_add_tail_rcu(&worker->all_list, &wqe->all_list);
-       worker->flags |= IO_WORKER_F_FREE;
-       if (index == IO_WQ_ACCT_BOUND)
-               worker->flags |= IO_WORKER_F_BOUND;
-       if (!acct->nr_workers && (worker->flags & IO_WORKER_F_BOUND))
-               worker->flags |= IO_WORKER_F_FIXED;
-       acct->nr_workers++;
-       raw_spin_unlock_irq(&wqe->lock);
-
-       if (index == IO_WQ_ACCT_UNBOUND)
-               atomic_inc(&wq->user->processes);
-
-       io_wqe_worker(data);
-       do_exit(0);
-}
-
-static int task_thread_bound(void *data)
-{
-       return task_thread(data, IO_WQ_ACCT_BOUND);
-}
-
-static int task_thread_unbound(void *data)
-{
-       return task_thread(data, IO_WQ_ACCT_UNBOUND);
-}
-
-static pid_t fork_thread(int (*fn)(void *), void *arg)
-{
-       unsigned long flags = CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|
-                               CLONE_IO|SIGCHLD;
-       struct kernel_clone_args args = {
-               .flags          = ((lower_32_bits(flags) | CLONE_VM |
-                                   CLONE_UNTRACED) & ~CSIGNAL),
-               .exit_signal    = (lower_32_bits(flags) & CSIGNAL),
-               .stack          = (unsigned long)fn,
-               .stack_size     = (unsigned long)arg,
-       };
-
-       return kernel_clone(&args);
-}
-
 static bool create_io_worker(struct io_wq *wq, struct io_wqe *wqe, int index)
 {
+       struct io_wqe_acct *acct = &wqe->acct[index];
        struct io_worker *worker;
-       pid_t pid;
+       struct task_struct *tsk;
+
+       __set_current_state(TASK_RUNNING);
 
        worker = kzalloc_node(sizeof(*worker), GFP_KERNEL, wqe->node);
        if (!worker)
@@ -620,16 +577,34 @@ static bool create_io_worker(struct io_wq *wq, struct io_wqe *wqe, int index)
        worker->nulls_node.pprev = NULL;
        worker->wqe = wqe;
        spin_lock_init(&worker->lock);
+       init_completion(&worker->ref_done);
 
-       if (index == IO_WQ_ACCT_BOUND)
-               pid = fork_thread(task_thread_bound, worker);
-       else
-               pid = fork_thread(task_thread_unbound, worker);
-       if (pid < 0) {
+       atomic_inc(&wq->worker_refs);
+
+       tsk = create_io_thread(io_wqe_worker, worker, wqe->node);
+       if (IS_ERR(tsk)) {
+               if (atomic_dec_and_test(&wq->worker_refs))
+                       complete(&wq->worker_done);
                kfree(worker);
                return false;
        }
-       refcount_inc(&wq->refs);
+
+       tsk->pf_io_worker = worker;
+       worker->task = tsk;
+       set_cpus_allowed_ptr(tsk, cpumask_of_node(wqe->node));
+       tsk->flags |= PF_NOFREEZE | PF_NO_SETAFFINITY;
+
+       raw_spin_lock_irq(&wqe->lock);
+       hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);
+       list_add_tail_rcu(&worker->all_list, &wqe->all_list);
+       worker->flags |= IO_WORKER_F_FREE;
+       if (index == IO_WQ_ACCT_BOUND)
+               worker->flags |= IO_WORKER_F_BOUND;
+       if (!acct->nr_workers && (worker->flags & IO_WORKER_F_BOUND))
+               worker->flags |= IO_WORKER_F_FIXED;
+       acct->nr_workers++;
+       raw_spin_unlock_irq(&wqe->lock);
+       wake_up_new_task(tsk);
        return true;
 }
 
@@ -638,6 +613,8 @@ static inline bool io_wqe_need_worker(struct io_wqe *wqe, int index)
 {
        struct io_wqe_acct *acct = &wqe->acct[index];
 
+       if (acct->nr_workers && test_bit(IO_WQ_BIT_EXIT, &wqe->wq->state))
+               return false;
        /* if we have available workers or no work, no need */
        if (!hlist_nulls_empty(&wqe->free_list) || !io_wqe_run_queue(wqe))
                return false;
@@ -671,10 +648,52 @@ static bool io_wq_for_each_worker(struct io_wqe *wqe,
 
 static bool io_wq_worker_wake(struct io_worker *worker, void *data)
 {
+       set_notify_signal(worker->task);
        wake_up_process(worker->task);
        return false;
 }
 
+static void io_wq_check_workers(struct io_wq *wq)
+{
+       int node;
+
+       for_each_node(node) {
+               struct io_wqe *wqe = wq->wqes[node];
+               bool fork_worker[2] = { false, false };
+
+               if (!node_online(node))
+                       continue;
+
+               raw_spin_lock_irq(&wqe->lock);
+               if (io_wqe_need_worker(wqe, IO_WQ_ACCT_BOUND))
+                       fork_worker[IO_WQ_ACCT_BOUND] = true;
+               if (io_wqe_need_worker(wqe, IO_WQ_ACCT_UNBOUND))
+                       fork_worker[IO_WQ_ACCT_UNBOUND] = true;
+               raw_spin_unlock_irq(&wqe->lock);
+               if (fork_worker[IO_WQ_ACCT_BOUND])
+                       create_io_worker(wq, wqe, IO_WQ_ACCT_BOUND);
+               if (fork_worker[IO_WQ_ACCT_UNBOUND])
+                       create_io_worker(wq, wqe, IO_WQ_ACCT_UNBOUND);
+       }
+}
+
+static bool io_wq_work_match_all(struct io_wq_work *work, void *data)
+{
+       return true;
+}
+
+static void io_wq_cancel_pending(struct io_wq *wq)
+{
+       struct io_cb_cancel_data match = {
+               .fn             = io_wq_work_match_all,
+               .cancel_all     = true,
+       };
+       int node;
+
+       for_each_node(node)
+               io_wqe_cancel_pending_work(wq->wqes[node], &match);
+}
+
 /*
  * Manager thread. Tasked with creating new workers, if we need them.
  */
@@ -686,71 +705,30 @@ static int io_wq_manager(void *data)
 
        sprintf(buf, "iou-mgr-%d", wq->task_pid);
        set_task_comm(current, buf);
-       current->flags |= PF_IO_WORKER;
-       wq->manager = current;
-
-       complete(&wq->done);
 
-       while (!test_bit(IO_WQ_BIT_EXIT, &wq->state)) {
-               for_each_node(node) {
-                       struct io_wqe *wqe = wq->wqes[node];
-                       bool fork_worker[2] = { false, false };
-
-                       if (!node_online(node))
-                               continue;
-
-                       raw_spin_lock_irq(&wqe->lock);
-                       if (io_wqe_need_worker(wqe, IO_WQ_ACCT_BOUND))
-                               fork_worker[IO_WQ_ACCT_BOUND] = true;
-                       if (io_wqe_need_worker(wqe, IO_WQ_ACCT_UNBOUND))
-                               fork_worker[IO_WQ_ACCT_UNBOUND] = true;
-                       raw_spin_unlock_irq(&wqe->lock);
-                       if (fork_worker[IO_WQ_ACCT_BOUND])
-                               create_io_worker(wq, wqe, IO_WQ_ACCT_BOUND);
-                       if (fork_worker[IO_WQ_ACCT_UNBOUND])
-                               create_io_worker(wq, wqe, IO_WQ_ACCT_UNBOUND);
-               }
+       do {
                set_current_state(TASK_INTERRUPTIBLE);
+               io_wq_check_workers(wq);
                schedule_timeout(HZ);
+               try_to_freeze();
                if (fatal_signal_pending(current))
                        set_bit(IO_WQ_BIT_EXIT, &wq->state);
-       }
-
-       if (refcount_dec_and_test(&wq->refs)) {
-               complete(&wq->done);
-               do_exit(0);
-       }
-       /* if ERROR is set and we get here, we have workers to wake */
-       if (test_bit(IO_WQ_BIT_ERROR, &wq->state)) {
-               rcu_read_lock();
-               for_each_node(node)
-                       io_wq_for_each_worker(wq->wqes[node], io_wq_worker_wake, NULL);
-               rcu_read_unlock();
-       }
-       do_exit(0);
-}
+       } while (!test_bit(IO_WQ_BIT_EXIT, &wq->state));
 
-static bool io_wq_can_queue(struct io_wqe *wqe, struct io_wqe_acct *acct,
-                           struct io_wq_work *work)
-{
-       bool free_worker;
-
-       if (!(work->flags & IO_WQ_WORK_UNBOUND))
-               return true;
-       if (atomic_read(&acct->nr_running))
-               return true;
+       io_wq_check_workers(wq);
 
        rcu_read_lock();
-       free_worker = !hlist_nulls_empty(&wqe->free_list);
+       for_each_node(node)
+               io_wq_for_each_worker(wq->wqes[node], io_wq_worker_wake, NULL);
        rcu_read_unlock();
-       if (free_worker)
-               return true;
 
-       if (atomic_read(&wqe->wq->user->processes) >= acct->max_workers &&
-           !(capable(CAP_SYS_RESOURCE) || capable(CAP_SYS_ADMIN)))
-               return false;
+       /* we might not ever have created any workers */
+       if (atomic_read(&wq->worker_refs))
+               wait_for_completion(&wq->worker_done);
 
-       return true;
+       io_wq_cancel_pending(wq);
+       complete(&wq->exited);
+       do_exit(0);
 }
 
 static void io_run_cancel(struct io_wq_work *work, struct io_wqe *wqe)
@@ -784,20 +762,35 @@ append:
        wq_list_add_after(&work->list, &tail->list, &wqe->work_list);
 }
 
+static int io_wq_fork_manager(struct io_wq *wq)
+{
+       struct task_struct *tsk;
+
+       if (wq->manager)
+               return 0;
+
+       reinit_completion(&wq->worker_done);
+       tsk = create_io_thread(io_wq_manager, wq, NUMA_NO_NODE);
+       if (!IS_ERR(tsk)) {
+               wq->manager = get_task_struct(tsk);
+               wake_up_new_task(tsk);
+               return 0;
+       }
+
+       return PTR_ERR(tsk);
+}
+
 static void io_wqe_enqueue(struct io_wqe *wqe, struct io_wq_work *work)
 {
        struct io_wqe_acct *acct = io_work_get_acct(wqe, work);
        int work_flags;
        unsigned long flags;
 
-       /*
-        * Do early check to see if we need a new unbound worker, and if we do,
-        * if we're allowed to do so. This isn't 100% accurate as there's a
-        * gap between this check and incrementing the value, but that's OK.
-        * It's close enough to not be an issue, fork() has the same delay.
-        */
-       if (unlikely(!io_wq_can_queue(wqe, acct, work))) {
-               io_run_cancel(work, wqe);
+       /* Can only happen if manager creation fails after exec */
+       if (io_wq_fork_manager(wqe->wq) ||
+           test_bit(IO_WQ_BIT_EXIT, &wqe->wq->state)) {
+               work->flags |= IO_WQ_WORK_CANCEL;
+               wqe->wq->do_work(work);
                return;
        }
 
@@ -831,14 +824,6 @@ void io_wq_hash_work(struct io_wq_work *work, void *val)
        work->flags |= (IO_WQ_WORK_HASHED | (bit << IO_WQ_HASH_SHIFT));
 }
 
-struct io_cb_cancel_data {
-       work_cancel_fn *fn;
-       void *data;
-       int nr_running;
-       int nr_pending;
-       bool cancel_all;
-};
-
 static bool io_wq_worker_cancel(struct io_worker *worker, void *data)
 {
        struct io_cb_cancel_data *match = data;
@@ -955,6 +940,24 @@ enum io_wq_cancel io_wq_cancel_cb(struct io_wq *wq, work_cancel_fn *cancel,
        return IO_WQ_CANCEL_NOTFOUND;
 }
 
+static int io_wqe_hash_wake(struct wait_queue_entry *wait, unsigned mode,
+                           int sync, void *key)
+{
+       struct io_wqe *wqe = container_of(wait, struct io_wqe, wait);
+       int ret;
+
+       list_del_init(&wait->entry);
+
+       rcu_read_lock();
+       ret = io_wqe_activate_free_worker(wqe);
+       rcu_read_unlock();
+
+       if (!ret)
+               wake_up_process(wqe->wq->manager);
+
+       return 1;
+}
+
 struct io_wq *io_wq_create(unsigned bounded, struct io_wq_data *data)
 {
        int ret = -ENOMEM, node;
@@ -975,12 +978,11 @@ struct io_wq *io_wq_create(unsigned bounded, struct io_wq_data *data)
        if (ret)
                goto err_wqes;
 
+       refcount_inc(&data->hash->refs);
+       wq->hash = data->hash;
        wq->free_work = data->free_work;
        wq->do_work = data->do_work;
 
-       /* caller must already hold a reference to this */
-       wq->user = data->user;
-
        ret = -ENOMEM;
        for_each_node(node) {
                struct io_wqe *wqe;
@@ -995,11 +997,11 @@ struct io_wq *io_wq_create(unsigned bounded, struct io_wq_data *data)
                wqe->node = alloc_node;
                wqe->acct[IO_WQ_ACCT_BOUND].max_workers = bounded;
                atomic_set(&wqe->acct[IO_WQ_ACCT_BOUND].nr_running, 0);
-               if (wq->user) {
-                       wqe->acct[IO_WQ_ACCT_UNBOUND].max_workers =
+               wqe->acct[IO_WQ_ACCT_UNBOUND].max_workers =
                                        task_rlimit(current, RLIMIT_NPROC);
-               }
                atomic_set(&wqe->acct[IO_WQ_ACCT_UNBOUND].nr_running, 0);
+               wqe->wait.func = io_wqe_hash_wake;
+               INIT_LIST_HEAD(&wqe->wait.entry);
                wqe->wq = wq;
                raw_spin_lock_init(&wqe->lock);
                INIT_WQ_LIST(&wqe->work_list);
@@ -1008,21 +1010,18 @@ struct io_wq *io_wq_create(unsigned bounded, struct io_wq_data *data)
        }
 
        wq->task_pid = current->pid;
-       init_completion(&wq->done);
+       init_completion(&wq->exited);
        refcount_set(&wq->refs, 1);
 
-       current->flags |= PF_IO_WORKER;
-       ret = fork_thread(io_wq_manager, wq);
-       current->flags &= ~PF_IO_WORKER;
-       if (ret >= 0) {
-               wait_for_completion(&wq->done);
-               reinit_completion(&wq->done);
+       init_completion(&wq->worker_done);
+       atomic_set(&wq->worker_refs, 0);
+
+       ret = io_wq_fork_manager(wq);
+       if (!ret)
                return wq;
-       }
 
-       if (refcount_dec_and_test(&wq->refs))
-               complete(&wq->done);
 err:
+       io_wq_put_hash(data->hash);
        cpuhp_state_remove_instance_nocalls(io_wq_online, &wq->cpuhp_node);
        for_each_node(node)
                kfree(wq->wqes[node]);
@@ -1033,29 +1032,52 @@ err_wq:
        return ERR_PTR(ret);
 }
 
-void io_wq_destroy(struct io_wq *wq)
+static void io_wq_destroy_manager(struct io_wq *wq)
+{
+       if (wq->manager) {
+               wake_up_process(wq->manager);
+               wait_for_completion(&wq->exited);
+               put_task_struct(wq->manager);
+               wq->manager = NULL;
+       }
+}
+
+static void io_wq_destroy(struct io_wq *wq)
 {
        int node;
 
        cpuhp_state_remove_instance_nocalls(io_wq_online, &wq->cpuhp_node);
 
        set_bit(IO_WQ_BIT_EXIT, &wq->state);
-       if (wq->manager)
-               wake_up_process(wq->manager);
-
-       rcu_read_lock();
-       for_each_node(node)
-               io_wq_for_each_worker(wq->wqes[node], io_wq_worker_wake, NULL);
-       rcu_read_unlock();
+       io_wq_destroy_manager(wq);
 
-       wait_for_completion(&wq->done);
+       spin_lock_irq(&wq->hash->wait.lock);
+       for_each_node(node) {
+               struct io_wqe *wqe = wq->wqes[node];
 
-       for_each_node(node)
-               kfree(wq->wqes[node]);
+               list_del_init(&wqe->wait.entry);
+               WARN_ON_ONCE(!wq_list_empty(&wqe->work_list));
+               kfree(wqe);
+       }
+       spin_unlock_irq(&wq->hash->wait.lock);
+       io_wq_put_hash(wq->hash);
        kfree(wq->wqes);
        kfree(wq);
 }
 
+void io_wq_put(struct io_wq *wq)
+{
+       if (refcount_dec_and_test(&wq->refs))
+               io_wq_destroy(wq);
+}
+
+void io_wq_put_and_exit(struct io_wq *wq)
+{
+       set_bit(IO_WQ_BIT_EXIT, &wq->state);
+       io_wq_destroy_manager(wq);
+       io_wq_put(wq);
+}
+
 static bool io_wq_worker_affinity(struct io_worker *worker, void *data)
 {
        struct task_struct *task = worker->task;