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