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