io_uring.c

// SPDX-License-Identifier: GPL-2.0
/*
 * Shared application/kernel submission and completion ring pairs, for
 * supporting fast/efficient IO.
 *
 * A note on the read/write ordering memory barriers that are matched between
 * the application and kernel side.
 *
 * After the application reads the CQ ring tail, it must use an
 * appropriate smp_rmb() to pair with the smp_wmb() the kernel uses
 * before writing the tail (using smp_load_acquire to read the tail will
 * do). It also needs a smp_mb() before updating CQ head (ordering the
 * entry load(s) with the head store), pairing with an implicit barrier
 * through a control-dependency in io_get_cqe (smp_store_release to
 * store head will do). Failure to do so could lead to reading invalid
 * CQ entries.
 *
 * Likewise, the application must use an appropriate smp_wmb() before
 * writing the SQ tail (ordering SQ entry stores with the tail store),
 * which pairs with smp_load_acquire in io_get_sqring (smp_store_release
 * to store the tail will do). And it needs a barrier ordering the SQ
 * head load before writing new SQ entries (smp_load_acquire to read
 * head will do).
 *
 * When using the SQ poll thread (IORING_SETUP_SQPOLL), the application
 * needs to check the SQ flags for IORING_SQ_NEED_WAKEUP *after*
 * updating the SQ tail; a full memory barrier smp_mb() is needed
 * between.
 *
 * Also see the examples in the liburing library:
 *
 *	git://git.kernel.org/pub/scm/linux/kernel/git/axboe/liburing.git
 *
 * io_uring also uses READ/WRITE_ONCE() for _any_ store or load that happens
 * from data shared between the kernel and application. This is done both
 * for ordering purposes, but also to ensure that once a value is loaded from
 * data that the application could potentially modify, it remains stable.
 *
 * Copyright (C) 2018-2019 Jens Axboe
 * Copyright (c) 2018-2019 Christoph Hellwig
 */
#include <linux/kernel.h>
#include <linux/errno.h>
#include <linux/syscalls.h>
#include <linux/refcount.h>
#include <linux/bits.h>

#include <linux/sched/signal.h>
#include <linux/fs.h>
#include <linux/mm.h>
#include <linux/percpu.h>
#include <linux/slab.h>
#include <linux/anon_inodes.h>
#include <linux/uaccess.h>
#include <linux/nospec.h>
#include <linux/task_work.h>
#include <linux/io_uring.h>
#include <linux/io_uring/cmd.h>
#include <linux/audit.h>
#include <linux/security.h>
#include <linux/jump_label.h>
#include <linux/kcov.h>

#define CREATE_TRACE_POINTS
#include <trace/events/io_uring.h>

#include <uapi/linux/io_uring.h>

#include "io-wq.h"

#include "filetable.h"
#include "io_uring.h"
#include "opdef.h"
#include "refs.h"
#include "tctx.h"
#include "register.h"
#include "sqpoll.h"
#include "fdinfo.h"
#include "kbuf.h"
#include "rsrc.h"
#include "cancel.h"
#include "net.h"
#include "notif.h"
#include "waitid.h"
#include "futex.h"
#include "napi.h"
#include "uring_cmd.h"
#include "msg_ring.h"
#include "memmap.h"
#include "zcrx.h"
#include "bpf-ops.h"

#include "timeout.h"
#include "poll.h"
#include "rw.h"
#include "alloc_cache.h"
#include "eventfd.h"
#include "wait.h"
#include "bpf_filter.h"
#include "loop.h"

#define SQE_COMMON_FLAGS (IOSQE_FIXED_FILE | IOSQE_IO_LINK | \
			  IOSQE_IO_HARDLINK | IOSQE_ASYNC)

#define IO_REQ_LINK_FLAGS (REQ_F_LINK | REQ_F_HARDLINK)

#define IO_REQ_CLEAN_FLAGS (REQ_F_BUFFER_SELECTED | REQ_F_NEED_CLEANUP | \
				REQ_F_INFLIGHT | REQ_F_CREDS | REQ_F_ASYNC_DATA)

#define IO_REQ_CLEAN_SLOW_FLAGS (REQ_F_REFCOUNT | IO_REQ_LINK_FLAGS | \
				 REQ_F_REISSUE | REQ_F_POLLED | \
				 IO_REQ_CLEAN_FLAGS)

#define IO_TCTX_REFS_CACHE_NR	(1U << 10)

#define IO_COMPL_BATCH			32
#define IO_REQ_ALLOC_BATCH		8

/* requests with any of those set should undergo io_disarm_next() */
#define IO_DISARM_MASK (REQ_F_ARM_LTIMEOUT | REQ_F_LINK_TIMEOUT | REQ_F_FAIL)

