From c84701cfc90a90a6a9dfbdb138706a6d79f5b186 Mon Sep 17 00:00:00 2001 From: Yi Xie Date: Thu, 14 May 2026 16:34:43 +0800 Subject: [PATCH 01/31] io_uring: parenthesize io_ring_head_to_buf() expansion Wrap the io_ring_head_to_buf() macro value in an extra pair of parentheses so it is safe when composed into larger expressions, and to satisfy scripts/checkpatch.pl. Signed-off-by: Yi Xie Link: https://patch.msgid.link/20260514083443.203387-1-xieyi@kylinos.cn Signed-off-by: Jens Axboe --- io_uring/kbuf.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/io_uring/kbuf.c b/io_uring/kbuf.c index 63061aa1cab9..dd54e43e9ddf 100644 --- a/io_uring/kbuf.c +++ b/io_uring/kbuf.c @@ -21,7 +21,7 @@ #define MAX_BIDS_PER_BGID (1 << 16) /* Mapped buffer ring, return io_uring_buf from head */ -#define io_ring_head_to_buf(br, head, mask) &(br)->bufs[(head) & (mask)] +#define io_ring_head_to_buf(br, head, mask) (&(br)->bufs[(head) & (mask)]) struct io_provide_buf { struct file *file; From 899bea8248cee35d54760a5e7d61a76af8e64411 Mon Sep 17 00:00:00 2001 From: Shouvik Kar Date: Tue, 12 May 2026 16:32:42 +0530 Subject: [PATCH 02/31] io_uring/net: allow filtering on IORING_OP_CONNECT This adds custom filtering for IORING_OP_CONNECT, where the target family is always exposed, and (for AF_INET / AF_INET6) port and address are exposed. port and v4_addr are in network byte order so filter authors can compare against on-wire constants. Skip population unless addr_len covers the populated fields, to avoid leaking stale io_async_msghdr data on short connects. Signed-off-by: Shouvik Kar Link: https://patch.msgid.link/20260512110242.26219-1-auxcorelabs@gmail.com Signed-off-by: Jens Axboe --- include/uapi/linux/io_uring/bpf_filter.h | 16 +++++++++ io_uring/net.c | 41 ++++++++++++++++++++++++ io_uring/net.h | 7 ++++ io_uring/opdef.c | 2 ++ 4 files changed, 66 insertions(+) diff --git a/include/uapi/linux/io_uring/bpf_filter.h b/include/uapi/linux/io_uring/bpf_filter.h index 1b461d792a7b..ce7d78ab13b3 100644 --- a/include/uapi/linux/io_uring/bpf_filter.h +++ b/include/uapi/linux/io_uring/bpf_filter.h @@ -27,6 +27,22 @@ struct io_uring_bpf_ctx { __u64 mode; __u64 resolve; } open; + /* + * For CONNECT: fields are populated only when addr_len covers + * them; unpopulated fields are zero from the caller-side memset + * in io_uring_populate_bpf_ctx(). port and v4_addr are network + * byte order. Filters may only issue BPF_LD|BPF_W|BPF_ABS at + * 4-byte aligned offsets; load + mask for sub-word fields. + */ + struct { + __u32 family; /* sa_family_t zero-extended */ + __be16 port; + __u8 pad[2]; + union { + __be32 v4_addr; + __u8 v6_addr[16]; + }; + } connect; }; }; diff --git a/io_uring/net.c b/io_uring/net.c index 30cd22c0b934..cceb5c1409ca 100644 --- a/io_uring/net.c +++ b/io_uring/net.c @@ -1674,6 +1674,47 @@ void io_socket_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req) bctx->socket.protocol = sock->protocol; } +void io_connect_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req) +{ + struct io_connect *conn = io_kiocb_to_cmd(req, struct io_connect); + struct io_async_msghdr *iomsg = req->async_data; + struct sockaddr_storage *ss = &iomsg->addr; + + /* + * move_addr_to_kernel() skips the copy for addr_len == 0, so + * iomsg->addr may hold stale data from a prior CONNECT. Bail + * unless addr_len covers the family discriminator. + */ + if (conn->addr_len < (int)sizeof(sa_family_t)) + return; + + bctx->connect.family = ss->ss_family; + switch (ss->ss_family) { + case AF_INET: { + struct sockaddr_in *sin = (struct sockaddr_in *)ss; + + if (conn->addr_len < (int)sizeof(*sin)) + break; + bctx->connect.port = sin->sin_port; + bctx->connect.v4_addr = sin->sin_addr.s_addr; + break; + } + case AF_INET6: { + struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)ss; + + if (conn->addr_len < (int)sizeof(*sin6)) + break; + bctx->connect.port = sin6->sin6_port; + memcpy(bctx->connect.v6_addr, &sin6->sin6_addr, + sizeof(bctx->connect.v6_addr)); + break; + } + default: + /* family is set; per-family fields stay zero - family-only filtering */ + break; + } +} + int io_socket_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) { struct io_socket *sock = io_kiocb_to_cmd(req, struct io_socket); diff --git a/io_uring/net.h b/io_uring/net.h index d4d1ddce50e3..51fda715d3c0 100644 --- a/io_uring/net.h +++ b/io_uring/net.h @@ -46,6 +46,7 @@ int io_accept(struct io_kiocb *req, unsigned int issue_flags); int io_socket_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe); int io_socket(struct io_kiocb *req, unsigned int issue_flags); void io_socket_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req); +void io_connect_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req); int io_connect_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe); int io_connect(struct io_kiocb *req, unsigned int issue_flags); @@ -69,4 +70,10 @@ static inline void io_socket_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req) { } + +static inline void io_connect_bpf_populate(struct io_uring_bpf_ctx *bctx, + struct io_kiocb *req) +{ +} + #endif diff --git a/io_uring/opdef.c b/io_uring/opdef.c index c3ef52b70811..8ea6bd274607 100644 --- a/io_uring/opdef.c +++ b/io_uring/opdef.c @@ -203,9 +203,11 @@ const struct io_issue_def io_issue_defs[] = { .unbound_nonreg_file = 1, .pollout = 1, #if defined(CONFIG_NET) + .filter_pdu_size = sizeof_field(struct io_uring_bpf_ctx, connect), .async_size = sizeof(struct io_async_msghdr), .prep = io_connect_prep, .issue = io_connect, + .filter_populate = io_connect_bpf_populate, #else .prep = io_eopnotsupp_prep, #endif From df0a52537c0f95d5441eb0ba1bdbd2d864e6def9 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Sat, 24 Jan 2026 10:02:41 -0700 Subject: [PATCH 03/31] io_uring/rsrc: add huge page accounting for registered buffers Track huge page references in a per-ring xarray to prevent double accounting when the same huge page is used by multiple registered buffers, either within the same ring or across cloned rings. When registering buffers backed by huge pages, we need to account for RLIMIT_MEMLOCK. But if multiple buffers share the same huge page (common with cloned buffers), we must not account for the same page multiple times. Similarly, we must only unaccount when the last reference to a huge page is released. Maintain a per-ring xarray (hpage_acct) that tracks reference counts for each huge page. When registering a buffer, for each unique huge page, increment its accounting reference count, and only account pages that are newly added. When unregistering a buffer, for each unique huge page, decrement its refcount. Once the refcount hits zero, the page is unaccounted. Note: any account is done against the ctx->user that was assigned when the ring was setup. As before, if root is running the operation, no accounting is done. With these changes, any use of imu->acct_pages is also dead, hence kill it from struct io_mapped_ubuf. This shrinks it from 56b to 48b on a 64-bit arch. Additionally, hpage_already_acct() is gone, which was an O(M*M) scan over current + previous registrations. Signed-off-by: Jens Axboe --- include/linux/io_uring_types.h | 3 + io_uring/io_uring.c | 3 + io_uring/rsrc.c | 264 +++++++++++++++++++++++++-------- io_uring/rsrc.h | 1 - 4 files changed, 208 insertions(+), 63 deletions(-) diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h index 244392026c6d..23b8891d5704 100644 --- a/include/linux/io_uring_types.h +++ b/include/linux/io_uring_types.h @@ -446,6 +446,9 @@ struct io_ring_ctx { /* Stores zcrx object pointers of type struct io_zcrx_ifq */ struct xarray zcrx_ctxs; + /* Used for accounting references on pages in registered buffers */ + struct xarray hpage_acct; + u32 pers_next; struct xarray personalities; diff --git a/io_uring/io_uring.c b/io_uring/io_uring.c index 4ed998d60c09..fb6ed52bae61 100644 --- a/io_uring/io_uring.c +++ b/io_uring/io_uring.c @@ -233,6 +233,7 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p) 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 @@ -302,6 +303,7 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p) 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; } @@ -2198,6 +2200,7 @@ static __cold void io_ring_ctx_free(struct io_ring_ctx *ctx) io_napi_free(ctx); kvfree(ctx->cancel_table.hbs); xa_destroy(&ctx->io_bl_xa); + xa_destroy(&ctx->hpage_acct); kfree(ctx); } diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c index 650303626be6..be7c5bf4e161 100644 --- a/io_uring/rsrc.c +++ b/io_uring/rsrc.c @@ -28,7 +28,52 @@ struct io_rsrc_update { }; static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx, - struct iovec *iov, struct page **last_hpage); + struct iovec *iov); + +static int hpage_acct_ref(struct io_ring_ctx *ctx, struct page *hpage, + bool *acct_new) +{ + unsigned long key = (unsigned long) hpage; + unsigned long count; + void *entry; + int ret; + + lockdep_assert_held(&ctx->uring_lock); + + entry = xa_load(&ctx->hpage_acct, key); + if (entry) { + *acct_new = false; + count = xa_to_value(entry) + 1; + } else { + ret = xa_reserve(&ctx->hpage_acct, key, GFP_KERNEL_ACCOUNT); + if (ret) + return ret; + *acct_new = true; + count = 1; + } + xa_store(&ctx->hpage_acct, key, xa_mk_value(count), GFP_KERNEL_ACCOUNT); + return 0; +} + +static bool hpage_acct_unref(struct io_ring_ctx *ctx, struct page *hpage) +{ + unsigned long key = (unsigned long) hpage; + unsigned long count; + void *entry; + + lockdep_assert_held(&ctx->uring_lock); + + entry = xa_load(&ctx->hpage_acct, key); + if (WARN_ON_ONCE(!entry)) + return false; + count = xa_to_value(entry); + if (count == 1) { + xa_erase(&ctx->hpage_acct, key); + return true; + } + xa_store(&ctx->hpage_acct, key, xa_mk_value(count - 1), GFP_KERNEL_ACCOUNT); + return false; +} /* only define max */ #define IORING_MAX_FIXED_FILES (1U << 20) @@ -124,15 +169,53 @@ static void io_free_imu(struct io_ring_ctx *ctx, struct io_mapped_ubuf *imu) kvfree(imu); } +static unsigned long io_buffer_unaccount_pages(struct io_ring_ctx *ctx, + struct io_mapped_ubuf *imu) +{ + struct page *seen = NULL; + unsigned long acct = 0; + int i; + + if (imu->flags & IO_REGBUF_F_KBUF || !ctx->user) + return 0; + + for (i = 0; i < imu->nr_bvecs; i++) { + struct page *page = imu->bvec[i].bv_page; + struct page *hpage; + + if (!PageCompound(page)) { + acct++; + continue; + } + + hpage = compound_head(page); + if (hpage == seen) + continue; + seen = hpage; + + /* Unaccount on last reference */ + if (hpage_acct_unref(ctx, hpage)) + acct += page_size(hpage) >> PAGE_SHIFT; + cond_resched(); + } + + return acct; +} + static void io_buffer_unmap(struct io_ring_ctx *ctx, struct io_mapped_ubuf *imu) { + unsigned long acct_pages = 0; + + /* Always decrement, so it works for cloned buffers too */ + acct_pages = io_buffer_unaccount_pages(ctx, imu); + if (unlikely(refcount_read(&imu->refs) > 1)) { if (!refcount_dec_and_test(&imu->refs)) return; } - if (imu->acct_pages) - io_unaccount_mem(ctx->user, ctx->mm_account, imu->acct_pages); + if (acct_pages) + io_unaccount_mem(ctx->user, ctx->mm_account, acct_pages); imu->release(imu->priv); io_free_imu(ctx, imu); } @@ -282,7 +365,6 @@ static int __io_sqe_buffers_update(struct io_ring_ctx *ctx, { u64 __user *tags = u64_to_user_ptr(up->tags); struct iovec fast_iov, *iov; - struct page *last_hpage = NULL; struct iovec __user *uvec; u64 user_data = up->data; __u32 done; @@ -307,7 +389,7 @@ static int __io_sqe_buffers_update(struct io_ring_ctx *ctx, err = -EFAULT; break; } - node = io_sqe_buffer_register(ctx, iov, &last_hpage); + node = io_sqe_buffer_register(ctx, iov); if (IS_ERR(node)) { err = PTR_ERR(node); break; @@ -605,76 +687,79 @@ int io_sqe_buffers_unregister(struct io_ring_ctx *ctx) } /* - * Not super efficient, but this is just a registration time. And we do cache - * the last compound head, so generally we'll only do a full search if we don't - * match that one. - * - * We check if the given compound head page has already been accounted, to - * avoid double accounting it. This allows us to account the full size of the - * page, not just the constituent pages of a huge page. + * Undo hpage_acct_ref() calls made during io_buffer_account_pin() on failure. + * This operates on the pages array since imu->bvec isn't populated yet. */ -static bool headpage_already_acct(struct io_ring_ctx *ctx, struct page **pages, - int nr_pages, struct page *hpage) +static void io_buffer_unaccount_hpages(struct io_ring_ctx *ctx, + struct page **pages, int nr_pages) { - int i, j; + struct page *seen = NULL; + int i; + + if (!ctx->user) + return; - /* check current page array */ for (i = 0; i < nr_pages; i++) { + struct page *hpage; + if (!PageCompound(pages[i])) continue; - if (compound_head(pages[i]) == hpage) - return true; - } - /* check previously registered pages */ - for (i = 0; i < ctx->buf_table.nr; i++) { - struct io_rsrc_node *node = ctx->buf_table.nodes[i]; - struct io_mapped_ubuf *imu; - - if (!node) + hpage = compound_head(pages[i]); + if (hpage == seen) continue; - imu = node->buf; - for (j = 0; j < imu->nr_bvecs; j++) { - if (!PageCompound(imu->bvec[j].bv_page)) - continue; - if (compound_head(imu->bvec[j].bv_page) == hpage) - return true; - } - } + seen = hpage; - return false; + hpage_acct_unref(ctx, hpage); + cond_resched(); + } } static int io_buffer_account_pin(struct io_ring_ctx *ctx, struct page **pages, - int nr_pages, struct io_mapped_ubuf *imu, - struct page **last_hpage) + int nr_pages) { + unsigned long acct_pages = 0; + struct page *seen = NULL; int i, ret; - imu->acct_pages = 0; - for (i = 0; i < nr_pages; i++) { - if (!PageCompound(pages[i])) { - imu->acct_pages++; - } else { - struct page *hpage; + if (!ctx->user) + return 0; - hpage = compound_head(pages[i]); - if (hpage == *last_hpage) - continue; - *last_hpage = hpage; - if (headpage_already_acct(ctx, pages, i, hpage)) - continue; - imu->acct_pages += page_size(hpage) >> PAGE_SHIFT; + for (i = 0; i < nr_pages; i++) { + struct page *hpage; + bool acct_new; + + if (!PageCompound(pages[i])) { + acct_pages++; + continue; + } + + hpage = compound_head(pages[i]); + if (hpage == seen) + continue; + seen = hpage; + + ret = hpage_acct_ref(ctx, hpage, &acct_new); + if (ret) { + io_buffer_unaccount_hpages(ctx, pages, i); + return ret; + } + if (acct_new) + acct_pages += page_size(hpage) >> PAGE_SHIFT; + cond_resched(); + } + + /* Try to account the memory */ + if (acct_pages) { + ret = io_account_mem(ctx->user, ctx->mm_account, acct_pages); + if (ret) { + /* Undo the refs we just added */ + io_buffer_unaccount_hpages(ctx, pages, nr_pages); + return ret; } } - if (!imu->acct_pages) - return 0; - - ret = io_account_mem(ctx->user, ctx->mm_account, imu->acct_pages); - if (ret) - imu->acct_pages = 0; - return ret; + return 0; } static bool io_coalesce_buffer(struct page ***pages, int *nr_pages, @@ -763,8 +848,7 @@ bool io_check_coalesce_buffer(struct page **page_array, int nr_pages, } static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx, - struct iovec *iov, - struct page **last_hpage) + struct iovec *iov) { struct io_mapped_ubuf *imu = NULL; struct page **pages = NULL; @@ -811,7 +895,7 @@ static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx, goto done; imu->nr_bvecs = nr_pages; - ret = io_buffer_account_pin(ctx, pages, nr_pages, imu, last_hpage); + ret = io_buffer_account_pin(ctx, pages, nr_pages); if (ret) goto done; @@ -861,7 +945,6 @@ static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx, int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg, unsigned int nr_args, u64 __user *tags) { - struct page *last_hpage = NULL; struct io_rsrc_data data; struct iovec fast_iov, *iov = &fast_iov; const struct iovec __user *uvec; @@ -904,7 +987,7 @@ int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg, } } - node = io_sqe_buffer_register(ctx, iov, &last_hpage); + node = io_sqe_buffer_register(ctx, iov); if (IS_ERR(node)) { ret = PTR_ERR(node); break; @@ -971,7 +1054,6 @@ int io_buffer_register_bvec(struct io_uring_cmd *cmd, struct request *rq, imu->ubuf = 0; imu->len = blk_rq_bytes(rq); - imu->acct_pages = 0; imu->folio_shift = PAGE_SHIFT; refcount_set(&imu->refs, 1); imu->release = release; @@ -1136,6 +1218,56 @@ int io_import_reg_buf(struct io_kiocb *req, struct iov_iter *iter, return io_import_fixed(ddir, iter, node->buf, buf_addr, len); } +static int io_buffer_acct_cloned_hpages(struct io_ring_ctx *ctx, + struct io_mapped_ubuf *imu) +{ + struct page *seen = NULL; + int i, ret = 0; + + if (imu->flags & IO_REGBUF_F_KBUF || !ctx->user) + return 0; + + for (i = 0; i < imu->nr_bvecs; i++) { + struct page *page = imu->bvec[i].bv_page; + struct page *hpage; + bool acct_new; + + if (!PageCompound(page)) + continue; + + hpage = compound_head(page); + if (hpage == seen) + continue; + seen = hpage; + + /* Atomically add reference for cloned buffer */ + ret = hpage_acct_ref(ctx, hpage, &acct_new); + if (ret) + break; + + cond_resched(); + } + + if (!ret) + return 0; + + /* Undo refs we added for bvecs [0..i) */ + seen = NULL; + for (int j = 0; j < i; j++) { + struct page *p = imu->bvec[j].bv_page; + struct page *hp; + + if (!PageCompound(p)) + continue; + hp = compound_head(p); + if (hp == seen) + continue; + seen = hp; + hpage_acct_unref(ctx, hp); + } + return ret; +} + /* Lock two rings at once. The rings must be different! */ static void lock_two_rings(struct io_ring_ctx *ctx1, struct io_ring_ctx *ctx2) { @@ -1218,6 +1350,14 @@ static int io_clone_buffers(struct io_ring_ctx *ctx, struct io_ring_ctx *src_ctx refcount_inc(&src_node->buf->refs); dst_node->buf = src_node->buf; + /* track compound references to clones */ + ret = io_buffer_acct_cloned_hpages(ctx, src_node->buf); + if (ret) { + refcount_dec(&src_node->buf->refs); + io_cache_free(&ctx->node_cache, dst_node); + io_rsrc_data_free(ctx, &data); + return ret; + } } data.nodes[off++] = dst_node; i++; diff --git a/io_uring/rsrc.h b/io_uring/rsrc.h index 44e3386f7c1c..c0f8a18ec767 100644 --- a/io_uring/rsrc.h +++ b/io_uring/rsrc.h @@ -38,7 +38,6 @@ struct io_mapped_ubuf { unsigned int nr_bvecs; unsigned int folio_shift; refcount_t refs; - unsigned long acct_pages; void (*release)(void *); void *priv; u8 flags; From ca76b56a2a2acf9875e5cad68612085f25b463bb Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Mon, 4 May 2026 05:40:16 -0600 Subject: [PATCH 04/31] io_uring/rsrc: bump struct io_mapped_ubuf length field to size_t In preparation for supporting bigger individual buffers, bump the length field to a full 8-bytes with size_t rather than an unsigned int. Signed-off-by: Jens Axboe --- io_uring/fdinfo.c | 2 +- io_uring/rsrc.h | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/io_uring/fdinfo.c b/io_uring/fdinfo.c index c2d3e45544bb..f0ff4bd01b6d 100644 --- a/io_uring/fdinfo.c +++ b/io_uring/fdinfo.c @@ -223,7 +223,7 @@ static void __io_uring_show_fdinfo(struct io_ring_ctx *ctx, struct seq_file *m) if (ctx->buf_table.nodes[i]) buf = ctx->buf_table.nodes[i]->buf; if (buf) - seq_printf(m, "%5u: 0x%llx/%u\n", i, buf->ubuf, buf->len); + seq_printf(m, "%5u: 0x%llx/%zu\n", i, buf->ubuf, buf->len); else seq_printf(m, "%5u: \n", i); } diff --git a/io_uring/rsrc.h b/io_uring/rsrc.h index c0f8a18ec767..98ae8ef51009 100644 --- a/io_uring/rsrc.h +++ b/io_uring/rsrc.h @@ -34,14 +34,14 @@ enum { struct io_mapped_ubuf { u64 ubuf; - unsigned int len; + size_t len; unsigned int nr_bvecs; unsigned int folio_shift; refcount_t refs; - void (*release)(void *); - void *priv; u8 flags; u8 dir; + void (*release)(void *); + void *priv; struct bio_vec bvec[] __counted_by(nr_bvecs); }; From b4e41050b212ca33c82fb0598a7b323d5b18f1bb Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Mon, 4 May 2026 05:42:51 -0600 Subject: [PATCH 05/31] io_uring/rsrc: raise registered buffer 1GB limit There's no real reason to have a limit, as the memory is accounted by the lockmem limits anyway, if any exist. io_pin_pages() will still restrict the maximum allowed limit per buffer, which is INT_MAX number of pages. Cap it a bit lower than that, at 1TB for a 64-bit system. Surely that should be enough for everyone. For now. Signed-off-by: Jens Axboe --- io_uring/rsrc.c | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c index be7c5bf4e161..7f553d115e36 100644 --- a/io_uring/rsrc.c +++ b/io_uring/rsrc.c @@ -133,9 +133,14 @@ int io_validate_user_buf_range(u64 uaddr, u64 ulen) unsigned long tmp, base = (unsigned long)uaddr; unsigned long acct_len = (unsigned long)PAGE_ALIGN(ulen); - /* arbitrary limit, but we need something */ - if (ulen > SZ_1G || !ulen) + if (!ulen) return -EFAULT; + /* 32-bit sanity checking */ + if (ulen > ULONG_MAX || uaddr > ULONG_MAX) + return -EFAULT; + /* cap to 1TB for 64-bit */ + if (ulen > SZ_1T) + return -EINVAL; if (check_add_overflow(base, acct_len, &tmp)) return -EOVERFLOW; return 0; From 67ee1f021a9b74ef289934be5e7a474e20031add Mon Sep 17 00:00:00 2001 From: Vineeth Pillai Date: Fri, 15 May 2026 09:59:03 -0400 Subject: [PATCH 06/31] io_uring: Use trace_call__##name() at guarded tracepoint call sites Replace trace_foo() with the new trace_call__foo() at sites already guarded by trace_foo_enabled(), avoiding a redundant static_branch_unlikely() re-evaluation inside the tracepoint. trace_call__foo() calls the tracepoint callbacks directly without utilizing the static branch again. Suggested-by: Steven Rostedt Suggested-by: Peter Zijlstra Signed-off-by: Vineeth Pillai (Google) Assisted-by: Claude:claude-sonnet-4-6 Link: https://patch.msgid.link/20260515135903.2238731-1-vineeth@bitbyteword.org Signed-off-by: Jens Axboe --- io_uring/io_uring.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/io_uring/io_uring.h b/io_uring/io_uring.h index e612a66ee80e..1b657b714373 100644 --- a/io_uring/io_uring.h +++ b/io_uring/io_uring.h @@ -312,7 +312,7 @@ static __always_inline bool io_fill_cqe_req(struct io_ring_ctx *ctx, } if (trace_io_uring_complete_enabled()) - trace_io_uring_complete(req->ctx, req, cqe); + trace_call__io_uring_complete(req->ctx, req, cqe); return true; } From 74fc9a9b50d43ed473ea2449682000da43e17175 Mon Sep 17 00:00:00 2001 From: Pavel Begunkov Date: Tue, 19 May 2026 12:44:27 +0100 Subject: [PATCH 07/31] io_uring/zcrx: make scrubbing more reliable Currently, scrubbing is done once before killing all recvzc requests. It's fine as those are cancelled and don't return buffers afterwards, but it'll be more reliable not to rely that much on cancellations. Signed-off-by: Pavel Begunkov Link: https://patch.msgid.link/c4ea127023494cbbedebd21a2b7ae5ff0448eb95.1779189667.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- io_uring/zcrx.c | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index 19837e0b5e91..a7eef37868cf 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -985,6 +985,12 @@ void io_unregister_zcrx(struct io_ring_ctx *ctx) } if (!ifq) break; + /* + * io_uring can run requests and return buffers to the user + * after termination, scrub it again. + */ + if (refcount_read(&ifq->user_refs) == 0) + io_zcrx_scrub(ifq); io_put_zcrx_ifq(ifq); } From e57b44039bc54bbdf3d1511021458356858a4a12 Mon Sep 17 00:00:00 2001 From: Pavel Begunkov Date: Tue, 19 May 2026 12:44:28 +0100 Subject: [PATCH 08/31] io_uring/zcrx: poison pointers on unregistration Nobody should be touching area and other pointers after zcrx destruction, poison them instead of zeroing. Signed-off-by: Pavel Begunkov Link: https://patch.msgid.link/19112d1412539dcfc04a0317b5812e968623bc51.1779189667.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- io_uring/zcrx.c | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index a7eef37868cf..4a1aea317287 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -245,14 +245,13 @@ static void io_release_area_mem(struct io_zcrx_mem *mem) { if (mem->is_dmabuf) { io_release_dmabuf(mem); - return; - } - if (mem->pages) { + } else if (mem->pages) { unpin_user_pages(mem->pages, mem->nr_folios); sg_free_table(mem->sgt); - mem->sgt = NULL; kvfree(mem->pages); } + mem->pages = IO_URING_PTR_POISON; + mem->sgt = IO_URING_PTR_POISON; } static int io_import_area(struct io_zcrx_ifq *ifq, @@ -403,8 +402,8 @@ static int io_allocate_rbuf_ring(struct io_ring_ctx *ctx, static void io_free_rbuf_ring(struct io_zcrx_ifq *ifq) { io_free_region(ifq->user, &ifq->rq_region); - ifq->rq.ring = NULL; - ifq->rq.rqes = NULL; + ifq->rq.ring = IO_URING_PTR_POISON; + ifq->rq.rqes = IO_URING_PTR_POISON; } static void io_zcrx_free_area(struct io_zcrx_ifq *ifq, From 98f07b0f74b65284ebe0d021505b461d4be6bf07 Mon Sep 17 00:00:00 2001 From: Pavel Begunkov Date: Tue, 19 May 2026 12:44:29 +0100 Subject: [PATCH 09/31] io_uring/zcrx: remove extra ifq close By the time io_zcrx_ifq_free() is called the interface queue should already be closed, so io_close_queue() will be a no-op. Remove the call and add a couple of warnings. Signed-off-by: Pavel Begunkov Link: https://patch.msgid.link/be6c4a283a5bab5440e22fbccafe7b885acb7abc.1779189667.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- io_uring/zcrx.c | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index 4a1aea317287..d610e0f96844 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -574,7 +574,10 @@ static void io_close_queue(struct io_zcrx_ifq *ifq) static void io_zcrx_ifq_free(struct io_zcrx_ifq *ifq) { - io_close_queue(ifq); + if (WARN_ON_ONCE(ifq->if_rxq != -1)) + return; + if (WARN_ON_ONCE(ifq->netdev != NULL)) + return; if (ifq->area) io_zcrx_free_area(ifq, ifq->area); From 84f7d0931c42cb0690615a431738cf6913d265f2 Mon Sep 17 00:00:00 2001 From: Bertie Tryner Date: Tue, 19 May 2026 12:44:30 +0100 Subject: [PATCH 10/31] io_uring/zcrx: reorder fd allocation in zcrx_export() Currently, zcrx_export() allocates a file descriptor and copies the control structure to userspace before the backing file is created. While the operation returns an error on failure, it is cleaner to follow the standard kernel pattern of performing the copy_to_user() and fd_install() only after all resource allocations (like the anon_inode) have succeeded. This aligns the code with other fd-publishing paths in the VFS. Signed-off-by: Bertie Tryner Signed-off-by: Pavel Begunkov Link: https://patch.msgid.link/1513a3f4ae7161692ca6e991b9f01278a6bc60e4.1779189667.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- io_uring/zcrx.c | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index d610e0f96844..7ac9eb6a0934 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -698,19 +698,10 @@ static int zcrx_export(struct io_ring_ctx *ctx, struct io_zcrx_ifq *ifq, { struct zcrx_ctrl_export *ce = &ctrl->zc_export; struct file *file; - int fd = -1; + int fd; if (!mem_is_zero(ce, sizeof(*ce))) return -EINVAL; - fd = get_unused_fd_flags(O_CLOEXEC); - if (fd < 0) - return fd; - - ce->zcrx_fd = fd; - if (copy_to_user(arg, ctrl, sizeof(*ctrl))) { - put_unused_fd(fd); - return -EFAULT; - } refcount_inc(&ifq->refs); refcount_inc(&ifq->user_refs); @@ -718,11 +709,23 @@ static int zcrx_export(struct io_ring_ctx *ctx, struct io_zcrx_ifq *ifq, file = anon_inode_create_getfile("[zcrx]", &zcrx_box_fops, ifq, O_CLOEXEC, NULL); if (IS_ERR(file)) { - put_unused_fd(fd); zcrx_unregister(ifq); return PTR_ERR(file); } + fd = get_unused_fd_flags(O_CLOEXEC); + if (fd < 0) { + fput(file); + return fd; + } + + ce->zcrx_fd = fd; + if (copy_to_user(arg, ctrl, sizeof(*ctrl))) { + fput(file); + put_unused_fd(fd); + return -EFAULT; + } + fd_install(fd, file); return 0; } From 8503f2de11f7fe78a7fdb87746255c8d02897279 Mon Sep 17 00:00:00 2001 From: Pavel Begunkov Date: Tue, 19 May 2026 12:44:31 +0100 Subject: [PATCH 11/31] io_uring/zcrx: add ctx pointer to zcrx zcrx will need to have a pointer to an owning ctx to communicate different events. Reference the ctx while it's attached to zcrx, and rely on zcrx termination to drop the ctx to avoid circular ref deps. Co-developed-by: Vishwanath Seshagiri Signed-off-by: Pavel Begunkov Signed-off-by: Vishwanath Seshagiri Link: https://patch.msgid.link/b60514b3d1bd92f571e3bd91751166f8c3599256.1779189667.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- io_uring/zcrx.c | 39 +++++++++++++++++++++++++++++++-------- io_uring/zcrx.h | 3 +++ 2 files changed, 34 insertions(+), 8 deletions(-) diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index 7ac9eb6a0934..1e194de7d6d2 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -44,6 +44,17 @@ static inline struct io_zcrx_area *io_zcrx_iov_to_area(const struct net_iov *nio return container_of(owner, struct io_zcrx_area, nia); } +static bool zcrx_set_ring_ctx(struct io_zcrx_ifq *zcrx, + struct io_ring_ctx *ctx) +{ + guard(spinlock_bh)(&zcrx->ctx_lock); + if (zcrx->master_ctx) + return false; + percpu_ref_get(&ctx->refs); + zcrx->master_ctx = ctx; + return true; +} + static inline struct page *io_zcrx_iov_page(const struct net_iov *niov) { struct io_zcrx_area *area = io_zcrx_iov_to_area(niov); @@ -529,6 +540,7 @@ static struct io_zcrx_ifq *io_zcrx_ifq_alloc(struct io_ring_ctx *ctx) return NULL; ifq->if_rxq = -1; + spin_lock_init(&ifq->ctx_lock); spin_lock_init(&ifq->rq.lock); mutex_init(&ifq->pp_lock); refcount_set(&ifq->refs, 1); @@ -578,6 +590,8 @@ static void io_zcrx_ifq_free(struct io_zcrx_ifq *ifq) return; if (WARN_ON_ONCE(ifq->netdev != NULL)) return; + if (WARN_ON_ONCE(ifq->master_ctx)) + return; if (ifq->area) io_zcrx_free_area(ifq, ifq->area); @@ -654,17 +668,24 @@ static void io_zcrx_scrub(struct io_zcrx_ifq *ifq) } } -static void zcrx_unregister_user(struct io_zcrx_ifq *ifq) +static void zcrx_unregister_user(struct io_zcrx_ifq *ifq, struct io_ring_ctx *ctx) { + scoped_guard(spinlock_bh, &ifq->ctx_lock) { + if (ctx && ifq->master_ctx == ctx) { + ifq->master_ctx = NULL; + percpu_ref_put(&ctx->refs); + } + } + if (refcount_dec_and_test(&ifq->user_refs)) { io_close_queue(ifq); io_zcrx_scrub(ifq); } } -static void zcrx_unregister(struct io_zcrx_ifq *ifq) +static void zcrx_unregister(struct io_zcrx_ifq *ifq, struct io_ring_ctx *ctx) { - zcrx_unregister_user(ifq); + zcrx_unregister_user(ifq, ctx); io_put_zcrx_ifq(ifq); } @@ -684,7 +705,7 @@ static int zcrx_box_release(struct inode *inode, struct file *file) if (WARN_ON_ONCE(!ifq)) return -EFAULT; - zcrx_unregister(ifq); + zcrx_unregister(ifq, NULL); return 0; } @@ -709,7 +730,7 @@ static int zcrx_export(struct io_ring_ctx *ctx, struct io_zcrx_ifq *ifq, file = anon_inode_create_getfile("[zcrx]", &zcrx_box_fops, ifq, O_CLOEXEC, NULL); if (IS_ERR(file)) { - zcrx_unregister(ifq); + zcrx_unregister(ifq, NULL); return PTR_ERR(file); } @@ -785,7 +806,7 @@ static int import_zcrx(struct io_ring_ctx *ctx, scoped_guard(mutex, &ctx->mmap_lock) xa_erase(&ctx->zcrx_ctxs, id); err: - zcrx_unregister(ifq); + zcrx_unregister(ifq, ctx); return ret; } @@ -930,12 +951,14 @@ int io_register_zcrx(struct io_ring_ctx *ctx, ret = -EFAULT; goto err; } + + zcrx_set_ring_ctx(ifq, ctx); return 0; err: scoped_guard(mutex, &ctx->mmap_lock) xa_erase(&ctx->zcrx_ctxs, id); ifq_free: - zcrx_unregister(ifq); + zcrx_unregister(ifq, ctx); return ret; } @@ -965,7 +988,7 @@ void io_terminate_zcrx(struct io_ring_ctx *ctx) break; set_zcrx_entry_mark(ctx, id); id++; - zcrx_unregister_user(ifq); + zcrx_unregister_user(ifq, ctx); } } diff --git a/io_uring/zcrx.h b/io_uring/zcrx.h index 75e0a4e6ef6e..76389a5dd50f 100644 --- a/io_uring/zcrx.h +++ b/io_uring/zcrx.h @@ -72,6 +72,9 @@ struct io_zcrx_ifq { */ struct mutex pp_lock; struct io_mapped_region rq_region; + + spinlock_t ctx_lock; + struct io_ring_ctx *master_ctx; }; #if defined(CONFIG_IO_URING_ZCRX) From 0719e10d826aa0ba4840917d0261986eaead9a51 Mon Sep 17 00:00:00 2001 From: Pavel Begunkov Date: Tue, 19 May 2026 12:44:32 +0100 Subject: [PATCH 12/31] io_uring/zcrx: notify user when out of buffers There are currently no easy ways for the user to know if zcrx is out of buffers and page pool fails to allocate. Add uapi for zcrx to communicate it back. It's implemented as a separate CQE, which for now is posted to the creator ctx. To use it, on registration the user space needs to pass an instance of struct zcrx_notification_desc, which tells the kernel the user_data for resulting CQEs and which event types are expected / allowed. When an allowed event happens, zcrx will post a CQE containing the specified user_data, and lower bits of cqe->res will be set to the event mask. Before the kernel could post another notification of the given type, the user needs to acknowledge that it processed the previous one by issuing IORING_REGISTER_ZCRX_CTRL with ZCRX_CTRL_ARM_NOTIFICATION. The only notification type the patch implements is ZCRX_NOTIF_NO_BUFFERS, but we'll need more of them in the future. Co-developed-by: Vishwanath Seshagiri Signed-off-by: Pavel Begunkov Signed-off-by: Vishwanath Seshagiri Link: https://patch.msgid.link/35cd307a03a43583838a2e151fc641c69abd786f.1779189667.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- include/uapi/linux/io_uring/zcrx.h | 24 ++++++++- io_uring/io_uring.c | 2 +- io_uring/io_uring.h | 1 + io_uring/zcrx.c | 86 +++++++++++++++++++++++++++++- io_uring/zcrx.h | 7 ++- 5 files changed, 115 insertions(+), 5 deletions(-) diff --git a/include/uapi/linux/io_uring/zcrx.h b/include/uapi/linux/io_uring/zcrx.h index 5ce02c7a6096..67185566ad3c 100644 --- a/include/uapi/linux/io_uring/zcrx.h +++ b/include/uapi/linux/io_uring/zcrx.h @@ -65,6 +65,20 @@ enum zcrx_features { * value in struct io_uring_zcrx_ifq_reg::rx_buf_len. */ ZCRX_FEATURE_RX_PAGE_SIZE = 1 << 0, + ZCRX_FEATURE_NOTIFICATION = 1 << 1, +}; + +enum zcrx_notification_type { + ZCRX_NOTIF_NO_BUFFERS, + + __ZCRX_NOTIF_TYPE_LAST, +}; + +struct zcrx_notification_desc { + __u64 user_data; + __u32 type_mask; + __u32 __resv1; + __u64 __resv2[10]; }; /* @@ -82,12 +96,14 @@ struct io_uring_zcrx_ifq_reg { struct io_uring_zcrx_offsets offsets; __u32 zcrx_id; __u32 rx_buf_len; - __u64 __resv[3]; + __u64 notif_desc; /* see struct zcrx_notification_desc */ + __u64 __resv[2]; }; enum zcrx_ctrl_op { ZCRX_CTRL_FLUSH_RQ, ZCRX_CTRL_EXPORT, + ZCRX_CTRL_ARM_NOTIFICATION, __ZCRX_CTRL_LAST, }; @@ -101,6 +117,11 @@ struct zcrx_ctrl_export { __u32 __resv1[11]; }; +struct zcrx_ctrl_arm_notif { + __u32 notif_type; + __u32 __resv[11]; +}; + struct zcrx_ctrl { __u32 zcrx_id; __u32 op; /* see enum zcrx_ctrl_op */ @@ -109,6 +130,7 @@ struct zcrx_ctrl { union { struct zcrx_ctrl_export zc_export; struct zcrx_ctrl_flush_rq zc_flush; + struct zcrx_ctrl_arm_notif zc_arm_notif; }; }; diff --git a/io_uring/io_uring.c b/io_uring/io_uring.c index fb6ed52bae61..02c02e14f392 100644 --- a/io_uring/io_uring.c +++ b/io_uring/io_uring.c @@ -160,7 +160,7 @@ static void io_poison_cached_req(struct io_kiocb *req) req->apoll = IO_URING_PTR_POISON; } -static void io_poison_req(struct io_kiocb *req) +void io_poison_req(struct io_kiocb *req) { io_poison_cached_req(req); req->async_data = IO_URING_PTR_POISON; diff --git a/io_uring/io_uring.h b/io_uring/io_uring.h index 1b657b714373..cb736b815422 100644 --- a/io_uring/io_uring.h +++ b/io_uring/io_uring.h @@ -213,6 +213,7 @@ bool __io_alloc_req_refill(struct io_ring_ctx *ctx); void io_activate_pollwq(struct io_ring_ctx *ctx); void io_restriction_clone(struct io_restriction *dst, struct io_restriction *src); +void io_poison_req(struct io_kiocb *req); static inline void io_lockdep_assert_cq_locked(struct io_ring_ctx *ctx) { diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index 1e194de7d6d2..2bca9a58dcf6 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -766,6 +766,8 @@ static int import_zcrx(struct io_ring_ctx *ctx, return -EINVAL; if (reg->if_rxq || reg->rq_entries || reg->area_ptr || reg->region_ptr) return -EINVAL; + if (reg->notif_desc) + return -EINVAL; if (reg->flags & ~ZCRX_REG_IMPORT) return -EINVAL; @@ -854,6 +856,7 @@ static int zcrx_register_netdev(struct io_zcrx_ifq *ifq, int io_register_zcrx(struct io_ring_ctx *ctx, struct io_uring_zcrx_ifq_reg __user *arg) { + struct zcrx_notification_desc notif; struct io_uring_zcrx_area_reg area; struct io_uring_zcrx_ifq_reg reg; struct io_uring_region_desc rd; @@ -897,10 +900,22 @@ int io_register_zcrx(struct io_ring_ctx *ctx, if (copy_from_user(&area, u64_to_user_ptr(reg.area_ptr), sizeof(area))) return -EFAULT; + memset(¬if, 0, sizeof(notif)); + if (reg.notif_desc && copy_from_user(¬if, u64_to_user_ptr(reg.notif_desc), + sizeof(notif))) + return -EFAULT; + if (notif.type_mask & ~ZCRX_NOTIF_TYPE_MASK) + return -EINVAL; + if (notif.__resv1 || !mem_is_zero(¬if.__resv2, sizeof(notif.__resv2))) + return -EINVAL; + ifq = io_zcrx_ifq_alloc(ctx); if (!ifq) return -ENOMEM; + ifq->notif_data = notif.user_data; + ifq->allowed_notif_mask = notif.type_mask; + if (ctx->user) { get_uid(ctx->user); ifq->user = ctx->user; @@ -952,7 +967,8 @@ int io_register_zcrx(struct io_ring_ctx *ctx, goto err; } - zcrx_set_ring_ctx(ifq, ctx); + if (notif.type_mask) + zcrx_set_ring_ctx(ifq, ctx); return 0; err: scoped_guard(mutex, &ctx->mmap_lock) @@ -1125,6 +1141,48 @@ static unsigned io_zcrx_refill_slow(struct page_pool *pp, struct io_zcrx_ifq *if return allocated; } +static void zcrx_notif_tw(struct io_tw_req tw_req, io_tw_token_t tw) +{ + struct io_kiocb *req = tw_req.req; + struct io_ring_ctx *ctx = req->ctx; + + io_post_aux_cqe(ctx, req->cqe.user_data, req->cqe.res, 0); + percpu_ref_put(&ctx->refs); + io_poison_req(req); + kmem_cache_free(req_cachep, req); +} + +static void zcrx_send_notif(struct io_zcrx_ifq *ifq, unsigned type) +{ + gfp_t gfp = GFP_ATOMIC | __GFP_NOWARN | __GFP_ZERO; + u32 type_mask = 1 << type; + struct io_kiocb *req; + + if (!(type_mask & ifq->allowed_notif_mask)) + return; + + guard(spinlock_bh)(&ifq->ctx_lock); + if (!ifq->master_ctx) + return; + if (type_mask & ifq->fired_notifs) + return; + + req = kmem_cache_alloc(req_cachep, gfp); + if (unlikely(!req)) + return; + + ifq->fired_notifs |= type_mask; + + req->opcode = IORING_OP_NOP; + req->cqe.user_data = ifq->notif_data; + req->cqe.res = type; + req->ctx = ifq->master_ctx; + percpu_ref_get(&req->ctx->refs); + req->tctx = NULL; + req->io_task_work.func = zcrx_notif_tw; + io_req_task_work_add(req); +} + static netmem_ref io_pp_zc_alloc_netmems(struct page_pool *pp, gfp_t gfp) { struct io_zcrx_ifq *ifq = io_pp_to_ifq(pp); @@ -1141,8 +1199,10 @@ static netmem_ref io_pp_zc_alloc_netmems(struct page_pool *pp, gfp_t gfp) goto out_return; allocated = io_zcrx_refill_slow(pp, ifq, netmems, to_alloc); - if (!allocated) + if (!allocated) { + zcrx_send_notif(ifq, ZCRX_NOTIF_NO_BUFFERS); return 0; + } out_return: zcrx_sync_for_device(pp, ifq, netmems, allocated); allocated--; @@ -1291,12 +1351,32 @@ static int zcrx_flush_rq(struct io_ring_ctx *ctx, struct io_zcrx_ifq *zcrx, return 0; } +static int zcrx_arm_notif(struct io_ring_ctx *ctx, struct io_zcrx_ifq *zcrx, + struct zcrx_ctrl *ctrl) +{ + const struct zcrx_ctrl_arm_notif *an = &ctrl->zc_arm_notif; + unsigned type_mask; + + if (an->notif_type >= __ZCRX_NOTIF_TYPE_LAST) + return -EINVAL; + if (!mem_is_zero(&an->__resv, sizeof(an->__resv))) + return -EINVAL; + + guard(spinlock_bh)(&zcrx->ctx_lock); + type_mask = 1U << an->notif_type; + if (type_mask & ~zcrx->fired_notifs) + return -EINVAL; + zcrx->fired_notifs &= ~type_mask; + return 0; +} + int io_zcrx_ctrl(struct io_ring_ctx *ctx, void __user *arg, unsigned nr_args) { struct zcrx_ctrl ctrl; struct io_zcrx_ifq *zcrx; BUILD_BUG_ON(sizeof(ctrl.zc_export) != sizeof(ctrl.zc_flush)); + BUILD_BUG_ON(sizeof(ctrl.zc_export) != sizeof(ctrl.zc_arm_notif)); if (nr_args) return -EINVAL; @@ -1314,6 +1394,8 @@ int io_zcrx_ctrl(struct io_ring_ctx *ctx, void __user *arg, unsigned nr_args) return zcrx_flush_rq(ctx, zcrx, &ctrl); case ZCRX_CTRL_EXPORT: return zcrx_export(ctx, zcrx, &ctrl, arg); + case ZCRX_CTRL_ARM_NOTIFICATION: + return zcrx_arm_notif(ctx, zcrx, &ctrl); } return -EOPNOTSUPP; diff --git a/io_uring/zcrx.h b/io_uring/zcrx.h index 76389a5dd50f..e8b7717d6adf 100644 --- a/io_uring/zcrx.h +++ b/io_uring/zcrx.h @@ -9,7 +9,9 @@ #include #define ZCRX_SUPPORTED_REG_FLAGS (ZCRX_REG_IMPORT | ZCRX_REG_NODEV) -#define ZCRX_FEATURES (ZCRX_FEATURE_RX_PAGE_SIZE) +#define ZCRX_FEATURES (ZCRX_FEATURE_RX_PAGE_SIZE |\ + ZCRX_FEATURE_NOTIFICATION) +#define ZCRX_NOTIF_TYPE_MASK (1U << ZCRX_NOTIF_NO_BUFFERS) struct io_zcrx_mem { unsigned long size; @@ -75,6 +77,9 @@ struct io_zcrx_ifq { spinlock_t ctx_lock; struct io_ring_ctx *master_ctx; + u32 allowed_notif_mask; + u32 fired_notifs; + u64 notif_data; }; #if defined(CONFIG_IO_URING_ZCRX) From 255180f7034f48aa5b0c8df70228307394bddbb9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20L=C3=A9ger?= Date: Tue, 19 May 2026 12:44:33 +0100 Subject: [PATCH 13/31] io_uring/zcrx: notify user on frag copy fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a ZCRX_NOTIF_COPY notification type to signal userspace when a received fragment could not be delivered using zero-copy and was instead copied into a buffer. Signed-off-by: Clément Léger Signed-off-by: Pavel Begunkov Link: https://patch.msgid.link/3d54bcd8bf10b3a1e88beb0cd39c40c3937bea4f.1779189667.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- include/uapi/linux/io_uring/zcrx.h | 1 + io_uring/zcrx.c | 7 ++++++- io_uring/zcrx.h | 2 +- 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/include/uapi/linux/io_uring/zcrx.h b/include/uapi/linux/io_uring/zcrx.h index 67185566ad3c..3f7b72b09878 100644 --- a/include/uapi/linux/io_uring/zcrx.h +++ b/include/uapi/linux/io_uring/zcrx.h @@ -70,6 +70,7 @@ enum zcrx_features { enum zcrx_notification_type { ZCRX_NOTIF_NO_BUFFERS, + ZCRX_NOTIF_COPY, __ZCRX_NOTIF_TYPE_LAST, }; diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index 2bca9a58dcf6..404f9e7574a2 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -1532,8 +1532,13 @@ static int io_zcrx_copy_frag(struct io_kiocb *req, struct io_zcrx_ifq *ifq, const skb_frag_t *frag, int off, int len) { struct page *page = skb_frag_page(frag); + int ret; - return io_zcrx_copy_chunk(req, ifq, page, off + skb_frag_off(frag), len); + ret = io_zcrx_copy_chunk(req, ifq, page, off + skb_frag_off(frag), len); + if (ret > 0) + zcrx_send_notif(ifq, ZCRX_NOTIF_COPY); + + return ret; } static int io_zcrx_recv_frag(struct io_kiocb *req, struct io_zcrx_ifq *ifq, diff --git a/io_uring/zcrx.h b/io_uring/zcrx.h index e8b7717d6adf..54d91b580eaf 100644 --- a/io_uring/zcrx.h +++ b/io_uring/zcrx.h @@ -11,7 +11,7 @@ #define ZCRX_SUPPORTED_REG_FLAGS (ZCRX_REG_IMPORT | ZCRX_REG_NODEV) #define ZCRX_FEATURES (ZCRX_FEATURE_RX_PAGE_SIZE |\ ZCRX_FEATURE_NOTIFICATION) -#define ZCRX_NOTIF_TYPE_MASK (1U << ZCRX_NOTIF_NO_BUFFERS) +#define ZCRX_NOTIF_TYPE_MASK ((1U << ZCRX_NOTIF_NO_BUFFERS) | (1U << ZCRX_NOTIF_COPY)) struct io_zcrx_mem { unsigned long size; From 6935f631465f5f60205978a59228a26db4723d51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20L=C3=A9ger?= Date: Tue, 19 May 2026 12:44:34 +0100 Subject: [PATCH 14/31] io_uring/zcrx: add shared-memory notification statistics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add support for an optional stats struct embedded in the refill queue region, allowing userspace to monitor copy-fallback in real-time. Userspace queries the stats struct size and alignment via IO_URING_QUERY_ZCRX_NOTIF (notif_stats_size / notif_stats_alignment), then provides a stats_offset in zcrx_notification_desc pointing to a location within the refill queue region. The kernel updates the stats counters in-place on every copy-fallback event. Signed-off-by: Clément Léger [pavel: rename io_uring_zcrx_notif_stats] Signed-off-by: Pavel Begunkov Link: https://patch.msgid.link/f6af5a21015efea4b733b9d77aba22c637788fe4.1779189667.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- include/uapi/linux/io_uring/query.h | 12 +++++++ include/uapi/linux/io_uring/zcrx.h | 15 ++++++-- io_uring/query.c | 16 +++++++++ io_uring/zcrx.c | 54 +++++++++++++++++++++++++++-- io_uring/zcrx.h | 1 + 5 files changed, 94 insertions(+), 4 deletions(-) diff --git a/include/uapi/linux/io_uring/query.h b/include/uapi/linux/io_uring/query.h index 95500759cc13..1a68eca7c6b4 100644 --- a/include/uapi/linux/io_uring/query.h +++ b/include/uapi/linux/io_uring/query.h @@ -23,6 +23,7 @@ enum { IO_URING_QUERY_OPCODES = 0, IO_URING_QUERY_ZCRX = 1, IO_URING_QUERY_SCQ = 2, + IO_URING_QUERY_ZCRX_NOTIF = 3, __IO_URING_QUERY_MAX, }; @@ -62,6 +63,17 @@ struct io_uring_query_zcrx { __u64 __resv2; }; +struct io_uring_query_zcrx_notif { + /* Bitmask of supported ZCRX_NOTIF_* flags */ + __u32 notif_flags; + /* Size of io_uring_zcrx_notif_stats */ + __u32 notif_stats_size; + /* Required alignment for the stats struct within the region (ie stats_offset) */ + __u32 notif_stats_off_alignment; + __u32 __resv1; + __u64 __resv2[4]; +}; + struct io_uring_query_scq { /* The SQ/CQ rings header size */ __u64 hdr_size; diff --git a/include/uapi/linux/io_uring/zcrx.h b/include/uapi/linux/io_uring/zcrx.h index 3f7b72b09878..15c05c45ce36 100644 --- a/include/uapi/linux/io_uring/zcrx.h +++ b/include/uapi/linux/io_uring/zcrx.h @@ -75,11 +75,22 @@ enum zcrx_notification_type { __ZCRX_NOTIF_TYPE_LAST, }; +enum zcrx_notification_desc_flags { + /* If set, stats_offset holds a valid offset to a notif_stats struct */ + ZCRX_NOTIF_DESC_FLAG_STATS = 1 << 0, +}; + +struct zcrx_notif_stats { + __u64 copy_count; /* cumulative copy-fallback CQEs */ + __u64 copy_bytes; /* cumulative bytes copied */ +}; + struct zcrx_notification_desc { __u64 user_data; __u32 type_mask; - __u32 __resv1; - __u64 __resv2[10]; + __u32 flags; /* see enum zcrx_notification_desc_flags */ + __u64 stats_offset; /* offset from the beginning of refill ring region for stats */ + __u64 __resv2[9]; }; /* diff --git a/io_uring/query.c b/io_uring/query.c index c1704d088374..d529d94aa8f4 100644 --- a/io_uring/query.c +++ b/io_uring/query.c @@ -9,6 +9,7 @@ union io_query_data { struct io_uring_query_opcode opcodes; struct io_uring_query_zcrx zcrx; + struct io_uring_query_zcrx_notif zcrx_notif; struct io_uring_query_scq scq; }; @@ -44,6 +45,18 @@ static ssize_t io_query_zcrx(union io_query_data *data) return sizeof(*e); } +static ssize_t io_query_zcrx_notif(union io_query_data *data) +{ + struct io_uring_query_zcrx_notif *e = &data->zcrx_notif; + + e->notif_flags = ZCRX_NOTIF_TYPE_MASK; + e->notif_stats_size = sizeof(struct zcrx_notif_stats); + e->notif_stats_off_alignment = __alignof__(struct zcrx_notif_stats); + e->__resv1 = 0; + memset(&e->__resv2, 0, sizeof(e->__resv2)); + return sizeof(*e); +} + static ssize_t io_query_scq(union io_query_data *data) { struct io_uring_query_scq *e = &data->scq; @@ -83,6 +96,9 @@ static int io_handle_query_entry(union io_query_data *data, void __user *uhdr, case IO_URING_QUERY_ZCRX: ret = io_query_zcrx(data); break; + case IO_URING_QUERY_ZCRX_NOTIF: + ret = io_query_zcrx_notif(data); + break; case IO_URING_QUERY_SCQ: ret = io_query_scq(data); break; diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index 404f9e7574a2..6bd71435e475 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -415,6 +415,7 @@ static void io_free_rbuf_ring(struct io_zcrx_ifq *ifq) io_free_region(ifq->user, &ifq->rq_region); ifq->rq.ring = IO_URING_PTR_POISON; ifq->rq.rqes = IO_URING_PTR_POISON; + ifq->notif_stats = IO_URING_PTR_POISON; } static void io_zcrx_free_area(struct io_zcrx_ifq *ifq, @@ -853,6 +854,33 @@ static int zcrx_register_netdev(struct io_zcrx_ifq *ifq, return ret; } +static int zcrx_validate_notif_stats(struct io_zcrx_ifq *ifq, + const struct io_uring_zcrx_ifq_reg *reg, + const struct zcrx_notification_desc *notif) +{ + size_t stats_off = notif->stats_offset; + size_t used, end; + + used = reg->offsets.rqes + + sizeof(struct io_uring_zcrx_rqe) * reg->rq_entries; + + if (!IS_ALIGNED(stats_off, __alignof__(struct zcrx_notif_stats))) + return -EINVAL; + if (stats_off < used) + return -ERANGE; + if (check_add_overflow(stats_off, + sizeof(struct zcrx_notif_stats), + &end)) + return -ERANGE; + if (end > io_region_size(&ifq->rq_region)) + return -ERANGE; + + ifq->notif_stats = io_region_get_ptr(&ifq->rq_region) + stats_off; + memset(ifq->notif_stats, 0, sizeof(*ifq->notif_stats)); + + return 0; +} + int io_register_zcrx(struct io_ring_ctx *ctx, struct io_uring_zcrx_ifq_reg __user *arg) { @@ -906,7 +934,13 @@ int io_register_zcrx(struct io_ring_ctx *ctx, return -EFAULT; if (notif.type_mask & ~ZCRX_NOTIF_TYPE_MASK) return -EINVAL; - if (notif.__resv1 || !mem_is_zero(¬if.__resv2, sizeof(notif.__resv2))) + if (notif.flags & ~ZCRX_NOTIF_DESC_FLAG_STATS) + return -EINVAL; + if (!(notif.flags & ZCRX_NOTIF_DESC_FLAG_STATS)) { + if (notif.stats_offset) + return -EINVAL; + } + if (!mem_is_zero(¬if.__resv2, sizeof(notif.__resv2))) return -EINVAL; ifq = io_zcrx_ifq_alloc(ctx); @@ -937,6 +971,12 @@ int io_register_zcrx(struct io_ring_ctx *ctx, if (ret) goto err; + if (notif.flags & ZCRX_NOTIF_DESC_FLAG_STATS) { + ret = zcrx_validate_notif_stats(ifq, ®, ¬if); + if (ret) + goto err; + } + ifq->kern_readable = !(area.flags & IORING_ZCRX_AREA_DMABUF); if (!(reg.flags & ZCRX_REG_NODEV)) { @@ -1152,6 +1192,11 @@ static void zcrx_notif_tw(struct io_tw_req tw_req, io_tw_token_t tw) kmem_cache_free(req_cachep, req); } +static void zcrx_stat_add(__u64 *p, s64 v) +{ + WRITE_ONCE(*p, READ_ONCE(*p) + v); +} + static void zcrx_send_notif(struct io_zcrx_ifq *ifq, unsigned type) { gfp_t gfp = GFP_ATOMIC | __GFP_NOWARN | __GFP_ZERO; @@ -1535,8 +1580,13 @@ static int io_zcrx_copy_frag(struct io_kiocb *req, struct io_zcrx_ifq *ifq, int ret; ret = io_zcrx_copy_chunk(req, ifq, page, off + skb_frag_off(frag), len); - if (ret > 0) + if (ret > 0) { + if (ifq->notif_stats) { + zcrx_stat_add(&ifq->notif_stats->copy_count, 1); + zcrx_stat_add(&ifq->notif_stats->copy_bytes, ret); + } zcrx_send_notif(ifq, ZCRX_NOTIF_COPY); + } return ret; } diff --git a/io_uring/zcrx.h b/io_uring/zcrx.h index 54d91b580eaf..fa00900e479e 100644 --- a/io_uring/zcrx.h +++ b/io_uring/zcrx.h @@ -80,6 +80,7 @@ struct io_zcrx_ifq { u32 allowed_notif_mask; u32 fired_notifs; u64 notif_data; + struct zcrx_notif_stats *notif_stats; }; #if defined(CONFIG_IO_URING_ZCRX) From ca55f98d6ff1ecb31a07b668a8d105b4e0829c6a Mon Sep 17 00:00:00 2001 From: liyouhong Date: Thu, 28 May 2026 10:49:36 +0800 Subject: [PATCH 15/31] io_uring/kbuf: align legacy buffer add limit with MAX_BIDS_PER_BGID io_provide_buffers_prep() accepts nbufs up to MAX_BIDS_PER_BGID, but io_add_buffers() stops when bl->nbufs reaches USHRT_MAX. This makes the effective add limit one lower than the validated limit. Use MAX_BIDS_PER_BGID in the add-side boundary check so validation and execution use the same limit, and update the comment to refer to the actual limit constant. Signed-off-by: liyouhong Link: https://patch.msgid.link/20260528024936.3672659-1-dayou5941@163.com Signed-off-by: Jens Axboe --- io_uring/kbuf.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/io_uring/kbuf.c b/io_uring/kbuf.c index dd54e43e9ddf..7a1c65f631c2 100644 --- a/io_uring/kbuf.c +++ b/io_uring/kbuf.c @@ -541,11 +541,11 @@ static int io_add_buffers(struct io_ring_ctx *ctx, struct io_provide_buf *pbuf, for (i = 0; i < pbuf->nbufs; i++) { /* - * Nonsensical to have more than sizeof(bid) buffers in a + * Nonsensical to have more than MAX_BIDS_PER_BGID buffers in a * buffer list, as the application then has no way of knowing * which duplicate bid refers to what buffer. */ - if (bl->nbufs == USHRT_MAX) { + if (bl->nbufs == MAX_BIDS_PER_BGID) { ret = -EOVERFLOW; break; } From 29bef9934b2521f787bb15dd1985d4c0d12ae02a Mon Sep 17 00:00:00 2001 From: Runyu Xiao Date: Thu, 28 May 2026 01:22:03 +0800 Subject: [PATCH 16/31] io_uring/io-wq: re-check IO_WQ_BIT_EXIT for each linked work item commit 10dc95939817 ("io_uring/io-wq: check IO_WQ_BIT_EXIT inside work run loop") fixed the obvious case where io_worker_handle_work() took one exit-bit snapshot before draining pending work, but the fix stops one level too early. io_worker_handle_work() now re-checks IO_WQ_BIT_EXIT in its outer work run loop, yet it still snapshots that bit once before processing a whole dependent linked-work chain. If io_wq_exit_start() sets IO_WQ_BIT_EXIT after the first linked item has started, the remaining linked items can still reuse stale do_kill = false, skip IO_WQ_WORK_CANCEL, and continue running after exit has begun. Move the check further inside, so it covers linked items too. Note: this is a syzbot special as it loves setting up tons of slow linked work on weird devices like msr that take forever to read, and immediately close the ring. Exit then takes a long time. Fixes: 10dc95939817 ("io_uring/io-wq: check IO_WQ_BIT_EXIT inside work run loop") Cc: stable@vger.kernel.org Signed-off-by: Runyu Xiao Link: https://patch.msgid.link/20260527172203.2043962-1-runyu.xiao@seu.edu.cn Signed-off-by: Jens Axboe --- io_uring/io-wq.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/io_uring/io-wq.c b/io_uring/io-wq.c index 7a9f94a0ce6f..15b35eb8d1f7 100644 --- a/io_uring/io-wq.c +++ b/io_uring/io-wq.c @@ -602,7 +602,6 @@ static void io_worker_handle_work(struct io_wq_acct *acct, struct io_wq *wq = worker->wq; do { - bool do_kill = test_bit(IO_WQ_BIT_EXIT, &wq->state); struct io_wq_work *work; /* @@ -638,6 +637,7 @@ static void io_worker_handle_work(struct io_wq_acct *acct, /* handle a whole dependent link */ do { + bool do_kill = test_bit(IO_WQ_BIT_EXIT, &wq->state); struct io_wq_work *next_hashed, *linked; unsigned int work_flags = atomic_read(&work->flags); unsigned int hash = __io_wq_is_hashed(work_flags) From ec02fe217fa66d79f8a65e8d28be9295c7f85093 Mon Sep 17 00:00:00 2001 From: Pavel Begunkov Date: Tue, 2 Jun 2026 11:08:25 +0100 Subject: [PATCH 17/31] io_uring/bpf-ops: restrict ctx access to BPF BPF programs should have no need in looking into struct io_ring_ctx, if anything, most of such cases would be anti patterns like looking up ring indices directly via the context. Replace it with a new empty structure, which is just an alias to struct io_ring_ctx. It'll create a new BTF type and fail verification if a BPF program tries to access it (beyond the first byte). It'll also give more flexibility for the future, and otherwise it can be made aligned with io_ring_ctx as before with struct groups if ever needed or extended in a different way. Fixes: d0e437b76bd3c ("io_uring/bpf-ops: implement loop_step with BPF struct_ops") Signed-off-by: Pavel Begunkov Link: https://patch.msgid.link/5f6ca3649e9e0bae8667db4357e28dd00cd07901.1780394491.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- include/linux/io_uring_types.h | 4 +++- io_uring/bpf-ops.c | 9 ++++++--- io_uring/bpf-ops.h | 2 +- io_uring/loop.c | 2 +- io_uring/loop.h | 10 ++++++++++ 5 files changed, 21 insertions(+), 6 deletions(-) diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h index 23b8891d5704..aa4d5477f859 100644 --- a/include/linux/io_uring_types.h +++ b/include/linux/io_uring_types.h @@ -290,6 +290,8 @@ enum { IO_RING_F_IOWQ_LIMITS_SET = BIT(12), }; +struct iou_ctx {}; + struct io_ring_ctx { /* const or read-mostly hot data */ struct { @@ -366,7 +368,7 @@ struct io_ring_ctx { struct io_alloc_cache rw_cache; struct io_alloc_cache cmd_cache; - int (*loop_step)(struct io_ring_ctx *ctx, + int (*loop_step)(struct iou_ctx *, struct iou_loop_params *); /* diff --git a/io_uring/bpf-ops.c b/io_uring/bpf-ops.c index 937e48bef40b..5a50f0675fe5 100644 --- a/io_uring/bpf-ops.c +++ b/io_uring/bpf-ops.c @@ -14,15 +14,18 @@ static const struct btf_type *loop_params_type; __bpf_kfunc_start_defs(); -__bpf_kfunc int bpf_io_uring_submit_sqes(struct io_ring_ctx *ctx, u32 nr) +__bpf_kfunc int bpf_io_uring_submit_sqes(struct iou_ctx *loop_ctx, u32 nr) { + struct io_ring_ctx *ctx = io_loop_demangle_ctx(loop_ctx); + return io_submit_sqes(ctx, nr); } __bpf_kfunc -__u8 *bpf_io_uring_get_region(struct io_ring_ctx *ctx, __u32 region_id, +__u8 *bpf_io_uring_get_region(struct iou_ctx *loop_ctx, __u32 region_id, const size_t rdwr_buf_size) { + struct io_ring_ctx *ctx = io_loop_demangle_ctx(loop_ctx); struct io_mapped_region *r; lockdep_assert_held(&ctx->uring_lock); @@ -58,7 +61,7 @@ static const struct btf_kfunc_id_set bpf_io_uring_kfunc_set = { .set = &io_uring_kfunc_set, }; -static int io_bpf_ops__loop_step(struct io_ring_ctx *ctx, +static int io_bpf_ops__loop_step(struct iou_ctx *ctx, struct iou_loop_params *lp) { return IOU_LOOP_STOP; diff --git a/io_uring/bpf-ops.h b/io_uring/bpf-ops.h index b39b3fd3acda..0b6d7894915e 100644 --- a/io_uring/bpf-ops.h +++ b/io_uring/bpf-ops.h @@ -11,7 +11,7 @@ enum { }; struct io_uring_bpf_ops { - int (*loop_step)(struct io_ring_ctx *ctx, struct iou_loop_params *lp); + int (*loop_step)(struct iou_ctx *, struct iou_loop_params *lp); __u32 ring_fd; void *priv; diff --git a/io_uring/loop.c b/io_uring/loop.c index 31843cc3e451..bbbb6ef14e6a 100644 --- a/io_uring/loop.c +++ b/io_uring/loop.c @@ -49,7 +49,7 @@ static int __io_run_loop(struct io_ring_ctx *ctx) if (unlikely(!ctx->loop_step)) return -EFAULT; - step_res = ctx->loop_step(ctx, &lp); + step_res = ctx->loop_step(io_loop_mangle_ctx(ctx), &lp); if (step_res == IOU_LOOP_STOP) break; if (step_res != IOU_LOOP_CONTINUE) diff --git a/io_uring/loop.h b/io_uring/loop.h index d7718b9ce61e..4dd4fb3aefef 100644 --- a/io_uring/loop.h +++ b/io_uring/loop.h @@ -24,4 +24,14 @@ static inline bool io_has_loop_ops(struct io_ring_ctx *ctx) int io_run_loop(struct io_ring_ctx *ctx); +static inline struct iou_ctx *io_loop_mangle_ctx(struct io_ring_ctx *ctx) +{ + return (struct iou_ctx *)ctx; +} + +static inline struct io_ring_ctx *io_loop_demangle_ctx(struct iou_ctx *ctx) +{ + return (struct io_ring_ctx *)ctx; +} + #endif From 3979840cd858f30f43ea9f4e7f7f1f56de82d698 Mon Sep 17 00:00:00 2001 From: Gabriel Krisman Bertazi Date: Tue, 2 Jun 2026 17:53:24 -0400 Subject: [PATCH 18/31] io_uring/net: Avoid msghdr on op_connect/op_bind async data Both IORING_OP_CONNECT and IORING_OP_BIND reuse the msghdr object just to store the sockaddr. Beyond allocating a much larger object than needed, msghdr can also wrap an iovec, which will be recycled unnecessarily. This uses the sockaddr directly. Signed-off-by: Gabriel Krisman Bertazi Link: https://patch.msgid.link/20260602215327.1885109-2-krisman@suse.de Signed-off-by: Jens Axboe --- io_uring/net.c | 30 ++++++++++++++---------------- io_uring/opdef.c | 4 ++-- 2 files changed, 16 insertions(+), 18 deletions(-) diff --git a/io_uring/net.c b/io_uring/net.c index cceb5c1409ca..7ff5975e118b 100644 --- a/io_uring/net.c +++ b/io_uring/net.c @@ -1677,8 +1677,7 @@ void io_socket_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req) void io_connect_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req) { struct io_connect *conn = io_kiocb_to_cmd(req, struct io_connect); - struct io_async_msghdr *iomsg = req->async_data; - struct sockaddr_storage *ss = &iomsg->addr; + struct sockaddr_storage *ss = req->async_data; /* * move_addr_to_kernel() skips the copy for addr_len == 0, so @@ -1772,7 +1771,7 @@ int io_socket(struct io_kiocb *req, unsigned int issue_flags) int io_connect_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) { struct io_connect *conn = io_kiocb_to_cmd(req, struct io_connect); - struct io_async_msghdr *io; + struct sockaddr_storage *addr; if (sqe->len || sqe->buf_index || sqe->rw_flags || sqe->splice_fd_in) return -EINVAL; @@ -1781,17 +1780,17 @@ int io_connect_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) conn->addr_len = READ_ONCE(sqe->addr2); conn->in_progress = conn->seen_econnaborted = false; - io = io_msg_alloc_async(req); - if (unlikely(!io)) + addr = io_uring_alloc_async_data(NULL, req); + if (unlikely(!addr)) return -ENOMEM; - return move_addr_to_kernel(conn->addr, conn->addr_len, &io->addr); + return move_addr_to_kernel(conn->addr, conn->addr_len, addr); } int io_connect(struct io_kiocb *req, unsigned int issue_flags) { struct io_connect *connect = io_kiocb_to_cmd(req, struct io_connect); - struct io_async_msghdr *io = req->async_data; + struct sockaddr_storage *addr = req->async_data; unsigned file_flags; int ret; bool force_nonblock = issue_flags & IO_URING_F_NONBLOCK; @@ -1805,8 +1804,7 @@ int io_connect(struct io_kiocb *req, unsigned int issue_flags) file_flags = force_nonblock ? O_NONBLOCK : 0; - ret = __sys_connect_file(req->file, &io->addr, connect->addr_len, - file_flags); + ret = __sys_connect_file(req->file, addr, connect->addr_len, file_flags); if ((ret == -EAGAIN || ret == -EINPROGRESS || ret == -ECONNABORTED) && force_nonblock) { if (ret == -EINPROGRESS) { @@ -1835,7 +1833,6 @@ int io_connect(struct io_kiocb *req, unsigned int issue_flags) out: if (ret < 0) req_set_fail(req); - io_req_msg_cleanup(req, issue_flags); io_req_set_res(req, ret, 0); return IOU_COMPLETE; } @@ -1844,7 +1841,7 @@ int io_bind_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) { struct io_bind *bind = io_kiocb_to_cmd(req, struct io_bind); struct sockaddr __user *uaddr; - struct io_async_msghdr *io; + struct sockaddr_storage *addr; if (sqe->len || sqe->buf_index || sqe->rw_flags || sqe->splice_fd_in) return -EINVAL; @@ -1852,16 +1849,17 @@ int io_bind_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) uaddr = u64_to_user_ptr(READ_ONCE(sqe->addr)); bind->addr_len = READ_ONCE(sqe->addr2); - io = io_msg_alloc_async(req); - if (unlikely(!io)) + addr = io_uring_alloc_async_data(NULL, req); + if (unlikely(!addr)) return -ENOMEM; - return move_addr_to_kernel(uaddr, bind->addr_len, &io->addr); + return move_addr_to_kernel(uaddr, bind->addr_len, addr); } + int io_bind(struct io_kiocb *req, unsigned int issue_flags) { struct io_bind *bind = io_kiocb_to_cmd(req, struct io_bind); - struct io_async_msghdr *io = req->async_data; + struct sockaddr_storage *addr = req->async_data; struct socket *sock; int ret; @@ -1869,7 +1867,7 @@ int io_bind(struct io_kiocb *req, unsigned int issue_flags) if (unlikely(!sock)) return -ENOTSOCK; - ret = __sys_bind_socket(sock, &io->addr, bind->addr_len); + ret = __sys_bind_socket(sock, addr, bind->addr_len); if (ret < 0) req_set_fail(req); io_req_set_res(req, ret, 0); diff --git a/io_uring/opdef.c b/io_uring/opdef.c index 8ea6bd274607..517bad015505 100644 --- a/io_uring/opdef.c +++ b/io_uring/opdef.c @@ -204,7 +204,7 @@ const struct io_issue_def io_issue_defs[] = { .pollout = 1, #if defined(CONFIG_NET) .filter_pdu_size = sizeof_field(struct io_uring_bpf_ctx, connect), - .async_size = sizeof(struct io_async_msghdr), + .async_size = sizeof(struct sockaddr_storage), .prep = io_connect_prep, .issue = io_connect, .filter_populate = io_connect_bpf_populate, @@ -505,7 +505,7 @@ const struct io_issue_def io_issue_defs[] = { .needs_file = 1, .prep = io_bind_prep, .issue = io_bind, - .async_size = sizeof(struct io_async_msghdr), + .async_size = sizeof(struct sockaddr_storage), #else .prep = io_eopnotsupp_prep, #endif From 4305fcf9c315e36e8ed05057fd9b170df92f2adc Mon Sep 17 00:00:00 2001 From: Gabriel Krisman Bertazi Date: Tue, 2 Jun 2026 17:53:25 -0400 Subject: [PATCH 19/31] io_uring/net: Remove async_size for OP_LISTEN OP_LISTEN does not use async_data. Remove it. Signed-off-by: Gabriel Krisman Bertazi Link: https://patch.msgid.link/20260602215327.1885109-3-krisman@suse.de Signed-off-by: Jens Axboe --- io_uring/opdef.c | 1 - 1 file changed, 1 deletion(-) diff --git a/io_uring/opdef.c b/io_uring/opdef.c index 517bad015505..88a45c7d897f 100644 --- a/io_uring/opdef.c +++ b/io_uring/opdef.c @@ -515,7 +515,6 @@ const struct io_issue_def io_issue_defs[] = { .needs_file = 1, .prep = io_listen_prep, .issue = io_listen, - .async_size = sizeof(struct io_async_msghdr), #else .prep = io_eopnotsupp_prep, #endif From 1f826f2db9991b4b625448fce14bbb2108f6bd07 Mon Sep 17 00:00:00 2001 From: Gabriel Krisman Bertazi Date: Tue, 2 Jun 2026 17:53:26 -0400 Subject: [PATCH 20/31] io_uring/nop: Drop a wrong comment in struct io_nop This was copy-pasted from io_rw, where the comment actually makes sense. Here, we don't have struct kiocb. Drop the comment. Signed-off-by: Gabriel Krisman Bertazi Link: https://patch.msgid.link/20260602215327.1885109-4-krisman@suse.de Signed-off-by: Jens Axboe --- io_uring/nop.c | 1 - 1 file changed, 1 deletion(-) diff --git a/io_uring/nop.c b/io_uring/nop.c index 3caf07878f8a..e3333b5ed4e0 100644 --- a/io_uring/nop.c +++ b/io_uring/nop.c @@ -12,7 +12,6 @@ #include "nop.h" struct io_nop { - /* NOTE: kiocb has the file as the first member, so don't do it here */ struct file *file; int result; int fd; From 57ed21fad4022d595c6654d3b4d2b2083a79ee25 Mon Sep 17 00:00:00 2001 From: Ming Lei Date: Mon, 8 Jun 2026 09:25:10 -0500 Subject: [PATCH 21/31] io_uring/net: support registered buffer for plain send and recv So far IORING_RECVSEND_FIXED_BUF is only honoured on the SEND_ZC path, even though the import wiring is already present for plain send and completely absent for recv. Targets such as ublk's NBD backend want to push/pull I/O data directly to/from an io_uring registered buffer over a plain send/recv on a TCP socket. Wire IORING_RECVSEND_FIXED_BUF into the plain IORING_OP_SEND and IORING_OP_RECV paths: - Accept the flag in SENDMSG_FLAGS / RECVMSG_FLAGS and, at prep time, restrict it to the non-vectorized IORING_OP_SEND / IORING_OP_RECV opcodes. It is mutually exclusive with buffer select, bundles and (for recv) multishot, and records sqe->buf_index. - For recv, set REQ_F_IMPORT_BUFFER in setup so the registered buffer is imported lazily at issue time, mirroring the send path. - In io_send()/io_recv(), import the registered buffer via io_import_reg_buf() (ITER_SOURCE for send, ITER_DEST for recv) and clear REQ_F_IMPORT_BUFFER. The resulting bvec iter persists in async_data, so MSG_WAITALL partial send/recv retries resume at the right offset. Signed-off-by: Ming Lei Link: https://patch.msgid.link/20260608142511.659240-2-ming.lei@redhat.com [axboe: combine flags checks] Signed-off-by: Jens Axboe --- io_uring/net.c | 46 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 44 insertions(+), 2 deletions(-) diff --git a/io_uring/net.c b/io_uring/net.c index 7ff5975e118b..5ae538b3c0f3 100644 --- a/io_uring/net.c +++ b/io_uring/net.c @@ -417,7 +417,8 @@ static int io_sendmsg_setup(struct io_kiocb *req, const struct io_uring_sqe *sqe return io_net_import_vec(req, kmsg, msg.msg_iov, msg.msg_iovlen, ITER_SOURCE); } -#define SENDMSG_FLAGS (IORING_RECVSEND_POLL_FIRST | IORING_RECVSEND_BUNDLE | IORING_SEND_VECTORIZED) +#define SENDMSG_FLAGS (IORING_RECVSEND_POLL_FIRST | IORING_RECVSEND_BUNDLE | \ + IORING_SEND_VECTORIZED | IORING_RECVSEND_FIXED_BUF) int io_sendmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) { @@ -430,6 +431,14 @@ int io_sendmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) sr->flags = READ_ONCE(sqe->ioprio); if (sr->flags & ~SENDMSG_FLAGS) return -EINVAL; + if (sr->flags & IORING_RECVSEND_FIXED_BUF) { + /* registered buffer send only supported for plain IORING_OP_SEND */ + if (req->opcode != IORING_OP_SEND || + (req->flags & REQ_F_BUFFER_SELECT) || + (sr->flags & (IORING_RECVSEND_BUNDLE|IORING_SEND_VECTORIZED))) + return -EINVAL; + req->buf_index = READ_ONCE(sqe->buf_index); + } sr->msg_flags = READ_ONCE(sqe->msg_flags) | MSG_NOSIGNAL; if (sr->msg_flags & MSG_DONTWAIT) req->flags |= REQ_F_NOWAIT; @@ -661,6 +670,15 @@ int io_send(struct io_kiocb *req, unsigned int issue_flags) (sr->flags & IORING_RECVSEND_POLL_FIRST)) return -EAGAIN; + if (req->flags & REQ_F_IMPORT_BUFFER) { + ret = io_import_reg_buf(req, &kmsg->msg.msg_iter, + (u64)(uintptr_t)sr->buf, sr->len, + ITER_SOURCE, issue_flags); + if (unlikely(ret)) + return ret; + req->flags &= ~REQ_F_IMPORT_BUFFER; + } + flags = sr->msg_flags; if (issue_flags & IO_URING_F_NONBLOCK) flags |= MSG_DONTWAIT; @@ -776,6 +794,10 @@ static int io_recvmsg_prep_setup(struct io_kiocb *req) if (req->flags & REQ_F_BUFFER_SELECT) return 0; + if (sr->flags & IORING_RECVSEND_FIXED_BUF) { + req->flags |= REQ_F_IMPORT_BUFFER; + return 0; + } return import_ubuf(ITER_DEST, sr->buf, sr->len, &kmsg->msg.msg_iter); } @@ -784,7 +806,7 @@ static int io_recvmsg_prep_setup(struct io_kiocb *req) } #define RECVMSG_FLAGS (IORING_RECVSEND_POLL_FIRST | IORING_RECV_MULTISHOT | \ - IORING_RECVSEND_BUNDLE) + IORING_RECVSEND_BUNDLE | IORING_RECVSEND_FIXED_BUF) int io_recvmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) { @@ -802,6 +824,14 @@ int io_recvmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe) sr->flags = READ_ONCE(sqe->ioprio); if (sr->flags & ~RECVMSG_FLAGS) return -EINVAL; + if (sr->flags & IORING_RECVSEND_FIXED_BUF) { + /* registered buffer recv only for plain IORING_OP_RECV */ + if (req->opcode != IORING_OP_RECV || + (req->flags & REQ_F_BUFFER_SELECT) || + (sr->flags & (IORING_RECV_MULTISHOT | IORING_RECVSEND_BUNDLE))) + return -EINVAL; + req->buf_index = READ_ONCE(sqe->buf_index); + } sr->msg_flags = READ_ONCE(sqe->msg_flags); if (sr->msg_flags & MSG_DONTWAIT) req->flags |= REQ_F_NOWAIT; @@ -1198,6 +1228,18 @@ int io_recv(struct io_kiocb *req, unsigned int issue_flags) if (force_nonblock) flags |= MSG_DONTWAIT; + if (req->flags & REQ_F_IMPORT_BUFFER) { + ret = io_import_reg_buf(req, &kmsg->msg.msg_iter, + (u64)(uintptr_t)sr->buf, sr->len, + ITER_DEST, issue_flags); + if (unlikely(ret)) { + kmsg->msg.msg_inq = -1; + sel.buf_list = NULL; + goto out_free; + } + req->flags &= ~REQ_F_IMPORT_BUFFER; + } + retry_multishot: sel.buf_list = NULL; if (io_do_buffer_select(req)) { From 46800585ae04863c623b1563b03d12e9381089f1 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Thu, 9 Apr 2026 11:22:43 -0600 Subject: [PATCH 22/31] io_uring/kbuf: validate ring provided buffer addresses with access_ok() Commit: 809b997a5ce9 ("x86-64/arm64/powerpc: clean up and rename __copy_from_user_flushcache") sanitized that any provided copy helper should separately validate destination and source addresses, but we should also ensure that anything that is retrieved from a buffer is validated upfront. For ring provided buffers, always include an access_ok() when grabbing a new buffer. Fixes: c7fb19428d67 ("io_uring: add support for ring mapped supplied buffers") Signed-off-by: Jens Axboe --- io_uring/kbuf.c | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/io_uring/kbuf.c b/io_uring/kbuf.c index 7a1c65f631c2..b9f4720ca22f 100644 --- a/io_uring/kbuf.c +++ b/io_uring/kbuf.c @@ -210,10 +210,14 @@ static struct io_br_sel io_ring_buffer_select(struct io_kiocb *req, size_t *len, buf_len = READ_ONCE(buf->len); if (*len == 0 || *len > buf_len) *len = buf_len; + sel.addr = u64_to_user_ptr(READ_ONCE(buf->addr)); + if (unlikely(!access_ok(sel.addr, *len))) { + sel.addr = NULL; + return sel; + } req->flags |= REQ_F_BUFFER_RING | REQ_F_BUFFERS_COMMIT; req->buf_index = READ_ONCE(buf->bid); sel.buf_list = bl; - sel.addr = u64_to_user_ptr(READ_ONCE(buf->addr)); if (io_should_commit(req, issue_flags)) { if (!io_kbuf_commit(req, sel.buf_list, *len, 1)) @@ -250,6 +254,7 @@ static int io_ring_buffers_peek(struct io_kiocb *req, struct buf_sel_arg *arg, struct io_buffer_list *bl) { struct io_uring_buf_ring *br = bl->buf_ring; + struct iovec *org_iovs = arg->iovs; struct iovec *iov = arg->iovs; int nr_iovs = arg->nr_iovs; __u16 nr_avail, tail, head; @@ -311,6 +316,11 @@ static int io_ring_buffers_peek(struct io_kiocb *req, struct buf_sel_arg *arg, iov->iov_base = u64_to_user_ptr(READ_ONCE(buf->addr)); iov->iov_len = len; + if (unlikely(!access_ok(iov->iov_base, len))) { + if (arg->iovs != org_iovs) + kfree(arg->iovs); + return -EFAULT; + } iov++; arg->out_len += len; From 753f81b3b8d78d1e34872b9c14dff3a841ada4b4 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Thu, 11 Jun 2026 12:27:42 -0600 Subject: [PATCH 23/31] io_uring/zcrx: kill dead 'sock' member in struct io_zcrx_args This member is only ever assigned, never read. Kill it. Signed-off-by: Jens Axboe --- io_uring/zcrx.c | 2 -- 1 file changed, 2 deletions(-) diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c index 6bd71435e475..49163f9c39df 100644 --- a/io_uring/zcrx.c +++ b/io_uring/zcrx.c @@ -339,7 +339,6 @@ static void zcrx_sync_for_device(struct page_pool *pp, struct io_zcrx_ifq *zcrx, struct io_zcrx_args { struct io_kiocb *req; struct io_zcrx_ifq *ifq; - struct socket *sock; unsigned nr_skbs; }; @@ -1733,7 +1732,6 @@ static int io_zcrx_tcp_recvmsg(struct io_kiocb *req, struct io_zcrx_ifq *ifq, struct io_zcrx_args args = { .req = req, .ifq = ifq, - .sock = sk->sk_socket, }; read_descriptor_t rd_desc = { .count = len ? len : UINT_MAX, From ed64f5c546b3d5e3a4840f6c055448ce90edf56c Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Thu, 11 Jun 2026 20:27:22 -0600 Subject: [PATCH 24/31] io_uring: grab RCU read lock marking task run Not required right now, as io_req_local_work_add() already calls this helper with the RCU read lock held. But in preparation for that not being the case, grab it locally. Reviewed-by: Caleb Sander Mateos Signed-off-by: Jens Axboe --- io_uring/tw.c | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/io_uring/tw.c b/io_uring/tw.c index 023d5e6bc491..f4335c8d50d9 100644 --- a/io_uring/tw.c +++ b/io_uring/tw.c @@ -158,11 +158,11 @@ void tctx_task_work(struct callback_head *cb) */ static void io_ctx_mark_taskrun(struct io_ring_ctx *ctx) { - lockdep_assert_in_rcu_read_lock(); - if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) { - struct io_rings *rings = rcu_dereference(ctx->rings_rcu); + struct io_rings *rings; + guard(rcu)(); + rings = rcu_dereference(ctx->rings_rcu); atomic_or(IORING_SQ_TASKRUN, &rings->sq_flags); } } From 50cb44bd0d5f243919a06b17b1d979fdcd72cb2b Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Wed, 10 Jun 2026 15:19:15 -0600 Subject: [PATCH 25/31] io_uring/mpscq: add lockless multi-producer, single-consumer FIFO queue Local task_work is currently using llists for managing the work, but that's a LIFO type of list. This means that running this task_work needs to reverse the list first, to ensure fairness in running the queued items. Add a lockless FIFO queued, based on Dmitry Vyukov's intrusive MPSC node-based queue algorithm, modified with an externally held consumer cursor and conditional stub reinsertion. See comments in the header. Producers are wait-free: a push is a single xchg() on the queue tail, which serializes concurrent producers and defines the FIFO order, plus a store linking the node to its predecessor. There are no cmpxchg retry loops, and pushing is safe from any context, including hardirq. The cost of linked list FIFO ordering is that a push publishes the node in two steps - the xchg() makes it visible as the new tail before the subsequent store links it into the chain that is reachable from the head. A consumer hitting that window gets a NULL from mpscq_pop() while mpscq_empty() reports false, and must retry later rather than treat the queue as empty. The window is two instructions wide, but a producer can get preempted inside it, so the consumer must not busy wait on it. The consumer side supports a single consumer at a time, with callers providing their own serialization. A stub node, which also defines the empty state (tail == stub), allows the consumer to detach the final node without racing against producer link stores: that node is only handed out once the stub has been cmpxchg'ed back in as the tail. This also guarantees that the previous tail returned by mpscq_push() cannot get freed before that push has linked it, making it always valid for comparisons. The consumer cursor is deliberately not part of the queue struct - the caller owns it and passes it to mpscq_pop(). This is done to separate the consumer and producers cacheline. The cursor is written for every popped entry, and keeping it on the same cacheline as ->tail would have the consumer invalidating the line that producers need for every push. Keeping it external lets the caller place it with its own consumer side data instead. Reviewed-by: Caleb Sander Mateos Signed-off-by: Jens Axboe --- include/linux/io_uring_types.h | 12 ++++ io_uring/mpscq.h | 125 +++++++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+) create mode 100644 io_uring/mpscq.h diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h index aa4d5477f859..85e12b4884a5 100644 --- a/include/linux/io_uring_types.h +++ b/include/linux/io_uring_types.h @@ -55,6 +55,18 @@ struct io_wq_work_list { struct io_wq_work_node *last; }; +/* + * Lockless multi-producer, single-consumer FIFO queue, see + * io_uring/mpscq.h for the implementation and rules. Defined here so + * that it can be embedded in io_ring_ctx. This is the producer side + * only - the consumer cursor is kept separately, on a cacheline that + * isn't dirtied by the producers. + */ +struct mpscq { + struct llist_node *tail; /* producers */ + struct llist_node stub; +}; + struct io_wq_work { struct io_wq_work_node list; atomic_t flags; diff --git a/io_uring/mpscq.h b/io_uring/mpscq.h new file mode 100644 index 000000000000..c801384c6a0a --- /dev/null +++ b/io_uring/mpscq.h @@ -0,0 +1,125 @@ +/* SPDX-License-Identifier: GPL-2.0 */ +#ifndef IOU_MPSCQ_H +#define IOU_MPSCQ_H + +#include + +/* + * mpscq - lockless multi-producer, single-consumer FIFO queue + * + * Unlike llist, which is LIFO ordered and hence needs an O(n) + * llist_reverse_order() pass before entries can be processed in queue order, + * this queue hands out nodes in the order they were pushed. + * + * The consumer cursor is held by the caller rather than in the queue struct + * (see below), and with the stub reinsertion done as a single cmpxchg attempt + * instead of an unconditional push, keeping tail == stub a reliable empty test + * while a producer is in the middle of a push. + * + * Producers may run in any context (task, softirq, hardirq) and are wait-free: + * a push is one xchg() plus one store, with no retry loops. FIFO order between + * producers is the order in which the xchg() on ->tail serializes them. + * + * The price for linked-list FIFO is that a push publishes the node in two + * steps: the xchg() makes it the new tail, and the subsequent store links it to + * its predecessor. In between, the tail end of the queue is not yet reachable + * from the head. mpscq_pop() detects this and returns NULL, while mpscq_empty() + * reports false. The consumer must not treat such a NULL as "queue empty" - it + * should retry later. The window is two instructions wide, but a producer can + * be preempted inside it, so the consumer must not spin on it while holding + * resources the producer might need to make progress. + * + * The consumer side only supports a single consumer at a time, callers must + * provide their own serialization for it. The stub node is what allows the + * consumer to detach the final node without racing with the link stores of + * producers. This scheme also guarantees that the previous tail observed by + * mpscq_push() cannot be freed by the consumer until the push has linked it, + * which is what makes the deferred link store safe. + * + * The queue struct only holds the producer side. The consumer keeps its cursor + * (the oldest not yet handed out node) externally and passes it to mpscq_pop(), + * so that it can be placed on a different cacheline: the cursor is written for + * every pop, and having it share a line with ->tail would have the consumer + * invalidating the line that producers need for every push. + */ +static inline void mpscq_init(struct mpscq *q, struct llist_node **headp) +{ + q->tail = *headp = &q->stub; + q->stub.next = NULL; +} + +/* + * Returns true if the queue holds no entries that mpscq_pop() hasn't handed out + * yet. May be called from any context. Note that !empty doesn't guarantee that + * mpscq_pop() will return an entry yet, see the in-flight producer window + * above. + */ +static inline bool mpscq_empty(struct mpscq *q) +{ + return READ_ONCE(q->tail) == &q->stub; +} + +/* + * Push a node onto the queue. Safe against concurrent pushes from any context, + * and against the (single) consumer. Returns true if the queue was empty + * before this push. + */ +static inline bool mpscq_push(struct mpscq *q, struct llist_node *node) +{ + struct llist_node *prev; + + node->next = NULL; + /* + * xchg() implies a full barrier, so the initialization of the + * entry (including ->next above) is visible before the node can + * be reached, either via ->tail or via ->next chasing from the + * head once the store below has linked it. + */ + prev = xchg(&q->tail, node); + WRITE_ONCE(prev->next, node); + return prev == &q->stub; +} + +/* + * Pop the oldest node off the queue, or return NULL if no node is available. + * NULL is returned both when the queue is empty and when a producer has + * published a node via ->tail but hasn't linked it yet; use mpscq_empty() to + * tell the two apart. Single consumer only, with headp being the consumer + * cursor that mpscq_init() set up. + */ +static inline struct llist_node *mpscq_pop(struct mpscq *q, + struct llist_node **headp) +{ + struct llist_node *head = *headp, *next; + + if (head == &q->stub) { + head = READ_ONCE(head->next); + if (!head) + return NULL; + /* + * The stub is now detached and stays quiescent until the + * consumer reinserts it as the tail, so reset its ->next here, + * ready for that. + */ + q->stub.next = NULL; + *headp = head; + } + next = READ_ONCE(head->next); + if (next) { + *headp = next; + return head; + } + /* + * 'head' is the last linked node, it can only be handed out once the + * stub has taken its place as the tail. If the cmpxchg fails, a + * producer has made a new node the tail but hasn't linked 'head' to + * it yet - bail and let the caller retry. + */ + if (try_cmpxchg(&q->tail, &head, &q->stub)) { + *headp = &q->stub; + return head; + } + return NULL; +} + +#endif /* IOU_MPSCQ_H */ From d46ab2c98ababa19b41a5709b6921d7b1add7f74 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Wed, 10 Jun 2026 15:19:35 -0600 Subject: [PATCH 26/31] io_uring: switch local task_work to a mpscq The local (DEFER_TASKRUN) task_work list is an llist, which is LIFO ordered, and hence __io_run_local_work() has to restore the right running order with an O(n) llist_reverse_order() pass first. On top of that, a batch that gets capped by max_events needs the leftover entries parked on a separate ->retry_llist, as they can't be pushed back to the shared list. Switch it to the FIFO mpscq. Adds are wait-free instead of a cmpxchg retry loop, entries are popped in queue order with no reversal pass, capping a run simply leaves the remainder on the queue, and ->retry_llist goes away entirely. The consumer cursor, ->work_head, lives with the rest of the ->uring_lock protected state rather than next to the queue, so that popping entries doesn't dirty the producer side cacheline. For low amounts of task_work, this ends up being a bit more efficient than the existing scheme. As an example of that, doing multishot receives for 8 clients has the following task_work overhead: 1.02% sock-test [kernel.kallsyms] [k] io_req_local_work_add 0.88% sock-test [kernel.kallsyms] [k] __io_run_local_work_loop 0.60% sock-test [kernel.kallsyms] [k] llist_reverse_order 0.14% sock-test [kernel.kallsyms] [k] __io_run_local_work 2.64% at ~46Gb/sec and after this change: 1.08% sock-test [kernel.kallsyms] [k] io_req_local_work_add 1.03% sock-test [kernel.kallsyms] [k] __io_run_local_work 2.11% at ~53Gb/sec which has less overhead even though that test run was faster. For a case of having 1024 clients on a single ring: 2.22% sock-test [kernel.kallsyms] [k] llist_reverse_order 0.84% sock-test [kernel.kallsyms] [k] __io_run_local_work_loop 0.42% sock-test [kernel.kallsyms] [k] io_req_local_work_add 0.02% sock-test [kernel.kallsyms] [k] __io_run_local_work 3.50% at ~24Gb/sec we start to see the llist reversing taking a considerable amount of time, and the total add+run task_work overhead is around 3.5%. After the change: 0.90% sock-test [kernel.kallsyms] [k] __io_run_local_work 0.42% sock-test [kernel.kallsyms] [k] io_req_local_work_add 1.32% at ~26Gb/sec most of that overhead is gone, and performance is better as well. Caleb Sander Mateos reports that it improves the performance of a ublk 4kb workload by 4% [1], while testing v1 of this patchset. [1] https://lore.kernel.org/io-uring/CADUfDZr-MMYBaP-e+y9+xuRhuiunO2sBTUCmwZyd7AgT8sVtiQ@mail.gmail.com/ Signed-off-by: Jens Axboe --- include/linux/io_uring_types.h | 13 ++- io_uring/io_uring.c | 2 +- io_uring/tw.c | 143 +++++++++++++++------------------ io_uring/tw.h | 4 +- io_uring/wait.c | 2 +- io_uring/wait.h | 12 +-- 6 files changed, 87 insertions(+), 89 deletions(-) diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h index 85e12b4884a5..3e07c7059d7b 100644 --- a/include/linux/io_uring_types.h +++ b/include/linux/io_uring_types.h @@ -360,6 +360,14 @@ struct io_ring_ctx { bool poll_multi_queue; struct list_head iopoll_list; + /* + * Consumer cursor for ->work_list, protected by ->uring_lock. + * Deliberately kept away from the producer side of the queue, + * as it's written for every popped entry, and the producer + * cacheline is contended enough as it is. + */ + struct llist_node *work_head; + struct io_file_table file_table; struct io_rsrc_data buf_table; struct io_alloc_cache node_cache; @@ -417,8 +425,7 @@ struct io_ring_ctx { */ struct { struct io_rings __rcu *rings_rcu; - struct llist_head work_llist; - struct llist_head retry_llist; + struct mpscq work_list; unsigned long check_cq; atomic_t cq_wait_nr; atomic_t cq_timeouts; @@ -742,8 +749,6 @@ struct io_kiocb { */ u16 buf_index; - unsigned nr_tw; - /* REQ_F_* flags */ io_req_flags_t flags; diff --git a/io_uring/io_uring.c b/io_uring/io_uring.c index 02c02e14f392..0809fc70c91d 100644 --- a/io_uring/io_uring.c +++ b/io_uring/io_uring.c @@ -280,7 +280,7 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p) INIT_LIST_HEAD(&ctx->defer_list); INIT_LIST_HEAD(&ctx->timeout_list); INIT_LIST_HEAD(&ctx->ltimeout_list); - init_llist_head(&ctx->work_llist); + 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; diff --git a/io_uring/tw.c b/io_uring/tw.c index f4335c8d50d9..79feeb4f671a 100644 --- a/io_uring/tw.c +++ b/io_uring/tw.c @@ -14,6 +14,7 @@ #include "rw.h" #include "eventfd.h" #include "wait.h" +#include "mpscq.h" void io_fallback_req_func(struct work_struct *work) { @@ -170,11 +171,7 @@ static void io_ctx_mark_taskrun(struct io_ring_ctx *ctx) void io_req_local_work_add(struct io_kiocb *req, unsigned flags) { struct io_ring_ctx *ctx = req->ctx; - 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); + int nr_wait; /* * We don't know how many requests there are in the link and whether @@ -183,56 +180,45 @@ void io_req_local_work_add(struct io_kiocb *req, unsigned flags) if (req->flags & IO_REQ_LINK_FLAGS) 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. + * The xchg() in mpscq_push() implies a full barrier, which pairs with + * the barrier in set_current_state() on the io_cqring_wait() side. This + * ensures that either we see the updated ->cq_wait_nr, or waiters going + * to sleep will observe the work added to the list, which is similar to + * the wait/wake task state sync. */ - - if (!head) { + if (mpscq_push(&ctx->work_list, &req->io_task_work.node)) { io_ctx_mark_taskrun(ctx); if (data_race(ctx->int_flags) & IO_RING_F_HAS_EVFD) io_eventfd_signal(ctx, false); } + /* + * No one is waiting (IO_CQ_WAKE_INIT), or this cycle's wake up has + * already been issued (zero or negative, see below). + */ nr_wait = atomic_read(&ctx->cq_wait_nr); - /* not enough or no one is waiting */ - if (nr_tw < nr_wait) + if (nr_wait <= 0) return; - /* the previous add has already woken it up */ - if (nr_tw_prev >= nr_wait) + if (flags & IOU_F_TWQ_LAZY_WAKE) { + /* + * ->cq_wait_nr counts down the number of lazy adds, once it + * hits zero we're good to wake the waiter. A producer that + * gets delayed between pushing its entry and getting here + * may count down a later wait cycle. That's OK, it'll be an + * early wake, not a lost one. + */ + if (!atomic_dec_and_test(&ctx->cq_wait_nr)) + return; + } else if (atomic_xchg(&ctx->cq_wait_nr, IO_CQ_WAKE_INIT) <= 0) { + /* + * Potentially raced with lazy add, claim the wake. A value + * <= 0 means a lazy add hit zero or another forced add + * claimed IO_CQ_WAKE_INIT. Either way, the wake up for this + * wait cycle has already been done. + */ return; + } wake_up_state(ctx->submitter_task, TASK_INTERRUPTIBLE); } @@ -273,21 +259,27 @@ void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags) void __cold io_move_task_work_from_local(struct io_ring_ctx *ctx) { - struct llist_node *node; + struct llist_node *node, *first = NULL, **tail = &first; /* - * Running the work items may utilize ->retry_llist as a means - * for capping the number of task_work entries run at the same - * time. But that list can potentially race with moving the work - * from here, if the task is exiting. As any normal task_work - * running holds ->uring_lock already, just guard this slow path - * with ->uring_lock to avoid racing on ->retry_llist. + * The work list consumer side is serialized by ->uring_lock, see + * __io_run_local_work(). Grab it to guard against racing with normal + * task_work running, as the task may be exiting. */ guard(mutex)(&ctx->uring_lock); - node = llist_del_all(&ctx->work_llist); - __io_fallback_tw(node, false); - node = llist_del_all(&ctx->retry_llist); - __io_fallback_tw(node, false); + + while (!mpscq_empty(&ctx->work_list)) { + node = mpscq_pop(&ctx->work_list, &ctx->work_head); + if (!node) { + /* a producer is mid-push, wait for it to link */ + cpu_relax(); + continue; + } + *tail = node; + tail = &node->next; + } + *tail = NULL; + __io_fallback_tw(first, false); } static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events, @@ -302,22 +294,23 @@ static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events, return false; } -static int __io_run_local_work_loop(struct llist_node **node, +static int __io_run_local_work_loop(struct io_ring_ctx *ctx, io_tw_token_t tw, int events) { int ret = 0; - while (*node) { - struct llist_node *next = (*node)->next; - struct io_kiocb *req = container_of(*node, struct io_kiocb, - io_task_work.node); + while (ret < events) { + struct llist_node *node = mpscq_pop(&ctx->work_list, &ctx->work_head); + struct io_kiocb *req; + + if (!node) + break; + 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, (struct io_tw_req){req}, tw); - *node = next; - if (++ret >= events) - break; + ret++; } return ret; @@ -326,7 +319,6 @@ static int __io_run_local_work_loop(struct llist_node **node, static int __io_run_local_work(struct io_ring_ctx *ctx, io_tw_token_t tw, int min_events, int max_events) { - struct llist_node *node; unsigned int loops = 0; int ret = 0; @@ -335,24 +327,21 @@ static int __io_run_local_work(struct io_ring_ctx *ctx, io_tw_token_t tw, if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) atomic_andnot(IORING_SQ_TASKRUN, &ctx->rings->sq_flags); again: + /* + * If the last loop made no progress while work is still pending, + * a producer has published a node but hasn't linked it into the + * queue yet (see mpscq_pop()). Give it a chance to finish rather + * than spinning on the queue. + */ + if (unlikely(loops && !ret)) + cond_resched(); tw.cancel = io_should_terminate_tw(ctx); min_events -= ret; - ret = __io_run_local_work_loop(&ctx->retry_llist.first, tw, max_events); - if (ctx->retry_llist.first) - goto retry_done; - - /* - * llists are in reverse order, flip it back the right way before - * running the pending items. - */ - node = llist_reverse_order(llist_del_all(&ctx->work_llist)); - ret += __io_run_local_work_loop(&node, tw, max_events - ret); - ctx->retry_llist.first = node; + ret = __io_run_local_work_loop(ctx, tw, max_events); loops++; if (io_run_local_work_continue(ctx, ret, min_events)) goto again; -retry_done: io_submit_flush_completions(ctx); if (io_run_local_work_continue(ctx, ret, min_events)) goto again; diff --git a/io_uring/tw.h b/io_uring/tw.h index 415e330fabde..f42db5fdbded 100644 --- a/io_uring/tw.h +++ b/io_uring/tw.h @@ -6,6 +6,8 @@ #include #include +#include "mpscq.h" + #define IO_LOCAL_TW_DEFAULT_MAX 20 /* @@ -89,7 +91,7 @@ static inline int io_run_task_work(void) static inline bool io_local_work_pending(struct io_ring_ctx *ctx) { - return !llist_empty(&ctx->work_llist) || !llist_empty(&ctx->retry_llist); + return !mpscq_empty(&ctx->work_list); } static inline bool io_task_work_pending(struct io_ring_ctx *ctx) diff --git a/io_uring/wait.c b/io_uring/wait.c index ec01e78a216d..c5fc34d2ce97 100644 --- a/io_uring/wait.c +++ b/io_uring/wait.c @@ -98,7 +98,7 @@ static enum hrtimer_restart io_cqring_min_timer_wakeup(struct hrtimer *timer) if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) { atomic_set(&ctx->cq_wait_nr, 1); smp_mb(); - if (!llist_empty(&ctx->work_llist)) + if (io_local_work_pending(ctx)) goto out_wake; } diff --git a/io_uring/wait.h b/io_uring/wait.h index a4274b137f81..b7b9c46b1b01 100644 --- a/io_uring/wait.h +++ b/io_uring/wait.h @@ -5,12 +5,14 @@ #include /* - * 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(). + * ->cq_wait_nr is armed with the number of lazy task_work adds the waiter + * still needs, and counted down by the add side, with the add reaching zero + * issuing the (single) wake up for this wait cycle. Zero and below means no + * wake up is to be issued: IO_CQ_WAKE_INIT when no task is waiting (also + * what a forced wake up resets it to when claiming one), zero once the + * countdown has fired. */ -#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) +#define IO_CQ_WAKE_INIT (-1) struct ext_arg { size_t argsz; From de7341ffe49ed30a1d75b254ac8c731b057247bf Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Thu, 11 Jun 2026 10:13:22 -0600 Subject: [PATCH 27/31] io_uring: switch normal task_work to a mpscq Like the local task_work list, the normal (tctx) task_work list is an llist, and hence needs the O(n) llist_reverse_order() pass before running entries in queue order. On top of that, capped runs - sqpoll processing IORING_TW_CAP_ENTRIES_VALUE entries at a time - need the claimed-but-unprocessed leftovers carried in a separate retry_list, as they can't be pushed back to the shared list. Switch tctx->task_list to a mpscq, like what was done for the DEFER_TASKRUN paths as well. Signed-off-by: Jens Axboe --- include/linux/io_uring_types.h | 12 ++- io_uring/sqpoll.c | 30 +++---- io_uring/tctx.c | 3 +- io_uring/tw.c | 146 ++++++++++++++++++++------------- io_uring/tw.h | 4 +- 5 files changed, 113 insertions(+), 82 deletions(-) diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h index 3e07c7059d7b..f511e96865b6 100644 --- a/include/linux/io_uring_types.h +++ b/include/linux/io_uring_types.h @@ -131,6 +131,11 @@ struct io_uring_task { const struct io_ring_ctx *last; struct task_struct *task; struct io_wq *io_wq; + /* + * Consumer cursor for ->task_list. Only popped by the task itself, + * or by ->fallback_work once the task can no longer run task_work. + */ + struct llist_node *task_head; struct file *registered_rings[IO_RINGFD_REG_MAX]; struct xarray xa; @@ -139,8 +144,13 @@ struct io_uring_task { atomic_t inflight_tracked; struct percpu_counter inflight; + /* drains ->task_list once the task can no longer run task_work */ + struct work_struct fallback_work; + struct { /* task_work */ - struct llist_head task_list; + struct mpscq task_list; + /* BIT(0) guards adding tw only once */ + unsigned long tw_pending; struct callback_head task_work; } ____cacheline_aligned_in_smp; }; diff --git a/io_uring/sqpoll.c b/io_uring/sqpoll.c index 46c12afec73e..2460bd605266 100644 --- a/io_uring/sqpoll.c +++ b/io_uring/sqpoll.c @@ -260,39 +260,29 @@ static bool io_sqd_handle_event(struct io_sq_data *sqd) } /* - * Run task_work, processing the retry_list first. The retry_list holds - * entries that we passed on in the previous run, if we had more task_work - * than we were asked to process. Newly queued task_work isn't run until the - * retry list has been fully processed. + * Run task_work, processing no more than max_entries at a time. If more + * than that is pending, it simply stays on the queue for the next run. */ -static unsigned int io_sq_tw(struct llist_node **retry_list, int max_entries) +static unsigned int io_sq_tw(int max_entries) { struct io_uring_task *tctx = current->io_uring; unsigned int count = 0; - if (*retry_list) { - *retry_list = io_handle_tw_list(*retry_list, &count, max_entries); - if (count >= max_entries) - goto out; - max_entries -= count; - } - *retry_list = tctx_task_work_run(tctx, max_entries, &count); -out: + tctx_task_work_run(tctx, max_entries, &count); if (task_work_pending(current)) task_work_run(); return count; } -static bool io_sq_tw_pending(struct llist_node *retry_list) +static bool io_sq_tw_pending(void) { struct io_uring_task *tctx = current->io_uring; - return retry_list || !llist_empty(&tctx->task_list); + return !mpscq_empty(&tctx->task_list); } static int io_sq_thread(void *data) { - struct llist_node *retry_list = NULL; struct io_sq_data *sqd = data; struct io_ring_ctx *ctx; unsigned long timeout = 0; @@ -347,7 +337,7 @@ static int io_sq_thread(void *data) if (!sqt_spin && (ret > 0 || !list_empty(&ctx->iopoll_list))) sqt_spin = true; } - if (io_sq_tw(&retry_list, IORING_TW_CAP_ENTRIES_VALUE)) + if (io_sq_tw(IORING_TW_CAP_ENTRIES_VALUE)) sqt_spin = true; list_for_each_entry(ctx, &sqd->ctx_list, sqd_list) { @@ -372,7 +362,7 @@ static int io_sq_thread(void *data) } prepare_to_wait(&sqd->wait, &wait, TASK_INTERRUPTIBLE); - if (!io_sqd_events_pending(sqd) && !io_sq_tw_pending(retry_list)) { + if (!io_sqd_events_pending(sqd) && !io_sq_tw_pending()) { bool needs_sched = true; list_for_each_entry(ctx, &sqd->ctx_list, sqd_list) { @@ -411,8 +401,8 @@ static int io_sq_thread(void *data) timeout = jiffies + sqd->sq_thread_idle; } - if (retry_list) - io_sq_tw(&retry_list, UINT_MAX); + if (io_sq_tw_pending()) + io_sq_tw(UINT_MAX); io_uring_cancel_generic(true, sqd); rcu_assign_pointer(sqd->thread, NULL); diff --git a/io_uring/tctx.c b/io_uring/tctx.c index 6af62ca9baba..8c6abc332b6c 100644 --- a/io_uring/tctx.c +++ b/io_uring/tctx.c @@ -103,7 +103,8 @@ __cold struct io_uring_task *io_uring_alloc_task_context(struct task_struct *tas init_waitqueue_head(&tctx->wait); atomic_set(&tctx->in_cancel, 0); atomic_set(&tctx->inflight_tracked, 0); - init_llist_head(&tctx->task_list); + mpscq_init(&tctx->task_list, &tctx->task_head); + INIT_WORK(&tctx->fallback_work, io_tctx_fallback_work); init_task_work(&tctx->task_work, tctx_task_work); return tctx; } diff --git a/io_uring/tw.c b/io_uring/tw.c index 79feeb4f671a..485d4c094a9f 100644 --- a/io_uring/tw.c +++ b/io_uring/tw.c @@ -46,46 +46,6 @@ static void ctx_flush_and_put(struct io_ring_ctx *ctx, io_tw_token_t tw) 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); - ts.cancel = io_should_terminate_tw(ctx); - } - INDIRECT_CALL_2(req->io_task_work.func, - io_poll_task_func, io_req_rw_complete, - (struct io_tw_req){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; -} - static __cold void __io_fallback_tw(struct llist_node *node, bool sync) { struct io_ring_ctx *last_ctx = NULL; @@ -114,43 +74,109 @@ static __cold void __io_fallback_tw(struct llist_node *node, bool sync) } } -static void io_fallback_tw(struct io_uring_task *tctx, bool sync) +void io_tctx_fallback_work(struct work_struct *work) { - struct llist_node *node = llist_del_all(&tctx->task_list); + struct io_uring_task *tctx = container_of(work, struct io_uring_task, + fallback_work); + struct llist_node *node, *first = NULL, **tail = &first; - __io_fallback_tw(node, sync); + /* see tctx_task_work() - a set bit must always have a run coming */ + clear_bit(0, &tctx->tw_pending); + smp_mb__after_atomic(); + + while (!mpscq_empty(&tctx->task_list)) { + node = mpscq_pop(&tctx->task_list, &tctx->task_head); + if (!node) { + /* a producer is mid-push, wait for it to link */ + cond_resched(); + continue; + } + *tail = node; + tail = &node->next; + } + *tail = NULL; + __io_fallback_tw(first, false); + put_task_struct(tctx->task); } -struct llist_node *tctx_task_work_run(struct io_uring_task *tctx, - unsigned int max_entries, - unsigned int *count) +static void io_fallback_tw(struct io_uring_task *tctx) { - struct llist_node *node; + /* + * The task ref both keeps ->task valid and, as __io_uring_free() is + * only called when the task itself is freed, ensures the tctx (and + * the queued work) stay around until the drain has run. + */ + get_task_struct(tctx->task); + if (!queue_work(system_unbound_wq, &tctx->fallback_work)) + put_task_struct(tctx->task); +} - node = llist_del_all(&tctx->task_list); - if (node) { - node = llist_reverse_order(node); - node = io_handle_tw_list(node, count, max_entries); +/* + * Run queued task_work, processing no more than max_entries, with the number + * of entries processed added to *count. If more entries than max_entries are + * available, the remainder simply stay on the queue for the next run. + */ +void tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries, + unsigned int *count) +{ + struct io_ring_ctx *ctx = NULL; + struct io_tw_state ts = { }; + + while (*count < max_entries) { + struct llist_node *node = mpscq_pop(&tctx->task_list, + &tctx->task_head); + struct io_kiocb *req; + + if (!node) { + if (mpscq_empty(&tctx->task_list)) + break; + /* + * A producer has published a node but hasn't + * linked it into the queue yet (see mpscq_pop()). + * Give it a chance to finish rather than spinning, + * and don't sit on the ctx lock while doing so. + */ + ctx_flush_and_put(ctx, ts); + ctx = NULL; + cond_resched(); + continue; + } + 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); + ts.cancel = io_should_terminate_tw(ctx); + } + INDIRECT_CALL_2(req->io_task_work.func, + io_poll_task_func, io_req_rw_complete, + (struct io_tw_req){req}, ts); + (*count)++; + if (unlikely(need_resched())) { + ctx_flush_and_put(ctx, ts); + ctx = NULL; + cond_resched(); + } } + ctx_flush_and_put(ctx, ts); /* 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); + clear_bit(0, &tctx->tw_pending); + smp_mb__after_atomic(); + tctx_task_work_run(tctx, UINT_MAX, &count); } /* @@ -228,7 +254,7 @@ void io_req_normal_work_add(struct io_kiocb *req) 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)) + if (!mpscq_push(&tctx->task_list, &req->io_task_work.node)) return; /* @@ -244,10 +270,14 @@ void io_req_normal_work_add(struct io_kiocb *req) return; } + /* task_work must only be added once */ + if (test_and_set_bit(0, &tctx->tw_pending)) + return; + if (likely(!task_work_add(tctx->task, &tctx->task_work, ctx->notify_method))) return; - io_fallback_tw(tctx, false); + io_fallback_tw(tctx); } void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags) diff --git a/io_uring/tw.h b/io_uring/tw.h index f42db5fdbded..387e52004da8 100644 --- a/io_uring/tw.h +++ b/io_uring/tw.h @@ -25,8 +25,8 @@ static inline bool io_should_terminate_tw(struct io_ring_ctx *ctx) } void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags); -struct llist_node *io_handle_tw_list(struct llist_node *node, unsigned int *count, unsigned int max_entries); void tctx_task_work(struct callback_head *cb); +void io_tctx_fallback_work(struct work_struct *work); int io_run_local_work(struct io_ring_ctx *ctx, int min_events, int max_events); int io_run_task_work_sig(struct io_ring_ctx *ctx); @@ -36,7 +36,7 @@ int io_run_local_work_locked(struct io_ring_ctx *ctx, int min_events); void io_req_local_work_add(struct io_kiocb *req, unsigned flags); void io_req_normal_work_add(struct io_kiocb *req); -struct llist_node *tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries, unsigned int *count); +void tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries, unsigned int *count); static inline void __io_req_task_work_add(struct io_kiocb *req, unsigned flags) { From df58c2161684b2a1d8db752339f0c0629ec68be5 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Thu, 11 Jun 2026 11:41:25 -0600 Subject: [PATCH 28/31] io_uring: run the tctx task_work fallback directly The fallback work drains the tctx queue only to redistribute the entries into the per-ctx fallback lists, bouncing them through a second (per-ctx) work item before they finally run. That made sense when the producer side did the draining and could be in any context, but the fallback work is a regular process context kworker: it can just run the entries itself. Reuse the normal run loop - if run from the fallback kernel thread, ts.cancel will get set, and the work terminated. Signed-off-by: Jens Axboe --- io_uring/tw.c | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/io_uring/tw.c b/io_uring/tw.c index 485d4c094a9f..a5ed57ec34e6 100644 --- a/io_uring/tw.c +++ b/io_uring/tw.c @@ -78,24 +78,18 @@ void io_tctx_fallback_work(struct work_struct *work) { struct io_uring_task *tctx = container_of(work, struct io_uring_task, fallback_work); - struct llist_node *node, *first = NULL, **tail = &first; + unsigned int count = 0; /* see tctx_task_work() - a set bit must always have a run coming */ clear_bit(0, &tctx->tw_pending); smp_mb__after_atomic(); - while (!mpscq_empty(&tctx->task_list)) { - node = mpscq_pop(&tctx->task_list, &tctx->task_head); - if (!node) { - /* a producer is mid-push, wait for it to link */ - cond_resched(); - continue; - } - *tail = node; - tail = &node->next; - } - *tail = NULL; - __io_fallback_tw(first, false); + /* + * Run the entries directly. We're in PF_KTHRED context, hence + * io_should_terminate_tw() is true and they will be marked as + * canceled. + */ + tctx_task_work_run(tctx, UINT_MAX, &count); put_task_struct(tctx->task); } @@ -161,8 +155,13 @@ void tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries, } ctx_flush_and_put(ctx, ts); - /* relaxed read is enough as only the task itself sets ->in_cancel */ - if (unlikely(atomic_read(&tctx->in_cancel))) + /* + * Relaxed read is enough as only the task itself sets ->in_cancel. + * The tctx may also be drained by io_tctx_fallback_work(), in which + * case current is a kworker that has no tctx refs to drop. + */ + if (unlikely(atomic_read(&tctx->in_cancel)) && + current->io_uring == tctx) io_uring_drop_tctx_refs(current); trace_io_uring_task_work_run(tctx, *count); From 576cce91480a949f5b83578300f37023b933e0a2 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Thu, 11 Jun 2026 11:44:47 -0600 Subject: [PATCH 29/31] io_uring: remove the per-ctx fallback task_work machinery With the tctx fallback running its entries directly, the per-ctx fallback work has a single user left: moving local (DEFER_TASKRUN) task_work entries out of a ring that is going away. Both of its call sites are process context and don't hold ->uring_lock, the same conditions the deferred fallback work itself ran under - so run the entries in cancel mode right there instead, and rename the helper to io_cancel_local_task_work() to match what it now does. With that, ->fallback_llist, ->fallback_work, io_fallback_req_func() and __io_fallback_tw() can all go away, along with the fallback work flushing in the ring exit and cancel paths. Requests that get orphaned by an exiting task now run via the tctx fallback work, which the ring exit side implicitly waits on through the ctx refs those requests hold. Signed-off-by: Jens Axboe --- include/linux/io_uring_types.h | 2 - io_uring/cancel.c | 2 - io_uring/io_uring.c | 7 +--- io_uring/tw.c | 67 +++++++--------------------------- io_uring/tw.h | 3 +- 5 files changed, 16 insertions(+), 65 deletions(-) diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h index f511e96865b6..6415a3353ee0 100644 --- a/include/linux/io_uring_types.h +++ b/include/linux/io_uring_types.h @@ -498,8 +498,6 @@ struct io_ring_ctx { struct mutex tctx_lock; /* ctx exit and cancelation */ - struct llist_head fallback_llist; - struct delayed_work fallback_work; struct work_struct exit_work; struct completion ref_comp; diff --git a/io_uring/cancel.c b/io_uring/cancel.c index 5e5eb9cfc7cd..b0259e74f678 100644 --- a/io_uring/cancel.c +++ b/io_uring/cancel.c @@ -565,8 +565,6 @@ __cold bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx, ret |= io_kill_timeouts(ctx, tctx, cancel_all); if (tctx) ret |= io_run_task_work() > 0; - else - ret |= flush_delayed_work(&ctx->fallback_work); return ret; } diff --git a/io_uring/io_uring.c b/io_uring/io_uring.c index 0809fc70c91d..5e687bbb973e 100644 --- a/io_uring/io_uring.c +++ b/io_uring/io_uring.c @@ -289,7 +289,6 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p) #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); @@ -1192,7 +1191,7 @@ __cold void io_iopoll_try_reap_events(struct io_ring_ctx *ctx) mutex_unlock(&ctx->uring_lock); if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) - io_move_task_work_from_local(ctx); + io_cancel_local_task_work(ctx); } static int io_iopoll_check(struct io_ring_ctx *ctx, unsigned int min_events) @@ -2334,7 +2333,7 @@ static __cold void io_ring_exit_work(struct work_struct *work) /* The SQPOLL thread never reaches this path */ do { if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) - io_move_task_work_from_local(ctx); + io_cancel_local_task_work(ctx); cond_resched(); } while (io_uring_try_cancel_requests(ctx, NULL, true, false)); @@ -2420,8 +2419,6 @@ static __cold void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx) 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_dfl_wq to avoid spawning tons of event kworkers diff --git a/io_uring/tw.c b/io_uring/tw.c index a5ed57ec34e6..e74372233f40 100644 --- a/io_uring/tw.c +++ b/io_uring/tw.c @@ -16,24 +16,6 @@ #include "wait.h" #include "mpscq.h" -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); - ts.cancel = io_should_terminate_tw(ctx); - llist_for_each_entry_safe(req, tmp, node, io_task_work.node) - req->io_task_work.func((struct io_tw_req){req}, ts); - io_submit_flush_completions(ctx); - mutex_unlock(&ctx->uring_lock); - percpu_ref_put(&ctx->refs); -} - static void ctx_flush_and_put(struct io_ring_ctx *ctx, io_tw_token_t tw) { if (!ctx) @@ -46,34 +28,6 @@ static void ctx_flush_and_put(struct io_ring_ctx *ctx, io_tw_token_t tw) percpu_ref_put(&ctx->refs); } -static __cold void __io_fallback_tw(struct llist_node *node, bool sync) -{ - 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 (last_ctx != req->ctx) { - if (last_ctx) { - if (sync) - 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, &last_ctx->fallback_llist)) - schedule_delayed_work(&last_ctx->fallback_work, 1); - } - - if (last_ctx) { - if (sync) - flush_delayed_work(&last_ctx->fallback_work); - percpu_ref_put(&last_ctx->refs); - } -} - void io_tctx_fallback_work(struct work_struct *work) { struct io_uring_task *tctx = container_of(work, struct io_uring_task, @@ -286,29 +240,34 @@ void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags) __io_req_task_work_add(req, flags); } -void __cold io_move_task_work_from_local(struct io_ring_ctx *ctx) +void __cold io_cancel_local_task_work(struct io_ring_ctx *ctx) { - struct llist_node *node, *first = NULL, **tail = &first; + struct io_tw_state ts = { .cancel = true }; + struct llist_node *node; /* * The work list consumer side is serialized by ->uring_lock, see * __io_run_local_work(). Grab it to guard against racing with normal - * task_work running, as the task may be exiting. + * task_work running, as the task may be exiting. The ring is going + * away, run the entries in cancel mode right here - the callers + * provide the same process context the per-ctx fallback work that + * they were previously punted to ran in. */ guard(mutex)(&ctx->uring_lock); while (!mpscq_empty(&ctx->work_list)) { + struct io_kiocb *req; + node = mpscq_pop(&ctx->work_list, &ctx->work_head); if (!node) { /* a producer is mid-push, wait for it to link */ - cpu_relax(); + cond_resched(); continue; } - *tail = node; - tail = &node->next; + req = container_of(node, struct io_kiocb, io_task_work.node); + req->io_task_work.func((struct io_tw_req){req}, ts); } - *tail = NULL; - __io_fallback_tw(first, false); + io_submit_flush_completions(ctx); } static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events, diff --git a/io_uring/tw.h b/io_uring/tw.h index 387e52004da8..3ade5ad577fd 100644 --- a/io_uring/tw.h +++ b/io_uring/tw.h @@ -30,8 +30,7 @@ void io_tctx_fallback_work(struct work_struct *work); int io_run_local_work(struct io_ring_ctx *ctx, int min_events, int max_events); int io_run_task_work_sig(struct io_ring_ctx *ctx); -__cold void io_fallback_req_func(struct work_struct *work); -__cold void io_move_task_work_from_local(struct io_ring_ctx *ctx); +__cold void io_cancel_local_task_work(struct io_ring_ctx *ctx); int io_run_local_work_locked(struct io_ring_ctx *ctx, int min_events); void io_req_local_work_add(struct io_kiocb *req, unsigned flags); From 856950322678906a0947c31205dd652830f9c628 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Fri, 29 May 2026 20:03:47 -0600 Subject: [PATCH 30/31] io_uring/net: make POLL_FIRST receive side checks consistent io_recv() and io_recvzc() are the odd ones out, as they checks for whether POLL_FIRST should be honored before checking if the file is a socket. It doesn't really matter, but might as well make it consistent across all receive and send types. Signed-off-by: Jens Axboe --- io_uring/net.c | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/io_uring/net.c b/io_uring/net.c index 5ae538b3c0f3..7deb62e3b4c0 100644 --- a/io_uring/net.c +++ b/io_uring/net.c @@ -1216,14 +1216,14 @@ int io_recv(struct io_kiocb *req, unsigned int issue_flags) bool force_nonblock = issue_flags & IO_URING_F_NONBLOCK; bool mshot_finished; - if (!(req->flags & REQ_F_POLLED) && - (sr->flags & IORING_RECVSEND_POLL_FIRST)) - return -EAGAIN; - sock = sock_from_file(req->file); if (unlikely(!sock)) return -ENOTSOCK; + if (!(req->flags & REQ_F_POLLED) && + (sr->flags & IORING_RECVSEND_POLL_FIRST)) + return -EAGAIN; + flags = sr->msg_flags; if (force_nonblock) flags |= MSG_DONTWAIT; @@ -1328,14 +1328,14 @@ int io_recvzc(struct io_kiocb *req, unsigned int issue_flags) unsigned int len; int ret; - if (!(req->flags & REQ_F_POLLED) && - (zc->flags & IORING_RECVSEND_POLL_FIRST)) - return -EAGAIN; - sock = sock_from_file(req->file); if (unlikely(!sock)) return -ENOTSOCK; + if (!(req->flags & REQ_F_POLLED) && + (zc->flags & IORING_RECVSEND_POLL_FIRST)) + return -EAGAIN; + len = zc->len; ret = io_zcrx_recv(req, zc->ifq, sock, 0, issue_flags, &zc->len); if (len && zc->len == 0) { From d9b710f683dc68b5c0b7dd0c6c64aeb5d27a1ac4 Mon Sep 17 00:00:00 2001 From: Pavel Begunkov Date: Fri, 12 Jun 2026 18:36:22 +0100 Subject: [PATCH 31/31] io_uring/bpf-ops: add a separate maintainer entry Add a maintainer entry for io_uring bpf struct_ops related files. Signed-off-by: Pavel Begunkov Link: https://patch.msgid.link/d89f3b89e77b09a18daa45476fd1a40f2ee253cd.1780930463.git.asml.silence@gmail.com Signed-off-by: Jens Axboe --- MAINTAINERS | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/MAINTAINERS b/MAINTAINERS index b2040011a386..31e70d5c0a28 100644 --- a/MAINTAINERS +++ b/MAINTAINERS @@ -13530,6 +13530,14 @@ T: git git://git.kernel.org/pub/scm/linux/kernel/git/axboe/linux.git S: Maintained F: io_uring/zcrx.* +IO_URING BPF-OPS +M: Pavel Begunkov +L: io-uring@vger.kernel.org +T: git git://git.kernel.org/pub/scm/linux/kernel/git/axboe/linux.git +S: Maintained +F: io_uring/bpf-ops.* +F: io_uring/loop.* + IPMI SUBSYSTEM M: Corey Minyard L: openipmi-developer@lists.sourceforge.net (moderated for non-subscribers)