[PATCH net v9 1/3] rxrpc: account encapsulated packets to UDP socket
Zihan Xi
zihanx at nebusec.ai
Wed Oct 14 09:58:34 PDT 2026
rxrpc_encap_rcv() queues encapsulated UDP packets on the RxRPC local
queue without charging them to the UDP socket. If the I/O thread cannot
keep up, the queue can grow without bound.
Charge each packet against the tunnel socket under
sk_receive_queue.lock. Derive the receive-buffer limit from the
advertised window, retaining headroom for control and error traffic.
Drop packets when the buffer or protocol memory budget is exhausted,
while preserving UDP's empty-queue exception.
Orphan charged skbs when the I/O thread dequeues them. Since
skb_set_owner_r() does not hold a socket reference, clear sk_user_data
under RCU and release the socket only after purging the local queues.
The remaining patches bound RxRPC queues that can retain skbs after the
UDP receive-memory charge is released.
Fixes: 446b3e14525b ("rxrpc: Move packet reception processing into I/O thread")
Cc: stable at vger.kernel.org
Link: https://lore.kernel.org/all/cover.1791428529.git.zihanx@nebusec.ai/
Reported-by: Vega <vega at nebusec.ai>
Assisted-by: LLM
Co-developed-by: Luxing Yin <root at tr0jan.top>
Signed-off-by: Luxing Yin <root at tr0jan.top>
Signed-off-by: Zihan Xi <zihanx at nebusec.ai>
---
changes in v9:
- Split the fix into UDP accounting, connection/OOB queue budgeting,
and DATA receive-queue budgeting patches.
- Correct the Fixes attribution for the asynchronous receive queues.
- v8 Link: https://lore.kernel.org/all/cover.1791428529.git.zihanx@nebusec.ai/
changes in v8:
- Rebase onto net/main at 6d25ffca055a and resend after the previous
posting became stale.
- v7 Link: https://lore.kernel.org/all/cover.1790427022.git.zihanx@nebusec.ai/
changes in v7:
- Rebase the fix onto net/main at a7bfaba4823e.
- v6 Link: https://lore.kernel.org/all/cover.1789608557.git.zihanx@nebusec.ai/
changes in v6:
- Refresh the tunnel socket receive-buffer cap when rxrpc_rx_window_size
increases, without shrinking an existing socket.
- Distinguish socket receive-buffer exhaustion from protocol memory
accounting failures in the drop reason.
- Cap the window-derived value at sysctl_rmem_max and only grow an
existing sk_rcvbuf.
- v5 Link: https://lore.kernel.org/all/cover.1789273347.git.zihanx@nebusec.ai/
changes in v5:
- Size the tunnel sk_rcvbuf from one advertised DATA window, with
additional headroom for ACKs and ICMP, capped by sysctl_rmem_max.
- Drop the dst instead of skb_dst_force(); keep skb->dev = NULL.
- Document the runtime receive-window sizing limitation.
- v4 Link: https://lore.kernel.org/all/cover.1788878590.git.zihanx@nebusec.ai/
changes in v4:
- Serialize UDP rmem charge and uncharge with sk_receive_queue.lock.
- Orphan only PACKET skbs charged by encap_rcv(); preserve error skb
ownership.
- Clarify the callback-listener setup and distinguish the unprivileged
flood from extra conditions used for the recorded OOM.
- v3 Link: https://lore.kernel.org/all/cover.1788539302.git.zihanx@nebusec.ai/
changes in v3:
- Orphan skbs when the I/O thread dequeues them so UDP rmem ownership
does not follow packets onto call or connection queues.
- Clarify the affected source files and the recorded OOM context.
- v2 Link: https://lore.kernel.org/all/cover.1785339953.git.zihanx@nebusec.ai/
changes in v2:
- Use sk_drops_inc(), correct the Fixes attribution, and refresh the
baseline reproducer and decoded crash log.
- Explain why packetdrill does not express the sustained unique-call
flood used by this reproducer.
- v1 Link: https://lore.kernel.org/all/cover.1784742007.git.zihanx@nebusec.ai/
---
net/rxrpc/ar-internal.h | 1 +
net/rxrpc/io_thread.c | 61 ++++++++++++++++++++++++++++++++++++++--
net/rxrpc/local_object.c | 24 ++++++++++++++--
3 files changed, 82 insertions(+), 4 deletions(-)
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index 865f05fe37ab9..b079ec98aaa74 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -1323,6 +1323,7 @@ void rxrpc_send_version_request(struct rxrpc_local *local,
* local_object.c
*/
void rxrpc_local_dont_fragment(const struct rxrpc_local *local, bool set);
+void rxrpc_adjust_rcvbuf(struct sock *sk);
struct rxrpc_local *rxrpc_lookup_local(struct net *, const struct sockaddr_rxrpc *);
struct rxrpc_local *rxrpc_get_local(struct rxrpc_local *, enum rxrpc_local_trace);
struct rxrpc_local *rxrpc_get_local_maybe(struct rxrpc_local *, enum rxrpc_local_trace);
diff --git a/net/rxrpc/io_thread.c b/net/rxrpc/io_thread.c
index dc5184a2fa9d1..634a6e1cab87a 100644
--- a/net/rxrpc/io_thread.c
+++ b/net/rxrpc/io_thread.c
@@ -7,12 +7,37 @@
#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
+#include <net/udp.h>
+
#include "ar-internal.h"
static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
struct sockaddr_rxrpc *peer_srx,
struct sk_buff *skb);
+/*
+ * Drop UDP rmem ownership for packets charged in encap_rcv().
+ * sk_forward_alloc is serialised by sk_receive_queue.lock.
+ */
+static void rxrpc_skb_orphan_udp(struct sk_buff *skb)
+{
+ struct sock *sk = skb->sk;
+
+ if (!sk)
+ return;
+
+ spin_lock_bh(&sk->sk_receive_queue.lock);
+ skb_orphan(skb);
+ spin_unlock_bh(&sk->sk_receive_queue.lock);
+}
+
+static void rxrpc_encap_rcv_drop(struct sock *udp_sk, struct sk_buff *skb,
+ enum skb_drop_reason reason)
+{
+ sk_drops_inc(udp_sk);
+ sk_skb_reason_drop(udp_sk, skb, reason);
+}
+
/*
* handle data received on the local endpoint
* - may be called in interrupt context
@@ -28,6 +53,10 @@ int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
struct sk_buff_head *rx_queue;
struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
struct task_struct *io_thread;
+ enum skb_drop_reason reason;
+ unsigned int headroom;
+ unsigned int rcvbuf;
+ unsigned int rmem;
if (unlikely(!local)) {
kfree_skb(skb);
@@ -41,8 +70,6 @@ int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
if (skb->tstamp == 0)
skb->tstamp = ktime_get_real();
- skb->mark = RXRPC_SKB_MARK_PACKET;
- rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
rx_queue = &local->rx_queue;
#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
if (rxrpc_inject_rx_delay ||
@@ -52,9 +79,36 @@ int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
}
#endif
+ headroom = SKB_TRUESIZE(RXRPC_JUMBO(1)) * 2;
+ spin_lock(&udp_sk->sk_receive_queue.lock);
+ rxrpc_adjust_rcvbuf(udp_sk);
+ rcvbuf = READ_ONCE(udp_sk->sk_rcvbuf);
+ rmem = atomic_read(&udp_sk->sk_rmem_alloc);
+ /* Permit one skb when there is no receive memory charged, as UDP does. */
+ if (rmem && rmem + skb->truesize + headroom >= rcvbuf) {
+ reason = SKB_DROP_REASON_SOCKET_RCVBUFF;
+ goto drop;
+ }
+ if (!sk_rmem_schedule(udp_sk, skb, skb->truesize)) {
+ reason = SKB_DROP_REASON_PROTO_MEM;
+ goto drop;
+ }
+
+ skb->dev = NULL;
+ skb_set_owner_r(skb, udp_sk);
+ spin_unlock(&udp_sk->sk_receive_queue.lock);
+ skb_dst_drop(skb);
+
+ skb->mark = RXRPC_SKB_MARK_PACKET;
+ rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
skb_queue_tail(rx_queue, skb);
wake_up_process(io_thread);
return 0;
+
+drop:
+ spin_unlock(&udp_sk->sk_receive_queue.lock);
+ rxrpc_encap_rcv_drop(udp_sk, skb, reason);
+ return 0;
}
/*
@@ -471,6 +525,9 @@ int rxrpc_io_thread(void *data)
/* Distribute packets and errors. */
while ((skb = __skb_dequeue(&rx_queue))) {
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+
+ if (skb->mark == RXRPC_SKB_MARK_PACKET)
+ rxrpc_skb_orphan_udp(skb);
switch (skb->mark) {
case RXRPC_SKB_MARK_PACKET:
skb->priority = 0;
diff --git a/net/rxrpc/local_object.c b/net/rxrpc/local_object.c
index 169f9dfdaa77f..f519a93059244 100644
--- a/net/rxrpc/local_object.c
+++ b/net/rxrpc/local_object.c
@@ -22,6 +22,19 @@
static void rxrpc_local_rcu(struct rcu_head *);
+void rxrpc_adjust_rcvbuf(struct sock *sk)
+{
+ u32 rcvbuf, old_rcvbuf;
+
+ rcvbuf = READ_ONCE(rxrpc_rx_window_size) *
+ SKB_TRUESIZE(RXRPC_JUMBO(1)) * 2;
+ rcvbuf += rcvbuf / 4;
+ rcvbuf = min_t(u32, rcvbuf, READ_ONCE(sysctl_rmem_max));
+ old_rcvbuf = READ_ONCE(sk->sk_rcvbuf);
+ if (rcvbuf > old_rcvbuf)
+ WRITE_ONCE(sk->sk_rcvbuf, rcvbuf);
+}
+
/*
* Handle an ICMP/ICMP6 error turning up at the tunnel. Push it through the
* usual mechanism so that it gets parsed and presented through the UDP
@@ -198,6 +211,9 @@ static int rxrpc_open_socket(struct rxrpc_local *local, struct net *net)
/* set the socket up */
usk = local->socket->sk;
+ spin_lock_bh(&usk->sk_receive_queue.lock);
+ rxrpc_adjust_rcvbuf(usk);
+ spin_unlock_bh(&usk->sk_receive_queue.lock);
usk->sk_error_report = rxrpc_error_report;
switch (srx->transport.family) {
@@ -437,8 +453,8 @@ void rxrpc_destroy_local(struct rxrpc_local *local)
if (socket) {
local->socket = NULL;
kernel_sock_shutdown(socket, SHUT_RDWR);
- socket->sk->sk_user_data = NULL;
- sock_release(socket);
+ rcu_assign_sk_user_data(socket->sk, NULL);
+ synchronize_rcu();
}
/* At this point, there should be no more packets coming in to the
@@ -448,6 +464,10 @@ void rxrpc_destroy_local(struct rxrpc_local *local)
rxrpc_purge_queue(&local->rx_delay_queue);
#endif
rxrpc_purge_queue(&local->rx_queue);
+
+ if (socket)
+ sock_release(socket);
+
rxrpc_purge_client_connections(local);
page_frag_cache_drain(&local->tx_alloc);
}
--
2.43.0
More information about the linux-afs
mailing list