static void io_queue_sqe(struct io_kiocb *req, unsigned int extra_flags);
static void __io_req_caches_free(struct io_ring_ctx *ctx);

static __read_mostly DEFINE_STATIC_KEY_DEFERRED_FALSE(io_key_has_sqarray, HZ);

struct kmem_cache *req_cachep;
static struct workqueue_struct *iou_wq __ro_after_init;

static int __read_mostly sysctl_io_uring_disabled;
static int __read_mostly sysctl_io_uring_group = -1;

#ifdef CONFIG_SYSCTL
static const struct ctl_table kernel_io_uring_disabled_table[] = {
	{
		.procname	= "io_uring_disabled",
		.data		= &sysctl_io_uring_disabled,
		.maxlen		= sizeof(sysctl_io_uring_disabled),
		.mode		= 0644,
		.proc_handler	= proc_dointvec_minmax,
		.extra1		= SYSCTL_ZERO,
		.extra2		= SYSCTL_TWO,
	},
	{
		.procname	= "io_uring_group",
		.data		= &sysctl_io_uring_group,
		.maxlen		= sizeof(gid_t),
		.mode		= 0644,
		.proc_handler	= proc_dointvec,
	},
};
#endif

static void io_poison_cached_req(struct io_kiocb *req)
{
	req->ctx = IO_URING_PTR_POISON;
	req->tctx = IO_URING_PTR_POISON;
	req->file = IO_URING_PTR_POISON;
	req->creds = IO_URING_PTR_POISON;
	req->io_task_work.func = IO_URING_PTR_POISON;
	req->apoll = IO_URING_PTR_POISON;
}

void io_poison_req(struct io_kiocb *req)
{
	io_poison_cached_req(req);
	req->async_data = IO_URING_PTR_POISON;
	req->kbuf = IO_URING_PTR_POISON;
	req->comp_list.next = IO_URING_PTR_POISON;
	req->file_node = IO_URING_PTR_POISON;
	req->link = IO_URING_PTR_POISON;
}

static inline void req_fail_link_node(struct io_kiocb *req, int res)
{
	req_set_fail(req);
	io_req_set_res(req, res, 0);
}

static inline void io_req_add_to_cache(struct io_kiocb *req, struct io_ring_ctx *ctx)
{
	if (IS_ENABLED(CONFIG_KASAN))
		io_poison_cached_req(req);
	wq_stack_add_head(&req->comp_list, &ctx->submit_state.free_list);
}

static __cold void io_ring_ctx_ref_free(struct percpu_ref *ref)
{
	struct io_ring_ctx *ctx = container_of(ref, struct io_ring_ctx, refs);

	complete(&ctx->ref_comp);
}

static int io_alloc_hash_table(struct io_hash_table *table, unsigned bits)
{
	unsigned int hash_buckets;
	int i;

	do {
		hash_buckets = 1U << bits;
		table->hbs = kvmalloc_objs(table->hbs[0], hash_buckets,
					   GFP_KERNEL_ACCOUNT);
		if (table->hbs)
			break;
		if (bits == 1)
			return -ENOMEM;
		bits--;
	} while (1);

	table->hash_bits = bits;
	for (i = 0; i < hash_buckets; i++)
		INIT_HLIST_HEAD(&table->hbs[i].list);
	return 0;
}

static void io_free_alloc_caches(struct io_ring_ctx *ctx)
{
	io_alloc_cache_free(&ctx->apoll_cache, kfree);
	io_alloc_cache_free(&ctx->netmsg_cache, io_netmsg_cache_free);
	io_alloc_cache_free(&ctx->rw_cache, io_rw_cache_free);
	io_alloc_cache_free(&ctx->cmd_cache, io_cmd_cache_free);
	io_futex_cache_free(ctx);
	io_rsrc_cache_free(ctx);
}

