1 // SPDX-License-Identifier: GPL-2.0
3 * Basic worker thread pool for io_uring
5 * Copyright (C) 2019 Jens Axboe
8 #include <linux/kernel.h>
9 #include <linux/init.h>
10 #include <linux/errno.h>
11 #include <linux/sched/signal.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 #include <linux/cpu.h>
24 #include "../kernel/sched/sched.h"
27 #define WORKER_IDLE_TIMEOUT (5 * HZ)
30 IO_WORKER_F_UP
= 1, /* up and active */
31 IO_WORKER_F_RUNNING
= 2, /* account as running */
32 IO_WORKER_F_FREE
= 4, /* worker on free list */
33 IO_WORKER_F_FIXED
= 8, /* static idle worker */
34 IO_WORKER_F_BOUND
= 16, /* is doing bounded work */
38 IO_WQ_BIT_EXIT
= 0, /* wq exiting */
39 IO_WQ_BIT_CANCEL
= 1, /* cancel work on list */
40 IO_WQ_BIT_ERROR
= 2, /* error on setup */
44 IO_WQE_FLAG_STALLED
= 1, /* stalled on hash */
48 * One for each thread in a wqe pool
53 struct hlist_nulls_node nulls_node
;
54 struct list_head all_list
;
55 struct task_struct
*task
;
58 struct io_wq_work
*cur_work
;
63 #ifdef CONFIG_BLK_CGROUP
64 struct cgroup_subsys_state
*blkcg_css
;
66 const struct cred
*cur_creds
;
67 const struct cred
*saved_creds
;
68 struct files_struct
*restore_files
;
69 struct nsproxy
*restore_nsproxy
;
70 struct fs_struct
*restore_fs
;
73 #if BITS_PER_LONG == 64
74 #define IO_WQ_HASH_ORDER 6
76 #define IO_WQ_HASH_ORDER 5
79 #define IO_WQ_NR_HASH_BUCKETS (1u << IO_WQ_HASH_ORDER)
93 * Per-node worker thread pool
98 struct io_wq_work_list work_list
;
99 unsigned long hash_map
;
101 } ____cacheline_aligned_in_smp
;
104 struct io_wqe_acct acct
[2];
106 struct hlist_nulls_head free_list
;
107 struct list_head all_list
;
110 struct io_wq_work
*hash_tail
[IO_WQ_NR_HASH_BUCKETS
];
117 struct io_wqe
**wqes
;
120 free_work_fn
*free_work
;
121 io_wq_work_fn
*do_work
;
123 struct task_struct
*manager
;
124 struct user_struct
*user
;
126 struct completion done
;
128 struct hlist_node cpuhp_node
;
133 static enum cpuhp_state io_wq_online
;
135 static bool io_worker_get(struct io_worker
*worker
)
137 return refcount_inc_not_zero(&worker
->ref
);
140 static void io_worker_release(struct io_worker
*worker
)
142 if (refcount_dec_and_test(&worker
->ref
))
143 wake_up_process(worker
->task
);
147 * Note: drops the wqe->lock if returning true! The caller must re-acquire
148 * the lock in that case. Some callers need to restart handling if this
149 * happens, so we can't just re-acquire the lock on behalf of the caller.
151 static bool __io_worker_unuse(struct io_wqe
*wqe
, struct io_worker
*worker
)
153 bool dropped_lock
= false;
155 if (worker
->saved_creds
) {
156 revert_creds(worker
->saved_creds
);
157 worker
->cur_creds
= worker
->saved_creds
= NULL
;
160 if (current
->files
!= worker
->restore_files
) {
161 __acquire(&wqe
->lock
);
162 raw_spin_unlock_irq(&wqe
->lock
);
166 current
->files
= worker
->restore_files
;
167 current
->nsproxy
= worker
->restore_nsproxy
;
168 task_unlock(current
);
171 if (current
->fs
!= worker
->restore_fs
)
172 current
->fs
= worker
->restore_fs
;
175 * If we have an active mm, we need to drop the wq lock before unusing
176 * it. If we do, return true and let the caller retry the idle loop.
180 __acquire(&wqe
->lock
);
181 raw_spin_unlock_irq(&wqe
->lock
);
184 __set_current_state(TASK_RUNNING
);
185 kthread_unuse_mm(worker
->mm
);
190 #ifdef CONFIG_BLK_CGROUP
191 if (worker
->blkcg_css
) {
192 kthread_associate_blkcg(NULL
);
193 worker
->blkcg_css
= NULL
;
196 if (current
->signal
->rlim
[RLIMIT_FSIZE
].rlim_cur
!= RLIM_INFINITY
)
197 current
->signal
->rlim
[RLIMIT_FSIZE
].rlim_cur
= RLIM_INFINITY
;
201 static inline struct io_wqe_acct
*io_work_get_acct(struct io_wqe
*wqe
,
202 struct io_wq_work
*work
)
204 if (work
->flags
& IO_WQ_WORK_UNBOUND
)
205 return &wqe
->acct
[IO_WQ_ACCT_UNBOUND
];
207 return &wqe
->acct
[IO_WQ_ACCT_BOUND
];
210 static inline struct io_wqe_acct
*io_wqe_get_acct(struct io_wqe
*wqe
,
211 struct io_worker
*worker
)
213 if (worker
->flags
& IO_WORKER_F_BOUND
)
214 return &wqe
->acct
[IO_WQ_ACCT_BOUND
];
216 return &wqe
->acct
[IO_WQ_ACCT_UNBOUND
];
219 static void io_worker_exit(struct io_worker
*worker
)
221 struct io_wqe
*wqe
= worker
->wqe
;
222 struct io_wqe_acct
*acct
= io_wqe_get_acct(wqe
, worker
);
225 * If we're not at zero, someone else is holding a brief reference
226 * to the worker. Wait for that to go away.
228 set_current_state(TASK_INTERRUPTIBLE
);
229 if (!refcount_dec_and_test(&worker
->ref
))
231 __set_current_state(TASK_RUNNING
);
234 current
->flags
&= ~PF_IO_WORKER
;
235 if (worker
->flags
& IO_WORKER_F_RUNNING
)
236 atomic_dec(&acct
->nr_running
);
237 if (!(worker
->flags
& IO_WORKER_F_BOUND
))
238 atomic_dec(&wqe
->wq
->user
->processes
);
242 raw_spin_lock_irq(&wqe
->lock
);
243 hlist_nulls_del_rcu(&worker
->nulls_node
);
244 list_del_rcu(&worker
->all_list
);
245 if (__io_worker_unuse(wqe
, worker
)) {
246 __release(&wqe
->lock
);
247 raw_spin_lock_irq(&wqe
->lock
);
250 raw_spin_unlock_irq(&wqe
->lock
);
252 kfree_rcu(worker
, rcu
);
253 if (refcount_dec_and_test(&wqe
->wq
->refs
))
254 complete(&wqe
->wq
->done
);
257 static inline bool io_wqe_run_queue(struct io_wqe
*wqe
)
258 __must_hold(wqe
->lock
)
260 if (!wq_list_empty(&wqe
->work_list
) &&
261 !(wqe
->flags
& IO_WQE_FLAG_STALLED
))
267 * Check head of free list for an available worker. If one isn't available,
268 * caller must wake up the wq manager to create one.
270 static bool io_wqe_activate_free_worker(struct io_wqe
*wqe
)
273 struct hlist_nulls_node
*n
;
274 struct io_worker
*worker
;
276 n
= rcu_dereference(hlist_nulls_first_rcu(&wqe
->free_list
));
280 worker
= hlist_nulls_entry(n
, struct io_worker
, nulls_node
);
281 if (io_worker_get(worker
)) {
282 wake_up_process(worker
->task
);
283 io_worker_release(worker
);
291 * We need a worker. If we find a free one, we're good. If not, and we're
292 * below the max number of workers, wake up the manager to create one.
294 static void io_wqe_wake_worker(struct io_wqe
*wqe
, struct io_wqe_acct
*acct
)
299 * Most likely an attempt to queue unbounded work on an io_wq that
300 * wasn't setup with any unbounded workers.
302 WARN_ON_ONCE(!acct
->max_workers
);
305 ret
= io_wqe_activate_free_worker(wqe
);
308 if (!ret
&& acct
->nr_workers
< acct
->max_workers
)
309 wake_up_process(wqe
->wq
->manager
);
312 static void io_wqe_inc_running(struct io_wqe
*wqe
, struct io_worker
*worker
)
314 struct io_wqe_acct
*acct
= io_wqe_get_acct(wqe
, worker
);
316 atomic_inc(&acct
->nr_running
);
319 static void io_wqe_dec_running(struct io_wqe
*wqe
, struct io_worker
*worker
)
320 __must_hold(wqe
->lock
)
322 struct io_wqe_acct
*acct
= io_wqe_get_acct(wqe
, worker
);
324 if (atomic_dec_and_test(&acct
->nr_running
) && io_wqe_run_queue(wqe
))
325 io_wqe_wake_worker(wqe
, acct
);
328 static void io_worker_start(struct io_wqe
*wqe
, struct io_worker
*worker
)
330 allow_kernel_signal(SIGINT
);
332 current
->flags
|= PF_IO_WORKER
;
334 worker
->flags
|= (IO_WORKER_F_UP
| IO_WORKER_F_RUNNING
);
335 worker
->restore_files
= current
->files
;
336 worker
->restore_nsproxy
= current
->nsproxy
;
337 worker
->restore_fs
= current
->fs
;
338 io_wqe_inc_running(wqe
, worker
);
342 * Worker will start processing some work. Move it to the busy list, if
343 * it's currently on the freelist
345 static void __io_worker_busy(struct io_wqe
*wqe
, struct io_worker
*worker
,
346 struct io_wq_work
*work
)
347 __must_hold(wqe
->lock
)
349 bool worker_bound
, work_bound
;
351 if (worker
->flags
& IO_WORKER_F_FREE
) {
352 worker
->flags
&= ~IO_WORKER_F_FREE
;
353 hlist_nulls_del_init_rcu(&worker
->nulls_node
);
357 * If worker is moving from bound to unbound (or vice versa), then
358 * ensure we update the running accounting.
360 worker_bound
= (worker
->flags
& IO_WORKER_F_BOUND
) != 0;
361 work_bound
= (work
->flags
& IO_WQ_WORK_UNBOUND
) == 0;
362 if (worker_bound
!= work_bound
) {
363 io_wqe_dec_running(wqe
, worker
);
365 worker
->flags
|= IO_WORKER_F_BOUND
;
366 wqe
->acct
[IO_WQ_ACCT_UNBOUND
].nr_workers
--;
367 wqe
->acct
[IO_WQ_ACCT_BOUND
].nr_workers
++;
368 atomic_dec(&wqe
->wq
->user
->processes
);
370 worker
->flags
&= ~IO_WORKER_F_BOUND
;
371 wqe
->acct
[IO_WQ_ACCT_UNBOUND
].nr_workers
++;
372 wqe
->acct
[IO_WQ_ACCT_BOUND
].nr_workers
--;
373 atomic_inc(&wqe
->wq
->user
->processes
);
375 io_wqe_inc_running(wqe
, worker
);
380 * No work, worker going to sleep. Move to freelist, and unuse mm if we
381 * have one attached. Dropping the mm may potentially sleep, so we drop
382 * the lock in that case and return success. Since the caller has to
383 * retry the loop in that case (we changed task state), we don't regrab
384 * the lock if we return success.
386 static bool __io_worker_idle(struct io_wqe
*wqe
, struct io_worker
*worker
)
387 __must_hold(wqe
->lock
)
389 if (!(worker
->flags
& IO_WORKER_F_FREE
)) {
390 worker
->flags
|= IO_WORKER_F_FREE
;
391 hlist_nulls_add_head_rcu(&worker
->nulls_node
, &wqe
->free_list
);
394 return __io_worker_unuse(wqe
, worker
);
397 static inline unsigned int io_get_work_hash(struct io_wq_work
*work
)
399 return work
->flags
>> IO_WQ_HASH_SHIFT
;
402 static struct io_wq_work
*io_get_next_work(struct io_wqe
*wqe
)
403 __must_hold(wqe
->lock
)
405 struct io_wq_work_node
*node
, *prev
;
406 struct io_wq_work
*work
, *tail
;
409 wq_list_for_each(node
, prev
, &wqe
->work_list
) {
410 work
= container_of(node
, struct io_wq_work
, list
);
412 /* not hashed, can run anytime */
413 if (!io_wq_is_hashed(work
)) {
414 wq_list_del(&wqe
->work_list
, node
, prev
);
418 /* hashed, can run if not already running */
419 hash
= io_get_work_hash(work
);
420 if (!(wqe
->hash_map
& BIT(hash
))) {
421 wqe
->hash_map
|= BIT(hash
);
422 /* all items with this hash lie in [work, tail] */
423 tail
= wqe
->hash_tail
[hash
];
424 wqe
->hash_tail
[hash
] = NULL
;
425 wq_list_cut(&wqe
->work_list
, &tail
->list
, prev
);
433 static void io_wq_switch_mm(struct io_worker
*worker
, struct io_wq_work
*work
)
436 kthread_unuse_mm(worker
->mm
);
441 if (mmget_not_zero(work
->identity
->mm
)) {
442 kthread_use_mm(work
->identity
->mm
);
443 worker
->mm
= work
->identity
->mm
;
447 /* failed grabbing mm, ensure work gets cancelled */
448 work
->flags
|= IO_WQ_WORK_CANCEL
;
451 static inline void io_wq_switch_blkcg(struct io_worker
*worker
,
452 struct io_wq_work
*work
)
454 #ifdef CONFIG_BLK_CGROUP
455 if (!(work
->flags
& IO_WQ_WORK_BLKCG
))
457 if (work
->identity
->blkcg_css
!= worker
->blkcg_css
) {
458 kthread_associate_blkcg(work
->identity
->blkcg_css
);
459 worker
->blkcg_css
= work
->identity
->blkcg_css
;
464 static void io_wq_switch_creds(struct io_worker
*worker
,
465 struct io_wq_work
*work
)
467 const struct cred
*old_creds
= override_creds(work
->identity
->creds
);
469 worker
->cur_creds
= work
->identity
->creds
;
470 if (worker
->saved_creds
)
471 put_cred(old_creds
); /* creds set by previous switch */
473 worker
->saved_creds
= old_creds
;
476 static void io_impersonate_work(struct io_worker
*worker
,
477 struct io_wq_work
*work
)
479 if ((work
->flags
& IO_WQ_WORK_FILES
) &&
480 current
->files
!= work
->identity
->files
) {
482 current
->files
= work
->identity
->files
;
483 current
->nsproxy
= work
->identity
->nsproxy
;
484 task_unlock(current
);
485 if (!work
->identity
->files
) {
486 /* failed grabbing files, ensure work gets cancelled */
487 work
->flags
|= IO_WQ_WORK_CANCEL
;
490 if ((work
->flags
& IO_WQ_WORK_FS
) && current
->fs
!= work
->identity
->fs
)
491 current
->fs
= work
->identity
->fs
;
492 if ((work
->flags
& IO_WQ_WORK_MM
) && work
->identity
->mm
!= worker
->mm
)
493 io_wq_switch_mm(worker
, work
);
494 if ((work
->flags
& IO_WQ_WORK_CREDS
) &&
495 worker
->cur_creds
!= work
->identity
->creds
)
496 io_wq_switch_creds(worker
, work
);
497 if (work
->flags
& IO_WQ_WORK_FSIZE
)
498 current
->signal
->rlim
[RLIMIT_FSIZE
].rlim_cur
= work
->identity
->fsize
;
499 else if (current
->signal
->rlim
[RLIMIT_FSIZE
].rlim_cur
!= RLIM_INFINITY
)
500 current
->signal
->rlim
[RLIMIT_FSIZE
].rlim_cur
= RLIM_INFINITY
;
501 io_wq_switch_blkcg(worker
, work
);
503 current
->loginuid
= work
->identity
->loginuid
;
504 current
->sessionid
= work
->identity
->sessionid
;
508 static void io_assign_current_work(struct io_worker
*worker
,
509 struct io_wq_work
*work
)
512 /* flush pending signals before assigning new work */
513 if (signal_pending(current
))
514 flush_signals(current
);
519 current
->loginuid
= KUIDT_INIT(AUDIT_UID_UNSET
);
520 current
->sessionid
= AUDIT_SID_UNSET
;
523 spin_lock_irq(&worker
->lock
);
524 worker
->cur_work
= work
;
525 spin_unlock_irq(&worker
->lock
);
528 static void io_wqe_enqueue(struct io_wqe
*wqe
, struct io_wq_work
*work
);
530 static void io_worker_handle_work(struct io_worker
*worker
)
531 __releases(wqe
->lock
)
533 struct io_wqe
*wqe
= worker
->wqe
;
534 struct io_wq
*wq
= wqe
->wq
;
537 struct io_wq_work
*work
;
540 * If we got some work, mark us as busy. If we didn't, but
541 * the list isn't empty, it means we stalled on hashed work.
542 * Mark us stalled so we don't keep looking for work when we
543 * can't make progress, any work completion or insertion will
544 * clear the stalled flag.
546 work
= io_get_next_work(wqe
);
548 __io_worker_busy(wqe
, worker
, work
);
549 else if (!wq_list_empty(&wqe
->work_list
))
550 wqe
->flags
|= IO_WQE_FLAG_STALLED
;
552 raw_spin_unlock_irq(&wqe
->lock
);
555 io_assign_current_work(worker
, work
);
557 /* handle a whole dependent link */
559 struct io_wq_work
*old_work
, *next_hashed
, *linked
;
560 unsigned int hash
= io_get_work_hash(work
);
562 next_hashed
= wq_next_work(work
);
563 io_impersonate_work(worker
, work
);
565 * OK to set IO_WQ_WORK_CANCEL even for uncancellable
566 * work, the worker function will do the right thing.
568 if (test_bit(IO_WQ_BIT_CANCEL
, &wq
->state
))
569 work
->flags
|= IO_WQ_WORK_CANCEL
;
572 linked
= wq
->do_work(work
);
575 if (!work
&& linked
&& !io_wq_is_hashed(linked
)) {
579 io_assign_current_work(worker
, work
);
580 wq
->free_work(old_work
);
583 io_wqe_enqueue(wqe
, linked
);
585 if (hash
!= -1U && !next_hashed
) {
586 raw_spin_lock_irq(&wqe
->lock
);
587 wqe
->hash_map
&= ~BIT_ULL(hash
);
588 wqe
->flags
&= ~IO_WQE_FLAG_STALLED
;
589 /* skip unnecessary unlock-lock wqe->lock */
592 raw_spin_unlock_irq(&wqe
->lock
);
596 raw_spin_lock_irq(&wqe
->lock
);
600 static int io_wqe_worker(void *data
)
602 struct io_worker
*worker
= data
;
603 struct io_wqe
*wqe
= worker
->wqe
;
604 struct io_wq
*wq
= wqe
->wq
;
606 io_worker_start(wqe
, worker
);
608 while (!test_bit(IO_WQ_BIT_EXIT
, &wq
->state
)) {
609 set_current_state(TASK_INTERRUPTIBLE
);
611 raw_spin_lock_irq(&wqe
->lock
);
612 if (io_wqe_run_queue(wqe
)) {
613 __set_current_state(TASK_RUNNING
);
614 io_worker_handle_work(worker
);
617 /* drops the lock on success, retry */
618 if (__io_worker_idle(wqe
, worker
)) {
619 __release(&wqe
->lock
);
622 raw_spin_unlock_irq(&wqe
->lock
);
623 if (signal_pending(current
))
624 flush_signals(current
);
625 if (schedule_timeout(WORKER_IDLE_TIMEOUT
))
627 /* timed out, exit unless we're the fixed worker */
628 if (test_bit(IO_WQ_BIT_EXIT
, &wq
->state
) ||
629 !(worker
->flags
& IO_WORKER_F_FIXED
))
633 if (test_bit(IO_WQ_BIT_EXIT
, &wq
->state
)) {
634 raw_spin_lock_irq(&wqe
->lock
);
635 if (!wq_list_empty(&wqe
->work_list
))
636 io_worker_handle_work(worker
);
638 raw_spin_unlock_irq(&wqe
->lock
);
641 io_worker_exit(worker
);
646 * Called when a worker is scheduled in. Mark us as currently running.
648 void io_wq_worker_running(struct task_struct
*tsk
)
650 struct io_worker
*worker
= kthread_data(tsk
);
651 struct io_wqe
*wqe
= worker
->wqe
;
653 if (!(worker
->flags
& IO_WORKER_F_UP
))
655 if (worker
->flags
& IO_WORKER_F_RUNNING
)
657 worker
->flags
|= IO_WORKER_F_RUNNING
;
658 io_wqe_inc_running(wqe
, worker
);
662 * Called when worker is going to sleep. If there are no workers currently
663 * running and we have work pending, wake up a free one or have the manager
666 void io_wq_worker_sleeping(struct task_struct
*tsk
)
668 struct io_worker
*worker
= kthread_data(tsk
);
669 struct io_wqe
*wqe
= worker
->wqe
;
671 if (!(worker
->flags
& IO_WORKER_F_UP
))
673 if (!(worker
->flags
& IO_WORKER_F_RUNNING
))
676 worker
->flags
&= ~IO_WORKER_F_RUNNING
;
678 raw_spin_lock_irq(&wqe
->lock
);
679 io_wqe_dec_running(wqe
, worker
);
680 raw_spin_unlock_irq(&wqe
->lock
);
683 static bool create_io_worker(struct io_wq
*wq
, struct io_wqe
*wqe
, int index
)
685 struct io_wqe_acct
*acct
= &wqe
->acct
[index
];
686 struct io_worker
*worker
;
688 worker
= kzalloc_node(sizeof(*worker
), GFP_KERNEL
, wqe
->node
);
692 refcount_set(&worker
->ref
, 1);
693 worker
->nulls_node
.pprev
= NULL
;
695 spin_lock_init(&worker
->lock
);
697 worker
->task
= kthread_create_on_node(io_wqe_worker
, worker
, wqe
->node
,
698 "io_wqe_worker-%d/%d", index
, wqe
->node
);
699 if (IS_ERR(worker
->task
)) {
703 kthread_bind_mask(worker
->task
, cpumask_of_node(wqe
->node
));
705 raw_spin_lock_irq(&wqe
->lock
);
706 hlist_nulls_add_head_rcu(&worker
->nulls_node
, &wqe
->free_list
);
707 list_add_tail_rcu(&worker
->all_list
, &wqe
->all_list
);
708 worker
->flags
|= IO_WORKER_F_FREE
;
709 if (index
== IO_WQ_ACCT_BOUND
)
710 worker
->flags
|= IO_WORKER_F_BOUND
;
711 if (!acct
->nr_workers
&& (worker
->flags
& IO_WORKER_F_BOUND
))
712 worker
->flags
|= IO_WORKER_F_FIXED
;
714 raw_spin_unlock_irq(&wqe
->lock
);
716 if (index
== IO_WQ_ACCT_UNBOUND
)
717 atomic_inc(&wq
->user
->processes
);
719 refcount_inc(&wq
->refs
);
720 wake_up_process(worker
->task
);
724 static inline bool io_wqe_need_worker(struct io_wqe
*wqe
, int index
)
725 __must_hold(wqe
->lock
)
727 struct io_wqe_acct
*acct
= &wqe
->acct
[index
];
729 /* if we have available workers or no work, no need */
730 if (!hlist_nulls_empty(&wqe
->free_list
) || !io_wqe_run_queue(wqe
))
732 return acct
->nr_workers
< acct
->max_workers
;
735 static bool io_wqe_worker_send_sig(struct io_worker
*worker
, void *data
)
737 send_sig(SIGINT
, worker
->task
, 1);
742 * Iterate the passed in list and call the specific function for each
743 * worker that isn't exiting
745 static bool io_wq_for_each_worker(struct io_wqe
*wqe
,
746 bool (*func
)(struct io_worker
*, void *),
749 struct io_worker
*worker
;
752 list_for_each_entry_rcu(worker
, &wqe
->all_list
, all_list
) {
753 if (io_worker_get(worker
)) {
754 /* no task if node is/was offline */
756 ret
= func(worker
, data
);
757 io_worker_release(worker
);
766 static bool io_wq_worker_wake(struct io_worker
*worker
, void *data
)
768 wake_up_process(worker
->task
);
773 * Manager thread. Tasked with creating new workers, if we need them.
775 static int io_wq_manager(void *data
)
777 struct io_wq
*wq
= data
;
780 /* create fixed workers */
781 refcount_set(&wq
->refs
, 1);
782 for_each_node(node
) {
783 if (!node_online(node
))
785 if (create_io_worker(wq
, wq
->wqes
[node
], IO_WQ_ACCT_BOUND
))
787 set_bit(IO_WQ_BIT_ERROR
, &wq
->state
);
788 set_bit(IO_WQ_BIT_EXIT
, &wq
->state
);
794 while (!kthread_should_stop()) {
795 if (current
->task_works
)
798 for_each_node(node
) {
799 struct io_wqe
*wqe
= wq
->wqes
[node
];
800 bool fork_worker
[2] = { false, false };
802 if (!node_online(node
))
805 raw_spin_lock_irq(&wqe
->lock
);
806 if (io_wqe_need_worker(wqe
, IO_WQ_ACCT_BOUND
))
807 fork_worker
[IO_WQ_ACCT_BOUND
] = true;
808 if (io_wqe_need_worker(wqe
, IO_WQ_ACCT_UNBOUND
))
809 fork_worker
[IO_WQ_ACCT_UNBOUND
] = true;
810 raw_spin_unlock_irq(&wqe
->lock
);
811 if (fork_worker
[IO_WQ_ACCT_BOUND
])
812 create_io_worker(wq
, wqe
, IO_WQ_ACCT_BOUND
);
813 if (fork_worker
[IO_WQ_ACCT_UNBOUND
])
814 create_io_worker(wq
, wqe
, IO_WQ_ACCT_UNBOUND
);
816 set_current_state(TASK_INTERRUPTIBLE
);
817 schedule_timeout(HZ
);
820 if (current
->task_works
)
824 if (refcount_dec_and_test(&wq
->refs
)) {
828 /* if ERROR is set and we get here, we have workers to wake */
829 if (test_bit(IO_WQ_BIT_ERROR
, &wq
->state
)) {
832 io_wq_for_each_worker(wq
->wqes
[node
], io_wq_worker_wake
, NULL
);
838 static bool io_wq_can_queue(struct io_wqe
*wqe
, struct io_wqe_acct
*acct
,
839 struct io_wq_work
*work
)
843 if (!(work
->flags
& IO_WQ_WORK_UNBOUND
))
845 if (atomic_read(&acct
->nr_running
))
849 free_worker
= !hlist_nulls_empty(&wqe
->free_list
);
854 if (atomic_read(&wqe
->wq
->user
->processes
) >= acct
->max_workers
&&
855 !(capable(CAP_SYS_RESOURCE
) || capable(CAP_SYS_ADMIN
)))
861 static void io_run_cancel(struct io_wq_work
*work
, struct io_wqe
*wqe
)
863 struct io_wq
*wq
= wqe
->wq
;
866 struct io_wq_work
*old_work
= work
;
868 work
->flags
|= IO_WQ_WORK_CANCEL
;
869 work
= wq
->do_work(work
);
870 wq
->free_work(old_work
);
874 static void io_wqe_insert_work(struct io_wqe
*wqe
, struct io_wq_work
*work
)
877 struct io_wq_work
*tail
;
879 if (!io_wq_is_hashed(work
)) {
881 wq_list_add_tail(&work
->list
, &wqe
->work_list
);
885 hash
= io_get_work_hash(work
);
886 tail
= wqe
->hash_tail
[hash
];
887 wqe
->hash_tail
[hash
] = work
;
891 wq_list_add_after(&work
->list
, &tail
->list
, &wqe
->work_list
);
894 static void io_wqe_enqueue(struct io_wqe
*wqe
, struct io_wq_work
*work
)
896 struct io_wqe_acct
*acct
= io_work_get_acct(wqe
, work
);
901 * Do early check to see if we need a new unbound worker, and if we do,
902 * if we're allowed to do so. This isn't 100% accurate as there's a
903 * gap between this check and incrementing the value, but that's OK.
904 * It's close enough to not be an issue, fork() has the same delay.
906 if (unlikely(!io_wq_can_queue(wqe
, acct
, work
))) {
907 io_run_cancel(work
, wqe
);
911 work_flags
= work
->flags
;
912 raw_spin_lock_irqsave(&wqe
->lock
, flags
);
913 io_wqe_insert_work(wqe
, work
);
914 wqe
->flags
&= ~IO_WQE_FLAG_STALLED
;
915 raw_spin_unlock_irqrestore(&wqe
->lock
, flags
);
917 if ((work_flags
& IO_WQ_WORK_CONCURRENT
) ||
918 !atomic_read(&acct
->nr_running
))
919 io_wqe_wake_worker(wqe
, acct
);
922 void io_wq_enqueue(struct io_wq
*wq
, struct io_wq_work
*work
)
924 struct io_wqe
*wqe
= wq
->wqes
[numa_node_id()];
926 io_wqe_enqueue(wqe
, work
);
930 * Work items that hash to the same value will not be done in parallel.
931 * Used to limit concurrent writes, generally hashed by inode.
933 void io_wq_hash_work(struct io_wq_work
*work
, void *val
)
937 bit
= hash_ptr(val
, IO_WQ_HASH_ORDER
);
938 work
->flags
|= (IO_WQ_WORK_HASHED
| (bit
<< IO_WQ_HASH_SHIFT
));
941 void io_wq_cancel_all(struct io_wq
*wq
)
945 set_bit(IO_WQ_BIT_CANCEL
, &wq
->state
);
948 for_each_node(node
) {
949 struct io_wqe
*wqe
= wq
->wqes
[node
];
951 io_wq_for_each_worker(wqe
, io_wqe_worker_send_sig
, NULL
);
956 struct io_cb_cancel_data
{
964 static bool io_wq_worker_cancel(struct io_worker
*worker
, void *data
)
966 struct io_cb_cancel_data
*match
= data
;
970 * Hold the lock to avoid ->cur_work going out of scope, caller
971 * may dereference the passed in work.
973 spin_lock_irqsave(&worker
->lock
, flags
);
974 if (worker
->cur_work
&&
975 !(worker
->cur_work
->flags
& IO_WQ_WORK_NO_CANCEL
) &&
976 match
->fn(worker
->cur_work
, match
->data
)) {
977 send_sig(SIGINT
, worker
->task
, 1);
980 spin_unlock_irqrestore(&worker
->lock
, flags
);
982 return match
->nr_running
&& !match
->cancel_all
;
985 static inline void io_wqe_remove_pending(struct io_wqe
*wqe
,
986 struct io_wq_work
*work
,
987 struct io_wq_work_node
*prev
)
989 unsigned int hash
= io_get_work_hash(work
);
990 struct io_wq_work
*prev_work
= NULL
;
992 if (io_wq_is_hashed(work
) && work
== wqe
->hash_tail
[hash
]) {
994 prev_work
= container_of(prev
, struct io_wq_work
, list
);
995 if (prev_work
&& io_get_work_hash(prev_work
) == hash
)
996 wqe
->hash_tail
[hash
] = prev_work
;
998 wqe
->hash_tail
[hash
] = NULL
;
1000 wq_list_del(&wqe
->work_list
, &work
->list
, prev
);
1003 static void io_wqe_cancel_pending_work(struct io_wqe
*wqe
,
1004 struct io_cb_cancel_data
*match
)
1006 struct io_wq_work_node
*node
, *prev
;
1007 struct io_wq_work
*work
;
1008 unsigned long flags
;
1011 raw_spin_lock_irqsave(&wqe
->lock
, flags
);
1012 wq_list_for_each(node
, prev
, &wqe
->work_list
) {
1013 work
= container_of(node
, struct io_wq_work
, list
);
1014 if (!match
->fn(work
, match
->data
))
1016 io_wqe_remove_pending(wqe
, work
, prev
);
1017 raw_spin_unlock_irqrestore(&wqe
->lock
, flags
);
1018 io_run_cancel(work
, wqe
);
1019 match
->nr_pending
++;
1020 if (!match
->cancel_all
)
1023 /* not safe to continue after unlock */
1026 raw_spin_unlock_irqrestore(&wqe
->lock
, flags
);
1029 static void io_wqe_cancel_running_work(struct io_wqe
*wqe
,
1030 struct io_cb_cancel_data
*match
)
1033 io_wq_for_each_worker(wqe
, io_wq_worker_cancel
, match
);
1037 enum io_wq_cancel
io_wq_cancel_cb(struct io_wq
*wq
, work_cancel_fn
*cancel
,
1038 void *data
, bool cancel_all
)
1040 struct io_cb_cancel_data match
= {
1043 .cancel_all
= cancel_all
,
1048 * First check pending list, if we're lucky we can just remove it
1049 * from there. CANCEL_OK means that the work is returned as-new,
1050 * no completion will be posted for it.
1052 for_each_node(node
) {
1053 struct io_wqe
*wqe
= wq
->wqes
[node
];
1055 io_wqe_cancel_pending_work(wqe
, &match
);
1056 if (match
.nr_pending
&& !match
.cancel_all
)
1057 return IO_WQ_CANCEL_OK
;
1061 * Now check if a free (going busy) or busy worker has the work
1062 * currently running. If we find it there, we'll return CANCEL_RUNNING
1063 * as an indication that we attempt to signal cancellation. The
1064 * completion will run normally in this case.
1066 for_each_node(node
) {
1067 struct io_wqe
*wqe
= wq
->wqes
[node
];
1069 io_wqe_cancel_running_work(wqe
, &match
);
1070 if (match
.nr_running
&& !match
.cancel_all
)
1071 return IO_WQ_CANCEL_RUNNING
;
1074 if (match
.nr_running
)
1075 return IO_WQ_CANCEL_RUNNING
;
1076 if (match
.nr_pending
)
1077 return IO_WQ_CANCEL_OK
;
1078 return IO_WQ_CANCEL_NOTFOUND
;
1081 static bool io_wq_io_cb_cancel_data(struct io_wq_work
*work
, void *data
)
1083 return work
== data
;
1086 enum io_wq_cancel
io_wq_cancel_work(struct io_wq
*wq
, struct io_wq_work
*cwork
)
1088 return io_wq_cancel_cb(wq
, io_wq_io_cb_cancel_data
, (void *)cwork
, false);
1091 struct io_wq
*io_wq_create(unsigned bounded
, struct io_wq_data
*data
)
1093 int ret
= -ENOMEM
, node
;
1096 if (WARN_ON_ONCE(!data
->free_work
|| !data
->do_work
))
1097 return ERR_PTR(-EINVAL
);
1099 wq
= kzalloc(sizeof(*wq
), GFP_KERNEL
);
1101 return ERR_PTR(-ENOMEM
);
1103 wq
->wqes
= kcalloc(nr_node_ids
, sizeof(struct io_wqe
*), GFP_KERNEL
);
1107 ret
= cpuhp_state_add_instance_nocalls(io_wq_online
, &wq
->cpuhp_node
);
1111 wq
->free_work
= data
->free_work
;
1112 wq
->do_work
= data
->do_work
;
1114 /* caller must already hold a reference to this */
1115 wq
->user
= data
->user
;
1118 for_each_node(node
) {
1120 int alloc_node
= node
;
1122 if (!node_online(alloc_node
))
1123 alloc_node
= NUMA_NO_NODE
;
1124 wqe
= kzalloc_node(sizeof(struct io_wqe
), GFP_KERNEL
, alloc_node
);
1127 wq
->wqes
[node
] = wqe
;
1128 wqe
->node
= alloc_node
;
1129 wqe
->acct
[IO_WQ_ACCT_BOUND
].max_workers
= bounded
;
1130 atomic_set(&wqe
->acct
[IO_WQ_ACCT_BOUND
].nr_running
, 0);
1132 wqe
->acct
[IO_WQ_ACCT_UNBOUND
].max_workers
=
1133 task_rlimit(current
, RLIMIT_NPROC
);
1135 atomic_set(&wqe
->acct
[IO_WQ_ACCT_UNBOUND
].nr_running
, 0);
1137 raw_spin_lock_init(&wqe
->lock
);
1138 INIT_WQ_LIST(&wqe
->work_list
);
1139 INIT_HLIST_NULLS_HEAD(&wqe
->free_list
, 0);
1140 INIT_LIST_HEAD(&wqe
->all_list
);
1143 init_completion(&wq
->done
);
1145 wq
->manager
= kthread_create(io_wq_manager
, wq
, "io_wq_manager");
1146 if (!IS_ERR(wq
->manager
)) {
1147 wake_up_process(wq
->manager
);
1148 wait_for_completion(&wq
->done
);
1149 if (test_bit(IO_WQ_BIT_ERROR
, &wq
->state
)) {
1153 refcount_set(&wq
->use_refs
, 1);
1154 reinit_completion(&wq
->done
);
1158 ret
= PTR_ERR(wq
->manager
);
1159 complete(&wq
->done
);
1161 cpuhp_state_remove_instance_nocalls(io_wq_online
, &wq
->cpuhp_node
);
1163 kfree(wq
->wqes
[node
]);
1168 return ERR_PTR(ret
);
1171 bool io_wq_get(struct io_wq
*wq
, struct io_wq_data
*data
)
1173 if (data
->free_work
!= wq
->free_work
|| data
->do_work
!= wq
->do_work
)
1176 return refcount_inc_not_zero(&wq
->use_refs
);
1179 static void __io_wq_destroy(struct io_wq
*wq
)
1183 cpuhp_state_remove_instance_nocalls(io_wq_online
, &wq
->cpuhp_node
);
1185 set_bit(IO_WQ_BIT_EXIT
, &wq
->state
);
1187 kthread_stop(wq
->manager
);
1191 io_wq_for_each_worker(wq
->wqes
[node
], io_wq_worker_wake
, NULL
);
1194 wait_for_completion(&wq
->done
);
1197 kfree(wq
->wqes
[node
]);
1202 void io_wq_destroy(struct io_wq
*wq
)
1204 if (refcount_dec_and_test(&wq
->use_refs
))
1205 __io_wq_destroy(wq
);
1208 struct task_struct
*io_wq_get_task(struct io_wq
*wq
)
1213 static bool io_wq_worker_affinity(struct io_worker
*worker
, void *data
)
1215 struct task_struct
*task
= worker
->task
;
1219 rq
= task_rq_lock(task
, &rf
);
1220 do_set_cpus_allowed(task
, cpumask_of_node(worker
->wqe
->node
));
1221 task
->flags
|= PF_NO_SETAFFINITY
;
1222 task_rq_unlock(rq
, task
, &rf
);
1226 static int io_wq_cpu_online(unsigned int cpu
, struct hlist_node
*node
)
1228 struct io_wq
*wq
= hlist_entry_safe(node
, struct io_wq
, cpuhp_node
);
1233 io_wq_for_each_worker(wq
->wqes
[i
], io_wq_worker_affinity
, NULL
);
1238 static __init
int io_wq_init(void)
1242 ret
= cpuhp_setup_state_multi(CPUHP_AP_ONLINE_DYN
, "io-wq/online",
1243 io_wq_cpu_online
, NULL
);
1249 subsys_initcall(io_wq_init
);