1// SPDX-License-Identifier: GPL-2.0
2/*
3 * Shared application/kernel submission and completion ring pairs, for
4 * supporting fast/efficient IO.
5 *
6 * A note on the read/write ordering memory barriers that are matched between
7 * the application and kernel side.
8 *
9 * After the application reads the CQ ring tail, it must use an
10 * appropriate smp_rmb() to pair with the smp_wmb() the kernel uses
11 * before writing the tail (using smp_load_acquire to read the tail will
12 * do). It also needs a smp_mb() before updating CQ head (ordering the
13 * entry load(s) with the head store), pairing with an implicit barrier
14 * through a control-dependency in io_get_cqe (smp_store_release to
15 * store head will do). Failure to do so could lead to reading invalid
16 * CQ entries.
17 *
18 * Likewise, the application must use an appropriate smp_wmb() before
19 * writing the SQ tail (ordering SQ entry stores with the tail store),
20 * which pairs with smp_load_acquire in io_get_sqring (smp_store_release
21 * to store the tail will do). And it needs a barrier ordering the SQ
22 * head load before writing new SQ entries (smp_load_acquire to read
23 * head will do).
24 *
25 * When using the SQ poll thread (IORING_SETUP_SQPOLL), the application
26 * needs to check the SQ flags for IORING_SQ_NEED_WAKEUP *after*
27 * updating the SQ tail; a full memory barrier smp_mb() is needed
28 * between.
29 *
30 * Also see the examples in the liburing library:
31 *
32 * git://git.kernel.org/pub/scm/linux/kernel/git/axboe/liburing.git
33 *
34 * io_uring also uses READ/WRITE_ONCE() for _any_ store or load that happens
35 * from data shared between the kernel and application. This is done both
36 * for ordering purposes, but also to ensure that once a value is loaded from
37 * data that the application could potentially modify, it remains stable.
38 *
39 * Copyright (C) 2018-2019 Jens Axboe
40 * Copyright (c) 2018-2019 Christoph Hellwig
41 */
42#include <linux/kernel.h>
43#include <linux/init.h>
44#include <linux/errno.h>
45#include <linux/syscalls.h>
46#include <net/compat.h>
47#include <linux/refcount.h>
48#include <linux/uio.h>
49#include <linux/bits.h>
50
51#include <linux/sched/signal.h>
52#include <linux/fs.h>
53#include <linux/file.h>
54#include <linux/mm.h>
55#include <linux/mman.h>
56#include <linux/percpu.h>
57#include <linux/slab.h>
58#include <linux/bvec.h>
59#include <linux/net.h>
60#include <net/sock.h>
61#include <linux/anon_inodes.h>
62#include <linux/sched/mm.h>
63#include <linux/uaccess.h>
64#include <linux/nospec.h>
65#include <linux/fsnotify.h>
66#include <linux/fadvise.h>
67#include <linux/task_work.h>
68#include <linux/io_uring.h>
69#include <linux/io_uring/cmd.h>
70#include <linux/audit.h>
71#include <linux/security.h>
72#include <linux/jump_label.h>
73#include <asm/shmparam.h>
74
75#define CREATE_TRACE_POINTS
76#include <trace/events/io_uring.h>
77
78#include <uapi/linux/io_uring.h>
79
80#include "io-wq.h"
81
82#include "filetable.h"
83#include "io_uring.h"
84#include "opdef.h"
85#include "refs.h"
86#include "tctx.h"
87#include "register.h"
88#include "sqpoll.h"
89#include "fdinfo.h"
90#include "kbuf.h"
91#include "rsrc.h"
92#include "cancel.h"
93#include "net.h"
94#include "notif.h"
95#include "waitid.h"
96#include "futex.h"
97#include "napi.h"
98#include "uring_cmd.h"
99#include "msg_ring.h"
100#include "memmap.h"
101#include "zcrx.h"
102
103#include "timeout.h"
104#include "poll.h"
105#include "rw.h"
106#include "alloc_cache.h"
107#include "eventfd.h"
108
109#define SQE_COMMON_FLAGS (IOSQE_FIXED_FILE | IOSQE_IO_LINK | \
110 IOSQE_IO_HARDLINK | IOSQE_ASYNC)
111
112#define IO_REQ_LINK_FLAGS (REQ_F_LINK | REQ_F_HARDLINK)
113
114#define IO_REQ_CLEAN_FLAGS (REQ_F_BUFFER_SELECTED | REQ_F_NEED_CLEANUP | \
115 REQ_F_INFLIGHT | REQ_F_CREDS | REQ_F_ASYNC_DATA)
116
117#define IO_REQ_CLEAN_SLOW_FLAGS (REQ_F_REFCOUNT | IO_REQ_LINK_FLAGS | \
118 REQ_F_REISSUE | REQ_F_POLLED | \
119 IO_REQ_CLEAN_FLAGS)
120
121#define IO_TCTX_REFS_CACHE_NR (1U << 10)
122
123#define IO_COMPL_BATCH 32
124#define IO_REQ_ALLOC_BATCH 8
125#define IO_LOCAL_TW_DEFAULT_MAX 20
126
127struct io_defer_entry {
128 struct list_head list;
129 struct io_kiocb *req;
130};
131
132/* requests with any of those set should undergo io_disarm_next() */
133#define IO_DISARM_MASK (REQ_F_ARM_LTIMEOUT | REQ_F_LINK_TIMEOUT | REQ_F_FAIL)
134
135/*
136 * No waiters. It's larger than any valid value of the tw counter
137 * so that tests against ->cq_wait_nr would fail and skip wake_up().
138 */
139#define IO_CQ_WAKE_INIT (-1U)
140/* Forced wake up if there is a waiter regardless of ->cq_wait_nr */
141#define IO_CQ_WAKE_FORCE (IO_CQ_WAKE_INIT >> 1)
142
143static bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx,
144 struct io_uring_task *tctx,
145 bool cancel_all,
146 bool is_sqpoll_thread);
147
148static void io_queue_sqe(struct io_kiocb *req, unsigned int extra_flags);
149static void __io_req_caches_free(struct io_ring_ctx *ctx);
150
151static __read_mostly DEFINE_STATIC_KEY_FALSE(io_key_has_sqarray);
152
153struct kmem_cache *req_cachep;
154static struct workqueue_struct *iou_wq __ro_after_init;
155
156static int __read_mostly sysctl_io_uring_disabled;
157static int __read_mostly sysctl_io_uring_group = -1;
158
159#ifdef CONFIG_SYSCTL
160static const struct ctl_table kernel_io_uring_disabled_table[] = {
161 {
162 .procname = "io_uring_disabled",
163 .data = &sysctl_io_uring_disabled,
164 .maxlen = sizeof(sysctl_io_uring_disabled),
165 .mode = 0644,
166 .proc_handler = proc_dointvec_minmax,
167 .extra1 = SYSCTL_ZERO,
168 .extra2 = SYSCTL_TWO,
169 },
170 {
171 .procname = "io_uring_group",
172 .data = &sysctl_io_uring_group,
173 .maxlen = sizeof(gid_t),
174 .mode = 0644,
175 .proc_handler = proc_dointvec,
176 },
177};
178#endif
179
180static void io_poison_cached_req(struct io_kiocb *req)
181{
182 req->ctx = IO_URING_PTR_POISON;
183 req->tctx = IO_URING_PTR_POISON;
184 req->file = IO_URING_PTR_POISON;
185 req->creds = IO_URING_PTR_POISON;
186 req->io_task_work.func = IO_URING_PTR_POISON;
187 req->apoll = IO_URING_PTR_POISON;
188}
189
190static void io_poison_req(struct io_kiocb *req)
191{
192 io_poison_cached_req(req);
193 req->async_data = IO_URING_PTR_POISON;
194 req->kbuf = IO_URING_PTR_POISON;
195 req->comp_list.next = IO_URING_PTR_POISON;
196 req->file_node = IO_URING_PTR_POISON;
197 req->link = IO_URING_PTR_POISON;
198}
199
200static inline unsigned int __io_cqring_events(struct io_ring_ctx *ctx)
201{
202 return ctx->cached_cq_tail - READ_ONCE(ctx->rings->cq.head);
203}
204
205static inline unsigned int __io_cqring_events_user(struct io_ring_ctx *ctx)
206{
207 return READ_ONCE(ctx->rings->cq.tail) - READ_ONCE(ctx->rings->cq.head);
208}
209
210static bool io_match_linked(struct io_kiocb *head)
211{
212 struct io_kiocb *req;
213
214 io_for_each_link(req, head) {
215 if (req->flags & REQ_F_INFLIGHT)
216 return true;
217 }
218 return false;
219}
220
221/*
222 * As io_match_task() but protected against racing with linked timeouts.
223 * User must not hold timeout_lock.
224 */
225bool io_match_task_safe(struct io_kiocb *head, struct io_uring_task *tctx,
226 bool cancel_all)
227{
228 bool matched;
229
230 if (tctx && head->tctx != tctx)
231 return false;
232 if (cancel_all)
233 return true;
234
235 if (head->flags & REQ_F_LINK_TIMEOUT) {
236 struct io_ring_ctx *ctx = head->ctx;
237
238 /* protect against races with linked timeouts */
239 raw_spin_lock_irq(&ctx->timeout_lock);
240 matched = io_match_linked(head);
241 raw_spin_unlock_irq(&ctx->timeout_lock);
242 } else {
243 matched = io_match_linked(head);
244 }
245 return matched;
246}
247
248static inline void req_fail_link_node(struct io_kiocb *req, int res)
249{
250 req_set_fail(req);
251 io_req_set_res(req, res, cflags: 0);
252}
253
254static inline void io_req_add_to_cache(struct io_kiocb *req, struct io_ring_ctx *ctx)
255{
256 if (IS_ENABLED(CONFIG_KASAN))
257 io_poison_cached_req(req);
258 wq_stack_add_head(node: &req->comp_list, stack: &ctx->submit_state.free_list);
259}
260
261static __cold void io_ring_ctx_ref_free(struct percpu_ref *ref)
262{
263 struct io_ring_ctx *ctx = container_of(ref, struct io_ring_ctx, refs);
264
265 complete(&ctx->ref_comp);
266}
267
268static __cold void io_fallback_req_func(struct work_struct *work)
269{
270 struct io_ring_ctx *ctx = container_of(work, struct io_ring_ctx,
271 fallback_work.work);
272 struct llist_node *node = llist_del_all(head: &ctx->fallback_llist);
273 struct io_kiocb *req, *tmp;
274 struct io_tw_state ts = {};
275
276 percpu_ref_get(ref: &ctx->refs);
277 mutex_lock(lock: &ctx->uring_lock);
278 llist_for_each_entry_safe(req, tmp, node, io_task_work.node)
279 req->io_task_work.func(req, ts);
280 io_submit_flush_completions(ctx);
281 mutex_unlock(lock: &ctx->uring_lock);
282 percpu_ref_put(ref: &ctx->refs);
283}
284
285static int io_alloc_hash_table(struct io_hash_table *table, unsigned bits)
286{
287 unsigned int hash_buckets;
288 int i;
289
290 do {
291 hash_buckets = 1U << bits;
292 table->hbs = kvmalloc_array(hash_buckets, sizeof(table->hbs[0]),
293 GFP_KERNEL_ACCOUNT);
294 if (table->hbs)
295 break;
296 if (bits == 1)
297 return -ENOMEM;
298 bits--;
299 } while (1);
300
301 table->hash_bits = bits;
302 for (i = 0; i < hash_buckets; i++)
303 INIT_HLIST_HEAD(&table->hbs[i].list);
304 return 0;
305}
306
307static void io_free_alloc_caches(struct io_ring_ctx *ctx)
308{
309 io_alloc_cache_free(cache: &ctx->apoll_cache, free: kfree);
310 io_alloc_cache_free(cache: &ctx->netmsg_cache, free: io_netmsg_cache_free);
311 io_alloc_cache_free(cache: &ctx->rw_cache, free: io_rw_cache_free);
312 io_alloc_cache_free(cache: &ctx->cmd_cache, free: io_cmd_cache_free);
313 io_futex_cache_free(ctx);
314 io_rsrc_cache_free(ctx);
315}
316
317static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
318{
319 struct io_ring_ctx *ctx;
320 int hash_bits;
321 bool ret;
322
323 ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
324 if (!ctx)
325 return NULL;
326
327 xa_init(xa: &ctx->io_bl_xa);
328
329 /*
330 * Use 5 bits less than the max cq entries, that should give us around
331 * 32 entries per hash list if totally full and uniformly spread, but
332 * don't keep too many buckets to not overconsume memory.
333 */
334 hash_bits = ilog2(p->cq_entries) - 5;
335 hash_bits = clamp(hash_bits, 1, 8);
336 if (io_alloc_hash_table(table: &ctx->cancel_table, bits: hash_bits))
337 goto err;
338 if (percpu_ref_init(ref: &ctx->refs, release: io_ring_ctx_ref_free,
339 flags: 0, GFP_KERNEL))
340 goto err;
341
342 ctx->flags = p->flags;
343 ctx->hybrid_poll_time = LLONG_MAX;
344 atomic_set(v: &ctx->cq_wait_nr, IO_CQ_WAKE_INIT);
345 init_waitqueue_head(&ctx->sqo_sq_wait);
346 INIT_LIST_HEAD(list: &ctx->sqd_list);
347 INIT_LIST_HEAD(list: &ctx->cq_overflow_list);
348 ret = io_alloc_cache_init(cache: &ctx->apoll_cache, IO_POLL_ALLOC_CACHE_MAX,
349 size: sizeof(struct async_poll), init_bytes: 0);
350 ret |= io_alloc_cache_init(cache: &ctx->netmsg_cache, IO_ALLOC_CACHE_MAX,
351 size: sizeof(struct io_async_msghdr),
352 offsetof(struct io_async_msghdr, clear));
353 ret |= io_alloc_cache_init(cache: &ctx->rw_cache, IO_ALLOC_CACHE_MAX,
354 size: sizeof(struct io_async_rw),
355 offsetof(struct io_async_rw, clear));
356 ret |= io_alloc_cache_init(cache: &ctx->cmd_cache, IO_ALLOC_CACHE_MAX,
357 size: sizeof(struct io_async_cmd),
358 init_bytes: sizeof(struct io_async_cmd));
359 ret |= io_futex_cache_init(ctx);
360 ret |= io_rsrc_cache_init(ctx);
361 if (ret)
362 goto free_ref;
363 init_completion(x: &ctx->ref_comp);
364 xa_init_flags(xa: &ctx->personalities, XA_FLAGS_ALLOC1);
365 mutex_init(&ctx->uring_lock);
366 init_waitqueue_head(&ctx->cq_wait);
367 init_waitqueue_head(&ctx->poll_wq);
368 spin_lock_init(&ctx->completion_lock);
369 raw_spin_lock_init(&ctx->timeout_lock);
370 INIT_WQ_LIST(&ctx->iopoll_list);
371 INIT_LIST_HEAD(list: &ctx->defer_list);
372 INIT_LIST_HEAD(list: &ctx->timeout_list);
373 INIT_LIST_HEAD(list: &ctx->ltimeout_list);
374 init_llist_head(list: &ctx->work_llist);
375 INIT_LIST_HEAD(list: &ctx->tctx_list);
376 ctx->submit_state.free_list.next = NULL;
377 INIT_HLIST_HEAD(&ctx->waitid_list);
378 xa_init_flags(xa: &ctx->zcrx_ctxs, XA_FLAGS_ALLOC);
379#ifdef CONFIG_FUTEX
380 INIT_HLIST_HEAD(&ctx->futex_list);
381#endif
382 INIT_DELAYED_WORK(&ctx->fallback_work, io_fallback_req_func);
383 INIT_WQ_LIST(&ctx->submit_state.compl_reqs);
384 INIT_HLIST_HEAD(&ctx->cancelable_uring_cmd);
385 io_napi_init(ctx);
386 mutex_init(&ctx->mmap_lock);
387
388 return ctx;
389
390free_ref:
391 percpu_ref_exit(ref: &ctx->refs);
392err:
393 io_free_alloc_caches(ctx);
394 kvfree(addr: ctx->cancel_table.hbs);
395 xa_destroy(&ctx->io_bl_xa);
396 kfree(objp: ctx);
397 return NULL;
398}
399
400static void io_clean_op(struct io_kiocb *req)
401{
402 if (unlikely(req->flags & REQ_F_BUFFER_SELECTED))
403 io_kbuf_drop_legacy(req);
404
405 if (req->flags & REQ_F_NEED_CLEANUP) {
406 const struct io_cold_def *def = &io_cold_defs[req->opcode];
407
408 if (def->cleanup)
409 def->cleanup(req);
410 }
411 if (req->flags & REQ_F_INFLIGHT)
412 atomic_dec(v: &req->tctx->inflight_tracked);
413 if (req->flags & REQ_F_CREDS)
414 put_cred(cred: req->creds);
415 if (req->flags & REQ_F_ASYNC_DATA) {
416 kfree(objp: req->async_data);
417 req->async_data = NULL;
418 }
419 req->flags &= ~IO_REQ_CLEAN_FLAGS;
420}
421
422/*
423 * Mark the request as inflight, so that file cancelation will find it.
424 * Can be used if the file is an io_uring instance, or if the request itself
425 * relies on ->mm being alive for the duration of the request.
426 */
427inline void io_req_track_inflight(struct io_kiocb *req)
428{
429 if (!(req->flags & REQ_F_INFLIGHT)) {
430 req->flags |= REQ_F_INFLIGHT;
431 atomic_inc(v: &req->tctx->inflight_tracked);
432 }
433}
434
435static struct io_kiocb *__io_prep_linked_timeout(struct io_kiocb *req)
436{
437 if (WARN_ON_ONCE(!req->link))
438 return NULL;
439
440 req->flags &= ~REQ_F_ARM_LTIMEOUT;
441 req->flags |= REQ_F_LINK_TIMEOUT;
442
443 /* linked timeouts should have two refs once prep'ed */
444 io_req_set_refcount(req);
445 __io_req_set_refcount(req: req->link, nr: 2);
446 return req->link;
447}
448
449static void io_prep_async_work(struct io_kiocb *req)
450{
451 const struct io_issue_def *def = &io_issue_defs[req->opcode];
452 struct io_ring_ctx *ctx = req->ctx;
453
454 if (!(req->flags & REQ_F_CREDS)) {
455 req->flags |= REQ_F_CREDS;
456 req->creds = get_current_cred();
457 }
458
459 req->work.list.next = NULL;
460 atomic_set(v: &req->work.flags, i: 0);
461 if (req->flags & REQ_F_FORCE_ASYNC)
462 atomic_or(i: IO_WQ_WORK_CONCURRENT, v: &req->work.flags);
463
464 if (req->file && !(req->flags & REQ_F_FIXED_FILE))
465 req->flags |= io_file_get_flags(file: req->file);
466
467 if (req->file && (req->flags & REQ_F_ISREG)) {
468 bool should_hash = def->hash_reg_file;
469
470 /* don't serialize this request if the fs doesn't need it */
471 if (should_hash && (req->file->f_flags & O_DIRECT) &&
472 (req->file->f_op->fop_flags & FOP_DIO_PARALLEL_WRITE))
473 should_hash = false;
474 if (should_hash || (ctx->flags & IORING_SETUP_IOPOLL))
475 io_wq_hash_work(work: &req->work, val: file_inode(f: req->file));
476 } else if (!req->file || !S_ISBLK(file_inode(req->file)->i_mode)) {
477 if (def->unbound_nonreg_file)
478 atomic_or(i: IO_WQ_WORK_UNBOUND, v: &req->work.flags);
479 }
480}
481
482static void io_prep_async_link(struct io_kiocb *req)
483{
484 struct io_kiocb *cur;
485
486 if (req->flags & REQ_F_LINK_TIMEOUT) {
487 struct io_ring_ctx *ctx = req->ctx;
488
489 raw_spin_lock_irq(&ctx->timeout_lock);
490 io_for_each_link(cur, req)
491 io_prep_async_work(req: cur);
492 raw_spin_unlock_irq(&ctx->timeout_lock);
493 } else {
494 io_for_each_link(cur, req)
495 io_prep_async_work(req: cur);
496 }
497}
498
499static void io_queue_iowq(struct io_kiocb *req)
500{
501 struct io_uring_task *tctx = req->tctx;
502
503 BUG_ON(!tctx);
504
505 if ((current->flags & PF_KTHREAD) || !tctx->io_wq) {
506 io_req_task_queue_fail(req, ret: -ECANCELED);
507 return;
508 }
509
510 /* init ->work of the whole link before punting */
511 io_prep_async_link(req);
512
513 /*
514 * Not expected to happen, but if we do have a bug where this _can_
515 * happen, catch it here and ensure the request is marked as
516 * canceled. That will make io-wq go through the usual work cancel
517 * procedure rather than attempt to run this request (or create a new
518 * worker for it).
519 */
520 if (WARN_ON_ONCE(!same_thread_group(tctx->task, current)))
521 atomic_or(i: IO_WQ_WORK_CANCEL, v: &req->work.flags);
522
523 trace_io_uring_queue_async_work(req, rw: io_wq_is_hashed(work: &req->work));
524 io_wq_enqueue(wq: tctx->io_wq, work: &req->work);
525}
526
527static void io_req_queue_iowq_tw(struct io_kiocb *req, io_tw_token_t tw)
528{
529 io_queue_iowq(req);
530}
531
532void io_req_queue_iowq(struct io_kiocb *req)
533{
534 req->io_task_work.func = io_req_queue_iowq_tw;
535 io_req_task_work_add(req);
536}
537
538static unsigned io_linked_nr(struct io_kiocb *req)
539{
540 struct io_kiocb *tmp;
541 unsigned nr = 0;
542
543 io_for_each_link(tmp, req)
544 nr++;
545 return nr;
546}
547
548static __cold noinline void io_queue_deferred(struct io_ring_ctx *ctx)
549{
550 bool drain_seen = false, first = true;
551
552 lockdep_assert_held(&ctx->uring_lock);
553 __io_req_caches_free(ctx);
554
555 while (!list_empty(head: &ctx->defer_list)) {
556 struct io_defer_entry *de = list_first_entry(&ctx->defer_list,
557 struct io_defer_entry, list);
558
559 drain_seen |= de->req->flags & REQ_F_IO_DRAIN;
560 if ((drain_seen || first) && ctx->nr_req_allocated != ctx->nr_drained)
561 return;
562
563 list_del_init(entry: &de->list);
564 ctx->nr_drained -= io_linked_nr(req: de->req);
565 io_req_task_queue(req: de->req);
566 kfree(objp: de);
567 first = false;
568 }
569}
570
571void __io_commit_cqring_flush(struct io_ring_ctx *ctx)
572{
573 if (ctx->poll_activated)
574 io_poll_wq_wake(ctx);
575 if (ctx->off_timeout_used)
576 io_flush_timeouts(ctx);
577 if (ctx->has_evfd)
578 io_eventfd_signal(ctx, cqe_event: true);
579}
580
581static inline void __io_cq_lock(struct io_ring_ctx *ctx)
582{
583 if (!ctx->lockless_cq)
584 spin_lock(lock: &ctx->completion_lock);
585}
586
587static inline void io_cq_lock(struct io_ring_ctx *ctx)
588 __acquires(ctx->completion_lock)
589{
590 spin_lock(lock: &ctx->completion_lock);
591}
592
593static inline void __io_cq_unlock_post(struct io_ring_ctx *ctx)
594{
595 io_commit_cqring(ctx);
596 if (!ctx->task_complete) {
597 if (!ctx->lockless_cq)
598 spin_unlock(lock: &ctx->completion_lock);
599 /* IOPOLL rings only need to wake up if it's also SQPOLL */
600 if (!ctx->syscall_iopoll)
601 io_cqring_wake(ctx);
602 }
603 io_commit_cqring_flush(ctx);
604}
605
606static void io_cq_unlock_post(struct io_ring_ctx *ctx)
607 __releases(ctx->completion_lock)
608{
609 io_commit_cqring(ctx);
610 spin_unlock(lock: &ctx->completion_lock);
611 io_cqring_wake(ctx);
612 io_commit_cqring_flush(ctx);
613}
614
615static void __io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool dying)
616{
617 lockdep_assert_held(&ctx->uring_lock);
618
619 /* don't abort if we're dying, entries must get freed */
620 if (!dying && __io_cqring_events(ctx) == ctx->cq_entries)
621 return;
622
623 io_cq_lock(ctx);
624 while (!list_empty(head: &ctx->cq_overflow_list)) {
625 size_t cqe_size = sizeof(struct io_uring_cqe);
626 struct io_uring_cqe *cqe;
627 struct io_overflow_cqe *ocqe;
628 bool is_cqe32 = false;
629
630 ocqe = list_first_entry(&ctx->cq_overflow_list,
631 struct io_overflow_cqe, list);
632 if (ocqe->cqe.flags & IORING_CQE_F_32 ||
633 ctx->flags & IORING_SETUP_CQE32) {
634 is_cqe32 = true;
635 cqe_size <<= 1;
636 }
637
638 if (!dying) {
639 if (!io_get_cqe_overflow(ctx, ret: &cqe, overflow: true, cqe32: is_cqe32))
640 break;
641 memcpy(to: cqe, from: &ocqe->cqe, len: cqe_size);
642 }
643 list_del(entry: &ocqe->list);
644 kfree(objp: ocqe);
645
646 /*
647 * For silly syzbot cases that deliberately overflow by huge
648 * amounts, check if we need to resched and drop and
649 * reacquire the locks if so. Nothing real would ever hit this.
650 * Ideally we'd have a non-posting unlock for this, but hard
651 * to care for a non-real case.
652 */
653 if (need_resched()) {
654 ctx->cqe_sentinel = ctx->cqe_cached;
655 io_cq_unlock_post(ctx);
656 mutex_unlock(lock: &ctx->uring_lock);
657 cond_resched();
658 mutex_lock(lock: &ctx->uring_lock);
659 io_cq_lock(ctx);
660 }
661 }
662
663 if (list_empty(head: &ctx->cq_overflow_list)) {
664 clear_bit(nr: IO_CHECK_CQ_OVERFLOW_BIT, addr: &ctx->check_cq);
665 atomic_andnot(IORING_SQ_CQ_OVERFLOW, v: &ctx->rings->sq_flags);
666 }
667 io_cq_unlock_post(ctx);
668}
669
670static void io_cqring_overflow_kill(struct io_ring_ctx *ctx)
671{
672 if (ctx->rings)
673 __io_cqring_overflow_flush(ctx, dying: true);
674}
675
676static void io_cqring_do_overflow_flush(struct io_ring_ctx *ctx)
677{
678 mutex_lock(lock: &ctx->uring_lock);
679 __io_cqring_overflow_flush(ctx, dying: false);
680 mutex_unlock(lock: &ctx->uring_lock);
681}
682
683/* must to be called somewhat shortly after putting a request */
684static inline void io_put_task(struct io_kiocb *req)
685{
686 struct io_uring_task *tctx = req->tctx;
687
688 if (likely(tctx->task == current)) {
689 tctx->cached_refs++;
690 } else {
691 percpu_counter_sub(fbc: &tctx->inflight, amount: 1);
692 if (unlikely(atomic_read(&tctx->in_cancel)))
693 wake_up(&tctx->wait);
694 put_task_struct(t: tctx->task);
695 }
696}
697
698void io_task_refs_refill(struct io_uring_task *tctx)
699{
700 unsigned int refill = -tctx->cached_refs + IO_TCTX_REFS_CACHE_NR;
701
702 percpu_counter_add(fbc: &tctx->inflight, amount: refill);
703 refcount_add(i: refill, r: &current->usage);
704 tctx->cached_refs += refill;
705}
706
707static __cold void io_uring_drop_tctx_refs(struct task_struct *task)
708{
709 struct io_uring_task *tctx = task->io_uring;
710 unsigned int refs = tctx->cached_refs;
711
712 if (refs) {
713 tctx->cached_refs = 0;
714 percpu_counter_sub(fbc: &tctx->inflight, amount: refs);
715 put_task_struct_many(t: task, nr: refs);
716 }
717}
718
719static __cold bool io_cqring_add_overflow(struct io_ring_ctx *ctx,
720 struct io_overflow_cqe *ocqe)
721{
722 lockdep_assert_held(&ctx->completion_lock);
723
724 if (!ocqe) {
725 struct io_rings *r = ctx->rings;
726
727 /*
728 * If we're in ring overflow flush mode, or in task cancel mode,
729 * or cannot allocate an overflow entry, then we need to drop it
730 * on the floor.
731 */
732 WRITE_ONCE(r->cq_overflow, READ_ONCE(r->cq_overflow) + 1);
733 set_bit(nr: IO_CHECK_CQ_DROPPED_BIT, addr: &ctx->check_cq);
734 return false;
735 }
736 if (list_empty(head: &ctx->cq_overflow_list)) {
737 set_bit(nr: IO_CHECK_CQ_OVERFLOW_BIT, addr: &ctx->check_cq);
738 atomic_or(IORING_SQ_CQ_OVERFLOW, v: &ctx->rings->sq_flags);
739
740 }
741 list_add_tail(new: &ocqe->list, head: &ctx->cq_overflow_list);
742 return true;
743}
744
745static struct io_overflow_cqe *io_alloc_ocqe(struct io_ring_ctx *ctx,
746 struct io_cqe *cqe,
747 struct io_big_cqe *big_cqe, gfp_t gfp)
748{
749 struct io_overflow_cqe *ocqe;
750 size_t ocq_size = sizeof(struct io_overflow_cqe);
751 bool is_cqe32 = false;
752
753 if (cqe->flags & IORING_CQE_F_32 || ctx->flags & IORING_SETUP_CQE32) {
754 is_cqe32 = true;
755 ocq_size += sizeof(struct io_uring_cqe);
756 }
757
758 ocqe = kzalloc(ocq_size, gfp | __GFP_ACCOUNT);
759 trace_io_uring_cqe_overflow(ctx, user_data: cqe->user_data, res: cqe->res, cflags: cqe->flags, ocqe);
760 if (ocqe) {
761 ocqe->cqe.user_data = cqe->user_data;
762 ocqe->cqe.res = cqe->res;
763 ocqe->cqe.flags = cqe->flags;
764 if (is_cqe32 && big_cqe) {
765 ocqe->cqe.big_cqe[0] = big_cqe->extra1;
766 ocqe->cqe.big_cqe[1] = big_cqe->extra2;
767 }
768 }
769 if (big_cqe)
770 big_cqe->extra1 = big_cqe->extra2 = 0;
771 return ocqe;
772}
773
774/*
775 * Fill an empty dummy CQE, in case alignment is off for posting a 32b CQE
776 * because the ring is a single 16b entry away from wrapping.
777 */
778static bool io_fill_nop_cqe(struct io_ring_ctx *ctx, unsigned int off)
779{
780 if (__io_cqring_events(ctx) < ctx->cq_entries) {
781 struct io_uring_cqe *cqe = &ctx->rings->cqes[off];
782
783 cqe->user_data = 0;
784 cqe->res = 0;
785 cqe->flags = IORING_CQE_F_SKIP;
786 ctx->cached_cq_tail++;
787 return true;
788 }
789 return false;
790}
791
792/*
793 * writes to the cq entry need to come after reading head; the
794 * control dependency is enough as we're using WRITE_ONCE to
795 * fill the cq entry
796 */
797bool io_cqe_cache_refill(struct io_ring_ctx *ctx, bool overflow, bool cqe32)
798{
799 struct io_rings *rings = ctx->rings;
800 unsigned int off = ctx->cached_cq_tail & (ctx->cq_entries - 1);
801 unsigned int free, queued, len;
802
803 /*
804 * Posting into the CQ when there are pending overflowed CQEs may break
805 * ordering guarantees, which will affect links, F_MORE users and more.
806 * Force overflow the completion.
807 */
808 if (!overflow && (ctx->check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT)))
809 return false;
810
811 /*
812 * Post dummy CQE if a 32b CQE is needed and there's only room for a
813 * 16b CQE before the ring wraps.
814 */
815 if (cqe32 && off + 1 == ctx->cq_entries) {
816 if (!io_fill_nop_cqe(ctx, off))
817 return false;
818 off = 0;
819 }
820
821 /* userspace may cheat modifying the tail, be safe and do min */
822 queued = min(__io_cqring_events(ctx), ctx->cq_entries);
823 free = ctx->cq_entries - queued;
824 /* we need a contiguous range, limit based on the current array offset */
825 len = min(free, ctx->cq_entries - off);
826 if (len < (cqe32 + 1))
827 return false;
828
829 if (ctx->flags & IORING_SETUP_CQE32) {
830 off <<= 1;
831 len <<= 1;
832 }
833
834 ctx->cqe_cached = &rings->cqes[off];
835 ctx->cqe_sentinel = ctx->cqe_cached + len;
836 return true;
837}
838
839static bool io_fill_cqe_aux32(struct io_ring_ctx *ctx,
840 struct io_uring_cqe src_cqe[2])
841{
842 struct io_uring_cqe *cqe;
843
844 if (WARN_ON_ONCE(!(ctx->flags & (IORING_SETUP_CQE32|IORING_SETUP_CQE_MIXED))))
845 return false;
846 if (unlikely(!io_get_cqe(ctx, &cqe, true)))
847 return false;
848
849 memcpy(to: cqe, from: src_cqe, len: 2 * sizeof(*cqe));
850 trace_io_uring_complete(ctx, NULL, cqe);
851 return true;
852}
853
854static bool io_fill_cqe_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res,
855 u32 cflags)
856{
857 bool cqe32 = cflags & IORING_CQE_F_32;
858 struct io_uring_cqe *cqe;
859
860 if (likely(io_get_cqe(ctx, &cqe, cqe32))) {
861 WRITE_ONCE(cqe->user_data, user_data);
862 WRITE_ONCE(cqe->res, res);
863 WRITE_ONCE(cqe->flags, cflags);
864
865 if (cqe32) {
866 WRITE_ONCE(cqe->big_cqe[0], 0);
867 WRITE_ONCE(cqe->big_cqe[1], 0);
868 }
869
870 trace_io_uring_complete(ctx, NULL, cqe);
871 return true;
872 }
873 return false;
874}
875
876static inline struct io_cqe io_init_cqe(u64 user_data, s32 res, u32 cflags)
877{
878 return (struct io_cqe) { .user_data = user_data, .res = res, .flags = cflags };
879}
880
881static __cold void io_cqe_overflow(struct io_ring_ctx *ctx, struct io_cqe *cqe,
882 struct io_big_cqe *big_cqe)
883{
884 struct io_overflow_cqe *ocqe;
885
886 ocqe = io_alloc_ocqe(ctx, cqe, big_cqe, GFP_KERNEL);
887 spin_lock(lock: &ctx->completion_lock);
888 io_cqring_add_overflow(ctx, ocqe);
889 spin_unlock(lock: &ctx->completion_lock);
890}
891
892static __cold bool io_cqe_overflow_locked(struct io_ring_ctx *ctx,
893 struct io_cqe *cqe,
894 struct io_big_cqe *big_cqe)
895{
896 struct io_overflow_cqe *ocqe;
897
898 ocqe = io_alloc_ocqe(ctx, cqe, big_cqe, GFP_ATOMIC);
899 return io_cqring_add_overflow(ctx, ocqe);
900}
901
902bool io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
903{
904 bool filled;
905
906 io_cq_lock(ctx);
907 filled = io_fill_cqe_aux(ctx, user_data, res, cflags);
908 if (unlikely(!filled)) {
909 struct io_cqe cqe = io_init_cqe(user_data, res, cflags);
910
911 filled = io_cqe_overflow_locked(ctx, cqe: &cqe, NULL);
912 }
913 io_cq_unlock_post(ctx);
914 return filled;
915}
916
917/*
918 * Must be called from inline task_work so we now a flush will happen later,
919 * and obviously with ctx->uring_lock held (tw always has that).
920 */
921void io_add_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
922{
923 lockdep_assert_held(&ctx->uring_lock);
924 lockdep_assert(ctx->lockless_cq);
925
926 if (!io_fill_cqe_aux(ctx, user_data, res, cflags)) {
927 struct io_cqe cqe = io_init_cqe(user_data, res, cflags);
928
929 io_cqe_overflow(ctx, cqe: &cqe, NULL);
930 }
931 ctx->submit_state.cq_flush = true;
932}
933
934/*
935 * A helper for multishot requests posting additional CQEs.
936 * Should only be used from a task_work including IO_URING_F_MULTISHOT.
937 */
938bool io_req_post_cqe(struct io_kiocb *req, s32 res, u32 cflags)
939{
940 struct io_ring_ctx *ctx = req->ctx;
941 bool posted;
942
943 /*
944 * If multishot has already posted deferred completions, ensure that
945 * those are flushed first before posting this one. If not, CQEs
946 * could get reordered.
947 */
948 if (!wq_list_empty(&ctx->submit_state.compl_reqs))
949 __io_submit_flush_completions(ctx);
950
951 lockdep_assert(!io_wq_current_is_worker());
952 lockdep_assert_held(&ctx->uring_lock);
953
954 if (!ctx->lockless_cq) {
955 spin_lock(lock: &ctx->completion_lock);
956 posted = io_fill_cqe_aux(ctx, user_data: req->cqe.user_data, res, cflags);
957 spin_unlock(lock: &ctx->completion_lock);
958 } else {
959 posted = io_fill_cqe_aux(ctx, user_data: req->cqe.user_data, res, cflags);
960 }
961
962 ctx->submit_state.cq_flush = true;
963 return posted;
964}
965
966/*
967 * A helper for multishot requests posting additional CQEs.
968 * Should only be used from a task_work including IO_URING_F_MULTISHOT.
969 */
970bool io_req_post_cqe32(struct io_kiocb *req, struct io_uring_cqe cqe[2])
971{
972 struct io_ring_ctx *ctx = req->ctx;
973 bool posted;
974
975 lockdep_assert(!io_wq_current_is_worker());
976 lockdep_assert_held(&ctx->uring_lock);
977
978 cqe[0].user_data = req->cqe.user_data;
979 if (!ctx->lockless_cq) {
980 spin_lock(lock: &ctx->completion_lock);
981 posted = io_fill_cqe_aux32(ctx, src_cqe: cqe);
982 spin_unlock(lock: &ctx->completion_lock);
983 } else {
984 posted = io_fill_cqe_aux32(ctx, src_cqe: cqe);
985 }
986
987 ctx->submit_state.cq_flush = true;
988 return posted;
989}
990
991static void io_req_complete_post(struct io_kiocb *req, unsigned issue_flags)
992{
993 struct io_ring_ctx *ctx = req->ctx;
994 bool completed = true;
995
996 /*
997 * All execution paths but io-wq use the deferred completions by
998 * passing IO_URING_F_COMPLETE_DEFER and thus should not end up here.
999 */
1000 if (WARN_ON_ONCE(!(issue_flags & IO_URING_F_IOWQ)))
1001 return;
1002
1003 /*
1004 * Handle special CQ sync cases via task_work. DEFER_TASKRUN requires
1005 * the submitter task context, IOPOLL protects with uring_lock.
1006 */
1007 if (ctx->lockless_cq || (req->flags & REQ_F_REISSUE)) {
1008defer_complete:
1009 req->io_task_work.func = io_req_task_complete;
1010 io_req_task_work_add(req);
1011 return;
1012 }
1013
1014 io_cq_lock(ctx);
1015 if (!(req->flags & REQ_F_CQE_SKIP))
1016 completed = io_fill_cqe_req(ctx, req);
1017 io_cq_unlock_post(ctx);
1018
1019 if (!completed)
1020 goto defer_complete;
1021
1022 /*
1023 * We don't free the request here because we know it's called from
1024 * io-wq only, which holds a reference, so it cannot be the last put.
1025 */
1026 req_ref_put(req);
1027}
1028
1029void io_req_defer_failed(struct io_kiocb *req, s32 res)
1030 __must_hold(&ctx->uring_lock)
1031{
1032 const struct io_cold_def *def = &io_cold_defs[req->opcode];
1033
1034 lockdep_assert_held(&req->ctx->uring_lock);
1035
1036 req_set_fail(req);
1037 io_req_set_res(req, res, cflags: io_put_kbuf(req, len: res, NULL));
1038 if (def->fail)
1039 def->fail(req);
1040 io_req_complete_defer(req);
1041}
1042
1043/*
1044 * A request might get retired back into the request caches even before opcode
1045 * handlers and io_issue_sqe() are done with it, e.g. inline completion path.
1046 * Because of that, io_alloc_req() should be called only under ->uring_lock
1047 * and with extra caution to not get a request that is still worked on.
1048 */
1049__cold bool __io_alloc_req_refill(struct io_ring_ctx *ctx)
1050 __must_hold(&ctx->uring_lock)
1051{
1052 gfp_t gfp = GFP_KERNEL | __GFP_NOWARN | __GFP_ZERO;
1053 void *reqs[IO_REQ_ALLOC_BATCH];
1054 int ret;
1055
1056 ret = kmem_cache_alloc_bulk(req_cachep, gfp, ARRAY_SIZE(reqs), reqs);
1057
1058 /*
1059 * Bulk alloc is all-or-nothing. If we fail to get a batch,
1060 * retry single alloc to be on the safe side.
1061 */
1062 if (unlikely(ret <= 0)) {
1063 reqs[0] = kmem_cache_alloc(req_cachep, gfp);
1064 if (!reqs[0])
1065 return false;
1066 ret = 1;
1067 }
1068
1069 percpu_ref_get_many(ref: &ctx->refs, nr: ret);
1070 ctx->nr_req_allocated += ret;
1071
1072 while (ret--) {
1073 struct io_kiocb *req = reqs[ret];
1074
1075 io_req_add_to_cache(req, ctx);
1076 }
1077 return true;
1078}
1079
1080__cold void io_free_req(struct io_kiocb *req)
1081{
1082 /* refs were already put, restore them for io_req_task_complete() */
1083 req->flags &= ~REQ_F_REFCOUNT;
1084 /* we only want to free it, don't post CQEs */
1085 req->flags |= REQ_F_CQE_SKIP;
1086 req->io_task_work.func = io_req_task_complete;
1087 io_req_task_work_add(req);
1088}
1089
1090static void __io_req_find_next_prep(struct io_kiocb *req)
1091{
1092 struct io_ring_ctx *ctx = req->ctx;
1093
1094 spin_lock(lock: &ctx->completion_lock);
1095 io_disarm_next(req);
1096 spin_unlock(lock: &ctx->completion_lock);
1097}
1098
1099static inline struct io_kiocb *io_req_find_next(struct io_kiocb *req)
1100{
1101 struct io_kiocb *nxt;
1102
1103 /*
1104 * If LINK is set, we have dependent requests in this chain. If we
1105 * didn't fail this request, queue the first one up, moving any other
1106 * dependencies to the next request. In case of failure, fail the rest
1107 * of the chain.
1108 */
1109 if (unlikely(req->flags & IO_DISARM_MASK))
1110 __io_req_find_next_prep(req);
1111 nxt = req->link;
1112 req->link = NULL;
1113 return nxt;
1114}
1115
1116static void ctx_flush_and_put(struct io_ring_ctx *ctx, io_tw_token_t tw)
1117{
1118 if (!ctx)
1119 return;
1120 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1121 atomic_andnot(IORING_SQ_TASKRUN, v: &ctx->rings->sq_flags);
1122
1123 io_submit_flush_completions(ctx);
1124 mutex_unlock(lock: &ctx->uring_lock);
1125 percpu_ref_put(ref: &ctx->refs);
1126}
1127
1128/*
1129 * Run queued task_work, returning the number of entries processed in *count.
1130 * If more entries than max_entries are available, stop processing once this
1131 * is reached and return the rest of the list.
1132 */
1133struct llist_node *io_handle_tw_list(struct llist_node *node,
1134 unsigned int *count,
1135 unsigned int max_entries)
1136{
1137 struct io_ring_ctx *ctx = NULL;
1138 struct io_tw_state ts = { };
1139
1140 do {
1141 struct llist_node *next = node->next;
1142 struct io_kiocb *req = container_of(node, struct io_kiocb,
1143 io_task_work.node);
1144
1145 if (req->ctx != ctx) {
1146 ctx_flush_and_put(ctx, tw: ts);
1147 ctx = req->ctx;
1148 mutex_lock(lock: &ctx->uring_lock);
1149 percpu_ref_get(ref: &ctx->refs);
1150 }
1151 INDIRECT_CALL_2(req->io_task_work.func,
1152 io_poll_task_func, io_req_rw_complete,
1153 req, ts);
1154 node = next;
1155 (*count)++;
1156 if (unlikely(need_resched())) {
1157 ctx_flush_and_put(ctx, tw: ts);
1158 ctx = NULL;
1159 cond_resched();
1160 }
1161 } while (node && *count < max_entries);
1162
1163 ctx_flush_and_put(ctx, tw: ts);
1164 return node;
1165}
1166
1167static __cold void __io_fallback_tw(struct llist_node *node, bool sync)
1168{
1169 struct io_ring_ctx *last_ctx = NULL;
1170 struct io_kiocb *req;
1171
1172 while (node) {
1173 req = container_of(node, struct io_kiocb, io_task_work.node);
1174 node = node->next;
1175 if (last_ctx != req->ctx) {
1176 if (last_ctx) {
1177 if (sync)
1178 flush_delayed_work(dwork: &last_ctx->fallback_work);
1179 percpu_ref_put(ref: &last_ctx->refs);
1180 }
1181 last_ctx = req->ctx;
1182 percpu_ref_get(ref: &last_ctx->refs);
1183 }
1184 if (llist_add(new: &req->io_task_work.node, head: &last_ctx->fallback_llist))
1185 schedule_delayed_work(dwork: &last_ctx->fallback_work, delay: 1);
1186 }
1187
1188 if (last_ctx) {
1189 if (sync)
1190 flush_delayed_work(dwork: &last_ctx->fallback_work);
1191 percpu_ref_put(ref: &last_ctx->refs);
1192 }
1193}
1194
1195static void io_fallback_tw(struct io_uring_task *tctx, bool sync)
1196{
1197 struct llist_node *node = llist_del_all(head: &tctx->task_list);
1198
1199 __io_fallback_tw(node, sync);
1200}
1201
1202struct llist_node *tctx_task_work_run(struct io_uring_task *tctx,
1203 unsigned int max_entries,
1204 unsigned int *count)
1205{
1206 struct llist_node *node;
1207
1208 if (unlikely(current->flags & PF_EXITING)) {
1209 io_fallback_tw(tctx, sync: true);
1210 return NULL;
1211 }
1212
1213 node = llist_del_all(head: &tctx->task_list);
1214 if (node) {
1215 node = llist_reverse_order(head: node);
1216 node = io_handle_tw_list(node, count, max_entries);
1217 }
1218
1219 /* relaxed read is enough as only the task itself sets ->in_cancel */
1220 if (unlikely(atomic_read(&tctx->in_cancel)))
1221 io_uring_drop_tctx_refs(current);
1222
1223 trace_io_uring_task_work_run(tctx, count: *count);
1224 return node;
1225}
1226
1227void tctx_task_work(struct callback_head *cb)
1228{
1229 struct io_uring_task *tctx;
1230 struct llist_node *ret;
1231 unsigned int count = 0;
1232
1233 tctx = container_of(cb, struct io_uring_task, task_work);
1234 ret = tctx_task_work_run(tctx, UINT_MAX, count: &count);
1235 /* can't happen */
1236 WARN_ON_ONCE(ret);
1237}
1238
1239static void io_req_local_work_add(struct io_kiocb *req, unsigned flags)
1240{
1241 struct io_ring_ctx *ctx = req->ctx;
1242 unsigned nr_wait, nr_tw, nr_tw_prev;
1243 struct llist_node *head;
1244
1245 /* See comment above IO_CQ_WAKE_INIT */
1246 BUILD_BUG_ON(IO_CQ_WAKE_FORCE <= IORING_MAX_CQ_ENTRIES);
1247
1248 /*
1249 * We don't know how many reuqests is there in the link and whether
1250 * they can even be queued lazily, fall back to non-lazy.
1251 */
1252 if (req->flags & IO_REQ_LINK_FLAGS)
1253 flags &= ~IOU_F_TWQ_LAZY_WAKE;
1254
1255 guard(rcu)();
1256
1257 head = READ_ONCE(ctx->work_llist.first);
1258 do {
1259 nr_tw_prev = 0;
1260 if (head) {
1261 struct io_kiocb *first_req = container_of(head,
1262 struct io_kiocb,
1263 io_task_work.node);
1264 /*
1265 * Might be executed at any moment, rely on
1266 * SLAB_TYPESAFE_BY_RCU to keep it alive.
1267 */
1268 nr_tw_prev = READ_ONCE(first_req->nr_tw);
1269 }
1270
1271 /*
1272 * Theoretically, it can overflow, but that's fine as one of
1273 * previous adds should've tried to wake the task.
1274 */
1275 nr_tw = nr_tw_prev + 1;
1276 if (!(flags & IOU_F_TWQ_LAZY_WAKE))
1277 nr_tw = IO_CQ_WAKE_FORCE;
1278
1279 req->nr_tw = nr_tw;
1280 req->io_task_work.node.next = head;
1281 } while (!try_cmpxchg(&ctx->work_llist.first, &head,
1282 &req->io_task_work.node));
1283
1284 /*
1285 * cmpxchg implies a full barrier, which pairs with the barrier
1286 * in set_current_state() on the io_cqring_wait() side. It's used
1287 * to ensure that either we see updated ->cq_wait_nr, or waiters
1288 * going to sleep will observe the work added to the list, which
1289 * is similar to the wait/wawke task state sync.
1290 */
1291
1292 if (!head) {
1293 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1294 atomic_or(IORING_SQ_TASKRUN, v: &ctx->rings->sq_flags);
1295 if (ctx->has_evfd)
1296 io_eventfd_signal(ctx, cqe_event: false);
1297 }
1298
1299 nr_wait = atomic_read(v: &ctx->cq_wait_nr);
1300 /* not enough or no one is waiting */
1301 if (nr_tw < nr_wait)
1302 return;
1303 /* the previous add has already woken it up */
1304 if (nr_tw_prev >= nr_wait)
1305 return;
1306 wake_up_state(tsk: ctx->submitter_task, TASK_INTERRUPTIBLE);
1307}
1308
1309static void io_req_normal_work_add(struct io_kiocb *req)
1310{
1311 struct io_uring_task *tctx = req->tctx;
1312 struct io_ring_ctx *ctx = req->ctx;
1313
1314 /* task_work already pending, we're done */
1315 if (!llist_add(new: &req->io_task_work.node, head: &tctx->task_list))
1316 return;
1317
1318 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1319 atomic_or(IORING_SQ_TASKRUN, v: &ctx->rings->sq_flags);
1320
1321 /* SQPOLL doesn't need the task_work added, it'll run it itself */
1322 if (ctx->flags & IORING_SETUP_SQPOLL) {
1323 __set_notify_signal(task: tctx->task);
1324 return;
1325 }
1326
1327 if (likely(!task_work_add(tctx->task, &tctx->task_work, ctx->notify_method)))
1328 return;
1329
1330 io_fallback_tw(tctx, sync: false);
1331}
1332
1333void __io_req_task_work_add(struct io_kiocb *req, unsigned flags)
1334{
1335 if (req->ctx->flags & IORING_SETUP_DEFER_TASKRUN)
1336 io_req_local_work_add(req, flags);
1337 else
1338 io_req_normal_work_add(req);
1339}
1340
1341void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags)
1342{
1343 if (WARN_ON_ONCE(!(req->ctx->flags & IORING_SETUP_DEFER_TASKRUN)))
1344 return;
1345 __io_req_task_work_add(req, flags);
1346}
1347
1348static void __cold io_move_task_work_from_local(struct io_ring_ctx *ctx)
1349{
1350 struct llist_node *node = llist_del_all(head: &ctx->work_llist);
1351
1352 __io_fallback_tw(node, sync: false);
1353 node = llist_del_all(head: &ctx->retry_llist);
1354 __io_fallback_tw(node, sync: false);
1355}
1356
1357static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events,
1358 int min_events)
1359{
1360 if (!io_local_work_pending(ctx))
1361 return false;
1362 if (events < min_events)
1363 return true;
1364 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1365 atomic_or(IORING_SQ_TASKRUN, v: &ctx->rings->sq_flags);
1366 return false;
1367}
1368
1369static int __io_run_local_work_loop(struct llist_node **node,
1370 io_tw_token_t tw,
1371 int events)
1372{
1373 int ret = 0;
1374
1375 while (*node) {
1376 struct llist_node *next = (*node)->next;
1377 struct io_kiocb *req = container_of(*node, struct io_kiocb,
1378 io_task_work.node);
1379 INDIRECT_CALL_2(req->io_task_work.func,
1380 io_poll_task_func, io_req_rw_complete,
1381 req, tw);
1382 *node = next;
1383 if (++ret >= events)
1384 break;
1385 }
1386
1387 return ret;
1388}
1389
1390static int __io_run_local_work(struct io_ring_ctx *ctx, io_tw_token_t tw,
1391 int min_events, int max_events)
1392{
1393 struct llist_node *node;
1394 unsigned int loops = 0;
1395 int ret = 0;
1396
1397 if (WARN_ON_ONCE(ctx->submitter_task != current))
1398 return -EEXIST;
1399 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1400 atomic_andnot(IORING_SQ_TASKRUN, v: &ctx->rings->sq_flags);
1401again:
1402 min_events -= ret;
1403 ret = __io_run_local_work_loop(node: &ctx->retry_llist.first, tw, events: max_events);
1404 if (ctx->retry_llist.first)
1405 goto retry_done;
1406
1407 /*
1408 * llists are in reverse order, flip it back the right way before
1409 * running the pending items.
1410 */
1411 node = llist_reverse_order(head: llist_del_all(head: &ctx->work_llist));
1412 ret += __io_run_local_work_loop(node: &node, tw, events: max_events - ret);
1413 ctx->retry_llist.first = node;
1414 loops++;
1415
1416 if (io_run_local_work_continue(ctx, events: ret, min_events))
1417 goto again;
1418retry_done:
1419 io_submit_flush_completions(ctx);
1420 if (io_run_local_work_continue(ctx, events: ret, min_events))
1421 goto again;
1422
1423 trace_io_uring_local_work_run(ctx, count: ret, loops);
1424 return ret;
1425}
1426
1427static inline int io_run_local_work_locked(struct io_ring_ctx *ctx,
1428 int min_events)
1429{
1430 struct io_tw_state ts = {};
1431
1432 if (!io_local_work_pending(ctx))
1433 return 0;
1434 return __io_run_local_work(ctx, tw: ts, min_events,
1435 max(IO_LOCAL_TW_DEFAULT_MAX, min_events));
1436}
1437
1438static int io_run_local_work(struct io_ring_ctx *ctx, int min_events,
1439 int max_events)
1440{
1441 struct io_tw_state ts = {};
1442 int ret;
1443
1444 mutex_lock(lock: &ctx->uring_lock);
1445 ret = __io_run_local_work(ctx, tw: ts, min_events, max_events);
1446 mutex_unlock(lock: &ctx->uring_lock);
1447 return ret;
1448}
1449
1450static void io_req_task_cancel(struct io_kiocb *req, io_tw_token_t tw)
1451{
1452 io_tw_lock(ctx: req->ctx, tw);
1453 io_req_defer_failed(req, res: req->cqe.res);
1454}
1455
1456void io_req_task_submit(struct io_kiocb *req, io_tw_token_t tw)
1457{
1458 struct io_ring_ctx *ctx = req->ctx;
1459
1460 io_tw_lock(ctx, tw);
1461 if (unlikely(io_should_terminate_tw(ctx)))
1462 io_req_defer_failed(req, res: -EFAULT);
1463 else if (req->flags & REQ_F_FORCE_ASYNC)
1464 io_queue_iowq(req);
1465 else
1466 io_queue_sqe(req, extra_flags: 0);
1467}
1468
1469void io_req_task_queue_fail(struct io_kiocb *req, int ret)
1470{
1471 io_req_set_res(req, res: ret, cflags: 0);
1472 req->io_task_work.func = io_req_task_cancel;
1473 io_req_task_work_add(req);
1474}
1475
1476void io_req_task_queue(struct io_kiocb *req)
1477{
1478 req->io_task_work.func = io_req_task_submit;
1479 io_req_task_work_add(req);
1480}
1481
1482void io_queue_next(struct io_kiocb *req)
1483{
1484 struct io_kiocb *nxt = io_req_find_next(req);
1485
1486 if (nxt)
1487 io_req_task_queue(req: nxt);
1488}
1489
1490static inline void io_req_put_rsrc_nodes(struct io_kiocb *req)
1491{
1492 if (req->file_node) {
1493 io_put_rsrc_node(ctx: req->ctx, node: req->file_node);
1494 req->file_node = NULL;
1495 }
1496 if (req->flags & REQ_F_BUF_NODE)
1497 io_put_rsrc_node(ctx: req->ctx, node: req->buf_node);
1498}
1499
1500static void io_free_batch_list(struct io_ring_ctx *ctx,
1501 struct io_wq_work_node *node)
1502 __must_hold(&ctx->uring_lock)
1503{
1504 do {
1505 struct io_kiocb *req = container_of(node, struct io_kiocb,
1506 comp_list);
1507
1508 if (unlikely(req->flags & IO_REQ_CLEAN_SLOW_FLAGS)) {
1509 if (req->flags & REQ_F_REISSUE) {
1510 node = req->comp_list.next;
1511 req->flags &= ~REQ_F_REISSUE;
1512 io_queue_iowq(req);
1513 continue;
1514 }
1515 if (req->flags & REQ_F_REFCOUNT) {
1516 node = req->comp_list.next;
1517 if (!req_ref_put_and_test(req))
1518 continue;
1519 }
1520 if ((req->flags & REQ_F_POLLED) && req->apoll) {
1521 struct async_poll *apoll = req->apoll;
1522
1523 if (apoll->double_poll)
1524 kfree(objp: apoll->double_poll);
1525 io_cache_free(cache: &ctx->apoll_cache, obj: apoll);
1526 req->flags &= ~REQ_F_POLLED;
1527 }
1528 if (req->flags & IO_REQ_LINK_FLAGS)
1529 io_queue_next(req);
1530 if (unlikely(req->flags & IO_REQ_CLEAN_FLAGS))
1531 io_clean_op(req);
1532 }
1533 io_put_file(req);
1534 io_req_put_rsrc_nodes(req);
1535 io_put_task(req);
1536
1537 node = req->comp_list.next;
1538 io_req_add_to_cache(req, ctx);
1539 } while (node);
1540}
1541
1542void __io_submit_flush_completions(struct io_ring_ctx *ctx)
1543 __must_hold(&ctx->uring_lock)
1544{
1545 struct io_submit_state *state = &ctx->submit_state;
1546 struct io_wq_work_node *node;
1547
1548 __io_cq_lock(ctx);
1549 __wq_list_for_each(node, &state->compl_reqs) {
1550 struct io_kiocb *req = container_of(node, struct io_kiocb,
1551 comp_list);
1552
1553 /*
1554 * Requests marked with REQUEUE should not post a CQE, they
1555 * will go through the io-wq retry machinery and post one
1556 * later.
1557 */
1558 if (!(req->flags & (REQ_F_CQE_SKIP | REQ_F_REISSUE)) &&
1559 unlikely(!io_fill_cqe_req(ctx, req))) {
1560 if (ctx->lockless_cq)
1561 io_cqe_overflow(ctx, cqe: &req->cqe, big_cqe: &req->big_cqe);
1562 else
1563 io_cqe_overflow_locked(ctx, cqe: &req->cqe, big_cqe: &req->big_cqe);
1564 }
1565 }
1566 __io_cq_unlock_post(ctx);
1567
1568 if (!wq_list_empty(&state->compl_reqs)) {
1569 io_free_batch_list(ctx, node: state->compl_reqs.first);
1570 INIT_WQ_LIST(&state->compl_reqs);
1571 }
1572
1573 if (unlikely(ctx->drain_active))
1574 io_queue_deferred(ctx);
1575
1576 ctx->submit_state.cq_flush = false;
1577}
1578
1579static unsigned io_cqring_events(struct io_ring_ctx *ctx)
1580{
1581 /* See comment at the top of this file */
1582 smp_rmb();
1583 return __io_cqring_events(ctx);
1584}
1585
1586/*
1587 * We can't just wait for polled events to come to us, we have to actively
1588 * find and complete them.
1589 */
1590static __cold void io_iopoll_try_reap_events(struct io_ring_ctx *ctx)
1591{
1592 if (!(ctx->flags & IORING_SETUP_IOPOLL))
1593 return;
1594
1595 mutex_lock(lock: &ctx->uring_lock);
1596 while (!wq_list_empty(&ctx->iopoll_list)) {
1597 /* let it sleep and repeat later if can't complete a request */
1598 if (io_do_iopoll(ctx, force_nonspin: true) == 0)
1599 break;
1600 /*
1601 * Ensure we allow local-to-the-cpu processing to take place,
1602 * in this case we need to ensure that we reap all events.
1603 * Also let task_work, etc. to progress by releasing the mutex
1604 */
1605 if (need_resched()) {
1606 mutex_unlock(lock: &ctx->uring_lock);
1607 cond_resched();
1608 mutex_lock(lock: &ctx->uring_lock);
1609 }
1610 }
1611 mutex_unlock(lock: &ctx->uring_lock);
1612
1613 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
1614 io_move_task_work_from_local(ctx);
1615}
1616
1617static int io_iopoll_check(struct io_ring_ctx *ctx, unsigned int min_events)
1618{
1619 unsigned int nr_events = 0;
1620 unsigned long check_cq;
1621
1622 min_events = min(min_events, ctx->cq_entries);
1623
1624 lockdep_assert_held(&ctx->uring_lock);
1625
1626 if (!io_allowed_run_tw(ctx))
1627 return -EEXIST;
1628
1629 check_cq = READ_ONCE(ctx->check_cq);
1630 if (unlikely(check_cq)) {
1631 if (check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT))
1632 __io_cqring_overflow_flush(ctx, dying: false);
1633 /*
1634 * Similarly do not spin if we have not informed the user of any
1635 * dropped CQE.
1636 */
1637 if (check_cq & BIT(IO_CHECK_CQ_DROPPED_BIT))
1638 return -EBADR;
1639 }
1640 /*
1641 * Don't enter poll loop if we already have events pending.
1642 * If we do, we can potentially be spinning for commands that
1643 * already triggered a CQE (eg in error).
1644 */
1645 if (io_cqring_events(ctx))
1646 return 0;
1647
1648 do {
1649 int ret = 0;
1650
1651 /*
1652 * If a submit got punted to a workqueue, we can have the
1653 * application entering polling for a command before it gets
1654 * issued. That app will hold the uring_lock for the duration
1655 * of the poll right here, so we need to take a breather every
1656 * now and then to ensure that the issue has a chance to add
1657 * the poll to the issued list. Otherwise we can spin here
1658 * forever, while the workqueue is stuck trying to acquire the
1659 * very same mutex.
1660 */
1661 if (wq_list_empty(&ctx->iopoll_list) ||
1662 io_task_work_pending(ctx)) {
1663 u32 tail = ctx->cached_cq_tail;
1664
1665 (void) io_run_local_work_locked(ctx, min_events);
1666
1667 if (task_work_pending(current) ||
1668 wq_list_empty(&ctx->iopoll_list)) {
1669 mutex_unlock(lock: &ctx->uring_lock);
1670 io_run_task_work();
1671 mutex_lock(lock: &ctx->uring_lock);
1672 }
1673 /* some requests don't go through iopoll_list */
1674 if (tail != ctx->cached_cq_tail ||
1675 wq_list_empty(&ctx->iopoll_list))
1676 break;
1677 }
1678 ret = io_do_iopoll(ctx, force_nonspin: !min_events);
1679 if (unlikely(ret < 0))
1680 return ret;
1681
1682 if (task_sigpending(current))
1683 return -EINTR;
1684 if (need_resched())
1685 break;
1686
1687 nr_events += ret;
1688 } while (nr_events < min_events);
1689
1690 return 0;
1691}
1692
1693void io_req_task_complete(struct io_kiocb *req, io_tw_token_t tw)
1694{
1695 io_req_complete_defer(req);
1696}
1697
1698/*
1699 * After the iocb has been issued, it's safe to be found on the poll list.
1700 * Adding the kiocb to the list AFTER submission ensures that we don't
1701 * find it from a io_do_iopoll() thread before the issuer is done
1702 * accessing the kiocb cookie.
1703 */
1704static void io_iopoll_req_issued(struct io_kiocb *req, unsigned int issue_flags)
1705{
1706 struct io_ring_ctx *ctx = req->ctx;
1707 const bool needs_lock = issue_flags & IO_URING_F_UNLOCKED;
1708
1709 /* workqueue context doesn't hold uring_lock, grab it now */
1710 if (unlikely(needs_lock))
1711 mutex_lock(lock: &ctx->uring_lock);
1712
1713 /*
1714 * Track whether we have multiple files in our lists. This will impact
1715 * how we do polling eventually, not spinning if we're on potentially
1716 * different devices.
1717 */
1718 if (wq_list_empty(&ctx->iopoll_list)) {
1719 ctx->poll_multi_queue = false;
1720 } else if (!ctx->poll_multi_queue) {
1721 struct io_kiocb *list_req;
1722
1723 list_req = container_of(ctx->iopoll_list.first, struct io_kiocb,
1724 comp_list);
1725 if (list_req->file != req->file)
1726 ctx->poll_multi_queue = true;
1727 }
1728
1729 /*
1730 * For fast devices, IO may have already completed. If it has, add
1731 * it to the front so we find it first.
1732 */
1733 if (READ_ONCE(req->iopoll_completed))
1734 wq_list_add_head(node: &req->comp_list, list: &ctx->iopoll_list);
1735 else
1736 wq_list_add_tail(node: &req->comp_list, list: &ctx->iopoll_list);
1737
1738 if (unlikely(needs_lock)) {
1739 /*
1740 * If IORING_SETUP_SQPOLL is enabled, sqes are either handle
1741 * in sq thread task context or in io worker task context. If
1742 * current task context is sq thread, we don't need to check
1743 * whether should wake up sq thread.
1744 */
1745 if ((ctx->flags & IORING_SETUP_SQPOLL) &&
1746 wq_has_sleeper(wq_head: &ctx->sq_data->wait))
1747 wake_up(&ctx->sq_data->wait);
1748
1749 mutex_unlock(lock: &ctx->uring_lock);
1750 }
1751}
1752
1753io_req_flags_t io_file_get_flags(struct file *file)
1754{
1755 io_req_flags_t res = 0;
1756
1757 BUILD_BUG_ON(REQ_F_ISREG_BIT != REQ_F_SUPPORT_NOWAIT_BIT + 1);
1758
1759 if (S_ISREG(file_inode(file)->i_mode))
1760 res |= REQ_F_ISREG;
1761 if ((file->f_flags & O_NONBLOCK) || (file->f_mode & FMODE_NOWAIT))
1762 res |= REQ_F_SUPPORT_NOWAIT;
1763 return res;
1764}
1765
1766static __cold void io_drain_req(struct io_kiocb *req)
1767 __must_hold(&ctx->uring_lock)
1768{
1769 struct io_ring_ctx *ctx = req->ctx;
1770 bool drain = req->flags & IOSQE_IO_DRAIN;
1771 struct io_defer_entry *de;
1772
1773 de = kmalloc(sizeof(*de), GFP_KERNEL_ACCOUNT);
1774 if (!de) {
1775 io_req_defer_failed(req, res: -ENOMEM);
1776 return;
1777 }
1778
1779 io_prep_async_link(req);
1780 trace_io_uring_defer(req);
1781 de->req = req;
1782
1783 ctx->nr_drained += io_linked_nr(req);
1784 list_add_tail(new: &de->list, head: &ctx->defer_list);
1785 io_queue_deferred(ctx);
1786 if (!drain && list_empty(head: &ctx->defer_list))
1787 ctx->drain_active = false;
1788}
1789
1790static bool io_assign_file(struct io_kiocb *req, const struct io_issue_def *def,
1791 unsigned int issue_flags)
1792{
1793 if (req->file || !def->needs_file)
1794 return true;
1795
1796 if (req->flags & REQ_F_FIXED_FILE)
1797 req->file = io_file_get_fixed(req, fd: req->cqe.fd, issue_flags);
1798 else
1799 req->file = io_file_get_normal(req, fd: req->cqe.fd);
1800
1801 return !!req->file;
1802}
1803
1804#define REQ_ISSUE_SLOW_FLAGS (REQ_F_CREDS | REQ_F_ARM_LTIMEOUT)
1805
1806static inline int __io_issue_sqe(struct io_kiocb *req,
1807 unsigned int issue_flags,
1808 const struct io_issue_def *def)
1809{
1810 const struct cred *creds = NULL;
1811 struct io_kiocb *link = NULL;
1812 int ret;
1813
1814 if (unlikely(req->flags & REQ_ISSUE_SLOW_FLAGS)) {
1815 if ((req->flags & REQ_F_CREDS) && req->creds != current_cred())
1816 creds = override_creds(override_cred: req->creds);
1817 if (req->flags & REQ_F_ARM_LTIMEOUT)
1818 link = __io_prep_linked_timeout(req);
1819 }
1820
1821 if (!def->audit_skip)
1822 audit_uring_entry(op: req->opcode);
1823
1824 ret = def->issue(req, issue_flags);
1825
1826 if (!def->audit_skip)
1827 audit_uring_exit(success: !ret, code: ret);
1828
1829 if (unlikely(creds || link)) {
1830 if (creds)
1831 revert_creds(revert_cred: creds);
1832 if (link)
1833 io_queue_linked_timeout(req: link);
1834 }
1835
1836 return ret;
1837}
1838
1839static int io_issue_sqe(struct io_kiocb *req, unsigned int issue_flags)
1840{
1841 const struct io_issue_def *def = &io_issue_defs[req->opcode];
1842 int ret;
1843
1844 if (unlikely(!io_assign_file(req, def, issue_flags)))
1845 return -EBADF;
1846
1847 ret = __io_issue_sqe(req, issue_flags, def);
1848
1849 if (ret == IOU_COMPLETE) {
1850 if (issue_flags & IO_URING_F_COMPLETE_DEFER)
1851 io_req_complete_defer(req);
1852 else
1853 io_req_complete_post(req, issue_flags);
1854
1855 return 0;
1856 }
1857
1858 if (ret == IOU_ISSUE_SKIP_COMPLETE) {
1859 ret = 0;
1860
1861 /* If the op doesn't have a file, we're not polling for it */
1862 if ((req->ctx->flags & IORING_SETUP_IOPOLL) && def->iopoll_queue)
1863 io_iopoll_req_issued(req, issue_flags);
1864 }
1865 return ret;
1866}
1867
1868int io_poll_issue(struct io_kiocb *req, io_tw_token_t tw)
1869{
1870 const unsigned int issue_flags = IO_URING_F_NONBLOCK |
1871 IO_URING_F_MULTISHOT |
1872 IO_URING_F_COMPLETE_DEFER;
1873 int ret;
1874
1875 io_tw_lock(ctx: req->ctx, tw);
1876
1877 WARN_ON_ONCE(!req->file);
1878 if (WARN_ON_ONCE(req->ctx->flags & IORING_SETUP_IOPOLL))
1879 return -EFAULT;
1880
1881 ret = __io_issue_sqe(req, issue_flags, def: &io_issue_defs[req->opcode]);
1882
1883 WARN_ON_ONCE(ret == IOU_ISSUE_SKIP_COMPLETE);
1884 return ret;
1885}
1886
1887struct io_wq_work *io_wq_free_work(struct io_wq_work *work)
1888{
1889 struct io_kiocb *req = container_of(work, struct io_kiocb, work);
1890 struct io_kiocb *nxt = NULL;
1891
1892 if (req_ref_put_and_test_atomic(req)) {
1893 if (req->flags & IO_REQ_LINK_FLAGS)
1894 nxt = io_req_find_next(req);
1895 io_free_req(req);
1896 }
1897 return nxt ? &nxt->work : NULL;
1898}
1899
1900void io_wq_submit_work(struct io_wq_work *work)
1901{
1902 struct io_kiocb *req = container_of(work, struct io_kiocb, work);
1903 const struct io_issue_def *def = &io_issue_defs[req->opcode];
1904 unsigned int issue_flags = IO_URING_F_UNLOCKED | IO_URING_F_IOWQ;
1905 bool needs_poll = false;
1906 int ret = 0, err = -ECANCELED;
1907
1908 /* one will be dropped by io_wq_free_work() after returning to io-wq */
1909 if (!(req->flags & REQ_F_REFCOUNT))
1910 __io_req_set_refcount(req, nr: 2);
1911 else
1912 req_ref_get(req);
1913
1914 /* either cancelled or io-wq is dying, so don't touch tctx->iowq */
1915 if (atomic_read(v: &work->flags) & IO_WQ_WORK_CANCEL) {
1916fail:
1917 io_req_task_queue_fail(req, ret: err);
1918 return;
1919 }
1920 if (!io_assign_file(req, def, issue_flags)) {
1921 err = -EBADF;
1922 atomic_or(i: IO_WQ_WORK_CANCEL, v: &work->flags);
1923 goto fail;
1924 }
1925
1926 /*
1927 * If DEFER_TASKRUN is set, it's only allowed to post CQEs from the
1928 * submitter task context. Final request completions are handed to the
1929 * right context, however this is not the case of auxiliary CQEs,
1930 * which is the main mean of operation for multishot requests.
1931 * Don't allow any multishot execution from io-wq. It's more restrictive
1932 * than necessary and also cleaner.
1933 */
1934 if (req->flags & (REQ_F_MULTISHOT|REQ_F_APOLL_MULTISHOT)) {
1935 err = -EBADFD;
1936 if (!io_file_can_poll(req))
1937 goto fail;
1938 if (req->file->f_flags & O_NONBLOCK ||
1939 req->file->f_mode & FMODE_NOWAIT) {
1940 err = -ECANCELED;
1941 if (io_arm_poll_handler(req, issue_flags) != IO_APOLL_OK)
1942 goto fail;
1943 return;
1944 } else {
1945 req->flags &= ~(REQ_F_APOLL_MULTISHOT|REQ_F_MULTISHOT);
1946 }
1947 }
1948
1949 if (req->flags & REQ_F_FORCE_ASYNC) {
1950 bool opcode_poll = def->pollin || def->pollout;
1951
1952 if (opcode_poll && io_file_can_poll(req)) {
1953 needs_poll = true;
1954 issue_flags |= IO_URING_F_NONBLOCK;
1955 }
1956 }
1957
1958 do {
1959 ret = io_issue_sqe(req, issue_flags);
1960 if (ret != -EAGAIN)
1961 break;
1962
1963 /*
1964 * If REQ_F_NOWAIT is set, then don't wait or retry with
1965 * poll. -EAGAIN is final for that case.
1966 */
1967 if (req->flags & REQ_F_NOWAIT)
1968 break;
1969
1970 /*
1971 * We can get EAGAIN for iopolled IO even though we're
1972 * forcing a sync submission from here, since we can't
1973 * wait for request slots on the block side.
1974 */
1975 if (!needs_poll) {
1976 if (!(req->ctx->flags & IORING_SETUP_IOPOLL))
1977 break;
1978 if (io_wq_worker_stopped())
1979 break;
1980 cond_resched();
1981 continue;
1982 }
1983
1984 if (io_arm_poll_handler(req, issue_flags) == IO_APOLL_OK)
1985 return;
1986 /* aborted or ready, in either case retry blocking */
1987 needs_poll = false;
1988 issue_flags &= ~IO_URING_F_NONBLOCK;
1989 } while (1);
1990
1991 /* avoid locking problems by failing it from a clean context */
1992 if (ret)
1993 io_req_task_queue_fail(req, ret);
1994}
1995
1996inline struct file *io_file_get_fixed(struct io_kiocb *req, int fd,
1997 unsigned int issue_flags)
1998{
1999 struct io_ring_ctx *ctx = req->ctx;
2000 struct io_rsrc_node *node;
2001 struct file *file = NULL;
2002
2003 io_ring_submit_lock(ctx, issue_flags);
2004 node = io_rsrc_node_lookup(data: &ctx->file_table.data, index: fd);
2005 if (node) {
2006 node->refs++;
2007 req->file_node = node;
2008 req->flags |= io_slot_flags(node);
2009 file = io_slot_file(node);
2010 }
2011 io_ring_submit_unlock(ctx, issue_flags);
2012 return file;
2013}
2014
2015struct file *io_file_get_normal(struct io_kiocb *req, int fd)
2016{
2017 struct file *file = fget(fd);
2018
2019 trace_io_uring_file_get(req, fd);
2020
2021 /* we don't allow fixed io_uring files */
2022 if (file && io_is_uring_fops(file))
2023 io_req_track_inflight(req);
2024 return file;
2025}
2026
2027static int io_req_sqe_copy(struct io_kiocb *req, unsigned int issue_flags)
2028{
2029 const struct io_cold_def *def = &io_cold_defs[req->opcode];
2030
2031 if (req->flags & REQ_F_SQE_COPIED)
2032 return 0;
2033 req->flags |= REQ_F_SQE_COPIED;
2034 if (!def->sqe_copy)
2035 return 0;
2036 if (WARN_ON_ONCE(!(issue_flags & IO_URING_F_INLINE)))
2037 return -EFAULT;
2038 def->sqe_copy(req);
2039 return 0;
2040}
2041
2042static void io_queue_async(struct io_kiocb *req, unsigned int issue_flags, int ret)
2043 __must_hold(&req->ctx->uring_lock)
2044{
2045 if (ret != -EAGAIN || (req->flags & REQ_F_NOWAIT)) {
2046fail:
2047 io_req_defer_failed(req, res: ret);
2048 return;
2049 }
2050
2051 ret = io_req_sqe_copy(req, issue_flags);
2052 if (unlikely(ret))
2053 goto fail;
2054
2055 switch (io_arm_poll_handler(req, issue_flags: 0)) {
2056 case IO_APOLL_READY:
2057 io_req_task_queue(req);
2058 break;
2059 case IO_APOLL_ABORTED:
2060 io_queue_iowq(req);
2061 break;
2062 case IO_APOLL_OK:
2063 break;
2064 }
2065}
2066
2067static inline void io_queue_sqe(struct io_kiocb *req, unsigned int extra_flags)
2068 __must_hold(&req->ctx->uring_lock)
2069{
2070 unsigned int issue_flags = IO_URING_F_NONBLOCK |
2071 IO_URING_F_COMPLETE_DEFER | extra_flags;
2072 int ret;
2073
2074 ret = io_issue_sqe(req, issue_flags);
2075
2076 /*
2077 * We async punt it if the file wasn't marked NOWAIT, or if the file
2078 * doesn't support non-blocking read/write attempts
2079 */
2080 if (unlikely(ret))
2081 io_queue_async(req, issue_flags, ret);
2082}
2083
2084static void io_queue_sqe_fallback(struct io_kiocb *req)
2085 __must_hold(&req->ctx->uring_lock)
2086{
2087 if (unlikely(req->flags & REQ_F_FAIL)) {
2088 /*
2089 * We don't submit, fail them all, for that replace hardlinks
2090 * with normal links. Extra REQ_F_LINK is tolerated.
2091 */
2092 req->flags &= ~REQ_F_HARDLINK;
2093 req->flags |= REQ_F_LINK;
2094 io_req_defer_failed(req, res: req->cqe.res);
2095 } else {
2096 /* can't fail with IO_URING_F_INLINE */
2097 io_req_sqe_copy(req, issue_flags: IO_URING_F_INLINE);
2098 if (unlikely(req->ctx->drain_active))
2099 io_drain_req(req);
2100 else
2101 io_queue_iowq(req);
2102 }
2103}
2104
2105/*
2106 * Check SQE restrictions (opcode and flags).
2107 *
2108 * Returns 'true' if SQE is allowed, 'false' otherwise.
2109 */
2110static inline bool io_check_restriction(struct io_ring_ctx *ctx,
2111 struct io_kiocb *req,
2112 unsigned int sqe_flags)
2113{
2114 if (!test_bit(req->opcode, ctx->restrictions.sqe_op))
2115 return false;
2116
2117 if ((sqe_flags & ctx->restrictions.sqe_flags_required) !=
2118 ctx->restrictions.sqe_flags_required)
2119 return false;
2120
2121 if (sqe_flags & ~(ctx->restrictions.sqe_flags_allowed |
2122 ctx->restrictions.sqe_flags_required))
2123 return false;
2124
2125 return true;
2126}
2127
2128static void io_init_drain(struct io_ring_ctx *ctx)
2129{
2130 struct io_kiocb *head = ctx->submit_state.link.head;
2131
2132 ctx->drain_active = true;
2133 if (head) {
2134 /*
2135 * If we need to drain a request in the middle of a link, drain
2136 * the head request and the next request/link after the current
2137 * link. Considering sequential execution of links,
2138 * REQ_F_IO_DRAIN will be maintained for every request of our
2139 * link.
2140 */
2141 head->flags |= REQ_F_IO_DRAIN | REQ_F_FORCE_ASYNC;
2142 ctx->drain_next = true;
2143 }
2144}
2145
2146static __cold int io_init_fail_req(struct io_kiocb *req, int err)
2147{
2148 /* ensure per-opcode data is cleared if we fail before prep */
2149 memset(s: &req->cmd.data, c: 0, n: sizeof(req->cmd.data));
2150 return err;
2151}
2152
2153static int io_init_req(struct io_ring_ctx *ctx, struct io_kiocb *req,
2154 const struct io_uring_sqe *sqe)
2155 __must_hold(&ctx->uring_lock)
2156{
2157 const struct io_issue_def *def;
2158 unsigned int sqe_flags;
2159 int personality;
2160 u8 opcode;
2161
2162 req->ctx = ctx;
2163 req->opcode = opcode = READ_ONCE(sqe->opcode);
2164 /* same numerical values with corresponding REQ_F_*, safe to copy */
2165 sqe_flags = READ_ONCE(sqe->flags);
2166 req->flags = (__force io_req_flags_t) sqe_flags;
2167 req->cqe.user_data = READ_ONCE(sqe->user_data);
2168 req->file = NULL;
2169 req->tctx = current->io_uring;
2170 req->cancel_seq_set = false;
2171 req->async_data = NULL;
2172
2173 if (unlikely(opcode >= IORING_OP_LAST)) {
2174 req->opcode = 0;
2175 return io_init_fail_req(req, err: -EINVAL);
2176 }
2177 opcode = array_index_nospec(opcode, IORING_OP_LAST);
2178
2179 def = &io_issue_defs[opcode];
2180 if (unlikely(sqe_flags & ~SQE_COMMON_FLAGS)) {
2181 /* enforce forwards compatibility on users */
2182 if (sqe_flags & ~SQE_VALID_FLAGS)
2183 return io_init_fail_req(req, err: -EINVAL);
2184 if (sqe_flags & IOSQE_BUFFER_SELECT) {
2185 if (!def->buffer_select)
2186 return io_init_fail_req(req, err: -EOPNOTSUPP);
2187 req->buf_index = READ_ONCE(sqe->buf_group);
2188 }
2189 if (sqe_flags & IOSQE_CQE_SKIP_SUCCESS)
2190 ctx->drain_disabled = true;
2191 if (sqe_flags & IOSQE_IO_DRAIN) {
2192 if (ctx->drain_disabled)
2193 return io_init_fail_req(req, err: -EOPNOTSUPP);
2194 io_init_drain(ctx);
2195 }
2196 }
2197 if (unlikely(ctx->restricted || ctx->drain_active || ctx->drain_next)) {
2198 if (ctx->restricted && !io_check_restriction(ctx, req, sqe_flags))
2199 return io_init_fail_req(req, err: -EACCES);
2200 /* knock it to the slow queue path, will be drained there */
2201 if (ctx->drain_active)
2202 req->flags |= REQ_F_FORCE_ASYNC;
2203 /* if there is no link, we're at "next" request and need to drain */
2204 if (unlikely(ctx->drain_next) && !ctx->submit_state.link.head) {
2205 ctx->drain_next = false;
2206 ctx->drain_active = true;
2207 req->flags |= REQ_F_IO_DRAIN | REQ_F_FORCE_ASYNC;
2208 }
2209 }
2210
2211 if (!def->ioprio && sqe->ioprio)
2212 return io_init_fail_req(req, err: -EINVAL);
2213 if (!def->iopoll && (ctx->flags & IORING_SETUP_IOPOLL))
2214 return io_init_fail_req(req, err: -EINVAL);
2215
2216 if (def->needs_file) {
2217 struct io_submit_state *state = &ctx->submit_state;
2218
2219 req->cqe.fd = READ_ONCE(sqe->fd);
2220
2221 /*
2222 * Plug now if we have more than 2 IO left after this, and the
2223 * target is potentially a read/write to block based storage.
2224 */
2225 if (state->need_plug && def->plug) {
2226 state->plug_started = true;
2227 state->need_plug = false;
2228 blk_start_plug_nr_ios(&state->plug, state->submit_nr);
2229 }
2230 }
2231
2232 personality = READ_ONCE(sqe->personality);
2233 if (personality) {
2234 int ret;
2235
2236 req->creds = xa_load(&ctx->personalities, index: personality);
2237 if (!req->creds)
2238 return io_init_fail_req(req, err: -EINVAL);
2239 get_cred(cred: req->creds);
2240 ret = security_uring_override_creds(new: req->creds);
2241 if (ret) {
2242 put_cred(cred: req->creds);
2243 return io_init_fail_req(req, err: ret);
2244 }
2245 req->flags |= REQ_F_CREDS;
2246 }
2247
2248 return def->prep(req, sqe);
2249}
2250
2251static __cold int io_submit_fail_init(const struct io_uring_sqe *sqe,
2252 struct io_kiocb *req, int ret)
2253{
2254 struct io_ring_ctx *ctx = req->ctx;
2255 struct io_submit_link *link = &ctx->submit_state.link;
2256 struct io_kiocb *head = link->head;
2257
2258 trace_io_uring_req_failed(sqe, req, error: ret);
2259
2260 /*
2261 * Avoid breaking links in the middle as it renders links with SQPOLL
2262 * unusable. Instead of failing eagerly, continue assembling the link if
2263 * applicable and mark the head with REQ_F_FAIL. The link flushing code
2264 * should find the flag and handle the rest.
2265 */
2266 req_fail_link_node(req, res: ret);
2267 if (head && !(head->flags & REQ_F_FAIL))
2268 req_fail_link_node(req: head, res: -ECANCELED);
2269
2270 if (!(req->flags & IO_REQ_LINK_FLAGS)) {
2271 if (head) {
2272 link->last->link = req;
2273 link->head = NULL;
2274 req = head;
2275 }
2276 io_queue_sqe_fallback(req);
2277 return ret;
2278 }
2279
2280 if (head)
2281 link->last->link = req;
2282 else
2283 link->head = req;
2284 link->last = req;
2285 return 0;
2286}
2287
2288static inline int io_submit_sqe(struct io_ring_ctx *ctx, struct io_kiocb *req,
2289 const struct io_uring_sqe *sqe)
2290 __must_hold(&ctx->uring_lock)
2291{
2292 struct io_submit_link *link = &ctx->submit_state.link;
2293 int ret;
2294
2295 ret = io_init_req(ctx, req, sqe);
2296 if (unlikely(ret))
2297 return io_submit_fail_init(sqe, req, ret);
2298
2299 trace_io_uring_submit_req(req);
2300
2301 /*
2302 * If we already have a head request, queue this one for async
2303 * submittal once the head completes. If we don't have a head but
2304 * IOSQE_IO_LINK is set in the sqe, start a new head. This one will be
2305 * submitted sync once the chain is complete. If none of those
2306 * conditions are true (normal request), then just queue it.
2307 */
2308 if (unlikely(link->head)) {
2309 trace_io_uring_link(req, target_req: link->last);
2310 io_req_sqe_copy(req, issue_flags: IO_URING_F_INLINE);
2311 link->last->link = req;
2312 link->last = req;
2313
2314 if (req->flags & IO_REQ_LINK_FLAGS)
2315 return 0;
2316 /* last request of the link, flush it */
2317 req = link->head;
2318 link->head = NULL;
2319 if (req->flags & (REQ_F_FORCE_ASYNC | REQ_F_FAIL))
2320 goto fallback;
2321
2322 } else if (unlikely(req->flags & (IO_REQ_LINK_FLAGS |
2323 REQ_F_FORCE_ASYNC | REQ_F_FAIL))) {
2324 if (req->flags & IO_REQ_LINK_FLAGS) {
2325 link->head = req;
2326 link->last = req;
2327 } else {
2328fallback:
2329 io_queue_sqe_fallback(req);
2330 }
2331 return 0;
2332 }
2333
2334 io_queue_sqe(req, extra_flags: IO_URING_F_INLINE);
2335 return 0;
2336}
2337
2338/*
2339 * Batched submission is done, ensure local IO is flushed out.
2340 */
2341static void io_submit_state_end(struct io_ring_ctx *ctx)
2342{
2343 struct io_submit_state *state = &ctx->submit_state;
2344
2345 if (unlikely(state->link.head))
2346 io_queue_sqe_fallback(req: state->link.head);
2347 /* flush only after queuing links as they can generate completions */
2348 io_submit_flush_completions(ctx);
2349 if (state->plug_started)
2350 blk_finish_plug(&state->plug);
2351}
2352
2353/*
2354 * Start submission side cache.
2355 */
2356static void io_submit_state_start(struct io_submit_state *state,
2357 unsigned int max_ios)
2358{
2359 state->plug_started = false;
2360 state->need_plug = max_ios > 2;
2361 state->submit_nr = max_ios;
2362 /* set only head, no need to init link_last in advance */
2363 state->link.head = NULL;
2364}
2365
2366static void io_commit_sqring(struct io_ring_ctx *ctx)
2367{
2368 struct io_rings *rings = ctx->rings;
2369
2370 /*
2371 * Ensure any loads from the SQEs are done at this point,
2372 * since once we write the new head, the application could
2373 * write new data to them.
2374 */
2375 smp_store_release(&rings->sq.head, ctx->cached_sq_head);
2376}
2377
2378/*
2379 * Fetch an sqe, if one is available. Note this returns a pointer to memory
2380 * that is mapped by userspace. This means that care needs to be taken to
2381 * ensure that reads are stable, as we cannot rely on userspace always
2382 * being a good citizen. If members of the sqe are validated and then later
2383 * used, it's important that those reads are done through READ_ONCE() to
2384 * prevent a re-load down the line.
2385 */
2386static bool io_get_sqe(struct io_ring_ctx *ctx, const struct io_uring_sqe **sqe)
2387{
2388 unsigned mask = ctx->sq_entries - 1;
2389 unsigned head = ctx->cached_sq_head++ & mask;
2390
2391 if (static_branch_unlikely(&io_key_has_sqarray) &&
2392 (!(ctx->flags & IORING_SETUP_NO_SQARRAY))) {
2393 head = READ_ONCE(ctx->sq_array[head]);
2394 if (unlikely(head >= ctx->sq_entries)) {
2395 WRITE_ONCE(ctx->rings->sq_dropped,
2396 READ_ONCE(ctx->rings->sq_dropped) + 1);
2397 return false;
2398 }
2399 head = array_index_nospec(head, ctx->sq_entries);
2400 }
2401
2402 /*
2403 * The cached sq head (or cq tail) serves two purposes:
2404 *
2405 * 1) allows us to batch the cost of updating the user visible
2406 * head updates.
2407 * 2) allows the kernel side to track the head on its own, even
2408 * though the application is the one updating it.
2409 */
2410
2411 /* double index for 128-byte SQEs, twice as long */
2412 if (ctx->flags & IORING_SETUP_SQE128)
2413 head <<= 1;
2414 *sqe = &ctx->sq_sqes[head];
2415 return true;
2416}
2417
2418int io_submit_sqes(struct io_ring_ctx *ctx, unsigned int nr)
2419 __must_hold(&ctx->uring_lock)
2420{
2421 unsigned int entries = io_sqring_entries(ctx);
2422 unsigned int left;
2423 int ret;
2424
2425 if (unlikely(!entries))
2426 return 0;
2427 /* make sure SQ entry isn't read before tail */
2428 ret = left = min(nr, entries);
2429 io_get_task_refs(nr: left);
2430 io_submit_state_start(state: &ctx->submit_state, max_ios: left);
2431
2432 do {
2433 const struct io_uring_sqe *sqe;
2434 struct io_kiocb *req;
2435
2436 if (unlikely(!io_alloc_req(ctx, &req)))
2437 break;
2438 if (unlikely(!io_get_sqe(ctx, &sqe))) {
2439 io_req_add_to_cache(req, ctx);
2440 break;
2441 }
2442
2443 /*
2444 * Continue submitting even for sqe failure if the
2445 * ring was setup with IORING_SETUP_SUBMIT_ALL
2446 */
2447 if (unlikely(io_submit_sqe(ctx, req, sqe)) &&
2448 !(ctx->flags & IORING_SETUP_SUBMIT_ALL)) {
2449 left--;
2450 break;
2451 }
2452 } while (--left);
2453
2454 if (unlikely(left)) {
2455 ret -= left;
2456 /* try again if it submitted nothing and can't allocate a req */
2457 if (!ret && io_req_cache_empty(ctx))
2458 ret = -EAGAIN;
2459 current->io_uring->cached_refs += left;
2460 }
2461
2462 io_submit_state_end(ctx);
2463 /* Commit SQ ring head once we've consumed and submitted all SQEs */
2464 io_commit_sqring(ctx);
2465 return ret;
2466}
2467
2468static int io_wake_function(struct wait_queue_entry *curr, unsigned int mode,
2469 int wake_flags, void *key)
2470{
2471 struct io_wait_queue *iowq = container_of(curr, struct io_wait_queue, wq);
2472
2473 /*
2474 * Cannot safely flush overflowed CQEs from here, ensure we wake up
2475 * the task, and the next invocation will do it.
2476 */
2477 if (io_should_wake(iowq) || io_has_work(ctx: iowq->ctx))
2478 return autoremove_wake_function(wq_entry: curr, mode, sync: wake_flags, key);
2479 return -1;
2480}
2481
2482int io_run_task_work_sig(struct io_ring_ctx *ctx)
2483{
2484 if (io_local_work_pending(ctx)) {
2485 __set_current_state(TASK_RUNNING);
2486 if (io_run_local_work(ctx, INT_MAX, IO_LOCAL_TW_DEFAULT_MAX) > 0)
2487 return 0;
2488 }
2489 if (io_run_task_work() > 0)
2490 return 0;
2491 if (task_sigpending(current))
2492 return -EINTR;
2493 return 0;
2494}
2495
2496static bool current_pending_io(void)
2497{
2498 struct io_uring_task *tctx = current->io_uring;
2499
2500 if (!tctx)
2501 return false;
2502 return percpu_counter_read_positive(fbc: &tctx->inflight);
2503}
2504
2505static enum hrtimer_restart io_cqring_timer_wakeup(struct hrtimer *timer)
2506{
2507 struct io_wait_queue *iowq = container_of(timer, struct io_wait_queue, t);
2508
2509 WRITE_ONCE(iowq->hit_timeout, 1);
2510 iowq->min_timeout = 0;
2511 wake_up_process(tsk: iowq->wq.private);
2512 return HRTIMER_NORESTART;
2513}
2514
2515/*
2516 * Doing min_timeout portion. If we saw any timeouts, events, or have work,
2517 * wake up. If not, and we have a normal timeout, switch to that and keep
2518 * sleeping.
2519 */
2520static enum hrtimer_restart io_cqring_min_timer_wakeup(struct hrtimer *timer)
2521{
2522 struct io_wait_queue *iowq = container_of(timer, struct io_wait_queue, t);
2523 struct io_ring_ctx *ctx = iowq->ctx;
2524
2525 /* no general timeout, or shorter (or equal), we are done */
2526 if (iowq->timeout == KTIME_MAX ||
2527 ktime_compare(cmp1: iowq->min_timeout, cmp2: iowq->timeout) >= 0)
2528 goto out_wake;
2529 /* work we may need to run, wake function will see if we need to wake */
2530 if (io_has_work(ctx))
2531 goto out_wake;
2532 /* got events since we started waiting, min timeout is done */
2533 if (iowq->cq_min_tail != READ_ONCE(ctx->rings->cq.tail))
2534 goto out_wake;
2535 /* if we have any events and min timeout expired, we're done */
2536 if (io_cqring_events(ctx))
2537 goto out_wake;
2538
2539 /*
2540 * If using deferred task_work running and application is waiting on
2541 * more than one request, ensure we reset it now where we are switching
2542 * to normal sleeps. Any request completion post min_wait should wake
2543 * the task and return.
2544 */
2545 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) {
2546 atomic_set(v: &ctx->cq_wait_nr, i: 1);
2547 smp_mb();
2548 if (!llist_empty(head: &ctx->work_llist))
2549 goto out_wake;
2550 }
2551
2552 hrtimer_update_function(timer: &iowq->t, function: io_cqring_timer_wakeup);
2553 hrtimer_set_expires(timer, time: iowq->timeout);
2554 return HRTIMER_RESTART;
2555out_wake:
2556 return io_cqring_timer_wakeup(timer);
2557}
2558
2559static int io_cqring_schedule_timeout(struct io_wait_queue *iowq,
2560 clockid_t clock_id, ktime_t start_time)
2561{
2562 ktime_t timeout;
2563
2564 if (iowq->min_timeout) {
2565 timeout = ktime_add_ns(iowq->min_timeout, start_time);
2566 hrtimer_setup_on_stack(timer: &iowq->t, function: io_cqring_min_timer_wakeup, clock_id,
2567 mode: HRTIMER_MODE_ABS);
2568 } else {
2569 timeout = iowq->timeout;
2570 hrtimer_setup_on_stack(timer: &iowq->t, function: io_cqring_timer_wakeup, clock_id,
2571 mode: HRTIMER_MODE_ABS);
2572 }
2573
2574 hrtimer_set_expires_range_ns(timer: &iowq->t, time: timeout, delta: 0);
2575 hrtimer_start_expires(timer: &iowq->t, mode: HRTIMER_MODE_ABS);
2576
2577 if (!READ_ONCE(iowq->hit_timeout))
2578 schedule();
2579
2580 hrtimer_cancel(timer: &iowq->t);
2581 destroy_hrtimer_on_stack(timer: &iowq->t);
2582 __set_current_state(TASK_RUNNING);
2583
2584 return READ_ONCE(iowq->hit_timeout) ? -ETIME : 0;
2585}
2586
2587struct ext_arg {
2588 size_t argsz;
2589 struct timespec64 ts;
2590 const sigset_t __user *sig;
2591 ktime_t min_time;
2592 bool ts_set;
2593 bool iowait;
2594};
2595
2596static int __io_cqring_wait_schedule(struct io_ring_ctx *ctx,
2597 struct io_wait_queue *iowq,
2598 struct ext_arg *ext_arg,
2599 ktime_t start_time)
2600{
2601 int ret = 0;
2602
2603 /*
2604 * Mark us as being in io_wait if we have pending requests, so cpufreq
2605 * can take into account that the task is waiting for IO - turns out
2606 * to be important for low QD IO.
2607 */
2608 if (ext_arg->iowait && current_pending_io())
2609 current->in_iowait = 1;
2610 if (iowq->timeout != KTIME_MAX || iowq->min_timeout)
2611 ret = io_cqring_schedule_timeout(iowq, clock_id: ctx->clockid, start_time);
2612 else
2613 schedule();
2614 current->in_iowait = 0;
2615 return ret;
2616}
2617
2618/* If this returns > 0, the caller should retry */
2619static inline int io_cqring_wait_schedule(struct io_ring_ctx *ctx,
2620 struct io_wait_queue *iowq,
2621 struct ext_arg *ext_arg,
2622 ktime_t start_time)
2623{
2624 if (unlikely(READ_ONCE(ctx->check_cq)))
2625 return 1;
2626 if (unlikely(io_local_work_pending(ctx)))
2627 return 1;
2628 if (unlikely(task_work_pending(current)))
2629 return 1;
2630 if (unlikely(task_sigpending(current)))
2631 return -EINTR;
2632 if (unlikely(io_should_wake(iowq)))
2633 return 0;
2634
2635 return __io_cqring_wait_schedule(ctx, iowq, ext_arg, start_time);
2636}
2637
2638/*
2639 * Wait until events become available, if we don't already have some. The
2640 * application must reap them itself, as they reside on the shared cq ring.
2641 */
2642static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events, u32 flags,
2643 struct ext_arg *ext_arg)
2644{
2645 struct io_wait_queue iowq;
2646 struct io_rings *rings = ctx->rings;
2647 ktime_t start_time;
2648 int ret;
2649
2650 min_events = min_t(int, min_events, ctx->cq_entries);
2651
2652 if (!io_allowed_run_tw(ctx))
2653 return -EEXIST;
2654 if (io_local_work_pending(ctx))
2655 io_run_local_work(ctx, min_events,
2656 max(IO_LOCAL_TW_DEFAULT_MAX, min_events));
2657 io_run_task_work();
2658
2659 if (unlikely(test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq)))
2660 io_cqring_do_overflow_flush(ctx);
2661 if (__io_cqring_events_user(ctx) >= min_events)
2662 return 0;
2663
2664 init_waitqueue_func_entry(wq_entry: &iowq.wq, func: io_wake_function);
2665 iowq.wq.private = current;
2666 INIT_LIST_HEAD(list: &iowq.wq.entry);
2667 iowq.ctx = ctx;
2668 iowq.cq_tail = READ_ONCE(ctx->rings->cq.head) + min_events;
2669 iowq.cq_min_tail = READ_ONCE(ctx->rings->cq.tail);
2670 iowq.nr_timeouts = atomic_read(v: &ctx->cq_timeouts);
2671 iowq.hit_timeout = 0;
2672 iowq.min_timeout = ext_arg->min_time;
2673 iowq.timeout = KTIME_MAX;
2674 start_time = io_get_time(ctx);
2675
2676 if (ext_arg->ts_set) {
2677 iowq.timeout = timespec64_to_ktime(ts: ext_arg->ts);
2678 if (!(flags & IORING_ENTER_ABS_TIMER))
2679 iowq.timeout = ktime_add(iowq.timeout, start_time);
2680 }
2681
2682 if (ext_arg->sig) {
2683#ifdef CONFIG_COMPAT
2684 if (in_compat_syscall())
2685 ret = set_compat_user_sigmask(umask: (const compat_sigset_t __user *)ext_arg->sig,
2686 sigsetsize: ext_arg->argsz);
2687 else
2688#endif
2689 ret = set_user_sigmask(umask: ext_arg->sig, sigsetsize: ext_arg->argsz);
2690
2691 if (ret)
2692 return ret;
2693 }
2694
2695 io_napi_busy_loop(ctx, iowq: &iowq);
2696
2697 trace_io_uring_cqring_wait(ctx, min_events);
2698 do {
2699 unsigned long check_cq;
2700 int nr_wait;
2701
2702 /* if min timeout has been hit, don't reset wait count */
2703 if (!iowq.hit_timeout)
2704 nr_wait = (int) iowq.cq_tail -
2705 READ_ONCE(ctx->rings->cq.tail);
2706 else
2707 nr_wait = 1;
2708
2709 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) {
2710 atomic_set(v: &ctx->cq_wait_nr, i: nr_wait);
2711 set_current_state(TASK_INTERRUPTIBLE);
2712 } else {
2713 prepare_to_wait_exclusive(wq_head: &ctx->cq_wait, wq_entry: &iowq.wq,
2714 TASK_INTERRUPTIBLE);
2715 }
2716
2717 ret = io_cqring_wait_schedule(ctx, iowq: &iowq, ext_arg, start_time);
2718 __set_current_state(TASK_RUNNING);
2719 atomic_set(v: &ctx->cq_wait_nr, IO_CQ_WAKE_INIT);
2720
2721 /*
2722 * Run task_work after scheduling and before io_should_wake().
2723 * If we got woken because of task_work being processed, run it
2724 * now rather than let the caller do another wait loop.
2725 */
2726 if (io_local_work_pending(ctx))
2727 io_run_local_work(ctx, min_events: nr_wait, max_events: nr_wait);
2728 io_run_task_work();
2729
2730 /*
2731 * Non-local task_work will be run on exit to userspace, but
2732 * if we're using DEFER_TASKRUN, then we could have waited
2733 * with a timeout for a number of requests. If the timeout
2734 * hits, we could have some requests ready to process. Ensure
2735 * this break is _after_ we have run task_work, to avoid
2736 * deferring running potentially pending requests until the
2737 * next time we wait for events.
2738 */
2739 if (ret < 0)
2740 break;
2741
2742 check_cq = READ_ONCE(ctx->check_cq);
2743 if (unlikely(check_cq)) {
2744 /* let the caller flush overflows, retry */
2745 if (check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT))
2746 io_cqring_do_overflow_flush(ctx);
2747 if (check_cq & BIT(IO_CHECK_CQ_DROPPED_BIT)) {
2748 ret = -EBADR;
2749 break;
2750 }
2751 }
2752
2753 if (io_should_wake(iowq: &iowq)) {
2754 ret = 0;
2755 break;
2756 }
2757 cond_resched();
2758 } while (1);
2759
2760 if (!(ctx->flags & IORING_SETUP_DEFER_TASKRUN))
2761 finish_wait(wq_head: &ctx->cq_wait, wq_entry: &iowq.wq);
2762 restore_saved_sigmask_unless(interrupted: ret == -EINTR);
2763
2764 return READ_ONCE(rings->cq.head) == READ_ONCE(rings->cq.tail) ? ret : 0;
2765}
2766
2767static void io_rings_free(struct io_ring_ctx *ctx)
2768{
2769 io_free_region(ctx, mr: &ctx->sq_region);
2770 io_free_region(ctx, mr: &ctx->ring_region);
2771 ctx->rings = NULL;
2772 ctx->sq_sqes = NULL;
2773}
2774
2775unsigned long rings_size(unsigned int flags, unsigned int sq_entries,
2776 unsigned int cq_entries, size_t *sq_offset)
2777{
2778 struct io_rings *rings;
2779 size_t off, sq_array_size;
2780
2781 off = struct_size(rings, cqes, cq_entries);
2782 if (off == SIZE_MAX)
2783 return SIZE_MAX;
2784 if (flags & IORING_SETUP_CQE32) {
2785 if (check_shl_overflow(off, 1, &off))
2786 return SIZE_MAX;
2787 }
2788 if (flags & IORING_SETUP_CQE_MIXED) {
2789 if (cq_entries < 2)
2790 return SIZE_MAX;
2791 }
2792
2793#ifdef CONFIG_SMP
2794 off = ALIGN(off, SMP_CACHE_BYTES);
2795 if (off == 0)
2796 return SIZE_MAX;
2797#endif
2798
2799 if (flags & IORING_SETUP_NO_SQARRAY) {
2800 *sq_offset = SIZE_MAX;
2801 return off;
2802 }
2803
2804 *sq_offset = off;
2805
2806 sq_array_size = array_size(sizeof(u32), sq_entries);
2807 if (sq_array_size == SIZE_MAX)
2808 return SIZE_MAX;
2809
2810 if (check_add_overflow(off, sq_array_size, &off))
2811 return SIZE_MAX;
2812
2813 return off;
2814}
2815
2816static __cold void __io_req_caches_free(struct io_ring_ctx *ctx)
2817{
2818 struct io_kiocb *req;
2819 int nr = 0;
2820
2821 while (!io_req_cache_empty(ctx)) {
2822 req = io_extract_req(ctx);
2823 io_poison_req(req);
2824 kmem_cache_free(s: req_cachep, objp: req);
2825 nr++;
2826 }
2827 if (nr) {
2828 ctx->nr_req_allocated -= nr;
2829 percpu_ref_put_many(ref: &ctx->refs, nr);
2830 }
2831}
2832
2833static __cold void io_req_caches_free(struct io_ring_ctx *ctx)
2834{
2835 guard(mutex)(T: &ctx->uring_lock);
2836 __io_req_caches_free(ctx);
2837}
2838
2839static __cold void io_ring_ctx_free(struct io_ring_ctx *ctx)
2840{
2841 io_sq_thread_finish(ctx);
2842
2843 mutex_lock(lock: &ctx->uring_lock);
2844 io_sqe_buffers_unregister(ctx);
2845 io_sqe_files_unregister(ctx);
2846 io_unregister_zcrx_ifqs(ctx);
2847 io_cqring_overflow_kill(ctx);
2848 io_eventfd_unregister(ctx);
2849 io_free_alloc_caches(ctx);
2850 io_destroy_buffers(ctx);
2851 io_free_region(ctx, mr: &ctx->param_region);
2852 mutex_unlock(lock: &ctx->uring_lock);
2853 if (ctx->sq_creds)
2854 put_cred(cred: ctx->sq_creds);
2855 if (ctx->submitter_task)
2856 put_task_struct(t: ctx->submitter_task);
2857
2858 WARN_ON_ONCE(!list_empty(&ctx->ltimeout_list));
2859
2860 if (ctx->mm_account) {
2861 mmdrop(mm: ctx->mm_account);
2862 ctx->mm_account = NULL;
2863 }
2864 io_rings_free(ctx);
2865
2866 if (!(ctx->flags & IORING_SETUP_NO_SQARRAY))
2867 static_branch_dec(&io_key_has_sqarray);
2868
2869 percpu_ref_exit(ref: &ctx->refs);
2870 free_uid(ctx->user);
2871 io_req_caches_free(ctx);
2872
2873 WARN_ON_ONCE(ctx->nr_req_allocated);
2874
2875 if (ctx->hash_map)
2876 io_wq_put_hash(hash: ctx->hash_map);
2877 io_napi_free(ctx);
2878 kvfree(addr: ctx->cancel_table.hbs);
2879 xa_destroy(&ctx->io_bl_xa);
2880 kfree(objp: ctx);
2881}
2882
2883static __cold void io_activate_pollwq_cb(struct callback_head *cb)
2884{
2885 struct io_ring_ctx *ctx = container_of(cb, struct io_ring_ctx,
2886 poll_wq_task_work);
2887
2888 mutex_lock(lock: &ctx->uring_lock);
2889 ctx->poll_activated = true;
2890 mutex_unlock(lock: &ctx->uring_lock);
2891
2892 /*
2893 * Wake ups for some events between start of polling and activation
2894 * might've been lost due to loose synchronisation.
2895 */
2896 wake_up_all(&ctx->poll_wq);
2897 percpu_ref_put(ref: &ctx->refs);
2898}
2899
2900__cold void io_activate_pollwq(struct io_ring_ctx *ctx)
2901{
2902 spin_lock(lock: &ctx->completion_lock);
2903 /* already activated or in progress */
2904 if (ctx->poll_activated || ctx->poll_wq_task_work.func)
2905 goto out;
2906 if (WARN_ON_ONCE(!ctx->task_complete))
2907 goto out;
2908 if (!ctx->submitter_task)
2909 goto out;
2910 /*
2911 * with ->submitter_task only the submitter task completes requests, we
2912 * only need to sync with it, which is done by injecting a tw
2913 */
2914 init_task_work(twork: &ctx->poll_wq_task_work, func: io_activate_pollwq_cb);
2915 percpu_ref_get(ref: &ctx->refs);
2916 if (task_work_add(task: ctx->submitter_task, twork: &ctx->poll_wq_task_work, mode: TWA_SIGNAL))
2917 percpu_ref_put(ref: &ctx->refs);
2918out:
2919 spin_unlock(lock: &ctx->completion_lock);
2920}
2921
2922static __poll_t io_uring_poll(struct file *file, poll_table *wait)
2923{
2924 struct io_ring_ctx *ctx = file->private_data;
2925 __poll_t mask = 0;
2926
2927 if (unlikely(!ctx->poll_activated))
2928 io_activate_pollwq(ctx);
2929 /*
2930 * provides mb() which pairs with barrier from wq_has_sleeper
2931 * call in io_commit_cqring
2932 */
2933 poll_wait(filp: file, wait_address: &ctx->poll_wq, p: wait);
2934
2935 if (!io_sqring_full(ctx))
2936 mask |= EPOLLOUT | EPOLLWRNORM;
2937
2938 /*
2939 * Don't flush cqring overflow list here, just do a simple check.
2940 * Otherwise there could possible be ABBA deadlock:
2941 * CPU0 CPU1
2942 * ---- ----
2943 * lock(&ctx->uring_lock);
2944 * lock(&ep->mtx);
2945 * lock(&ctx->uring_lock);
2946 * lock(&ep->mtx);
2947 *
2948 * Users may get EPOLLIN meanwhile seeing nothing in cqring, this
2949 * pushes them to do the flush.
2950 */
2951
2952 if (__io_cqring_events_user(ctx) || io_has_work(ctx))
2953 mask |= EPOLLIN | EPOLLRDNORM;
2954
2955 return mask;
2956}
2957
2958struct io_tctx_exit {
2959 struct callback_head task_work;
2960 struct completion completion;
2961 struct io_ring_ctx *ctx;
2962};
2963
2964static __cold void io_tctx_exit_cb(struct callback_head *cb)
2965{
2966 struct io_uring_task *tctx = current->io_uring;
2967 struct io_tctx_exit *work;
2968
2969 work = container_of(cb, struct io_tctx_exit, task_work);
2970 /*
2971 * When @in_cancel, we're in cancellation and it's racy to remove the
2972 * node. It'll be removed by the end of cancellation, just ignore it.
2973 * tctx can be NULL if the queueing of this task_work raced with
2974 * work cancelation off the exec path.
2975 */
2976 if (tctx && !atomic_read(v: &tctx->in_cancel))
2977 io_uring_del_tctx_node(index: (unsigned long)work->ctx);
2978 complete(&work->completion);
2979}
2980
2981static __cold bool io_cancel_ctx_cb(struct io_wq_work *work, void *data)
2982{
2983 struct io_kiocb *req = container_of(work, struct io_kiocb, work);
2984
2985 return req->ctx == data;
2986}
2987
2988static __cold void io_ring_exit_work(struct work_struct *work)
2989{
2990 struct io_ring_ctx *ctx = container_of(work, struct io_ring_ctx, exit_work);
2991 unsigned long timeout = jiffies + HZ * 60 * 5;
2992 unsigned long interval = HZ / 20;
2993 struct io_tctx_exit exit;
2994 struct io_tctx_node *node;
2995 int ret;
2996
2997 /*
2998 * If we're doing polled IO and end up having requests being
2999 * submitted async (out-of-line), then completions can come in while
3000 * we're waiting for refs to drop. We need to reap these manually,
3001 * as nobody else will be looking for them.
3002 */
3003 do {
3004 if (test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq)) {
3005 mutex_lock(lock: &ctx->uring_lock);
3006 io_cqring_overflow_kill(ctx);
3007 mutex_unlock(lock: &ctx->uring_lock);
3008 }
3009 if (!xa_empty(xa: &ctx->zcrx_ctxs)) {
3010 mutex_lock(lock: &ctx->uring_lock);
3011 io_shutdown_zcrx_ifqs(ctx);
3012 mutex_unlock(lock: &ctx->uring_lock);
3013 }
3014
3015 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
3016 io_move_task_work_from_local(ctx);
3017
3018 /* The SQPOLL thread never reaches this path */
3019 while (io_uring_try_cancel_requests(ctx, NULL, cancel_all: true, is_sqpoll_thread: false))
3020 cond_resched();
3021
3022 if (ctx->sq_data) {
3023 struct io_sq_data *sqd = ctx->sq_data;
3024 struct task_struct *tsk;
3025
3026 io_sq_thread_park(sqd);
3027 tsk = sqpoll_task_locked(sqd);
3028 if (tsk && tsk->io_uring && tsk->io_uring->io_wq)
3029 io_wq_cancel_cb(wq: tsk->io_uring->io_wq,
3030 cancel: io_cancel_ctx_cb, data: ctx, cancel_all: true);
3031 io_sq_thread_unpark(sqd);
3032 }
3033
3034 io_req_caches_free(ctx);
3035
3036 if (WARN_ON_ONCE(time_after(jiffies, timeout))) {
3037 /* there is little hope left, don't run it too often */
3038 interval = HZ * 60;
3039 }
3040 /*
3041 * This is really an uninterruptible wait, as it has to be
3042 * complete. But it's also run from a kworker, which doesn't
3043 * take signals, so it's fine to make it interruptible. This
3044 * avoids scenarios where we knowingly can wait much longer
3045 * on completions, for example if someone does a SIGSTOP on
3046 * a task that needs to finish task_work to make this loop
3047 * complete. That's a synthetic situation that should not
3048 * cause a stuck task backtrace, and hence a potential panic
3049 * on stuck tasks if that is enabled.
3050 */
3051 } while (!wait_for_completion_interruptible_timeout(x: &ctx->ref_comp, timeout: interval));
3052
3053 init_completion(x: &exit.completion);
3054 init_task_work(twork: &exit.task_work, func: io_tctx_exit_cb);
3055 exit.ctx = ctx;
3056
3057 mutex_lock(lock: &ctx->uring_lock);
3058 while (!list_empty(head: &ctx->tctx_list)) {
3059 WARN_ON_ONCE(time_after(jiffies, timeout));
3060
3061 node = list_first_entry(&ctx->tctx_list, struct io_tctx_node,
3062 ctx_node);
3063 /* don't spin on a single task if cancellation failed */
3064 list_rotate_left(head: &ctx->tctx_list);
3065 ret = task_work_add(task: node->task, twork: &exit.task_work, mode: TWA_SIGNAL);
3066 if (WARN_ON_ONCE(ret))
3067 continue;
3068
3069 mutex_unlock(lock: &ctx->uring_lock);
3070 /*
3071 * See comment above for
3072 * wait_for_completion_interruptible_timeout() on why this
3073 * wait is marked as interruptible.
3074 */
3075 wait_for_completion_interruptible(x: &exit.completion);
3076 mutex_lock(lock: &ctx->uring_lock);
3077 }
3078 mutex_unlock(lock: &ctx->uring_lock);
3079 spin_lock(lock: &ctx->completion_lock);
3080 spin_unlock(lock: &ctx->completion_lock);
3081
3082 /* pairs with RCU read section in io_req_local_work_add() */
3083 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
3084 synchronize_rcu();
3085
3086 io_ring_ctx_free(ctx);
3087}
3088
3089static __cold void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
3090{
3091 unsigned long index;
3092 struct creds *creds;
3093
3094 mutex_lock(lock: &ctx->uring_lock);
3095 percpu_ref_kill(ref: &ctx->refs);
3096 xa_for_each(&ctx->personalities, index, creds)
3097 io_unregister_personality(ctx, id: index);
3098 mutex_unlock(lock: &ctx->uring_lock);
3099
3100 flush_delayed_work(dwork: &ctx->fallback_work);
3101
3102 INIT_WORK(&ctx->exit_work, io_ring_exit_work);
3103 /*
3104 * Use system_dfl_wq to avoid spawning tons of event kworkers
3105 * if we're exiting a ton of rings at the same time. It just adds
3106 * noise and overhead, there's no discernable change in runtime
3107 * over using system_percpu_wq.
3108 */
3109 queue_work(wq: iou_wq, work: &ctx->exit_work);
3110}
3111
3112static int io_uring_release(struct inode *inode, struct file *file)
3113{
3114 struct io_ring_ctx *ctx = file->private_data;
3115
3116 file->private_data = NULL;
3117 io_ring_ctx_wait_and_kill(ctx);
3118 return 0;
3119}
3120
3121struct io_task_cancel {
3122 struct io_uring_task *tctx;
3123 bool all;
3124};
3125
3126static bool io_cancel_task_cb(struct io_wq_work *work, void *data)
3127{
3128 struct io_kiocb *req = container_of(work, struct io_kiocb, work);
3129 struct io_task_cancel *cancel = data;
3130
3131 return io_match_task_safe(head: req, tctx: cancel->tctx, cancel_all: cancel->all);
3132}
3133
3134static __cold bool io_cancel_defer_files(struct io_ring_ctx *ctx,
3135 struct io_uring_task *tctx,
3136 bool cancel_all)
3137{
3138 struct io_defer_entry *de;
3139 LIST_HEAD(list);
3140
3141 list_for_each_entry_reverse(de, &ctx->defer_list, list) {
3142 if (io_match_task_safe(head: de->req, tctx, cancel_all)) {
3143 list_cut_position(list: &list, head: &ctx->defer_list, entry: &de->list);
3144 break;
3145 }
3146 }
3147 if (list_empty(head: &list))
3148 return false;
3149
3150 while (!list_empty(head: &list)) {
3151 de = list_first_entry(&list, struct io_defer_entry, list);
3152 list_del_init(entry: &de->list);
3153 ctx->nr_drained -= io_linked_nr(req: de->req);
3154 io_req_task_queue_fail(req: de->req, ret: -ECANCELED);
3155 kfree(objp: de);
3156 }
3157 return true;
3158}
3159
3160static __cold bool io_uring_try_cancel_iowq(struct io_ring_ctx *ctx)
3161{
3162 struct io_tctx_node *node;
3163 enum io_wq_cancel cret;
3164 bool ret = false;
3165
3166 mutex_lock(lock: &ctx->uring_lock);
3167 list_for_each_entry(node, &ctx->tctx_list, ctx_node) {
3168 struct io_uring_task *tctx = node->task->io_uring;
3169
3170 /*
3171 * io_wq will stay alive while we hold uring_lock, because it's
3172 * killed after ctx nodes, which requires to take the lock.
3173 */
3174 if (!tctx || !tctx->io_wq)
3175 continue;
3176 cret = io_wq_cancel_cb(wq: tctx->io_wq, cancel: io_cancel_ctx_cb, data: ctx, cancel_all: true);
3177 ret |= (cret != IO_WQ_CANCEL_NOTFOUND);
3178 }
3179 mutex_unlock(lock: &ctx->uring_lock);
3180
3181 return ret;
3182}
3183
3184static __cold bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx,
3185 struct io_uring_task *tctx,
3186 bool cancel_all,
3187 bool is_sqpoll_thread)
3188{
3189 struct io_task_cancel cancel = { .tctx = tctx, .all = cancel_all, };
3190 enum io_wq_cancel cret;
3191 bool ret = false;
3192
3193 /* set it so io_req_local_work_add() would wake us up */
3194 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) {
3195 atomic_set(v: &ctx->cq_wait_nr, i: 1);
3196 smp_mb();
3197 }
3198
3199 /* failed during ring init, it couldn't have issued any requests */
3200 if (!ctx->rings)
3201 return false;
3202
3203 if (!tctx) {
3204 ret |= io_uring_try_cancel_iowq(ctx);
3205 } else if (tctx->io_wq) {
3206 /*
3207 * Cancels requests of all rings, not only @ctx, but
3208 * it's fine as the task is in exit/exec.
3209 */
3210 cret = io_wq_cancel_cb(wq: tctx->io_wq, cancel: io_cancel_task_cb,
3211 data: &cancel, cancel_all: true);
3212 ret |= (cret != IO_WQ_CANCEL_NOTFOUND);
3213 }
3214
3215 /* SQPOLL thread does its own polling */
3216 if ((!(ctx->flags & IORING_SETUP_SQPOLL) && cancel_all) ||
3217 is_sqpoll_thread) {
3218 while (!wq_list_empty(&ctx->iopoll_list)) {
3219 io_iopoll_try_reap_events(ctx);
3220 ret = true;
3221 cond_resched();
3222 }
3223 }
3224
3225 if ((ctx->flags & IORING_SETUP_DEFER_TASKRUN) &&
3226 io_allowed_defer_tw_run(ctx))
3227 ret |= io_run_local_work(ctx, INT_MAX, INT_MAX) > 0;
3228 mutex_lock(lock: &ctx->uring_lock);
3229 ret |= io_cancel_defer_files(ctx, tctx, cancel_all);
3230 ret |= io_poll_remove_all(ctx, tctx, cancel_all);
3231 ret |= io_waitid_remove_all(ctx, tctx, cancel_all);
3232 ret |= io_futex_remove_all(ctx, tctx, cancel_all);
3233 ret |= io_uring_try_cancel_uring_cmd(ctx, tctx, cancel_all);
3234 mutex_unlock(lock: &ctx->uring_lock);
3235 ret |= io_kill_timeouts(ctx, tctx, cancel_all);
3236 if (tctx)
3237 ret |= io_run_task_work() > 0;
3238 else
3239 ret |= flush_delayed_work(dwork: &ctx->fallback_work);
3240 return ret;
3241}
3242
3243static s64 tctx_inflight(struct io_uring_task *tctx, bool tracked)
3244{
3245 if (tracked)
3246 return atomic_read(v: &tctx->inflight_tracked);
3247 return percpu_counter_sum(fbc: &tctx->inflight);
3248}
3249
3250/*
3251 * Find any io_uring ctx that this task has registered or done IO on, and cancel
3252 * requests. @sqd should be not-null IFF it's an SQPOLL thread cancellation.
3253 */
3254__cold void io_uring_cancel_generic(bool cancel_all, struct io_sq_data *sqd)
3255{
3256 struct io_uring_task *tctx = current->io_uring;
3257 struct io_ring_ctx *ctx;
3258 struct io_tctx_node *node;
3259 unsigned long index;
3260 s64 inflight;
3261 DEFINE_WAIT(wait);
3262
3263 WARN_ON_ONCE(sqd && sqpoll_task_locked(sqd) != current);
3264
3265 if (!current->io_uring)
3266 return;
3267 if (tctx->io_wq)
3268 io_wq_exit_start(wq: tctx->io_wq);
3269
3270 atomic_inc(v: &tctx->in_cancel);
3271 do {
3272 bool loop = false;
3273
3274 io_uring_drop_tctx_refs(current);
3275 if (!tctx_inflight(tctx, tracked: !cancel_all))
3276 break;
3277
3278 /* read completions before cancelations */
3279 inflight = tctx_inflight(tctx, tracked: false);
3280 if (!inflight)
3281 break;
3282
3283 if (!sqd) {
3284 xa_for_each(&tctx->xa, index, node) {
3285 /* sqpoll task will cancel all its requests */
3286 if (node->ctx->sq_data)
3287 continue;
3288 loop |= io_uring_try_cancel_requests(ctx: node->ctx,
3289 current->io_uring,
3290 cancel_all,
3291 is_sqpoll_thread: false);
3292 }
3293 } else {
3294 list_for_each_entry(ctx, &sqd->ctx_list, sqd_list)
3295 loop |= io_uring_try_cancel_requests(ctx,
3296 current->io_uring,
3297 cancel_all,
3298 is_sqpoll_thread: true);
3299 }
3300
3301 if (loop) {
3302 cond_resched();
3303 continue;
3304 }
3305
3306 prepare_to_wait(wq_head: &tctx->wait, wq_entry: &wait, TASK_INTERRUPTIBLE);
3307 io_run_task_work();
3308 io_uring_drop_tctx_refs(current);
3309 xa_for_each(&tctx->xa, index, node) {
3310 if (io_local_work_pending(ctx: node->ctx)) {
3311 WARN_ON_ONCE(node->ctx->submitter_task &&
3312 node->ctx->submitter_task != current);
3313 goto end_wait;
3314 }
3315 }
3316 /*
3317 * If we've seen completions, retry without waiting. This
3318 * avoids a race where a completion comes in before we did
3319 * prepare_to_wait().
3320 */
3321 if (inflight == tctx_inflight(tctx, tracked: !cancel_all))
3322 schedule();
3323end_wait:
3324 finish_wait(wq_head: &tctx->wait, wq_entry: &wait);
3325 } while (1);
3326
3327 io_uring_clean_tctx(tctx);
3328 if (cancel_all) {
3329 /*
3330 * We shouldn't run task_works after cancel, so just leave
3331 * ->in_cancel set for normal exit.
3332 */
3333 atomic_dec(v: &tctx->in_cancel);
3334 /* for exec all current's requests should be gone, kill tctx */
3335 __io_uring_free(current);
3336 }
3337}
3338
3339void __io_uring_cancel(bool cancel_all)
3340{
3341 io_uring_unreg_ringfd();
3342 io_uring_cancel_generic(cancel_all, NULL);
3343}
3344
3345static struct io_uring_reg_wait *io_get_ext_arg_reg(struct io_ring_ctx *ctx,
3346 const struct io_uring_getevents_arg __user *uarg)
3347{
3348 unsigned long size = sizeof(struct io_uring_reg_wait);
3349 unsigned long offset = (uintptr_t)uarg;
3350 unsigned long end;
3351
3352 if (unlikely(offset % sizeof(long)))
3353 return ERR_PTR(error: -EFAULT);
3354
3355 /* also protects from NULL ->cq_wait_arg as the size would be 0 */
3356 if (unlikely(check_add_overflow(offset, size, &end) ||
3357 end > ctx->cq_wait_size))
3358 return ERR_PTR(error: -EFAULT);
3359
3360 offset = array_index_nospec(offset, ctx->cq_wait_size - size);
3361 return ctx->cq_wait_arg + offset;
3362}
3363
3364static int io_validate_ext_arg(struct io_ring_ctx *ctx, unsigned flags,
3365 const void __user *argp, size_t argsz)
3366{
3367 struct io_uring_getevents_arg arg;
3368
3369 if (!(flags & IORING_ENTER_EXT_ARG))
3370 return 0;
3371 if (flags & IORING_ENTER_EXT_ARG_REG)
3372 return -EINVAL;
3373 if (argsz != sizeof(arg))
3374 return -EINVAL;
3375 if (copy_from_user(to: &arg, from: argp, n: sizeof(arg)))
3376 return -EFAULT;
3377 return 0;
3378}
3379
3380static int io_get_ext_arg(struct io_ring_ctx *ctx, unsigned flags,
3381 const void __user *argp, struct ext_arg *ext_arg)
3382{
3383 const struct io_uring_getevents_arg __user *uarg = argp;
3384 struct io_uring_getevents_arg arg;
3385
3386 ext_arg->iowait = !(flags & IORING_ENTER_NO_IOWAIT);
3387
3388 /*
3389 * If EXT_ARG isn't set, then we have no timespec and the argp pointer
3390 * is just a pointer to the sigset_t.
3391 */
3392 if (!(flags & IORING_ENTER_EXT_ARG)) {
3393 ext_arg->sig = (const sigset_t __user *) argp;
3394 return 0;
3395 }
3396
3397 if (flags & IORING_ENTER_EXT_ARG_REG) {
3398 struct io_uring_reg_wait *w;
3399
3400 if (ext_arg->argsz != sizeof(struct io_uring_reg_wait))
3401 return -EINVAL;
3402 w = io_get_ext_arg_reg(ctx, uarg: argp);
3403 if (IS_ERR(ptr: w))
3404 return PTR_ERR(ptr: w);
3405
3406 if (w->flags & ~IORING_REG_WAIT_TS)
3407 return -EINVAL;
3408 ext_arg->min_time = READ_ONCE(w->min_wait_usec) * NSEC_PER_USEC;
3409 ext_arg->sig = u64_to_user_ptr(READ_ONCE(w->sigmask));
3410 ext_arg->argsz = READ_ONCE(w->sigmask_sz);
3411 if (w->flags & IORING_REG_WAIT_TS) {
3412 ext_arg->ts.tv_sec = READ_ONCE(w->ts.tv_sec);
3413 ext_arg->ts.tv_nsec = READ_ONCE(w->ts.tv_nsec);
3414 ext_arg->ts_set = true;
3415 }
3416 return 0;
3417 }
3418
3419 /*
3420 * EXT_ARG is set - ensure we agree on the size of it and copy in our
3421 * timespec and sigset_t pointers if good.
3422 */
3423 if (ext_arg->argsz != sizeof(arg))
3424 return -EINVAL;
3425#ifdef CONFIG_64BIT
3426 if (!user_access_begin(uarg, sizeof(*uarg)))
3427 return -EFAULT;
3428 unsafe_get_user(arg.sigmask, &uarg->sigmask, uaccess_end);
3429 unsafe_get_user(arg.sigmask_sz, &uarg->sigmask_sz, uaccess_end);
3430 unsafe_get_user(arg.min_wait_usec, &uarg->min_wait_usec, uaccess_end);
3431 unsafe_get_user(arg.ts, &uarg->ts, uaccess_end);
3432 user_access_end();
3433#else
3434 if (copy_from_user(&arg, uarg, sizeof(arg)))
3435 return -EFAULT;
3436#endif
3437 ext_arg->min_time = arg.min_wait_usec * NSEC_PER_USEC;
3438 ext_arg->sig = u64_to_user_ptr(arg.sigmask);
3439 ext_arg->argsz = arg.sigmask_sz;
3440 if (arg.ts) {
3441 if (get_timespec64(ts: &ext_arg->ts, u64_to_user_ptr(arg.ts)))
3442 return -EFAULT;
3443 ext_arg->ts_set = true;
3444 }
3445 return 0;
3446#ifdef CONFIG_64BIT
3447uaccess_end:
3448 user_access_end();
3449 return -EFAULT;
3450#endif
3451}
3452
3453SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
3454 u32, min_complete, u32, flags, const void __user *, argp,
3455 size_t, argsz)
3456{
3457 struct io_ring_ctx *ctx;
3458 struct file *file;
3459 long ret;
3460
3461 if (unlikely(flags & ~IORING_ENTER_FLAGS))
3462 return -EINVAL;
3463
3464 /*
3465 * Ring fd has been registered via IORING_REGISTER_RING_FDS, we
3466 * need only dereference our task private array to find it.
3467 */
3468 if (flags & IORING_ENTER_REGISTERED_RING) {
3469 struct io_uring_task *tctx = current->io_uring;
3470
3471 if (unlikely(!tctx || fd >= IO_RINGFD_REG_MAX))
3472 return -EINVAL;
3473 fd = array_index_nospec(fd, IO_RINGFD_REG_MAX);
3474 file = tctx->registered_rings[fd];
3475 if (unlikely(!file))
3476 return -EBADF;
3477 } else {
3478 file = fget(fd);
3479 if (unlikely(!file))
3480 return -EBADF;
3481 ret = -EOPNOTSUPP;
3482 if (unlikely(!io_is_uring_fops(file)))
3483 goto out;
3484 }
3485
3486 ctx = file->private_data;
3487 ret = -EBADFD;
3488 if (unlikely(ctx->flags & IORING_SETUP_R_DISABLED))
3489 goto out;
3490
3491 /*
3492 * For SQ polling, the thread will do all submissions and completions.
3493 * Just return the requested submit count, and wake the thread if
3494 * we were asked to.
3495 */
3496 ret = 0;
3497 if (ctx->flags & IORING_SETUP_SQPOLL) {
3498 if (unlikely(ctx->sq_data->thread == NULL)) {
3499 ret = -EOWNERDEAD;
3500 goto out;
3501 }
3502 if (flags & IORING_ENTER_SQ_WAKEUP)
3503 wake_up(&ctx->sq_data->wait);
3504 if (flags & IORING_ENTER_SQ_WAIT)
3505 io_sqpoll_wait_sq(ctx);
3506
3507 ret = to_submit;
3508 } else if (to_submit) {
3509 ret = io_uring_add_tctx_node(ctx);
3510 if (unlikely(ret))
3511 goto out;
3512
3513 mutex_lock(lock: &ctx->uring_lock);
3514 ret = io_submit_sqes(ctx, nr: to_submit);
3515 if (ret != to_submit) {
3516 mutex_unlock(lock: &ctx->uring_lock);
3517 goto out;
3518 }
3519 if (flags & IORING_ENTER_GETEVENTS) {
3520 if (ctx->syscall_iopoll)
3521 goto iopoll_locked;
3522 /*
3523 * Ignore errors, we'll soon call io_cqring_wait() and
3524 * it should handle ownership problems if any.
3525 */
3526 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
3527 (void)io_run_local_work_locked(ctx, min_events: min_complete);
3528 }
3529 mutex_unlock(lock: &ctx->uring_lock);
3530 }
3531
3532 if (flags & IORING_ENTER_GETEVENTS) {
3533 int ret2;
3534
3535 if (ctx->syscall_iopoll) {
3536 /*
3537 * We disallow the app entering submit/complete with
3538 * polling, but we still need to lock the ring to
3539 * prevent racing with polled issue that got punted to
3540 * a workqueue.
3541 */
3542 mutex_lock(lock: &ctx->uring_lock);
3543iopoll_locked:
3544 ret2 = io_validate_ext_arg(ctx, flags, argp, argsz);
3545 if (likely(!ret2))
3546 ret2 = io_iopoll_check(ctx, min_events: min_complete);
3547 mutex_unlock(lock: &ctx->uring_lock);
3548 } else {
3549 struct ext_arg ext_arg = { .argsz = argsz };
3550
3551 ret2 = io_get_ext_arg(ctx, flags, argp, ext_arg: &ext_arg);
3552 if (likely(!ret2))
3553 ret2 = io_cqring_wait(ctx, min_events: min_complete, flags,
3554 ext_arg: &ext_arg);
3555 }
3556
3557 if (!ret) {
3558 ret = ret2;
3559
3560 /*
3561 * EBADR indicates that one or more CQE were dropped.
3562 * Once the user has been informed we can clear the bit
3563 * as they are obviously ok with those drops.
3564 */
3565 if (unlikely(ret2 == -EBADR))
3566 clear_bit(nr: IO_CHECK_CQ_DROPPED_BIT,
3567 addr: &ctx->check_cq);
3568 }
3569 }
3570out:
3571 if (!(flags & IORING_ENTER_REGISTERED_RING))
3572 fput(file);
3573 return ret;
3574}
3575
3576static const struct file_operations io_uring_fops = {
3577 .release = io_uring_release,
3578 .mmap = io_uring_mmap,
3579 .get_unmapped_area = io_uring_get_unmapped_area,
3580#ifndef CONFIG_MMU
3581 .mmap_capabilities = io_uring_nommu_mmap_capabilities,
3582#endif
3583 .poll = io_uring_poll,
3584#ifdef CONFIG_PROC_FS
3585 .show_fdinfo = io_uring_show_fdinfo,
3586#endif
3587};
3588
3589bool io_is_uring_fops(struct file *file)
3590{
3591 return file->f_op == &io_uring_fops;
3592}
3593
3594static __cold int io_allocate_scq_urings(struct io_ring_ctx *ctx,
3595 struct io_uring_params *p)
3596{
3597 struct io_uring_region_desc rd;
3598 struct io_rings *rings;
3599 size_t size, sq_array_offset;
3600 int ret;
3601
3602 /* make sure these are sane, as we already accounted them */
3603 ctx->sq_entries = p->sq_entries;
3604 ctx->cq_entries = p->cq_entries;
3605
3606 size = rings_size(flags: ctx->flags, sq_entries: p->sq_entries, cq_entries: p->cq_entries,
3607 sq_offset: &sq_array_offset);
3608 if (size == SIZE_MAX)
3609 return -EOVERFLOW;
3610
3611 memset(s: &rd, c: 0, n: sizeof(rd));
3612 rd.size = PAGE_ALIGN(size);
3613 if (ctx->flags & IORING_SETUP_NO_MMAP) {
3614 rd.user_addr = p->cq_off.user_addr;
3615 rd.flags |= IORING_MEM_REGION_TYPE_USER;
3616 }
3617 ret = io_create_region(ctx, mr: &ctx->ring_region, reg: &rd, IORING_OFF_CQ_RING);
3618 if (ret)
3619 return ret;
3620 ctx->rings = rings = io_region_get_ptr(mr: &ctx->ring_region);
3621
3622 if (!(ctx->flags & IORING_SETUP_NO_SQARRAY))
3623 ctx->sq_array = (u32 *)((char *)rings + sq_array_offset);
3624 rings->sq_ring_mask = p->sq_entries - 1;
3625 rings->cq_ring_mask = p->cq_entries - 1;
3626 rings->sq_ring_entries = p->sq_entries;
3627 rings->cq_ring_entries = p->cq_entries;
3628
3629 if (p->flags & IORING_SETUP_SQE128)
3630 size = array_size(2 * sizeof(struct io_uring_sqe), p->sq_entries);
3631 else
3632 size = array_size(sizeof(struct io_uring_sqe), p->sq_entries);
3633 if (size == SIZE_MAX) {
3634 io_rings_free(ctx);
3635 return -EOVERFLOW;
3636 }
3637
3638 memset(s: &rd, c: 0, n: sizeof(rd));
3639 rd.size = PAGE_ALIGN(size);
3640 if (ctx->flags & IORING_SETUP_NO_MMAP) {
3641 rd.user_addr = p->sq_off.user_addr;
3642 rd.flags |= IORING_MEM_REGION_TYPE_USER;
3643 }
3644 ret = io_create_region(ctx, mr: &ctx->sq_region, reg: &rd, IORING_OFF_SQES);
3645 if (ret) {
3646 io_rings_free(ctx);
3647 return ret;
3648 }
3649 ctx->sq_sqes = io_region_get_ptr(mr: &ctx->sq_region);
3650 return 0;
3651}
3652
3653static int io_uring_install_fd(struct file *file)
3654{
3655 int fd;
3656
3657 fd = get_unused_fd_flags(O_RDWR | O_CLOEXEC);
3658 if (fd < 0)
3659 return fd;
3660 fd_install(fd, file);
3661 return fd;
3662}
3663
3664/*
3665 * Allocate an anonymous fd, this is what constitutes the application
3666 * visible backing of an io_uring instance. The application mmaps this
3667 * fd to gain access to the SQ/CQ ring details.
3668 */
3669static struct file *io_uring_get_file(struct io_ring_ctx *ctx)
3670{
3671 /* Create a new inode so that the LSM can block the creation. */
3672 return anon_inode_create_getfile(name: "[io_uring]", fops: &io_uring_fops, priv: ctx,
3673 O_RDWR | O_CLOEXEC, NULL);
3674}
3675
3676static int io_uring_sanitise_params(struct io_uring_params *p)
3677{
3678 unsigned flags = p->flags;
3679
3680 /* There is no way to mmap rings without a real fd */
3681 if ((flags & IORING_SETUP_REGISTERED_FD_ONLY) &&
3682 !(flags & IORING_SETUP_NO_MMAP))
3683 return -EINVAL;
3684
3685 if (flags & IORING_SETUP_SQPOLL) {
3686 /* IPI related flags don't make sense with SQPOLL */
3687 if (flags & (IORING_SETUP_COOP_TASKRUN |
3688 IORING_SETUP_TASKRUN_FLAG |
3689 IORING_SETUP_DEFER_TASKRUN))
3690 return -EINVAL;
3691 }
3692
3693 if (flags & IORING_SETUP_TASKRUN_FLAG) {
3694 if (!(flags & (IORING_SETUP_COOP_TASKRUN |
3695 IORING_SETUP_DEFER_TASKRUN)))
3696 return -EINVAL;
3697 }
3698
3699 /* HYBRID_IOPOLL only valid with IOPOLL */
3700 if ((flags & IORING_SETUP_HYBRID_IOPOLL) && !(flags & IORING_SETUP_IOPOLL))
3701 return -EINVAL;
3702
3703 /*
3704 * For DEFER_TASKRUN we require the completion task to be the same as
3705 * the submission task. This implies that there is only one submitter.
3706 */
3707 if ((flags & IORING_SETUP_DEFER_TASKRUN) &&
3708 !(flags & IORING_SETUP_SINGLE_ISSUER))
3709 return -EINVAL;
3710
3711 /*
3712 * Nonsensical to ask for CQE32 and mixed CQE support, it's not
3713 * supported to post 16b CQEs on a ring setup with CQE32.
3714 */
3715 if ((flags & (IORING_SETUP_CQE32|IORING_SETUP_CQE_MIXED)) ==
3716 (IORING_SETUP_CQE32|IORING_SETUP_CQE_MIXED))
3717 return -EINVAL;
3718
3719 return 0;
3720}
3721
3722int io_uring_fill_params(unsigned entries, struct io_uring_params *p)
3723{
3724 if (!entries)
3725 return -EINVAL;
3726 if (entries > IORING_MAX_ENTRIES) {
3727 if (!(p->flags & IORING_SETUP_CLAMP))
3728 return -EINVAL;
3729 entries = IORING_MAX_ENTRIES;
3730 }
3731
3732 /*
3733 * Use twice as many entries for the CQ ring. It's possible for the
3734 * application to drive a higher depth than the size of the SQ ring,
3735 * since the sqes are only used at submission time. This allows for
3736 * some flexibility in overcommitting a bit. If the application has
3737 * set IORING_SETUP_CQSIZE, it will have passed in the desired number
3738 * of CQ ring entries manually.
3739 */
3740 p->sq_entries = roundup_pow_of_two(entries);
3741 if (p->flags & IORING_SETUP_CQSIZE) {
3742 /*
3743 * If IORING_SETUP_CQSIZE is set, we do the same roundup
3744 * to a power-of-two, if it isn't already. We do NOT impose
3745 * any cq vs sq ring sizing.
3746 */
3747 if (!p->cq_entries)
3748 return -EINVAL;
3749 if (p->cq_entries > IORING_MAX_CQ_ENTRIES) {
3750 if (!(p->flags & IORING_SETUP_CLAMP))
3751 return -EINVAL;
3752 p->cq_entries = IORING_MAX_CQ_ENTRIES;
3753 }
3754 p->cq_entries = roundup_pow_of_two(p->cq_entries);
3755 if (p->cq_entries < p->sq_entries)
3756 return -EINVAL;
3757 } else {
3758 p->cq_entries = 2 * p->sq_entries;
3759 }
3760
3761 p->sq_off.head = offsetof(struct io_rings, sq.head);
3762 p->sq_off.tail = offsetof(struct io_rings, sq.tail);
3763 p->sq_off.ring_mask = offsetof(struct io_rings, sq_ring_mask);
3764 p->sq_off.ring_entries = offsetof(struct io_rings, sq_ring_entries);
3765 p->sq_off.flags = offsetof(struct io_rings, sq_flags);
3766 p->sq_off.dropped = offsetof(struct io_rings, sq_dropped);
3767 p->sq_off.resv1 = 0;
3768 if (!(p->flags & IORING_SETUP_NO_MMAP))
3769 p->sq_off.user_addr = 0;
3770
3771 p->cq_off.head = offsetof(struct io_rings, cq.head);
3772 p->cq_off.tail = offsetof(struct io_rings, cq.tail);
3773 p->cq_off.ring_mask = offsetof(struct io_rings, cq_ring_mask);
3774 p->cq_off.ring_entries = offsetof(struct io_rings, cq_ring_entries);
3775 p->cq_off.overflow = offsetof(struct io_rings, cq_overflow);
3776 p->cq_off.cqes = offsetof(struct io_rings, cqes);
3777 p->cq_off.flags = offsetof(struct io_rings, cq_flags);
3778 p->cq_off.resv1 = 0;
3779 if (!(p->flags & IORING_SETUP_NO_MMAP))
3780 p->cq_off.user_addr = 0;
3781
3782 return 0;
3783}
3784
3785static __cold int io_uring_create(unsigned entries, struct io_uring_params *p,
3786 struct io_uring_params __user *params)
3787{
3788 struct io_ring_ctx *ctx;
3789 struct io_uring_task *tctx;
3790 struct file *file;
3791 int ret;
3792
3793 ret = io_uring_sanitise_params(p);
3794 if (ret)
3795 return ret;
3796
3797 ret = io_uring_fill_params(entries, p);
3798 if (unlikely(ret))
3799 return ret;
3800
3801 ctx = io_ring_ctx_alloc(p);
3802 if (!ctx)
3803 return -ENOMEM;
3804
3805 ctx->clockid = CLOCK_MONOTONIC;
3806 ctx->clock_offset = 0;
3807
3808 if (!(ctx->flags & IORING_SETUP_NO_SQARRAY))
3809 static_branch_inc(&io_key_has_sqarray);
3810
3811 if ((ctx->flags & IORING_SETUP_DEFER_TASKRUN) &&
3812 !(ctx->flags & IORING_SETUP_IOPOLL) &&
3813 !(ctx->flags & IORING_SETUP_SQPOLL))
3814 ctx->task_complete = true;
3815
3816 if (ctx->task_complete || (ctx->flags & IORING_SETUP_IOPOLL))
3817 ctx->lockless_cq = true;
3818
3819 /*
3820 * lazy poll_wq activation relies on ->task_complete for synchronisation
3821 * purposes, see io_activate_pollwq()
3822 */
3823 if (!ctx->task_complete)
3824 ctx->poll_activated = true;
3825
3826 /*
3827 * When SETUP_IOPOLL and SETUP_SQPOLL are both enabled, user
3828 * space applications don't need to do io completion events
3829 * polling again, they can rely on io_sq_thread to do polling
3830 * work, which can reduce cpu usage and uring_lock contention.
3831 */
3832 if (ctx->flags & IORING_SETUP_IOPOLL &&
3833 !(ctx->flags & IORING_SETUP_SQPOLL))
3834 ctx->syscall_iopoll = 1;
3835
3836 ctx->compat = in_compat_syscall();
3837 if (!ns_capable_noaudit(ns: &init_user_ns, CAP_IPC_LOCK))
3838 ctx->user = get_uid(current_user());
3839
3840 /*
3841 * For SQPOLL, we just need a wakeup, always. For !SQPOLL, if
3842 * COOP_TASKRUN is set, then IPIs are never needed by the app.
3843 */
3844 if (ctx->flags & (IORING_SETUP_SQPOLL|IORING_SETUP_COOP_TASKRUN))
3845 ctx->notify_method = TWA_SIGNAL_NO_IPI;
3846 else
3847 ctx->notify_method = TWA_SIGNAL;
3848
3849 /*
3850 * This is just grabbed for accounting purposes. When a process exits,
3851 * the mm is exited and dropped before the files, hence we need to hang
3852 * on to this mm purely for the purposes of being able to unaccount
3853 * memory (locked/pinned vm). It's not used for anything else.
3854 */
3855 mmgrab(current->mm);
3856 ctx->mm_account = current->mm;
3857
3858 ret = io_allocate_scq_urings(ctx, p);
3859 if (ret)
3860 goto err;
3861
3862 if (!(p->flags & IORING_SETUP_NO_SQARRAY))
3863 p->sq_off.array = (char *)ctx->sq_array - (char *)ctx->rings;
3864
3865 ret = io_sq_offload_create(ctx, p);
3866 if (ret)
3867 goto err;
3868
3869 p->features = IORING_FEAT_FLAGS;
3870
3871 if (copy_to_user(to: params, from: p, n: sizeof(*p))) {
3872 ret = -EFAULT;
3873 goto err;
3874 }
3875
3876 if (ctx->flags & IORING_SETUP_SINGLE_ISSUER
3877 && !(ctx->flags & IORING_SETUP_R_DISABLED)) {
3878 /*
3879 * Unlike io_register_enable_rings(), don't need WRITE_ONCE()
3880 * since ctx isn't yet accessible from other tasks
3881 */
3882 ctx->submitter_task = get_task_struct(current);
3883 }
3884
3885 file = io_uring_get_file(ctx);
3886 if (IS_ERR(ptr: file)) {
3887 ret = PTR_ERR(ptr: file);
3888 goto err;
3889 }
3890
3891 ret = __io_uring_add_tctx_node(ctx);
3892 if (ret)
3893 goto err_fput;
3894 tctx = current->io_uring;
3895
3896 /*
3897 * Install ring fd as the very last thing, so we don't risk someone
3898 * having closed it before we finish setup
3899 */
3900 if (p->flags & IORING_SETUP_REGISTERED_FD_ONLY)
3901 ret = io_ring_add_registered_file(tctx, file, start: 0, IO_RINGFD_REG_MAX);
3902 else
3903 ret = io_uring_install_fd(file);
3904 if (ret < 0)
3905 goto err_fput;
3906
3907 trace_io_uring_create(fd: ret, ctx, sq_entries: p->sq_entries, cq_entries: p->cq_entries, flags: p->flags);
3908 return ret;
3909err:
3910 io_ring_ctx_wait_and_kill(ctx);
3911 return ret;
3912err_fput:
3913 fput(file);
3914 return ret;
3915}
3916
3917/*
3918 * Sets up an aio uring context, and returns the fd. Applications asks for a
3919 * ring size, we return the actual sq/cq ring sizes (among other things) in the
3920 * params structure passed in.
3921 */
3922static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
3923{
3924 struct io_uring_params p;
3925 int i;
3926
3927 if (copy_from_user(to: &p, from: params, n: sizeof(p)))
3928 return -EFAULT;
3929 for (i = 0; i < ARRAY_SIZE(p.resv); i++) {
3930 if (p.resv[i])
3931 return -EINVAL;
3932 }
3933
3934 if (p.flags & ~IORING_SETUP_FLAGS)
3935 return -EINVAL;
3936 return io_uring_create(entries, p: &p, params);
3937}
3938
3939static inline int io_uring_allowed(void)
3940{
3941 int disabled = READ_ONCE(sysctl_io_uring_disabled);
3942 kgid_t io_uring_group;
3943
3944 if (disabled == 2)
3945 return -EPERM;
3946
3947 if (disabled == 0 || capable(CAP_SYS_ADMIN))
3948 goto allowed_lsm;
3949
3950 io_uring_group = make_kgid(from: &init_user_ns, gid: sysctl_io_uring_group);
3951 if (!gid_valid(gid: io_uring_group))
3952 return -EPERM;
3953
3954 if (!in_group_p(io_uring_group))
3955 return -EPERM;
3956
3957allowed_lsm:
3958 return security_uring_allowed();
3959}
3960
3961SYSCALL_DEFINE2(io_uring_setup, u32, entries,
3962 struct io_uring_params __user *, params)
3963{
3964 int ret;
3965
3966 ret = io_uring_allowed();
3967 if (ret)
3968 return ret;
3969
3970 return io_uring_setup(entries, params);
3971}
3972
3973static int __init io_uring_init(void)
3974{
3975 struct kmem_cache_args kmem_args = {
3976 .useroffset = offsetof(struct io_kiocb, cmd.data),
3977 .usersize = sizeof_field(struct io_kiocb, cmd.data),
3978 .freeptr_offset = offsetof(struct io_kiocb, work),
3979 .use_freeptr_offset = true,
3980 };
3981
3982#define __BUILD_BUG_VERIFY_OFFSET_SIZE(stype, eoffset, esize, ename) do { \
3983 BUILD_BUG_ON(offsetof(stype, ename) != eoffset); \
3984 BUILD_BUG_ON(sizeof_field(stype, ename) != esize); \
3985} while (0)
3986
3987#define BUILD_BUG_SQE_ELEM(eoffset, etype, ename) \
3988 __BUILD_BUG_VERIFY_OFFSET_SIZE(struct io_uring_sqe, eoffset, sizeof(etype), ename)
3989#define BUILD_BUG_SQE_ELEM_SIZE(eoffset, esize, ename) \
3990 __BUILD_BUG_VERIFY_OFFSET_SIZE(struct io_uring_sqe, eoffset, esize, ename)
3991 BUILD_BUG_ON(sizeof(struct io_uring_sqe) != 64);
3992 BUILD_BUG_SQE_ELEM(0, __u8, opcode);
3993 BUILD_BUG_SQE_ELEM(1, __u8, flags);
3994 BUILD_BUG_SQE_ELEM(2, __u16, ioprio);
3995 BUILD_BUG_SQE_ELEM(4, __s32, fd);
3996 BUILD_BUG_SQE_ELEM(8, __u64, off);
3997 BUILD_BUG_SQE_ELEM(8, __u64, addr2);
3998 BUILD_BUG_SQE_ELEM(8, __u32, cmd_op);
3999 BUILD_BUG_SQE_ELEM(12, __u32, __pad1);
4000 BUILD_BUG_SQE_ELEM(16, __u64, addr);
4001 BUILD_BUG_SQE_ELEM(16, __u64, splice_off_in);
4002 BUILD_BUG_SQE_ELEM(24, __u32, len);
4003 BUILD_BUG_SQE_ELEM(28, __kernel_rwf_t, rw_flags);
4004 BUILD_BUG_SQE_ELEM(28, /* compat */ int, rw_flags);
4005 BUILD_BUG_SQE_ELEM(28, /* compat */ __u32, rw_flags);
4006 BUILD_BUG_SQE_ELEM(28, __u32, fsync_flags);
4007 BUILD_BUG_SQE_ELEM(28, /* compat */ __u16, poll_events);
4008 BUILD_BUG_SQE_ELEM(28, __u32, poll32_events);
4009 BUILD_BUG_SQE_ELEM(28, __u32, sync_range_flags);
4010 BUILD_BUG_SQE_ELEM(28, __u32, msg_flags);
4011 BUILD_BUG_SQE_ELEM(28, __u32, timeout_flags);
4012 BUILD_BUG_SQE_ELEM(28, __u32, accept_flags);
4013 BUILD_BUG_SQE_ELEM(28, __u32, cancel_flags);
4014 BUILD_BUG_SQE_ELEM(28, __u32, open_flags);
4015 BUILD_BUG_SQE_ELEM(28, __u32, statx_flags);
4016 BUILD_BUG_SQE_ELEM(28, __u32, fadvise_advice);
4017 BUILD_BUG_SQE_ELEM(28, __u32, splice_flags);
4018 BUILD_BUG_SQE_ELEM(28, __u32, rename_flags);
4019 BUILD_BUG_SQE_ELEM(28, __u32, unlink_flags);
4020 BUILD_BUG_SQE_ELEM(28, __u32, hardlink_flags);
4021 BUILD_BUG_SQE_ELEM(28, __u32, xattr_flags);
4022 BUILD_BUG_SQE_ELEM(28, __u32, msg_ring_flags);
4023 BUILD_BUG_SQE_ELEM(32, __u64, user_data);
4024 BUILD_BUG_SQE_ELEM(40, __u16, buf_index);
4025 BUILD_BUG_SQE_ELEM(40, __u16, buf_group);
4026 BUILD_BUG_SQE_ELEM(42, __u16, personality);
4027 BUILD_BUG_SQE_ELEM(44, __s32, splice_fd_in);
4028 BUILD_BUG_SQE_ELEM(44, __u32, file_index);
4029 BUILD_BUG_SQE_ELEM(44, __u16, addr_len);
4030 BUILD_BUG_SQE_ELEM(44, __u8, write_stream);
4031 BUILD_BUG_SQE_ELEM(45, __u8, __pad4[0]);
4032 BUILD_BUG_SQE_ELEM(46, __u16, __pad3[0]);
4033 BUILD_BUG_SQE_ELEM(48, __u64, addr3);
4034 BUILD_BUG_SQE_ELEM_SIZE(48, 0, cmd);
4035 BUILD_BUG_SQE_ELEM(48, __u64, attr_ptr);
4036 BUILD_BUG_SQE_ELEM(56, __u64, attr_type_mask);
4037 BUILD_BUG_SQE_ELEM(56, __u64, __pad2);
4038
4039 BUILD_BUG_ON(sizeof(struct io_uring_files_update) !=
4040 sizeof(struct io_uring_rsrc_update));
4041 BUILD_BUG_ON(sizeof(struct io_uring_rsrc_update) >
4042 sizeof(struct io_uring_rsrc_update2));
4043
4044 /* ->buf_index is u16 */
4045 BUILD_BUG_ON(offsetof(struct io_uring_buf_ring, bufs) != 0);
4046 BUILD_BUG_ON(offsetof(struct io_uring_buf, resv) !=
4047 offsetof(struct io_uring_buf_ring, tail));
4048
4049 /* should fit into one byte */
4050 BUILD_BUG_ON(SQE_VALID_FLAGS >= (1 << 8));
4051 BUILD_BUG_ON(SQE_COMMON_FLAGS >= (1 << 8));
4052 BUILD_BUG_ON((SQE_VALID_FLAGS | SQE_COMMON_FLAGS) != SQE_VALID_FLAGS);
4053
4054 BUILD_BUG_ON(__REQ_F_LAST_BIT > 8 * sizeof_field(struct io_kiocb, flags));
4055
4056 BUILD_BUG_ON(sizeof(atomic_t) != sizeof(u32));
4057
4058 /* top 8bits are for internal use */
4059 BUILD_BUG_ON((IORING_URING_CMD_MASK & 0xff000000) != 0);
4060
4061 io_uring_optable_init();
4062
4063 /* imu->dir is u8 */
4064 BUILD_BUG_ON((IO_IMU_DEST | IO_IMU_SOURCE) > U8_MAX);
4065
4066 /*
4067 * Allow user copy in the per-command field, which starts after the
4068 * file in io_kiocb and until the opcode field. The openat2 handling
4069 * requires copying in user memory into the io_kiocb object in that
4070 * range, and HARDENED_USERCOPY will complain if we haven't
4071 * correctly annotated this range.
4072 */
4073 req_cachep = kmem_cache_create("io_kiocb", sizeof(struct io_kiocb), &kmem_args,
4074 SLAB_HWCACHE_ALIGN | SLAB_PANIC | SLAB_ACCOUNT |
4075 SLAB_TYPESAFE_BY_RCU);
4076
4077 iou_wq = alloc_workqueue("iou_exit", WQ_UNBOUND, 64);
4078 BUG_ON(!iou_wq);
4079
4080#ifdef CONFIG_SYSCTL
4081 register_sysctl_init("kernel", kernel_io_uring_disabled_table);
4082#endif
4083
4084 return 0;
4085};
4086__initcall(io_uring_init);
4087