* [RFC PATCH net-next 01/12] net: add an exclusive TCP receive claim
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
@ 2026-11-10 13:19 ` Alex Markuze
2026-11-10 13:19 ` [RFC PATCH net-next 02/12] io_uring/opaque: add data structures and the UAPI header Alex Markuze
` (10 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:19 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
The io_uring opaque object collector, added later in this series,
consumes the receive queue of a TCP socket through tcp_read_sock() with
its own actor. It reports stream offsets that count from the first byte
that was unread when it attached. If another reader consumes bytes while
the collector is attached, those offsets no longer match the data, and
the application selects the wrong ranges.
Add a receive owner to struct sock: a pointer to a struct sock_rx_owner
that names a tcp_read_sock() actor and its data. The owner is published
and cleared under the socket lock and sk_callback_lock. While an owner is
set:
- tcp_read_sock() and tcp_read_sock_noack() return -EBUSY unless the
caller passes the owning actor and data. This also stops splice().
- tcp_recvmsg_locked(), tcp_zerocopy_receive() and tcp_read_skb() return
-EBUSY.
- sk_wait_data() also returns when an owner appears. A reader that sleeps
in tcp_recvmsg_locked() then stops with the bytes that it copied, or
with -EBUSY. The claimant wakes sleeping readers after it publishes the
owner.
- tcp_disconnect(), connect(AF_UNSPEC), tcp_set_ulp(), sk_psock_init()
and TCP_REPAIR return -EBUSY, because each of them resets or takes
over the receive sequence. TCP_REPAIR checks the owner before its
permission check, so a caller without the permission also gets -EBUSY.
- sk_clone() starts the child socket without an owner.
CONFIG_SOCK_RX_OWNER is a hidden option that the io_uring feature
selects. Without it, struct sock does not change and the checks compile
away. With it, struct sock grows by one pointer. Each affected receive
path reads the pointer once with READ_ONCE(), and tcp_recvmsg_locked()
reads it once per loop iteration.
An optional data_ready callback notifies the owner independently of
ordinary TCP low-water readiness. TCP invokes it under the read side
of sk_callback_lock, which pins the owner and its data; the callback
must not sleep. With receive ownership disabled, the helper is empty.
This patch adds no user of the claim or callback.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
include/linux/net.h | 8 ++++++++
include/net/sock.h | 46 ++++++++++++++++++++++++++++++++++++++++++++
net/Kconfig | 7 +++++++
net/core/skmsg.c | 4 ++++
net/core/sock.c | 6 +++++-
net/ipv4/af_inet.c | 2 ++
net/ipv4/tcp.c | 22 ++++++++++++++++++++-
net/ipv4/tcp_input.c | 1 +
net/ipv4/tcp_ulp.c | 2 ++
9 files changed, 96 insertions(+), 2 deletions(-)
diff --git a/include/linux/net.h b/include/linux/net.h
index e2a866fbcfa5..d8fe1812e02b 100644
--- a/include/linux/net.h
+++ b/include/linux/net.h
@@ -230,6 +230,14 @@ struct sk_buff;
struct proto_accept_arg;
typedef int (*sk_read_actor_t)(read_descriptor_t *, struct sk_buff *,
unsigned int, size_t);
+
+/* Published under socket and callback locks; controls the receive sequence. */
+struct sock_rx_owner {
+ sk_read_actor_t actor;
+ void *data;
+ /* TCP receive context, under sk_callback_lock: must not sleep. */
+ void (*data_ready)(struct sock *sk, void *data);
+};
typedef int (*skb_read_actor_t)(struct sock *, struct sk_buff *);
diff --git a/include/net/sock.h b/include/net/sock.h
index 173b37b2b8bb..4da1796b3b8c 100644
--- a/include/net/sock.h
+++ b/include/net/sock.h
@@ -326,6 +326,7 @@ struct sk_filter;
* @sk_zckey: counter to order MSG_ZEROCOPY notifications
* @sk_socket: Identd and reporting IO signals
* @sk_user_data: RPC layer private data. Write-protected by @sk_callback_lock.
+ * @sk_rx_owner: Exclusive receive owner under socket and callback locks.
* @sk_frag: cached page frag
* @sk_peek_off: current peek_offset value
* @sk_send_head: front of stuff to transmit
@@ -575,6 +576,9 @@ struct sock {
u8 sk_bpf_cb_flags;
void *sk_user_data;
+#ifdef CONFIG_SOCK_RX_OWNER
+ struct sock_rx_owner *sk_rx_owner;
+#endif
#ifdef CONFIG_SECURITY
void *sk_security;
#endif
@@ -617,6 +621,48 @@ enum sk_pacing {
SK_PACING_FQ = 2,
};
+static inline bool sock_rx_owned(const struct sock *sk)
+{
+#ifdef CONFIG_SOCK_RX_OWNER
+ return unlikely(READ_ONCE(sk->sk_rx_owner));
+#else
+ return false;
+#endif
+}
+
+/*
+ * Notify the receive owner independently of ordinary socket low-water wakes.
+ * The callback lock pins both owner and data until the callback returns;
+ * publication and removal take its write side while owning the socket.
+ */
+static inline void sock_rx_owner_data_ready(struct sock *sk)
+{
+#ifdef CONFIG_SOCK_RX_OWNER
+ struct sock_rx_owner *owner;
+
+ if (!sock_rx_owned(sk))
+ return;
+ read_lock_bh(&sk->sk_callback_lock);
+ owner = sk->sk_rx_owner;
+ if (owner && owner->data_ready)
+ owner->data_ready(sk, owner->data);
+ read_unlock_bh(&sk->sk_callback_lock);
+#endif
+}
+
+static inline bool sock_rx_owner_conflict(const struct sock *sk,
+ read_descriptor_t *desc,
+ sk_read_actor_t actor)
+{
+#ifdef CONFIG_SOCK_RX_OWNER
+ const struct sock_rx_owner *owner = READ_ONCE(sk->sk_rx_owner);
+
+ return owner && (owner->actor != actor || owner->data != desc->arg.data);
+#else
+ return false;
+#endif
+}
+
/* flag bits in sk_user_data
*
* - SK_USER_DATA_NOCOPY: Pointer stored in sk_user_data might
diff --git a/net/Kconfig b/net/Kconfig
index 76ab44aa439a..6ece51ab26ba 100644
--- a/net/Kconfig
+++ b/net/Kconfig
@@ -26,6 +26,13 @@ menuconfig NET
if NET
+config SOCK_RX_OWNER
+ bool
+ depends on INET
+ help
+ Allow a TCP stream actor to claim exclusive receive sequence access.
+ Receive paths reject competing readers while the claim is held.
+
config WANT_COMPAT_NETLINK_MESSAGES
bool
help
diff --git a/net/core/skmsg.c b/net/core/skmsg.c
index df385a5a961e..5c753bcb808c 100644
--- a/net/core/skmsg.c
+++ b/net/core/skmsg.c
@@ -754,6 +754,10 @@ struct sk_psock *sk_psock_init(struct sock *sk, int node)
write_lock_bh(&sk->sk_callback_lock);
+ if (sock_rx_owned(sk)) {
+ psock = ERR_PTR(-EBUSY);
+ goto out;
+ }
if (sk_is_inet(sk) && inet_csk_has_ulp(sk)) {
psock = ERR_PTR(-EINVAL);
goto out;
diff --git a/net/core/sock.c b/net/core/sock.c
index e7405be8941f..3fcdc56e624e 100644
--- a/net/core/sock.c
+++ b/net/core/sock.c
@@ -2614,6 +2614,9 @@ struct sock *sk_clone(const struct sock *sk, const gfp_t priority,
*/
if (sk_user_data_is_nocopy(newsk))
newsk->sk_user_data = NULL;
+#ifdef CONFIG_SOCK_RX_OWNER
+ newsk->sk_rx_owner = NULL;
+#endif
newsk->sk_err = 0;
newsk->sk_err_soft = 0;
@@ -3389,7 +3392,8 @@ int sk_wait_data(struct sock *sk, long *timeo, const struct sk_buff *skb)
add_wait_queue(sk_sleep(sk), &wait);
sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk);
- rc = sk_wait_event(sk, timeo, skb_peek_tail(&sk->sk_receive_queue) != skb, &wait);
+ rc = sk_wait_event(sk, timeo, sock_rx_owned(sk) ||
+ skb_peek_tail(&sk->sk_receive_queue) != skb, &wait);
sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk);
remove_wait_queue(sk_sleep(sk), &wait);
return rc;
diff --git a/net/ipv4/af_inet.c b/net/ipv4/af_inet.c
index 7677b0343abd..54c30a59bbce 100644
--- a/net/ipv4/af_inet.c
+++ b/net/ipv4/af_inet.c
@@ -649,6 +649,8 @@ int __inet_stream_connect(struct socket *sock, struct sockaddr_unsized *uaddr,
return -EINVAL;
if (uaddr->sa_family == AF_UNSPEC) {
+ if (sock_rx_owned(sk))
+ return -EBUSY;
sk->sk_disconnects++;
err = sk->sk_prot->disconnect(sk, flags);
sock->state = err ? SS_DISCONNECTING : SS_UNCONNECTED;
diff --git a/net/ipv4/tcp.c b/net/ipv4/tcp.c
index b9de7c302ba8..6de6ce153fda 100644
--- a/net/ipv4/tcp.c
+++ b/net/ipv4/tcp.c
@@ -1753,6 +1753,8 @@ out:
int tcp_read_sock(struct sock *sk, read_descriptor_t *desc,
sk_read_actor_t recv_actor)
{
+ if (sock_rx_owner_conflict(sk, desc, recv_actor))
+ return -EBUSY;
return __tcp_read_sock(sk, desc, recv_actor, false,
&tcp_sk(sk)->copied_seq);
}
@@ -1762,6 +1764,8 @@ int tcp_read_sock_noack(struct sock *sk, read_descriptor_t *desc,
sk_read_actor_t recv_actor, bool noack,
u32 *copied_seq)
{
+ if (sock_rx_owner_conflict(sk, desc, recv_actor))
+ return -EBUSY;
return __tcp_read_sock(sk, desc, recv_actor, noack, copied_seq);
}
@@ -1770,6 +1774,8 @@ int tcp_read_skb(struct sock *sk, skb_read_actor_t recv_actor)
struct sk_buff *skb;
int copied = 0;
+ if (sock_rx_owned(sk))
+ return -EBUSY;
if (sk->sk_state == TCP_LISTEN)
return -ENOTCONN;
@@ -2224,6 +2230,8 @@ static int tcp_zerocopy_receive(struct sock *sk,
zc->copybuf_len = 0;
zc->msg_flags = 0;
+ if (sock_rx_owned(sk))
+ return -EBUSY;
if (address & (PAGE_SIZE - 1) || address != zc->address)
return -EINVAL;
@@ -2678,6 +2686,8 @@ static int tcp_recvmsg_locked(struct sock *sk, struct msghdr *msg, size_t len,
u32 peek_offset = 0;
u32 urg_hole = 0;
+ if (sock_rx_owned(sk))
+ return -EBUSY;
err = -ENOTCONN;
if (sk->sk_state == TCP_LISTEN)
goto out;
@@ -2717,6 +2727,11 @@ static int tcp_recvmsg_locked(struct sock *sk, struct msghdr *msg, size_t len,
do {
u32 offset;
+ /* Ownership may have changed while this reader slept for data. */
+ if (sock_rx_owned(sk)) {
+ copied = copied ?: -EBUSY;
+ break;
+ }
/* Are we at urgent data? Stop if we have read anything or have SIGURG pending. */
if (unlikely(tp->urg_data) && tp->urg_seq == *seq) {
if (copied)
@@ -3374,6 +3389,9 @@ int tcp_disconnect(struct sock *sk, int flags)
struct request_sock *req;
u32 seq;
+ if (sock_rx_owned(sk))
+ return -EBUSY;
+
if (old_state != TCP_CLOSE)
tcp_set_state(sk, TCP_CLOSE);
@@ -3988,7 +4006,9 @@ int do_tcp_setsockopt(struct sock *sk, int level, int optname,
break;
case TCP_REPAIR:
- if (!tcp_can_repair_sock(sk))
+ if (sock_rx_owned(sk))
+ err = -EBUSY;
+ else if (!tcp_can_repair_sock(sk))
err = -EPERM;
else if (val == TCP_REPAIR_ON) {
tp->repair = 1;
diff --git a/net/ipv4/tcp_input.c b/net/ipv4/tcp_input.c
index e1f2d9765f81..2d17fcdf73dd 100644
--- a/net/ipv4/tcp_input.c
+++ b/net/ipv4/tcp_input.c
@@ -5637,6 +5637,7 @@ err:
void tcp_data_ready(struct sock *sk)
{
+ sock_rx_owner_data_ready(sk);
if (tcp_epollin_ready(sk, sk->sk_rcvlowat) || sock_flag(sk, SOCK_DONE))
READ_ONCE(sk->sk_data_ready)(sk);
}
diff --git a/net/ipv4/tcp_ulp.c b/net/ipv4/tcp_ulp.c
index b58045df101e..77abc21bcb63 100644
--- a/net/ipv4/tcp_ulp.c
+++ b/net/ipv4/tcp_ulp.c
@@ -164,6 +164,8 @@ int tcp_set_ulp(struct sock *sk, const char *name)
sock_owned_by_me(sk);
+ if (sock_rx_owned(sk))
+ return -EBUSY;
ulp_ops = __tcp_ulp_find_autoload(name);
if (!ulp_ops)
return -ENOENT;
base-commit: 4dd16e31135ea8437a720222df34e8745f385d6d
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 02/12] io_uring/opaque: add data structures and the UAPI header
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
2026-11-10 13:19 ` [RFC PATCH net-next 01/12] net: add an exclusive TCP receive claim Alex Markuze
@ 2026-11-10 13:19 ` Alex Markuze
2026-11-10 13:19 ` [RFC PATCH net-next 03/12] io_uring/opaque: add backing accounting and immutable versions Alex Markuze
` (9 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:19 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Define immutable versions, backing extents, stores and request state.
TX queues record FIFO positions ordered by enqueue under the store's TX
lock, including positions with reference acquisition still pending, and
carry a busy socket identity pin.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
include/linux/io_uring_types.h | 10 +
include/uapi/linux/io_uring/opaque.h | 126 +++++++++++++
io_uring/opaque.h | 155 +++++++++++++++
io_uring/opaque_internal.h | 269 +++++++++++++++++++++++++++
4 files changed, 560 insertions(+)
create mode 100644 include/uapi/linux/io_uring/opaque.h
create mode 100644 io_uring/opaque.h
create mode 100644 io_uring/opaque_internal.h
diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h
index 9140c21ca819..198c6bdb3e2c 100644
--- a/include/linux/io_uring_types.h
+++ b/include/linux/io_uring_types.h
@@ -412,6 +412,9 @@ struct io_ring_ctx {
struct io_alloc_cache netmsg_cache;
struct io_alloc_cache rw_cache;
struct io_alloc_cache cmd_cache;
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+ struct io_alloc_cache opaque_cache;
+#endif
int (*loop_step)(struct iou_ctx *,
struct iou_loop_params *);
@@ -421,6 +424,9 @@ struct io_ring_ctx {
* ->uring_cmd() by io_uring_cmd_insert_cancelable()
*/
struct hlist_head cancelable_uring_cmd;
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+ struct list_head opaque_waits;
+#endif
/*
* For Hybrid IOPOLL, runtime in hybrid polling, without
* scheduling time
@@ -491,6 +497,10 @@ struct io_ring_ctx {
/* Stores zcrx object pointers of type struct io_zcrx_ifq */
struct xarray zcrx_ctxs;
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+ struct xarray opaque_stores;
+ bool opaque_stores_dying;
+#endif
/* Used for accounting references on pages in registered buffers */
struct xarray hpage_acct;
diff --git a/include/uapi/linux/io_uring/opaque.h b/include/uapi/linux/io_uring/opaque.h
new file mode 100644
index 000000000000..692daa7d7e32
--- /dev/null
+++ b/include/uapi/linux/io_uring/opaque.h
@@ -0,0 +1,126 @@
+/* SPDX-License-Identifier: (GPL-2.0 WITH Linux-syscall-note) OR MIT */
+#ifndef LINUX_IO_URING_OPAQUE_H
+#define LINUX_IO_URING_OPAQUE_H
+
+#include <linux/types.h>
+
+/* IORING_REGISTER_OPAQUE_STORE, nr_args == 1. size includes reserved bytes. */
+struct io_uring_opaque_register {
+ __u32 size;
+ __u32 flags;
+ __s32 fd;
+ __u32 store_id;
+ __u64 config;
+ __u32 config_size;
+ __u32 __resv0;
+ __u64 __resv[2];
+};
+
+/* config/config_size select a size-versioned configuration on creation. */
+struct io_uring_opaque_config {
+ __u64 hard_limit;
+ __u64 compact_headroom;
+ __u32 max_object_size;
+ __u32 max_objects;
+ __u32 max_streams;
+ __u32 max_extents;
+ __u32 flags;
+ __u32 __resv0;
+ __u64 __resv[2];
+ __u64 high_watermark;
+ __u64 low_watermark;
+ __u64 reply_reserve;
+ __u32 __resv2;
+ __u32 __resv1;
+};
+
+/* IORING_OP_OPAQUE_OBJ: sqe->ioprio selects the operation. */
+enum io_uring_opaque_op {
+ /* Nonconsuming available window; len is capacity, never WAITALL. */
+ IORING_OPAQUE_INSPECT,
+ /* Complete-only object-ready CQE for the declared future stream range. */
+ IORING_OPAQUE_RECV_OBJECT,
+ IORING_OPAQUE_DISCARD,
+ /* Copy from an already complete immutable object. */
+ IORING_OPAQUE_READ_OBJECT,
+ IORING_OPAQUE_FREE,
+ IORING_OPAQUE_STAT,
+ IORING_OPAQUE_OBJECT_STAT,
+ IORING_OPAQUE_STREAM_STAT,
+ IORING_OPAQUE_STREAM_CLOSE,
+ IORING_OPAQUE_COMPACT,
+ IORING_OPAQUE_SET_POLICY,
+ /* Copy userspace bytes into a new complete immutable object. */
+ IORING_OPAQUE_CREATE,
+ /* Nonconsuming contiguous byte count, without copying payload. */
+ IORING_OPAQUE_AVAILABLE,
+};
+
+/* Optional SEND sqe->addr3 framing; size includes zero reserved extensions. */
+struct io_uring_opaque_frame {
+ __u32 size;
+ __u32 flags;
+ __u64 prefix;
+ __u64 suffix;
+ __u32 prefix_length;
+ __u32 suffix_length;
+ __u64 __resv[2];
+};
+
+/* Combined copied prefix and suffix, independently of object body length. */
+#define IORING_OPAQUE_FRAME_MAX (64U * 1024U)
+
+#define IORING_OPAQUE_OBJECT_F_COMPACTED (1U << 0)
+#define IORING_OPAQUE_OBJECT_F_COMPACTING (1U << 1)
+#define IORING_OPAQUE_OBJECT_F_FORWARD_ONLY (1U << 2)
+#define IORING_OPAQUE_STREAM_F_EOF (1U << 0)
+
+enum io_uring_opaque_memory_mode {
+ IORING_OPAQUE_MEMORY_NORMAL,
+ IORING_OPAQUE_MEMORY_LOWMEM,
+};
+
+struct io_uring_opaque_stat {
+ __u32 size;
+ __u32 __resv;
+ __u64 backing_bytes;
+ __u64 hard_limit;
+ __u64 copied_bytes;
+ __u64 compacted_bytes;
+ __u64 compact_count;
+ __u32 objects;
+ __u32 streams;
+ __u32 extents;
+ __u32 memory_mode;
+ __u64 framing_bytes;
+ __u64 framing_copied_bytes;
+};
+
+struct io_uring_opaque_object_stat {
+ __u32 size;
+ __u32 flags;
+ __u64 length;
+ __u64 backing_bytes;
+ __u32 extents;
+ __u32 __resv;
+};
+
+struct io_uring_opaque_stream_stat {
+ __u32 size;
+ __u32 flags;
+ __u64 rx_next;
+ __u64 undecided_bytes;
+ __s32 error;
+ __u32 __resv;
+};
+
+struct io_uring_opaque_policy {
+ __u32 flags;
+ __u32 __resv0;
+ __u64 bytes_per_second;
+ __u64 burst_bytes;
+ __u64 max_temporary_bytes;
+ __u64 __resv[2];
+};
+
+#endif /* LINUX_IO_URING_OPAQUE_H */
diff --git a/io_uring/opaque.h b/io_uring/opaque.h
new file mode 100644
index 000000000000..a7210f4bee5e
--- /dev/null
+++ b/io_uring/opaque.h
@@ -0,0 +1,155 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+#ifndef IO_URING_OPAQUE_H
+#define IO_URING_OPAQUE_H
+
+#include <linux/io_uring_types.h>
+#include <linux/rbtree.h>
+#include <linux/uio.h>
+#include <uapi/linux/io_uring/opaque.h>
+
+struct io_opaque_store;
+struct io_cancel_data;
+struct io_uring_opaque_config;
+
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+struct io_opaque_stream;
+struct io_opaque_data;
+struct io_opaque_tx;
+struct io_opaque_vec;
+
+enum io_opaque_phase {
+ IO_OPAQUE_IDLE,
+ IO_OPAQUE_RANGE_WAIT,
+ IO_OPAQUE_BUDGET_POLL,
+ IO_OPAQUE_SEND_WAIT,
+ IO_OPAQUE_COPY_WORK,
+ IO_OPAQUE_QUEUED,
+};
+
+struct io_opaque_req {
+ struct io_kiocb *req;
+ struct io_opaque_store *store;
+ struct io_opaque_stream *stream;
+ struct io_opaque_data *data;
+ struct list_head wait;
+ struct list_head cancel;
+ u64 handle;
+ u64 offset;
+ u64 progress;
+ u64 need_bytes;
+ /* Unallocated CREATE charge; must coexist with the reserved slot. */
+ u64 create_reserved;
+ unsigned long retry_at;
+ void __user *addr;
+ u32 length;
+ int result;
+ u16 op;
+ u16 collect_flags;
+ enum io_opaque_phase phase;
+ bool canceled;
+ bool slot_reserved;
+ bool consumed;
+ bool resolved;
+ bool budget_reported;
+ union {
+ struct {
+ struct rb_node decision;
+ struct list_head read;
+ u32 slot;
+ };
+ struct {
+ struct iov_iter iter;
+ struct list_head send;
+ struct io_opaque_tx *tx;
+ void *framing;
+ u64 prefix;
+ u64 suffix;
+ u32 frame_charge;
+ u32 prefix_length;
+ u32 suffix_length;
+ u32 total_length;
+ u32 msg_flags;
+ u16 send_flags;
+ bool frame_reserved;
+ };
+ struct {
+ struct io_opaque_data *replacement;
+ u64 reserved;
+ u64 transient;
+ };
+ struct io_uring_opaque_policy policy;
+ };
+ /*
+ * Block of a vectored send, kept when the request returns to the ring's
+ * cache so that the next one has none to allocate. It is not part of the
+ * union and is the last member: allocation clears everything before it.
+ * Only a request with IORING_OPAQUE_SEND_VECTOR uses it, whatever it holds.
+ */
+ struct io_opaque_vec *vec;
+};
+
+bool io_opaque_cache_init(struct io_ring_ctx *ctx);
+void io_opaque_cache_free(struct io_ring_ctx *ctx);
+struct io_opaque_store *io_opaque_alloc(struct io_ring_ctx *ctx,
+ const struct io_uring_opaque_config *config);
+void io_opaque_get(struct io_opaque_store *store);
+void io_opaque_user_put(struct io_opaque_store *store);
+void io_opaque_put(struct io_opaque_store *store);
+int io_register_opaque(struct io_ring_ctx *ctx, void __user *arg);
+void io_terminate_opaque(struct io_ring_ctx *ctx);
+void io_unregister_opaque(struct io_ring_ctx *ctx);
+void io_opaque_stop(struct io_opaque_store *store);
+void io_opaque_free(struct io_opaque_store *store);
+int io_opaque_recv_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
+int io_opaque_recv(struct io_kiocb *req, unsigned int issue_flags);
+int io_opaque_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
+int io_opaque_issue(struct io_kiocb *req, unsigned int issue_flags);
+int io_opaque_send_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
+int io_opaque_send(struct io_kiocb *req, unsigned int issue_flags);
+void io_opaque_cleanup(struct io_kiocb *req);
+void io_opaque_fail(struct io_kiocb *req);
+int io_opaque_cancel(struct io_ring_ctx *ctx, struct io_cancel_data *cd,
+ unsigned int issue_flags);
+bool io_opaque_cancel_all(struct io_ring_ctx *ctx, struct io_uring_task *tctx,
+ bool cancel_all);
+#else
+static inline bool io_opaque_cache_init(struct io_ring_ctx *ctx)
+{
+ return false;
+}
+
+static inline void io_opaque_cache_free(struct io_ring_ctx *ctx)
+{
+}
+
+static inline int io_register_opaque(struct io_ring_ctx *ctx, void __user *arg)
+{
+ return -EOPNOTSUPP;
+}
+
+static inline void io_terminate_opaque(struct io_ring_ctx *ctx)
+{
+}
+
+static inline void io_unregister_opaque(struct io_ring_ctx *ctx)
+{
+}
+
+static inline void io_opaque_cleanup(struct io_kiocb *req)
+{
+}
+
+static inline int io_opaque_cancel(struct io_ring_ctx *ctx,
+ struct io_cancel_data *cd,
+ unsigned int issue_flags)
+{
+ return -ENOENT;
+}
+
+static inline bool io_opaque_cancel_all(struct io_ring_ctx *ctx,
+ struct io_uring_task *tctx, bool cancel_all)
+{
+ return false;
+}
+#endif
+#endif
diff --git a/io_uring/opaque_internal.h b/io_uring/opaque_internal.h
new file mode 100644
index 000000000000..f6b372800c76
--- /dev/null
+++ b/io_uring/opaque_internal.h
@@ -0,0 +1,269 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+#ifndef IO_URING_OPAQUE_INTERNAL_H
+#define IO_URING_OPAQUE_INTERNAL_H
+
+#include <linux/bitmap.h>
+#include <linux/bvec.h>
+#include <linux/net.h>
+#include <linux/sizes.h>
+#include <linux/workqueue.h>
+#include <uapi/linux/io_uring/opaque.h>
+
+#include "opaque.h"
+
+#define IO_OPAQUE_READ_MAX SZ_64K
+#define IO_OPAQUE_RX_BATCH SZ_256K
+#define IO_OPAQUE_MAX_ORDER (PAGE_SHIFT < 16 ? 16 - PAGE_SHIFT : 0)
+#define IO_OPAQUE_ALLOC_RETRY_MS 50
+#define IO_OPAQUE_WAKE_BATCH 32
+#define IO_OPAQUE_INDEX_BITS 14
+#define IO_OPAQUE_INDEX_MASK ((1U << IO_OPAQUE_INDEX_BITS) - 1)
+/* Stream tokens carry the tag bit; object handles and generations never do. */
+#define IO_OPAQUE_TOKEN_TAG BIT_ULL(63)
+#define IO_OPAQUE_GENERATION_MAX (U64_MAX >> (IO_OPAQUE_INDEX_BITS + 1))
+/* Unused per-socket TX queues kept per store; the oldest is freed beyond this. */
+#define IO_OPAQUE_TX_IDLE_MAX 64
+/* Backing segments all ranges of one vectored send may span; more is EMSGSIZE. */
+#define IO_OPAQUE_SEND_VEC_SEGS 4096
+/*
+ * A version of more segments than this records where every this-many-th of
+ * them starts, so that a range is found with a walk of about this many steps
+ * and a binary search, however large the version is and wherever the walk
+ * began. A smaller one is walked, which is as short.
+ */
+#define IO_OPAQUE_SEEK_STRIDE 16
+/* Slots of the table of handles that a policy change walks while it holds tables. */
+#define IO_OPAQUE_POLICY_CHUNK 256
+
+/* What io_opaque_tx_join() returns besides an error. */
+/* Queued behind another frame and parked, with its references taken. */
+#define IO_OPAQUE_TX_WAIT 0
+/* The head of its queue, with its references taken. */
+#define IO_OPAQUE_TX_HEAD 1
+/* The head of its queue, still to take its references. */
+#define IO_OPAQUE_TX_SETTLE 2
+
+struct io_opaque_backing {
+ refcount_t refs;
+ struct page *page;
+ u64 charge;
+ u32 filled;
+ bool copied;
+ bool accounted;
+};
+
+struct io_opaque_extent {
+ struct list_head list;
+ struct io_opaque_backing *backing;
+ u64 start;
+ u32 offset;
+ u32 length;
+ bool embedded;
+};
+
+/* One allocation for the common capture; splits allocate only an extent. */
+struct io_opaque_chunk {
+ struct io_opaque_backing backing;
+ struct io_opaque_extent extent;
+};
+
+struct io_opaque_data {
+ refcount_t refs;
+ struct list_head extents;
+ struct bio_vec *bvec;
+ /*
+ * Where segments 0, IO_OPAQUE_SEEK_STRIDE, twice that and so on start,
+ * as bytes into the object, when it has more than that many segments;
+ * else NULL. It is allocated with bvec and freed with it.
+ */
+ u64 *anchor;
+ u64 length;
+ u64 charge;
+ u32 nr;
+ bool dense;
+ bool forward_only;
+};
+
+struct io_opaque_slot {
+ struct io_opaque_data *data;
+ u64 generation;
+ bool reserved;
+ bool compacting;
+ struct list_head candidate;
+ unsigned long born;
+ unsigned long retry;
+};
+
+struct io_opaque_stream_slot {
+ struct io_opaque_stream *stream;
+ u64 generation;
+};
+
+struct io_opaque_tx {
+ struct rb_node node;
+ struct sock *sk;
+ /* First-issue order, including pending admissions; only the head sends. */
+ struct list_head requests;
+ /* On store->txs_idle while no request is queued. */
+ struct list_head idle;
+ bool failed;
+ /* A busy queue owns its socket identity independently of request files. */
+ bool sk_held;
+};
+
+/*
+ * Where one range of a vectored send starts in the version it pins. Admission
+ * finds it under tables; the cursor is built from it once tables is dropped.
+ */
+struct io_opaque_vec_start {
+ /* The first segment of the range. */
+ u32 seg;
+ /* Bytes of that segment before the range. */
+ u32 skip;
+};
+
+/*
+ * What a vectored send needs besides the request: its entries, copied at prep,
+ * the version each pins, where each starts in it and the segments of the
+ * combined cursor. One block holds the header and the four arrays, so that
+ * growing it is one allocation.
+ */
+struct io_opaque_vec {
+ /* Entries, from the SQE. */
+ u32 nr;
+ /*
+ * Entries of pin[] that admission has set. The version of each is kept
+ * alive by the send, by a reference of the entry's own or, unless the
+ * vector is LAST, by that of the entry before it when both name the
+ * same handle.
+ */
+ u32 pinned;
+ u32 cap_nr;
+ u32 cap_segs;
+ struct io_uring_opaque_send_entry *ent;
+ struct io_opaque_data **pin;
+ struct io_opaque_vec_start *start;
+ struct bio_vec *bv;
+};
+
+struct io_opaque_stream {
+ refcount_t refs;
+ /* Extents, range decisions and collector state. */
+ struct mutex lock;
+ struct io_opaque_store *store;
+ struct file *file;
+ struct sock_rx_owner owner;
+ struct list_head extents;
+ struct list_head reads;
+ struct rb_root_cached decisions;
+ struct io_opaque_req *collector;
+ u64 token;
+ u64 next;
+ u64 undecided;
+ u64 need_bytes;
+ int error;
+ bool eof;
+ bool closed;
+ /* Collection pass may not take undecided bytes off the socket. */
+ bool hold_undecided;
+};
+
+struct io_opaque_store {
+ refcount_t refs;
+ refcount_t users;
+ struct io_uring_opaque_config config;
+ /*
+ * Handles and candidates; never held during copies or I/O, and for a time
+ * that the request bounds: what admission does under it is a lookup per
+ * entry, the segments its ranges span up to the limit, and for each range
+ * a walk that the version's anchors keep to about two strides and a
+ * binary search. Attach and the admission of a send take it with the
+ * socket owned, so nothing may take a socket lock while holding it, or
+ * wait for anything that holds one for long.
+ */
+ struct mutex tables;
+ struct io_opaque_slot *objects;
+ struct io_opaque_stream_slot *streams;
+ unsigned long *object_used;
+ unsigned long *stream_used;
+ u32 object_cursor;
+ u32 stream_cursor;
+ u32 nr_objects;
+ u32 nr_streams;
+ struct rb_root txs;
+ struct list_head txs_idle;
+ u32 nr_txs_idle;
+ struct user_struct *user;
+ struct mm_struct *mm;
+ struct mem_cgroup *memcg;
+ struct obj_cgroup *objcg;
+ atomic64_t bytes;
+ atomic64_t copied;
+ atomic64_t compacted;
+ atomic64_t compactions;
+ atomic64_t framing;
+ atomic64_t framing_copied;
+ /* Framing held in the reply reserve, outside bytes. */
+ atomic64_t reserve_framing;
+ /*
+ * Charged in bytes but not payload held: unpublished compaction copies
+ * and framing in ordinary capacity. Watermarks subtract it. Changes
+ * that move bytes and transient together are one seqcount write.
+ */
+ atomic64_t transient;
+ /* Serializes transient_seq writers. */
+ spinlock_t transient_lock;
+ seqcount_spinlock_t transient_seq;
+ atomic_t extents;
+ atomic_t memory_mode;
+ /* Budget waiters and remote wakeup/cancellation arbitration. */
+ spinlock_t wait_lock;
+ /* TX queues: the tree, every queue's requests and failed flag. Never with tables. */
+ spinlock_t tx_lock;
+ struct list_head budget_waits;
+ struct list_head copies;
+ /* One dispatcher serializes manual and automatic copies. */
+ struct delayed_work compact_work;
+ struct delayed_work retry_work;
+ struct list_head candidates;
+ struct io_uring_opaque_policy policy;
+ /* Counts policy changes; a walk of the table that is overtaken by another stops. */
+ u32 policy_seq;
+ u64 tokens;
+ unsigned long token_time;
+ unsigned long auto_at;
+ int dead;
+};
+
+bool io_opaque_charge(struct io_opaque_store *store, u64 bytes, bool rx);
+void io_opaque_uncharge(struct io_opaque_store *store, u64 bytes);
+bool io_opaque_charge_transient(struct io_opaque_store *store, u64 bytes, u64 limit);
+void io_opaque_release_transient(struct io_opaque_store *store, u64 uncharge,
+ u64 transient);
+bool io_opaque_charge_frame(struct io_opaque_store *store, u64 bytes, bool *reserved);
+void io_opaque_uncharge_frame(struct io_opaque_store *store, u64 bytes, bool reserved);
+int io_opaque_memory_mode(struct io_opaque_store *store);
+
+static inline u64 io_opaque_rx_limit(struct io_opaque_store *store)
+{
+ return store->config.hard_limit - store->config.compact_headroom -
+ store->config.reply_reserve;
+}
+
+struct io_opaque_extent *io_opaque_extent_new(struct io_opaque_store *store,
+ struct page *page, u32 offset,
+ u32 length, u64 start,
+ bool reserved);
+struct io_opaque_data *io_opaque_data_alloc(void);
+void io_opaque_data_put(struct io_opaque_store *store,
+ struct io_opaque_data *data);
+int io_opaque_vector(struct io_opaque_data *data);
+struct page *io_opaque_owned_page(unsigned int *order);
+struct page *io_opaque_dest_page(struct io_opaque_store *store, u64 pages_left,
+ unsigned int *order);
+u64 io_opaque_handle(u32 index, u64 generation);
+struct io_opaque_slot *io_opaque_lookup(struct io_opaque_store *store,
+ u64 handle);
+void io_opaque_wait(struct io_opaque_req *op, enum io_opaque_phase phase);
+void io_opaque_queue_ready(struct io_opaque_req *op, int result);
+#endif
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 03/12] io_uring/opaque: add backing accounting and immutable versions
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
2026-11-10 13:19 ` [RFC PATCH net-next 01/12] net: add an exclusive TCP receive claim Alex Markuze
2026-11-10 13:19 ` [RFC PATCH net-next 02/12] io_uring/opaque: add data structures and the UAPI header Alex Markuze
@ 2026-11-10 13:19 ` Alex Markuze
2026-11-10 13:19 ` [RFC PATCH net-next 04/12] io_uring/opaque: add owned-page helpers for dense copies Alex Markuze
` (8 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:19 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Add bounded backing-byte and extent accounting, immutable version
references, vectors and generation handles. Separate ordinary capacity,
reply reserve and transient reservations; update byte/transient pairs
coherently for watermark hysteresis. Capacity returns precede batched
collector wakes. The feature selects the exclusive receive-owner support.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
io_uring/Kconfig | 11 ++
io_uring/Makefile | 1 +
io_uring/opaque.c | 399 ++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 411 insertions(+)
create mode 100644 io_uring/opaque.c
diff --git a/io_uring/Kconfig b/io_uring/Kconfig
index a283d9e53787..5b696cf266f6 100644
--- a/io_uring/Kconfig
+++ b/io_uring/Kconfig
@@ -15,6 +15,17 @@ config IO_URING_BPF
depends on BPF
depends on NET
+config IO_URING_OPAQUE_OBJ
+ bool "io_uring opaque TCP payload objects"
+ default n
+ depends on IO_URING && INET
+ select SOCK_RX_OWNER
+ help
+ Retain immutable TCP payloads as kernel-resident objects that can
+ be inspected, transmitted and compacted through io_uring. No network
+ device zerocopy support or mapped receive area is required.
+ Applications manage payload lifetime through generation handles.
+
config IO_URING_BPF_OPS
def_bool y
depends on IO_URING
diff --git a/io_uring/Makefile b/io_uring/Makefile
index 965bb60e9f7b..81710127046c 100644
--- a/io_uring/Makefile
+++ b/io_uring/Makefile
@@ -17,6 +17,7 @@ obj-$(CONFIG_IO_URING) += io_uring.o opdef.o kbuf.o rsrc.o notif.o \
query.o loop.o
obj-$(CONFIG_IO_URING_ZCRX) += zcrx.o
+obj-$(CONFIG_IO_URING_OPAQUE_OBJ) += opaque.o
obj-$(CONFIG_IO_WQ) += io-wq.o
obj-$(CONFIG_FUTEX) += futex.o
obj-$(CONFIG_EPOLL) += epoll.o
diff --git a/io_uring/opaque.c b/io_uring/opaque.c
new file mode 100644
index 000000000000..2fd9db898e67
--- /dev/null
+++ b/io_uring/opaque.c
@@ -0,0 +1,399 @@
+// SPDX-License-Identifier: GPL-2.0
+/* Kernel-resident immutable TCP payloads. */
+#include <linux/bitmap.h>
+#include <linux/bvec.h>
+#include <linux/file.h>
+#include <linux/highmem.h>
+#include <linux/memcontrol.h>
+#include <linux/overflow.h>
+#include <linux/rbtree.h>
+#include <linux/sched/mm.h>
+#include <linux/skbuff.h>
+#include <linux/sizes.h>
+#include <linux/slab.h>
+#include <linux/workqueue.h>
+#include <net/tcp.h>
+#include <uapi/linux/io_uring/opaque.h>
+
+#include "io_uring.h"
+#include "cancel.h"
+#include "opaque_internal.h"
+#include "rsrc.h"
+#include "tw.h"
+
+static bool io_opaque_budget_ready(struct io_opaque_req *op, unsigned long now)
+{
+ struct io_opaque_store *store = op->store;
+ u64 limit = io_opaque_rx_limit(store);
+ u64 used = atomic64_read(&store->bytes);
+
+ lockdep_assert_held(&store->wait_lock);
+ return !time_before(now, op->retry_at) && used <= limit &&
+ op->need_bytes <= limit - used &&
+ atomic_read(&store->extents) < store->config.max_extents - 2;
+}
+
+/*
+ * Let in the waiters for budget that the budget lets go on, at most
+ * IO_OPAQUE_WAKE_BATCH of them in a hold of wait_lock. Those left over are for
+ * the retry work, and that is why this says whether there were any: a caller
+ * that wants them in now repeats it. The work comes a jiffy later, or as late
+ * as a timer that is already pending, which queueing never moves earlier.
+ */
+static bool io_opaque_wake_budget(struct io_opaque_store *store)
+{
+ struct io_opaque_req *op, *next;
+ unsigned long delay = MAX_JIFFY_OFFSET;
+ unsigned long now = jiffies;
+ unsigned int woken = 0;
+ bool stopped, more = false;
+
+ guard(spinlock)(&store->wait_lock);
+ stopped = READ_ONCE(store->dead);
+ list_for_each_entry_safe(op, next, &store->budget_waits, wait) {
+ struct socket *sock = sock_from_file(op->req->file);
+
+ if (!stopped && !READ_ONCE(op->stream->closed)) {
+ if (time_before(now, op->retry_at)) {
+ delay = min(delay, op->retry_at - now);
+ continue;
+ }
+ if (!io_opaque_budget_ready(op, now))
+ continue;
+ if (woken == IO_OPAQUE_WAKE_BATCH) {
+ delay = 1;
+ more = true;
+ continue;
+ }
+ }
+ list_del_init(&op->wait);
+ /* A wake is not a reservation; the collector rechecks on entry. */
+ woken++;
+ /* Native multishot polling also observes FIN, RST and urgent data. */
+ wake_up_interruptible_poll(sk_sleep(sock->sk), EPOLLIN);
+ }
+ if (!stopped && delay != MAX_JIFFY_OFFSET)
+ queue_delayed_work(system_dfl_wq, &store->retry_work, delay);
+ return more;
+}
+
+/* As io_opaque_wake_budget(), until the batch leaves nobody behind. */
+static void io_opaque_wake_budget_all(struct io_opaque_store *store)
+{
+ while (io_opaque_wake_budget(store))
+ ;
+}
+
+int io_opaque_memory_mode(struct io_opaque_store *store)
+{
+ unsigned int seq;
+ int mode, next;
+ u64 used, transient;
+
+ if (!store->config.high_watermark)
+ return IORING_OPAQUE_MEMORY_NORMAL;
+ for (;;) {
+ mode = atomic_read(&store->memory_mode);
+ /* Watermarks measure payload held: compaction copies and framing are not. */
+ do {
+ seq = read_seqcount_begin(&store->transient_seq);
+ used = atomic64_read(&store->bytes);
+ transient = atomic64_read(&store->transient);
+ } while (read_seqcount_retry(&store->transient_seq, seq));
+ used = used > transient ? used - transient : 0;
+ next = mode;
+ if (mode == IORING_OPAQUE_MEMORY_NORMAL && used >= store->config.high_watermark)
+ next = IORING_OPAQUE_MEMORY_LOWMEM;
+ else if (mode == IORING_OPAQUE_MEMORY_LOWMEM && used <= store->config.low_watermark)
+ next = IORING_OPAQUE_MEMORY_NORMAL;
+ if (next == mode)
+ return mode;
+ /* Recheck occupancy after winning a concurrent mode transition. */
+ atomic_try_cmpxchg(&store->memory_mode, &mode, next);
+ }
+}
+
+bool io_opaque_charge(struct io_opaque_store *store, u64 bytes, bool rx)
+{
+ u64 limit = store->config.hard_limit - store->config.reply_reserve;
+ s64 used = atomic64_read(&store->bytes);
+
+ if (rx)
+ limit -= store->config.compact_headroom;
+ for (;;) {
+ if (bytes > limit || used > limit - bytes)
+ return false;
+ if (atomic64_try_cmpxchg(&store->bytes, &used, used + bytes)) {
+ io_opaque_memory_mode(store);
+ return true;
+ }
+ }
+}
+
+/* Give bytes back to the store, and let nobody in on them yet. */
+static void io_opaque_uncharge_quiet(struct io_opaque_store *store, u64 bytes)
+{
+ atomic64_sub(bytes, &store->bytes);
+ io_opaque_memory_mode(store);
+}
+
+void io_opaque_uncharge(struct io_opaque_store *store, u64 bytes)
+{
+ io_opaque_uncharge_quiet(store, bytes);
+ io_opaque_wake_budget(store);
+}
+
+/* Restore both budgets before waking collectors after a failed reservation. */
+static void io_opaque_extent_unreserve(struct io_opaque_store *store, u64 bytes)
+{
+ atomic_dec(&store->extents);
+ if (bytes)
+ io_opaque_uncharge_quiet(store, bytes);
+ io_opaque_wake_budget(store);
+}
+
+/* Charge bytes that watermarks ignore; bytes and transient change in one write. */
+bool io_opaque_charge_transient(struct io_opaque_store *store, u64 bytes, u64 limit)
+{
+ s64 used;
+ bool charged = false;
+
+ spin_lock(&store->transient_lock);
+ write_seqcount_begin(&store->transient_seq);
+ used = atomic64_read(&store->bytes);
+ for (;;) {
+ if (bytes > limit || used > limit - bytes)
+ break;
+ if (atomic64_try_cmpxchg(&store->bytes, &used, used + bytes)) {
+ atomic64_add(bytes, &store->transient);
+ charged = true;
+ break;
+ }
+ }
+ write_seqcount_end(&store->transient_seq);
+ spin_unlock(&store->transient_lock);
+ return charged;
+}
+
+/* Uncharge and stop excluding transient bytes; the two amounts may differ. */
+void io_opaque_release_transient(struct io_opaque_store *store, u64 uncharge,
+ u64 transient)
+{
+ spin_lock(&store->transient_lock);
+ write_seqcount_begin(&store->transient_seq);
+ atomic64_sub(uncharge, &store->bytes);
+ atomic64_sub(transient, &store->transient);
+ write_seqcount_end(&store->transient_seq);
+ spin_unlock(&store->transient_lock);
+ io_opaque_memory_mode(store);
+ if (uncharge)
+ io_opaque_wake_budget(store);
+}
+
+static bool io_opaque_extent_reserve(struct io_opaque_store *store)
+{
+ int nr = atomic_read(&store->extents);
+
+ while (nr < store->config.max_extents) {
+ if (atomic_try_cmpxchg(&store->extents, &nr, nr + 1))
+ return true;
+ }
+ return false;
+}
+
+static int io_opaque_memcg_charge(struct io_opaque_store *store, u64 bytes)
+{
+#ifdef CONFIG_MEMCG
+ if (store->objcg)
+ return obj_cgroup_charge(store->objcg,
+ GFP_KERNEL | __GFP_NORETRY | __GFP_NOWARN, bytes);
+#endif
+ return 0;
+}
+
+static void io_opaque_memcg_uncharge(struct io_opaque_store *store, u64 bytes)
+{
+#ifdef CONFIG_MEMCG
+ if (store->objcg)
+ obj_cgroup_uncharge(store->objcg, bytes);
+#endif
+}
+
+struct io_opaque_extent *io_opaque_extent_new(struct io_opaque_store *store,
+ struct page *page, u32 offset,
+ u32 length, u64 start,
+ bool reserved)
+{
+ struct io_opaque_chunk *chunk;
+ u64 charge = page_size(compound_head(page));
+
+ if (!io_opaque_extent_reserve(store))
+ return ERR_PTR(-ENOBUFS);
+ if (!reserved && !io_opaque_charge(store, charge, true))
+ goto no_space;
+ if (!reserved && io_opaque_memcg_charge(store, charge))
+ goto no_memory;
+ chunk = kzalloc_obj(*chunk, GFP_KERNEL_ACCOUNT);
+ if (!chunk) {
+ if (!reserved)
+ io_opaque_memcg_uncharge(store, charge);
+ goto no_memory;
+ }
+ refcount_set(&chunk->backing.refs, 1);
+ chunk->backing.page = compound_head(page);
+ chunk->backing.charge = charge;
+ chunk->backing.accounted = !reserved && store->objcg;
+ chunk->extent.backing = &chunk->backing;
+ chunk->extent.start = start;
+ chunk->extent.offset = offset + (page - compound_head(page)) * PAGE_SIZE;
+ chunk->extent.length = length;
+ chunk->extent.embedded = true;
+ INIT_LIST_HEAD(&chunk->extent.list);
+ return &chunk->extent;
+no_memory:
+ io_opaque_extent_unreserve(store, reserved ? 0 : charge);
+ return ERR_PTR(-ENOMEM);
+no_space:
+ io_opaque_extent_unreserve(store, 0);
+ return ERR_PTR(-ENOBUFS);
+}
+
+/*
+ * Let go of @extent, and of its backing if it was the last to use it. Returns
+ * the charge of the backing that went, for the caller to give back to the
+ * store: giving it back lets those waiting for budget in, which one free does
+ * once and a list of them once, not once per extent.
+ */
+static u64 io_opaque_extent_release(struct io_opaque_store *store, struct io_opaque_extent *extent)
+{
+ struct io_opaque_backing *backing = extent->backing;
+ u64 charge = 0;
+
+ list_del(&extent->list);
+ if (!extent->embedded)
+ kfree(extent);
+ atomic_dec(&store->extents);
+ if (refcount_dec_and_test(&backing->refs)) {
+ charge = backing->charge;
+
+ if (backing->accounted)
+ io_opaque_memcg_uncharge(store, charge);
+ put_page(backing->page);
+ kfree(container_of(backing, struct io_opaque_chunk, backing));
+ }
+ return charge;
+}
+
+/*
+ * Free the extents of @list. What they held is returned to the store in one
+ * step, and the waiters for budget are looked at in one go: each look takes
+ * wait_lock and walks the list of them, and a 1 MB object that is not one
+ * mapping can have 700 extents. The look is repeated while it leaves ready
+ * waiters behind, so that they are let in by the free and not by the retry
+ * work.
+ */
+static void io_opaque_extents_free(struct io_opaque_store *store,
+ struct list_head *list)
+{
+ struct io_opaque_extent *extent, *next;
+ bool any = false;
+ u64 charge = 0;
+
+ list_for_each_entry_safe(extent, next, list, list) {
+ charge += io_opaque_extent_release(store, extent);
+ any = true;
+ }
+ if (charge)
+ io_opaque_uncharge_quiet(store, charge);
+ if (any)
+ io_opaque_wake_budget_all(store);
+}
+
+struct io_opaque_data *io_opaque_data_alloc(void)
+{
+ struct io_opaque_data *data = kzalloc_obj(*data, GFP_KERNEL_ACCOUNT);
+
+ if (data) {
+ refcount_set(&data->refs, 1);
+ INIT_LIST_HEAD(&data->extents);
+ }
+ return data;
+}
+
+void io_opaque_data_put(struct io_opaque_store *store,
+ struct io_opaque_data *data)
+{
+ if (data && refcount_dec_and_test(&data->refs)) {
+ io_opaque_extents_free(store, &data->extents);
+ kvfree(data->bvec);
+ kfree(data);
+ }
+}
+
+/*
+ * Build the version's segments, and for a version of more than
+ * IO_OPAQUE_SEEK_STRIDE of them the anchors that find one without walking
+ * from the first, in one allocation. Both are made once, when the version is,
+ * and never change. The anchors follow the segments at the next multiple of
+ * eight bytes: a segment is 16 bytes, but only 12 on 32 bits, where an odd
+ * number of them would leave the anchors four-byte aligned.
+ */
+int io_opaque_vector(struct io_opaque_data *data)
+{
+ struct io_opaque_extent *extent;
+ struct io_opaque_backing *previous = NULL;
+ u32 nr = 0, anchors;
+ size_t bytes;
+ u64 at = 0;
+
+ data->charge = 0;
+ list_for_each_entry(extent, &data->extents, list) {
+ nr++;
+ if (extent->backing != previous)
+ data->charge += extent->backing->charge;
+ previous = extent->backing;
+ }
+ anchors = nr > IO_OPAQUE_SEEK_STRIDE ? DIV_ROUND_UP(nr, IO_OPAQUE_SEEK_STRIDE) : 0;
+ /* Where the anchors start. An overflow saturates, and the allocation fails. */
+ bytes = size_add(array_size(nr, sizeof(*data->bvec)), sizeof(*data->anchor) - 1);
+ bytes &= ~(sizeof(*data->anchor) - 1);
+ data->bvec = kvmalloc(size_add(bytes, array_size(anchors, sizeof(*data->anchor))),
+ GFP_KERNEL_ACCOUNT);
+ if (!data->bvec)
+ return -ENOMEM;
+ data->anchor = anchors ? (void *)data->bvec + bytes : NULL;
+ data->nr = nr;
+ nr = 0;
+ list_for_each_entry(extent, &data->extents, list) {
+ if (anchors && !(nr % IO_OPAQUE_SEEK_STRIDE))
+ data->anchor[nr / IO_OPAQUE_SEEK_STRIDE] = at;
+ bvec_set_page(&data->bvec[nr++], extent->backing->page,
+ extent->length, extent->offset);
+ at += extent->length;
+ }
+ return 0;
+}
+
+/* Handle encoding is internal; zero is never a published handle. */
+u64 io_opaque_handle(u32 index, u64 generation)
+{
+ return generation << IO_OPAQUE_INDEX_BITS | index;
+}
+
+static u32 io_opaque_index(u64 handle)
+{
+ return handle & IO_OPAQUE_INDEX_MASK;
+}
+
+struct io_opaque_slot *io_opaque_lookup(struct io_opaque_store *store,
+ u64 handle)
+{
+ u32 index = io_opaque_index(handle);
+
+ lockdep_assert_held(&store->tables);
+ if (!handle || index >= store->config.max_objects)
+ return NULL;
+ if (store->objects[index].generation != handle >> IO_OPAQUE_INDEX_BITS)
+ return NULL;
+ return &store->objects[index];
+}
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 04/12] io_uring/opaque: add owned-page helpers for dense copies
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (2 preceding siblings ...)
2026-11-10 13:19 ` [RFC PATCH net-next 03/12] io_uring/opaque: add backing accounting and immutable versions Alex Markuze
@ 2026-11-10 13:19 ` Alex Markuze
2026-11-10 13:19 ` [RFC PATCH net-next 05/12] io_uring/opaque: add store registration and IORING_OP_OPAQUE_OBJ Alex Markuze
` (7 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:19 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Add the accounted destination-page allocator used by CREATE and compaction.
Prefer compound allocations up to 64 KiB, or one base page when larger,
and fall back to smaller orders.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
io_uring/Makefile | 2 +-
io_uring/opaque_compact.c | 33 +++++++++++++++++++++++++++++++++
2 files changed, 34 insertions(+), 1 deletion(-)
create mode 100644 io_uring/opaque_compact.c
diff --git a/io_uring/Makefile b/io_uring/Makefile
index 81710127046c..191316972e04 100644
--- a/io_uring/Makefile
+++ b/io_uring/Makefile
@@ -17,7 +17,7 @@ obj-$(CONFIG_IO_URING) += io_uring.o opdef.o kbuf.o rsrc.o notif.o \
query.o loop.o
obj-$(CONFIG_IO_URING_ZCRX) += zcrx.o
-obj-$(CONFIG_IO_URING_OPAQUE_OBJ) += opaque.o
+obj-$(CONFIG_IO_URING_OPAQUE_OBJ) += opaque.o opaque_compact.o
obj-$(CONFIG_IO_WQ) += io-wq.o
obj-$(CONFIG_FUTEX) += futex.o
obj-$(CONFIG_EPOLL) += epoll.o
diff --git a/io_uring/opaque_compact.c b/io_uring/opaque_compact.c
new file mode 100644
index 000000000000..b59c7cb95d42
--- /dev/null
+++ b/io_uring/opaque_compact.c
@@ -0,0 +1,33 @@
+// SPDX-License-Identifier: GPL-2.0
+/* Workers copy immutable backing without holding object table locks. */
+#include <linux/highmem.h>
+#include <linux/memcontrol.h>
+#include <linux/mm.h>
+#include <linux/overflow.h>
+#include <linux/sched.h>
+#include <linux/sizes.h>
+#include <linux/slab.h>
+
+#include "io_uring.h"
+#include "opaque_internal.h"
+
+struct page *io_opaque_owned_page(unsigned int *order)
+{
+ struct page *page;
+
+ for (;;) {
+ page = alloc_pages(GFP_KERNEL_ACCOUNT | __GFP_COMP | __GFP_NORETRY |
+ __GFP_NOWARN, *order);
+ if (page || !*order)
+ return page;
+ (*order)--;
+ }
+}
+
+/* Destination allocation for dense copies and CREATE. */
+struct page *io_opaque_dest_page(struct io_opaque_store *store, u64 pages_left,
+ unsigned int *order)
+{
+ *order = min_t(unsigned int, IO_OPAQUE_MAX_ORDER, ilog2(pages_left));
+ return io_opaque_owned_page(order);
+}
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 05/12] io_uring/opaque: add store registration and IORING_OP_OPAQUE_OBJ
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (3 preceding siblings ...)
2026-11-10 13:19 ` [RFC PATCH net-next 04/12] io_uring/opaque: add owned-page helpers for dense copies Alex Markuze
@ 2026-11-10 13:19 ` Alex Markuze
2026-11-10 13:20 ` [RFC PATCH net-next 06/12] io_uring/opaque: allow rings to share a store Alex Markuze
` (6 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:19 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Add IORING_REGISTER_OPAQUE_STORE and the object controls, together with
their request and store lifetimes.
Registration requires DEFER_TASKRUN and either CQE32 or CQE_MIXED. The
configuration supplies limits for backing bytes, object size, objects,
streams and extents, with optional watermarks and a reply reserve.
Creation accounts the hard byte limit against the ring's user and mm,
as registered buffers do. The creating task's live memory cgroup owns
the backing charges; the store retains that reference for its lifetime.
Flags and reserved fields must be zero. An xarray assigns the ring-local
store ID, removed if copying the registration result to userspace fails.
IORING_OP_OPAQUE_OBJ takes the control in sqe->ioprio, the store ID in
sqe->opaque_store_id, a handle or stream token in sqe->addr, a byte
offset in sqe->off, a length in sqe->len and a user pointer in sqe->addr3.
- INSPECT copies up to 64 KiB of retained stream bytes without consuming
them, waiting if no byte is available at the requested offset.
- AVAILABLE returns the contiguous retained length without copying,
stopping at the next pending receive or discard decision.
- RECV_OBJECT reserves a slot and assembles a declared stream range.
Its completed CQE returns the length, handle and forwarding-only flag.
- DISCARD consumes a declared range without retaining an object.
- READ_OBJECT copies up to 64 KiB from a published object; FREE removes
its handle without invalidating references already held by requests.
- STAT, OBJECT_STAT and STREAM_STAT return size-versioned statistics.
- STREAM_CLOSE ends a stream.
Pending receive and discard ranges are non-overlapping and ordered by
offset. A failed or canceled receive returns its private prefix while
the stream is open, allowing inspection, retry or discard. Cancellation,
linked timeouts and request cleanup release all acquired references.
Request state is cached per ring.
Handles carry a slot generation; stream tokens additionally carry a tag.
Stale or wrong-kind names fail with -ESTALE. Exhausted generations retire
their slots. A store stops when its user count reaches zero, closing its
streams and releasing table-owned objects. Its reference count retains
memory and accounting until outstanding requests finish.
The ring lock serializes issue, task work, cancellation and cleanup.
The table mutex protects handle tables; the stream mutex protects one
stream's extents and decisions; wait_lock protects request phases and
wait lists. The order is uring_lock, then tables or stream->lock, then
wait_lock. The table and stream mutexes never nest.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
include/uapi/linux/io_uring.h | 5 +
io_uring/Makefile | 2 +-
io_uring/cancel.c | 5 +
io_uring/io_uring.c | 9 +
io_uring/opaque.c | 1200 +++++++++++++++++++++++++++++++++
io_uring/opaque_register.c | 90 +++
io_uring/opdef.c | 16 +
io_uring/register.c | 7 +
8 files changed, 1333 insertions(+), 1 deletion(-)
create mode 100644 io_uring/opaque_register.c
diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
index b456f45fa8de..081f30605410 100644
--- a/include/uapi/linux/io_uring.h
+++ b/include/uapi/linux/io_uring.h
@@ -91,6 +91,7 @@ struct io_uring_sqe {
__s32 splice_fd_in;
__u32 file_index;
__u32 zcrx_ifq_idx;
+ __u32 opaque_store_id;
__u32 optlen;
struct {
__u16 addr_len;
@@ -319,6 +320,7 @@ enum io_uring_op {
IORING_OP_PIPE,
IORING_OP_NOP128,
IORING_OP_URING_CMD128,
+ IORING_OP_OPAQUE_OBJ,
/* this goes last, obviously */
IORING_OP_LAST,
@@ -724,6 +726,9 @@ enum io_uring_register_op {
/* register bpf filtering programs */
IORING_REGISTER_BPF_FILTER = 37,
+ /* register, import or export a kernel opaque object store */
+ IORING_REGISTER_OPAQUE_STORE = 38,
+
/* this goes last */
IORING_REGISTER_LAST,
diff --git a/io_uring/Makefile b/io_uring/Makefile
index 191316972e04..912b1861c0a5 100644
--- a/io_uring/Makefile
+++ b/io_uring/Makefile
@@ -17,7 +17,7 @@ obj-$(CONFIG_IO_URING) += io_uring.o opdef.o kbuf.o rsrc.o notif.o \
query.o loop.o
obj-$(CONFIG_IO_URING_ZCRX) += zcrx.o
-obj-$(CONFIG_IO_URING_OPAQUE_OBJ) += opaque.o opaque_compact.o
+obj-$(CONFIG_IO_URING_OPAQUE_OBJ) += opaque.o opaque_register.o opaque_compact.o
obj-$(CONFIG_IO_WQ) += io-wq.o
obj-$(CONFIG_FUTEX) += futex.o
obj-$(CONFIG_EPOLL) += epoll.o
diff --git a/io_uring/cancel.c b/io_uring/cancel.c
index 6f7a6cf356c2..e35720f41721 100644
--- a/io_uring/cancel.c
+++ b/io_uring/cancel.c
@@ -19,6 +19,7 @@
#include "waitid.h"
#include "futex.h"
#include "cancel.h"
+#include "opaque.h"
#include "wait.h"
struct io_cancel {
@@ -130,6 +131,9 @@ int io_try_cancel(struct io_uring_task *tctx, struct io_cancel_data *cd,
cd->running = true;
ret = io_poll_cancel(ctx, cd, issue_flags);
+ if (ret != -ENOENT)
+ return ret;
+ ret = io_opaque_cancel(ctx, cd, issue_flags);
if (ret != -ENOENT)
return ret;
@@ -582,6 +586,7 @@ __cold bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx,
mutex_lock(&ctx->uring_lock);
ret |= io_cancel_defer_files(ctx, tctx, cancel_all);
ret |= io_poll_remove_all(ctx, tctx, cancel_all);
+ ret |= io_opaque_cancel_all(ctx, tctx, cancel_all);
ret |= io_waitid_remove_all(ctx, tctx, cancel_all);
ret |= io_futex_remove_all(ctx, tctx, cancel_all);
ret |= io_uring_try_cancel_uring_cmd(ctx, tctx);
diff --git a/io_uring/io_uring.c b/io_uring/io_uring.c
index 8ffe0c9a4e29..fddf9e0ebc8b 100644
--- a/io_uring/io_uring.c
+++ b/io_uring/io_uring.c
@@ -88,6 +88,7 @@
#include "msg_ring.h"
#include "memmap.h"
#include "zcrx.h"
+#include "opaque.h"
#include "bpf-ops.h"
#include "timeout.h"
@@ -220,6 +221,7 @@ static void io_free_alloc_caches(struct io_ring_ctx *ctx)
io_alloc_cache_free(&ctx->rw_cache, io_rw_cache_free);
io_alloc_cache_free(&ctx->cmd_cache, io_cmd_cache_free);
io_futex_cache_free(ctx);
+ io_opaque_cache_free(ctx);
io_rsrc_cache_free(ctx);
}
@@ -267,6 +269,7 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
sizeof(struct io_async_cmd),
sizeof(struct io_async_cmd));
ret |= io_futex_cache_init(ctx);
+ ret |= io_opaque_cache_init(ctx);
ret |= io_rsrc_cache_init(ctx);
if (ret)
goto free_ref;
@@ -287,6 +290,10 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
ctx->submit_state.free_list.next = NULL;
INIT_HLIST_HEAD(&ctx->waitid_list);
xa_init_flags(&ctx->zcrx_ctxs, XA_FLAGS_ALLOC);
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+ xa_init_flags(&ctx->opaque_stores, XA_FLAGS_ALLOC);
+ INIT_LIST_HEAD(&ctx->opaque_waits);
+#endif
#ifdef CONFIG_FUTEX
INIT_HLIST_HEAD(&ctx->futex_list);
#endif
@@ -2294,6 +2301,7 @@ static __cold void io_ring_ctx_free(struct io_ring_ctx *ctx)
mutex_lock(&ctx->uring_lock);
io_unregister_zcrx(ctx);
+ io_unregister_opaque(ctx);
io_cqring_overflow_kill(ctx);
io_eventfd_unregister(ctx);
io_free_alloc_caches(ctx);
@@ -2638,6 +2646,7 @@ static __cold void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
xa_for_each(&ctx->personalities, index, creds)
io_unregister_personality(ctx, index);
io_terminate_zcrx(ctx);
+ io_terminate_opaque(ctx);
/* Drop these now, rather than async, to unpin the files */
io_sqe_buffers_unregister(ctx);
io_sqe_files_unregister(ctx);
diff --git a/io_uring/opaque.c b/io_uring/opaque.c
index 2fd9db898e67..81fbf2011446 100644
--- a/io_uring/opaque.c
+++ b/io_uring/opaque.c
@@ -21,6 +21,20 @@
#include "rsrc.h"
#include "tw.h"
+bool io_opaque_cache_init(struct io_ring_ctx *ctx)
+{
+ return io_alloc_cache_init(&ctx->opaque_cache, IO_ALLOC_CACHE_MAX,
+ sizeof(struct io_opaque_req), sizeof(struct io_opaque_req));
+}
+
+void io_opaque_cache_free(struct io_ring_ctx *ctx)
+{
+ io_alloc_cache_free(&ctx->opaque_cache, kfree);
+}
+
+static void io_opaque_ready(struct io_tw_req tw_req, io_tw_token_t tw);
+static void io_opaque_stream_close(struct io_opaque_stream *stream);
+
static bool io_opaque_budget_ready(struct io_opaque_req *op, unsigned long now)
{
struct io_opaque_store *store = op->store;
@@ -33,6 +47,30 @@ static bool io_opaque_budget_ready(struct io_opaque_req *op, unsigned long now)
atomic_read(&store->extents) < store->config.max_extents - 2;
}
+/* A DISCARD at the capture cursor consumes TCP bytes without allocating. */
+static bool io_opaque_discard_ready(struct io_opaque_stream *stream)
+{
+ struct rb_node *node = rb_first_cached(&stream->decisions);
+ struct io_opaque_req *op;
+
+ lockdep_assert_held(&stream->lock);
+ if (!node)
+ return false;
+ op = rb_entry(node, struct io_opaque_req, decision);
+ return op->op == IORING_OPAQUE_DISCARD &&
+ op->offset + op->progress == stream->next &&
+ op->progress < op->length;
+}
+
+static bool io_opaque_collect_blocked(struct io_opaque_req *op)
+{
+ lockdep_assert_held(&op->stream->lock);
+ lockdep_assert_held(&op->store->wait_lock);
+ return op->phase == IO_OPAQUE_BUDGET_POLL &&
+ !io_opaque_budget_ready(op, jiffies) &&
+ !io_opaque_discard_ready(op->stream);
+}
+
/*
* Let in the waiters for budget that the budget lets go on, at most
* IO_OPAQUE_WAKE_BATCH of them in a hold of wait_lock. Those left over are for
@@ -84,6 +122,31 @@ static void io_opaque_wake_budget_all(struct io_opaque_store *store)
;
}
+static void io_opaque_retry_work(struct work_struct *work)
+{
+ struct io_opaque_store *store = container_of(to_delayed_work(work),
+ struct io_opaque_store, retry_work);
+
+ io_opaque_wake_budget(store);
+}
+
+static void io_opaque_wake_stream(struct io_opaque_stream *stream)
+{
+ struct io_opaque_req *op = stream->collector;
+ struct socket *sock = sock_from_file(stream->file);
+
+ lockdep_assert_held(&stream->lock);
+ if (!op)
+ return;
+ scoped_guard(spinlock, &stream->store->wait_lock) {
+ if (!stream->closed && !READ_ONCE(stream->store->dead) &&
+ io_opaque_collect_blocked(op))
+ return;
+ list_del_init(&op->wait);
+ }
+ wake_up_interruptible_poll(sk_sleep(sock->sk), EPOLLIN);
+}
+
int io_opaque_memory_mode(struct io_opaque_store *store)
{
unsigned int seq;
@@ -284,6 +347,17 @@ static u64 io_opaque_extent_release(struct io_opaque_store *store, struct io_opa
return charge;
}
+static void io_opaque_extent_free(struct io_opaque_store *store,
+ struct io_opaque_extent *extent)
+{
+ u64 charge = io_opaque_extent_release(store, extent);
+
+ if (charge)
+ io_opaque_uncharge(store, charge);
+ else
+ io_opaque_wake_budget(store);
+}
+
/*
* Free the extents of @list. What they held is returned to the store in one
* step, and the waiters for budget are looked at in one go: each look takes
@@ -385,6 +459,32 @@ static u32 io_opaque_index(u64 handle)
return handle & IO_OPAQUE_INDEX_MASK;
}
+/* Stream tokens carry IO_OPAQUE_TOKEN_TAG; strip it to get the generation. */
+static u64 io_opaque_token_generation(u64 token)
+{
+ return (token & ~IO_OPAQUE_TOKEN_TAG) >> IO_OPAQUE_INDEX_BITS;
+}
+
+static u32 io_opaque_find_slot(unsigned long *used, u32 count, u32 *cursor)
+{
+ u32 index = find_next_zero_bit(used, count, *cursor);
+
+ if (index == count && *cursor) {
+ index = find_first_zero_bit(used, *cursor);
+ if (index == *cursor)
+ return count;
+ }
+ if (index < count)
+ *cursor = (index + 1) % count;
+ return index;
+}
+
+static void io_opaque_release_slot(unsigned long *used, u32 index, u64 generation)
+{
+ if (generation != IO_OPAQUE_GENERATION_MAX)
+ __clear_bit(index, used);
+}
+
struct io_opaque_slot *io_opaque_lookup(struct io_opaque_store *store,
u64 handle)
{
@@ -397,3 +497,1103 @@ struct io_opaque_slot *io_opaque_lookup(struct io_opaque_store *store,
return NULL;
return &store->objects[index];
}
+
+/* Drop handle visibility, transferring the table reference to the caller. */
+static struct io_opaque_data *io_opaque_detach(struct io_opaque_store *store,
+ struct io_opaque_slot *slot)
+{
+ struct io_opaque_data *data;
+
+ lockdep_assert_held(&store->tables);
+ if (!slot || !slot->data)
+ return NULL;
+ data = slot->data;
+ slot->data = NULL;
+ store->nr_objects--;
+ io_opaque_release_slot(store->object_used, slot - store->objects, slot->generation);
+ slot->compacting = false;
+ list_del_init(&slot->candidate);
+ return data;
+}
+
+static int io_opaque_slot_reserve(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_slot *slot;
+ u32 index;
+
+ guard(mutex)(&store->tables);
+ for (;;) {
+ index = io_opaque_find_slot(store->object_used, store->config.max_objects,
+ &store->object_cursor);
+ if (index >= store->config.max_objects)
+ return -ENOSPC;
+ slot = &store->objects[index];
+ __set_bit(index, store->object_used);
+ if (slot->generation != IO_OPAQUE_GENERATION_MAX)
+ break;
+ }
+ slot->reserved = true;
+ slot->generation++;
+ if (op->data)
+ op->data->forward_only =
+ io_opaque_memory_mode(store) == IORING_OPAQUE_MEMORY_LOWMEM;
+ op->slot = index;
+ op->handle = io_opaque_handle(index, slot->generation);
+ op->slot_reserved = true;
+ return 0;
+}
+
+static void io_opaque_stream_put(struct io_opaque_stream *stream)
+{
+ if (stream && refcount_dec_and_test(&stream->refs)) {
+ io_opaque_extents_free(stream->store, &stream->extents);
+ fput(stream->file);
+ mutex_destroy(&stream->lock);
+ kfree(stream);
+ }
+}
+
+struct io_opaque_store *io_opaque_alloc(struct io_ring_ctx *ctx,
+ const struct io_uring_opaque_config *config)
+{
+ struct io_uring_opaque_config cfg = *config;
+ struct io_opaque_store *store;
+ int ret;
+ u32 i;
+
+ if (cfg.flags || cfg.__resv0 || cfg.__resv1 || cfg.__resv2 ||
+ !mem_is_zero(cfg.__resv, sizeof(cfg.__resv)))
+ return ERR_PTR(-EINVAL);
+ if (!cfg.hard_limit || cfg.hard_limit > SZ_1T ||
+ !IS_ALIGNED(cfg.hard_limit, PAGE_SIZE) ||
+ !IS_ALIGNED(cfg.compact_headroom, PAGE_SIZE) ||
+ cfg.compact_headroom >= cfg.hard_limit ||
+ !cfg.max_object_size || cfg.max_object_size > INT_MAX ||
+ !cfg.max_objects || cfg.max_objects > 16384 ||
+ !cfg.max_streams || cfg.max_streams > 1024 ||
+ cfg.max_extents < 4 || cfg.max_extents > 65536)
+ return ERR_PTR(-EINVAL);
+ if (!IS_ALIGNED(cfg.reply_reserve, PAGE_SIZE) ||
+ cfg.reply_reserve >= cfg.hard_limit - cfg.compact_headroom ||
+ (!cfg.high_watermark && cfg.low_watermark) ||
+ (cfg.high_watermark &&
+ (!cfg.reply_reserve ||
+ !IS_ALIGNED(cfg.high_watermark, PAGE_SIZE) ||
+ !IS_ALIGNED(cfg.low_watermark, PAGE_SIZE) ||
+ cfg.low_watermark >= cfg.high_watermark ||
+ cfg.high_watermark > cfg.hard_limit - cfg.compact_headroom - cfg.reply_reserve)))
+ return ERR_PTR(-EINVAL);
+ store = kzalloc_obj(*store, GFP_KERNEL_ACCOUNT);
+ if (!store)
+ return ERR_PTR(-ENOMEM);
+ refcount_set(&store->refs, 1);
+ refcount_set(&store->users, 1);
+ store->config = cfg;
+ store->objects = kvcalloc(cfg.max_objects, sizeof(*store->objects),
+ GFP_KERNEL_ACCOUNT);
+ store->streams = kvcalloc(cfg.max_streams, sizeof(*store->streams),
+ GFP_KERNEL_ACCOUNT);
+ store->object_used = bitmap_zalloc(cfg.max_objects, GFP_KERNEL_ACCOUNT);
+ store->stream_used = bitmap_zalloc(cfg.max_streams, GFP_KERNEL_ACCOUNT);
+ if (!store->objects || !store->streams || !store->object_used || !store->stream_used) {
+ ret = -ENOMEM;
+ goto free_store;
+ }
+ ret = io_account_mem(ctx->user, ctx->mm_account, cfg.hard_limit >> PAGE_SHIFT);
+ if (ret)
+ goto free_store;
+ store->user = ctx->user ? get_uid(ctx->user) : NULL;
+ store->mm = ctx->mm_account;
+ if (store->mm)
+ mmgrab(store->mm);
+ /* The registering task's memcg owns backing charges, including after import. */
+ store->memcg = get_mem_cgroup_from_current();
+#ifdef CONFIG_MEMCG
+ if (store->memcg && !mem_cgroup_is_root(store->memcg)) {
+ struct mem_cgroup *old = set_active_memcg(store->memcg);
+
+ store->objcg = get_obj_cgroup_from_current();
+ set_active_memcg(old);
+ }
+#endif
+ mutex_init(&store->tables);
+ spin_lock_init(&store->wait_lock);
+ spin_lock_init(&store->tx_lock);
+ spin_lock_init(&store->transient_lock);
+ seqcount_spinlock_init(&store->transient_seq, &store->transient_lock);
+ INIT_LIST_HEAD(&store->budget_waits);
+ INIT_LIST_HEAD(&store->copies);
+ INIT_LIST_HEAD(&store->candidates);
+ store->txs = RB_ROOT;
+ INIT_LIST_HEAD(&store->txs_idle);
+ for (i = 0; i < cfg.max_objects; i++)
+ INIT_LIST_HEAD(&store->objects[i].candidate);
+ INIT_DELAYED_WORK(&store->retry_work, io_opaque_retry_work);
+ return store;
+free_store:
+ bitmap_free(store->stream_used);
+ bitmap_free(store->object_used);
+ kvfree(store->streams);
+ kvfree(store->objects);
+ kfree(store);
+ return ERR_PTR(ret);
+}
+
+void io_opaque_stop(struct io_opaque_store *store)
+{
+ u32 i;
+
+ if (!store)
+ return;
+ /* Close enrollment before draining the compaction dispatcher. */
+ scoped_guard(mutex, &store->tables) {
+ guard(spinlock)(&store->wait_lock);
+
+ if (store->dead)
+ return;
+ WRITE_ONCE(store->dead, 1);
+ }
+ cancel_delayed_work_sync(&store->retry_work);
+ for (i = 0; i < store->config.max_streams; i++) {
+ struct io_opaque_stream *stream;
+
+ mutex_lock(&store->tables);
+ stream = store->streams[i].stream;
+ store->streams[i].stream = NULL;
+ if (stream) {
+ store->nr_streams--;
+ io_opaque_release_slot(store->stream_used, i, store->streams[i].generation);
+ }
+ mutex_unlock(&store->tables);
+ if (stream) {
+ io_opaque_stream_close(stream);
+ io_opaque_stream_put(stream);
+ }
+ }
+ for (i = 0; i < store->config.max_objects; i++) {
+ struct io_opaque_data *data;
+
+ mutex_lock(&store->tables);
+ data = io_opaque_detach(store, &store->objects[i]);
+ mutex_unlock(&store->tables);
+ io_opaque_data_put(store, data);
+ }
+ io_opaque_wake_budget(store);
+}
+
+void io_opaque_free(struct io_opaque_store *store)
+{
+ if (!store)
+ return;
+ io_opaque_stop(store);
+ WARN_ON_ONCE(atomic64_read(&store->bytes));
+ WARN_ON_ONCE(atomic_read(&store->extents));
+ io_unaccount_mem(store->user, store->mm,
+ store->config.hard_limit >> PAGE_SHIFT);
+ obj_cgroup_put(store->objcg);
+ mem_cgroup_put(store->memcg);
+ if (store->mm)
+ mmdrop(store->mm);
+ free_uid(store->user);
+ bitmap_free(store->object_used);
+ bitmap_free(store->stream_used);
+ kvfree(store->objects);
+ kvfree(store->streams);
+ mutex_destroy(&store->tables);
+ kfree(store);
+}
+
+void io_opaque_get(struct io_opaque_store *store)
+{
+ refcount_inc(&store->refs);
+ refcount_inc(&store->users);
+}
+
+void io_opaque_user_put(struct io_opaque_store *store)
+{
+ if (refcount_dec_and_test(&store->users))
+ io_opaque_stop(store);
+}
+
+void io_opaque_put(struct io_opaque_store *store)
+{
+ if (refcount_dec_and_test(&store->refs))
+ io_opaque_free(store);
+}
+
+static struct io_opaque_req *io_opaque_req_alloc(struct io_kiocb *req,
+ struct io_opaque_store *store)
+{
+ struct io_opaque_req *op;
+
+ if (READ_ONCE(store->dead))
+ return ERR_PTR(-ESHUTDOWN);
+ op = io_uring_alloc_async_data(&req->ctx->opaque_cache, req);
+ if (!op)
+ return ERR_PTR(-ENOMEM);
+ memset(op, 0, sizeof(*op));
+ op->op = U16_MAX;
+ op->req = req;
+ op->store = store;
+ op->retry_at = jiffies;
+ INIT_LIST_HEAD(&op->wait);
+ INIT_LIST_HEAD(&op->cancel);
+ req->flags |= REQ_F_NEED_CLEANUP;
+ return op;
+}
+
+/* The ring lock protects enrollment; wait_lock arbitrates remote wakeups. */
+static void io_opaque_enroll(struct io_opaque_req *op)
+{
+ lockdep_assert_held(&op->req->ctx->uring_lock);
+ if (list_empty(&op->cancel))
+ list_add_tail(&op->cancel, &op->req->ctx->opaque_waits);
+}
+
+void io_opaque_wait(struct io_opaque_req *op, enum io_opaque_phase phase)
+{
+ io_opaque_enroll(op);
+ guard(spinlock)(&op->store->wait_lock);
+ op->phase = phase;
+}
+
+void io_opaque_queue_ready(struct io_opaque_req *op, int result)
+{
+ guard(spinlock)(&op->store->wait_lock);
+ op->result = result;
+ op->phase = IO_OPAQUE_QUEUED;
+ op->req->io_task_work.func = io_opaque_ready;
+ io_req_task_work_add(op->req);
+}
+
+static struct io_opaque_extent *io_opaque_extent_split(struct io_opaque_store *store,
+ struct io_opaque_extent *extent,
+ u64 at)
+{
+ struct io_opaque_extent *right;
+ u32 left = at - extent->start;
+
+ if (!io_opaque_extent_reserve(store))
+ return ERR_PTR(-ENOBUFS);
+ right = kmalloc_obj(*right, GFP_KERNEL_ACCOUNT);
+ if (!right) {
+ io_opaque_extent_unreserve(store, 0);
+ return ERR_PTR(-ENOMEM);
+ }
+ *right = *extent;
+ right->embedded = false;
+ refcount_inc(&right->backing->refs);
+ right->start = at;
+ right->offset += left;
+ right->length -= left;
+ extent->length = left;
+ list_add(&right->list, &extent->list);
+ return right;
+}
+
+static int io_opaque_available(struct io_opaque_stream *stream, u64 off, u64 len)
+{
+ struct io_opaque_extent *extent;
+ u64 end = off + len;
+ u64 start = off;
+
+ list_for_each_entry(extent, &stream->extents, list) {
+ if (extent->start + extent->length <= off)
+ continue;
+ if (extent->start > off)
+ break;
+ off = min(end, extent->start + extent->length);
+ if (off == end)
+ break;
+ }
+ if (off != start)
+ return off - start;
+ return off < stream->next ? -ENODATA : 0;
+}
+
+/*
+ * AVAILABLE must not count bytes reserved by a pending decision, even if
+ * pressure has left that decision's prefix in the stream extents.
+ */
+static int io_opaque_unclassified(struct io_opaque_stream *stream, u64 off, u64 len)
+{
+ struct rb_node *node;
+
+ for (node = rb_first_cached(&stream->decisions); node; node = rb_next(node)) {
+ struct io_opaque_req *decision = rb_entry(node, struct io_opaque_req, decision);
+
+ if (decision->offset > off) {
+ len = min(len, decision->offset - off);
+ break;
+ }
+ }
+ return io_opaque_available(stream, off, len);
+}
+
+static int io_opaque_snapshot(struct io_opaque_req *op)
+{
+ struct io_opaque_extent *extent;
+ int ret = io_opaque_available(op->stream, op->offset, op->length);
+ u64 end;
+
+ if (ret <= 0)
+ return ret;
+ end = op->offset + ret;
+ list_for_each_entry(extent, &op->stream->extents, list) {
+ u64 first = max(op->offset, extent->start);
+ u64 last = min(end, extent->start + extent->length);
+ struct io_opaque_extent *copy;
+
+ if (first >= last)
+ continue;
+ if (!io_opaque_extent_reserve(op->store))
+ return -ENOBUFS;
+ copy = kmalloc_obj(*copy, GFP_KERNEL_ACCOUNT);
+ if (!copy) {
+ io_opaque_extent_unreserve(op->store, 0);
+ return -ENOMEM;
+ }
+ *copy = *extent;
+ copy->embedded = false;
+ refcount_inc(©->backing->refs);
+ copy->start = first - op->offset;
+ copy->offset += first - extent->start;
+ copy->length = last - first;
+ list_add_tail(©->list, &op->data->extents);
+ }
+ op->data->length = ret;
+ return ret;
+}
+
+static int io_opaque_take(struct io_opaque_req *op)
+{
+ struct io_opaque_stream *stream = op->stream;
+ u64 off = op->offset + op->progress;
+ u64 end = op->offset + op->length;
+
+ while (off < end && off < stream->next) {
+ struct io_opaque_extent *extent, *found = NULL;
+ u64 last;
+
+ list_for_each_entry(extent, &stream->extents, list) {
+ if (extent->start + extent->length <= off)
+ continue;
+ if (extent->start <= off)
+ found = extent;
+ break;
+ }
+ if (!found)
+ return -ENODATA;
+ extent = found;
+ if (extent->start < off) {
+ extent = io_opaque_extent_split(op->store, extent, off);
+ if (IS_ERR(extent))
+ return PTR_ERR(extent);
+ }
+ last = min(end, extent->start + extent->length);
+ if (last < extent->start + extent->length) {
+ struct io_opaque_extent *right;
+
+ right = io_opaque_extent_split(op->store, extent, last);
+ if (IS_ERR(right))
+ return PTR_ERR(right);
+ }
+ stream->undecided -= last - off;
+ if (op->op == IORING_OPAQUE_RECV_OBJECT) {
+ struct io_opaque_extent *tail;
+
+ tail = list_last_entry_or_null(&op->data->extents,
+ struct io_opaque_extent, list);
+ if (!tail || tail->backing != extent->backing)
+ op->data->charge += extent->backing->charge;
+ extent->start -= op->offset;
+ list_move_tail(&extent->list, &op->data->extents);
+ op->data->length += last - off;
+ } else {
+ io_opaque_extent_free(op->store, extent);
+ }
+ op->progress += last - off;
+ off = last;
+ }
+ return off == end;
+}
+
+/* An unsuccessful KEEP returns its private prefix to the undecided stream. */
+static void io_opaque_restore(struct io_opaque_req *op)
+{
+ struct io_opaque_stream *stream = op->stream;
+ struct io_opaque_extent *extent, *next;
+ struct list_head *before = &stream->extents;
+
+ lockdep_assert_held(&stream->lock);
+ if (!op->data || !op->data->length || stream->closed)
+ return;
+ list_for_each_entry(extent, &stream->extents, list) {
+ if (extent->start > op->offset) {
+ before = &extent->list;
+ break;
+ }
+ }
+ list_for_each_entry_safe(extent, next, &op->data->extents, list) {
+ extent->start += op->offset;
+ list_move_tail(&extent->list, before);
+ }
+ stream->undecided += op->data->length;
+ op->data->length = 0;
+ op->data->charge = 0;
+ op->progress = 0;
+}
+
+static u64 io_opaque_dense_left(struct io_opaque_req *op)
+{
+ struct io_opaque_extent *tail;
+ u64 left = op->length - op->progress;
+
+ tail = list_last_entry_or_null(&op->data->extents, struct io_opaque_extent, list);
+ if (tail && tail->backing->copied &&
+ tail->offset + tail->length == tail->backing->filled) {
+ u32 room = PAGE_SIZE - tail->backing->filled;
+
+ left -= min_t(u64, left, room);
+ }
+ return ALIGN(left, PAGE_SIZE);
+}
+
+static bool io_opaque_overlap(u64 a, u64 alen, u64 b, u64 blen)
+{
+ return a < b + blen && b < a + alen;
+}
+
+static int io_opaque_range_insert(struct io_opaque_req *op)
+{
+ struct io_opaque_stream *stream = op->stream;
+ struct rb_node **link = &stream->decisions.rb_root.rb_node, *parent = NULL;
+ struct io_opaque_req *other;
+ bool leftmost = true;
+
+ while (*link) {
+ parent = *link;
+ other = rb_entry(parent, struct io_opaque_req, decision);
+ if (io_opaque_overlap(op->offset, op->length, other->offset, other->length))
+ return -EBUSY;
+ if (op->offset < other->offset) {
+ link = &parent->rb_left;
+ } else {
+ link = &parent->rb_right;
+ leftmost = false;
+ }
+ }
+ list_for_each_entry(other, &stream->reads, read)
+ if (io_opaque_overlap(op->offset, op->length, other->offset, 1))
+ return -EBUSY;
+ rb_link_node(&op->decision, parent, link);
+ rb_insert_color_cached(&op->decision, &stream->decisions, leftmost);
+ return 0;
+}
+
+/* Validate a blocked TAKE without taking and restoring its retained prefix. */
+static int io_opaque_defer_take(struct io_opaque_req *op)
+{
+ struct io_opaque_stream *stream = op->stream;
+ struct io_opaque_extent *extent, *tail = NULL;
+ u64 off = op->offset, end = min(off + op->length, stream->next);
+ u64 charge = 0, left;
+
+ /* Allocation backoff can leave an already private prefix in place. */
+ if (op->progress)
+ return 0;
+ list_for_each_entry(extent, &stream->extents, list) {
+ if (extent->start + extent->length <= off)
+ continue;
+ if (extent->start > off)
+ break;
+ if (!tail || tail->backing != extent->backing)
+ charge += extent->backing->charge;
+ tail = extent;
+ off = min(end, extent->start + extent->length);
+ if (off == end)
+ break;
+ }
+ if (off < end)
+ return -ENODATA;
+ left = op->length - (off - op->offset);
+ if (tail && tail->backing->copied &&
+ tail->offset + tail->length == tail->backing->filled)
+ left -= min_t(u64, left, PAGE_SIZE - tail->backing->filled);
+ return charge + ALIGN(left, PAGE_SIZE) > io_opaque_rx_limit(op->store) ?
+ -ENOBUFS : 0;
+}
+
+static void io_opaque_stream_process(struct io_opaque_stream *stream)
+{
+ struct io_opaque_req *op, *next;
+ struct rb_node *node, *after;
+ int ret;
+
+ list_for_each_entry_safe(op, next, &stream->reads, read) {
+ ret = stream->closed ? -ECANCELED :
+ op->op == IORING_OPAQUE_AVAILABLE ?
+ io_opaque_unclassified(stream, op->offset, op->length) :
+ io_opaque_snapshot(op);
+ if (!ret && !stream->eof && !stream->closed && !stream->error)
+ continue;
+ list_del_init(&op->read);
+ if (!ret && stream->error)
+ ret = stream->error;
+ io_opaque_queue_ready(op, ret);
+ }
+ for (node = rb_first_cached(&stream->decisions); node; node = after) {
+ op = rb_entry(node, struct io_opaque_req, decision);
+ after = rb_next(node);
+ if (op->op == IORING_OPAQUE_RECV_OBJECT &&
+ op->offset + op->length > stream->next &&
+ stream->collector &&
+ stream->collector->phase == IO_OPAQUE_BUDGET_POLL &&
+ !stream->closed && !stream->eof && !stream->error)
+ ret = io_opaque_defer_take(op);
+ else
+ ret = stream->closed ? -ECANCELED : io_opaque_take(op);
+ if (!ret && op->op == IORING_OPAQUE_RECV_OBJECT &&
+ op->data->charge + io_opaque_dense_left(op) >
+ io_opaque_rx_limit(op->store))
+ ret = -ENOBUFS;
+ if (!ret && !stream->eof && !stream->closed && !stream->error)
+ continue;
+ rb_erase_cached(node, &stream->decisions);
+ RB_CLEAR_NODE(node);
+ if (!ret)
+ ret = stream->error ?: -ENODATA;
+ if (ret < 0 && op->op == IORING_OPAQUE_RECV_OBJECT)
+ io_opaque_restore(op);
+ io_opaque_queue_ready(op, ret == 1 ? op->length : ret);
+ }
+}
+
+static void io_opaque_stream_close(struct io_opaque_stream *stream)
+{
+ struct socket *sock = sock_from_file(stream->file);
+
+ guard(mutex)(&stream->lock);
+ stream->closed = true;
+ stream->error = -ECANCELED;
+ lock_sock(sock->sk);
+ write_lock_bh(&sock->sk->sk_callback_lock);
+ if (sock->sk->sk_rx_owner == &stream->owner)
+ WRITE_ONCE(sock->sk->sk_rx_owner, NULL);
+ write_unlock_bh(&sock->sk->sk_callback_lock);
+ release_sock(sock->sk);
+ io_opaque_extents_free(stream->store, &stream->extents);
+ stream->undecided = 0;
+ io_opaque_stream_process(stream);
+ wake_up_interruptible_poll(sk_sleep(sock->sk), EPOLLIN);
+ io_opaque_wake_budget(stream->store);
+}
+
+static int io_opaque_copy(struct io_opaque_data *data, u64 off, u32 length,
+ void __user *dest)
+{
+ struct io_opaque_extent *extent;
+ u64 end = off + length, done = off;
+
+ if (end > data->length)
+ return -ENODATA;
+
+ list_for_each_entry(extent, &data->extents, list) {
+ u64 first = max(off, extent->start);
+ u64 last = min(end, extent->start + extent->length);
+ u32 pos = extent->offset + first - extent->start;
+
+ if (first >= last)
+ continue;
+ if (first != done)
+ return -ENODATA;
+ while (first < last) {
+ u32 offset = offset_in_page(pos);
+ u32 chunk = min_t(u64, last - first, PAGE_SIZE - offset);
+ void *mapped = kmap_local_page(extent->backing->page +
+ (pos >> PAGE_SHIFT));
+ unsigned long left;
+
+ left = copy_to_user(dest + first - off, mapped + offset, chunk);
+ kunmap_local(mapped);
+ if (left)
+ return -EFAULT;
+ pos += chunk;
+ first += chunk;
+ }
+ done = last;
+ }
+ return done == end ? length : -ENODATA;
+}
+
+static int io_opaque_publish(struct io_opaque_req *op, u64 *object_flags)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_slot *slot = &store->objects[op->slot];
+
+ guard(mutex)(&store->tables);
+ if (store->dead || READ_ONCE(op->canceled))
+ return -ECANCELED;
+ if (!slot->reserved ||
+ slot->generation != op->handle >> IO_OPAQUE_INDEX_BITS)
+ return -ESTALE;
+ slot->data = op->data;
+ if (op->data->forward_only)
+ *object_flags = IORING_OPAQUE_OBJECT_F_FORWARD_ONLY;
+ store->nr_objects++;
+ op->data = NULL;
+ slot->reserved = false;
+ op->slot_reserved = false;
+ slot->born = jiffies;
+ slot->retry = jiffies;
+ return op->length;
+}
+
+static void io_opaque_ready(struct io_tw_req tw_req, io_tw_token_t tw)
+{
+ struct io_kiocb *req = tw_req.req;
+ struct io_opaque_req *op = req->async_data;
+ struct io_opaque_store *store = op->store;
+ int ret = op->result;
+ u64 object_flags = 0;
+
+ io_tw_lock(req->ctx, tw);
+ list_del_init(&op->cancel);
+ op->phase = IO_OPAQUE_IDLE;
+ if (tw.cancel || READ_ONCE(op->canceled) || READ_ONCE(store->dead))
+ ret = -ECANCELED;
+ if (ret >= 0 && op->op == IORING_OPAQUE_INSPECT)
+ ret = io_opaque_copy(op->data, 0, op->data->length, op->addr);
+ if (ret >= 0 && op->op == IORING_OPAQUE_RECV_OBJECT) {
+ ret = op->data->length == op->length ?
+ io_opaque_vector(op->data) : -ENODATA;
+ if (!ret)
+ ret = io_opaque_publish(op, &object_flags);
+ }
+ if (ret < 0 && op->op == IORING_OPAQUE_RECV_OBJECT) {
+ mutex_lock(&op->stream->lock);
+ io_opaque_restore(op);
+ io_opaque_stream_process(op->stream);
+ mutex_unlock(&op->stream->lock);
+ }
+ if (ret < 0)
+ req_set_fail(req);
+ if (ret >= 0 && op->op == IORING_OPAQUE_RECV_OBJECT)
+ io_req_set_res32(req, ret, 0, op->handle, object_flags);
+ else
+ io_req_set_res(req, ret, 0);
+ io_req_task_complete(tw_req, tw);
+}
+
+static struct io_opaque_store *io_opaque_get_store(struct io_kiocb *req, u32 id)
+{
+ return xa_load(&req->ctx->opaque_stores, id);
+}
+
+static bool io_opaque_stream_op(u16 op)
+{
+ return op == IORING_OPAQUE_INSPECT || op == IORING_OPAQUE_AVAILABLE ||
+ op == IORING_OPAQUE_RECV_OBJECT ||
+ op == IORING_OPAQUE_DISCARD || op == IORING_OPAQUE_STREAM_STAT ||
+ op == IORING_OPAQUE_STREAM_CLOSE;
+}
+
+int io_opaque_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
+{
+ struct io_opaque_store *store = io_opaque_get_store(req, READ_ONCE(sqe->opaque_store_id));
+ struct io_opaque_req *op;
+ u64 end;
+ u16 cmd = READ_ONCE(sqe->ioprio);
+
+ if (!store)
+ return -ENXIO;
+ if (READ_ONCE(sqe->fd) != -1 || READ_ONCE(sqe->rw_flags) ||
+ READ_ONCE(sqe->buf_index) || READ_ONCE(sqe->__pad2[0]) ||
+ cmd > IORING_OPAQUE_AVAILABLE ||
+ cmd == IORING_OPAQUE_COMPACT || cmd == IORING_OPAQUE_SET_POLICY ||
+ cmd == IORING_OPAQUE_CREATE ||
+ (req->flags & (REQ_F_CQE_SKIP | REQ_F_FIXED_FILE)))
+ return -EINVAL;
+ op = io_opaque_req_alloc(req, store);
+ if (IS_ERR(op))
+ return PTR_ERR(op);
+ op->op = cmd;
+ if (cmd <= IORING_OPAQUE_DISCARD || cmd == IORING_OPAQUE_AVAILABLE) {
+ INIT_LIST_HEAD(&op->read);
+ RB_CLEAR_NODE(&op->decision);
+ }
+ op->handle = READ_ONCE(sqe->addr);
+ op->offset = READ_ONCE(sqe->off);
+ op->length = READ_ONCE(sqe->len);
+ op->addr = u64_to_user_ptr(READ_ONCE(sqe->addr3));
+ if (check_add_overflow(op->offset, (u64)op->length, &end))
+ return -EOVERFLOW;
+ if (cmd <= IORING_OPAQUE_READ_OBJECT || cmd == IORING_OPAQUE_AVAILABLE) {
+ if (!op->length || op->length > INT_MAX)
+ return -EINVAL;
+ if ((cmd == IORING_OPAQUE_INSPECT || cmd == IORING_OPAQUE_READ_OBJECT) &&
+ (op->length > IO_OPAQUE_READ_MAX || !op->addr))
+ return -EINVAL;
+ if ((cmd == IORING_OPAQUE_RECV_OBJECT || cmd == IORING_OPAQUE_DISCARD ||
+ cmd == IORING_OPAQUE_AVAILABLE) && op->addr)
+ return -EINVAL;
+ if (cmd == IORING_OPAQUE_RECV_OBJECT &&
+ (op->length > store->config.max_object_size ||
+ op->length > io_opaque_rx_limit(store)))
+ return -EMSGSIZE;
+ } else if (op->offset) {
+ return -EINVAL;
+ }
+ if (cmd == IORING_OPAQUE_FREE || cmd == IORING_OPAQUE_STREAM_CLOSE) {
+ if (op->length || op->addr)
+ return -EINVAL;
+ }
+ if (cmd == IORING_OPAQUE_STAT) {
+ if (op->handle || !op->addr)
+ return -EINVAL;
+ }
+ return 0;
+}
+
+/* Resolve resources when links permit execution, never while preparing SQEs. */
+static int io_opaque_resolve(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_slot *slot;
+ u16 cmd = op->op;
+ int ret = 0;
+
+ if (op->resolved)
+ return 0;
+ mutex_lock(&store->tables);
+ if (io_opaque_stream_op(cmd)) {
+ u32 index = io_opaque_index(op->handle);
+
+ if (!(op->handle & IO_OPAQUE_TOKEN_TAG) || index >= store->config.max_streams ||
+ store->streams[index].generation != io_opaque_token_generation(op->handle) ||
+ !store->streams[index].stream) {
+ ret = -ESTALE;
+ } else {
+ op->stream = store->streams[index].stream;
+ refcount_inc(&op->stream->refs);
+ op->req->file = get_file(op->stream->file);
+ }
+ } else if (cmd == IORING_OPAQUE_READ_OBJECT || cmd == IORING_OPAQUE_OBJECT_STAT) {
+ slot = io_opaque_lookup(store, op->handle);
+ if (!slot || !slot->data) {
+ ret = -ESTALE;
+ } else {
+ op->data = slot->data;
+ refcount_inc(&op->data->refs);
+ if (cmd == IORING_OPAQUE_READ_OBJECT &&
+ (op->offset > op->data->length ||
+ op->length > op->data->length - op->offset))
+ ret = -ERANGE;
+ }
+ }
+ mutex_unlock(&store->tables);
+ if (ret)
+ return ret;
+ if (cmd == IORING_OPAQUE_INSPECT || cmd == IORING_OPAQUE_RECV_OBJECT) {
+ op->data = io_opaque_data_alloc();
+ if (!op->data)
+ return -ENOMEM;
+ }
+ if (cmd == IORING_OPAQUE_RECV_OBJECT) {
+ ret = io_opaque_slot_reserve(op);
+ if (ret)
+ return ret;
+ }
+ op->resolved = true;
+ return 0;
+}
+
+static int io_opaque_range_start(struct io_opaque_req *op)
+{
+ struct io_opaque_stream *stream = op->stream;
+ struct rb_node *node;
+ int ret = 0;
+
+ guard(mutex)(&stream->lock);
+ if (stream->closed)
+ return -ECANCELED;
+ if (op->op == IORING_OPAQUE_INSPECT || op->op == IORING_OPAQUE_AVAILABLE) {
+ for (node = rb_first_cached(&stream->decisions); node; node = rb_next(node)) {
+ struct io_opaque_req *other;
+
+ other = rb_entry(node, struct io_opaque_req, decision);
+
+ if (io_opaque_overlap(op->offset, 1, other->offset, other->length))
+ return -EBUSY;
+ }
+ list_add_tail(&op->read, &stream->reads);
+ } else {
+ ret = io_opaque_range_insert(op);
+ if (ret)
+ return ret;
+ }
+ io_opaque_wait(op, IO_OPAQUE_RANGE_WAIT);
+ io_opaque_stream_process(stream);
+ /* A decision, or an INSPECT still waiting for bytes, may let collection continue. */
+ if ((op->op != IORING_OPAQUE_INSPECT && op->op != IORING_OPAQUE_AVAILABLE) ||
+ !list_empty(&op->read))
+ io_opaque_wake_stream(stream);
+ return IOU_ISSUE_SKIP_COMPLETE;
+}
+
+static int io_opaque_stat_copy(struct io_opaque_req *op, const void *stat, u32 size)
+{
+ if (!op->addr || op->length < sizeof(u32) || op->length > PAGE_SIZE)
+ return -EINVAL;
+ return copy_struct_to_user(op->addr, op->length, stat, size, NULL);
+}
+
+static int io_opaque_stat(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_uring_opaque_stat stat = {
+ .size = sizeof(stat),
+ .backing_bytes = atomic64_read(&store->bytes),
+ .hard_limit = store->config.hard_limit,
+ .copied_bytes = atomic64_read(&store->copied),
+ .compacted_bytes = atomic64_read(&store->compacted),
+ .compact_count = atomic64_read(&store->compactions),
+ .extents = atomic_read(&store->extents),
+ .framing_bytes = atomic64_read(&store->framing),
+ .framing_copied_bytes = atomic64_read(&store->framing_copied),
+ .memory_mode = io_opaque_memory_mode(store),
+ };
+
+ /* Framing in the reply reserve is charged outside bytes. */
+ stat.backing_bytes += atomic64_read(&store->reserve_framing);
+ mutex_lock(&store->tables);
+ stat.objects = store->nr_objects;
+ stat.streams = store->nr_streams;
+ mutex_unlock(&store->tables);
+ return io_opaque_stat_copy(op, &stat, sizeof(stat));
+}
+
+int io_opaque_issue(struct io_kiocb *req, unsigned int issue_flags)
+{
+ struct io_opaque_req *op = req->async_data;
+ struct io_opaque_store *store = op->store;
+ int ret;
+
+ io_ring_submit_lock(req->ctx, issue_flags);
+ /* Pair with the cancellation search before any opaque wait enrollment. */
+ if (io_wq_cancel_seen(req, issue_flags)) {
+ ret = -ECANCELED;
+ goto out;
+ }
+ if (READ_ONCE(store->dead)) {
+ ret = -ESHUTDOWN;
+ goto out;
+ }
+ ret = io_opaque_resolve(op);
+ if (ret)
+ goto out;
+ switch (op->op) {
+ case IORING_OPAQUE_INSPECT:
+ case IORING_OPAQUE_AVAILABLE:
+ case IORING_OPAQUE_RECV_OBJECT:
+ case IORING_OPAQUE_DISCARD:
+ ret = io_opaque_range_start(op);
+ break;
+ case IORING_OPAQUE_READ_OBJECT:
+ ret = io_opaque_copy(op->data, op->offset, op->length, op->addr);
+ break;
+ case IORING_OPAQUE_FREE:
+ mutex_lock(&store->tables);
+ {
+ struct io_opaque_data *data = io_opaque_detach(store,
+ io_opaque_lookup(store, op->handle));
+
+ ret = data ? 0 : -ESTALE;
+ mutex_unlock(&store->tables);
+ io_opaque_data_put(store, data);
+ }
+ break;
+ case IORING_OPAQUE_STAT:
+ ret = io_opaque_stat(op);
+ break;
+ case IORING_OPAQUE_OBJECT_STAT:
+ {
+ struct io_uring_opaque_object_stat stat = {
+ .size = sizeof(stat),
+ .length = op->data->length,
+ .backing_bytes = op->data->charge,
+ .extents = op->data->nr,
+ .flags = op->data->dense ? IORING_OPAQUE_OBJECT_F_COMPACTED : 0,
+ };
+ struct io_opaque_slot *slot;
+
+ if (op->data->forward_only)
+ stat.flags |= IORING_OPAQUE_OBJECT_F_FORWARD_ONLY;
+ mutex_lock(&store->tables);
+ slot = io_opaque_lookup(store, op->handle);
+ if (slot && slot->compacting)
+ stat.flags |= IORING_OPAQUE_OBJECT_F_COMPACTING;
+ mutex_unlock(&store->tables);
+ ret = io_opaque_stat_copy(op, &stat, sizeof(stat));
+ }
+ break;
+ case IORING_OPAQUE_STREAM_STAT:
+ {
+ struct io_uring_opaque_stream_stat stat = { .size = sizeof(stat) };
+
+ mutex_lock(&op->stream->lock);
+ stat.rx_next = op->stream->next;
+ stat.undecided_bytes = op->stream->undecided;
+ stat.flags = op->stream->eof ? IORING_OPAQUE_STREAM_F_EOF : 0;
+ stat.error = op->stream->error;
+ mutex_unlock(&op->stream->lock);
+ ret = io_opaque_stat_copy(op, &stat, sizeof(stat));
+ }
+ break;
+ case IORING_OPAQUE_STREAM_CLOSE:
+ mutex_lock(&store->tables);
+ if (store->streams[io_opaque_index(op->handle)].stream == op->stream) {
+ store->streams[io_opaque_index(op->handle)].stream = NULL;
+ store->nr_streams--;
+ io_opaque_release_slot(store->stream_used, io_opaque_index(op->handle),
+ io_opaque_token_generation(op->handle));
+ mutex_unlock(&store->tables);
+ io_opaque_stream_close(op->stream);
+ io_opaque_stream_put(op->stream);
+ ret = 0;
+ } else {
+ mutex_unlock(&store->tables);
+ ret = -ESTALE;
+ }
+ break;
+ default:
+ ret = -EINVAL;
+ }
+out:
+ if (ret != IOU_ISSUE_SKIP_COMPLETE) {
+ if (ret < 0)
+ req_set_fail(req);
+ io_req_set_res(req, ret, 0);
+ ret = IOU_COMPLETE;
+ }
+ io_ring_submit_unlock(req->ctx, issue_flags);
+ return ret;
+}
+
+static bool io_opaque_cancel_req(struct io_opaque_req *op)
+{
+ enum io_opaque_phase phase;
+ bool queue;
+
+ if (op->stream)
+ mutex_lock(&op->stream->lock);
+ spin_lock(&op->store->wait_lock);
+ phase = op->phase;
+ if (op->canceled || phase == IO_OPAQUE_IDLE) {
+ spin_unlock(&op->store->wait_lock);
+ if (op->stream)
+ mutex_unlock(&op->stream->lock);
+ return false;
+ }
+ WRITE_ONCE(op->canceled, true);
+ /* A queued copy is owned by the shared worker until it reports readiness. */
+ if (phase != IO_OPAQUE_COPY_WORK)
+ list_del_init(&op->wait);
+ /* Claim task work before a remote TX head can promote this waiter. */
+ queue = phase != IO_OPAQUE_QUEUED && phase != IO_OPAQUE_COPY_WORK;
+ if (queue)
+ op->phase = IO_OPAQUE_QUEUED;
+ spin_unlock(&op->store->wait_lock);
+ if (phase == IO_OPAQUE_RANGE_WAIT) {
+ list_del_init(&op->read);
+ if (!RB_EMPTY_NODE(&op->decision)) {
+ rb_erase_cached(&op->decision, &op->stream->decisions);
+ RB_CLEAR_NODE(&op->decision);
+ }
+ }
+ if (op->stream) {
+ if (op->op == IORING_OPAQUE_RECV_OBJECT)
+ io_opaque_restore(op);
+ mutex_unlock(&op->stream->lock);
+ }
+ if (queue)
+ io_opaque_queue_ready(op, -ECANCELED);
+ return true;
+}
+
+int io_opaque_cancel(struct io_ring_ctx *ctx, struct io_cancel_data *cd,
+ unsigned int issue_flags)
+{
+ struct io_opaque_req *op;
+ int nr = 0;
+
+ io_ring_submit_lock(ctx, issue_flags);
+ list_for_each_entry(op, &ctx->opaque_waits, cancel) {
+ if (!io_cancel_req_match(op->req, cd) || !io_opaque_cancel_req(op))
+ continue;
+ nr++;
+ break;
+ }
+ io_ring_submit_unlock(ctx, issue_flags);
+ return nr ? 0 : -ENOENT;
+}
+
+bool io_opaque_cancel_all(struct io_ring_ctx *ctx, struct io_uring_task *tctx,
+ bool cancel_all)
+{
+ struct io_opaque_req *op;
+ bool found = false;
+
+ lockdep_assert_held(&ctx->uring_lock);
+ list_for_each_entry(op, &ctx->opaque_waits, cancel)
+ if (io_match_task_safe(op->req, tctx, cancel_all))
+ found |= io_opaque_cancel_req(op);
+ return found;
+}
+
+void io_opaque_cleanup(struct io_kiocb *req)
+{
+ struct io_opaque_req *op = req->async_data;
+ struct io_opaque_store *store;
+ bool close = false;
+
+ if (!op)
+ return;
+ store = op->store;
+ list_del_init(&op->cancel);
+ scoped_guard(spinlock, &store->wait_lock)
+ list_del_init(&op->wait);
+ /* Waiters have stopped accessing this request before cleanup. */
+ if (op->stream) {
+ mutex_lock(&op->stream->lock);
+ if (op->op <= IORING_OPAQUE_DISCARD || op->op == IORING_OPAQUE_AVAILABLE) {
+ list_del_init(&op->read);
+ if (!RB_EMPTY_NODE(&op->decision)) {
+ rb_erase_cached(&op->decision, &op->stream->decisions);
+ RB_CLEAR_NODE(&op->decision);
+ }
+ if (op->op == IORING_OPAQUE_RECV_OBJECT)
+ io_opaque_restore(op);
+ }
+ if (op->stream->collector == op) {
+ op->stream->collector = NULL;
+ close = !op->stream->eof && !op->stream->closed && !op->stream->error;
+ }
+ mutex_unlock(&op->stream->lock);
+ if (close)
+ io_opaque_stream_close(op->stream);
+ }
+ if (op->slot_reserved) {
+ mutex_lock(&store->tables);
+ store->objects[op->slot].reserved = false;
+ io_opaque_release_slot(store->object_used, op->slot,
+ store->objects[op->slot].generation);
+ mutex_unlock(&store->tables);
+ }
+ io_opaque_data_put(store, op->data);
+ io_opaque_stream_put(op->stream);
+ lockdep_assert_held(&req->ctx->uring_lock);
+ io_cache_free(&req->ctx->opaque_cache, op);
+ io_req_async_data_clear(req, 0);
+}
diff --git a/io_uring/opaque_register.c b/io_uring/opaque_register.c
new file mode 100644
index 000000000000..9e39d1ee5072
--- /dev/null
+++ b/io_uring/opaque_register.c
@@ -0,0 +1,90 @@
+// SPDX-License-Identifier: GPL-2.0
+/* Ring-local registration IDs for shared kernel-resident object stores. */
+#include <linux/anon_inodes.h>
+#include <linux/file.h>
+#include <linux/slab.h>
+#include <linux/uaccess.h>
+#include <uapi/linux/io_uring/opaque.h>
+
+#include "io_uring.h"
+#include "opaque.h"
+
+static void io_opaque_unregister_user(struct io_opaque_store *store)
+{
+ io_opaque_user_put(store);
+ io_opaque_put(store);
+}
+
+int io_register_opaque(struct io_ring_ctx *ctx, void __user *arg)
+{
+ struct io_uring_opaque_register reg;
+ struct io_uring_opaque_config config;
+ struct io_opaque_store *store;
+ u32 size, id;
+ int ret;
+
+ lockdep_assert_held(&ctx->uring_lock);
+ if (!(ctx->flags & IORING_SETUP_DEFER_TASKRUN) ||
+ !(ctx->flags & (IORING_SETUP_CQE32 | IORING_SETUP_CQE_MIXED)))
+ return -EINVAL;
+ if (get_user(size, &((struct io_uring_opaque_register __user *)arg)->size))
+ return -EFAULT;
+ if (size < offsetof(struct io_uring_opaque_register, __resv) || size > PAGE_SIZE)
+ return -EINVAL;
+ ret = copy_struct_from_user(®, sizeof(reg), arg, size);
+ if (ret)
+ return ret;
+ if (reg.size != size || reg.__resv0 ||
+ !mem_is_zero(reg.__resv, sizeof(reg.__resv)))
+ return -EINVAL;
+ reg.size = sizeof(reg);
+ if (reg.store_id || reg.flags)
+ return -EINVAL;
+ if (reg.fd != -1 || !reg.config ||
+ reg.config_size < offsetof(struct io_uring_opaque_config, __resv0) ||
+ reg.config_size > PAGE_SIZE)
+ return -EINVAL;
+ ret = copy_struct_from_user(&config, sizeof(config),
+ u64_to_user_ptr(reg.config), reg.config_size);
+ if (ret)
+ return ret;
+ store = io_opaque_alloc(ctx, &config);
+ if (IS_ERR(store))
+ return PTR_ERR(store);
+ ret = xa_alloc(&ctx->opaque_stores, &id, store, xa_limit_31b, GFP_KERNEL);
+ if (ret)
+ goto put_store;
+ reg.store_id = id;
+ ret = copy_struct_to_user(arg, size, ®, sizeof(reg), NULL);
+ if (!ret)
+ return 0;
+ xa_erase(&ctx->opaque_stores, id);
+put_store:
+ io_opaque_unregister_user(store);
+ return ret;
+}
+
+void io_terminate_opaque(struct io_ring_ctx *ctx)
+{
+ struct io_opaque_store *store;
+ unsigned long index;
+
+ lockdep_assert_held(&ctx->uring_lock);
+ if (ctx->opaque_stores_dying)
+ return;
+ ctx->opaque_stores_dying = true;
+ xa_for_each(&ctx->opaque_stores, index, store)
+ io_opaque_user_put(store);
+}
+
+void io_unregister_opaque(struct io_ring_ctx *ctx)
+{
+ struct io_opaque_store *store;
+ unsigned long index;
+
+ lockdep_assert_held(&ctx->uring_lock);
+ io_terminate_opaque(ctx);
+ xa_for_each(&ctx->opaque_stores, index, store)
+ io_opaque_put(store);
+ xa_destroy(&ctx->opaque_stores);
+}
diff --git a/io_uring/opdef.c b/io_uring/opdef.c
index cf3aa2242cd7..256c7c533118 100644
--- a/io_uring/opdef.c
+++ b/io_uring/opdef.c
@@ -38,6 +38,7 @@
#include "futex.h"
#include "truncate.h"
#include "zcrx.h"
+#include "opaque.h"
static int io_no_issue(struct io_kiocb *req, unsigned int issue_flags)
{
@@ -591,6 +592,17 @@ const struct io_issue_def io_issue_defs[] = {
.prep = io_uring_cmd_prep,
.issue = io_uring_cmd,
},
+ [IORING_OP_OPAQUE_OBJ] = {
+ .audit_skip = 1,
+ .ioprio = 1,
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+ .async_size = sizeof(struct io_opaque_req),
+ .prep = io_opaque_prep,
+ .issue = io_opaque_issue,
+#else
+ .prep = io_eopnotsupp_prep,
+#endif
+ },
};
const struct io_cold_def io_cold_defs[] = {
@@ -849,6 +861,10 @@ const struct io_cold_def io_cold_defs[] = {
.sqe_copy = io_uring_cmd_sqe_copy,
.cleanup = io_uring_cmd_cleanup,
},
+ [IORING_OP_OPAQUE_OBJ] = {
+ .name = "OPAQUE_OBJ",
+ .cleanup = io_opaque_cleanup,
+ },
};
const char *io_uring_get_opcode(u8 opcode)
diff --git a/io_uring/register.c b/io_uring/register.c
index 2ce36623e25a..1e450554ca38 100644
--- a/io_uring/register.c
+++ b/io_uring/register.c
@@ -32,6 +32,7 @@
#include "msg_ring.h"
#include "memmap.h"
#include "zcrx.h"
+#include "opaque.h"
#include "query.h"
#include "bpf_filter.h"
@@ -1013,6 +1014,12 @@ static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
case IORING_REGISTER_QUERY:
ret = io_query(arg, nr_args);
break;
+ case IORING_REGISTER_OPAQUE_STORE:
+ ret = -EINVAL;
+ if (!arg || nr_args != 1)
+ break;
+ ret = io_register_opaque(ctx, arg);
+ break;
case IORING_REGISTER_ZCRX_CTRL:
ret = io_zcrx_ctrl(ctx, arg, nr_args);
break;
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 06/12] io_uring/opaque: allow rings to share a store
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (4 preceding siblings ...)
2026-11-10 13:19 ` [RFC PATCH net-next 05/12] io_uring/opaque: add store registration and IORING_OP_OPAQUE_OBJ Alex Markuze
@ 2026-11-10 13:20 ` Alex Markuze
2026-11-10 13:20 ` [RFC PATCH net-next 07/12] io_uring/opaque: add IORING_OP_OPAQUE_OBJ_SEND Alex Markuze
` (5 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:20 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
A multi-threaded server often uses one ring per thread. Allow an object
received or created through one ring to be used through another ring
without copying its payload.
Add EXPORT and IMPORT controls to IORING_REGISTER_OPAQUE_STORE. EXPORT
returns a close-on-exec anonymous-inode descriptor; IMPORT registers its
store under a new ring-local ID. The descriptor cannot read, map or poll
payload. Handles and stream tokens belong to the store and are valid
through every ring sharing it.
Each registered ring entry and export file holds one user reference and
one lifetime reference. The last user stops the store; the last lifetime
reference releases its memory and accounting. An export file can keep a
store alive after the creating ring exits. Import preserves the store's
memory-cgroup ownership and original ring user/mm accounting and adds no
backing charge.
Export reserves the descriptor and creates the file, copies the number
to userspace, and installs it only after that copy succeeds. Failure
unwinds the references and descriptor reservation. Import erases the
new ring-local ID if copying its result to userspace fails.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
include/uapi/linux/io_uring/opaque.h | 3 +
io_uring/opaque_register.c | 84 ++++++++++++++++++++++++----
2 files changed, 76 insertions(+), 11 deletions(-)
diff --git a/include/uapi/linux/io_uring/opaque.h b/include/uapi/linux/io_uring/opaque.h
index 692daa7d7e32..9a129b5da5af 100644
--- a/include/uapi/linux/io_uring/opaque.h
+++ b/include/uapi/linux/io_uring/opaque.h
@@ -16,6 +16,9 @@ struct io_uring_opaque_register {
__u64 __resv[2];
};
+#define IORING_OPAQUE_REG_IMPORT (1U << 0)
+#define IORING_OPAQUE_REG_EXPORT (1U << 1)
+
/* config/config_size select a size-versioned configuration on creation. */
struct io_uring_opaque_config {
__u64 hard_limit;
diff --git a/io_uring/opaque_register.c b/io_uring/opaque_register.c
index 9e39d1ee5072..472c739a17de 100644
--- a/io_uring/opaque_register.c
+++ b/io_uring/opaque_register.c
@@ -15,6 +15,50 @@ static void io_opaque_unregister_user(struct io_opaque_store *store)
io_opaque_put(store);
}
+static int io_opaque_box_release(struct inode *inode, struct file *file)
+{
+ io_opaque_unregister_user(file->private_data);
+ return 0;
+}
+
+static const struct file_operations io_opaque_box_fops = {
+ .owner = THIS_MODULE,
+ .release = io_opaque_box_release,
+};
+
+static int io_opaque_export(struct io_ring_ctx *ctx, void __user *arg,
+ struct io_uring_opaque_register *reg, u32 size)
+{
+ struct io_opaque_store *store = xa_load(&ctx->opaque_stores, reg->store_id);
+ struct file *file;
+ int fd, ret;
+
+ if (!store)
+ return -ENXIO;
+ fd = get_unused_fd_flags(O_CLOEXEC);
+ if (fd < 0)
+ return fd;
+ io_opaque_get(store);
+ file = anon_inode_create_getfile("[io_uring-opaque]", &io_opaque_box_fops,
+ store, O_CLOEXEC, NULL);
+ if (IS_ERR(file)) {
+ ret = PTR_ERR(file);
+ io_opaque_unregister_user(store);
+ goto put_fd;
+ }
+ reg->fd = fd;
+ ret = copy_struct_to_user(arg, size, reg, sizeof(*reg), NULL);
+ if (ret) {
+ fput(file);
+ goto put_fd;
+ }
+ fd_install(fd, file);
+ return 0;
+put_fd:
+ put_unused_fd(fd);
+ return ret;
+}
+
int io_register_opaque(struct io_ring_ctx *ctx, void __user *arg)
{
struct io_uring_opaque_register reg;
@@ -38,19 +82,37 @@ int io_register_opaque(struct io_ring_ctx *ctx, void __user *arg)
!mem_is_zero(reg.__resv, sizeof(reg.__resv)))
return -EINVAL;
reg.size = sizeof(reg);
- if (reg.store_id || reg.flags)
+ if (reg.flags == IORING_OPAQUE_REG_EXPORT) {
+ if (reg.fd != -1 || reg.config || reg.config_size)
+ return -EINVAL;
+ return io_opaque_export(ctx, arg, ®, size);
+ }
+ if (reg.store_id)
return -EINVAL;
- if (reg.fd != -1 || !reg.config ||
- reg.config_size < offsetof(struct io_uring_opaque_config, __resv0) ||
- reg.config_size > PAGE_SIZE)
+ if (reg.flags == IORING_OPAQUE_REG_IMPORT) {
+ CLASS(fd, f)(reg.fd);
+
+ if (reg.config || reg.config_size)
+ return -EINVAL;
+ if (fd_empty(f) || fd_file(f)->f_op != &io_opaque_box_fops)
+ return -EBADF;
+ store = fd_file(f)->private_data;
+ io_opaque_get(store);
+ } else if (!reg.flags) {
+ if (reg.fd != -1 || !reg.config ||
+ reg.config_size < offsetof(struct io_uring_opaque_config, __resv0) ||
+ reg.config_size > PAGE_SIZE)
+ return -EINVAL;
+ ret = copy_struct_from_user(&config, sizeof(config),
+ u64_to_user_ptr(reg.config), reg.config_size);
+ if (ret)
+ return ret;
+ store = io_opaque_alloc(ctx, &config);
+ if (IS_ERR(store))
+ return PTR_ERR(store);
+ } else {
return -EINVAL;
- ret = copy_struct_from_user(&config, sizeof(config),
- u64_to_user_ptr(reg.config), reg.config_size);
- if (ret)
- return ret;
- store = io_opaque_alloc(ctx, &config);
- if (IS_ERR(store))
- return PTR_ERR(store);
+ }
ret = xa_alloc(&ctx->opaque_stores, &id, store, xa_limit_31b, GFP_KERNEL);
if (ret)
goto put_store;
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 07/12] io_uring/opaque: add IORING_OP_OPAQUE_OBJ_SEND
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (5 preceding siblings ...)
2026-11-10 13:20 ` [RFC PATCH net-next 06/12] io_uring/opaque: allow rings to share a store Alex Markuze
@ 2026-11-10 13:20 ` Alex Markuze
2026-11-10 13:20 ` [RFC PATCH net-next 08/12] io_uring/opaque: add vectored IORING_OP_OPAQUE_OBJ_SEND Alex Markuze
` (4 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:20 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Add an operation that queues an optional prefix, an object range and an
optional suffix to a TCP socket, reporting the queued byte count in one
CQE.
sqe->fd names the socket; addr, off and len select the object range;
addr3 optionally points to an io_uring_opaque_frame descriptor. Combined
framing is limited to 64 KiB. ioprio accepts SEND_LAST and msg_flags
accepts MSG_MORE and MSG_DONTWAIT. With a zero handle, offset and length,
the request sends only framing bytes as an object-free reply.
Preparation copies the frame descriptor when the kernel consumes this
SQE. Its storage must survive until that consumption, which a short or
failed submission may leave pending. Prefix and suffix bytes are copied
at first issue, before handle acquisition, and must remain valid until
then. The request owns the resulting copy, charged to the store and its
memory cgroup; framing can use the reply reserve.
The body uses MSG_SPLICE_PAGES over the immutable version's bio_vecs.
Indexed segment lookup finds the range's start. One cursor tracks prefix,
body and suffix so retries continue at the first unqueued byte. Without
MSG_DONTWAIT, the request waits for socket space until it finishes or an
error, signal or cancellation ends it. The result counts bytes queued to
TCP, not delivered to the peer.
Sends through one store to one socket enter a FIFO under the store's TX
lock after frame import during first issue. The ordering is by enqueue,
not SQE submission or the start of issue on concurrent rings. Only
the head sends and polls; followers resume through their own ring's task
work. The head leaves when its frame finishes or terminates. If it ends
after queuing only part of the frame, its followers complete -ECANCELED,
including admissions still acquiring object references. The failure
remains visible until both linked requests and admissions drain. A head
that queued nothing does not fail its followers. Cancellation of linked
followers proceeds in batches of 32 outside the queue lock between
batches. Writers outside this store do not participate in the FIFO.
The store indexes queues by socket address under tx_lock. Up to 64 empty
queues are retained for reuse. Their keys are compared, never
dereferenced; a reused address finds an empty, nonfailed queue.
An ordinary send pins the current version while leaving the handle
usable. SEND_LAST removes the handle at acquisition and transfers its
reference to the request. CONSUMED reports that transfer even if the
request later fails, is canceled or completes short. A failure before
acquisition consumes nothing. Competing LAST requests cannot both acquire
the same handle.
For plain TCP, the head acquires its references, sends each frame part,
advances the queue and drops nonfinal references under one socket lock.
Final version release follows release_sock(). The issue performs the
security hook, RPS recording and send-length tracing normally supplied
by sock_sendmsg(); the hook runs before lock_sock(), since it may take
that lock. A replaced TCP protocol uses sock_sendmsg() instead, and the
direct path rechecks the protocol between frame parts after TCP may have
dropped the socket lock.
With IO_URING_F_UNLOCKED, the ring lock is dropped before acquiring the
socket lock. Inline issue retains its ring lock. Nothing may wait for a
socket lock while holding the table mutex. Short and failed sends fail
their link chain.
Reserve the FIFO position in the first TX lock hold, before acquiring
references, including queue creation, but after frame import. Queue metadata
uses accounted nonblocking allocation; ENOMEM precedes handle acquisition
or LAST. A busy queue holds a socket reference independently of request
files. That reference is released outside the TX lock when its last
position drains. Idle queues pin no socket. Pending failures keep their
position through cleanup, preserving
the failure barrier without allowing socket-address reuse.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
include/uapi/linux/io_uring.h | 4 +
include/uapi/linux/io_uring/opaque.h | 3 +
io_uring/opaque.c | 933 ++++++++++++++++++++++++++-
io_uring/opdef.c | 21 +
4 files changed, 958 insertions(+), 3 deletions(-)
diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
index 081f30605410..1759f4f805b1 100644
--- a/include/uapi/linux/io_uring.h
+++ b/include/uapi/linux/io_uring.h
@@ -321,6 +321,7 @@ enum io_uring_op {
IORING_OP_NOP128,
IORING_OP_URING_CMD128,
IORING_OP_OPAQUE_OBJ,
+ IORING_OP_OPAQUE_OBJ_SEND,
/* this goes last, obviously */
IORING_OP_LAST,
@@ -534,6 +535,8 @@ struct io_uring_cqe {
* if a large CQE is attempted posted when the ring has
* just a single small CQE worth of space left before
* wrapping.
+ * IORING_CQE_F_OPAQUE_CONSUMED OPAQUE_OBJ_SEND took ownership of the entire
+ * object handle. Also valid on short or failed completions.
* IORING_CQE_F_32 If set, this is a 32b/big-cqe posting. Use with rings
* setup in a mixed CQE mode, where both 16b and 32b
* CQEs may be posted to the CQ ring.
@@ -544,6 +547,7 @@ struct io_uring_cqe {
#define IORING_CQE_F_NOTIF (1U << 3)
#define IORING_CQE_F_BUF_MORE (1U << 4)
#define IORING_CQE_F_SKIP (1U << 5)
+#define IORING_CQE_F_OPAQUE_CONSUMED (1U << 6)
#define IORING_CQE_F_32 (1U << 15)
#define IORING_CQE_BUFFER_SHIFT 16
diff --git a/include/uapi/linux/io_uring/opaque.h b/include/uapi/linux/io_uring/opaque.h
index 9a129b5da5af..2a862d76a40e 100644
--- a/include/uapi/linux/io_uring/opaque.h
+++ b/include/uapi/linux/io_uring/opaque.h
@@ -59,6 +59,9 @@ enum io_uring_opaque_op {
IORING_OPAQUE_AVAILABLE,
};
+/* IORING_OP_OPAQUE_OBJ_SEND: sqe->ioprio contains these flags. */
+#define IORING_OPAQUE_SEND_LAST (1U << 0)
+
/* Optional SEND sqe->addr3 framing; size includes zero reserved extensions. */
struct io_uring_opaque_frame {
__u32 size;
diff --git a/io_uring/opaque.c b/io_uring/opaque.c
index 81fbf2011446..b62c4be92fea 100644
--- a/io_uring/opaque.c
+++ b/io_uring/opaque.c
@@ -12,7 +12,9 @@
#include <linux/sizes.h>
#include <linux/slab.h>
#include <linux/workqueue.h>
+#include <net/rps.h>
#include <net/tcp.h>
+#include <trace/events/sock.h>
#include <uapi/linux/io_uring/opaque.h>
#include "io_uring.h"
@@ -32,8 +34,10 @@ void io_opaque_cache_free(struct io_ring_ctx *ctx)
io_alloc_cache_free(&ctx->opaque_cache, kfree);
}
+static void io_opaque_resume(struct io_tw_req tw_req, io_tw_token_t tw);
static void io_opaque_ready(struct io_tw_req tw_req, io_tw_token_t tw);
static void io_opaque_stream_close(struct io_opaque_stream *stream);
+static void io_opaque_tx_put(struct io_opaque_req *op);
static bool io_opaque_budget_ready(struct io_opaque_req *op, unsigned long now)
{
@@ -253,6 +257,37 @@ void io_opaque_release_transient(struct io_opaque_store *store, u64 uncharge,
io_opaque_wake_budget(store);
}
+/* Framing uses ordinary capacity first and the reply reserve only when that is full. */
+bool io_opaque_charge_frame(struct io_opaque_store *store, u64 bytes, bool *reserved)
+{
+ u64 limit = store->config.reply_reserve;
+ s64 used;
+
+ if (io_opaque_charge_transient(store, bytes, io_opaque_rx_limit(store))) {
+ *reserved = false;
+ } else {
+ used = atomic64_read(&store->reserve_framing);
+ for (;;) {
+ if (bytes > limit || used > limit - bytes)
+ return false;
+ if (atomic64_try_cmpxchg(&store->reserve_framing, &used, used + bytes))
+ break;
+ }
+ *reserved = true;
+ }
+ atomic64_add(bytes, &store->framing);
+ return true;
+}
+
+void io_opaque_uncharge_frame(struct io_opaque_store *store, u64 bytes, bool reserved)
+{
+ atomic64_sub(bytes, &store->framing);
+ if (reserved)
+ atomic64_sub(bytes, &store->reserve_framing);
+ else
+ io_opaque_release_transient(store, bytes, bytes);
+}
+
static bool io_opaque_extent_reserve(struct io_opaque_store *store)
{
int nr = atomic_read(&store->extents);
@@ -682,6 +717,21 @@ void io_opaque_stop(struct io_opaque_store *store)
io_opaque_wake_budget(store);
}
+/* Free every queue; no request may be left on one. */
+static void io_opaque_txs_free(struct io_opaque_store *store)
+{
+ struct io_opaque_tx *tx, *next;
+
+ guard(spinlock)(&store->tx_lock);
+ rbtree_postorder_for_each_entry_safe(tx, next, &store->txs, node) {
+ WARN_ON_ONCE(!list_empty(&tx->requests) || tx->sk_held);
+ kfree(tx);
+ }
+ store->txs = RB_ROOT;
+ INIT_LIST_HEAD(&store->txs_idle);
+ store->nr_txs_idle = 0;
+}
+
void io_opaque_free(struct io_opaque_store *store)
{
if (!store)
@@ -689,6 +739,7 @@ void io_opaque_free(struct io_opaque_store *store)
io_opaque_stop(store);
WARN_ON_ONCE(atomic64_read(&store->bytes));
WARN_ON_ONCE(atomic_read(&store->extents));
+ io_opaque_txs_free(store);
io_unaccount_mem(store->user, store->mm,
store->config.hard_limit >> PAGE_SHIFT);
obj_cgroup_put(store->objcg);
@@ -767,6 +818,20 @@ void io_opaque_queue_ready(struct io_opaque_req *op, int result)
io_req_task_work_add(op->req);
}
+static void io_opaque_resume(struct io_tw_req tw_req, io_tw_token_t tw)
+{
+ struct io_kiocb *req = tw_req.req;
+ struct io_opaque_req *op = req->async_data;
+
+ io_tw_lock(req->ctx, tw);
+ list_del_init(&op->cancel);
+ op->phase = IO_OPAQUE_IDLE;
+ if (tw.cancel || READ_ONCE(op->canceled) || READ_ONCE(op->store->dead))
+ io_req_defer_failed(req, -ECANCELED);
+ else
+ io_req_task_submit(tw_req, tw);
+}
+
static struct io_opaque_extent *io_opaque_extent_split(struct io_opaque_store *store,
struct io_opaque_extent *extent,
u64 at)
@@ -1179,10 +1244,12 @@ static void io_opaque_ready(struct io_tw_req tw_req, io_tw_token_t tw)
}
if (ret < 0)
req_set_fail(req);
+ if (ret < 0 && req->opcode == IORING_OP_OPAQUE_OBJ_SEND && op->progress)
+ ret = op->progress;
if (ret >= 0 && op->op == IORING_OPAQUE_RECV_OBJECT)
io_req_set_res32(req, ret, 0, op->handle, object_flags);
else
- io_req_set_res(req, ret, 0);
+ io_req_set_res(req, ret, op->consumed ? IORING_CQE_F_OPAQUE_CONSUMED : 0);
io_req_task_complete(tw_req, tw);
}
@@ -1562,8 +1629,19 @@ void io_opaque_cleanup(struct io_kiocb *req)
return;
store = op->store;
list_del_init(&op->cancel);
- scoped_guard(spinlock, &store->wait_lock)
- list_del_init(&op->wait);
+ /*
+ * Taking wait_lock is how cleanup waits out a waker that is still in
+ * its hold: budget_waits and copies are walked under it, using the
+ * request's file and stream after its entry has been unlinked. A send
+ * is on neither list. What reaches one from outside comes through its
+ * TX queue, which io_opaque_tx_put() below waits for, or from
+ * cancellation, which holds the ring lock as cleanup does. It has no
+ * waker to wait out, and takes the lock only if its entry is linked.
+ */
+ if (req->opcode != IORING_OP_OPAQUE_OBJ_SEND || !list_empty(&op->wait)) {
+ scoped_guard(spinlock, &store->wait_lock)
+ list_del_init(&op->wait);
+ }
/* Waiters have stopped accessing this request before cleanup. */
if (op->stream) {
mutex_lock(&op->stream->lock);
@@ -1591,9 +1669,858 @@ void io_opaque_cleanup(struct io_kiocb *req)
store->objects[op->slot].generation);
mutex_unlock(&store->tables);
}
+ if (req->opcode == IORING_OP_OPAQUE_OBJ_SEND) {
+ io_opaque_tx_put(op);
+ kfree(op->framing);
+ if (op->frame_charge)
+ io_opaque_uncharge_frame(store, op->frame_charge, op->frame_reserved);
+ }
io_opaque_data_put(store, op->data);
io_opaque_stream_put(op->stream);
lockdep_assert_held(&req->ctx->uring_lock);
io_cache_free(&req->ctx->opaque_cache, op);
io_req_async_data_clear(req, 0);
}
+
+void io_opaque_fail(struct io_kiocb *req)
+{
+ struct io_opaque_req *op = req->async_data;
+
+ if (op && req->opcode == IORING_OP_OPAQUE_OBJ_SEND) {
+ if (op->progress)
+ req->cqe.res = op->progress;
+ if (op->consumed)
+ req->cqe.flags |= IORING_CQE_F_OPAQUE_CONSUMED;
+ }
+}
+
+/* Copy and check the framing descriptor at prep; prefix and suffix are send data. */
+static int io_opaque_frame_prep(struct io_opaque_req *op)
+{
+ struct io_uring_opaque_frame frame = {};
+ u32 size, bytes;
+ int ret;
+
+ op->total_length = op->length;
+ if (op->addr) {
+ if (get_user(size, (u32 __user *)op->addr))
+ return -EFAULT;
+ if (size < offsetof(struct io_uring_opaque_frame, __resv) ||
+ size > PAGE_SIZE)
+ return -EINVAL;
+ ret = copy_struct_from_user(&frame, sizeof(frame), op->addr, size);
+ if (ret)
+ return ret;
+ if (frame.size != size || frame.flags ||
+ !mem_is_zero(frame.__resv, sizeof(frame.__resv)) ||
+ (!frame.prefix_length && frame.prefix) ||
+ (!frame.suffix_length && frame.suffix) ||
+ check_add_overflow(frame.prefix_length, frame.suffix_length, &bytes) ||
+ bytes > IORING_OPAQUE_FRAME_MAX ||
+ check_add_overflow(op->length, bytes, &op->total_length) ||
+ op->total_length > INT_MAX)
+ return -EINVAL;
+ op->prefix = frame.prefix;
+ op->suffix = frame.suffix;
+ op->prefix_length = frame.prefix_length;
+ op->suffix_length = frame.suffix_length;
+ }
+ if (!op->total_length ||
+ (!op->length && (op->handle || op->offset ||
+ (op->send_flags & IORING_OPAQUE_SEND_LAST))))
+ return -EINVAL;
+ return 0;
+}
+
+/*
+ * Snapshot framing before TX admission can consume a LAST handle. Only a send
+ * with a frame gets here: one without is resolved at prep.
+ */
+static int io_opaque_frame_import(struct io_opaque_req *op)
+{
+ u32 bytes = op->prefix_length + op->suffix_length, charge;
+ struct mem_cgroup *old;
+
+ if (bytes) {
+ charge = kmalloc_size_roundup(bytes);
+ if (!io_opaque_charge_frame(op->store, charge, &op->frame_reserved))
+ return -ENOBUFS;
+ op->frame_charge = charge;
+ old = set_active_memcg(op->store->memcg);
+ op->framing = kmalloc(charge, GFP_KERNEL_ACCOUNT);
+ set_active_memcg(old);
+ if (!op->framing)
+ return -ENOMEM;
+ if ((op->prefix_length &&
+ copy_from_user(op->framing, u64_to_user_ptr(op->prefix),
+ op->prefix_length)) ||
+ (op->suffix_length &&
+ copy_from_user(op->framing + op->prefix_length,
+ u64_to_user_ptr(op->suffix), op->suffix_length)))
+ return -EFAULT;
+ atomic64_add(bytes, &op->store->framing_copied);
+ }
+ op->resolved = true;
+ return 0;
+}
+
+/*
+ * A send without a frame reads nothing from user memory at first issue: its
+ * total is its body, which prep already checked to be positive.
+ */
+static void io_opaque_send_resolve(struct io_opaque_req *op)
+{
+ if (op->addr)
+ return;
+ op->total_length = op->length;
+ op->resolved = true;
+}
+
+/*
+ * Find the segment of @data that holds byte @off, and return how far into it
+ * @off is. @seg is a segment of the object that starts @pos bytes in and not
+ * after the one wanted: the walk goes on from there, and leaves @seg and @pos
+ * at the start of the segment it found.
+ *
+ * A segment within IO_OPAQUE_SEEK_STRIDE steps is walked to. A farther one
+ * is reached by jumping to the last anchor that is at or before @off, if that
+ * is ahead of where the walk is, and walking on from it: under a stride more.
+ * No seek takes more than two strides and a binary search over the anchors,
+ * wherever it starts, however large the object is.
+ */
+static u32 io_opaque_vec_seek(const struct io_opaque_data *data, u64 off, u32 *seg, u64 *pos)
+{
+ const struct bio_vec *src = data->bvec;
+ u32 i = *seg, steps = 0;
+ u64 at = *pos;
+
+ while (steps < IO_OPAQUE_SEEK_STRIDE && i < data->nr && at + src[i].bv_len <= off) {
+ at += src[i++].bv_len;
+ steps++;
+ }
+ if (data->anchor && i < data->nr && at + src[i].bv_len <= off) {
+ u32 lo = 0, hi = DIV_ROUND_UP(data->nr, IO_OPAQUE_SEEK_STRIDE) - 1;
+
+ /* The last anchor that starts at or before @off; the first always does. */
+ while (lo < hi) {
+ u32 mid = lo + (hi - lo + 1) / 2;
+
+ if (data->anchor[mid] <= off)
+ lo = mid;
+ else
+ hi = mid - 1;
+ }
+ if (data->anchor[lo] > at) {
+ i = lo * IO_OPAQUE_SEEK_STRIDE;
+ at = data->anchor[lo];
+ }
+ }
+ while (i < data->nr && at + src[i].bv_len <= off)
+ at += src[i++].bv_len;
+ *seg = i;
+ *pos = at;
+ return off - at;
+}
+
+/* The handle must hold a version, and the range must lie within it. */
+static int io_opaque_send_range(const struct io_opaque_slot *slot, u64 off, u32 len)
+{
+ if (!slot || !slot->data)
+ return -ESTALE;
+ if (off > slot->data->length || len > slot->data->length - off)
+ return -ERANGE;
+ return 0;
+}
+
+/*
+ * Take a send's reference on the version @slot holds: the table reference when
+ * the send is LAST, which also removes the handle, else an extra one. Under
+ * tables, after io_opaque_send_range().
+ */
+static struct io_opaque_data *io_opaque_send_pin(struct io_opaque_store *store,
+ struct io_opaque_slot *slot, bool last)
+{
+ if (last)
+ return io_opaque_detach(store, slot);
+ refcount_inc(&slot->data->refs);
+ return slot->data;
+}
+
+/* Drop the references a send took at admission; a no-op when it took none. */
+static void io_opaque_send_unpin(struct io_opaque_req *op)
+{
+ io_opaque_data_put(op->store, op->data);
+ op->data = NULL;
+}
+
+/*
+ * The part of io_opaque_send_unpin() that a socket lock can wait for: drop
+ * every reference of the send that is not the last on its version, which is a
+ * decrement and cannot free anything, and say whether any is left. What is
+ * left is for io_opaque_send_unpin() once the socket is released, since the
+ * last reference frees the version, every extent of it, and that is not for
+ * the lock to wait out. Nobody can take a reference on a version whose count
+ * is one, which is not in the table, so the last one is still the last then.
+ */
+static bool io_opaque_send_unpin_cheap(struct io_opaque_req *op)
+{
+ bool left = false;
+
+ if (op->data) {
+ if (refcount_dec_not_one(&op->data->refs))
+ op->data = NULL;
+ else
+ left = true;
+ }
+ return left;
+}
+
+/*
+ * Point the send's cursor at the range it selects: @len bytes in @nr segments
+ * from @bv, of which the range starts @skip bytes in and lasts for op->length.
+ */
+static void io_opaque_send_iter(struct io_opaque_req *op, const struct bio_vec *bv,
+ unsigned int nr, size_t len, size_t skip)
+{
+ iov_iter_bvec(&op->iter, ITER_SOURCE, bv, nr, len);
+ iov_iter_advance(&op->iter, skip);
+ iov_iter_truncate(&op->iter, op->length);
+}
+
+/* Resolve what a send names, take its references and set its cursor, under tables. */
+static int io_opaque_send_acquire(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ bool last = op->send_flags & IORING_OPAQUE_SEND_LAST;
+ struct io_opaque_slot *slot = io_opaque_lookup(store, op->handle);
+ u64 pos = 0;
+ u32 seg = 0, skip;
+ int ret;
+
+ if (store->dead)
+ return -ESHUTDOWN;
+ if (!op->length)
+ return 0;
+ ret = io_opaque_send_range(slot, op->offset, op->length);
+ if (ret)
+ return ret;
+ op->data = io_opaque_send_pin(store, slot, last);
+ if (last)
+ op->consumed = true;
+ /* The cursor starts at the segment that holds the range. */
+ skip = io_opaque_vec_seek(op->data, op->offset, &seg, &pos);
+ io_opaque_send_iter(op, op->data->bvec + seg, op->data->nr - seg, op->data->length - pos,
+ skip);
+ return 0;
+}
+
+int io_opaque_send_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
+{
+ struct io_opaque_store *store = io_opaque_get_store(req, READ_ONCE(sqe->opaque_store_id));
+ struct io_opaque_req *op;
+ u16 send_flags = READ_ONCE(sqe->ioprio);
+ u32 msg_flags = READ_ONCE(sqe->msg_flags);
+ u64 end;
+ int ret;
+
+ if (!store)
+ return -ENXIO;
+ if ((send_flags & ~IORING_OPAQUE_SEND_LAST) ||
+ ((send_flags & IORING_OPAQUE_SEND_LAST) && (req->flags & REQ_F_CQE_SKIP)) ||
+ READ_ONCE(sqe->buf_index) || READ_ONCE(sqe->__pad2[0]) ||
+ (msg_flags & ~(MSG_MORE | MSG_DONTWAIT)))
+ return -EINVAL;
+ op = io_opaque_req_alloc(req, store);
+ if (IS_ERR(op))
+ return PTR_ERR(op);
+ op->op = U16_MAX;
+ INIT_LIST_HEAD(&op->send);
+ op->send_flags = send_flags;
+ op->handle = READ_ONCE(sqe->addr);
+ op->offset = READ_ONCE(sqe->off);
+ op->length = READ_ONCE(sqe->len);
+ op->addr = u64_to_user_ptr(READ_ONCE(sqe->addr3));
+ op->msg_flags = msg_flags;
+ if (op->msg_flags & MSG_DONTWAIT)
+ req->flags |= REQ_F_NOWAIT;
+ if ((!op->length && !op->addr) || op->length > INT_MAX ||
+ check_add_overflow(op->offset, (u64)op->length, &end))
+ return -EINVAL;
+ ret = io_opaque_frame_prep(op);
+ if (ret)
+ return ret;
+ io_opaque_send_resolve(op);
+ return 0;
+}
+
+static struct io_opaque_req *io_opaque_tx_head(struct io_opaque_tx *tx)
+{
+ return list_first_entry_or_null(&tx->requests, struct io_opaque_req, send);
+}
+
+/* Set under tx_lock, so that a release cannot pass a queued request by. */
+static void io_opaque_tx_park(struct io_opaque_req *op)
+{
+ guard(spinlock)(&op->store->wait_lock);
+ op->phase = IO_OPAQUE_SEND_WAIT;
+}
+
+static struct io_opaque_tx *io_opaque_tx_find(struct io_opaque_store *store,
+ struct sock *sk,
+ struct rb_node ***linkp,
+ struct rb_node **parentp)
+{
+ struct rb_node **link = &store->txs.rb_node, *parent = NULL;
+ struct io_opaque_tx *tx = NULL;
+
+ lockdep_assert_held(&store->tx_lock);
+ while (*link) {
+ parent = *link;
+ tx = rb_entry(parent, struct io_opaque_tx, node);
+ if ((unsigned long)sk < (unsigned long)tx->sk)
+ link = &parent->rb_left;
+ else if ((unsigned long)sk > (unsigned long)tx->sk)
+ link = &parent->rb_right;
+ else
+ break;
+ }
+ /* On a miss these are the insertion point. */
+ *linkp = link;
+ *parentp = parent;
+ return *link ? tx : NULL;
+}
+
+/* A queue nobody uses stays for the next frame on its socket, up to a cap. */
+static struct sock *io_opaque_tx_unused(struct io_opaque_store *store, struct io_opaque_tx *tx)
+{
+ struct sock *sk;
+
+ lockdep_assert_held(&store->tx_lock);
+ if (!list_empty(&tx->requests))
+ return NULL;
+ /* Transfer the busy identity reference; the caller drops it after tx_lock. */
+ WARN_ON_ONCE(!tx->sk_held);
+ sk = tx->sk;
+ tx->sk_held = false;
+ /* Reuse a failed queue only after its followers and admissions drain. */
+ tx->failed = false;
+ list_add_tail(&tx->idle, &store->txs_idle);
+ if (++store->nr_txs_idle <= IO_OPAQUE_TX_IDLE_MAX)
+ return sk;
+ tx = list_first_entry(&store->txs_idle, struct io_opaque_tx, idle);
+ list_del(&tx->idle);
+ store->nr_txs_idle--;
+ rb_erase(&tx->node, &store->txs);
+ kfree(tx);
+ return sk;
+}
+
+/* A queue that is about to hold a request is not idle. */
+static void io_opaque_tx_busy(struct io_opaque_store *store, struct io_opaque_tx *tx)
+{
+ lockdep_assert_held(&store->tx_lock);
+ if (list_empty(&tx->requests)) {
+ WARN_ON_ONCE(tx->sk_held);
+ sock_hold(tx->sk);
+ tx->sk_held = true;
+ }
+ if (!list_empty(&tx->idle)) {
+ list_del_init(&tx->idle);
+ store->nr_txs_idle--;
+ }
+}
+
+static struct io_opaque_tx *io_opaque_tx_alloc(struct io_opaque_store *store,
+ gfp_t gfp)
+{
+ return kzalloc_obj(struct io_opaque_tx, gfp);
+}
+
+/* Resolve what a send names and take its references, under tables. */
+static int io_opaque_tx_acquire(struct io_opaque_req *op)
+{
+ int ret;
+
+ scoped_guard(mutex, &op->store->tables)
+ ret = io_opaque_send_acquire(op);
+ return ret;
+}
+
+/*
+ * The admission of a request that found frames ahead of it. It takes its
+ * references before it waits, so that a LAST send parked behind another has
+ * consumed its object. Its FIFO position is already reserved. Until this
+ * second tx_lock hold it stays IDLE: a release must not wake an admission
+ * that is still issuing. If it became head meanwhile, it continues here.
+ * A queue that fails in between cancels it. It holds tables alone, not the
+ * socket. Returns as io_opaque_tx_join() does.
+ */
+static int io_opaque_tx_settle(struct io_opaque_req *op, struct io_opaque_tx *tx)
+{
+ struct io_opaque_store *store = op->store;
+ int ret = io_opaque_tx_acquire(op);
+
+ spin_lock(&store->tx_lock);
+ if (ret || tx->failed) {
+ bool pinned = !ret;
+
+ /* Normal cleanup releases the pending position and wakes its successor. */
+ spin_unlock(&store->tx_lock);
+ if (!pinned)
+ return ret;
+ /* The head failed after the check: the pin goes, LAST stays consumed. */
+ io_opaque_send_unpin(op);
+ return -ECANCELED;
+ }
+ ret = io_opaque_tx_head(tx) == op;
+ if (!ret)
+ io_opaque_tx_park(op);
+ spin_unlock(&store->tx_lock);
+ return ret;
+}
+
+/*
+ * Queue @op behind the frames already issuing for @sk. Returns
+ * IO_OPAQUE_TX_WAIT when it waits for one (its phase is set before tx_lock is
+ * dropped, so a release cannot miss it), IO_OPAQUE_TX_HEAD when it is the head
+ * and holds its references, IO_OPAQUE_TX_SETTLE when it is the head and has
+ * not taken them yet, or a negative errno.
+ *
+ * Every new request links itself in the first tx_lock hold, before acquiring
+ * references. A request that finds its queue empty takes them later, from
+ * io_opaque_tx_settle_head(), with the socket owned. Nothing can join the
+ * queue ahead of it, and a head cannot fail the queue, so nothing that
+ * settling does is undone by the queue. A request that finds frames ahead
+ * takes its references now, and may find itself the head when it is done
+ * (io_opaque_tx_settle()). A queue that has failed refuses it.
+ *
+ * @op->req is not used; the caller enrolls a waiting request for cancellation.
+ */
+static int io_opaque_tx_join(struct io_opaque_req *op, struct sock *sk)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_tx *tx;
+ struct rb_node **link, *parent;
+ int ret = 0;
+
+ spin_lock(&store->tx_lock);
+ tx = op->tx;
+ if (tx) {
+ /* A resumed or retried request keeps its place. */
+ ret = io_opaque_tx_head(tx) == op;
+ if (!ret)
+ io_opaque_tx_park(op);
+ spin_unlock(&store->tx_lock);
+ return ret;
+ }
+ tx = io_opaque_tx_find(store, sk, &link, &parent);
+ if (!tx) {
+ /* First issue is authoritative on a miss too: never drop tx_lock here. */
+ tx = io_opaque_tx_alloc(store, GFP_NOWAIT | __GFP_ACCOUNT);
+ if (!tx) {
+ spin_unlock(&store->tx_lock);
+ return -ENOMEM;
+ }
+ tx->sk = sk;
+ INIT_LIST_HEAD(&tx->requests);
+ INIT_LIST_HEAD(&tx->idle);
+ rb_link_node(&tx->node, parent, link);
+ rb_insert_color(&tx->node, &store->txs);
+ }
+ if (tx->failed) {
+ ret = -ECANCELED;
+ } else {
+ io_opaque_tx_busy(store, tx);
+ ret = list_empty(&tx->requests) ? IO_OPAQUE_TX_SETTLE : IO_OPAQUE_TX_WAIT;
+ op->tx = tx;
+ list_add_tail(&op->send, &tx->requests);
+ }
+ spin_unlock(&store->tx_lock);
+ if (ret)
+ return ret;
+ return io_opaque_tx_settle(op, tx);
+}
+
+/*
+ * Take the references of a head that io_opaque_tx_join() left to settle, with
+ * the socket owned where it can be. Nothing is taken and nothing removed when
+ * this fails, and the request is still the head: it leaves the queue through
+ * io_opaque_tx_put(), which hands the queue to whoever waits behind it and does
+ * not fail it, since nothing of this frame was sent.
+ */
+static int io_opaque_tx_settle_head(struct io_opaque_req *op)
+{
+ return io_opaque_tx_acquire(op);
+}
+
+/*
+ * Cancel and wake the frames queued behind @op, a head that queued part of its
+ * frame and ended short. @op has left @tx, which has failed already, and goes
+ * back in for the walk. At most IO_OPAQUE_WAKE_BATCH followers are looked at
+ * in a hold of tx_lock, and they are woken after it: how many there are is up
+ * to the process, and every socket and ring of the store takes tx_lock.
+ *
+ * The followers stay in the queue and leave it as they always did, each when
+ * its request is cleaned up, so the queue stays failed until the last has
+ * gone. @op is where the walk has got to. It moves behind each batch and
+ * leaves with the last, and nobody else moves it. It is not parked, and a
+ * release hands the queue only to a parked follower, so a canceled follower
+ * that is released as the first of the queue cannot promote one that is not
+ * canceled yet.
+ *
+ * A follower is touched in a hold of tx_lock, where it cannot be cleaned up,
+ * for its cleanup leaves the queue first. After the hold the walk touches only
+ * those it made queued: nobody else wakes them, and a request ends only once
+ * it is woken. A follower that was queued before is a cancellation's to wake.
+ * It is marked here, and left alone.
+ */
+static noinline_for_stack struct sock *io_opaque_tx_fail(struct io_opaque_store *store,
+ struct io_opaque_tx *tx,
+ struct io_opaque_req *op)
+{
+ struct io_opaque_req *woken[IO_OPAQUE_WAKE_BATCH];
+
+ lockdep_assert_held(&store->tx_lock);
+ op->tx = tx;
+ list_add(&op->send, &tx->requests);
+ while (op->send.next != &tx->requests) {
+ struct list_head *pos = op->send.next;
+ unsigned int seen, nr = 0, i;
+
+ for (seen = 0; seen < IO_OPAQUE_WAKE_BATCH && pos != &tx->requests; seen++) {
+ struct io_opaque_req *next = list_entry(pos, struct io_opaque_req, send);
+ bool queue;
+
+ scoped_guard(spinlock, &store->wait_lock) {
+ WRITE_ONCE(next->canceled, true);
+ queue = next->phase == IO_OPAQUE_SEND_WAIT;
+ if (queue)
+ next->phase = IO_OPAQUE_QUEUED;
+ }
+ if (queue)
+ woken[nr++] = next;
+ pos = pos->next;
+ }
+ /* Behind the ones looked at: what follows is what is left to do. */
+ list_move(&op->send, pos->prev);
+ spin_unlock(&store->tx_lock);
+ for (i = 0; i < nr; i++)
+ io_opaque_queue_ready(woken[i], -ECANCELED);
+ spin_lock(&store->tx_lock);
+ }
+ list_del_init(&op->send);
+ op->tx = NULL;
+ /* The followers may all have gone while the walk was out of the lock. */
+ return io_opaque_tx_unused(store, tx);
+}
+
+/*
+ * Remove @op from its queue. A head that queued part of its frame and ended
+ * short fails the queue: every follower is canceled and woken
+ * (io_opaque_tx_fail()), and nothing is returned. Publish failure even when
+ * only in-progress admissions remain. Otherwise the follower that
+ * becomes head is returned with its phase already queued, for the caller to
+ * wake. Only the failure wakeups reach an io_kiocb.
+ */
+static struct io_opaque_req *io_opaque_tx_release(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_tx *tx = op->tx;
+ struct io_opaque_req *next = NULL;
+ struct sock *put = NULL;
+ bool head;
+
+ spin_lock(&store->tx_lock);
+ head = io_opaque_tx_head(tx) == op;
+ list_del_init(&op->send);
+ op->tx = NULL;
+ /* Only a send that queued part of its frame breaks the following frames. */
+ if (head && op->progress && op->progress != op->total_length) {
+ tx->failed = true;
+ put = io_opaque_tx_fail(store, tx, op);
+ goto unlock;
+ }
+ if (list_empty(&tx->requests)) {
+ put = io_opaque_tx_unused(store, tx);
+ goto unlock;
+ }
+ if (!head || tx->failed)
+ goto unlock;
+ scoped_guard(spinlock, &store->wait_lock) {
+ struct io_opaque_req *candidate = io_opaque_tx_head(tx);
+
+ if (candidate->phase == IO_OPAQUE_SEND_WAIT) {
+ candidate->phase = IO_OPAQUE_QUEUED;
+ next = candidate;
+ }
+ }
+unlock:
+ spin_unlock(&store->tx_lock);
+ if (put)
+ sock_put(put);
+ return next;
+}
+
+static void io_opaque_tx_put(struct io_opaque_req *op)
+{
+ struct io_opaque_req *next;
+
+ if (!op->tx)
+ return;
+ next = io_opaque_tx_release(op);
+ if (next) {
+ next->req->io_task_work.func = io_opaque_resume;
+ io_req_task_work_add(next->req);
+ }
+}
+
+/*
+ * Whether a send may call tcp_sendmsg_locked() on @sk where sock_sendmsg()
+ * would end up in tcp_sendmsg(): plain TCP, whose sendmsg no ULP (kTLS,
+ * espintcp) or socket map has replaced, which they do with a copy of tcp_prot
+ * that sk_prot_creator does not follow. A socket with no local port has not
+ * connected. TCP's proto has inet_send_prepare() leave it unbound, so this
+ * does not change what a send does, and only keeps such a socket on the
+ * ordinary path.
+ *
+ * kTLS and espintcp install their proto under the socket lock, and a socket
+ * map under the bottom-half lock alone, so the answer is a hint until the
+ * socket is owned and is no more than true then. A swap that comes while
+ * tcp_sendmsg_locked() is in progress is what it is for a stock tcp_sendmsg()
+ * in flight: that call goes on with the proto it began with, and the next one
+ * takes the new, here the next part of the frame, which asks again.
+ */
+static bool io_opaque_send_direct(const struct sock *sk)
+{
+ const struct proto *prot;
+ u16 port;
+
+ prot = READ_ONCE(sk->sk_prot);
+ /* Written when the socket binds or connects; a stale zero only spoils a hint. */
+ port = data_race(inet_sk(sk)->inet_num);
+ return prot->sendmsg == tcp_sendmsg && prot == sk->sk_prot_creator && port;
+}
+
+static u32 io_opaque_send_msg_flags(const struct io_opaque_req *op, unsigned int issue_flags)
+{
+ u32 flags = MSG_NOSIGNAL | op->msg_flags;
+
+ if (issue_flags & IO_URING_F_NONBLOCK)
+ flags |= MSG_DONTWAIT;
+ return flags;
+}
+
+/*
+ * The security hook, for an issue that sends on the socket directly. The
+ * first sock_sendmsg() of an issue asks it for the part of the frame it sends,
+ * and a refusal ends that call before any byte goes: a send that has taken its
+ * references, so that LAST is consumed and a frame behind it fails with it if
+ * part of this one was sent, or is told to try again if the hook says EAGAIN.
+ * Here it is asked once, for what is left of the frame. The caller keeps the
+ * verdict and applies it where the first sock_sendmsg() would have met it, so
+ * that the issue ends in the same way. The ordinary path asks for every part,
+ * which a policy that depends on the socket and not on the part cannot tell
+ * apart.
+ *
+ * It is asked with the socket not owned, because a hook may take the socket
+ * lock itself: AppArmor's finds the address with getname(), which does, and
+ * would wait for the owner of the socket for ever.
+ */
+static int io_opaque_send_security(struct socket *sock, const struct io_opaque_req *op,
+ unsigned int issue_flags)
+{
+ struct msghdr msg = { .msg_flags = io_opaque_send_msg_flags(op, issue_flags) };
+
+ lockdep_assert_not_held(&sock->sk->sk_lock);
+ return security_socket_sendmsg(sock, &msg, op->total_length - op->progress);
+}
+
+/*
+ * One issue of an object send. A request that is not the head of its
+ * destination's queue parks. The head sends its frame, prefix, body and
+ * suffix, from where it stopped, and the frame leaves the queue once all of it
+ * is queued to TCP.
+ *
+ * On a plain TCP socket (io_opaque_send_direct()) the head asks the security
+ * hook, and then does all of the rest under one lock_sock(): it takes the
+ * references it has not taken yet, applies what the hook said, calls
+ * tcp_sendmsg_locked() for each part of the frame, hands the queue on, lets go
+ * of the references that are not the last and only then releases the socket,
+ * and after it puts the last ones. While the socket is
+ * unowned, ACKs and the transmit that autocorking left queued run under its
+ * spinlock, for as long as they take, and the sender's next lock_sock() waits
+ * for them; the less the sender does in that window, the less often it does.
+ * Any other socket takes its references first and goes through sock_sendmsg()
+ * for each part, with the ring lock dropped around it.
+ *
+ * Under IO_URING_F_UNLOCKED the ring lock is dropped before the socket is
+ * taken, and an issue that owned the socket returns without taking it again,
+ * as io-wq expects; an inline issue holds it throughout, as it holds it across
+ * sock_sendmsg().
+ */
+int io_opaque_send(struct io_kiocb *req, unsigned int issue_flags)
+{
+ struct io_opaque_req *op = req->async_data;
+ struct socket *sock = sock_from_file(req->file);
+ struct sock *sk = NULL;
+ struct msghdr msg;
+ struct kvec vec;
+ u32 body_end;
+ bool body = false, fresh, left, owned = false;
+ int ret, sec = 0;
+
+ io_ring_submit_lock(req->ctx, issue_flags);
+ /* A worker cancellation may have searched opaque_waits before us. */
+ if (io_wq_cancel_seen(req, issue_flags)) {
+ ret = -ECANCELED;
+ goto finish;
+ }
+ if (!sock || sock->type != SOCK_STREAM || sock->sk->sk_protocol != IPPROTO_TCP) {
+ ret = -EOPNOTSUPP;
+ goto finish;
+ }
+ if (READ_ONCE(op->store->dead)) {
+ ret = -ESHUTDOWN;
+ goto finish;
+ }
+ if (!op->resolved) {
+ ret = io_opaque_frame_import(op);
+ if (ret)
+ goto finish;
+ }
+ sk = sock->sk;
+ ret = io_opaque_tx_join(op, sk);
+ if (ret < 0)
+ goto out;
+ if (ret == IO_OPAQUE_TX_WAIT) {
+ /* The ring lock keeps the wakeup from running before the request is listed. */
+ io_opaque_enroll(op);
+ ret = IOU_ISSUE_SKIP_COMPLETE;
+ goto out;
+ }
+ fresh = ret == IO_OPAQUE_TX_SETTLE;
+
+ if (io_opaque_send_direct(sk)) {
+ /*
+ * The hint says plain TCP; the socket decides once it is owned.
+ * The security hook is asked now, with it not owned, and what it
+ * says is applied below, after the references are taken.
+ */
+ io_ring_submit_unlock(req->ctx, issue_flags);
+ sec = io_opaque_send_security(sock, op, issue_flags);
+ sock_rps_record_flow(sk);
+ lock_sock(sk);
+ owned = io_opaque_send_direct(sk);
+ if (!owned) {
+ release_sock(sk);
+ io_ring_submit_lock(req->ctx, issue_flags);
+ }
+ }
+ if (fresh) {
+ ret = io_opaque_tx_settle_head(op);
+ if (ret) {
+ /* Nothing was taken or sent: the frame behind this one goes now. */
+ io_opaque_tx_put(op);
+ goto leave;
+ }
+ }
+ if (owned && sec) {
+ /* As the first sock_sendmsg() would have ended with it. */
+ ret = sec;
+ goto sent;
+ }
+
+again:
+ memset(&msg, 0, sizeof(msg));
+ msg.msg_flags = io_opaque_send_msg_flags(op, issue_flags);
+ body_end = op->prefix_length + op->length;
+ body = op->length && op->progress >= op->prefix_length &&
+ op->progress < body_end;
+ if (!op->length) {
+ vec.iov_base = op->framing + op->progress;
+ vec.iov_len = op->total_length - op->progress;
+ } else if (op->progress < op->prefix_length) {
+ vec.iov_base = op->framing + op->progress;
+ vec.iov_len = op->prefix_length - op->progress;
+ msg.msg_flags |= MSG_MORE;
+ } else if (body) {
+ msg.msg_iter = op->iter;
+ msg.msg_flags |= MSG_SPLICE_PAGES;
+ if (op->suffix_length)
+ msg.msg_flags |= MSG_MORE;
+ } else {
+ vec.iov_base = op->framing + op->prefix_length + op->progress - body_end;
+ vec.iov_len = op->total_length - op->progress;
+ }
+ if (!body)
+ iov_iter_kvec(&msg.msg_iter, ITER_SOURCE, &vec, 1, vec.iov_len);
+ if (owned) {
+ ret = tcp_sendmsg_locked(sk, &msg, iov_iter_count(&msg.msg_iter));
+ if (trace_sock_send_length_enabled())
+ trace_sock_send_length(sk, ret, 0);
+ } else {
+ /* io-wq may block in TCP; other requests must retain access to the ring. */
+ io_ring_submit_unlock(req->ctx, issue_flags);
+ ret = sock_sendmsg(sock, &msg);
+ io_ring_submit_lock(req->ctx, issue_flags);
+ }
+sent:
+ if (ret > 0) {
+ if (body)
+ op->iter = msg.msg_iter;
+ op->progress += ret;
+ if (op->progress < op->total_length && !iov_iter_count(&msg.msg_iter)) {
+ /*
+ * TCP lets go of the socket while it waits for room, and a
+ * protocol may have taken it over by then: ask again.
+ */
+ if (owned && !io_opaque_send_direct(sk)) {
+ release_sock(sk);
+ io_ring_submit_lock(req->ctx, issue_flags);
+ owned = false;
+ }
+ goto again;
+ }
+ if (op->progress < op->total_length && !(req->flags & REQ_F_NOWAIT)) {
+ ret = -EAGAIN;
+ goto leave;
+ }
+ } else if (ret == -EAGAIN && (issue_flags & IO_URING_F_NONBLOCK) &&
+ !(req->flags & REQ_F_NOWAIT)) {
+ goto leave;
+ }
+finish:
+ if (ret == -ERESTARTSYS)
+ ret = -EINTR;
+ if (ret < 0 || (op->progress && op->progress < op->total_length))
+ req_set_fail(req);
+ if (op->progress)
+ ret = op->progress;
+ /* The frame is queued or abandoned: the next send goes now, not at cleanup. */
+ io_opaque_tx_put(op);
+ /* The references go with the socket, the last of a version after it. */
+ left = owned && io_opaque_send_unpin_cheap(op);
+ io_req_set_res(req, ret, op->consumed ? IORING_CQE_F_OPAQUE_CONSUMED : 0);
+ if (owned) {
+ /*
+ * Nothing is left for the ring lock, which an io-wq issue dropped
+ * for the socket and an inline one still holds: taking it only to
+ * give it up would be a round trip for the caller.
+ */
+ release_sock(sk);
+ if (left)
+ io_opaque_send_unpin(op);
+ return IOU_COMPLETE;
+ }
+ ret = IOU_COMPLETE;
+ goto out;
+leave:
+ /* The request keeps whatever it still holds, and the ring lock is as the caller left it. */
+ if (owned) {
+ release_sock(sk);
+ return ret;
+ }
+out:
+ io_ring_submit_unlock(req->ctx, issue_flags);
+ return ret;
+}
diff --git a/io_uring/opdef.c b/io_uring/opdef.c
index 256c7c533118..a80a56b19581 100644
--- a/io_uring/opdef.c
+++ b/io_uring/opdef.c
@@ -601,6 +601,20 @@ const struct io_issue_def io_issue_defs[] = {
.issue = io_opaque_issue,
#else
.prep = io_eopnotsupp_prep,
+#endif
+ },
+ [IORING_OP_OPAQUE_OBJ_SEND] = {
+ .audit_skip = 1,
+ .ioprio = 1,
+ .needs_file = 1,
+ .unbound_nonreg_file = 1,
+ .pollout = 1,
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+ .async_size = sizeof(struct io_opaque_req),
+ .prep = io_opaque_send_prep,
+ .issue = io_opaque_send,
+#else
+ .prep = io_eopnotsupp_prep,
#endif
},
};
@@ -865,6 +879,13 @@ const struct io_cold_def io_cold_defs[] = {
.name = "OPAQUE_OBJ",
.cleanup = io_opaque_cleanup,
},
+ [IORING_OP_OPAQUE_OBJ_SEND] = {
+ .name = "OPAQUE_OBJ_SEND",
+ .cleanup = io_opaque_cleanup,
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+ .fail = io_opaque_fail,
+#endif
+ },
};
const char *io_uring_get_opcode(u8 opcode)
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 08/12] io_uring/opaque: add vectored IORING_OP_OPAQUE_OBJ_SEND
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (6 preceding siblings ...)
2026-11-10 13:20 ` [RFC PATCH net-next 07/12] io_uring/opaque: add IORING_OP_OPAQUE_OBJ_SEND Alex Markuze
@ 2026-11-10 13:20 ` Alex Markuze
2026-11-10 13:20 ` [RFC PATCH net-next 09/12] io_uring/opaque: add IORING_OP_OPAQUE_COLLECT Alex Markuze
` (3 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:20 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Allow one send request to concatenate several object ranges, amortizing
request, queue and CQE work across small objects.
With SEND_VECTOR in sqe->ioprio, addr points to len entries and off must
be zero. A vector contains 1 to 1024 io_uring_opaque_send_entry records,
each with a handle, offset, nonzero length and zero reserved word. Handles
and ranges may repeat. The optional frame wraps the concatenated body;
prefix, body and suffix together must fit in INT_MAX bytes.
Preparation copies the entry array when the kernel consumes this SQE.
Keep the array valid until that consumption, including across short or
failed submissions. The resulting request has one queue entry, one
cursor and one completion; retry and ordering follow single-range sends.
Admission resolves and validates every range under the table mutex before
taking any reference. Indexed segment lookup uses short walks and binary
search; ranges together may span at most 4096 backing segments. A second
pass acquires references and, for LAST, removes each distinct handle once.
Acquisition is all-or-none; the queued-byte result may still be short.
Repeated entries retain the references needed to use the same immutable
version.
One request-owned block holds entries, pinned versions, range starts and
cursor segments. If it must grow, admission drops the mutex and resolves
the handles again after allocation. Cursor construction uses only the
versions already pinned by the request.
Ordinary-sized blocks remain with cached requests for reuse. Cleanup
frees blocks exceeding 256 entries or segments, and frees every block
under KASAN. The VECTOR flag identifies the operation independently of
whether a recycled request still carries a block. As with single-range
sends, final version release occurs after the socket is unlocked.
Vector admission uses the same already-reserved TX FIFO position as scalar
admission. Reference/cursor construction may finish after a head fails;
cleanup retains the linked failure barrier and socket identity until the
position drains. The vector caller never overtakes an earlier admission.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
include/uapi/linux/io_uring/opaque.h | 12 +
io_uring/opaque.c | 405 ++++++++++++++++++++++++++-
2 files changed, 407 insertions(+), 10 deletions(-)
diff --git a/include/uapi/linux/io_uring/opaque.h b/include/uapi/linux/io_uring/opaque.h
index 2a862d76a40e..c04eaed8ce5f 100644
--- a/include/uapi/linux/io_uring/opaque.h
+++ b/include/uapi/linux/io_uring/opaque.h
@@ -61,6 +61,18 @@ enum io_uring_opaque_op {
/* IORING_OP_OPAQUE_OBJ_SEND: sqe->ioprio contains these flags. */
#define IORING_OPAQUE_SEND_LAST (1U << 0)
+/* addr points at len struct io_uring_opaque_send_entry, off is zero: one send, many ranges. */
+#define IORING_OPAQUE_SEND_VECTOR (1U << 1)
+/* Entries of one vectored send, as many as SEND_ZC takes iovecs (UIO_MAXIOV). */
+#define IORING_OPAQUE_SEND_VEC_MAX 1024
+
+/* One range of a vectored send, named as addr, off and len name it in a single one. */
+struct io_uring_opaque_send_entry {
+ __u64 handle;
+ __u64 off;
+ __u32 len;
+ __u32 __resv;
+};
/* Optional SEND sqe->addr3 framing; size includes zero reserved extensions. */
struct io_uring_opaque_frame {
diff --git a/io_uring/opaque.c b/io_uring/opaque.c
index b62c4be92fea..0c58900073f7 100644
--- a/io_uring/opaque.c
+++ b/io_uring/opaque.c
@@ -29,13 +29,23 @@ bool io_opaque_cache_init(struct io_ring_ctx *ctx)
sizeof(struct io_opaque_req), sizeof(struct io_opaque_req));
}
+/* Frees a request left in the cache, with the vector block it may carry. */
+static void io_opaque_req_destroy(const void *entry)
+{
+ const struct io_opaque_req *op = entry;
+
+ kvfree(op->vec);
+ kfree(entry);
+}
+
void io_opaque_cache_free(struct io_ring_ctx *ctx)
{
- io_alloc_cache_free(&ctx->opaque_cache, kfree);
+ io_alloc_cache_free(&ctx->opaque_cache, io_opaque_req_destroy);
}
static void io_opaque_resume(struct io_tw_req tw_req, io_tw_token_t tw);
static void io_opaque_ready(struct io_tw_req tw_req, io_tw_token_t tw);
+static void io_opaque_vec_cleanup(struct io_opaque_req *op);
static void io_opaque_stream_close(struct io_opaque_stream *stream);
static void io_opaque_tx_put(struct io_opaque_req *op);
@@ -783,7 +793,8 @@ static struct io_opaque_req *io_opaque_req_alloc(struct io_kiocb *req,
op = io_uring_alloc_async_data(&req->ctx->opaque_cache, req);
if (!op)
return ERR_PTR(-ENOMEM);
- memset(op, 0, sizeof(*op));
+ /* Not the vector block, which a recycled request brings along. */
+ memset(op, 0, offsetof(struct io_opaque_req, vec));
op->op = U16_MAX;
op->req = req;
op->store = store;
@@ -794,6 +805,18 @@ static struct io_opaque_req *io_opaque_req_alloc(struct io_kiocb *req,
return op;
}
+/*
+ * A finished request goes back to its ring's cache, with the vector block it
+ * carries, or is freed with it when the cache is full.
+ */
+static void io_opaque_req_free(struct io_ring_ctx *ctx, struct io_opaque_req *op)
+{
+ if (!io_alloc_cache_put(&ctx->opaque_cache, op)) {
+ kvfree(op->vec);
+ kvfree(op);
+ }
+}
+
/* The ring lock protects enrollment; wait_lock arbitrates remote wakeups. */
static void io_opaque_enroll(struct io_opaque_req *op)
{
@@ -1675,10 +1698,12 @@ void io_opaque_cleanup(struct io_kiocb *req)
if (op->frame_charge)
io_opaque_uncharge_frame(store, op->frame_charge, op->frame_reserved);
}
+ if (unlikely(op->vec))
+ io_opaque_vec_cleanup(op);
io_opaque_data_put(store, op->data);
io_opaque_stream_put(op->stream);
lockdep_assert_held(&req->ctx->uring_lock);
- io_cache_free(&req->ctx->opaque_cache, op);
+ io_opaque_req_free(req->ctx, op);
io_req_async_data_clear(req, 0);
}
@@ -1776,17 +1801,154 @@ static void io_opaque_send_resolve(struct io_opaque_req *op)
op->resolved = true;
}
+/*
+ * Check @nr entries copied from user memory; return the body they add up to.
+ * A handle may repeat, and a range may too.
+ */
+static int io_opaque_vec_check(const struct io_uring_opaque_send_entry *ent, u32 nr,
+ u32 *total)
+{
+ u64 sum = 0, end;
+ u32 i;
+
+ BUILD_BUG_ON(sizeof(*ent) != 24);
+ if (!nr || nr > IORING_OPAQUE_SEND_VEC_MAX)
+ return -EINVAL;
+ for (i = 0; i < nr; i++) {
+ if (ent[i].__resv || !ent[i].len ||
+ check_add_overflow(ent[i].off, (u64)ent[i].len, &end))
+ return -EINVAL;
+ sum += ent[i].len;
+ if (sum > INT_MAX)
+ return -EINVAL;
+ }
+ *total = sum;
+ return 0;
+}
+
+/*
+ * Make room in the request's block for @nr entries and @segs segments, keeping
+ * its first @keep entries. A block that is big enough stays as it is. This
+ * allocates, so tables must not be held, and the send must hold no reference
+ * yet: what admission recorded about the versions is not carried over.
+ */
+static int io_opaque_vec_grow(struct io_opaque_req *op, u32 nr, u32 segs, u32 keep)
+{
+ struct io_opaque_vec *old = op->vec, *vec;
+
+ lockdep_assert_not_held(&op->store->tables);
+ if (old) {
+ WARN_ON_ONCE(old->pinned);
+ if (old->cap_nr >= nr && old->cap_segs >= segs)
+ return 0;
+ nr = max(nr, old->cap_nr);
+ segs = max(segs, old->cap_segs);
+ }
+ vec = kvmalloc(size_add(sizeof(*vec),
+ size_add(array_size(nr, sizeof(*vec->ent) + sizeof(*vec->pin) +
+ sizeof(*vec->start)),
+ array_size(segs, sizeof(*vec->bv)))),
+ GFP_KERNEL_ACCOUNT);
+ if (!vec)
+ return -ENOMEM;
+ vec->cap_nr = nr;
+ vec->cap_segs = segs;
+ vec->ent = (void *)(vec + 1);
+ vec->pin = (void *)(vec->ent + nr);
+ vec->start = (void *)(vec->pin + nr);
+ vec->bv = (void *)(vec->start + nr);
+ vec->pinned = 0;
+ if (old) {
+ vec->nr = old->nr;
+ memcpy(vec->ent, old->ent, array_size(keep, sizeof(*vec->ent)));
+ kvfree(old);
+ } else {
+ vec->nr = 0;
+ }
+ op->vec = vec;
+ return 0;
+}
+
+/*
+ * The @nr entries are in the block: validate them and make the request what
+ * its length and offset say about a send, a body of their total at no offset.
+ */
+static int io_opaque_vec_commit(struct io_opaque_req *op, u32 nr)
+{
+ struct io_opaque_vec *vec = op->vec;
+ u32 total;
+ int ret;
+
+ ret = io_opaque_vec_check(vec->ent, nr, &total);
+ if (ret)
+ return ret;
+ vec->nr = nr;
+ vec->pinned = 0;
+ op->handle = 0;
+ op->offset = 0;
+ op->length = total;
+ return 0;
+}
+
+/*
+ * Whether entry @i of an admitted vector holds a reference of its own. A LAST
+ * vector takes one per entry. Any other takes one per run of entries that name
+ * the same handle: within one hold of tables the same handle is the same
+ * version, so the rest of the run shares the first entry's pin. This looks at
+ * the entries, which never change, and not at the pins, which may be freed by
+ * the time the run's owner has been released.
+ */
+static bool io_opaque_vec_owns(const struct io_opaque_req *op, u32 i)
+{
+ const struct io_opaque_vec *vec = op->vec;
+
+ return (op->send_flags & IORING_OPAQUE_SEND_LAST) || !i ||
+ vec->ent[i].handle != vec->ent[i - 1].handle;
+}
+
+/* Drop every version reference the send holds; one that took none drops none. */
+static void io_opaque_vec_release(struct io_opaque_req *op)
+{
+ struct io_opaque_vec *vec = op->vec;
+ u32 i;
+
+ for (i = 0; i < vec->pinned; i++)
+ if (io_opaque_vec_owns(op, i))
+ io_opaque_data_put(op->store, vec->pin[i]);
+ vec->pinned = 0;
+}
+
+/*
+ * Request cleanup: release what a vectored send pinned and let go of the block
+ * unless the request takes it back to the cache. A big block is not worth
+ * keeping, and under KASAN an entry in the cache is cleared when it is taken,
+ * which would leak it.
+ */
+static void io_opaque_vec_cleanup(struct io_opaque_req *op)
+{
+ struct io_opaque_vec *vec = op->vec;
+
+ io_opaque_vec_release(op);
+ if (IS_ENABLED(CONFIG_KASAN) || vec->cap_nr > IO_VEC_CACHE_SOFT_CAP ||
+ vec->cap_segs > IO_VEC_CACHE_SOFT_CAP) {
+ kvfree(vec);
+ op->vec = NULL;
+ }
+}
+
/*
* Find the segment of @data that holds byte @off, and return how far into it
* @off is. @seg is a segment of the object that starts @pos bytes in and not
* after the one wanted: the walk goes on from there, and leaves @seg and @pos
* at the start of the segment it found.
*
- * A segment within IO_OPAQUE_SEEK_STRIDE steps is walked to. A farther one
+ * A segment within IO_OPAQUE_SEEK_STRIDE steps is walked to, which is all that
+ * a range after the previous one in the same object ever takes. A farther one
* is reached by jumping to the last anchor that is at or before @off, if that
* is ahead of where the walk is, and walking on from it: under a stride more.
* No seek takes more than two strides and a binary search over the anchors,
- * wherever it starts, however large the object is.
+ * wherever it starts, however large the object is, and so a vector of any
+ * ranges keeps tables for a time that its entries bound.
*/
static u32 io_opaque_vec_seek(const struct io_opaque_data *data, u64 off, u32 *seg, u64 *pos)
{
@@ -1822,6 +1984,39 @@ static u32 io_opaque_vec_seek(const struct io_opaque_data *data, u64 off, u32 *s
return off - at;
}
+/*
+ * The segments of the range of @len bytes that starts @skip bytes into segment
+ * @seg of @data, as a cursor over the whole object would visit them: the first
+ * starts inside its extent, the last ends inside its own. They are stored at
+ * @bv, and all of them counted, unless it is NULL: then only the count is
+ * wanted, and it stops at @room + 1, since a caller that has room for no more
+ * has no use for how many more there are. The range must lie within the
+ * object.
+ */
+static u32 io_opaque_vec_span(const struct io_opaque_data *data, u32 seg, u32 skip, u32 len,
+ struct bio_vec *bv, u64 room)
+{
+ const struct bio_vec *src = data->bvec;
+ u32 n = 0;
+
+ while (len && seg < data->nr) {
+ u32 take = min(src[seg].bv_len - skip, len);
+
+ if (bv) {
+ bv[n] = src[seg];
+ bv[n].bv_offset += skip;
+ bv[n].bv_len = take;
+ }
+ n++;
+ if (!bv && n > room)
+ break;
+ len -= take;
+ skip = 0;
+ seg++;
+ }
+ return n;
+}
+
/* The handle must hold a version, and the range must lie within it. */
static int io_opaque_send_range(const struct io_opaque_slot *slot, u64 off, u32 len)
{
@@ -1849,6 +2044,8 @@ static struct io_opaque_data *io_opaque_send_pin(struct io_opaque_store *store,
/* Drop the references a send took at admission; a no-op when it took none. */
static void io_opaque_send_unpin(struct io_opaque_req *op)
{
+ if (op->send_flags & IORING_OPAQUE_SEND_VECTOR)
+ io_opaque_vec_release(op);
io_opaque_data_put(op->store, op->data);
op->data = NULL;
}
@@ -1861,11 +2058,28 @@ static void io_opaque_send_unpin(struct io_opaque_req *op)
* last reference frees the version, every extent of it, and that is not for
* the lock to wait out. Nobody can take a reference on a version whose count
* is one, which is not in the table, so the last one is still the last then.
+ * A vector's entries that were put are cleared, so that the rest cannot put
+ * them again.
*/
static bool io_opaque_send_unpin_cheap(struct io_opaque_req *op)
{
bool left = false;
+ if (op->send_flags & IORING_OPAQUE_SEND_VECTOR) {
+ struct io_opaque_vec *vec = op->vec;
+ u32 i;
+
+ for (i = 0; i < vec->pinned; i++) {
+ if (!vec->pin[i] || !io_opaque_vec_owns(op, i))
+ continue;
+ if (refcount_dec_not_one(&vec->pin[i]->refs))
+ vec->pin[i] = NULL;
+ else
+ left = true;
+ }
+ if (!left)
+ vec->pinned = 0;
+ }
if (op->data) {
if (refcount_dec_not_one(&op->data->refs))
op->data = NULL;
@@ -1887,16 +2101,130 @@ static void io_opaque_send_iter(struct io_opaque_req *op, const struct bio_vec *
iov_iter_truncate(&op->iter, op->length);
}
-/* Resolve what a send names, take its references and set its cursor, under tables. */
+/*
+ * Admission of a vectored send, under tables; every entry sees the versions
+ * as they are in this one hold, and nothing in here sleeps. Pass one resolves
+ * and counts, and fails without having pinned or detached anything, so a
+ * vector that cannot go consumes nothing. It finds where each range starts by
+ * walking the segments of its version, and an entry that follows another in
+ * the same version, not before the segment that one started in, walks on from
+ * there instead of from the first: ranges in ascending order walk each object
+ * once. One that is more than a stride from where the walk is jumps to the
+ * version's anchors (see io_opaque_vec_seek()), so that ranges that go back
+ * or alternate between large objects cost a few dozen steps each, not an
+ * object. Counting the segments of the ranges stops as soon as the limit is
+ * passed. What tables is held for is therefore bounded by the entries and the
+ * limit, whatever the ranges. A block with too few segments is not grown here,
+ * as that allocates; the number needed is returned, and io_opaque_vec_finish()
+ * grows the block and asks again. Pass two takes the references and looks
+ * nothing up: pass one recorded every version in pin[], and tables has been
+ * held since. A LAST vector detaches each handle at its first entry, and an
+ * entry that repeats the handle takes one more reference on the version it
+ * found. Any other vector takes one reference for each run of entries that
+ * name the same handle (see io_opaque_vec_owns()), so eight ranges of one
+ * object cost one reference, and a handle that returns after another one pins
+ * again. Nothing can fail from there on. The cursor over the combined segments
+ * is io_opaque_vec_build()'s, once tables is dropped.
+ */
+static int io_opaque_vec_acquire(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ bool last = op->send_flags & IORING_OPAQUE_SEND_LAST;
+ struct io_opaque_vec *vec = op->vec;
+ const struct io_opaque_data *prev = NULL;
+ u64 segs = 0, pos = 0;
+ u32 i, seg = 0;
+ int ret;
+
+ if (store->dead)
+ return -ESHUTDOWN;
+ for (i = 0; i < vec->nr; i++) {
+ const struct io_uring_opaque_send_entry *ent = &vec->ent[i];
+ struct io_opaque_slot *slot = io_opaque_lookup(store, ent->handle);
+ struct io_opaque_data *data;
+ u32 skip, n;
+
+ ret = io_opaque_send_range(slot, ent->off, ent->len);
+ if (ret)
+ return ret;
+ data = slot->data;
+ if (data != prev || ent->off < pos) {
+ prev = data;
+ seg = 0;
+ pos = 0;
+ }
+ skip = io_opaque_vec_seek(data, ent->off, &seg, &pos);
+ vec->start[i].seg = seg;
+ vec->start[i].skip = skip;
+ n = io_opaque_vec_span(data, seg, skip, ent->len, NULL,
+ IO_OPAQUE_SEND_VEC_SEGS - segs);
+ segs += n;
+ if (segs > IO_OPAQUE_SEND_VEC_SEGS)
+ return -EMSGSIZE;
+ /* Not yet a reference: pass two takes it, and a repeated handle needs it. */
+ vec->pin[i] = data;
+ }
+ /* Within the limit, so it fits an int. */
+ if (segs > vec->cap_segs)
+ return segs;
+ for (i = 0; i < vec->nr; i++) {
+ struct io_opaque_slot *slot;
+
+ if (!last) {
+ if (io_opaque_vec_owns(op, i))
+ refcount_inc(&vec->pin[i]->refs);
+ continue;
+ }
+ slot = io_opaque_lookup(store, vec->ent[i].handle);
+ if (!slot->data)
+ refcount_inc(&vec->pin[i]->refs);
+ else
+ vec->pin[i] = io_opaque_send_pin(store, slot, true);
+ }
+ vec->pinned = vec->nr;
+ if (last)
+ op->consumed = true;
+ return 0;
+}
+
+/*
+ * The cursor over the combined segments, from the starts that admission
+ * recorded. It takes no lock and is done without tables, which is the point of
+ * it: each version is held by the send's own reference, and the segments of a
+ * version never change.
+ */
+static void io_opaque_vec_build(struct io_opaque_req *op)
+{
+ struct io_opaque_vec *vec = op->vec;
+ u32 i, filled = 0;
+
+ lockdep_assert_not_held(&op->store->tables);
+ for (i = 0; i < vec->nr; i++)
+ filled += io_opaque_vec_span(vec->pin[i], vec->start[i].seg, vec->start[i].skip,
+ vec->ent[i].len, vec->bv + filled, U64_MAX);
+ io_opaque_send_iter(op, vec->bv, filled, op->length, 0);
+}
+
+/*
+ * Resolve what a send names and take its references, under tables. A send of
+ * one range also gets its cursor. A vector returns 0 with its references taken,
+ * a negative errno with none taken and nothing removed, or the number of
+ * segments its ranges span when its block holds fewer, to be settled by
+ * io_opaque_vec_finish() once tables is dropped.
+ */
static int io_opaque_send_acquire(struct io_opaque_req *op)
{
struct io_opaque_store *store = op->store;
bool last = op->send_flags & IORING_OPAQUE_SEND_LAST;
- struct io_opaque_slot *slot = io_opaque_lookup(store, op->handle);
+ struct io_opaque_slot *slot;
u64 pos = 0;
u32 seg = 0, skip;
int ret;
+ if (unlikely(op->send_flags & IORING_OPAQUE_SEND_VECTOR))
+ return io_opaque_vec_acquire(op);
+ slot = io_opaque_lookup(store, op->handle);
+
if (store->dead)
return -ESHUTDOWN;
if (!op->length)
@@ -1907,13 +2235,58 @@ static int io_opaque_send_acquire(struct io_opaque_req *op)
op->data = io_opaque_send_pin(store, slot, last);
if (last)
op->consumed = true;
- /* The cursor starts at the segment that holds the range. */
+ /* The cursor starts at the segment that holds the range, found as a vector's are. */
skip = io_opaque_vec_seek(op->data, op->offset, &seg, &pos);
io_opaque_send_iter(op, op->data->bvec + seg, op->data->nr - seg, op->data->length - pos,
skip);
return 0;
}
+/*
+ * What admission of a vector does without tables, after the hold in which
+ * io_opaque_send_acquire() returned @ret. A block with too few segments (@ret
+ * of them are needed) is grown here, where allocating is fine, and the vector
+ * is resolved again in a new hold: nothing was pinned, and the versions may
+ * have changed meanwhile, so every entry is looked at again. Once the vector
+ * is admitted, the cursor is built.
+ */
+static int io_opaque_vec_finish(struct io_opaque_req *op, int ret)
+{
+ while (ret > 0) {
+ ret = io_opaque_vec_grow(op, op->vec->nr, ret, op->vec->nr);
+ if (ret)
+ return ret;
+ scoped_guard(mutex, &op->store->tables)
+ ret = io_opaque_send_acquire(op);
+ }
+ if (!ret)
+ io_opaque_vec_build(op);
+ return ret;
+}
+
+/*
+ * A vectored send: addr names len entries and off is zero. They are copied
+ * now, like SEND_ZC's iovecs, and the array need not stay valid. The request
+ * still holds the SQE's addr, off and len in handle, offset and length; the
+ * commit turns them into the body, whose handle and offset are none.
+ */
+static int io_opaque_send_vec_prep(struct io_opaque_req *op)
+{
+ u32 nr = op->length;
+ int ret;
+
+ if (op->offset || !nr || nr > IORING_OPAQUE_SEND_VEC_MAX)
+ return -EINVAL;
+ /* Every entry spans a segment; admission grows the block if some span more. */
+ ret = io_opaque_vec_grow(op, nr, nr, 0);
+ if (ret)
+ return ret;
+ if (copy_from_user(op->vec->ent, u64_to_user_ptr(op->handle),
+ array_size(nr, sizeof(*op->vec->ent))))
+ return -EFAULT;
+ return io_opaque_vec_commit(op, nr);
+}
+
int io_opaque_send_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
{
struct io_opaque_store *store = io_opaque_get_store(req, READ_ONCE(sqe->opaque_store_id));
@@ -1925,7 +2298,7 @@ int io_opaque_send_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
if (!store)
return -ENXIO;
- if ((send_flags & ~IORING_OPAQUE_SEND_LAST) ||
+ if ((send_flags & ~(IORING_OPAQUE_SEND_LAST | IORING_OPAQUE_SEND_VECTOR)) ||
((send_flags & IORING_OPAQUE_SEND_LAST) && (req->flags & REQ_F_CQE_SKIP)) ||
READ_ONCE(sqe->buf_index) || READ_ONCE(sqe->__pad2[0]) ||
(msg_flags & ~(MSG_MORE | MSG_DONTWAIT)))
@@ -1943,6 +2316,11 @@ int io_opaque_send_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
op->msg_flags = msg_flags;
if (op->msg_flags & MSG_DONTWAIT)
req->flags |= REQ_F_NOWAIT;
+ if (unlikely(send_flags & IORING_OPAQUE_SEND_VECTOR)) {
+ ret = io_opaque_send_vec_prep(op);
+ if (ret)
+ return ret;
+ }
if ((!op->length && !op->addr) || op->length > INT_MAX ||
check_add_overflow(op->offset, (u64)op->length, &end))
return -EINVAL;
@@ -2036,13 +2414,20 @@ static struct io_opaque_tx *io_opaque_tx_alloc(struct io_opaque_store *store,
return kzalloc_obj(struct io_opaque_tx, gfp);
}
-/* Resolve what a send names and take its references, under tables. */
+/*
+ * Resolve what @op names and take its references: the handles under tables,
+ * and then, without it, the block and the cursor. A vector leaves tables once
+ * its references are taken; one whose block is too small leaves it earlier, to
+ * grow the block, and is resolved again.
+ */
static int io_opaque_tx_acquire(struct io_opaque_req *op)
{
int ret;
scoped_guard(mutex, &op->store->tables)
ret = io_opaque_send_acquire(op);
+ if (unlikely(op->send_flags & IORING_OPAQUE_SEND_VECTOR))
+ ret = io_opaque_vec_finish(op, ret);
return ret;
}
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 09/12] io_uring/opaque: add IORING_OP_OPAQUE_COLLECT
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (7 preceding siblings ...)
2026-11-10 13:20 ` [RFC PATCH net-next 08/12] io_uring/opaque: add vectored IORING_OP_OPAQUE_OBJ_SEND Alex Markuze
@ 2026-11-10 13:20 ` Alex Markuze
2026-11-10 13:20 ` [RFC PATCH net-next 10/12] io_uring/opaque: add IORING_OPAQUE_CREATE Alex Markuze
` (2 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:20 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Add a multishot request that attaches a connected TCP socket to a store
and collects its receive stream. sqe->fd names the socket and
opaque_store_id the store. ioprio accepts POLL_FIRST; IOSQE_ASYNC and
successful-CQE suppression are rejected.
Attachment requires ESTABLISHED or CLOSE_WAIT, the original TCP protocol,
no receive owner, repair mode, urgent data or sk_user_data. Authorize
receive access under the request's credentials before attachment and
before taking the socket lock. This authorization covers the collector's
lifetime; subsequent passes do not recheck receive policy. Attachment
posts the stream-token CQE before publishing the receive claim. If that
CQE cannot be posted, no claim or byte is taken. Stream offset zero is
the first unread byte at attachment.
Each issue captures at most 256 KiB through tcp_read_sock(). The actor
follows the earliest pending stream decision: DISCARD consumes without
retention, RECV_OBJECT builds private backing, and other bytes become
undecided. Eligible unshared receive pages are retained by reference;
other data is packed into owned copies. Undecided bytes normally stop
further collection so TCP flow control can apply. Decisions and waiting
inspection requests can make collection resume.
Byte or extent pressure produces a nonterminal -ENOBUFS CQE, at most
once per unread offset, and enrolls the collector on the store's budget
wait list. Allocation failure imposes a 50-ms retry deadline. Wakeups
grant no capacity: reentry checks both budgets and the deadline again.
An infeasible partial RECV_OBJECT returns its private prefix and fails
instead of holding capacity while waiting for the rest of itself.
Use native multishot socket polling between collection passes. The
receive-owner callback, decisions, closure and capacity returns set
a persistent private collector predicate before waking the socket
waitqueue with kernel-only POLL_RX_OWNER (bit 26). Native poll enrolls
before checking the predicate and consumes it only on collector entry.
Filter the private key before taking poll ownership: ordinary polls
and output async polls ignore it. Eligible payload-read async polls
recheck current receive ownership even after coalesced notifications,
so the real receive handler reports the owner conflict. Error-queue
receives retain their existing behavior. No UAPI poll event is added.
Retain a pass after both initial and multishot batch exhaustion. A
worker poll before first issue permits initial attachment unless
POLL_FIRST requested ordinary socket readiness. Set NO_LAZY because
one collector task-work enqueue can complete several range requests.
While budget-blocked, the collector still observes errors, urgent data
and EOF, consumes a payload-free FIN, and can execute a DISCARD at its
capture cursor without allocating.
EOF, socket error, urgent data, cancellation, stream close or inability
to post a pressure CQE ends the collector. Cancellation closes its stream
and releases retained undecided data, including cancellation after a
native poll-setup failure. A token is not a recovery promise after
that failure. EOF and socket errors leave retained data available for
pending decisions. The stream token remains until
STREAM_CLOSE or store teardown. The stream retains its socket file.
Capture takes stream->lock, then the socket lock, then wait_lock.
Attachment takes the socket lock before tables and sk_callback_lock;
no table holder waits for the socket lock.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
include/linux/poll.h | 3 +
include/uapi/linux/io_uring.h | 1 +
io_uring/opaque.c | 599 +++++++++++++++++++++++++++++++++-
io_uring/opaque.h | 6 +
io_uring/opaque_internal.h | 3 +
io_uring/opdef.c | 18 +
io_uring/poll.c | 58 +++-
7 files changed, 680 insertions(+), 8 deletions(-)
diff --git a/include/linux/poll.h b/include/linux/poll.h
index 12bb18e8b978..b2a85a503fd9 100644
--- a/include/linux/poll.h
+++ b/include/linux/poll.h
@@ -12,6 +12,9 @@
#include <uapi/linux/poll.h>
#include <uapi/linux/eventpoll.h>
+/* Wake-only receive-owner notification, not file readiness or a UAPI event. */
+#define POLL_RX_OWNER ((__force __poll_t)(1U << 26))
+
/* ~832 bytes of stack space used max in sys_select/sys_poll before allocating
additional memory. */
#define MAX_STACK_ALLOC 832
diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
index 1759f4f805b1..6469ab391eec 100644
--- a/include/uapi/linux/io_uring.h
+++ b/include/uapi/linux/io_uring.h
@@ -322,6 +322,7 @@ enum io_uring_op {
IORING_OP_URING_CMD128,
IORING_OP_OPAQUE_OBJ,
IORING_OP_OPAQUE_OBJ_SEND,
+ IORING_OP_OPAQUE_COLLECT,
/* this goes last, obviously */
IORING_OP_LAST,
diff --git a/io_uring/opaque.c b/io_uring/opaque.c
index 0c58900073f7..4610e212a800 100644
--- a/io_uring/opaque.c
+++ b/io_uring/opaque.c
@@ -8,6 +8,7 @@
#include <linux/overflow.h>
#include <linux/rbtree.h>
#include <linux/sched/mm.h>
+#include <linux/security.h>
#include <linux/skbuff.h>
#include <linux/sizes.h>
#include <linux/slab.h>
@@ -49,6 +50,38 @@ static void io_opaque_vec_cleanup(struct io_opaque_req *op);
static void io_opaque_stream_close(struct io_opaque_stream *stream);
static void io_opaque_tx_put(struct io_opaque_req *op);
+/*
+ * Publish before waking; native poll enrolls before reading collect_ready.
+ * This is not TCP readiness: only COLLECT and receive-owner-affected apoll
+ * recheck their predicates. Ordinary poll and output apoll ignore the key.
+ */
+static void io_opaque_collect_wake(struct io_opaque_stream *stream)
+{
+ struct socket *sock = sock_from_file(stream->file);
+
+ atomic_set(&stream->collect_ready, 1);
+ wake_up_interruptible_poll(sk_sleep(sock->sk), POLL_RX_OWNER);
+}
+
+static void io_opaque_data_ready(struct sock *sk, void *data)
+{
+ struct io_opaque_stream *stream = data;
+
+ /* sk_callback_lock pins stream; do not take stream or socket locks here. */
+ atomic_set(&stream->collect_ready, 1);
+ wake_up_interruptible_poll(sk_sleep(sk), POLL_RX_OWNER);
+}
+
+__poll_t io_opaque_collect_poll(struct io_kiocb *req)
+{
+ struct io_opaque_req *op = req->async_data;
+
+ /* io-wq may poll before the first issue, including a trailing link. */
+ if (!op->stream)
+ return op->collect_flags & IORING_RECVSEND_POLL_FIRST ? 0 : EPOLLIN;
+ return atomic_read(&op->stream->collect_ready) ? EPOLLIN : 0;
+}
+
static bool io_opaque_budget_ready(struct io_opaque_req *op, unsigned long now)
{
struct io_opaque_store *store = op->store;
@@ -103,8 +136,6 @@ static bool io_opaque_wake_budget(struct io_opaque_store *store)
guard(spinlock)(&store->wait_lock);
stopped = READ_ONCE(store->dead);
list_for_each_entry_safe(op, next, &store->budget_waits, wait) {
- struct socket *sock = sock_from_file(op->req->file);
-
if (!stopped && !READ_ONCE(op->stream->closed)) {
if (time_before(now, op->retry_at)) {
delay = min(delay, op->retry_at - now);
@@ -122,7 +153,7 @@ static bool io_opaque_wake_budget(struct io_opaque_store *store)
/* A wake is not a reservation; the collector rechecks on entry. */
woken++;
/* Native multishot polling also observes FIN, RST and urgent data. */
- wake_up_interruptible_poll(sk_sleep(sock->sk), EPOLLIN);
+ io_opaque_collect_wake(op->stream);
}
if (!stopped && delay != MAX_JIFFY_OFFSET)
queue_delayed_work(system_dfl_wq, &store->retry_work, delay);
@@ -147,7 +178,6 @@ static void io_opaque_retry_work(struct work_struct *work)
static void io_opaque_wake_stream(struct io_opaque_stream *stream)
{
struct io_opaque_req *op = stream->collector;
- struct socket *sock = sock_from_file(stream->file);
lockdep_assert_held(&stream->lock);
if (!op)
@@ -158,7 +188,7 @@ static void io_opaque_wake_stream(struct io_opaque_stream *stream)
return;
list_del_init(&op->wait);
}
- wake_up_interruptible_poll(sk_sleep(sock->sk), EPOLLIN);
+ io_opaque_collect_wake(stream);
}
int io_opaque_memory_mode(struct io_opaque_store *store)
@@ -1158,6 +1188,269 @@ static void io_opaque_stream_process(struct io_opaque_stream *stream)
}
}
+/* Only the earliest undecided range is consulted on the capture hot path. */
+static struct io_opaque_req *io_opaque_next_decision(struct io_opaque_stream *stream)
+{
+ struct rb_node *node = rb_first_cached(&stream->decisions);
+
+ return node ? rb_entry(node, struct io_opaque_req, decision) : NULL;
+}
+
+/* Undecided bytes are collected only when nothing is pending or a request needs later bytes. */
+static bool io_opaque_collect_demand(struct io_opaque_stream *stream)
+{
+ struct io_opaque_req *op = io_opaque_next_decision(stream);
+
+ lockdep_assert_held(&stream->lock);
+ return !stream->undecided || !list_empty(&stream->reads) ||
+ (op && op->offset > stream->next);
+}
+
+static struct io_opaque_req *io_opaque_capture_keep(struct io_opaque_stream *stream)
+{
+ struct io_opaque_req *op = io_opaque_next_decision(stream);
+
+ if (op && op->op == IORING_OPAQUE_RECV_OBJECT && stream->next >= op->offset &&
+ stream->next < op->offset + op->length)
+ return op;
+ return NULL;
+}
+
+static struct io_opaque_extent *io_opaque_capture_tail(struct io_opaque_stream *stream)
+{
+ struct io_opaque_req *op = io_opaque_capture_keep(stream);
+ struct list_head *list = op ? &op->data->extents : &stream->extents;
+
+ return list_last_entry_or_null(list, struct io_opaque_extent, list);
+}
+
+static int io_opaque_append(struct io_opaque_stream *stream, struct page *page,
+ u32 offset, u32 length, bool take_ref, bool copied)
+{
+ struct io_opaque_req *op = io_opaque_capture_keep(stream);
+ struct list_head *list = &stream->extents;
+ struct io_opaque_extent *extent, *tail;
+ u64 start = stream->next;
+ bool new_backing = false;
+
+ if (op) {
+ list = &op->data->extents;
+ start -= op->offset;
+ }
+ tail = list_last_entry_or_null(list, struct io_opaque_extent, list);
+ offset += (page - compound_head(page)) * PAGE_SIZE;
+ page = compound_head(page);
+ if (tail && tail->backing->page == page) {
+ if (tail->start + tail->length == start &&
+ tail->offset + tail->length == offset) {
+ tail->length += length;
+ extent = tail;
+ goto accounted;
+ }
+ }
+ stream->need_bytes = page_size(page);
+ if (atomic_read(&stream->store->extents) >= stream->store->config.max_extents - 2)
+ return -ENOBUFS;
+ if (tail && tail->backing->page == page) {
+ if (!io_opaque_extent_reserve(stream->store))
+ return -ENOBUFS;
+ extent = kmalloc_obj(*extent, GFP_KERNEL_ACCOUNT);
+ if (!extent) {
+ io_opaque_extent_unreserve(stream->store, 0);
+ return -ENOMEM;
+ }
+ *extent = (struct io_opaque_extent) {
+ .backing = tail->backing, .start = start,
+ .offset = offset, .length = length,
+ };
+ refcount_inc(&extent->backing->refs);
+ } else {
+ u64 left = op ? ALIGN(op->length - op->progress - length, PAGE_SIZE) : 0;
+
+ /* Leave room for dense copies of a private KEEP's remaining bytes. */
+ if (!copied && op && op->data->charge + stream->need_bytes + left >
+ io_opaque_rx_limit(stream->store))
+ return -ENOBUFS;
+ extent = io_opaque_extent_new(stream->store, page, offset, length, start, copied);
+ if (IS_ERR(extent))
+ return PTR_ERR(extent);
+ if (take_ref)
+ get_page(page);
+ extent->backing->copied = copied;
+ new_backing = true;
+ }
+ list_add_tail(&extent->list, list);
+accounted:
+ if (op) {
+ op->progress += length;
+ op->data->length += length;
+ if (new_backing)
+ op->data->charge += extent->backing->charge;
+ } else {
+ stream->undecided += length;
+ }
+ if (copied) {
+ extent->backing->filled = offset + length;
+ atomic64_add(length, &stream->store->copied);
+ }
+ return length;
+}
+
+static int io_opaque_capture(struct io_opaque_stream *stream,
+ const struct sk_buff *skb, u32 offset, u32 len,
+ bool shared)
+{
+ u32 consumed = 0, head = skb_headlen(skb), start = head;
+ bool copy_frags;
+ struct sk_buff *child;
+ u64 retain_limit = io_opaque_rx_limit(stream->store);
+ u32 chunk;
+ int i, ret = -EFAULT;
+
+ shared |= skb_shared(skb);
+ copy_frags = shared || skb_cloned(skb) || skb_has_shared_frag(skb);
+ if (!skb_frags_readable(skb))
+ return -EIO;
+ if (offset < head) {
+ chunk = min(head - offset, len);
+ if (!shared && !skb_head_is_locked(skb)) {
+ void *data = skb->data + offset;
+ struct page *page = virt_to_page(data);
+
+ /* A compound allocation larger than the limit can never fit. */
+ if (page_size(compound_head(page)) > retain_limit)
+ goto copy;
+ ret = io_opaque_append(stream, page, offset_in_page(data),
+ chunk, true, false);
+ } else {
+ goto copy;
+ }
+ if (ret == -ENOBUFS)
+ goto copy;
+ goto head_done;
+ }
+ for (i = 0; i < skb_shinfo(skb)->nr_frags; i++) {
+ const skb_frag_t *frag = &skb_shinfo(skb)->frags[i];
+ u32 size = skb_frag_size(frag);
+
+ if (offset >= start + size) {
+ start += size;
+ continue;
+ }
+ chunk = min(start + size - offset, len);
+ if (copy_frags || page_size(compound_head(skb_frag_page(frag))) > retain_limit)
+ goto copy;
+ ret = io_opaque_append(stream, skb_frag_page(frag),
+ skb_frag_off(frag) + offset - start, chunk, true, false);
+ if (ret == -ENOBUFS)
+ goto copy;
+ goto head_done;
+ }
+ skb_walk_frags(skb, child) {
+ if (offset >= start + child->len) {
+ start += child->len;
+ continue;
+ }
+ return io_opaque_capture(stream, child, offset - start,
+ min(start + child->len - offset, len), copy_frags);
+ }
+ return -EFAULT;
+copy:
+ {
+ struct io_opaque_extent *tail = io_opaque_capture_tail(stream);
+ struct page *page;
+ u32 pos = 0;
+ bool allocated = true;
+ void *mapped;
+
+ if (tail && tail->backing->copied && tail->backing->filled < PAGE_SIZE &&
+ tail->offset + tail->length == tail->backing->filled) {
+ page = tail->backing->page;
+ pos = tail->backing->filled;
+ allocated = false;
+ } else {
+ stream->need_bytes = PAGE_SIZE;
+ if (!io_opaque_charge(stream->store, PAGE_SIZE, true))
+ return -ENOBUFS;
+ page = alloc_page(GFP_KERNEL_ACCOUNT);
+ if (!page) {
+ io_opaque_uncharge(stream->store, PAGE_SIZE);
+ return -ENOMEM;
+ }
+ }
+ chunk = min_t(u32, chunk, PAGE_SIZE - pos);
+ mapped = kmap_local_page(page);
+ ret = skb_copy_bits(skb, offset, mapped + pos, chunk);
+ kunmap_local(mapped);
+ if (!ret)
+ ret = io_opaque_append(stream, page, pos, chunk, false, true);
+ if (ret < 0 && allocated) {
+ put_page(page);
+ io_opaque_uncharge(stream->store, PAGE_SIZE);
+ }
+ }
+head_done:
+ if (ret > 0)
+ consumed += ret;
+ return consumed ?: ret;
+}
+
+static int io_opaque_actor(read_descriptor_t *desc, struct sk_buff *skb,
+ unsigned int offset, size_t length)
+{
+ struct io_opaque_stream *stream = desc->arg.data;
+ u32 consumed = 0;
+ int ret = 0;
+
+ /* The first capture error ends the pass; later bytes wait for the next one. */
+ if (desc->error)
+ return 0;
+ length = min(length, desc->count);
+ length = min_t(u64, length, U64_MAX - stream->next);
+ while (length) {
+ struct io_opaque_req *op = io_opaque_next_decision(stream);
+ u32 chunk = length;
+
+ if (op && stream->next >= op->offset) {
+ chunk = min_t(u64, chunk, op->offset + op->length - stream->next);
+ if (op->op == IORING_OPAQUE_DISCARD) {
+ op->progress += chunk;
+ ret = chunk;
+ goto consumed;
+ }
+ } else {
+ /*
+ * Undecided bytes stay on the socket until someone asks for them.
+ * A decision further in the stream asks for the gap before it,
+ * including one that became the earliest during this pass.
+ */
+ if (stream->hold_undecided && !op)
+ break;
+ if (op)
+ chunk = min_t(u64, chunk, op->offset - stream->next);
+ }
+ ret = io_opaque_capture(stream, skb, offset, chunk, false);
+ if (ret <= 0) {
+ desc->error = ret;
+ break;
+ }
+consumed:
+ stream->next += ret;
+ consumed += ret;
+ offset += ret;
+ length -= ret;
+ desc->count -= ret;
+ if (op && op->progress == op->length) {
+ rb_erase_cached(&op->decision, &stream->decisions);
+ RB_CLEAR_NODE(&op->decision);
+ io_opaque_queue_ready(op, op->length);
+ }
+ }
+ if (!length && stream->next == U64_MAX)
+ stream->error = -EOVERFLOW;
+ return consumed ?: ret;
+}
+
static void io_opaque_stream_close(struct io_opaque_stream *stream)
{
struct socket *sock = sock_from_file(stream->file);
@@ -1174,10 +1467,299 @@ static void io_opaque_stream_close(struct io_opaque_stream *stream)
io_opaque_extents_free(stream->store, &stream->extents);
stream->undecided = 0;
io_opaque_stream_process(stream);
- wake_up_interruptible_poll(sk_sleep(sock->sk), EPOLLIN);
+ io_opaque_collect_wake(stream);
io_opaque_wake_budget(stream->store);
}
+int io_opaque_recv_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
+{
+ struct io_opaque_store *store = xa_load(&req->ctx->opaque_stores,
+ READ_ONCE(sqe->opaque_store_id));
+ struct io_opaque_req *op;
+ u16 flags = READ_ONCE(sqe->ioprio);
+
+ if (!store)
+ return -ENXIO;
+ if (READ_ONCE(sqe->addr) || READ_ONCE(sqe->off) || READ_ONCE(sqe->addr3) ||
+ READ_ONCE(sqe->len) || READ_ONCE(sqe->msg_flags) ||
+ READ_ONCE(sqe->buf_index) || READ_ONCE(sqe->__pad2[0]) ||
+ (flags & ~IORING_RECVSEND_POLL_FIRST) ||
+ (req->flags & (REQ_F_CQE_SKIP | REQ_F_FORCE_ASYNC)))
+ return -EINVAL;
+ op = io_opaque_req_alloc(req, store);
+ if (IS_ERR(op))
+ return PTR_ERR(op);
+ op->op = U16_MAX - 1;
+ op->collect_flags = flags;
+ /* One native poll task-work enqueue may produce several range CQEs. */
+ req->flags |= REQ_F_APOLL_MULTISHOT | REQ_F_POLL_NO_LAZY;
+ return 0;
+}
+
+/*
+ * Make @op the collector of its socket. The socket is taken before tables,
+ * which is the order a send admits in, with the socket owned. Nothing may wait
+ * for a socket lock while it holds tables.
+ */
+static int io_opaque_attach(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_stream *stream;
+ struct socket *sock = sock_from_file(op->req->file);
+ struct msghdr msg = { .msg_flags = MSG_DONTWAIT };
+ struct io_uring_cqe cqe[2] = {};
+ u32 i;
+ int ret;
+
+ if (!sock || sock->type != SOCK_STREAM || sock->sk->sk_protocol != IPPROTO_TCP)
+ return -EOPNOTSUPP;
+ /* Authorize the collector's lifetime before taking the socket lock. */
+ ret = security_socket_recvmsg(sock, &msg, 0, MSG_DONTWAIT);
+ if (ret)
+ return ret;
+ stream = kzalloc_obj(*stream, GFP_KERNEL_ACCOUNT);
+ if (!stream)
+ return -ENOMEM;
+ refcount_set(&stream->refs, 1);
+ atomic_set(&stream->collect_ready, 0);
+ mutex_init(&stream->lock);
+ INIT_LIST_HEAD(&stream->extents);
+ INIT_LIST_HEAD(&stream->reads);
+ stream->decisions = RB_ROOT_CACHED;
+ stream->store = store;
+ stream->file = get_file(op->req->file);
+ stream->owner.actor = io_opaque_actor;
+ stream->owner.data = stream;
+ stream->owner.data_ready = io_opaque_data_ready;
+ stream->collector = op;
+ lock_sock(sock->sk);
+ mutex_lock(&store->tables);
+ if (sock->sk->sk_rx_owner || sock->sk->sk_prot != sock->sk->sk_prot_creator ||
+ tcp_sk(sock->sk)->repair || tcp_sk(sock->sk)->urg_data) {
+ ret = -EBUSY;
+ goto unlock;
+ }
+ if (sock->sk->sk_state != TCP_ESTABLISHED && sock->sk->sk_state != TCP_CLOSE_WAIT) {
+ ret = -ENOTCONN;
+ goto unlock;
+ }
+ ret = -ENOSPC;
+ for (;;) {
+ struct io_opaque_stream_slot *slot;
+
+ i = io_opaque_find_slot(store->stream_used, store->config.max_streams,
+ &store->stream_cursor);
+ if (i >= store->config.max_streams)
+ break;
+ slot = &store->streams[i];
+ if (slot->generation == IO_OPAQUE_GENERATION_MAX) {
+ __set_bit(i, store->stream_used);
+ continue;
+ }
+ write_lock_bh(&sock->sk->sk_callback_lock);
+ if (sock->sk->sk_user_data) {
+ write_unlock_bh(&sock->sk->sk_callback_lock);
+ ret = -EBUSY;
+ break;
+ }
+ slot->generation++;
+ stream->token = io_opaque_handle(i, slot->generation) | IO_OPAQUE_TOKEN_TAG;
+ cqe[0].user_data = op->req->cqe.user_data;
+ cqe[0].flags = IORING_CQE_F_MORE | ctx_cqe32_flags(op->req->ctx);
+ memcpy(&cqe[1], &stream->token, sizeof(stream->token));
+ if (!io_req_post_cqe32(op->req, cqe)) {
+ write_unlock_bh(&sock->sk->sk_callback_lock);
+ ret = -ENOSPC;
+ break;
+ }
+ WRITE_ONCE(sock->sk->sk_rx_owner, &stream->owner);
+ write_unlock_bh(&sock->sk->sk_callback_lock);
+ slot->stream = stream;
+ __set_bit(i, store->stream_used);
+ store->nr_streams++;
+ refcount_inc(&stream->refs);
+ op->stream = stream;
+ ret = 0;
+ break;
+ }
+unlock:
+ mutex_unlock(&store->tables);
+ release_sock(sock->sk);
+ /* Ownership publication is not an ordinary TCP readiness event either. */
+ if (!ret)
+ wake_up_interruptible_poll(sk_sleep(sock->sk), POLL_RX_OWNER);
+ if (ret)
+ io_opaque_stream_put(stream);
+ return ret;
+}
+
+/* Report a budget stall once per unread offset, without ending collection. */
+static int io_opaque_budget_cqe(struct io_opaque_req *op)
+{
+ struct io_opaque_stream *stream = op->stream;
+ struct io_uring_cqe cqe[2] = {};
+ u64 extra[2] = { stream->token, stream->next };
+
+ if (op->budget_reported && op->offset == stream->next)
+ return 0;
+ cqe[0].res = -ENOBUFS;
+ cqe[0].flags = IORING_CQE_F_MORE | ctx_cqe32_flags(op->req->ctx);
+ memcpy(&cqe[1], extra, sizeof(extra));
+ if (!io_req_post_cqe32(op->req, cqe))
+ return -ENOSPC;
+ op->budget_reported = true;
+ op->offset = stream->next;
+ return 0;
+}
+
+static void io_opaque_budget_wait(struct io_opaque_req *op, int error)
+{
+ lockdep_assert_held(&op->stream->lock);
+ scoped_guard(spinlock, &op->store->wait_lock) {
+ op->need_bytes = error == -ENOMEM ? 0 : op->stream->need_bytes;
+ if (error == -ENOMEM && !time_before(jiffies, op->retry_at))
+ op->retry_at = jiffies + msecs_to_jiffies(IO_OPAQUE_ALLOC_RETRY_MS);
+ else if (error == -ENOBUFS)
+ op->retry_at = jiffies;
+ op->phase = IO_OPAQUE_BUDGET_POLL;
+ list_add_tail(&op->wait, &op->store->budget_waits);
+ }
+ /* Recheck after enrollment to close the release/enrollment race. */
+ io_opaque_wake_budget(op->store);
+}
+
+int io_opaque_recv(struct io_kiocb *req, unsigned int issue_flags)
+{
+ struct io_opaque_req *op = req->async_data;
+ struct io_opaque_stream *stream;
+ struct socket *sock = sock_from_file(req->file);
+ read_descriptor_t desc = { .count = IO_OPAQUE_RX_BATCH };
+ struct mem_cgroup *old;
+ bool blocked;
+ int ret;
+
+ io_ring_submit_lock(req->ctx, issue_flags);
+ if (!(req->flags & REQ_F_POLLED) &&
+ (op->collect_flags & IORING_RECVSEND_POLL_FIRST)) {
+ ret = -EAGAIN;
+ goto out;
+ }
+ if (!op->stream && (issue_flags & IO_URING_F_UNLOCKED)) {
+ io_req_task_queue(req);
+ ret = IOU_ISSUE_SKIP_COMPLETE;
+ goto out;
+ }
+ if (!op->stream) {
+ ret = io_opaque_attach(op);
+ if (ret)
+ goto out;
+ }
+ stream = op->stream;
+ mutex_lock(&stream->lock);
+ /* A concurrent data wake after this exchange belongs to the next pass. */
+ atomic_xchg(&stream->collect_ready, 0);
+ if (stream->closed || READ_ONCE(op->store->dead)) {
+ ret = -ECANCELED;
+ goto unlock_stream;
+ }
+ scoped_guard(spinlock, &op->store->wait_lock) {
+ blocked = io_opaque_collect_blocked(op);
+ if (blocked) {
+ /* Capacity may have been taken since the wake removed us. */
+ if (list_empty(&op->wait))
+ list_add_tail(&op->wait, &op->store->budget_waits);
+ } else {
+ list_del_init(&op->wait);
+ op->phase = IO_OPAQUE_IDLE;
+ }
+ }
+ desc.arg.data = stream;
+ old = set_active_memcg(op->store->memcg);
+ if (!blocked) {
+ /* Retained prefixes precede bytes captured directly into a TAKE. */
+ io_opaque_stream_process(stream);
+ stream->hold_undecided = !io_opaque_collect_demand(stream);
+ }
+ lock_sock(sock->sk);
+ if (tcp_sk(sock->sk)->urg_data) {
+ ret = -EOPNOTSUPP;
+ } else if (blocked) {
+ ret = 0;
+ if (sock->sk->sk_shutdown & RCV_SHUTDOWN) {
+ /* Consume a FIN-only skb, never payload, while out of budget. */
+ desc.count = 0;
+ ret = tcp_read_sock(sock->sk, &desc, io_opaque_actor);
+ }
+ } else {
+ ret = tcp_read_sock(sock->sk, &desc, io_opaque_actor);
+ }
+ if (desc.error)
+ ret = desc.error;
+ if (sock->sk->sk_err)
+ stream->error = sock_error(sock->sk);
+ if ((sock->sk->sk_shutdown & RCV_SHUTDOWN) &&
+ skb_queue_empty(&sock->sk->sk_receive_queue))
+ stream->eof = true;
+ release_sock(sock->sk);
+ set_active_memcg(old);
+ if (ret < 0 && ret != -ENOBUFS && ret != -ENOMEM)
+ stream->error = ret;
+ if (blocked && !stream->error && !stream->eof) {
+ /* Close release/reenrollment races without retrying TCP capture. */
+ io_opaque_wake_budget(op->store);
+ ret = IOU_RETRY;
+ goto unlock_stream;
+ }
+ if (blocked) {
+ guard(spinlock)(&op->store->wait_lock);
+
+ list_del_init(&op->wait);
+ op->phase = IO_OPAQUE_IDLE;
+ }
+ io_opaque_stream_process(stream);
+ if (ret == -ENOBUFS) {
+ struct io_opaque_req *keep = io_opaque_capture_keep(stream);
+
+ /* A private prefix must not pin quota indefinitely awaiting itself. */
+ if (keep && keep->progress) {
+ rb_erase_cached(&keep->decision, &stream->decisions);
+ RB_CLEAR_NODE(&keep->decision);
+ io_opaque_restore(keep);
+ io_opaque_queue_ready(keep, -ENOBUFS);
+ }
+ }
+ if (ret == -ENOBUFS && !stream->error && !stream->eof) {
+ int err = io_opaque_budget_cqe(op);
+
+ if (err) {
+ /* Retain the token and captured prefix for explicit recovery. */
+ stream->error = err;
+ io_opaque_stream_process(stream);
+ }
+ }
+ if (stream->error || stream->eof) {
+ ret = stream->error;
+ } else if (ret == -ENOBUFS || ret == -ENOMEM) {
+ io_opaque_budget_wait(op, ret);
+ ret = IOU_RETRY;
+ } else if (!desc.count) {
+ /* Also retain the next pass when the initial issue exhausts a batch. */
+ atomic_set(&stream->collect_ready, 1);
+ ret = issue_flags & IO_URING_F_MULTISHOT ? IOU_REQUEUE : IOU_RETRY;
+ } else {
+ ret = IOU_RETRY;
+ }
+unlock_stream:
+ mutex_unlock(&stream->lock);
+out:
+ if (ret >= 0 && ret != IOU_ISSUE_SKIP_COMPLETE) {
+ io_req_set_res(req, ret, 0);
+ ret = IOU_COMPLETE;
+ }
+ io_ring_submit_unlock(req->ctx, issue_flags);
+ return ret;
+}
+
static int io_opaque_copy(struct io_opaque_data *data, u64 off, u32 length,
void __user *dest)
{
@@ -1679,6 +2261,11 @@ void io_opaque_cleanup(struct io_kiocb *req)
}
if (op->stream->collector == op) {
op->stream->collector = NULL;
+ /*
+ * Native poll setup failure may cancel COLLECT too. As for
+ * explicit cancellation, closing discards undecided capture;
+ * a token is not a promise of recovery after this failure.
+ */
close = !op->stream->eof && !op->stream->closed && !op->stream->error;
}
mutex_unlock(&op->stream->lock);
diff --git a/io_uring/opaque.h b/io_uring/opaque.h
index a7210f4bee5e..4001a4484ff9 100644
--- a/io_uring/opaque.h
+++ b/io_uring/opaque.h
@@ -102,6 +102,7 @@ void io_opaque_stop(struct io_opaque_store *store);
void io_opaque_free(struct io_opaque_store *store);
int io_opaque_recv_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
int io_opaque_recv(struct io_kiocb *req, unsigned int issue_flags);
+__poll_t io_opaque_collect_poll(struct io_kiocb *req);
int io_opaque_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
int io_opaque_issue(struct io_kiocb *req, unsigned int issue_flags);
int io_opaque_send_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
@@ -113,6 +114,11 @@ int io_opaque_cancel(struct io_ring_ctx *ctx, struct io_cancel_data *cd,
bool io_opaque_cancel_all(struct io_ring_ctx *ctx, struct io_uring_task *tctx,
bool cancel_all);
#else
+static inline __poll_t io_opaque_collect_poll(struct io_kiocb *req)
+{
+ return 0;
+}
+
static inline bool io_opaque_cache_init(struct io_ring_ctx *ctx)
{
return false;
diff --git a/io_uring/opaque_internal.h b/io_uring/opaque_internal.h
index f6b372800c76..2d5f69ee8925 100644
--- a/io_uring/opaque_internal.h
+++ b/io_uring/opaque_internal.h
@@ -2,6 +2,7 @@
#ifndef IO_URING_OPAQUE_INTERNAL_H
#define IO_URING_OPAQUE_INTERNAL_H
+#include <linux/atomic.h>
#include <linux/bitmap.h>
#include <linux/bvec.h>
#include <linux/net.h>
@@ -148,6 +149,8 @@ struct io_opaque_vec {
struct io_opaque_stream {
refcount_t refs;
+ /* Consumed only on entry to a collector pass, never by native poll. */
+ atomic_t collect_ready;
/* Extents, range decisions and collector state. */
struct mutex lock;
struct io_opaque_store *store;
diff --git a/io_uring/opdef.c b/io_uring/opdef.c
index a80a56b19581..628f908a66ad 100644
--- a/io_uring/opdef.c
+++ b/io_uring/opdef.c
@@ -592,6 +592,20 @@ const struct io_issue_def io_issue_defs[] = {
.prep = io_uring_cmd_prep,
.issue = io_uring_cmd,
},
+ [IORING_OP_OPAQUE_COLLECT] = {
+ .audit_skip = 1,
+ .ioprio = 1,
+ .needs_file = 1,
+ .unbound_nonreg_file = 1,
+ .pollin = 1,
+#ifdef CONFIG_IO_URING_OPAQUE_OBJ
+ .async_size = sizeof(struct io_opaque_req),
+ .prep = io_opaque_recv_prep,
+ .issue = io_opaque_recv,
+#else
+ .prep = io_eopnotsupp_prep,
+#endif
+ },
[IORING_OP_OPAQUE_OBJ] = {
.audit_skip = 1,
.ioprio = 1,
@@ -875,6 +889,10 @@ const struct io_cold_def io_cold_defs[] = {
.sqe_copy = io_uring_cmd_sqe_copy,
.cleanup = io_uring_cmd_cleanup,
},
+ [IORING_OP_OPAQUE_COLLECT] = {
+ .name = "OPAQUE_COLLECT",
+ .cleanup = io_opaque_cleanup,
+ },
[IORING_OP_OPAQUE_OBJ] = {
.name = "OPAQUE_OBJ",
.cleanup = io_opaque_cleanup,
diff --git a/io_uring/poll.c b/io_uring/poll.c
index 4f04bda42f7f..25a202f3f3ae 100644
--- a/io_uring/poll.c
+++ b/io_uring/poll.c
@@ -9,6 +9,10 @@
#include <linux/hashtable.h>
#include <linux/io_uring.h>
+#ifdef CONFIG_SOCK_RX_OWNER
+#include <net/sock.h>
+#endif
+
#include <trace/events/io_uring.h>
#include <uapi/linux/io_uring.h>
@@ -21,6 +25,7 @@
#include "kbuf.h"
#include "poll.h"
#include "cancel.h"
+#include "opaque.h"
struct io_poll_update {
struct file *file;
@@ -41,6 +46,47 @@ struct io_poll_table {
__poll_t result_mask;
};
+static bool io_poll_rx_owned(struct io_kiocb *req)
+{
+#ifdef CONFIG_SOCK_RX_OWNER
+ struct socket *sock;
+
+ /* Error-queue receives do not encounter the payload-owner refusal. */
+ if (req->flags & REQ_F_CLEAR_POLLIN)
+ return false;
+ switch (req->opcode) {
+ case IORING_OP_READ:
+ case IORING_OP_READV:
+ case IORING_OP_READ_FIXED:
+ case IORING_OP_READV_FIXED:
+ case IORING_OP_READ_MULTISHOT:
+ case IORING_OP_RECV:
+ case IORING_OP_RECVMSG:
+ case IORING_OP_RECV_ZC:
+ break;
+ default:
+ return false;
+ }
+ sock = sock_from_file(req->file);
+ return sock && sock->sk && sock_rx_owned(sock->sk);
+#else
+ return false;
+#endif
+}
+
+static __poll_t io_poll_mask(struct io_kiocb *req, struct poll_table_struct *pt)
+{
+ /* Enroll before checking persistent readiness: a later change must wake us. */
+ __poll_t mask = vfs_poll(req->file, pt);
+
+ if (req->opcode == IORING_OP_OPAQUE_COLLECT)
+ mask |= io_opaque_collect_poll(req);
+ else if (io_poll_rx_owned(req))
+ /* A coalesced claim wake must still issue the real receive handler. */
+ mask |= EPOLLIN | EPOLLRDNORM;
+ return mask;
+}
+
#define IO_POLL_CANCEL_FLAG BIT(31)
#define IO_POLL_RETRY_FLAG BIT(30)
#define IO_POLL_REF_MASK GENMASK(29, 0)
@@ -283,7 +329,7 @@ static int io_poll_check_events(struct io_kiocb *req, io_tw_token_t tw)
__poll_t events = req->apoll_events;
struct poll_table_struct pt = { ._key = events };
- req->cqe.res = vfs_poll(req->file, &pt) & events;
+ req->cqe.res = io_poll_mask(req, &pt) & events;
/*
* We got woken with a mask, but someone else got to
* it first. The above vfs_poll() doesn't add us back
@@ -417,6 +463,14 @@ static int io_poll_wake(struct wait_queue_entry *wait, unsigned mode, int sync,
if (unlikely(mask & POLLFREE))
return io_pollfree_wake(req, poll);
+ if (mask & POLL_RX_OWNER) {
+ if (req->opcode != IORING_OP_OPAQUE_COLLECT &&
+ !io_poll_rx_owned(req))
+ return 0;
+ /* Recheck current ownership/readiness, never cache a private event. */
+ mask = 0;
+ }
+
/* for instances that support it check for an event match first */
if (mask && !(mask & (poll->events & ~IO_ASYNC_POLL_COMMON)))
return 0;
@@ -597,7 +651,7 @@ static int __io_arm_poll_handler(struct io_kiocb *req,
if (poll->events & EPOLLEXCLUSIVE)
req->flags |= REQ_F_POLL_NO_LAZY;
- mask = vfs_poll(req->file, &ipt->pt) & poll->events;
+ mask = io_poll_mask(req, &ipt->pt) & poll->events;
if (unlikely(ipt->error || !ipt->nr_entries)) {
io_poll_remove_entries(req);
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 10/12] io_uring/opaque: add IORING_OPAQUE_CREATE
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (8 preceding siblings ...)
2026-11-10 13:20 ` [RFC PATCH net-next 09/12] io_uring/opaque: add IORING_OP_OPAQUE_COLLECT Alex Markuze
@ 2026-11-10 13:20 ` Alex Markuze
2026-11-10 13:20 ` [RFC PATCH net-next 11/12] io_uring/opaque: add manual and automatic compaction Alex Markuze
2026-11-10 13:20 ` [RFC PATCH net-next 12/12] io_uring/opaque: add basic KUnit tests Alex Markuze
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:20 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Allow an application to populate an immutable object from its own memory,
for example to retain a generated reply for repeated sends. Add CREATE
to IORING_OP_OPAQUE_OBJ; its objects support the same reads, sends and
FREE as received objects.
sqe->addr3 points to the source and len supplies 1 to INT_MAX bytes;
addr and off must be zero. Preparation rejects malformed fields, sizes
beyond the store's object or capture limits, and inaccessible or
overflowing source ranges. Source memory remains valid through execution.
Issue reserves an object slot and page-rounded backing charge, then
copies synchronously under the ring lock. It uses the compaction page
allocator, preferring compound allocations and falling back to smaller
orders. Each allocation owns an extent and part of the reservation.
Copying proceeds by base page and yields between destination allocations.
CREATE fails immediately on budget exhaustion rather than waiting.
Success returns the length, handle and forwarding-only flag; OBJECT_STAT
marks the dense version as compacted. STAT counts the copied bytes. On
failure no handle is published, and cleanup releases the slot, backing
and unused reservation. A large CREATE holds the ring lock through the
copy and can delay other requests of that ring. There is no per-store
pending-request limit.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
io_uring/opaque.c | 129 +++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 127 insertions(+), 2 deletions(-)
diff --git a/io_uring/opaque.c b/io_uring/opaque.c
index 4610e212a800..78ae19a6ad82 100644
--- a/io_uring/opaque.c
+++ b/io_uring/opaque.c
@@ -1884,7 +1884,6 @@ int io_opaque_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
READ_ONCE(sqe->buf_index) || READ_ONCE(sqe->__pad2[0]) ||
cmd > IORING_OPAQUE_AVAILABLE ||
cmd == IORING_OPAQUE_COMPACT || cmd == IORING_OPAQUE_SET_POLICY ||
- cmd == IORING_OPAQUE_CREATE ||
(req->flags & (REQ_F_CQE_SKIP | REQ_F_FIXED_FILE)))
return -EINVAL;
op = io_opaque_req_alloc(req, store);
@@ -1899,6 +1898,21 @@ int io_opaque_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
op->offset = READ_ONCE(sqe->off);
op->length = READ_ONCE(sqe->len);
op->addr = u64_to_user_ptr(READ_ONCE(sqe->addr3));
+ if (cmd == IORING_OPAQUE_CREATE) {
+ u64 pointer = READ_ONCE(sqe->addr3);
+
+ if (op->handle || op->offset || !op->length ||
+ op->length > INT_MAX || !pointer)
+ return -EINVAL;
+ if (op->length > store->config.max_object_size ||
+ ALIGN((u64)op->length, PAGE_SIZE) > io_opaque_rx_limit(store))
+ return -EMSGSIZE;
+ if (check_add_overflow(pointer, (u64)op->length, &end) ||
+ pointer > ULONG_MAX || end > ULONG_MAX ||
+ !access_ok(op->addr, op->length))
+ return -EFAULT;
+ return 0;
+ }
if (check_add_overflow(op->offset, (u64)op->length, &end))
return -EOVERFLOW;
if (cmd <= IORING_OPAQUE_READ_OBJECT || cmd == IORING_OPAQUE_AVAILABLE) {
@@ -2046,12 +2060,110 @@ static int io_opaque_stat(struct io_opaque_req *op)
return io_opaque_stat_copy(op, &stat, sizeof(stat));
}
+/* Construction is synchronous under the ring lock, like READ_OBJECT copies. */
+static int io_opaque_create_build(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct mem_cgroup *old = set_active_memcg(store->memcg);
+ u64 done = 0;
+ int ret = 0;
+
+ while (done < op->length) {
+ u64 pages_left = DIV_ROUND_UP_ULL(op->length - done, PAGE_SIZE);
+ struct io_opaque_extent *extent;
+ unsigned int order;
+ struct page *page;
+ u32 chunk, written = 0;
+
+ page = io_opaque_dest_page(store, pages_left, &order);
+ if (!page) {
+ ret = -ENOMEM;
+ goto out;
+ }
+ chunk = min_t(u64, op->length - done, PAGE_SIZE << order);
+ extent = io_opaque_extent_new(store, page, 0, chunk, done, true);
+ if (IS_ERR(extent)) {
+ put_page(page);
+ ret = PTR_ERR(extent);
+ goto out;
+ }
+ op->create_reserved -= page_size(page);
+ extent->backing->copied = true;
+ list_add_tail(&extent->list, &op->data->extents);
+ while (written < chunk) {
+ u32 bytes = min_t(u32, chunk - written, PAGE_SIZE);
+ unsigned long left;
+ void *mapped;
+
+ mapped = kmap_local_page(page + (written >> PAGE_SHIFT));
+ left = copy_from_user(mapped, op->addr + done + written, bytes);
+ kunmap_local(mapped);
+ atomic64_add(bytes - left, &store->copied);
+ if (left) {
+ ret = -EFAULT;
+ goto out;
+ }
+ written += bytes;
+ }
+ done += chunk;
+ op->data->length = done;
+ cond_resched();
+ }
+ ret = io_opaque_vector(op->data);
+ if (!ret)
+ op->data->dense = true;
+out:
+ set_active_memcg(old);
+ return ret;
+}
+
+static int io_opaque_create(struct io_opaque_req *op, unsigned int issue_flags)
+{
+ struct io_ring_ctx *ctx = op->req->ctx;
+ u64 charge = ALIGN((u64)op->length, PAGE_SIZE);
+ u64 object_flags = 0;
+ struct mem_cgroup *old;
+ int ret;
+
+ io_ring_submit_lock(ctx, issue_flags);
+ if (READ_ONCE(op->store->dead)) {
+ ret = -ESHUTDOWN;
+ goto unlock;
+ }
+ old = set_active_memcg(op->store->memcg);
+ op->data = io_opaque_data_alloc();
+ set_active_memcg(old);
+ if (!op->data) {
+ ret = -ENOMEM;
+ goto unlock;
+ }
+ ret = io_opaque_slot_reserve(op);
+ if (ret)
+ goto unlock;
+ if (!io_opaque_charge(op->store, charge, true)) {
+ ret = -ENOBUFS;
+ goto unlock;
+ }
+ op->create_reserved = charge;
+ ret = io_opaque_create_build(op);
+ if (!ret)
+ ret = io_opaque_publish(op, &object_flags);
+unlock:
+ io_req_set_res32(op->req, ret, 0, ret >= 0 ? op->handle : 0, object_flags);
+ if (ret < 0)
+ req_set_fail(op->req);
+ io_ring_submit_unlock(ctx, issue_flags);
+ return IOU_COMPLETE;
+}
+
int io_opaque_issue(struct io_kiocb *req, unsigned int issue_flags)
{
struct io_opaque_req *op = req->async_data;
struct io_opaque_store *store = op->store;
int ret;
+ if (op->op == IORING_OPAQUE_CREATE)
+ return io_opaque_create(op, issue_flags);
io_ring_submit_lock(req->ctx, issue_flags);
/* Pair with the cancellation search before any opaque wait enrollment. */
if (io_wq_cancel_seen(req, issue_flags)) {
@@ -2224,6 +2336,16 @@ bool io_opaque_cancel_all(struct io_ring_ctx *ctx, struct io_uring_task *tctx,
return found;
}
+static void io_opaque_create_release(struct io_opaque_req *op)
+{
+ io_opaque_data_put(op->store, op->data);
+ op->data = NULL;
+ if (op->create_reserved) {
+ io_opaque_uncharge(op->store, op->create_reserved);
+ op->create_reserved = 0;
+ }
+}
+
void io_opaque_cleanup(struct io_kiocb *req)
{
struct io_opaque_req *op = req->async_data;
@@ -2287,7 +2409,10 @@ void io_opaque_cleanup(struct io_kiocb *req)
}
if (unlikely(op->vec))
io_opaque_vec_cleanup(op);
- io_opaque_data_put(store, op->data);
+ if (op->op == IORING_OPAQUE_CREATE)
+ io_opaque_create_release(op);
+ else
+ io_opaque_data_put(store, op->data);
io_opaque_stream_put(op->stream);
lockdep_assert_held(&req->ctx->uring_lock);
io_opaque_req_free(req->ctx, op);
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 11/12] io_uring/opaque: add manual and automatic compaction
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (9 preceding siblings ...)
2026-11-10 13:20 ` [RFC PATCH net-next 10/12] io_uring/opaque: add IORING_OPAQUE_CREATE Alex Markuze
@ 2026-11-10 13:20 ` Alex Markuze
2026-11-10 13:20 ` [RFC PATCH net-next 12/12] io_uring/opaque: add basic KUnit tests Alex Markuze
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:20 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Sparse retained receive allocations can cost much more backing capacity
than their payload length. Add COMPACT to copy an object into dense
backing and publish the replacement under the same handle. Requests
already using the old version keep their references. A dense object
completes immediately; concurrent compaction of the same object returns
-EBUSY.
SET_POLICY configures automatic compaction with a flag, byte rate, burst
size and per-copy temporary-backing limit. Preparation snapshots the
policy when its SQE is consumed. Updating candidates walks the object
table in chunks of 256, dropping the mutex and yielding between chunks.
Newly published objects see the new policy immediately; a newer update
stops an older walk. Store shutdown ends the update with -ESHUTDOWN.
One delayed work item serializes copies for each store. A run handles up
to eight manual requests and one automatic copy. Automatic selection
examines up to 32 candidates and skips young, forwarding-only and dense
objects. It estimates a nonnegative saving from current charge minus the
page-rounded payload length, considering an estimate of at least a quarter
of the charge or multiple source extents with zero estimated saving.
Selection does not guarantee fewer destination extents or an immediate
reduction in retained charge: allocations can fall back to smaller orders,
and old versions or shared backing may remain pinned. Rate tokens and a
100-ms scan interval bound automatic work. These rules are implementation
heuristics, not ABI guarantees.
Every copy reserves destination capacity under the hard store limit.
Publication requires that the handle still name the source version; if
FREE or LAST removed it, the replacement is discarded. Compaction never
evicts an object. Disabling automatic work preserves accepted manual
requests; shutdown completes them and waits for the work item to finish.
Run reserve/build, publication and accounting settlement in the existing
per-store dispatcher before readiness enqueue. A deferred ring owns only
delivery of the decided result, not active accounting or source pins, so
independent shared-backing copies cannot be excluded by idle ring task work.
Only concurrent compaction of the same slot returns EBUSY.
Prepare whole extent releases before final version zero, including shared
versions and releases started before tagging. Retain backing metadata
through prepared puts, credit source charge after physical release, and
settle credits, unused reservation and transient exclusion coherently.
Keep a private published-replacement pin through the handoff. Predicate
waitqueues drain already-started frees, not live readers; physical frees,
mode updates and wakes occur outside accounting spinlocks.
Failed setup detaches partial tags before returning. Automatic setup
failure refunds tokens within the same policy generation and burst.
Publication and cancellation share a bounded commit decision; committed
copies cannot be retroactively canceled or published twice by task work.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
include/uapi/linux/io_uring/opaque.h | 1 +
io_uring/opaque.c | 363 ++++++++++++++++++++++--
io_uring/opaque.h | 6 +
io_uring/opaque_compact.c | 394 ++++++++++++++++++++++++++-
io_uring/opaque_internal.h | 25 +-
5 files changed, 757 insertions(+), 32 deletions(-)
diff --git a/include/uapi/linux/io_uring/opaque.h b/include/uapi/linux/io_uring/opaque.h
index c04eaed8ce5f..410061906684 100644
--- a/include/uapi/linux/io_uring/opaque.h
+++ b/include/uapi/linux/io_uring/opaque.h
@@ -92,6 +92,7 @@ struct io_uring_opaque_frame {
#define IORING_OPAQUE_OBJECT_F_COMPACTING (1U << 1)
#define IORING_OPAQUE_OBJECT_F_FORWARD_ONLY (1U << 2)
#define IORING_OPAQUE_STREAM_F_EOF (1U << 0)
+#define IORING_OPAQUE_POLICY_F_AUTO_COMPACT (1U << 0)
enum io_uring_opaque_memory_mode {
IORING_OPAQUE_MEMORY_NORMAL,
diff --git a/io_uring/opaque.c b/io_uring/opaque.c
index 78ae19a6ad82..59752db9a041 100644
--- a/io_uring/opaque.c
+++ b/io_uring/opaque.c
@@ -282,17 +282,30 @@ bool io_opaque_charge_transient(struct io_opaque_store *store, u64 bytes, u64 li
return charged;
}
-/* Uncharge and stop excluding transient bytes; the two amounts may differ. */
-void io_opaque_release_transient(struct io_opaque_store *store, u64 uncharge,
- u64 transient)
+static void io_opaque_release_transient_locked(struct io_opaque_store *store,
+ u64 uncharge, u64 transient)
{
- spin_lock(&store->transient_lock);
+ lockdep_assert_held(&store->transient_lock);
write_seqcount_begin(&store->transient_seq);
atomic64_sub(uncharge, &store->bytes);
atomic64_sub(transient, &store->transient);
write_seqcount_end(&store->transient_seq);
+}
+
+static void io_opaque_release_transient_quiet(struct io_opaque_store *store,
+ u64 uncharge, u64 transient)
+{
+ spin_lock(&store->transient_lock);
+ io_opaque_release_transient_locked(store, uncharge, transient);
spin_unlock(&store->transient_lock);
io_opaque_memory_mode(store);
+}
+
+/* Uncharge and stop excluding transient bytes; the two amounts may differ. */
+void io_opaque_release_transient(struct io_opaque_store *store, u64 uncharge,
+ u64 transient)
+{
+ io_opaque_release_transient_quiet(store, uncharge, transient);
if (uncharge)
io_opaque_wake_budget(store);
}
@@ -396,6 +409,157 @@ no_space:
return ERR_PTR(-ENOBUFS);
}
+static void io_opaque_handoff_put(struct io_opaque_handoff *handoff)
+{
+ if (handoff && refcount_dec_and_test(&handoff->refs))
+ kfree(handoff);
+}
+
+/* Wake outside transient_lock, with a reference spanning a racing settler. */
+static void io_opaque_handoff_wake(struct io_opaque_handoff *handoff)
+{
+ if (handoff) {
+ wake_up_all(&handoff->wait);
+ io_opaque_handoff_put(handoff);
+ }
+}
+
+/* A failed setup owns no physical source free: its version pin is still held. */
+static void io_opaque_compact_unreserve(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_handoff *handoff = op->handoff;
+ struct io_opaque_extent *extent;
+
+ if (op->data) {
+ list_for_each_entry(extent, &op->data->extents, list) {
+ struct io_opaque_backing *backing = extent->backing;
+ bool put = false;
+
+ spin_lock(&store->transient_lock);
+ if (backing->handoff == handoff) {
+ handoff->pending -= backing->pending;
+ backing->handoff = NULL;
+ put = true;
+ }
+ spin_unlock(&store->transient_lock);
+ if (put)
+ io_opaque_handoff_put(handoff);
+ }
+ }
+ op->handoff = NULL;
+ io_opaque_handoff_put(handoff);
+}
+
+/* Tag while the source pin still owns every backing, before allocating a copy. */
+int io_opaque_compact_reserve(struct io_opaque_req *op, u64 charge)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_handoff *handoff;
+ struct io_opaque_extent *extent;
+ struct mem_cgroup *old;
+
+ old = set_active_memcg(store->memcg);
+ handoff = kzalloc_obj(*handoff, GFP_KERNEL_ACCOUNT);
+ set_active_memcg(old);
+ if (!handoff)
+ return -ENOMEM;
+ refcount_set(&handoff->refs, 1);
+ init_waitqueue_head(&handoff->wait);
+ handoff->active = true;
+ op->handoff = handoff;
+ if (op->data) {
+ list_for_each_entry(extent, &op->data->extents, list) {
+ struct io_opaque_backing *backing = extent->backing;
+ struct io_opaque_handoff *previous;
+
+ spin_lock(&store->transient_lock);
+ previous = backing->handoff;
+ /* The dispatcher settles each copy before starting the next. */
+ if (previous != handoff) {
+ refcount_inc(&handoff->refs);
+ backing->handoff = handoff;
+ handoff->pending += backing->pending;
+ }
+ spin_unlock(&store->transient_lock);
+ if (previous != handoff)
+ io_opaque_handoff_put(previous);
+ cond_resched();
+ }
+ }
+ if (!io_opaque_charge_transient(store, charge,
+ store->config.hard_limit - store->config.reply_reserve)) {
+ io_opaque_compact_unreserve(op);
+ return -ENOBUFS;
+ }
+ op->transient = charge;
+ op->reserved = charge;
+ return 0;
+}
+
+static void io_opaque_backing_release(struct io_opaque_store *store,
+ struct io_opaque_backing *backing)
+{
+ if (backing->accounted)
+ io_opaque_memcg_uncharge(store, backing->charge);
+ put_page(backing->page);
+}
+
+/*
+ * Finish one prepared put. Its pending count keeps backing metadata alive even
+ * if another put releases the last physical reference before this one returns.
+ */
+static u64 io_opaque_backing_done(struct io_opaque_store *store,
+ struct io_opaque_backing *backing, bool released)
+{
+ struct io_opaque_handoff *handoff, *wake = NULL;
+ u64 charge = released ? backing->charge : 0;
+ bool free;
+
+ spin_lock(&store->transient_lock);
+ handoff = backing->handoff;
+ if (released)
+ backing->released = true;
+ if (handoff && handoff->active) {
+ handoff->released += charge;
+ charge = 0;
+ if (!--handoff->pending) {
+ refcount_inc(&handoff->refs);
+ wake = handoff;
+ }
+ }
+ free = !--backing->pending && backing->released;
+ spin_unlock(&store->transient_lock);
+ io_opaque_handoff_wake(wake);
+ if (free) {
+ io_opaque_handoff_put(handoff);
+ kfree(container_of(backing, struct io_opaque_chunk, backing));
+ }
+ return charge;
+}
+
+/* Register even without a tag: a later copy adopts already-started releases. */
+static void io_opaque_extent_release_begin(struct io_opaque_store *store,
+ struct io_opaque_extent *extent)
+{
+ struct io_opaque_handoff *handoff;
+
+ guard(spinlock)(&store->transient_lock);
+ extent->backing->pending++;
+ handoff = extent->backing->handoff;
+ if (handoff && handoff->active)
+ handoff->pending++;
+}
+
+static void io_opaque_extents_release_begin(struct io_opaque_store *store,
+ struct list_head *list)
+{
+ struct io_opaque_extent *extent;
+
+ list_for_each_entry(extent, list, list)
+ io_opaque_extent_release_begin(store, extent);
+}
+
/*
* Let go of @extent, and of its backing if it was the last to use it. Returns
* the charge of the backing that went, for the caller to give back to the
@@ -405,34 +569,46 @@ no_space:
static u64 io_opaque_extent_release(struct io_opaque_store *store, struct io_opaque_extent *extent)
{
struct io_opaque_backing *backing = extent->backing;
- u64 charge = 0;
+ bool last = false;
list_del(&extent->list);
if (!extent->embedded)
kfree(extent);
atomic_dec(&store->extents);
- if (refcount_dec_and_test(&backing->refs)) {
- charge = backing->charge;
-
- if (backing->accounted)
- io_opaque_memcg_uncharge(store, charge);
- put_page(backing->page);
- kfree(container_of(backing, struct io_opaque_chunk, backing));
+ if (!refcount_dec_not_one(&backing->refs)) {
+ spin_lock(&store->transient_lock);
+ last = refcount_dec_and_test(&backing->refs);
+ spin_unlock(&store->transient_lock);
}
- return charge;
+ if (last)
+ io_opaque_backing_release(store, backing);
+ return io_opaque_backing_done(store, backing, last);
}
static void io_opaque_extent_free(struct io_opaque_store *store,
struct io_opaque_extent *extent)
{
- u64 charge = io_opaque_extent_release(store, extent);
+ u64 charge;
+ io_opaque_extent_release_begin(store, extent);
+ charge = io_opaque_extent_release(store, extent);
if (charge)
io_opaque_uncharge(store, charge);
else
io_opaque_wake_budget(store);
}
+/* Release physical backing, retaining its store reservation until the caller settles it. */
+static u64 io_opaque_extents_release(struct io_opaque_store *store, struct list_head *list)
+{
+ struct io_opaque_extent *extent, *next;
+ u64 charge = 0;
+
+ list_for_each_entry_safe(extent, next, list, list)
+ charge += io_opaque_extent_release(store, extent);
+ return charge;
+}
+
/*
* Free the extents of @list. What they held is returned to the store in one
* step, and the waiters for budget are looked at in one go: each look takes
@@ -444,14 +620,11 @@ static void io_opaque_extent_free(struct io_opaque_store *store,
static void io_opaque_extents_free(struct io_opaque_store *store,
struct list_head *list)
{
- struct io_opaque_extent *extent, *next;
- bool any = false;
- u64 charge = 0;
+ bool any = !list_empty(list);
+ u64 charge;
- list_for_each_entry_safe(extent, next, list, list) {
- charge += io_opaque_extent_release(store, extent);
- any = true;
- }
+ io_opaque_extents_release_begin(store, list);
+ charge = io_opaque_extents_release(store, list);
if (charge)
io_opaque_uncharge_quiet(store, charge);
if (any)
@@ -469,14 +642,122 @@ struct io_opaque_data *io_opaque_data_alloc(void)
return data;
}
+static u64 io_opaque_data_free(struct io_opaque_store *store,
+ struct io_opaque_data *data, bool *extents)
+{
+ u64 charge;
+
+ *extents |= !list_empty(&data->extents);
+ charge = io_opaque_extents_release(store, &data->extents);
+ kvfree(data->bvec);
+ kfree(data);
+ return charge;
+}
+
+static u64 io_opaque_data_release(struct io_opaque_store *store,
+ struct io_opaque_data *data, bool *extents)
+{
+ bool last;
+
+retry:
+ if (!data || refcount_dec_not_one(&data->refs))
+ return 0;
+ /* Announce the entire underway release before exposing version ref zero. */
+ io_opaque_extents_release_begin(store, &data->extents);
+ /* Serialize potential final drops with the published copy's private pin. */
+ spin_lock(&store->transient_lock);
+ last = refcount_dec_if_one(&data->refs);
+ spin_unlock(&store->transient_lock);
+ if (last) {
+ smp_acquire__after_ctrl_dep();
+ return io_opaque_data_free(store, data, extents);
+ }
+ /* A reference acquired during preparation keeps the version live. */
+ {
+ struct io_opaque_extent *extent;
+
+ list_for_each_entry(extent, &data->extents, list)
+ io_opaque_backing_done(store, extent->backing, false);
+ }
+ goto retry;
+}
+
void io_opaque_data_put(struct io_opaque_store *store,
struct io_opaque_data *data)
{
- if (data && refcount_dec_and_test(&data->refs)) {
- io_opaque_extents_free(store, &data->extents);
- kvfree(data->bvec);
- kfree(data);
+ bool extents = false;
+ u64 charge = io_opaque_data_release(store, data, &extents);
+
+ if (charge)
+ io_opaque_uncharge_quiet(store, charge);
+ if (extents)
+ io_opaque_wake_budget_all(store);
+}
+
+static u64 io_opaque_compact_source_release(struct io_opaque_req *op,
+ bool *extents)
+{
+ u64 charge;
+
+ charge = io_opaque_data_release(op->store, op->data, extents);
+ op->data = NULL;
+ return charge;
+}
+
+static bool io_opaque_handoff_idle(struct io_opaque_handoff *handoff)
+{
+ return !READ_ONCE(handoff->pending);
+}
+
+static void io_opaque_handoff_wait(struct io_opaque_handoff *handoff)
+{
+ wait_event(handoff->wait, io_opaque_handoff_idle(handoff));
+}
+
+/* Settle source credits, replacement ownership and its exclusion together. */
+void io_opaque_compact_end(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_handoff *handoff = op->handoff;
+ bool extents = false;
+ u64 charge = op->reserved;
+
+ charge += io_opaque_compact_source_release(op, &extents);
+ charge += io_opaque_data_release(store, op->replacement, &extents);
+ op->replacement = NULL;
+ for (;;) {
+ struct io_opaque_data *published = op->published;
+
+ spin_lock(&store->transient_lock);
+ if (handoff && handoff->pending) {
+ spin_unlock(&store->transient_lock);
+ io_opaque_handoff_wait(handoff);
+ continue;
+ }
+ if (published) {
+ op->published = NULL;
+ if (!refcount_dec_not_one(&published->refs)) {
+ spin_unlock(&store->transient_lock);
+ charge += io_opaque_data_release(store, published, &extents);
+ continue;
+ }
+ }
+ if (handoff) {
+ charge += handoff->released;
+ handoff->active = false;
+ }
+ io_opaque_release_transient_locked(store, charge, op->transient);
+ spin_unlock(&store->transient_lock);
+ break;
}
+ if (charge || op->transient)
+ io_opaque_memory_mode(store);
+ op->handoff = NULL;
+ io_opaque_handoff_put(handoff);
+ op->reserved = 0;
+ op->transient = 0;
+ if (charge || extents)
+ io_opaque_wake_budget_all(store);
}
/*
@@ -705,6 +986,7 @@ struct io_opaque_store *io_opaque_alloc(struct io_ring_ctx *ctx,
for (i = 0; i < cfg.max_objects; i++)
INIT_LIST_HEAD(&store->objects[i].candidate);
INIT_DELAYED_WORK(&store->retry_work, io_opaque_retry_work);
+ INIT_DELAYED_WORK(&store->compact_work, io_opaque_compact_work);
return store;
free_store:
bitmap_free(store->stream_used);
@@ -729,6 +1011,7 @@ void io_opaque_stop(struct io_opaque_store *store)
return;
WRITE_ONCE(store->dead, 1);
}
+ flush_delayed_work(&store->compact_work);
cancel_delayed_work_sync(&store->retry_work);
for (i = 0; i < store->config.max_streams; i++) {
struct io_opaque_stream *stream;
@@ -777,6 +1060,8 @@ void io_opaque_free(struct io_opaque_store *store)
if (!store)
return;
io_opaque_stop(store);
+ /* The dispatcher can requeue itself while stop flushes it. */
+ cancel_delayed_work_sync(&store->compact_work);
WARN_ON_ONCE(atomic64_read(&store->bytes));
WARN_ON_ONCE(atomic_read(&store->extents));
io_opaque_txs_free(store);
@@ -1817,6 +2102,12 @@ static int io_opaque_publish(struct io_opaque_req *op, u64 *object_flags)
op->slot_reserved = false;
slot->born = jiffies;
slot->retry = jiffies;
+ if (!slot->data->dense && !slot->data->forward_only &&
+ (store->policy.flags & IORING_OPAQUE_POLICY_F_AUTO_COMPACT)) {
+ list_add_tail(&slot->candidate, &store->candidates);
+ queue_delayed_work(system_dfl_wq, &store->compact_work,
+ msecs_to_jiffies(100));
+ }
return op->length;
}
@@ -1831,7 +2122,8 @@ static void io_opaque_ready(struct io_tw_req tw_req, io_tw_token_t tw)
io_tw_lock(req->ctx, tw);
list_del_init(&op->cancel);
op->phase = IO_OPAQUE_IDLE;
- if (tw.cancel || READ_ONCE(op->canceled) || READ_ONCE(store->dead))
+ if ((tw.cancel || READ_ONCE(op->canceled) || READ_ONCE(store->dead)) &&
+ !(op->op == IORING_OPAQUE_COMPACT && op->copy_committed))
ret = -ECANCELED;
if (ret >= 0 && op->op == IORING_OPAQUE_INSPECT)
ret = io_opaque_copy(op->data, 0, op->data->length, op->addr);
@@ -1883,7 +2175,6 @@ int io_opaque_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
if (READ_ONCE(sqe->fd) != -1 || READ_ONCE(sqe->rw_flags) ||
READ_ONCE(sqe->buf_index) || READ_ONCE(sqe->__pad2[0]) ||
cmd > IORING_OPAQUE_AVAILABLE ||
- cmd == IORING_OPAQUE_COMPACT || cmd == IORING_OPAQUE_SET_POLICY ||
(req->flags & (REQ_F_CQE_SKIP | REQ_F_FIXED_FILE)))
return -EINVAL;
op = io_opaque_req_alloc(req, store);
@@ -1931,14 +2222,17 @@ int io_opaque_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
} else if (op->offset) {
return -EINVAL;
}
- if (cmd == IORING_OPAQUE_FREE || cmd == IORING_OPAQUE_STREAM_CLOSE) {
+ if (cmd == IORING_OPAQUE_FREE || cmd == IORING_OPAQUE_STREAM_CLOSE ||
+ cmd == IORING_OPAQUE_COMPACT) {
if (op->length || op->addr)
return -EINVAL;
}
- if (cmd == IORING_OPAQUE_STAT) {
+ if (cmd == IORING_OPAQUE_STAT || cmd == IORING_OPAQUE_SET_POLICY) {
if (op->handle || !op->addr)
return -EINVAL;
}
+ if (cmd == IORING_OPAQUE_SET_POLICY)
+ return io_opaque_policy_prep(op);
return 0;
}
@@ -2251,6 +2545,12 @@ int io_opaque_issue(struct io_kiocb *req, unsigned int issue_flags)
ret = -ESTALE;
}
break;
+ case IORING_OPAQUE_COMPACT:
+ ret = io_opaque_compact_start(op);
+ break;
+ case IORING_OPAQUE_SET_POLICY:
+ ret = io_opaque_set_policy(op);
+ break;
default:
ret = -EINVAL;
}
@@ -2274,7 +2574,8 @@ static bool io_opaque_cancel_req(struct io_opaque_req *op)
mutex_lock(&op->stream->lock);
spin_lock(&op->store->wait_lock);
phase = op->phase;
- if (op->canceled || phase == IO_OPAQUE_IDLE) {
+ if (op->canceled || phase == IO_OPAQUE_IDLE ||
+ (op->op == IORING_OPAQUE_COMPACT && op->copy_committed)) {
spin_unlock(&op->store->wait_lock);
if (op->stream)
mutex_unlock(&op->stream->lock);
@@ -2411,6 +2712,8 @@ void io_opaque_cleanup(struct io_kiocb *req)
io_opaque_vec_cleanup(op);
if (op->op == IORING_OPAQUE_CREATE)
io_opaque_create_release(op);
+ else if (op->op == IORING_OPAQUE_COMPACT)
+ io_opaque_compact_end(op);
else
io_opaque_data_put(store, op->data);
io_opaque_stream_put(op->stream);
diff --git a/io_uring/opaque.h b/io_uring/opaque.h
index 4001a4484ff9..f63729c19af5 100644
--- a/io_uring/opaque.h
+++ b/io_uring/opaque.h
@@ -14,6 +14,7 @@ struct io_uring_opaque_config;
#ifdef CONFIG_IO_URING_OPAQUE_OBJ
struct io_opaque_stream;
struct io_opaque_data;
+struct io_opaque_handoff;
struct io_opaque_tx;
struct io_opaque_vec;
@@ -74,6 +75,11 @@ struct io_opaque_req {
};
struct {
struct io_opaque_data *replacement;
+ /* Private pin retained after the table takes the replacement. */
+ struct io_opaque_data *published;
+ struct io_opaque_handoff *handoff;
+ bool copy_committed;
+ bool copy_reserved;
u64 reserved;
u64 transient;
};
diff --git a/io_uring/opaque_compact.c b/io_uring/opaque_compact.c
index b59c7cb95d42..5bde1701f189 100644
--- a/io_uring/opaque_compact.c
+++ b/io_uring/opaque_compact.c
@@ -11,6 +11,48 @@
#include "io_uring.h"
#include "opaque_internal.h"
+static void io_opaque_auto_compact(struct io_opaque_store *store);
+
+struct io_opaque_cursor {
+ struct io_opaque_extent *extent;
+ u32 offset;
+};
+
+static int io_opaque_compact_copy(struct io_opaque_req *op,
+ struct io_opaque_cursor *cursor,
+ struct page *page, u32 length)
+{
+ u32 written = 0;
+
+ while (written < length) {
+ struct io_opaque_extent *src = cursor->extent;
+ u32 pos = src->offset + cursor->offset;
+ u32 in = offset_in_page(pos), out = offset_in_page(written);
+ u32 chunk = min(length - written, src->length - cursor->offset);
+ void *from, *to;
+
+ if (READ_ONCE(op->canceled) || READ_ONCE(op->store->dead))
+ return -ECANCELED;
+ chunk = min_t(u32, chunk, PAGE_SIZE - in);
+ chunk = min_t(u32, chunk, PAGE_SIZE - out);
+ to = kmap_local_page(page + (written >> PAGE_SHIFT));
+ from = kmap_local_page(src->backing->page + (pos >> PAGE_SHIFT));
+ memcpy(to + out, from + in, chunk);
+ kunmap_local(from);
+ kunmap_local(to);
+ written += chunk;
+ cursor->offset += chunk;
+ if (cursor->offset == src->length) {
+ cursor->offset = 0;
+ if (!list_is_last(&src->list, &op->data->extents))
+ cursor->extent = list_next_entry(src, list);
+ else if (written < length)
+ return -EIO;
+ }
+ }
+ return 0;
+}
+
struct page *io_opaque_owned_page(unsigned int *order)
{
struct page *page;
@@ -24,10 +66,360 @@ struct page *io_opaque_owned_page(unsigned int *order)
}
}
-/* Destination allocation for dense copies and CREATE. */
+/*
+ * Destination chunk for compaction and CREATE: the largest order within the
+ * remaining size, falling back to smaller orders.
+ */
struct page *io_opaque_dest_page(struct io_opaque_store *store, u64 pages_left,
unsigned int *order)
{
*order = min_t(unsigned int, IO_OPAQUE_MAX_ORDER, ilog2(pages_left));
return io_opaque_owned_page(order);
}
+
+static int io_opaque_compact_build(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_cursor cursor;
+ u64 done = 0, length, charge;
+ struct mem_cgroup *old;
+ int ret = 0;
+
+ length = op->data->length;
+ charge = ALIGN(length, PAGE_SIZE);
+ if (READ_ONCE(op->canceled) || READ_ONCE(store->dead))
+ return -ECANCELED;
+ ret = io_opaque_compact_reserve(op, charge);
+ if (ret)
+ return ret;
+ op->copy_reserved = true;
+ old = set_active_memcg(store->memcg);
+ op->replacement = io_opaque_data_alloc();
+ if (!op->replacement) {
+ ret = -ENOMEM;
+ goto out;
+ }
+ op->replacement->forward_only = op->data->forward_only;
+ cursor.extent = list_first_entry(&op->data->extents, struct io_opaque_extent, list);
+ cursor.offset = 0;
+ while (done < length) {
+ u64 pages_left = DIV_ROUND_UP_ULL(length - done, PAGE_SIZE);
+ struct io_opaque_extent *extent;
+ unsigned int order;
+ struct page *page;
+ u32 chunk;
+
+ if (READ_ONCE(op->canceled) || READ_ONCE(store->dead)) {
+ ret = -ECANCELED;
+ goto out;
+ }
+ page = io_opaque_dest_page(store, pages_left, &order);
+ if (!page) {
+ ret = -ENOMEM;
+ goto out;
+ }
+ chunk = min_t(u64, length - done, PAGE_SIZE << order);
+ extent = io_opaque_extent_new(store, page, 0, chunk, done, true);
+ if (IS_ERR(extent)) {
+ put_page(page);
+ ret = PTR_ERR(extent);
+ goto out;
+ }
+ op->reserved -= page_size(page);
+ list_add_tail(&extent->list, &op->replacement->extents);
+ ret = io_opaque_compact_copy(op, &cursor, page, chunk);
+ if (ret)
+ goto out;
+ done += chunk;
+ op->replacement->length = done;
+ cond_resched();
+ }
+ ret = io_opaque_vector(op->replacement);
+ if (!ret)
+ op->replacement->dense = true;
+out:
+ set_active_memcg(old);
+ return ret;
+}
+
+int io_opaque_compact_publish(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_slot *slot;
+ struct io_opaque_data *old;
+ int ret = -ESTALE;
+
+ mutex_lock(&store->tables);
+ slot = io_opaque_lookup(store, op->handle);
+ /* Cancellation shares this bounded publication decision, not the frees. */
+ spin_lock(&store->wait_lock);
+ if (slot && slot->data == op->data && !store->dead &&
+ !READ_ONCE(op->canceled)) {
+ old = slot->data;
+ op->published = op->replacement;
+ refcount_inc(&op->published->refs);
+ slot->data = op->replacement;
+ op->replacement = NULL;
+ op->copy_committed = true;
+ slot->compacting = false;
+ list_del_init(&slot->candidate);
+ atomic64_add(old->length, &store->compacted);
+ atomic64_inc(&store->compactions);
+ ret = 0;
+ } else {
+ old = NULL;
+ }
+ spin_unlock(&store->wait_lock);
+ mutex_unlock(&store->tables);
+ io_opaque_data_put(store, old);
+ return ret;
+}
+
+/* Complete accounting in the existing dispatcher, never in deferred ring work. */
+static int io_opaque_compact_run(struct io_opaque_req *op, bool automatic)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_slot *slot;
+ int ret = io_opaque_compact_build(op);
+
+ if (!ret)
+ ret = io_opaque_compact_publish(op);
+ mutex_lock(&store->tables);
+ slot = io_opaque_lookup(store, op->handle);
+ if (slot && slot->data == op->data) {
+ slot->compacting = false;
+ if (automatic && ret)
+ slot->retry = jiffies + HZ;
+ }
+ mutex_unlock(&store->tables);
+ io_opaque_compact_end(op);
+ return ret;
+}
+
+void io_opaque_compact_work(struct work_struct *work)
+{
+ struct io_opaque_store *store = container_of(to_delayed_work(work),
+ struct io_opaque_store, compact_work);
+ unsigned long delay = MAX_JIFFY_OFFSET;
+ unsigned long now;
+ unsigned int i;
+
+ /* Once shutdown is observed, drain every accepted request in this run. */
+ for (i = 0; i < 8 || READ_ONCE(store->dead); i++) {
+ struct io_opaque_req *op;
+ int ret;
+
+ scoped_guard(spinlock, &store->wait_lock) {
+ op = list_first_entry_or_null(&store->copies, struct io_opaque_req, wait);
+ if (op)
+ list_del_init(&op->wait);
+ }
+ if (!op)
+ break;
+ ret = io_opaque_compact_run(op, false);
+ /* Cancellation cannot complete a request still owned by this worker. */
+ io_opaque_queue_ready(op, ret);
+ }
+ io_opaque_auto_compact(store);
+ /* Keep automatic scans paced even when manual requests wake us early. */
+ guard(mutex)(&store->tables);
+ now = jiffies;
+ if (!store->dead && (store->policy.flags & IORING_OPAQUE_POLICY_F_AUTO_COMPACT) &&
+ !list_empty(&store->candidates))
+ delay = time_before(now, store->auto_at) ? store->auto_at - now : 0;
+ guard(spinlock)(&store->wait_lock);
+ if (!list_empty(&store->copies))
+ delay = 0;
+ if (delay != MAX_JIFFY_OFFSET)
+ queue_delayed_work(system_dfl_wq, &store->compact_work, delay);
+}
+
+int io_opaque_compact_start(struct io_opaque_req *op)
+{
+ struct io_opaque_store *store = op->store;
+ struct io_opaque_slot *slot;
+
+ guard(mutex)(&store->tables);
+ if (store->dead)
+ return -ESHUTDOWN;
+ slot = io_opaque_lookup(store, op->handle);
+ if (!slot || !slot->data)
+ return -ESTALE;
+ if (slot->compacting)
+ return -EBUSY;
+ if (slot->data->dense)
+ return 0;
+ op->data = slot->data;
+ refcount_inc(&op->data->refs);
+ slot->compacting = true;
+ io_opaque_wait(op, IO_OPAQUE_COPY_WORK);
+ scoped_guard(spinlock, &store->wait_lock)
+ list_add_tail(&op->wait, &store->copies);
+ mod_delayed_work(system_dfl_wq, &store->compact_work, 0);
+ return IOU_ISSUE_SKIP_COMPLETE;
+}
+
+/*
+ * Install @policy and mark the objects it concerns as candidates, or not. Where
+ * the table of handles is long this is the longest thing done under tables, so
+ * the walk drops it every IO_OPAQUE_POLICY_CHUNK slots and those waiting
+ * for it wait for a chunk. What happens between chunks cannot hurt it. The
+ * policy is in force from the first one, so an object published meanwhile joins
+ * the candidates by itself, or does not, and one freed leaves them; the walk
+ * adds a slot only if it is not on the list. A second change that starts
+ * meanwhile walks the whole table itself, so the first gives up when it comes
+ * back and finds it was overtaken, as it does if the store has stopped.
+ */
+static int io_opaque_apply_policy(struct io_opaque_store *store,
+ const struct io_uring_opaque_policy *policy)
+{
+ bool automatic = policy->flags & IORING_OPAQUE_POLICY_F_AUTO_COMPACT;
+ u32 i, seq;
+
+ mutex_lock(&store->tables);
+ if (store->dead) {
+ mutex_unlock(&store->tables);
+ return -ESHUTDOWN;
+ }
+ store->policy = *policy;
+ store->tokens = policy->burst_bytes;
+ store->token_time = jiffies;
+ store->auto_at = jiffies;
+ seq = ++store->policy_seq;
+ for (i = 0; i < store->config.max_objects; i++) {
+ struct io_opaque_slot *slot = &store->objects[i];
+
+ if (automatic && slot->data && !slot->data->dense && !slot->data->forward_only &&
+ list_empty(&slot->candidate)) {
+ list_add_tail(&slot->candidate, &store->candidates);
+ slot->retry = jiffies;
+ } else if (!automatic) {
+ list_del_init(&slot->candidate);
+ }
+ if ((i + 1) % IO_OPAQUE_POLICY_CHUNK || i + 1 == store->config.max_objects)
+ continue;
+ mutex_unlock(&store->tables);
+ cond_resched();
+ mutex_lock(&store->tables);
+ if (store->dead) {
+ mutex_unlock(&store->tables);
+ return -ESHUTDOWN;
+ }
+ if (store->policy_seq != seq) {
+ mutex_unlock(&store->tables);
+ return 0;
+ }
+ }
+ if (automatic) {
+ mod_delayed_work(system_dfl_wq, &store->compact_work, 0);
+ } else {
+ guard(spinlock)(&store->wait_lock);
+ /* Policy changes cannot cancel an accepted manual request. */
+ if (list_empty(&store->copies))
+ cancel_delayed_work(&store->compact_work);
+ }
+ mutex_unlock(&store->tables);
+ return 0;
+}
+
+/* Copy and check the policy at prep, so its struct need not outlive submission. */
+int io_opaque_policy_prep(struct io_opaque_req *op)
+{
+ struct io_uring_opaque_policy *policy = &op->policy;
+ u64 hard_limit = op->store->config.hard_limit;
+ int ret;
+
+ if (op->length < offsetof(struct io_uring_opaque_policy, __resv) ||
+ op->length > PAGE_SIZE)
+ return -EINVAL;
+ ret = copy_struct_from_user(policy, sizeof(*policy), op->addr, op->length);
+ if (ret)
+ return ret;
+ if ((policy->flags & ~IORING_OPAQUE_POLICY_F_AUTO_COMPACT) ||
+ policy->__resv0 || !mem_is_zero(policy->__resv, sizeof(policy->__resv)) ||
+ policy->bytes_per_second > SZ_1T || policy->burst_bytes > hard_limit ||
+ policy->max_temporary_bytes > hard_limit)
+ return -EINVAL;
+ if ((policy->flags & IORING_OPAQUE_POLICY_F_AUTO_COMPACT) &&
+ (!policy->bytes_per_second || !policy->burst_bytes ||
+ policy->max_temporary_bytes < PAGE_SIZE))
+ return -EINVAL;
+ return 0;
+}
+
+int io_opaque_set_policy(struct io_opaque_req *op)
+{
+ return io_opaque_apply_policy(op->store, &op->policy);
+}
+
+/* The same dispatcher handles bounded scans and the store-wide token bucket. */
+static void io_opaque_auto_compact(struct io_opaque_store *store)
+{
+ struct io_opaque_req op = { .store = store };
+ struct io_opaque_slot *slot = NULL;
+ struct io_uring_opaque_policy policy;
+ u64 minted, elapsed;
+ u64 length = 0;
+ u32 policy_seq;
+ unsigned int scanned = 0;
+ int ret;
+
+ mutex_lock(&store->tables);
+ policy = store->policy;
+ policy_seq = store->policy_seq;
+ if (store->dead || !(policy.flags & IORING_OPAQUE_POLICY_F_AUTO_COMPACT) ||
+ time_before(jiffies, store->auto_at))
+ goto unlock;
+ store->auto_at = jiffies + msecs_to_jiffies(100);
+ elapsed = jiffies - store->token_time;
+ if (check_mul_overflow(elapsed, policy.bytes_per_second, &minted))
+ minted = policy.burst_bytes;
+ else
+ minted = div_u64(minted, HZ);
+ store->tokens = min(policy.burst_bytes, store->tokens + min(minted, policy.burst_bytes));
+ store->token_time = jiffies;
+ while (!list_empty(&store->candidates) && scanned++ < 32) {
+ u64 dense, saving;
+
+ slot = list_first_entry(&store->candidates, struct io_opaque_slot, candidate);
+ list_move_tail(&slot->candidate, &store->candidates);
+ if (!slot->data || slot->data->dense) {
+ list_del_init(&slot->candidate);
+ continue;
+ }
+ if (slot->compacting || time_before(jiffies, slot->retry) ||
+ time_before(jiffies, slot->born + msecs_to_jiffies(100)))
+ continue;
+ dense = ALIGN(slot->data->length, PAGE_SIZE);
+ saving = slot->data->charge > dense ? slot->data->charge - dense : 0;
+ /* Selection heuristics are implementation details, not UAPI. */
+ if ((!saving && slot->data->nr == 1) ||
+ (saving && saving * 100 < slot->data->charge * 25) ||
+ dense > policy.max_temporary_bytes || slot->data->length > policy.burst_bytes) {
+ /* Immutable backing cannot become eligible without a new policy. */
+ list_del_init(&slot->candidate);
+ continue;
+ }
+ if (slot->data->length > store->tokens)
+ continue;
+ slot->compacting = true;
+ store->tokens -= slot->data->length;
+ op.data = slot->data;
+ length = op.data->length;
+ refcount_inc(&op.data->refs);
+ op.handle = io_opaque_handle(slot - store->objects, slot->generation);
+ break;
+ }
+unlock:
+ mutex_unlock(&store->tables);
+ if (op.data) {
+ ret = io_opaque_compact_run(&op, true);
+ /* No reservation means no copy was attempted; return its tokens. */
+ if (ret && !op.copy_reserved) {
+ guard(mutex)(&store->tables);
+
+ if (store->policy_seq == policy_seq)
+ store->tokens = min(policy.burst_bytes, store->tokens + length);
+ }
+ }
+}
diff --git a/io_uring/opaque_internal.h b/io_uring/opaque_internal.h
index 2d5f69ee8925..f27c30d32e13 100644
--- a/io_uring/opaque_internal.h
+++ b/io_uring/opaque_internal.h
@@ -7,6 +7,7 @@
#include <linux/bvec.h>
#include <linux/net.h>
#include <linux/sizes.h>
+#include <linux/wait.h>
#include <linux/workqueue.h>
#include <uapi/linux/io_uring/opaque.h>
@@ -44,8 +45,22 @@
/* The head of its queue, still to take its references. */
#define IO_OPAQUE_TX_SETTLE 2
+/* One copy's deferred source credits; mutable fields use transient_lock. */
+struct io_opaque_handoff {
+ refcount_t refs;
+ u64 released;
+ unsigned int pending;
+ bool active;
+ wait_queue_head_t wait;
+};
+
struct io_opaque_backing {
refcount_t refs;
+ /* Owns a handoff reference until retagged or physically released. */
+ struct io_opaque_handoff *handoff;
+ /* Prepared extent puts keep metadata alive through physical release. */
+ unsigned int pending;
+ bool released;
struct page *page;
u64 charge;
u32 filled;
@@ -214,7 +229,7 @@ struct io_opaque_store {
* that move bytes and transient together are one seqcount write.
*/
atomic64_t transient;
- /* Serializes transient_seq writers. */
+ /* Transient writers, backing handoffs and potential last-version drops. */
spinlock_t transient_lock;
seqcount_spinlock_t transient_seq;
atomic_t extents;
@@ -269,4 +284,12 @@ struct io_opaque_slot *io_opaque_lookup(struct io_opaque_store *store,
u64 handle);
void io_opaque_wait(struct io_opaque_req *op, enum io_opaque_phase phase);
void io_opaque_queue_ready(struct io_opaque_req *op, int result);
+int io_opaque_compact_reserve(struct io_opaque_req *op, u64 charge);
+int io_opaque_compact_publish(struct io_opaque_req *op);
+void io_opaque_compact_end(struct io_opaque_req *op);
+void io_opaque_compact_work(struct work_struct *work);
+int io_opaque_compact_start(struct io_opaque_req *op);
+int io_opaque_policy_prep(struct io_opaque_req *op);
+int io_opaque_set_policy(struct io_opaque_req *op);
+
#endif
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread* [RFC PATCH net-next 12/12] io_uring/opaque: add basic KUnit tests
2026-11-10 13:19 [RFC PATCH net-next 00/12] io_uring: opaque TCP payload objects Alex Markuze
` (10 preceding siblings ...)
2026-11-10 13:20 ` [RFC PATCH net-next 11/12] io_uring/opaque: add manual and automatic compaction Alex Markuze
@ 2026-11-10 13:20 ` Alex Markuze
11 siblings, 0 replies; 13+ messages in thread
From: Alex Markuze @ 2026-11-10 13:20 UTC (permalink / raw)
To: Jens Axboe, David S . Miller, Eric Dumazet, Jakub Kicinski, Paolo Abeni
Cc: Pavel Begunkov, Simon Horman, Kuniyuki Iwashima,
Willem de Bruijn, Neal Cardwell, John Fastabend, Jakub Sitnicki,
Jiayuan Chen, io-uring, netdev, bpf, linux-kernel
Check receive-budget admission and watermark transitions, then build a
page-backed immutable version and check its vector, payload, held version
reference and final backing/reference release.
Use existing internal helpers without adding production test hooks. These
are basic helper checks; the matching liburing example exercises a real
ring and loopback TCP connection.
Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
io_uring/Kconfig | 10 ++++
io_uring/Makefile | 1 +
io_uring/opaque_test.c | 128 +++++++++++++++++++++++++++++++++++++++++
3 files changed, 139 insertions(+)
create mode 100644 io_uring/opaque_test.c
diff --git a/io_uring/Kconfig b/io_uring/Kconfig
index 5b696cf266f6..db6e3c5a317d 100644
--- a/io_uring/Kconfig
+++ b/io_uring/Kconfig
@@ -30,3 +30,13 @@ config IO_URING_BPF_OPS
def_bool y
depends on IO_URING
depends on BPF_SYSCALL && BPF_JIT && DEBUG_INFO_BTF
+
+config IO_URING_OPAQUE_OBJ_KUNIT_TEST
+ bool "KUnit smoke tests for opaque objects" if !KUNIT_ALL_TESTS
+ depends on KUNIT=y && IO_URING_OPAQUE_OBJ
+ default KUNIT_ALL_TESTS
+ help
+ Check receive-budget admission and watermark transitions, and the
+ vector representation and reference lifetime of an immutable version.
+ The tests call internal helpers without a userspace ring or TCP socket.
+ They are a small smoke test, not a pressure or concurrency suite.
diff --git a/io_uring/Makefile b/io_uring/Makefile
index 912b1861c0a5..b7fcc2d3c714 100644
--- a/io_uring/Makefile
+++ b/io_uring/Makefile
@@ -18,6 +18,7 @@ obj-$(CONFIG_IO_URING) += io_uring.o opdef.o kbuf.o rsrc.o notif.o \
obj-$(CONFIG_IO_URING_ZCRX) += zcrx.o
obj-$(CONFIG_IO_URING_OPAQUE_OBJ) += opaque.o opaque_register.o opaque_compact.o
+obj-$(CONFIG_IO_URING_OPAQUE_OBJ_KUNIT_TEST) += opaque_test.o
obj-$(CONFIG_IO_WQ) += io-wq.o
obj-$(CONFIG_FUTEX) += futex.o
obj-$(CONFIG_EPOLL) += epoll.o
diff --git a/io_uring/opaque_test.c b/io_uring/opaque_test.c
new file mode 100644
index 000000000000..cba5829de05f
--- /dev/null
+++ b/io_uring/opaque_test.c
@@ -0,0 +1,128 @@
+// SPDX-License-Identifier: GPL-2.0
+#include <kunit/test.h>
+#include <linux/highmem.h>
+
+#include "opaque_internal.h"
+
+struct opaque_basic {
+ struct io_opaque_store *store;
+ struct io_opaque_data *data;
+ struct page *page;
+};
+
+static void opaque_basic_release(void *arg)
+{
+ struct opaque_basic *c = arg;
+
+ if (!IS_ERR_OR_NULL(c->store)) {
+ io_opaque_data_put(c->store, c->data);
+ /* Also release a budget assertion's unfinished reservation. */
+ if (atomic64_read(&c->store->bytes) > 0)
+ io_opaque_uncharge(c->store, atomic64_read(&c->store->bytes));
+ io_opaque_free(c->store);
+ }
+ if (c->page)
+ put_page(c->page);
+}
+
+static struct opaque_basic *opaque_basic_new(struct kunit *test)
+{
+ struct io_uring_opaque_config config = {
+ .hard_limit = 8 * PAGE_SIZE,
+ .compact_headroom = 2 * PAGE_SIZE,
+ .reply_reserve = PAGE_SIZE,
+ .high_watermark = 4 * PAGE_SIZE,
+ .low_watermark = 2 * PAGE_SIZE,
+ .max_object_size = PAGE_SIZE,
+ .max_objects = 4,
+ .max_streams = 1,
+ .max_extents = 8,
+ };
+ struct opaque_basic *c = kunit_kzalloc(test, sizeof(*c), GFP_KERNEL);
+ struct io_ring_ctx *ctx;
+
+ KUNIT_ASSERT_NOT_NULL(test, c);
+ KUNIT_ASSERT_EQ(test,
+ kunit_add_action_or_reset(test, opaque_basic_release, c), 0);
+ /* No user/mm accounting in this internal-helper fixture. */
+ ctx = kunit_kzalloc(test, sizeof(*ctx), GFP_KERNEL);
+ KUNIT_ASSERT_NOT_NULL(test, ctx);
+ c->store = io_opaque_alloc(ctx, &config);
+ KUNIT_ASSERT_NOT_ERR_OR_NULL(test, c->store);
+ return c;
+}
+
+static void opaque_basic_budget(struct kunit *test)
+{
+ struct opaque_basic *c = opaque_basic_new(test);
+ struct io_opaque_store *store = c->store;
+
+ KUNIT_ASSERT_TRUE(test, io_opaque_charge(store, 3 * PAGE_SIZE, true));
+ KUNIT_EXPECT_EQ(test, io_opaque_memory_mode(store), IORING_OPAQUE_MEMORY_NORMAL);
+ KUNIT_ASSERT_TRUE(test, io_opaque_charge(store, PAGE_SIZE, true));
+ KUNIT_EXPECT_EQ(test, io_opaque_memory_mode(store), IORING_OPAQUE_MEMORY_LOWMEM);
+ KUNIT_EXPECT_FALSE(test, io_opaque_charge(store, 2 * PAGE_SIZE, true));
+ KUNIT_EXPECT_EQ(test, atomic64_read(&store->bytes), (s64)(4 * PAGE_SIZE));
+ io_opaque_uncharge(store, 2 * PAGE_SIZE);
+ KUNIT_EXPECT_EQ(test, io_opaque_memory_mode(store), IORING_OPAQUE_MEMORY_NORMAL);
+ io_opaque_uncharge(store, 2 * PAGE_SIZE);
+ KUNIT_EXPECT_EQ(test, atomic64_read(&store->bytes), 0LL);
+}
+
+static void opaque_basic_version(struct kunit *test)
+{
+ static const char payload[] = "opaque";
+ struct opaque_basic *c = opaque_basic_new(test);
+ struct io_opaque_extent *extent;
+ void *mapped;
+
+ c->data = io_opaque_data_alloc();
+ KUNIT_ASSERT_NOT_NULL(test, c->data);
+ c->page = alloc_page(GFP_KERNEL | __GFP_ZERO);
+ KUNIT_ASSERT_NOT_NULL(test, c->page);
+ mapped = kmap_local_page(c->page);
+ memcpy(mapped + 17, payload, sizeof(payload));
+ kunmap_local(mapped);
+ get_page(c->page);
+ extent = io_opaque_extent_new(c->store, c->page, 17, sizeof(payload), 0, false);
+ if (IS_ERR(extent))
+ put_page(c->page);
+ KUNIT_ASSERT_NOT_ERR_OR_NULL(test, extent);
+ list_add_tail(&extent->list, &c->data->extents);
+ c->data->length = sizeof(payload);
+ KUNIT_ASSERT_EQ(test, io_opaque_vector(c->data), 0);
+ KUNIT_ASSERT_EQ(test, c->data->nr, 1U);
+ KUNIT_EXPECT_PTR_EQ(test, c->data->bvec[0].bv_page, c->page);
+ KUNIT_EXPECT_EQ(test, c->data->bvec[0].bv_offset, 17U);
+ KUNIT_EXPECT_EQ(test, c->data->bvec[0].bv_len, (unsigned int)sizeof(payload));
+ KUNIT_EXPECT_EQ(test, c->data->charge, (u64)PAGE_SIZE);
+ KUNIT_EXPECT_EQ(test, atomic64_read(&c->store->bytes), (s64)PAGE_SIZE);
+ KUNIT_EXPECT_EQ(test, atomic_read(&c->store->extents), 1);
+ refcount_inc(&c->data->refs);
+ io_opaque_data_put(c->store, c->data);
+ KUNIT_EXPECT_EQ(test, refcount_read(&c->data->refs), 1U);
+ KUNIT_EXPECT_EQ(test, page_ref_count(c->page), 2);
+ mapped = kmap_local_page(c->data->bvec[0].bv_page);
+ KUNIT_EXPECT_MEMEQ(test, mapped + c->data->bvec[0].bv_offset, payload, sizeof(payload));
+ kunmap_local(mapped);
+ io_opaque_data_put(c->store, c->data);
+ c->data = NULL;
+ KUNIT_EXPECT_EQ(test, page_ref_count(c->page), 1);
+ KUNIT_EXPECT_EQ(test, atomic64_read(&c->store->bytes), 0LL);
+ KUNIT_EXPECT_EQ(test, atomic_read(&c->store->extents), 0);
+}
+
+static struct kunit_case opaque_basic_cases[] = {
+ KUNIT_CASE(opaque_basic_budget),
+ KUNIT_CASE(opaque_basic_version),
+ {}
+};
+
+static struct kunit_suite opaque_basic_suite = {
+ .name = "io_opaque_basic",
+ .test_cases = opaque_basic_cases,
+};
+
+kunit_test_suite(opaque_basic_suite);
+
+MODULE_LICENSE("GPL");
--
2.34.1
^ permalink raw reply [flat|nested] 13+ messages in thread