static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
{
	struct io_ring_ctx *ctx;
	int hash_bits;
	bool ret;

	ctx = kzalloc_obj(*ctx);
	if (!ctx)
		return NULL;

	xa_init(&ctx->io_bl_xa);
	xa_init(&ctx->hpage_acct);

	/*
	 * Use 5 bits less than the max cq entries, that should give us around
	 * 32 entries per hash list if totally full and uniformly spread, but
	 * don't keep too many buckets to not overconsume memory.
	 */
	hash_bits = ilog2(p->cq_entries) - 5;
	hash_bits = clamp(hash_bits, 1, 8);
	if (io_alloc_hash_table(&ctx->cancel_table, hash_bits))
		goto err;
	if (percpu_ref_init(&ctx->refs, io_ring_ctx_ref_free,
			    0, GFP_KERNEL))
		goto err;

	ctx->flags = p->flags;
	ctx->hybrid_poll_time = LLONG_MAX;
	atomic_set(&ctx->cq_wait_nr, IO_CQ_WAKE_INIT);
	init_waitqueue_head(&ctx->sqo_sq_wait);
	INIT_LIST_HEAD(&ctx->sqd_list);
	INIT_LIST_HEAD(&ctx->cq_overflow_list);
	ret = io_alloc_cache_init(&ctx->apoll_cache, IO_POLL_ALLOC_CACHE_MAX,
			    sizeof(struct async_poll), 0);
	ret |= io_alloc_cache_init(&ctx->netmsg_cache, IO_ALLOC_CACHE_MAX,
			    sizeof(struct io_async_msghdr),
			    offsetof(struct io_async_msghdr, clear));
	ret |= io_alloc_cache_init(&ctx->rw_cache, IO_ALLOC_CACHE_MAX,
			    sizeof(struct io_async_rw),
			    offsetof(struct io_async_rw, clear));
	ret |= io_alloc_cache_init(&ctx->cmd_cache, IO_ALLOC_CACHE_MAX,
			    sizeof(struct io_async_cmd),
			    sizeof(struct io_async_cmd));
	ret |= io_futex_cache_init(ctx);
	ret |= io_rsrc_cache_init(ctx);
	if (ret)
		goto free_ref;
	init_completion(&ctx->ref_comp);
	xa_init_flags(&ctx->personalities, XA_FLAGS_ALLOC1);
	mutex_init(&ctx->uring_lock);
	init_waitqueue_head(&ctx->cq_wait);
	init_waitqueue_head(&ctx->poll_wq);
	spin_lock_init(&ctx->completion_lock);
	raw_spin_lock_init(&ctx->timeout_lock);
	INIT_LIST_HEAD(&ctx->iopoll_list);
	INIT_LIST_HEAD(&ctx->defer_list);
	INIT_LIST_HEAD(&ctx->timeout_list);
	INIT_LIST_HEAD(&ctx->ltimeout_list);
	mpscq_init(&ctx->work_list, &ctx->work_head);
	INIT_LIST_HEAD(&ctx->tctx_list);
	mutex_init(&ctx->tctx_lock);
	ctx->submit_state.free_list.next = NULL;
	INIT_HLIST_HEAD(&ctx->waitid_list);
	xa_init_flags(&ctx->zcrx_ctxs, XA_FLAGS_ALLOC);
#ifdef CONFIG_FUTEX
	INIT_HLIST_HEAD(&ctx->futex_list);
#endif
	INIT_WQ_LIST(&ctx->submit_state.compl_reqs);
	INIT_HLIST_HEAD(&ctx->cancelable_uring_cmd);
	io_napi_init(ctx);
	mutex_init(&ctx->mmap_lock);
	ctx->kcov_handle = kcov_common_handle();

	return ctx;

free_ref:
	percpu_ref_exit(&ctx->refs);
err:
	io_free_alloc_caches(ctx);
	kvfree(ctx->cancel_table.hbs);
	xa_destroy(&ctx->io_bl_xa);
	xa_destroy(&ctx->hpage_acct);
	kfree(ctx);
	return NULL;
}

static void io_clean_op(struct io_kiocb *req)
{
	if (unlikely(req->flags & REQ_F_BUFFER_SELECTED))
		io_kbuf_drop_legacy(req);

	if (req->flags & REQ_F_NEED_CLEANUP) {
		const struct io_cold_def *def = &io_cold_defs[req->opcode];

		if (def->cleanup)
			def->cleanup(req);
	}
	if (req->flags & REQ_F_INFLIGHT)
		atomic_dec(&req->tctx->inflight_tracked);
	if (req->flags & REQ_F_CREDS)
		put_cred(req->creds);
	if (req->flags & REQ_F_ASYNC_DATA) {
		kfree(req->async_data);
		req->async_data = NULL;
	}
	req->flags &= ~IO_REQ_CLEAN_FLAGS;
}

/*
 * Mark the request as inflight, so that file cancelation will find it.
 * Can be used if the file is an io_uring instance, or if the request itself
 * relies on ->mm being alive for the duration of the request.
 */
inline void io_req_track_inflight(struct io_kiocb *req)
{
	if (!(req->flags & REQ_F_INFLIGHT)) {
		req->flags |= REQ_F_INFLIGHT;
		atomic_inc(&req->tctx->inflight_tracked);
	}
}

static struct io_kiocb *__io_prep_linked_timeout(struct io_kiocb *req)
{
	if (WARN_ON_ONCE(!req->link))
		return NULL;

