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@vger.kernel.org Link: https://lore.kernel.org/all/cover.1791994057.git.zihanx@nebusec.ai/ Reported-by: Vega Assisted-by: LLM Co-developed-by: Luxing Yin Signed-off-by: Luxing Yin Signed-off-by: Zihan Xi --- changes in v10: - No code changes; carry this patch forward unchanged in the updated three-patch series. - v9 Link: https://lore.kernel.org/all/cover.1791994057.git.zihanx@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 + #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