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