	req->flags &= ~REQ_F_ARM_LTIMEOUT;
	req->flags |= REQ_F_LINK_TIMEOUT;

	/* linked timeouts should have two refs once prep'ed */
	io_req_set_refcount(req);
	__io_req_set_refcount(req->link, 2);
	return req->link;
}

static void io_prep_async_work(struct io_kiocb *req)
{
	const struct io_issue_def *def = &io_issue_defs[req->opcode];

	if (!(req->flags & REQ_F_CREDS)) {
		req->flags |= REQ_F_CREDS;
		req->creds = get_current_cred();
	}

	req->work.list.next = NULL;
	atomic_set(&req->work.flags, 0);
	if (req->flags & REQ_F_FORCE_ASYNC)
		atomic_or(IO_WQ_WORK_CONCURRENT, &req->work.flags);

	if (req->file && !(req->flags & REQ_F_FIXED_FILE))
		req->flags |= io_file_get_flags(req->file);

	if (req->file && (req->flags & REQ_F_ISREG)) {
		bool should_hash = def->hash_reg_file;

		/* don't serialize this request if the fs doesn't need it */
		if (should_hash && (req->file->f_flags & O_DIRECT) &&
		    (req->file->f_op->fop_flags & FOP_DIO_PARALLEL_WRITE))
			should_hash = false;
		if (should_hash || (req->flags & REQ_F_IOPOLL))
			io_wq_hash_work(&req->work, file_inode(req->file));
	} else if (!req->file || !S_ISBLK(file_inode(req->file)->i_mode)) {
		if (def->unbound_nonreg_file)
			atomic_or(IO_WQ_WORK_UNBOUND, &req->work.flags);
	}
}

static void io_prep_async_link(struct io_kiocb *req)
{
	struct io_kiocb *cur;

	if (req->flags & REQ_F_LINK_TIMEOUT) {
		struct io_ring_ctx *ctx = req->ctx;

		raw_spin_lock_irq(&ctx->timeout_lock);
		io_for_each_link(cur, req)
			io_prep_async_work(cur);
		raw_spin_unlock_irq(&ctx->timeout_lock);
	} else {
		io_for_each_link(cur, req)
			io_prep_async_work(cur);
	}
}

void io_queue_iowq(struct io_kiocb *req)
{
	struct io_uring_task *tctx = req->tctx;

	BUG_ON(!tctx);

	if ((current->flags & PF_KTHREAD) || !tctx->io_wq) {
		io_req_task_queue_fail(req, -ECANCELED);
		return;
	}

	/* init ->work of the whole link before punting */
	io_prep_async_link(req);

	/*
	 * Not expected to happen, but if we do have a bug where this _can_
	 * happen, catch it here and ensure the request is marked as
	 * canceled. That will make io-wq go through the usual work cancel
	 * procedure rather than attempt to run this request (or create a new
	 * worker for it).
	 */
	if (WARN_ON_ONCE(!same_thread_group(tctx->task, current)))
		atomic_or(IO_WQ_WORK_CANCEL, &req->work.flags);

	trace_io_uring_queue_async_work(req, io_wq_is_hashed(&req->work));
	io_wq_enqueue(tctx->io_wq, &req->work);
}

unsigned io_linked_nr(struct io_kiocb *req)
{
	struct io_kiocb *tmp;
	unsigned nr = 0;

	io_for_each_link(tmp, req)
		nr++;
	return nr;
}

static __cold noinline void io_queue_deferred(struct io_ring_ctx *ctx)
{
	bool drain_seen = false, first = true;

	lockdep_assert_held(&ctx->uring_lock);
	__io_req_caches_free(ctx);

	while (!list_empty(&ctx->defer_list)) {
		struct io_defer_entry *de = list_first_entry(&ctx->defer_list,
						struct io_defer_entry, list);

		drain_seen |= de->req->flags & REQ_F_IO_DRAIN;
		if ((drain_seen || first) && ctx->nr_req_allocated != ctx->nr_drained)
			return;

		list_del_init(&de->list);
		ctx->nr_drained -= io_linked_nr(de->req);
		io_req_task_queue(de->req);
		kfree(de);
		first = false;
	}
}

void __io_commit_cqring_flush(struct io_ring_ctx *ctx)
{
	if (ctx->int_flags & IO_RING_F_POLL_ACTIVATED)
		io_poll_wq_wake(ctx);
	if (ctx->int_flags & IO_RING_F_OFF_TIMEOUT_USED)
		io_flush_timeouts(ctx);
	if (ctx->int_flags & IO_RING_F_HAS_EVFD)
		io_eventfd_signal(ctx, true, false);
}

