io_uring.c

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