| // 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.dk/liburing |
| * |
| * 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/init.h> |
| #include <linux/errno.h> |
| #include <linux/syscalls.h> |
| #include <net/compat.h> |
| #include <linux/refcount.h> |
| #include <linux/uio.h> |
| #include <linux/bits.h> |
| |
| #include <linux/sched/signal.h> |
| #include <linux/fs.h> |
| #include <linux/file.h> |
| #include <linux/fdtable.h> |
| #include <linux/mm.h> |
| #include <linux/mman.h> |
| #include <linux/percpu.h> |
| #include <linux/slab.h> |
| #include <linux/bvec.h> |
| #include <linux/net.h> |
| #include <net/sock.h> |
| #include <linux/anon_inodes.h> |
| #include <linux/sched/mm.h> |
| #include <linux/uaccess.h> |
| #include <linux/nospec.h> |
| #include <linux/fsnotify.h> |
| #include <linux/fadvise.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 <asm/shmparam.h> |
| |
| #define CREATE_TRACE_POINTS |
| #include <trace/events/io_uring.h> |
| |
| #include <uapi/linux/io_uring.h> |
| |
| #include "io-wq.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 "timeout.h" |
| #include "poll.h" |
| #include "rw.h" |
| #include "alloc_cache.h" |
| #include "eventfd.h" |
| |
| #define IORING_MAX_ENTRIES 32768 |
| #define IORING_MAX_CQ_ENTRIES (2 * IORING_MAX_ENTRIES) |
| |
| #define SQE_COMMON_FLAGS (IOSQE_FIXED_FILE | IOSQE_IO_LINK | \ |
| IOSQE_IO_HARDLINK | IOSQE_ASYNC) |
| |
| #define SQE_VALID_FLAGS (SQE_COMMON_FLAGS | IOSQE_BUFFER_SELECT | \ |
| IOSQE_IO_DRAIN | IOSQE_CQE_SKIP_SUCCESS) |
| |
| #define IO_REQ_CLEAN_FLAGS (REQ_F_BUFFER_SELECTED | REQ_F_NEED_CLEANUP | \ |
| REQ_F_POLLED | REQ_F_INFLIGHT | REQ_F_CREDS | \ |
| REQ_F_ASYNC_DATA) |
| |
| #define IO_REQ_CLEAN_SLOW_FLAGS (REQ_F_REFCOUNT | REQ_F_LINK | REQ_F_HARDLINK |\ |
| IO_REQ_CLEAN_FLAGS) |
| |
| #define IO_TCTX_REFS_CACHE_NR (1U << 10) |
| |
| #define IO_COMPL_BATCH 32 |
| #define IO_REQ_ALLOC_BATCH 8 |
| |
| struct io_defer_entry { |
| struct list_head list; |
| struct io_kiocb *req; |
| u32 seq; |
| }; |
| |
| /* 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) |
| #define IO_REQ_LINK_FLAGS (REQ_F_LINK | REQ_F_HARDLINK) |
| |
| /* |
| * No waiters. It's larger than any valid value of the tw counter |
| * so that tests against ->cq_wait_nr would fail and skip wake_up(). |
| */ |
| #define IO_CQ_WAKE_INIT (-1U) |
| /* Forced wake up if there is a waiter regardless of ->cq_wait_nr */ |
| #define IO_CQ_WAKE_FORCE (IO_CQ_WAKE_INIT >> 1) |
| |
| static bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx, |
| struct task_struct *task, |
| bool cancel_all); |
| |
| static void io_queue_sqe(struct io_kiocb *req); |
| |
| 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 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 inline unsigned int __io_cqring_events(struct io_ring_ctx *ctx) |
| { |
| return ctx->cached_cq_tail - READ_ONCE(ctx->rings->cq.head); |
| } |
| |
| static inline unsigned int __io_cqring_events_user(struct io_ring_ctx *ctx) |
| { |
| return READ_ONCE(ctx->rings->cq.tail) - READ_ONCE(ctx->rings->cq.head); |
| } |
| |
| static bool io_match_linked(struct io_kiocb *head) |
| { |
| struct io_kiocb *req; |
| |
| io_for_each_link(req, head) { |
| if (req->flags & REQ_F_INFLIGHT) |
| return true; |
| } |
| return false; |
| } |
| |
| /* |
| * As io_match_task() but protected against racing with linked timeouts. |
| * User must not hold timeout_lock. |
| */ |
| bool io_match_task_safe(struct io_kiocb *head, struct task_struct *task, |
| bool cancel_all) |
| { |
| bool matched; |
| |
| if (task && head->task != task) |
| return false; |
| if (cancel_all) |
| return true; |
| |
| if (head->flags & REQ_F_LINK_TIMEOUT) { |
| struct io_ring_ctx *ctx = head->ctx; |
| |
| /* protect against races with linked timeouts */ |
| spin_lock_irq(&ctx->timeout_lock); |
| matched = io_match_linked(head); |
| spin_unlock_irq(&ctx->timeout_lock); |
| } else { |
| matched = io_match_linked(head); |
| } |
| return matched; |
| } |
| |
| 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) |
| { |
| 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 __cold void io_fallback_req_func(struct work_struct *work) |
| { |
| struct io_ring_ctx *ctx = container_of(work, struct io_ring_ctx, |
| fallback_work.work); |
| struct llist_node *node = llist_del_all(&ctx->fallback_llist); |
| struct io_kiocb *req, *tmp; |
| struct io_tw_state ts = {}; |
| |
| percpu_ref_get(&ctx->refs); |
| mutex_lock(&ctx->uring_lock); |
| llist_for_each_entry_safe(req, tmp, node, io_task_work.node) |
| req->io_task_work.func(req, &ts); |
| io_submit_flush_completions(ctx); |
| mutex_unlock(&ctx->uring_lock); |
| percpu_ref_put(&ctx->refs); |
| } |
| |
| static int io_alloc_hash_table(struct io_hash_table *table, unsigned bits) |
| { |
| unsigned hash_buckets = 1U << bits; |
| size_t hash_size = hash_buckets * sizeof(table->hbs[0]); |
| |
| table->hbs = kmalloc(hash_size, GFP_KERNEL); |
| if (!table->hbs) |
| return -ENOMEM; |
| |
| table->hash_bits = bits; |
| init_hash_table(table, hash_buckets); |
| return 0; |
| } |
| |
| 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(sizeof(*ctx), GFP_KERNEL); |
| if (!ctx) |
| return NULL; |
| |
| xa_init(&ctx->io_bl_xa); |
| |
| /* |
| * 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 (io_alloc_hash_table(&ctx->cancel_table_locked, hash_bits)) |
| goto err; |
| if (percpu_ref_init(&ctx->refs, io_ring_ctx_ref_free, |
| 0, GFP_KERNEL)) |
| goto err; |
| |
| ctx->flags = p->flags; |
| 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); |
| INIT_LIST_HEAD(&ctx->io_buffers_cache); |
| ret = io_alloc_cache_init(&ctx->rsrc_node_cache, IO_NODE_ALLOC_CACHE_MAX, |
| sizeof(struct io_rsrc_node)); |
| ret |= io_alloc_cache_init(&ctx->apoll_cache, IO_POLL_ALLOC_CACHE_MAX, |
| sizeof(struct async_poll)); |
| ret |= io_alloc_cache_init(&ctx->netmsg_cache, IO_ALLOC_CACHE_MAX, |
| sizeof(struct io_async_msghdr)); |
| ret |= io_alloc_cache_init(&ctx->rw_cache, IO_ALLOC_CACHE_MAX, |
| sizeof(struct io_async_rw)); |
| ret |= io_alloc_cache_init(&ctx->uring_cache, IO_ALLOC_CACHE_MAX, |
| sizeof(struct uring_cache)); |
| spin_lock_init(&ctx->msg_lock); |
| ret |= io_alloc_cache_init(&ctx->msg_cache, IO_ALLOC_CACHE_MAX, |
| sizeof(struct io_kiocb)); |
| ret |= io_futex_cache_init(ctx); |
| if (ret) |
| goto err; |
| 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); |
| init_waitqueue_head(&ctx->rsrc_quiesce_wq); |
| spin_lock_init(&ctx->completion_lock); |
| spin_lock_init(&ctx->timeout_lock); |
| INIT_WQ_LIST(&ctx->iopoll_list); |
| INIT_LIST_HEAD(&ctx->io_buffers_comp); |
| INIT_LIST_HEAD(&ctx->defer_list); |
| INIT_LIST_HEAD(&ctx->timeout_list); |
| INIT_LIST_HEAD(&ctx->ltimeout_list); |
| INIT_LIST_HEAD(&ctx->rsrc_ref_list); |
| init_llist_head(&ctx->work_llist); |
| INIT_LIST_HEAD(&ctx->tctx_list); |
| ctx->submit_state.free_list.next = NULL; |
| INIT_HLIST_HEAD(&ctx->waitid_list); |
| #ifdef CONFIG_FUTEX |
| INIT_HLIST_HEAD(&ctx->futex_list); |
| #endif |
| INIT_DELAYED_WORK(&ctx->fallback_work, io_fallback_req_func); |
| INIT_WQ_LIST(&ctx->submit_state.compl_reqs); |
| INIT_HLIST_HEAD(&ctx->cancelable_uring_cmd); |
| io_napi_init(ctx); |
| |
| return ctx; |
| err: |
| io_alloc_cache_free(&ctx->rsrc_node_cache, kfree); |
| 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->uring_cache, kfree); |
| io_alloc_cache_free(&ctx->msg_cache, io_msg_cache_free); |
| io_futex_cache_free(ctx); |
| kfree(ctx->cancel_table.hbs); |
| kfree(ctx->cancel_table_locked.hbs); |
| xa_destroy(&ctx->io_bl_xa); |
| kfree(ctx); |
| return NULL; |
| } |
| |
| static void io_account_cq_overflow(struct io_ring_ctx *ctx) |
| { |
| struct io_rings *r = ctx->rings; |
| |
| WRITE_ONCE(r->cq_overflow, READ_ONCE(r->cq_overflow) + 1); |
| ctx->cq_extra--; |
| } |
| |
| static bool req_need_defer(struct io_kiocb *req, u32 seq) |
| { |
| if (unlikely(req->flags & REQ_F_IO_DRAIN)) { |
| struct io_ring_ctx *ctx = req->ctx; |
| |
| return seq + READ_ONCE(ctx->cq_extra) != ctx->cached_cq_tail; |
| } |
| |
| return false; |
| } |
| |
| static void io_clean_op(struct io_kiocb *req) |
| { |
| if (req->flags & REQ_F_BUFFER_SELECTED) { |
| spin_lock(&req->ctx->completion_lock); |
| io_kbuf_drop(req); |
| spin_unlock(&req->ctx->completion_lock); |
| } |
| |
| 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_POLLED) && req->apoll) { |
| kfree(req->apoll->double_poll); |
| kfree(req->apoll); |
| req->apoll = NULL; |
| } |
| if (req->flags & REQ_F_INFLIGHT) { |
| struct io_uring_task *tctx = req->task->io_uring; |
| |
| atomic_dec(&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; |
| } |
| |
| static inline void io_req_track_inflight(struct io_kiocb *req) |
| { |
| if (!(req->flags & REQ_F_INFLIGHT)) { |
| req->flags |= REQ_F_INFLIGHT; |
| atomic_inc(&req->task->io_uring->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 inline struct io_kiocb *io_prep_linked_timeout(struct io_kiocb *req) |
| { |
| if (likely(!(req->flags & REQ_F_ARM_LTIMEOUT))) |
| return NULL; |
| return __io_prep_linked_timeout(req); |
| } |
| |
| static noinline void __io_arm_ltimeout(struct io_kiocb *req) |
| { |
| io_queue_linked_timeout(__io_prep_linked_timeout(req)); |
| } |
| |
| static inline void io_arm_ltimeout(struct io_kiocb *req) |
| { |
| if (unlikely(req->flags & REQ_F_ARM_LTIMEOUT)) |
| __io_arm_ltimeout(req); |
| } |
| |
| static void io_prep_async_work(struct io_kiocb *req) |
| { |
| const struct io_issue_def *def = &io_issue_defs[req->opcode]; |
| struct io_ring_ctx *ctx = req->ctx; |
| |
| 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 || (ctx->flags & IORING_SETUP_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; |
| |
| spin_lock_irq(&ctx->timeout_lock); |
| io_for_each_link(cur, req) |
| io_prep_async_work(cur); |
| spin_unlock_irq(&ctx->timeout_lock); |
| } else { |
| io_for_each_link(cur, req) |
| io_prep_async_work(cur); |
| } |
| } |
| |
| static void io_queue_iowq(struct io_kiocb *req) |
| { |
| struct io_kiocb *link = io_prep_linked_timeout(req); |
| struct io_uring_task *tctx = req->task->io_uring; |
| |
| BUG_ON(!tctx); |
| BUG_ON(!tctx->io_wq); |
| |
| /* 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(req->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); |
| if (link) |
| io_queue_linked_timeout(link); |
| } |
| |
| static __cold void io_queue_deferred(struct io_ring_ctx *ctx) |
| { |
| while (!list_empty(&ctx->defer_list)) { |
| struct io_defer_entry *de = list_first_entry(&ctx->defer_list, |
| struct io_defer_entry, list); |
| |
| if (req_need_defer(de->req, de->seq)) |
| break; |
| list_del_init(&de->list); |
| io_req_task_queue(de->req); |
| kfree(de); |
| } |
| } |
| |
| void __io_commit_cqring_flush(struct io_ring_ctx *ctx) |
| { |
| if (ctx->poll_activated) |
| io_poll_wq_wake(ctx); |
| if (ctx->off_timeout_used) |
| io_flush_timeouts(ctx); |
| if (ctx->drain_active) { |
| spin_lock(&ctx->completion_lock); |
| io_queue_deferred(ctx); |
| spin_unlock(&ctx->completion_lock); |
| } |
| if (ctx->has_evfd) |
| io_eventfd_flush_signal(ctx); |
| } |
| |
| static inline void __io_cq_lock(struct io_ring_ctx *ctx) |
| { |
| if (!ctx->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->task_complete) { |
| if (!ctx->lockless_cq) |
| spin_unlock(&ctx->completion_lock); |
| /* IOPOLL rings only need to wake up if it's also SQPOLL */ |
| if (!ctx->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) |
| { |
| size_t cqe_size = sizeof(struct io_uring_cqe); |
| |
| 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; |
| |
| if (ctx->flags & IORING_SETUP_CQE32) |
| cqe_size <<= 1; |
| |
| io_cq_lock(ctx); |
| while (!list_empty(&ctx->cq_overflow_list)) { |
| struct io_uring_cqe *cqe; |
| struct io_overflow_cqe *ocqe; |
| |
| ocqe = list_first_entry(&ctx->cq_overflow_list, |
| struct io_overflow_cqe, list); |
| |
| if (!dying) { |
| if (!io_get_cqe_overflow(ctx, &cqe, true)) |
| break; |
| memcpy(cqe, &ocqe->cqe, cqe_size); |
| } |
| list_del(&ocqe->list); |
| kfree(ocqe); |
| } |
| |
| 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); |
| } |
| |
| static 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); |
| } |
| |
| /* can be called by any task */ |
| static void io_put_task_remote(struct task_struct *task) |
| { |
| struct io_uring_task *tctx = task->io_uring; |
| |
| percpu_counter_sub(&tctx->inflight, 1); |
| if (unlikely(atomic_read(&tctx->in_cancel))) |
| wake_up(&tctx->wait); |
| put_task_struct(task); |
| } |
| |
| /* used by a task to put its own references */ |
| static void io_put_task_local(struct task_struct *task) |
| { |
| task->io_uring->cached_refs++; |
| } |
| |
| /* must to be called somewhat shortly after putting a request */ |
| static inline void io_put_task(struct task_struct *task) |
| { |
| if (likely(task == current)) |
| io_put_task_local(task); |
| else |
| io_put_task_remote(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, ¤t->usage); |
| tctx->cached_refs += refill; |
| } |
| |
| static __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 bool io_cqring_event_overflow(struct io_ring_ctx *ctx, u64 user_data, |
| s32 res, u32 cflags, u64 extra1, u64 extra2) |
| { |
| struct io_overflow_cqe *ocqe; |
| size_t ocq_size = sizeof(struct io_overflow_cqe); |
| bool is_cqe32 = (ctx->flags & IORING_SETUP_CQE32); |
| |
| lockdep_assert_held(&ctx->completion_lock); |
| |
| if (is_cqe32) |
| ocq_size += sizeof(struct io_uring_cqe); |
| |
| ocqe = kmalloc(ocq_size, GFP_ATOMIC | __GFP_ACCOUNT); |
| trace_io_uring_cqe_overflow(ctx, user_data, res, cflags, ocqe); |
| if (!ocqe) { |
| /* |
| * 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. |
| */ |
| io_account_cq_overflow(ctx); |
| 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); |
| |
| } |
| ocqe->cqe.user_data = user_data; |
| ocqe->cqe.res = res; |
| ocqe->cqe.flags = cflags; |
| if (is_cqe32) { |
| ocqe->cqe.big_cqe[0] = extra1; |
| ocqe->cqe.big_cqe[1] = extra2; |
| } |
| list_add_tail(&ocqe->list, &ctx->cq_overflow_list); |
| return true; |
| } |
| |
| static void io_req_cqe_overflow(struct io_kiocb *req) |
| { |
| io_cqring_event_overflow(req->ctx, req->cqe.user_data, |
| req->cqe.res, req->cqe.flags, |
| req->big_cqe.extra1, req->big_cqe.extra2); |
| memset(&req->big_cqe, 0, sizeof(req->big_cqe)); |
| } |
| |
| /* |
| * 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) |
| { |
| struct io_rings *rings = ctx->rings; |
| unsigned int off = ctx->cached_cq_tail & (ctx->cq_entries - 1); |
| unsigned int free, queued, 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; |
| |
| /* userspace may cheat modifying the tail, be safe and do min */ |
| queued = min(__io_cqring_events(ctx), ctx->cq_entries); |
| free = ctx->cq_entries - queued; |
| /* we need a contiguous range, limit based on the current array offset */ |
| len = min(free, ctx->cq_entries - off); |
| if (!len) |
| 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_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res, |
| u32 cflags) |
| { |
| struct io_uring_cqe *cqe; |
| |
| ctx->cq_extra++; |
| |
| /* |
| * If we can't get a cq entry, userspace overflowed the |
| * submission (by quite a lot). Increment the overflow count in |
| * the ring. |
| */ |
| if (likely(io_get_cqe(ctx, &cqe))) { |
| trace_io_uring_complete(ctx, NULL, user_data, res, cflags, 0, 0); |
| |
| WRITE_ONCE(cqe->user_data, user_data); |
| WRITE_ONCE(cqe->res, res); |
| WRITE_ONCE(cqe->flags, cflags); |
| |
| if (ctx->flags & IORING_SETUP_CQE32) { |
| WRITE_ONCE(cqe->big_cqe[0], 0); |
| WRITE_ONCE(cqe->big_cqe[1], 0); |
| } |
| return true; |
| } |
| return false; |
| } |
| |
| static bool __io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, |
| u32 cflags) |
| { |
| bool filled; |
| |
| filled = io_fill_cqe_aux(ctx, user_data, res, cflags); |
| if (!filled) |
| filled = io_cqring_event_overflow(ctx, user_data, res, cflags, 0, 0); |
| |
| return filled; |
| } |
| |
| 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_post_aux_cqe(ctx, user_data, res, cflags); |
| io_cq_unlock_post(ctx); |
| return filled; |
| } |
| |
| /* |
| * Must be called from inline task_work so we now 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) |
| { |
| if (!io_fill_cqe_aux(ctx, user_data, res, cflags)) { |
| spin_lock(&ctx->completion_lock); |
| io_cqring_event_overflow(ctx, user_data, res, cflags, 0, 0); |
| spin_unlock(&ctx->completion_lock); |
| } |
| 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; |
| |
| lockdep_assert(!io_wq_current_is_worker()); |
| lockdep_assert_held(&ctx->uring_lock); |
| |
| __io_cq_lock(ctx); |
| posted = io_fill_cqe_aux(ctx, req->cqe.user_data, res, cflags); |
| ctx->submit_state.cq_flush = true; |
| __io_cq_unlock_post(ctx); |
| return posted; |
| } |
| |
| static void io_req_complete_post(struct io_kiocb *req, unsigned issue_flags) |
| { |
| struct io_ring_ctx *ctx = req->ctx; |
| |
| /* |
| * 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->task_complete || (ctx->flags & IORING_SETUP_IOPOLL)) { |
| 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)) { |
| if (!io_fill_cqe_req(ctx, req)) |
| io_req_cqe_overflow(req); |
| } |
| io_cq_unlock_post(ctx); |
| |
| /* |
| * 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, IO_URING_F_UNLOCKED)); |
| if (def->fail) |
| def->fail(req); |
| io_req_complete_defer(req); |
| } |
| |
| /* |
| * Don't initialise the fields below on every allocation, but do that in |
| * advance and keep them valid across allocations. |
| */ |
| static void io_preinit_req(struct io_kiocb *req, struct io_ring_ctx *ctx) |
| { |
| req->ctx = ctx; |
| req->link = NULL; |
| req->async_data = NULL; |
| /* not necessary, but safer to zero */ |
| memset(&req->cqe, 0, sizeof(req->cqe)); |
| memset(&req->big_cqe, 0, sizeof(req->big_cqe)); |
| } |
| |
| /* |
| * 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; |
| void *reqs[IO_REQ_ALLOC_BATCH]; |
| int ret; |
| |
| ret = kmem_cache_alloc_bulk(req_cachep, gfp, ARRAY_SIZE(reqs), reqs); |
| |
| /* |
| * Bulk alloc is all-or-nothing. If we fail to get a batch, |
| * retry single alloc to be on the safe side. |
| */ |
| if (unlikely(ret <= 0)) { |
| reqs[0] = kmem_cache_alloc(req_cachep, gfp); |
| if (!reqs[0]) |
| return false; |
| ret = 1; |
| } |
| |
| percpu_ref_get_many(&ctx->refs, ret); |
| while (ret--) { |
| struct io_kiocb *req = reqs[ret]; |
| |
| io_preinit_req(req, ctx); |
| io_req_add_to_cache(req, 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; |
| io_req_task_work_add(req); |
| } |
| |
| static void __io_req_find_next_prep(struct io_kiocb *req) |
| { |
| struct io_ring_ctx *ctx = req->ctx; |
| |
| spin_lock(&ctx->completion_lock); |
| io_disarm_next(req); |
| spin_unlock(&ctx->completion_lock); |
| } |
| |
| static inline struct io_kiocb *io_req_find_next(struct io_kiocb *req) |
| { |
| struct io_kiocb *nxt; |
| |
| /* |
| * If LINK is set, we have dependent requests in this chain. If we |
| * didn't fail this request, queue the first one up, moving any other |
| * dependencies to the next request. In case of failure, fail the rest |
| * of the chain. |
| */ |
| if (unlikely(req->flags & IO_DISARM_MASK)) |
| __io_req_find_next_prep(req); |
| nxt = req->link; |
| req->link = NULL; |
| return nxt; |
| } |
| |
| static void ctx_flush_and_put(struct io_ring_ctx *ctx, struct io_tw_state *ts) |
| { |
| if (!ctx) |
| return; |
| if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) |
| atomic_andnot(IORING_SQ_TASKRUN, &ctx->rings->sq_flags); |
| |
| io_submit_flush_completions(ctx); |
| mutex_unlock(&ctx->uring_lock); |
| percpu_ref_put(&ctx->refs); |
| } |
| |
| /* |
| * Run queued task_work, returning the number of entries processed in *count. |
| * If more entries than max_entries are available, stop processing once this |
| * is reached and return the rest of the list. |
| */ |
| struct llist_node *io_handle_tw_list(struct llist_node *node, |
| unsigned int *count, |
| unsigned int max_entries) |
| { |
| struct io_ring_ctx *ctx = NULL; |
| struct io_tw_state ts = { }; |
| |
| do { |
| struct llist_node *next = node->next; |
| struct io_kiocb *req = container_of(node, struct io_kiocb, |
| io_task_work.node); |
| |
| if (req->ctx != ctx) { |
| ctx_flush_and_put(ctx, &ts); |
| ctx = req->ctx; |
| mutex_lock(&ctx->uring_lock); |
| percpu_ref_get(&ctx->refs); |
| } |
| INDIRECT_CALL_2(req->io_task_work.func, |
| io_poll_task_func, io_req_rw_complete, |
| req, &ts); |
| node = next; |
| (*count)++; |
| if (unlikely(need_resched())) { |
| ctx_flush_and_put(ctx, &ts); |
| ctx = NULL; |
| cond_resched(); |
| } |
| } while (node && *count < max_entries); |
| |
| ctx_flush_and_put(ctx, &ts); |
| return node; |
| } |
| |
| /** |
| * io_llist_xchg - swap all entries in a lock-less list |
| * @head: the head of lock-less list to delete all entries |
| * @new: new entry as the head of the list |
| * |
| * If list is empty, return NULL, otherwise, return the pointer to the first entry. |
| * The order of entries returned is from the newest to the oldest added one. |
| */ |
| static inline struct llist_node *io_llist_xchg(struct llist_head *head, |
| struct llist_node *new) |
| { |
| return xchg(&head->first, new); |
| } |
| |
| static __cold void io_fallback_tw(struct io_uring_task *tctx, bool sync) |
| { |
| struct llist_node *node = llist_del_all(&tctx->task_list); |
| struct io_ring_ctx *last_ctx = NULL; |
| struct io_kiocb *req; |
| |
| while (node) { |
| req = container_of(node, struct io_kiocb, io_task_work.node); |
| node = node->next; |
| if (sync && last_ctx != req->ctx) { |
| if (last_ctx) { |
| flush_delayed_work(&last_ctx->fallback_work); |
| percpu_ref_put(&last_ctx->refs); |
| } |
| last_ctx = req->ctx; |
| percpu_ref_get(&last_ctx->refs); |
| } |
| if (llist_add(&req->io_task_work.node, |
| &req->ctx->fallback_llist)) |
| schedule_delayed_work(&req->ctx->fallback_work, 1); |
| } |
| |
| if (last_ctx) { |
| flush_delayed_work(&last_ctx->fallback_work); |
| percpu_ref_put(&last_ctx->refs); |
| } |
| } |
| |
| struct llist_node *tctx_task_work_run(struct io_uring_task *tctx, |
| unsigned int max_entries, |
| unsigned int *count) |
| { |
| struct llist_node *node; |
| |
| if (unlikely(current->flags & PF_EXITING)) { |
| io_fallback_tw(tctx, true); |
| return NULL; |
| } |
| |
| node = llist_del_all(&tctx->task_list); |
| if (node) { |
| node = llist_reverse_order(node); |
| node = io_handle_tw_list(node, count, max_entries); |
| } |
| |
| /* relaxed read is enough as only the task itself sets ->in_cancel */ |
| if (unlikely(atomic_read(&tctx->in_cancel))) |
| io_uring_drop_tctx_refs(current); |
| |
| trace_io_uring_task_work_run(tctx, *count); |
| return node; |
| } |
| |
| void tctx_task_work(struct callback_head *cb) |
| { |
| struct io_uring_task *tctx; |
| struct llist_node *ret; |
| unsigned int count = 0; |
| |
| tctx = container_of(cb, struct io_uring_task, task_work); |
| ret = tctx_task_work_run(tctx, UINT_MAX, &count); |
| /* can't happen */ |
| WARN_ON_ONCE(ret); |
| } |
| |
| static inline void io_req_local_work_add(struct io_kiocb *req, |
| struct io_ring_ctx *ctx, |
| unsigned flags) |
| { |
| unsigned nr_wait, nr_tw, nr_tw_prev; |
| struct llist_node *head; |
| |
| /* See comment above IO_CQ_WAKE_INIT */ |
| BUILD_BUG_ON(IO_CQ_WAKE_FORCE <= IORING_MAX_CQ_ENTRIES); |
| |
| /* |
| * We don't know how many reuqests is there in the link and whether |
| * they can even be queued lazily, fall back to non-lazy. |
| */ |
| if (req->flags & (REQ_F_LINK | REQ_F_HARDLINK)) |
| flags &= ~IOU_F_TWQ_LAZY_WAKE; |
| |
| guard(rcu)(); |
| |
| head = READ_ONCE(ctx->work_llist.first); |
| do { |
| nr_tw_prev = 0; |
| if (head) { |
| struct io_kiocb *first_req = container_of(head, |
| struct io_kiocb, |
| io_task_work.node); |
| /* |
| * Might be executed at any moment, rely on |
| * SLAB_TYPESAFE_BY_RCU to keep it alive. |
| */ |
| nr_tw_prev = READ_ONCE(first_req->nr_tw); |
| } |
| |
| /* |
| * Theoretically, it can overflow, but that's fine as one of |
| * previous adds should've tried to wake the task. |
| */ |
| nr_tw = nr_tw_prev + 1; |
| if (!(flags & IOU_F_TWQ_LAZY_WAKE)) |
| nr_tw = IO_CQ_WAKE_FORCE; |
| |
| req->nr_tw = nr_tw; |
| req->io_task_work.node.next = head; |
| } while (!try_cmpxchg(&ctx->work_llist.first, &head, |
| &req->io_task_work.node)); |
| |
| /* |
| * cmpxchg implies a full barrier, which pairs with the barrier |
| * in set_current_state() on the io_cqring_wait() side. It's used |
| * to ensure that either we see updated ->cq_wait_nr, or waiters |
| * going to sleep will observe the work added to the list, which |
| * is similar to the wait/wawke task state sync. |
| */ |
| |
| if (!head) { |
| if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) |
| atomic_or(IORING_SQ_TASKRUN, &ctx->rings->sq_flags); |
| if (ctx->has_evfd) |
| io_eventfd_signal(ctx); |
| } |
| |
| nr_wait = atomic_read(&ctx->cq_wait_nr); |
| /* not enough or no one is waiting */ |
| if (nr_tw < nr_wait) |
| return; |
| /* the previous add has already woken it up */ |
| if (nr_tw_prev >= nr_wait) |
| return; |
| wake_up_state(ctx->submitter_task, TASK_INTERRUPTIBLE); |
| } |
| |
| static void io_req_normal_work_add(struct io_kiocb *req) |
| { |
| struct io_uring_task *tctx = req->task->io_uring; |
| struct io_ring_ctx *ctx = req->ctx; |
| |
| /* task_work already pending, we're done */ |
| if (!llist_add(&req->io_task_work.node, &tctx->task_list)) |
| return; |
| |
| if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) |
| atomic_or(IORING_SQ_TASKRUN, &ctx->rings->sq_flags); |
| |
| /* SQPOLL doesn't need the task_work added, it'll run it itself */ |
| if (ctx->flags & IORING_SETUP_SQPOLL) { |
| struct io_sq_data *sqd = ctx->sq_data; |
| |
| if (sqd->thread) |
| __set_notify_signal(sqd->thread); |
| return; |
| } |
| |
| if (likely(!task_work_add(req->task, &tctx->task_work, ctx->notify_method))) |
| return; |
| |
| io_fallback_tw(tctx, false); |
| } |
| |
| void __io_req_task_work_add(struct io_kiocb *req, unsigned flags) |
| { |
| if (req->ctx->flags & IORING_SETUP_DEFER_TASKRUN) |
| io_req_local_work_add(req, req->ctx, flags); |
| else |
| io_req_normal_work_add(req); |
| } |
| |
| void io_req_task_work_add_remote(struct io_kiocb *req, struct io_ring_ctx *ctx, |
| unsigned flags) |
| { |
| if (WARN_ON_ONCE(!(ctx->flags & IORING_SETUP_DEFER_TASKRUN))) |
| return; |
| io_req_local_work_add(req, ctx, flags); |
| } |
| |
| static void __cold io_move_task_work_from_local(struct io_ring_ctx *ctx) |
| { |
| struct llist_node *node; |
| |
| node = llist_del_all(&ctx->work_llist); |
| while (node) { |
| struct io_kiocb *req = container_of(node, struct io_kiocb, |
| io_task_work.node); |
| |
| node = node->next; |
| io_req_normal_work_add(req); |
| } |
| } |
| |
| static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events, |
| int min_events) |
| { |
| if (llist_empty(&ctx->work_llist)) |
| return false; |
| if (events < min_events) |
| return true; |
| if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) |
| atomic_or(IORING_SQ_TASKRUN, &ctx->rings->sq_flags); |
| return false; |
| } |
| |
| static int __io_run_local_work(struct io_ring_ctx *ctx, struct io_tw_state *ts, |
| int min_events) |
| { |
| struct llist_node *node; |
| unsigned int loops = 0; |
| int ret = 0; |
| |
| if (WARN_ON_ONCE(ctx->submitter_task != current)) |
| return -EEXIST; |
| if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) |
| atomic_andnot(IORING_SQ_TASKRUN, &ctx->rings->sq_flags); |
| again: |
| /* |
| * llists are in reverse order, flip it back the right way before |
| * running the pending items. |
| */ |
| node = llist_reverse_order(io_llist_xchg(&ctx->work_llist, NULL)); |
| while (node) { |
| struct llist_node *next = node->next; |
| struct io_kiocb *req = container_of(node, struct io_kiocb, |
| io_task_work.node); |
| INDIRECT_CALL_2(req->io_task_work.func, |
| io_poll_task_func, io_req_rw_complete, |
| req, ts); |
| ret++; |
| node = next; |
| } |
| loops++; |
| |
| if (io_run_local_work_continue(ctx, ret, min_events)) |
| goto again; |
| io_submit_flush_completions(ctx); |
| if (io_run_local_work_continue(ctx, ret, min_events)) |
| goto again; |
| |
| trace_io_uring_local_work_run(ctx, ret, loops); |
| return ret; |
| } |
| |
| static inline int io_run_local_work_locked(struct io_ring_ctx *ctx, |
| int min_events) |
| { |
| struct io_tw_state ts = {}; |
| |
| if (llist_empty(&ctx->work_llist)) |
| return 0; |
| return __io_run_local_work(ctx, &ts, min_events); |
| } |
| |
| static int io_run_local_work(struct io_ring_ctx *ctx, int min_events) |
| { |
| struct io_tw_state ts = {}; |
| int ret; |
| |
| mutex_lock(&ctx->uring_lock); |
| ret = __io_run_local_work(ctx, &ts, min_events); |
| mutex_unlock(&ctx->uring_lock); |
| return ret; |
| } |
| |
| static void io_req_task_cancel(struct io_kiocb *req, struct io_tw_state *ts) |
| { |
| io_tw_lock(req->ctx, ts); |
| io_req_defer_failed(req, req->cqe.res); |
| } |
| |
| void io_req_task_submit(struct io_kiocb *req, struct io_tw_state *ts) |
| { |
| io_tw_lock(req->ctx, ts); |
| /* req->task == current here, checking PF_EXITING is safe */ |
| if (unlikely(req->task->flags & PF_EXITING)) |
| io_req_defer_failed(req, -EFAULT); |
| else if (req->flags & REQ_F_FORCE_ASYNC) |
| io_queue_iowq(req); |
| else |
| io_queue_sqe(req); |
| } |
| |
| void io_req_task_queue_fail(struct io_kiocb *req, int ret) |
| { |
| io_req_set_res(req, ret, 0); |
| req->io_task_work.func = io_req_task_cancel; |
| io_req_task_work_add(req); |
| } |
| |
| void io_req_task_queue(struct io_kiocb *req) |
| { |
| req->io_task_work.func = io_req_task_submit; |
| io_req_task_work_add(req); |
| } |
| |
| void io_queue_next(struct io_kiocb *req) |
| { |
| struct io_kiocb *nxt = io_req_find_next(req); |
| |
| if (nxt) |
| io_req_task_queue(nxt); |
| } |
| |
| static void io_free_batch_list(struct io_ring_ctx *ctx, |
| struct io_wq_work_node *node) |
| __must_hold(&ctx->uring_lock) |
| { |
| do { |
| struct io_kiocb *req = container_of(node, struct io_kiocb, |
| comp_list); |
| |
| if (unlikely(req->flags & IO_REQ_CLEAN_SLOW_FLAGS)) { |
| if (req->flags & REQ_F_REFCOUNT) { |
| node = req->comp_list.next; |
| if (!req_ref_put_and_test(req)) |
| continue; |
| } |
| if ((req->flags & REQ_F_POLLED) && req->apoll) { |
| struct async_poll *apoll = req->apoll; |
| |
| if (apoll->double_poll) |
| kfree(apoll->double_poll); |
| if (!io_alloc_cache_put(&ctx->apoll_cache, apoll)) |
| kfree(apoll); |
| req->flags &= ~REQ_F_POLLED; |
| } |
| if (req->flags & IO_REQ_LINK_FLAGS) |
| io_queue_next(req); |
| if (unlikely(req->flags & IO_REQ_CLEAN_FLAGS)) |
| io_clean_op(req); |
| } |
| io_put_file(req); |
| io_put_rsrc_node(ctx, req->rsrc_node); |
| io_put_task(req->task); |
| |
| node = req->comp_list.next; |
| io_req_add_to_cache(req, ctx); |
| } while (node); |
| } |
| |
| void __io_submit_flush_completions(struct io_ring_ctx *ctx) |
| __must_hold(&ctx->uring_lock) |
| { |
| struct io_submit_state *state = &ctx->submit_state; |
| struct io_wq_work_node *node; |
| |
| __io_cq_lock(ctx); |
| __wq_list_for_each(node, &state->compl_reqs) { |
| struct io_kiocb *req = container_of(node, struct io_kiocb, |
| comp_list); |
| |
| if (!(req->flags & REQ_F_CQE_SKIP) && |
| unlikely(!io_fill_cqe_req(ctx, req))) { |
| if (ctx->lockless_cq) { |
| spin_lock(&ctx->completion_lock); |
| io_req_cqe_overflow(req); |
| spin_unlock(&ctx->completion_lock); |
| } else { |
| io_req_cqe_overflow(req); |
| } |
| } |
| } |
| __io_cq_unlock_post(ctx); |
| |
| if (!wq_list_empty(&state->compl_reqs)) { |
| io_free_batch_list(ctx, state->compl_reqs.first); |
| INIT_WQ_LIST(&state->compl_reqs); |
| } |
| ctx->submit_state.cq_flush = false; |
| } |
| |
| static unsigned io_cqring_events(struct io_ring_ctx *ctx) |
| { |
| /* See comment at the top of this file */ |
| smp_rmb(); |
| return __io_cqring_events(ctx); |
| } |
| |
| /* |
| * We can't just wait for polled events to come to us, we have to actively |
| * find and complete them. |
| */ |
| static __cold void io_iopoll_try_reap_events(struct io_ring_ctx *ctx) |
| { |
| if (!(ctx->flags & IORING_SETUP_IOPOLL)) |
| return; |
| |
| mutex_lock(&ctx->uring_lock); |
| while (!wq_list_empty(&ctx->iopoll_list)) { |
| /* let it sleep and repeat later if can't complete a request */ |
| if (io_do_iopoll(ctx, true) == 0) |
| break; |
| /* |
| * Ensure we allow local-to-the-cpu processing to take place, |
| * in this case we need to ensure that we reap all events. |
| * Also let task_work, etc. to progress by releasing the mutex |
| */ |
| if (need_resched()) { |
| mutex_unlock(&ctx->uring_lock); |
| cond_resched(); |
| mutex_lock(&ctx->uring_lock); |
| } |
| } |
| mutex_unlock(&ctx->uring_lock); |
| } |
| |
| static int io_iopoll_check(struct io_ring_ctx *ctx, long min) |
| { |
| unsigned int nr_events = 0; |
| unsigned long check_cq; |
| |
| lockdep_assert_held(&ctx->uring_lock); |
| |
| if (!io_allowed_run_tw(ctx)) |
| return -EEXIST; |
| |
| check_cq = READ_ONCE(ctx->check_cq); |
| if (unlikely(check_cq)) { |
| if (check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT)) |
| __io_cqring_overflow_flush(ctx, false); |
| /* |
| * Similarly do not spin if we have not informed the user of any |
| * dropped CQE. |
| */ |
| if (check_cq & BIT(IO_CHECK_CQ_DROPPED_BIT)) |
| return -EBADR; |
| } |
| /* |
| * Don't enter poll loop if we already have events pending. |
| * If we do, we can potentially be spinning for commands that |
| * already triggered a CQE (eg in error). |
| */ |
| if (io_cqring_events(ctx)) |
| return 0; |
| |
| do { |
| int ret = 0; |
| |
| /* |
| * If a submit got punted to a workqueue, we can have the |
| * application entering polling for a command before it gets |
| * issued. That app will hold the uring_lock for the duration |
| * of the poll right here, so we need to take a breather every |
| * now and then to ensure that the issue has a chance to add |
| * the poll to the issued list. Otherwise we can spin here |
| * forever, while the workqueue is stuck trying to acquire the |
| * very same mutex. |
| */ |
| if (wq_list_empty(&ctx->iopoll_list) || |
| io_task_work_pending(ctx)) { |
| u32 tail = ctx->cached_cq_tail; |
| |
| (void) io_run_local_work_locked(ctx, min); |
| |
| if (task_work_pending(current) || |
| wq_list_empty(&ctx->iopoll_list)) { |
| mutex_unlock(&ctx->uring_lock); |
| io_run_task_work(); |
| mutex_lock(&ctx->uring_lock); |
| } |
| /* some requests don't go through iopoll_list */ |
| if (tail != ctx->cached_cq_tail || |
| wq_list_empty(&ctx->iopoll_list)) |
| break; |
| } |
| ret = io_do_iopoll(ctx, !min); |
| if (unlikely(ret < 0)) |
| return ret; |
| |
| if (task_sigpending(current)) |
| return -EINTR; |
| if (need_resched()) |
| break; |
| |
| nr_events += ret; |
| } while (nr_events < min); |
| |
| return 0; |
| } |
| |
| void io_req_task_complete(struct io_kiocb *req, struct io_tw_state *ts) |
| { |
| io_req_complete_defer(req); |
| } |
| |
| /* |
| * After the iocb has been issued, it's safe to be found on the poll list. |
| * Adding the kiocb to the list AFTER submission ensures that we don't |
| * find it from a io_do_iopoll() thread before the issuer is done |
| * accessing the kiocb cookie. |
| */ |
| static void io_iopoll_req_issued(struct io_kiocb *req, unsigned int issue_flags) |
| { |
| struct io_ring_ctx *ctx = req->ctx; |
| const bool needs_lock = issue_flags & IO_URING_F_UNLOCKED; |
| |
| /* workqueue context doesn't hold uring_lock, grab it now */ |
| if (unlikely(needs_lock)) |
| mutex_lock(&ctx->uring_lock); |
| |
| /* |
| * Track whether we have multiple files in our lists. This will impact |
| * how we do polling eventually, not spinning if we're on potentially |
| * different devices. |
| */ |
| if (wq_list_empty(&ctx->iopoll_list)) { |
| ctx->poll_multi_queue = false; |
| } else if (!ctx->poll_multi_queue) { |
| struct io_kiocb *list_req; |
| |
| list_req = container_of(ctx->iopoll_list.first, struct io_kiocb, |
| comp_list); |
| if (list_req->file != req->file) |
| ctx->poll_multi_queue = true; |
| } |
| |
| /* |
| * For fast devices, IO may have already completed. If it has, add |
| * it to the front so we find it first. |
| */ |
| if (READ_ONCE(req->iopoll_completed)) |
| wq_list_add_head(&req->comp_list, &ctx->iopoll_list); |
| else |
| wq_list_add_tail(&req->comp_list, &ctx->iopoll_list); |
| |
| if (unlikely(needs_lock)) { |
| /* |
| * If IORING_SETUP_SQPOLL is enabled, sqes are either handle |
| * in sq thread task context or in io worker task context. If |
| * current task context is sq thread, we don't need to check |
| * whether should wake up sq thread. |
| */ |
| if ((ctx->flags & IORING_SETUP_SQPOLL) && |
| wq_has_sleeper(&ctx->sq_data->wait)) |
| wake_up(&ctx->sq_data->wait); |
| |
| mutex_unlock(&ctx->uring_lock); |
| } |
| } |
| |
| io_req_flags_t io_file_get_flags(struct file *file) |
| { |
| io_req_flags_t res = 0; |
| |
| if (S_ISREG(file_inode(file)->i_mode)) |
| res |= REQ_F_ISREG; |
| if ((file->f_flags & O_NONBLOCK) || (file->f_mode & FMODE_NOWAIT)) |
| res |= REQ_F_SUPPORT_NOWAIT; |
| return res; |
| } |
| |
| bool io_alloc_async_data(struct io_kiocb *req) |
| { |
| const struct io_issue_def *def = &io_issue_defs[req->opcode]; |
| |
| WARN_ON_ONCE(!def->async_size); |
| req->async_data = kmalloc(def->async_size, GFP_KERNEL); |
| if (req->async_data) { |
| req->flags |= REQ_F_ASYNC_DATA; |
| return false; |
| } |
| return true; |
| } |
| |
| static u32 io_get_sequence(struct io_kiocb *req) |
| { |
| u32 seq = req->ctx->cached_sq_head; |
| struct io_kiocb *cur; |
| |
| /* need original cached_sq_head, but it was increased for each req */ |
| io_for_each_link(cur, req) |
| seq--; |
| return seq; |
| } |
| |
| static __cold void io_drain_req(struct io_kiocb *req) |
| __must_hold(&ctx->uring_lock) |
| { |
| struct io_ring_ctx *ctx = req->ctx; |
| struct io_defer_entry *de; |
| int ret; |
| u32 seq = io_get_sequence(req); |
| |
| /* Still need defer if there is pending req in defer list. */ |
| spin_lock(&ctx->completion_lock); |
| if (!req_need_defer(req, seq) && list_empty_careful(&ctx->defer_list)) { |
| spin_unlock(&ctx->completion_lock); |
| queue: |
| ctx->drain_active = false; |
| io_req_task_queue(req); |
| return; |
| } |
| spin_unlock(&ctx->completion_lock); |
| |
| io_prep_async_link(req); |
| de = kmalloc(sizeof(*de), GFP_KERNEL); |
| if (!de) { |
| ret = -ENOMEM; |
| io_req_defer_failed(req, ret); |
| return; |
| } |
| |
| spin_lock(&ctx->completion_lock); |
| if (!req_need_defer(req, seq) && list_empty(&ctx->defer_list)) { |
| spin_unlock(&ctx->completion_lock); |
| kfree(de); |
| goto queue; |
| } |
| |
| trace_io_uring_defer(req); |
| de->req = req; |
| de->seq = seq; |
| list_add_tail(&de->list, &ctx->defer_list); |
| spin_unlock(&ctx->completion_lock); |
| } |
| |
| static bool io_assign_file(struct io_kiocb *req, const struct io_issue_def *def, |
| unsigned int issue_flags) |
| { |
| if (req->file || !def->needs_file) |
| return true; |
| |
| if (req->flags & REQ_F_FIXED_FILE) |
| req->file = io_file_get_fixed(req, req->cqe.fd, issue_flags); |
| else |
| req->file = io_file_get_normal(req, req->cqe.fd); |
| |
| return !!req->file; |
| } |
| |
| static int io_issue_sqe(struct io_kiocb *req, unsigned int issue_flags) |
| { |
| const struct io_issue_def *def = &io_issue_defs[req->opcode]; |
| const struct cred *creds = NULL; |
| int ret; |
| |
| if (unlikely(!io_assign_file(req, def, issue_flags))) |
| return -EBADF; |
| |
| if (unlikely((req->flags & REQ_F_CREDS) && req->creds != current_cred())) |
| creds = override_creds(req->creds); |
| |
| if (!def->audit_skip) |
| audit_uring_entry(req->opcode); |
| |
| ret = def->issue(req, issue_flags); |
| |
| if (!def->audit_skip) |
| audit_uring_exit(!ret, ret); |
| |
| if (creds) |
| revert_creds(creds); |
| |
| if (ret == IOU_OK) { |
| if (issue_flags & IO_URING_F_COMPLETE_DEFER) |
| io_req_complete_defer(req); |
| else |
| io_req_complete_post(req, issue_flags); |
| |
| return 0; |
| } |
| |
| if (ret == IOU_ISSUE_SKIP_COMPLETE) { |
| ret = 0; |
| io_arm_ltimeout(req); |
| |
| /* If the op doesn't have a file, we're not polling for it */ |
| if ((req->ctx->flags & IORING_SETUP_IOPOLL) && def->iopoll_queue) |
| io_iopoll_req_issued(req, issue_flags); |
| } |
| return ret; |
| } |
| |
| int io_poll_issue(struct io_kiocb *req, struct io_tw_state *ts) |
| { |
| io_tw_lock(req->ctx, ts); |
| return io_issue_sqe(req, IO_URING_F_NONBLOCK|IO_URING_F_MULTISHOT| |
| IO_URING_F_COMPLETE_DEFER); |
| } |
| |
| struct io_wq_work *io_wq_free_work(struct io_wq_work *work) |
| { |
| struct io_kiocb *req = container_of(work, struct io_kiocb, work); |
| struct io_kiocb *nxt = NULL; |
| |
| if (req_ref_put_and_test(req)) { |
| if (req->flags & IO_REQ_LINK_FLAGS) |
| nxt = io_req_find_next(req); |
| io_free_req(req); |
| } |
| return nxt ? &nxt->work : NULL; |
| } |
| |
| void io_wq_submit_work(struct io_wq_work *work) |
| { |
| struct io_kiocb *req = container_of(work, struct io_kiocb, work); |
| const struct io_issue_def *def = &io_issue_defs[req->opcode]; |
| unsigned int issue_flags = IO_URING_F_UNLOCKED | IO_URING_F_IOWQ; |
| bool needs_poll = false; |
| int ret = 0, err = -ECANCELED; |
| |
| /* one will be dropped by ->io_wq_free_work() after returning to io-wq */ |
| if (!(req->flags & REQ_F_REFCOUNT)) |
| __io_req_set_refcount(req, 2); |
| else |
| req_ref_get(req); |
| |
| io_arm_ltimeout(req); |
| |
| /* either cancelled or io-wq is dying, so don't touch tctx->iowq */ |
| if (atomic_read(&work->flags) & IO_WQ_WORK_CANCEL) { |
| fail: |
| io_req_task_queue_fail(req, err); |
| return; |
| } |
| if (!io_assign_file(req, def, issue_flags)) { |
| err = -EBADF; |
| atomic_or(IO_WQ_WORK_CANCEL, &work->flags); |
| goto fail; |
| } |
| |
| /* |
| * If DEFER_TASKRUN is set, it's only allowed to post CQEs from the |
| * submitter task context. Final request completions are handed to the |
| * right context, however this is not the case of auxiliary CQEs, |
| * which is the main mean of operation for multishot requests. |
| * Don't allow any multishot execution from io-wq. It's more restrictive |
| * than necessary and also cleaner. |
| */ |
| if (req->flags & REQ_F_APOLL_MULTISHOT) { |
| err = -EBADFD; |
| if (!io_file_can_poll(req)) |
| goto fail; |
| if (req->file->f_flags & O_NONBLOCK || |
| req->file->f_mode & FMODE_NOWAIT) { |
| err = -ECANCELED; |
| if (io_arm_poll_handler(req, issue_flags) != IO_APOLL_OK) |
| goto fail; |
| return; |
| } else { |
| req->flags &= ~REQ_F_APOLL_MULTISHOT; |
| } |
| } |
| |
| if (req->flags & REQ_F_FORCE_ASYNC) { |
| bool opcode_poll = def->pollin || def->pollout; |
| |
| if (opcode_poll && io_file_can_poll(req)) { |
| needs_poll = true; |
| issue_flags |= IO_URING_F_NONBLOCK; |
| } |
| } |
| |
| do { |
| ret = io_issue_sqe(req, issue_flags); |
| if (ret != -EAGAIN) |
| break; |
| |
| /* |
| * If REQ_F_NOWAIT is set, then don't wait or retry with |
| * poll. -EAGAIN is final for that case. |
| */ |
| if (req->flags & REQ_F_NOWAIT) |
| break; |
| |
| /* |
| * We can get EAGAIN for iopolled IO even though we're |
| * forcing a sync submission from here, since we can't |
| * wait for request slots on the block side. |
| */ |
| if (!needs_poll) { |
| if (!(req->ctx->flags & IORING_SETUP_IOPOLL)) |
| break; |
| if (io_wq_worker_stopped()) |
| break; |
| cond_resched(); |
| continue; |
| } |
| |
| if (io_arm_poll_handler(req, issue_flags) == IO_APOLL_OK) |
| return; |
| /* aborted or ready, in either case retry blocking */ |
| needs_poll = false; |
| issue_flags &= ~IO_URING_F_NONBLOCK; |
| } while (1); |
| |
| /* avoid locking problems by failing it from a clean context */ |
| if (ret) |
| io_req_task_queue_fail(req, ret); |
| } |
| |
| inline struct file *io_file_get_fixed(struct io_kiocb *req, int fd, |
| unsigned int issue_flags) |
| { |
| struct io_ring_ctx *ctx = req->ctx; |
| struct io_fixed_file *slot; |
| struct file *file = NULL; |
| |
| io_ring_submit_lock(ctx, issue_flags); |
| |
| if (unlikely((unsigned int)fd >= ctx->nr_user_files)) |
| goto out; |
| fd = array_index_nospec(fd, ctx->nr_user_files); |
| slot = io_fixed_file_slot(&ctx->file_table, fd); |
| if (!req->rsrc_node) |
| __io_req_set_rsrc_node(req, ctx); |
| req->flags |= io_slot_flags(slot); |
| file = io_slot_file(slot); |
| out: |
| io_ring_submit_unlock(ctx, issue_flags); |
| return file; |
| } |
| |
| struct file *io_file_get_normal(struct io_kiocb *req, int fd) |
| { |
| struct file *file = fget(fd); |
| |
| trace_io_uring_file_get(req, fd); |
| |
| /* we don't allow fixed io_uring files */ |
| if (file && io_is_uring_fops(file)) |
| io_req_track_inflight(req); |
| return file; |
| } |
| |
| static void io_queue_async(struct io_kiocb *req, int ret) |
| __must_hold(&req->ctx->uring_lock) |
| { |
| struct io_kiocb *linked_timeout; |
| |
| if (ret != -EAGAIN || (req->flags & REQ_F_NOWAIT)) { |
| io_req_defer_failed(req, ret); |
| return; |
| } |
| |
| linked_timeout = io_prep_linked_timeout(req); |
| |
| switch (io_arm_poll_handler(req, 0)) { |
| case IO_APOLL_READY: |
| io_kbuf_recycle(req, 0); |
| io_req_task_queue(req); |
| break; |
| case IO_APOLL_ABORTED: |
| io_kbuf_recycle(req, 0); |
| io_queue_iowq(req); |
| break; |
| case IO_APOLL_OK: |
| break; |
| } |
| |
| if (linked_timeout) |
| io_queue_linked_timeout(linked_timeout); |
| } |
| |
| static inline void io_queue_sqe(struct io_kiocb *req) |
| __must_hold(&req->ctx->uring_lock) |
| { |
| int ret; |
| |
| ret = io_issue_sqe(req, IO_URING_F_NONBLOCK|IO_URING_F_COMPLETE_DEFER); |
| |
| /* |
| * We async punt it if the file wasn't marked NOWAIT, or if the file |
| * doesn't support non-blocking read/write attempts |
| */ |
| if (unlikely(ret)) |
| io_queue_async(req, ret); |
| } |
| |
| static void io_queue_sqe_fallback(struct io_kiocb *req) |
| __must_hold(&req->ctx->uring_lock) |
| { |
| if (unlikely(req->flags & REQ_F_FAIL)) { |
| /* |
| * We don't submit, fail them all, for that replace hardlinks |
| * with normal links. Extra REQ_F_LINK is tolerated. |
| */ |
| req->flags &= ~REQ_F_HARDLINK; |
| req->flags |= REQ_F_LINK; |
| io_req_defer_failed(req, req->cqe.res); |
| } else { |
| if (unlikely(req->ctx->drain_active)) |
| io_drain_req(req); |
| else |
| io_queue_iowq(req); |
| } |
| } |
| |
| /* |
| * Check SQE restrictions (opcode and flags). |
| * |
| * Returns 'true' if SQE is allowed, 'false' otherwise. |
| */ |
| static inline bool io_check_restriction(struct io_ring_ctx *ctx, |
| struct io_kiocb *req, |
| unsigned int sqe_flags) |
| { |
| if (!test_bit(req->opcode, ctx->restrictions.sqe_op)) |
| return false; |
| |
| if ((sqe_flags & ctx->restrictions.sqe_flags_required) != |
| ctx->restrictions.sqe_flags_required) |
| return false; |
| |
| if (sqe_flags & ~(ctx->restrictions.sqe_flags_allowed | |
| ctx->restrictions.sqe_flags_required)) |
| return false; |
| |
| return true; |
| } |
| |
| static void io_init_req_drain(struct io_kiocb *req) |
| { |
| struct io_ring_ctx *ctx = req->ctx; |
| struct io_kiocb *head = ctx->submit_state.link.head; |
| |
| ctx->drain_active = true; |
| if (head) { |
| /* |
| * If we need to drain a request in the middle of a link, drain |
| * the head request and the next request/link after the current |
| * link. Considering sequential execution of links, |
| * REQ_F_IO_DRAIN will be maintained for every request of our |
| * link. |
| */ |
| head->flags |= REQ_F_IO_DRAIN | REQ_F_FORCE_ASYNC; |
| ctx->drain_next = true; |
| } |
| } |
| |
| static __cold int io_init_fail_req(struct io_kiocb *req, int err) |
| { |
| /* ensure per-opcode data is cleared if we fail before prep */ |
| memset(&req->cmd.data, 0, sizeof(req->cmd.data)); |
| return err; |
| } |
| |
| static int io_init_req(struct io_ring_ctx *ctx, struct io_kiocb *req, |
| const struct io_uring_sqe *sqe) |
| __must_hold(&ctx->uring_lock) |
| { |
| const struct io_issue_def *def; |
| unsigned int sqe_flags; |
| int personality; |
| u8 opcode; |
| |
| /* req is partially pre-initialised, see io_preinit_req() */ |
| req->opcode = opcode = READ_ONCE(sqe->opcode); |
| /* same numerical values with corresponding REQ_F_*, safe to copy */ |
| sqe_flags = READ_ONCE(sqe->flags); |
| req->flags = (io_req_flags_t) sqe_flags; |
| req->cqe.user_data = READ_ONCE(sqe->user_data); |
| req->file = NULL; |
| req->rsrc_node = NULL; |
| req->task = current; |
| req->cancel_seq_set = false; |
| |
| if (unlikely(opcode >= IORING_OP_LAST)) { |
| req->opcode = 0; |
| return io_init_fail_req(req, -EINVAL); |
| } |
| def = &io_issue_defs[opcode]; |
| if (unlikely(sqe_flags & ~SQE_COMMON_FLAGS)) { |
| /* enforce forwards compatibility on users */ |
| if (sqe_flags & ~SQE_VALID_FLAGS) |
| return io_init_fail_req(req, -EINVAL); |
| if (sqe_flags & IOSQE_BUFFER_SELECT) { |
| if (!def->buffer_select) |
| return io_init_fail_req(req, -EOPNOTSUPP); |
| req->buf_index = READ_ONCE(sqe->buf_group); |
| } |
| if (sqe_flags & IOSQE_CQE_SKIP_SUCCESS) |
| ctx->drain_disabled = true; |
| if (sqe_flags & IOSQE_IO_DRAIN) { |
| if (ctx->drain_disabled) |
| return io_init_fail_req(req, -EOPNOTSUPP); |
| io_init_req_drain(req); |
| } |
| } |
| if (unlikely(ctx->restricted || ctx->drain_active || ctx->drain_next)) { |
| if (ctx->restricted && !io_check_restriction(ctx, req, sqe_flags)) |
| return io_init_fail_req(req, -EACCES); |
| /* knock it to the slow queue path, will be drained there */ |
| if (ctx->drain_active) |
| req->flags |= REQ_F_FORCE_ASYNC; |
| /* if there is no link, we're at "next" request and need to drain */ |
| if (unlikely(ctx->drain_next) && !ctx->submit_state.link.head) { |
| ctx->drain_next = false; |
| ctx->drain_active = true; |
| req->flags |= REQ_F_IO_DRAIN | REQ_F_FORCE_ASYNC; |
| } |
| } |
| |
| if (!def->ioprio && sqe->ioprio) |
| return io_init_fail_req(req, -EINVAL); |
| if (!def->iopoll && (ctx->flags & IORING_SETUP_IOPOLL)) |
| return io_init_fail_req(req, -EINVAL); |
| |
| if (def->needs_file) { |
| struct io_submit_state *state = &ctx->submit_state; |
| |
| req->cqe.fd = READ_ONCE(sqe->fd); |
| |
| /* |
| * Plug now if we have more than 2 IO left after this, and the |
| * target is potentially a read/write to block based storage. |
| */ |
| if (state->need_plug && def->plug) { |
| state->plug_started = true; |
| state->need_plug = false; |
| blk_start_plug_nr_ios(&state->plug, state->submit_nr); |
| } |
| } |
| |
| personality = READ_ONCE(sqe->personality); |
| if (personality) { |
| int ret; |
| |
| req->creds = xa_load(&ctx->personalities, personality); |
| if (!req->creds) |
| return io_init_fail_req(req, -EINVAL); |
| get_cred(req->creds); |
| ret = security_uring_override_creds(req->creds); |
| if (ret) { |
| put_cred(req->creds); |
| return io_init_fail_req(req, ret); |
| } |
| req->flags |= REQ_F_CREDS; |
| } |
| |
| return def->prep(req, sqe); |
| } |
| |
| static __cold int io_submit_fail_init(const struct io_uring_sqe *sqe, |
| struct io_kiocb *req, int ret) |
| { |
| struct io_ring_ctx *ctx = req->ctx; |
| struct io_submit_link *link = &ctx->submit_state.link; |
| struct io_kiocb *head = link->head; |
| |
| trace_io_uring_req_failed(sqe, req, ret); |
| |
| /* |
| * Avoid breaking links in the middle as it renders links with SQPOLL |
| * unusable. Instead of failing eagerly, continue assembling the link if |
| * applicable and mark the head with REQ_F_FAIL. The link flushing code |
| * should find the flag and handle the rest. |
| */ |
| req_fail_link_node(req, ret); |
| if (head && !(head->flags & REQ_F_FAIL)) |
| req_fail_link_node(head, -ECANCELED); |
| |
| if (!(req->flags & IO_REQ_LINK_FLAGS)) { |
| if (head) { |
| link->last->link = req; |
| link->head = NULL; |
| req = head; |
| } |
| io_queue_sqe_fallback(req); |
| return ret; |
| } |
| |
| if (head) |
| link->last->link = req; |
| else |
| link->head = req; |
| link->last = req; |
| return 0; |
| } |
| |
| static inline int io_submit_sqe(struct io_ring_ctx *ctx, struct io_kiocb *req, |
| const struct io_uring_sqe *sqe) |
| __must_hold(&ctx->uring_lock) |
| { |
| struct io_submit_link *link = &ctx->submit_state.link; |
| int ret; |
| |
| ret = io_init_req(ctx, req, sqe); |
| if (unlikely(ret)) |
| return io_submit_fail_init(sqe, req, ret); |
| |
| trace_io_uring_submit_req(req); |
| |
| /* |
| * If we already have a head request, queue this one for async |
| * submittal once the head completes. If we don't have a head but |
| * IOSQE_IO_LINK is set in the sqe, start a new head. This one will be |
| * submitted sync once the chain is complete. If none of those |
| * conditions are true (normal request), then just queue it. |
| */ |
| if (unlikely(link->head)) { |
| trace_io_uring_link(req, link->head); |
| link->last->link = req; |
| link->last = req; |
| |
| if (req->flags & IO_REQ_LINK_FLAGS) |
| return 0; |
| /* last request of the link, flush it */ |
| req = link->head; |
| link->head = NULL; |
| if (req->flags & (REQ_F_FORCE_ASYNC | REQ_F_FAIL)) |
| goto fallback; |
| |
| } else if (unlikely(req->flags & (IO_REQ_LINK_FLAGS | |
| REQ_F_FORCE_ASYNC | REQ_F_FAIL))) { |
| if (req->flags & IO_REQ_LINK_FLAGS) { |
| link->head = req; |
| link->last = req; |
| } else { |
| fallback: |
| io_queue_sqe_fallback(req); |
| } |
| return 0; |
| } |
| |
| io_queue_sqe(req); |
| return 0; |
| } |
| |
| /* |
| * Batched submission is done, ensure local IO is flushed out. |
| */ |
| static void io_submit_state_end(struct io_ring_ctx *ctx) |
| { |
| struct io_submit_state *state = &ctx->submit_state; |
| |
| if (unlikely(state->link.head)) |
| io_queue_sqe_fallback(state->link.head); |
| /* flush only after queuing links as they can generate completions */ |
| io_submit_flush_completions(ctx); |
| if (state->plug_started) |
| blk_finish_plug(&state->plug); |
| } |
| |
| /* |
| * Start submission side cache. |
| */ |
| static void io_submit_state_start(struct io_submit_state *state, |
| unsigned int max_ios) |
| { |
| state->plug_started = false; |
| state->need_plug = max_ios > 2; |
| state->submit_nr = max_ios; |
| /* set only head, no need to init link_last in advance */ |
| state->link.head = NULL; |
| } |
| |
| static void io_commit_sqring(struct io_ring_ctx *ctx) |
| { |
| struct io_rings *rings = ctx->rings; |
| |
| /* |
| * Ensure any loads from the SQEs are done at this point, |
| * since once we write the new head, the application could |
| * write new data to them. |
| */ |
| smp_store_release(&rings->sq.head, ctx->cached_sq_head); |
| } |
| |
| /* |
| * Fetch an sqe, if one is available. Note this returns a pointer to memory |
| * that is mapped by userspace. This means that care needs to be taken to |
| * ensure that reads are stable, as we cannot rely on userspace always |
| * being a good citizen. If members of the sqe are validated and then later |
| * used, it's important that those reads are done through READ_ONCE() to |
| * prevent a re-load down the line. |
| */ |
| static bool io_get_sqe(struct io_ring_ctx *ctx, const struct io_uring_sqe **sqe) |
| { |
| unsigned mask = ctx->sq_entries - 1; |
| unsigned head = ctx->cached_sq_head++ & mask; |
| |
| if (!(ctx->flags & IORING_SETUP_NO_SQARRAY)) { |
| head = READ_ONCE(ctx->sq_array[head]); |
| if (unlikely(head >= ctx->sq_entries)) { |
| /* drop invalid entries */ |
| spin_lock(&ctx->completion_lock); |
| ctx->cq_extra--; |
| spin_unlock(&ctx->completion_lock); |
| WRITE_ONCE(ctx->rings->sq_dropped, |
| READ_ONCE(ctx->rings->sq_dropped) + 1); |
| return false; |
| } |
| } |
| |
| /* |
| * The cached sq head (or cq tail) serves two purposes: |
| * |
| * 1) allows us to batch the cost of updating the user visible |
| * head updates. |
| * 2) allows the kernel side to track the head on its own, even |
| * though the application is the one updating it. |
| */ |
| |
| /* double index for 128-byte SQEs, twice as long */ |
| if (ctx->flags & IORING_SETUP_SQE128) |
| head <<= 1; |
| *sqe = &ctx->sq_sqes[head]; |
| return true; |
| } |
| |
| int io_submit_sqes(struct io_ring_ctx *ctx, unsigned int nr) |
| __must_hold(&ctx->uring_lock) |
| { |
| unsigned int entries = io_sqring_entries(ctx); |
| unsigned int left; |
| int ret; |
| |
| if (unlikely(!entries)) |
| return 0; |
| /* make sure SQ entry isn't read before tail */ |
| ret = left = min(nr, entries); |
| io_get_task_refs(left); |
| io_submit_state_start(&ctx->submit_state, left); |
| |
| do { |
| const struct io_uring_sqe *sqe; |
| struct io_kiocb *req; |
| |
| if (unlikely(!io_alloc_req(ctx, &req))) |
| break; |
| if (unlikely(!io_get_sqe(ctx, &sqe))) { |
| io_req_add_to_cache(req, ctx); |
| break; |
| } |
| |
| /* |
| * Continue submitting even for sqe failure if the |
| * ring was setup with IORING_SETUP_SUBMIT_ALL |
| */ |
| if (unlikely(io_submit_sqe(ctx, req, sqe)) && |
| !(ctx->flags & IORING_SETUP_SUBMIT_ALL)) { |
| left--; |
| break; |
| } |
| } while (--left); |
| |
| if (unlikely(left)) { |
| ret -= left; |
| /* try again if it submitted nothing and can't allocate a req */ |
| if (!ret && io_req_cache_empty(ctx)) |
| ret = -EAGAIN; |
| current->io_uring->cached_refs += left; |
| } |
| |
| io_submit_state_end(ctx); |
| /* Commit SQ ring head once we've consumed and submitted all SQEs */ |
| io_commit_sqring(ctx); |
| return ret; |
| } |
| |
| static int io_wake_function(struct wait_queue_entry *curr, unsigned int mode, |
| int wake_flags, void *key) |
| { |
| struct io_wait_queue *iowq = container_of(curr, struct io_wait_queue, wq); |
| |
| /* |
| * Cannot safely flush overflowed CQEs from here, ensure we wake up |
| * the task, and the next invocation will do it. |
| */ |
| if (io_should_wake(iowq) || io_has_work(iowq->ctx)) |
| return autoremove_wake_function(curr, mode, wake_flags, key); |
| return -1; |
| } |
| |
| int io_run_task_work_sig(struct io_ring_ctx *ctx) |
| { |
| if (!llist_empty(&ctx->work_llist)) { |
| __set_current_state(TASK_RUNNING); |
| if (io_run_local_work(ctx, INT_MAX) > 0) |
| return 0; |
| } |
| if (io_run_task_work() > 0) |
| return 0; |
| if (task_sigpending(current)) |
| return -EINTR; |
| return 0; |
| } |
| |
| static bool current_pending_io(void) |
| { |
| struct io_uring_task *tctx = current->io_uring; |
| |
| if (!tctx) |
| return false; |
| return percpu_counter_read_positive(&tctx->inflight); |
| } |
| |
| /* when returns >0, the caller should retry */ |
| static inline int io_cqring_wait_schedule(struct io_ring_ctx *ctx, |
| struct io_wait_queue *iowq) |
| { |
| int ret; |
| |
| if (unlikely(READ_ONCE(ctx->check_cq))) |
| return 1; |
| if (unlikely(!llist_empty(&ctx->work_llist))) |
| return 1; |
| if (unlikely(test_thread_flag(TIF_NOTIFY_SIGNAL))) |
| return 1; |
| if (unlikely(task_sigpending(current))) |
| return -EINTR; |
| if (unlikely(io_should_wake(iowq))) |
| return 0; |
| |
| /* |
| * Mark us as being in io_wait if we have pending requests, so cpufreq |
| * can take into account that the task is waiting for IO - turns out |
| * to be important for low QD IO. |
| */ |
| if (current_pending_io()) |
| current->in_iowait = 1; |
| ret = 0; |
| if (iowq->timeout == KTIME_MAX) |
| schedule(); |
| else if (!schedule_hrtimeout(&iowq->timeout, HRTIMER_MODE_ABS)) |
| ret = -ETIME; |
| current->in_iowait = 0; |
| return ret; |
| } |
| |
| /* |
| * Wait until events become available, if we don't already have some. The |
| * application must reap them itself, as they reside on the shared cq ring. |
| */ |
| static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events, |
| const sigset_t __user *sig, size_t sigsz, |
| struct __kernel_timespec __user *uts) |
| { |
| struct io_wait_queue iowq; |
| struct io_rings *rings = ctx->rings; |
| int ret; |
| |
| if (!io_allowed_run_tw(ctx)) |
| return -EEXIST; |
| if (!llist_empty(&ctx->work_llist)) |
| io_run_local_work(ctx, min_events); |
| io_run_task_work(); |
| |
| if (unlikely(test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq))) |
| io_cqring_do_overflow_flush(ctx); |
| if (__io_cqring_events_user(ctx) >= min_events) |
| return 0; |
| |
| init_waitqueue_func_entry(&iowq.wq, io_wake_function); |
| iowq.wq.private = current; |
| INIT_LIST_HEAD(&iowq.wq.entry); |
| iowq.ctx = ctx; |
| iowq.nr_timeouts = atomic_read(&ctx->cq_timeouts); |
| iowq.cq_tail = READ_ONCE(ctx->rings->cq.head) + min_events; |
| iowq.timeout = KTIME_MAX; |
| |
| if (uts) { |
| struct timespec64 ts; |
| ktime_t dt; |
| |
| if (get_timespec64(&ts, uts)) |
| return -EFAULT; |
| |
| dt = timespec64_to_ktime(ts); |
| iowq.timeout = ktime_add(dt, ktime_get()); |
| io_napi_adjust_timeout(ctx, &iowq, dt); |
| } |
| |
| if (sig) { |
| #ifdef CONFIG_COMPAT |
| if (in_compat_syscall()) |
| ret = set_compat_user_sigmask((const compat_sigset_t __user *)sig, |
| sigsz); |
| else |
| #endif |
| ret = set_user_sigmask(sig, sigsz); |
| |
| if (ret) |
| return ret; |
| } |
| |
| io_napi_busy_loop(ctx, &iowq); |
| |
| trace_io_uring_cqring_wait(ctx, min_events); |
| do { |
| int nr_wait = (int) iowq.cq_tail - READ_ONCE(ctx->rings->cq.tail); |
| unsigned long check_cq; |
| |
| if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) { |
| atomic_set(&ctx->cq_wait_nr, nr_wait); |
| set_current_state(TASK_INTERRUPTIBLE); |
| } else { |
| prepare_to_wait_exclusive(&ctx->cq_wait, &iowq.wq, |
| TASK_INTERRUPTIBLE); |
| } |
| |
| ret = io_cqring_wait_schedule(ctx, &iowq); |
| __set_current_state(TASK_RUNNING); |
| atomic_set(&ctx->cq_wait_nr, IO_CQ_WAKE_INIT); |
| |
| /* |
| * Run task_work after scheduling and before io_should_wake(). |
| * If we got woken because of task_work being processed, run it |
| * now rather than let the caller do another wait loop. |
| */ |
| io_run_task_work(); |
| if (!llist_empty(&ctx->work_llist)) |
| io_run_local_work(ctx, nr_wait); |
| |
| /* |
| * Non-local task_work will be run on exit to userspace, but |
| * if we're using DEFER_TASKRUN, then we could have waited |
| * with a timeout for a number of requests. If the timeout |
| * hits, we could have some requests ready to process. Ensure |
| * this break is _after_ we have run task_work, to avoid |
| * deferring running potentially pending requests until the |
| * next time we wait for events. |
| */ |
| if (ret < 0) |
| break; |
| |
| check_cq = READ_ONCE(ctx->check_cq); |
| if (unlikely(check_cq)) { |
| /* let the caller flush overflows, retry */ |
| if (check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT)) |
| io_cqring_do_overflow_flush(ctx); |
| if (check_cq & BIT(IO_CHECK_CQ_DROPPED_BIT)) { |
| ret = -EBADR; |
| break; |
| } |
| } |
| |
| if (io_should_wake(&iowq)) { |
| ret = 0; |
| break; |
| } |
| cond_resched(); |
| } while (1); |
| |
| if (!(ctx->flags & IORING_SETUP_DEFER_TASKRUN)) |
| finish_wait(&ctx->cq_wait, &iowq.wq); |
| restore_saved_sigmask_unless(ret == -EINTR); |
| |
| return READ_ONCE(rings->cq.head) == READ_ONCE(rings->cq.tail) ? ret : 0; |
| } |
| |
| static void *io_rings_map(struct io_ring_ctx *ctx, unsigned long uaddr, |
| size_t size) |
| { |
| return __io_uaddr_map(&ctx->ring_pages, &ctx->n_ring_pages, uaddr, |
| size); |
| } |
| |
| static void *io_sqes_map(struct io_ring_ctx *ctx, unsigned long uaddr, |
| size_t size) |
| { |
| return __io_uaddr_map(&ctx->sqe_pages, &ctx->n_sqe_pages, uaddr, |
| size); |
| } |
| |
| static void io_rings_free(struct io_ring_ctx *ctx) |
| { |
| if (!(ctx->flags & IORING_SETUP_NO_MMAP)) { |
| io_pages_unmap(ctx->rings, &ctx->ring_pages, &ctx->n_ring_pages, |
| true); |
| io_pages_unmap(ctx->sq_sqes, &ctx->sqe_pages, &ctx->n_sqe_pages, |
| true); |
| } else { |
| io_pages_free(&ctx->ring_pages, ctx->n_ring_pages); |
| ctx->n_ring_pages = 0; |
| io_pages_free(&ctx->sqe_pages, ctx->n_sqe_pages); |
| ctx->n_sqe_pages = 0; |
| vunmap(ctx->rings); |
| vunmap(ctx->sq_sqes); |
| } |
| |
| ctx->rings = NULL; |
| ctx->sq_sqes = NULL; |
| } |
| |
| static unsigned long rings_size(struct io_ring_ctx *ctx, unsigned int sq_entries, |
| unsigned int cq_entries, size_t *sq_offset) |
| { |
| struct io_rings *rings; |
| size_t off, sq_array_size; |
| |
| off = struct_size(rings, cqes, cq_entries); |
| if (off == SIZE_MAX) |
| return SIZE_MAX; |
| if (ctx->flags & IORING_SETUP_CQE32) { |
| if (check_shl_overflow(off, 1, &off)) |
| return SIZE_MAX; |
| } |
| |
| #ifdef CONFIG_SMP |
| off = ALIGN(off, SMP_CACHE_BYTES); |
| if (off == 0) |
| return SIZE_MAX; |
| #endif |
| |
| if (ctx->flags & IORING_SETUP_NO_SQARRAY) { |
| *sq_offset = SIZE_MAX; |
| return off; |
| } |
| |
| *sq_offset = off; |
| |
| sq_array_size = array_size(sizeof(u32), sq_entries); |
| if (sq_array_size == SIZE_MAX) |
| return SIZE_MAX; |
| |
| if (check_add_overflow(off, sq_array_size, &off)) |
| return SIZE_MAX; |
| |
| return off; |
| } |
| |
| static void io_req_caches_free(struct io_ring_ctx *ctx) |
| { |
| struct io_kiocb *req; |
| int nr = 0; |
| |
| mutex_lock(&ctx->uring_lock); |
| |
| while (!io_req_cache_empty(ctx)) { |
| req = io_extract_req(ctx); |
| kmem_cache_free(req_cachep, req); |
| nr++; |
| } |
| if (nr) |
| percpu_ref_put_many(&ctx->refs, nr); |
| mutex_unlock(&ctx->uring_lock); |
| } |
| |
| static __cold void io_ring_ctx_free(struct io_ring_ctx *ctx) |
| { |
| io_sq_thread_finish(ctx); |
| /* __io_rsrc_put_work() may need uring_lock to progress, wait w/o it */ |
| if (WARN_ON_ONCE(!list_empty(&ctx->rsrc_ref_list))) |
| return; |
| |
| mutex_lock(&ctx->uring_lock); |
| if (ctx->buf_data) |
| __io_sqe_buffers_unregister(ctx); |
| if (ctx->file_data) |
| __io_sqe_files_unregister(ctx); |
| io_cqring_overflow_kill(ctx); |
| io_eventfd_unregister(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->uring_cache, kfree); |
| io_alloc_cache_free(&ctx->msg_cache, io_msg_cache_free); |
| io_futex_cache_free(ctx); |
| io_destroy_buffers(ctx); |
| mutex_unlock(&ctx->uring_lock); |
| if (ctx->sq_creds) |
| put_cred(ctx->sq_creds); |
| if (ctx->submitter_task) |
| put_task_struct(ctx->submitter_task); |
| |
| /* there are no registered resources left, nobody uses it */ |
| if (ctx->rsrc_node) |
| io_rsrc_node_destroy(ctx, ctx->rsrc_node); |
| |
| WARN_ON_ONCE(!list_empty(&ctx->rsrc_ref_list)); |
| WARN_ON_ONCE(!list_empty(&ctx->ltimeout_list)); |
| |
| io_alloc_cache_free(&ctx->rsrc_node_cache, kfree); |
| if (ctx->mm_account) { |
| mmdrop(ctx->mm_account); |
| ctx->mm_account = NULL; |
| } |
| io_rings_free(ctx); |
| |
| percpu_ref_exit(&ctx->refs); |
| free_uid(ctx->user); |
| io_req_caches_free(ctx); |
| if (ctx->hash_map) |
| io_wq_put_hash(ctx->hash_map); |
| io_napi_free(ctx); |
| kfree(ctx->cancel_table.hbs); |
| kfree(ctx->cancel_table_locked.hbs); |
| xa_destroy(&ctx->io_bl_xa); |
| kfree(ctx); |
| } |
| |
| static __cold void io_activate_pollwq_cb(struct callback_head *cb) |
| { |
| struct io_ring_ctx *ctx = container_of(cb, struct io_ring_ctx, |
| poll_wq_task_work); |
| |
| mutex_lock(&ctx->uring_lock); |
| ctx->poll_activated = true; |
| mutex_unlock(&ctx->uring_lock); |
| |
| /* |
| * Wake ups for some events between start of polling and activation |
| * might've been lost due to loose synchronisation. |
| */ |
| wake_up_all(&ctx->poll_wq); |
| percpu_ref_put(&ctx->refs); |
| } |
| |
| __cold void io_activate_pollwq(struct io_ring_ctx *ctx) |
| { |
| spin_lock(&ctx->completion_lock); |
| /* already activated or in progress */ |
| if (ctx->poll_activated || ctx->poll_wq_task_work.func) |
| goto out; |
| if (WARN_ON_ONCE(!ctx->task_complete)) |
| goto out; |
| if (!ctx->submitter_task) |
| goto out; |
| /* |
| * with ->submitter_task only the submitter task completes requests, we |
| * only need to sync with it, which is done by injecting a tw |
| */ |
| init_task_work(&ctx->poll_wq_task_work, io_activate_pollwq_cb); |
| percpu_ref_get(&ctx->refs); |
| if (task_work_add(ctx->submitter_task, &ctx->poll_wq_task_work, TWA_SIGNAL)) |
| percpu_ref_put(&ctx->refs); |
| out: |
| spin_unlock(&ctx->completion_lock); |
| } |
| |
| static __poll_t io_uring_poll(struct file *file, poll_table *wait) |
| { |
| struct io_ring_ctx *ctx = file->private_data; |
| __poll_t mask = 0; |
| |
| if (unlikely(!ctx->poll_activated)) |
| io_activate_pollwq(ctx); |
| |
| poll_wait(file, &ctx->poll_wq, wait); |
| /* |
| * synchronizes with barrier from wq_has_sleeper call in |
| * io_commit_cqring |
| */ |
| smp_rmb(); |
| if (!io_sqring_full(ctx)) |
| mask |= EPOLLOUT | EPOLLWRNORM; |
| |
| /* |
| * Don't flush cqring overflow list here, just do a simple check. |
| * Otherwise there could possible be ABBA deadlock: |
| * CPU0 CPU1 |
| * ---- ---- |
| * lock(&ctx->uring_lock); |
| * lock(&ep->mtx); |
| * lock(&ctx->uring_lock); |
| * lock(&ep->mtx); |
| * |
| * Users may get EPOLLIN meanwhile seeing nothing in cqring, this |
| * pushes them to do the flush. |
| */ |
| |
| if (__io_cqring_events_user(ctx) || io_has_work(ctx)) |
| mask |= EPOLLIN | EPOLLRDNORM; |
| |
| return mask; |
| } |
| |
| struct io_tctx_exit { |
| struct callback_head task_work; |
| struct completion completion; |
| struct io_ring_ctx *ctx; |
| }; |
| |
| static __cold void io_tctx_exit_cb(struct callback_head *cb) |
| { |
| struct io_uring_task *tctx = current->io_uring; |
| struct io_tctx_exit *work; |
| |
| work = container_of(cb, struct io_tctx_exit, task_work); |
| /* |
| * When @in_cancel, we're in cancellation and it's racy to remove the |
| * node. It'll be removed by the end of cancellation, just ignore it. |
| * tctx can be NULL if the queueing of this task_work raced with |
| * work cancelation off the exec path. |
| */ |
| if (tctx && !atomic_read(&tctx->in_cancel)) |
| io_uring_del_tctx_node((unsigned long)work->ctx); |
| complete(&work->completion); |
| } |
| |
| static __cold bool io_cancel_ctx_cb(struct io_wq_work *work, void *data) |
| { |
| struct io_kiocb *req = container_of(work, struct io_kiocb, work); |
| |
| return req->ctx == data; |
| } |
| |
| static __cold void io_ring_exit_work(struct work_struct *work) |
| { |
| struct io_ring_ctx *ctx = container_of(work, struct io_ring_ctx, exit_work); |
| unsigned long timeout = jiffies + HZ * 60 * 5; |
| unsigned long interval = HZ / 20; |
| struct io_tctx_exit exit; |
| struct io_tctx_node *node; |
| int ret; |
| |
| /* |
| * If we're doing polled IO and end up having requests being |
| * submitted async (out-of-line), then completions can come in while |
| * we're waiting for refs to drop. We need to reap these manually, |
| * as nobody else will be looking for them. |
| */ |
| do { |
| if (test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq)) { |
| mutex_lock(&ctx->uring_lock); |
| io_cqring_overflow_kill(ctx); |
| mutex_unlock(&ctx->uring_lock); |
| } |
| |
| if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) |
| io_move_task_work_from_local(ctx); |
| |
| while (io_uring_try_cancel_requests(ctx, NULL, true)) |
| cond_resched(); |
| |
| if (ctx->sq_data) { |
| struct io_sq_data *sqd = ctx->sq_data; |
| struct task_struct *tsk; |
| |
| io_sq_thread_park(sqd); |
| tsk = sqd->thread; |
| if (tsk && tsk->io_uring && tsk->io_uring->io_wq) |
| io_wq_cancel_cb(tsk->io_uring->io_wq, |
| io_cancel_ctx_cb, ctx, true); |
| io_sq_thread_unpark(sqd); |
| } |
| |
| io_req_caches_free(ctx); |
| |
| if (WARN_ON_ONCE(time_after(jiffies, timeout))) { |
| /* there is little hope left, don't run it too often */ |
| interval = HZ * 60; |
| } |
| /* |
| * This is really an uninterruptible wait, as it has to be |
| * complete. But it's also run from a kworker, which doesn't |
| * take signals, so it's fine to make it interruptible. This |
| * avoids scenarios where we knowingly can wait much longer |
| * on completions, for example if someone does a SIGSTOP on |
| * a task that needs to finish task_work to make this loop |
| * complete. That's a synthetic situation that should not |
| * cause a stuck task backtrace, and hence a potential panic |
| * on stuck tasks if that is enabled. |
| */ |
| } while (!wait_for_completion_interruptible_timeout(&ctx->ref_comp, interval)); |
| |
| init_completion(&exit.completion); |
| init_task_work(&exit.task_work, io_tctx_exit_cb); |
| exit.ctx = ctx; |
| |
| mutex_lock(&ctx->uring_lock); |
| while (!list_empty(&ctx->tctx_list)) { |
| WARN_ON_ONCE(time_after(jiffies, timeout)); |
| |
| node = list_first_entry(&ctx->tctx_list, struct io_tctx_node, |
| ctx_node); |
| /* don't spin on a single task if cancellation failed */ |
| list_rotate_left(&ctx->tctx_list); |
| ret = task_work_add(node->task, &exit.task_work, TWA_SIGNAL); |
| if (WARN_ON_ONCE(ret)) |
| continue; |
| |
| mutex_unlock(&ctx->uring_lock); |
| /* |
| * See comment above for |
| * wait_for_completion_interruptible_timeout() on why this |
| * wait is marked as interruptible. |
| */ |
| wait_for_completion_interruptible(&exit.completion); |
| mutex_lock(&ctx->uring_lock); |
| } |
| mutex_unlock(&ctx->uring_lock); |
| spin_lock(&ctx->completion_lock); |
| spin_unlock(&ctx->completion_lock); |
| |
| /* pairs with RCU read section in io_req_local_work_add() */ |
| if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) |
| synchronize_rcu(); |
| |
| io_ring_ctx_free(ctx); |
| } |
| |
| static __cold void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx) |
| { |
| unsigned long index; |
| struct creds *creds; |
| |
| mutex_lock(&ctx->uring_lock); |
| percpu_ref_kill(&ctx->refs); |
| xa_for_each(&ctx->personalities, index, creds) |
| io_unregister_personality(ctx, index); |
| mutex_unlock(&ctx->uring_lock); |
| |
| flush_delayed_work(&ctx->fallback_work); |
| |
| INIT_WORK(&ctx->exit_work, io_ring_exit_work); |
| /* |
| * Use system_unbound_wq to avoid spawning tons of event kworkers |
| * if we're exiting a ton of rings at the same time. It just adds |
| * noise and overhead, there's no discernable change in runtime |
| * over using system_wq. |
| */ |
| queue_work(iou_wq, &ctx->exit_work); |
| } |
| |
| static int io_uring_release(struct inode *inode, struct file *file) |
| { |
| struct io_ring_ctx *ctx = file->private_data; |
| |
| file->private_data = NULL; |
| io_ring_ctx_wait_and_kill(ctx); |
| return 0; |
| } |
| |
| struct io_task_cancel { |
| struct task_struct *task; |
| bool all; |
| }; |
| |
| static bool io_cancel_task_cb(struct io_wq_work *work, void *data) |
| { |
| struct io_kiocb *req = container_of(work, struct io_kiocb, work); |
| struct io_task_cancel *cancel = data; |
| |
| return io_match_task_safe(req, cancel->task, cancel->all); |
| } |
| |
| static __cold bool io_cancel_defer_files(struct io_ring_ctx *ctx, |
| struct task_struct *task, |
| bool cancel_all) |
| { |
| struct io_defer_entry *de; |
| LIST_HEAD(list); |
| |
| spin_lock(&ctx->completion_lock); |
| list_for_each_entry_reverse(de, &ctx->defer_list, list) { |
| if (io_match_task_safe(de->req, task, cancel_all)) { |
| list_cut_position(&list, &ctx->defer_list, &de->list); |
| break; |
| } |
| } |
| spin_unlock(&ctx->completion_lock); |
| if (list_empty(&list)) |
| return false; |
| |
| while (!list_empty(&list)) { |
| de = list_first_entry(&list, struct io_defer_entry, list); |
| list_del_init(&de->list); |
| io_req_task_queue_fail(de->req, -ECANCELED); |
| kfree(de); |
| } |
| return true; |
| } |
| |
| static __cold bool io_uring_try_cancel_iowq(struct io_ring_ctx *ctx) |
| { |
| struct io_tctx_node *node; |
| enum io_wq_cancel cret; |
| bool ret = false; |
| |
| mutex_lock(&ctx->uring_lock); |
| list_for_each_entry(node, &ctx->tctx_list, ctx_node) { |
| struct io_uring_task *tctx = node->task->io_uring; |
| |
| /* |
| * io_wq will stay alive while we hold uring_lock, because it's |
| * killed after ctx nodes, which requires to take the lock. |
| */ |
| if (!tctx || !tctx->io_wq) |
| continue; |
| cret = io_wq_cancel_cb(tctx->io_wq, io_cancel_ctx_cb, ctx, true); |
| ret |= (cret != IO_WQ_CANCEL_NOTFOUND); |
| } |
| mutex_unlock(&ctx->uring_lock); |
| |
| return ret; |
| } |
| |
| static __cold bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx, |
| struct task_struct *task, |
| bool cancel_all) |
| { |
| struct io_task_cancel cancel = { .task = task, .all = cancel_all, }; |
| struct io_uring_task *tctx = task ? task->io_uring : NULL; |
| enum io_wq_cancel cret; |
| bool ret = false; |
| |
| /* set it so io_req_local_work_add() would wake us up */ |
| if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) { |
| atomic_set(&ctx->cq_wait_nr, 1); |
| smp_mb(); |
| } |
| |
| /* failed during ring init, it couldn't have issued any requests */ |
| if (!ctx->rings) |
| return false; |
| |
| if (!task) { |
| ret |= io_uring_try_cancel_iowq(ctx); |
| } else if (tctx && tctx->io_wq) { |
| /* |
| * Cancels requests of all rings, not only @ctx, but |
| * it's fine as the task is in exit/exec. |
| */ |
| cret = io_wq_cancel_cb(tctx->io_wq, io_cancel_task_cb, |
| &cancel, true); |
| ret |= (cret != IO_WQ_CANCEL_NOTFOUND); |
| } |
| |
| /* SQPOLL thread does its own polling */ |
| if ((!(ctx->flags & IORING_SETUP_SQPOLL) && cancel_all) || |
| (ctx->sq_data && ctx->sq_data->thread == current)) { |
| while (!wq_list_empty(&ctx->iopoll_list)) { |
| io_iopoll_try_reap_events(ctx); |
| ret = true; |
| cond_resched(); |
| } |
| } |
| |
| if ((ctx->flags & IORING_SETUP_DEFER_TASKRUN) && |
| io_allowed_defer_tw_run(ctx)) |
| ret |= io_run_local_work(ctx, INT_MAX) > 0; |
| ret |= io_cancel_defer_files(ctx, task, cancel_all); |
| mutex_lock(&ctx->uring_lock); |
| ret |= io_poll_remove_all(ctx, task, cancel_all); |
| ret |= io_waitid_remove_all(ctx, task, cancel_all); |
| ret |= io_futex_remove_all(ctx, task, cancel_all); |
| ret |= io_uring_try_cancel_uring_cmd(ctx, task, cancel_all); |
| mutex_unlock(&ctx->uring_lock); |
| ret |= io_kill_timeouts(ctx, task, cancel_all); |
| if (task) |
| ret |= io_run_task_work() > 0; |
| else |
| ret |= flush_delayed_work(&ctx->fallback_work); |
| return ret; |
| } |
| |
| static s64 tctx_inflight(struct io_uring_task *tctx, bool tracked) |
| { |
| if (tracked) |
| return atomic_read(&tctx->inflight_tracked); |
| return percpu_counter_sum(&tctx->inflight); |
| } |
| |
| /* |
| * Find any io_uring ctx that this task has registered or done IO on, and cancel |
| * requests. @sqd should be not-null IFF it's an SQPOLL thread cancellation. |
| */ |
| __cold void io_uring_cancel_generic(bool cancel_all, struct io_sq_data *sqd) |
| { |
| struct io_uring_task *tctx = current->io_uring; |
| struct io_ring_ctx *ctx; |
| struct io_tctx_node *node; |
| unsigned long index; |
| s64 inflight; |
| DEFINE_WAIT(wait); |
| |
| WARN_ON_ONCE(sqd && sqd->thread != current); |
| |
| if (!current->io_uring) |
| return; |
| if (tctx->io_wq) |
| io_wq_exit_start(tctx->io_wq); |
| |
| atomic_inc(&tctx->in_cancel); |
| do { |
| bool loop = false; |
| |
| io_uring_drop_tctx_refs(current); |
| if (!tctx_inflight(tctx, !cancel_all)) |
| break; |
| |
| /* read completions before cancelations */ |
| inflight = tctx_inflight(tctx, false); |
| if (!inflight) |
| break; |
| |
| if (!sqd) { |
| xa_for_each(&tctx->xa, index, node) { |
| /* sqpoll task will cancel all its requests */ |
| if (node->ctx->sq_data) |
| continue; |
| loop |= io_uring_try_cancel_requests(node->ctx, |
| current, cancel_all); |
| } |
| } else { |
| list_for_each_entry(ctx, &sqd->ctx_list, sqd_list) |
| loop |= io_uring_try_cancel_requests(ctx, |
| current, |
| cancel_all); |
| } |
| |
| if (loop) { |
| cond_resched(); |
| continue; |
| } |
| |
| prepare_to_wait(&tctx->wait, &wait, TASK_INTERRUPTIBLE); |
| io_run_task_work(); |
| io_uring_drop_tctx_refs(current); |
| xa_for_each(&tctx->xa, index, node) { |
| if (!llist_empty(&node->ctx->work_llist)) { |
| WARN_ON_ONCE(node->ctx->submitter_task && |
| node->ctx->submitter_task != current); |
| goto end_wait; |
| } |
| } |
| /* |
| * If we've seen completions, retry without waiting. This |
| * avoids a race where a completion comes in before we did |
| * prepare_to_wait(). |
| */ |
| if (inflight == tctx_inflight(tctx, !cancel_all)) |
| schedule(); |
| end_wait: |
| finish_wait(&tctx->wait, &wait); |
| } while (1); |
| |
| |