static inline void __io_cq_lock(struct io_ring_ctx *ctx)
{
	if (!(ctx->int_flags & IO_RING_F_LOCKLESS_CQ))
		spin_lock(&ctx->completion_lock);
}

static inline void io_cq_lock(struct io_ring_ctx *ctx)
	__acquires(ctx->completion_lock)
{
	spin_lock(&ctx->completion_lock);
}

static inline void __io_cq_unlock_post(struct io_ring_ctx *ctx)
{
	io_commit_cqring(ctx);
	if (!(ctx->int_flags & IO_RING_F_TASK_COMPLETE)) {
		if (!(ctx->int_flags & IO_RING_F_LOCKLESS_CQ))
			spin_unlock(&ctx->completion_lock);
		/* IOPOLL rings only need to wake up if it's also SQPOLL */
		if (!(ctx->int_flags & IO_RING_F_SYSCALL_IOPOLL))
			io_cqring_wake(ctx);
	}
	io_commit_cqring_flush(ctx);
}

static void io_cq_unlock_post(struct io_ring_ctx *ctx)
	__releases(ctx->completion_lock)
{
	io_commit_cqring(ctx);
	spin_unlock(&ctx->completion_lock);
	io_cqring_wake(ctx);
	io_commit_cqring_flush(ctx);
}

static void __io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool dying)
{
	lockdep_assert_held(&ctx->uring_lock);

	/* don't abort if we're dying, entries must get freed */
	if (!dying && __io_cqring_events(ctx) == ctx->cq_entries)
		return;

	io_cq_lock(ctx);
	while (!list_empty(&ctx->cq_overflow_list)) {
		size_t cqe_size = sizeof(struct io_uring_cqe);
		struct io_uring_cqe *cqe;
		struct io_overflow_cqe *ocqe;
		bool is_cqe32 = false;

		ocqe = list_first_entry(&ctx->cq_overflow_list,
					struct io_overflow_cqe, list);
		if (ocqe->cqe.flags & IORING_CQE_F_32 ||
		    ctx->flags & IORING_SETUP_CQE32) {
			is_cqe32 = true;
			cqe_size <<= 1;
		}
		if (ctx->flags & IORING_SETUP_CQE32)
			is_cqe32 = false;

		if (!dying) {
			if (!io_get_cqe_overflow(ctx, &cqe, true, is_cqe32))
				break;
			memcpy(cqe, &ocqe->cqe, cqe_size);
		}
		list_del(&ocqe->list);
		kfree(ocqe);

		/*
		 * For silly syzbot cases that deliberately overflow by huge
		 * amounts, check if we need to resched and drop and
		 * reacquire the locks if so. Nothing real would ever hit this.
		 * Ideally we'd have a non-posting unlock for this, but hard
		 * to care for a non-real case.
		 */
		if (need_resched()) {
			ctx->cqe_sentinel = ctx->cqe_cached;
			io_cq_unlock_post(ctx);
			mutex_unlock(&ctx->uring_lock);
			cond_resched();
			mutex_lock(&ctx->uring_lock);
			io_cq_lock(ctx);
		}
	}

	if (list_empty(&ctx->cq_overflow_list)) {
		clear_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq);
		atomic_andnot(IORING_SQ_CQ_OVERFLOW, &ctx->rings->sq_flags);
	}
	io_cq_unlock_post(ctx);
}

static void io_cqring_overflow_kill(struct io_ring_ctx *ctx)
{
	if (ctx->rings)
		__io_cqring_overflow_flush(ctx, true);
}

void io_cqring_do_overflow_flush(struct io_ring_ctx *ctx)
{
	mutex_lock(&ctx->uring_lock);
	__io_cqring_overflow_flush(ctx, false);
	mutex_unlock(&ctx->uring_lock);
}

void io_cqring_overflow_flush_locked(struct io_ring_ctx *ctx)
{
	__io_cqring_overflow_flush(ctx, false);
}

/* must to be called somewhat shortly after putting a request */
static inline void io_put_task(struct io_kiocb *req)
{
	struct io_uring_task *tctx = req->tctx;

	if (likely(tctx->task == current)) {
		tctx->cached_refs++;
	} else {
		percpu_counter_sub(&tctx->inflight, 1);
		if (unlikely(atomic_read(&tctx->in_cancel)))
			wake_up(&tctx->wait);
		put_task_struct(tctx->task);
	}
}

void io_task_refs_refill(struct io_uring_task *tctx)
{
	unsigned int refill = -tctx->cached_refs + IO_TCTX_REFS_CACHE_NR;

	percpu_counter_add(&tctx->inflight, refill);
	refcount_add(refill, &current->usage);
	tctx->cached_refs += refill;
}

