xref: /linux/io_uring/io_uring.c (revision 91928e0d3cc29789f4483bffee5f36218f23942b)
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, &current->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