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