__cold void io_uring_drop_tctx_refs(struct task_struct *task)
{
	struct io_uring_task *tctx = task->io_uring;
	unsigned int refs = tctx->cached_refs;

	if (refs) {
		tctx->cached_refs = 0;
		percpu_counter_sub(&tctx->inflight, refs);
		put_task_struct_many(task, refs);
	}
}

static __cold bool io_cqring_add_overflow(struct io_ring_ctx *ctx,
					  struct io_overflow_cqe *ocqe)
{
	lockdep_assert_held(&ctx->completion_lock);

	if (!ocqe) {
		struct io_rings *r = ctx->rings;

		/*
		 * If we're in ring overflow flush mode, or in task cancel mode,
		 * or cannot allocate an overflow entry, then we need to drop it
		 * on the floor.
		 */
		WRITE_ONCE(r->cq_overflow, READ_ONCE(r->cq_overflow) + 1);
		set_bit(IO_CHECK_CQ_DROPPED_BIT, &ctx->check_cq);
		return false;
	}
	if (list_empty(&ctx->cq_overflow_list)) {
		set_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq);
		atomic_or(IORING_SQ_CQ_OVERFLOW, &ctx->rings->sq_flags);

	}
	list_add_tail(&ocqe->list, &ctx->cq_overflow_list);
	return true;
}

static struct io_overflow_cqe *io_alloc_ocqe(struct io_ring_ctx *ctx,
					     struct io_cqe *cqe,
					     struct io_big_cqe *big_cqe, gfp_t gfp)
{
	struct io_overflow_cqe *ocqe;
	size_t ocq_size = sizeof(struct io_overflow_cqe);
	bool is_cqe32 = false;

	if (cqe->flags & IORING_CQE_F_32 || ctx->flags & IORING_SETUP_CQE32) {
		is_cqe32 = true;
		ocq_size += sizeof(struct io_uring_cqe);
	}

	ocqe = kzalloc(ocq_size, gfp | __GFP_ACCOUNT);
	trace_io_uring_cqe_overflow(ctx, cqe->user_data, cqe->res, cqe->flags, ocqe);
	if (ocqe) {
		ocqe->cqe.user_data = cqe->user_data;
		ocqe->cqe.res = cqe->res;
		ocqe->cqe.flags = cqe->flags;
		if (is_cqe32 && big_cqe) {
			ocqe->cqe.big_cqe[0] = big_cqe->extra1;
			ocqe->cqe.big_cqe[1] = big_cqe->extra2;
		}
	}
	if (big_cqe)
		big_cqe->extra1 = big_cqe->extra2 = 0;
	return ocqe;
}

/*
 * Compute queued CQEs for free-space calculation, clamped to cq_entries.
 */
static unsigned int io_cqring_queued(struct io_ring_ctx *ctx)
{
	struct io_rings *rings = io_get_rings(ctx);
	int diff;

	diff = (int)(ctx->cached_cq_tail - READ_ONCE(rings->cq.head));
	if (diff >= 0)
		return min((unsigned int)diff, ctx->cq_entries);
	return 0;
}

/*
 * Fill an empty dummy CQE, in case alignment is off for posting a 32b CQE
 * because the ring is a single 16b entry away from wrapping.
 */
static bool io_fill_nop_cqe(struct io_ring_ctx *ctx, unsigned int off)
{
	if (io_cqring_queued(ctx) < ctx->cq_entries) {
		struct io_uring_cqe *cqe = &ctx->rings->cqes[off];

		cqe->user_data = 0;
		cqe->res = 0;
		cqe->flags = IORING_CQE_F_SKIP;
		ctx->cached_cq_tail++;
		return true;
	}
	return false;
}

/*
 * writes to the cq entry need to come after reading head; the
 * control dependency is enough as we're using WRITE_ONCE to
 * fill the cq entry
 */
bool io_cqe_cache_refill(struct io_ring_ctx *ctx, bool overflow, bool cqe32)
{
	struct io_rings *rings = ctx->rings;
	unsigned int off = ctx->cached_cq_tail & (ctx->cq_entries - 1);
	unsigned int free, len;

	/*
	 * Posting into the CQ when there are pending overflowed CQEs may break
	 * ordering guarantees, which will affect links, F_MORE users and more.
	 * Force overflow the completion.
	 */
	if (!overflow && (ctx->check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT)))
		return false;

	/*
	 * Post dummy CQE if a 32b CQE is needed and there's only room for a
	 * 16b CQE before the ring wraps.
	 */
	if (cqe32 && off + 1 == ctx->cq_entries) {
		if (!io_fill_nop_cqe(ctx, off))
			return false;
		off = 0;
	}

	free = ctx->cq_entries - io_cqring_queued(ctx);
	/* we need a contiguous range, limit based on the current array offset */
	len = min(free, ctx->cq_entries - off);
	if (len < (cqe32 + 1))
		return false;

	if (ctx->flags & IORING_SETUP_CQE32) {
		off <<= 1;
		len <<= 1;
	}

	ctx->cqe_cached = &rings->cqes[off];
	ctx->cqe_sentinel = ctx->cqe_cached + len;
	return true;
}

static bool io_fill_cqe_aux32(struct io_ring_ctx *ctx,
			      struct io_uring_cqe src_cqe[2])
{
	struct io_uring_cqe *cqe;

	if (WARN_ON_ONCE(!(ctx->flags & (IORING_SETUP_CQE32|IORING_SETUP_CQE_MIXED))))
		return false;
	if (unlikely(!io_get_cqe(ctx, &cqe, true)))
		return false;

	memcpy(cqe, src_cqe, 2 * sizeof(*cqe));
	trace_io_uring_complete(ctx, NULL, cqe);
	return true;
}

static bool io_fill_cqe_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res,
			      u32 cflags)
{
	bool cqe32 = cflags & IORING_CQE_F_32;
	struct io_uring_cqe *cqe;

	if (likely(io_get_cqe(ctx, &cqe, cqe32))) {
		WRITE_ONCE(cqe->user_data, user_data);
		WRITE_ONCE(cqe->res, res);
		WRITE_ONCE(cqe->flags, cflags);

		if (cqe32) {
			WRITE_ONCE(cqe->big_cqe[0], 0);
			WRITE_ONCE(cqe->big_cqe[1], 0);
		}

		trace_io_uring_complete(ctx, NULL, cqe);
		return true;
	}
	return false;
}

static inline struct io_cqe io_init_cqe(u64 user_data, s32 res, u32 cflags)
{
	return (struct io_cqe) { .user_data = user_data, .res = res, .flags = cflags };
}

static __cold void io_cqe_overflow(struct io_ring_ctx *ctx, struct io_cqe *cqe,
				   struct io_big_cqe *big_cqe)
{
	struct io_overflow_cqe *ocqe;

	ocqe = io_alloc_ocqe(ctx, cqe, big_cqe, GFP_KERNEL);
	spin_lock(&ctx->completion_lock);
	io_cqring_add_overflow(ctx, ocqe);
	spin_unlock(&ctx->completion_lock);
}

static __cold bool io_cqe_overflow_locked(struct io_ring_ctx *ctx,
					  struct io_cqe *cqe,
					  struct io_big_cqe *big_cqe)
{
	struct io_overflow_cqe *ocqe;

	ocqe = io_alloc_ocqe(ctx, cqe, big_cqe, GFP_NOWAIT);
	return io_cqring_add_overflow(ctx, ocqe);
}

bool io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
{
	bool filled;

	io_cq_lock(ctx);
	filled = io_fill_cqe_aux(ctx, user_data, res, cflags);
	if (unlikely(!filled)) {
		struct io_cqe cqe = io_init_cqe(user_data, res, cflags);

		filled = io_cqe_overflow_locked(ctx, &cqe, NULL);
	}
	io_cq_unlock_post(ctx);
	return filled;
}

/*
 * Must be called from inline task_work so we know a flush will happen later,
 * and obviously with ctx->uring_lock held (tw always has that).
 */
void io_add_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
{
	lockdep_assert_held(&ctx->uring_lock);
	lockdep_assert(ctx->int_flags & IO_RING_F_LOCKLESS_CQ);

	if (!io_fill_cqe_aux(ctx, user_data, res, cflags)) {
		struct io_cqe cqe = io_init_cqe(user_data, res, cflags);

		io_cqe_overflow(ctx, &cqe, NULL);
	}
	ctx->submit_state.cq_flush = true;
}

/*
 * A helper for multishot requests posting additional CQEs.
 * Should only be used from a task_work including IO_URING_F_MULTISHOT.
 */
bool io_req_post_cqe(struct io_kiocb *req, s32 res, u32 cflags)
{
	struct io_ring_ctx *ctx = req->ctx;
	bool posted;

	/*
	 * If multishot has already posted deferred completions, ensure that
	 * those are flushed first before posting this one. If not, CQEs
	 * could get reordered.
	 */
	if (!wq_list_empty(&ctx->submit_state.compl_reqs))
		__io_submit_flush_completions(ctx);

	lockdep_assert(!io_wq_current_is_worker());
	lockdep_assert_held(&ctx->uring_lock);

	if (!(ctx->int_flags & IO_RING_F_LOCKLESS_CQ)) {
		spin_lock(&ctx->completion_lock);
		posted = io_fill_cqe_aux(ctx, req->cqe.user_data, res, cflags);
		spin_unlock(&ctx->completion_lock);
	} else {
		posted = io_fill_cqe_aux(ctx, req->cqe.user_data, res, cflags);
	}

	ctx->submit_state.cq_flush = true;
	return posted;
}

/*
 * A helper for multishot requests posting additional CQEs.
 * Should only be used from a task_work including IO_URING_F_MULTISHOT.
 */
bool io_req_post_cqe32(struct io_kiocb *req, struct io_uring_cqe cqe[2])
{
	struct io_ring_ctx *ctx = req->ctx;
	bool posted;

	lockdep_assert(!io_wq_current_is_worker());
	lockdep_assert_held(&ctx->uring_lock);

	cqe[0].user_data = req->cqe.user_data;
	if (!(ctx->int_flags & IO_RING_F_LOCKLESS_CQ)) {
		spin_lock(&ctx->completion_lock);
		posted = io_fill_cqe_aux32(ctx, cqe);
		spin_unlock(&ctx->completion_lock);
	} else {
		posted = io_fill_cqe_aux32(ctx, cqe);
	}

	ctx->submit_state.cq_flush = true;
	return posted;
}

static void io_req_complete_post(struct io_kiocb *req, unsigned issue_flags)
{
	struct io_ring_ctx *ctx = req->ctx;
	bool completed = true;

	/*
	 * All execution paths but io-wq use the deferred completions by
	 * passing IO_URING_F_COMPLETE_DEFER and thus should not end up here.
	 */
	if (WARN_ON_ONCE(!(issue_flags & IO_URING_F_IOWQ)))
		return;

	/*
	 * Handle special CQ sync cases via task_work. DEFER_TASKRUN requires
	 * the submitter task context, IOPOLL protects with uring_lock.
	 */
	if ((ctx->int_flags & IO_RING_F_LOCKLESS_CQ) || (req->flags & REQ_F_REISSUE)) {
defer_complete:
		req->io_task_work.func = io_req_task_complete;
		io_req_task_work_add(req);
		return;
	}

	io_cq_lock(ctx);
	if (!(req->flags & REQ_F_CQE_SKIP))
		completed = io_fill_cqe_req(ctx, req);
	io_cq_unlock_post(ctx);

	if (!completed)
		goto defer_complete;

	/*
	 * We don't free the request here because we know it's called from
	 * io-wq only, which holds a reference, so it cannot be the last put.
	 */
	req_ref_put(req);
}

void io_req_defer_failed(struct io_kiocb *req, s32 res)
	__must_hold(&ctx->uring_lock)
{
	const struct io_cold_def *def = &io_cold_defs[req->opcode];

	lockdep_assert_held(&req->ctx->uring_lock);

	req_set_fail(req);
	io_req_set_res(req, res, io_put_kbuf(req, res, NULL));
	if (def->fail)
		def->fail(req);
	io_req_complete_defer(req);
}

/*
 * A request might get retired back into the request caches even before opcode
 * handlers and io_issue_sqe() are done with it, e.g. inline completion path.
 * Because of that, io_alloc_req() should be called only under ->uring_lock
 * and with extra caution to not get a request that is still worked on.
 */
__cold bool __io_alloc_req_refill(struct io_ring_ctx *ctx)
	__must_hold(&ctx->uring_lock)
{
	gfp_t gfp = GFP_KERNEL | __GFP_NOWARN | __GFP_ZERO;
	void *reqs[IO_REQ_ALLOC_BATCH];
	int nr_reqs = ARRAY_SIZE(reqs);

	/*
	 * Bulk alloc is all-or-nothing. If we fail to get a batch, retry a
	 * single allocation to be on the safe side.
	 */
	if (!kmem_cache_alloc_bulk(req_cachep, gfp, nr_reqs, reqs)) {
		reqs[0] = kmem_cache_alloc(req_cachep, gfp);
		if (!reqs[0])
			return false;
		nr_reqs = 1;
	}

	percpu_ref_get_many(&ctx->refs, nr_reqs);
	ctx->nr_req_allocated += nr_reqs;

	while (nr_reqs--)
		io_req_add_to_cache(reqs[nr_reqs], ctx);
	return true;
}

__cold void io_free_req(struct io_kiocb *req)
{
	/* refs were already put, restore them for io_req_task_complete() */
	req->flags &= ~REQ_F_REFCOUNT;
	/* we only want to free it, don't post CQEs */
	req->flags |= REQ_F_CQE_SKIP;
	req->io_task_work.func = io_req_task_complete;