[PATCH net v9 2/3] rxrpc: bound connection and OOB receive queues
Zihan Xi
zihanx at nebusec.ai
Wed Oct 14 09:58:35 PDT 2026
After rxrpc_io_thread() dequeues a packet, the UDP receive-memory charge
is released. Connection-level packets can remain queued on conn->rx_queue,
while managed RXGK challenges can remain on the socket's OOB queue and
pending-response set.
Account both paths against a per-local receive budget. Release the charge
when a packet is processed, answered, freed or purged, and retain it when a
secured service packet is returned to the local I/O queue.
Fixes: 17926a79320a ("[AF_RXRPC]: Provide secure RxRPC sockets for use by userspace and kernel both")
Fixes: 5800b1cf3fd8 ("rxrpc: Allow CHALLENGEs to the passed to the app for a RESPONSE")
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 connection/OOB queue budgeting from DATA receive-queue
budgeting.
- Add the Fixes attribution for the RXGK challenge OOB path.
- Order receive-budget helper declarations per netdev reverse-xmas style.
- v8 Link: https://lore.kernel.org/all/cover.1791428529.git.zihanx@nebusec.ai/
---
net/rxrpc/ar-internal.h | 3 +++
net/rxrpc/conn_event.c | 22 ++++++++++++++++----
net/rxrpc/conn_object.c | 19 +++++++++++++++---
net/rxrpc/io_thread.c | 2 ++
net/rxrpc/local_object.c | 1 +
net/rxrpc/oob.c | 5 ++++-
net/rxrpc/skbuff.c | 43 +++++++++++++++++++++++++++++++++++++++-
7 files changed, 86 insertions(+), 9 deletions(-)
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index b079ec98aaa74..36fc4609181e1 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -360,6 +360,7 @@ struct rxrpc_local {
struct sk_buff_head rx_delay_queue; /* Delay injection queue */
#endif
struct sk_buff_head rx_queue; /* Received packets */
+ atomic_long_t rx_async_queue_bytes; /* Outstanding async Rx bytes */
struct list_head conn_attend_q; /* Conns requiring immediate attention */
struct list_head call_attend_q; /* Calls requiring immediate attention */
@@ -1516,6 +1517,8 @@ void rxrpc_new_skb(struct sk_buff *, enum rxrpc_skb_trace);
void rxrpc_see_skb(struct sk_buff *, enum rxrpc_skb_trace);
void rxrpc_get_skb(struct sk_buff *, enum rxrpc_skb_trace);
void rxrpc_free_skb(struct sk_buff *, enum rxrpc_skb_trace);
+bool rxrpc_charge_async_queue(struct rxrpc_local *, struct sk_buff *);
+void rxrpc_uncharge_async_queue(struct rxrpc_local *, struct sk_buff *);
void rxrpc_purge_queue(struct sk_buff_head *);
/*
diff --git a/net/rxrpc/conn_event.c b/net/rxrpc/conn_event.c
index 611c790bc6d0c..4baddd178cfa5 100644
--- a/net/rxrpc/conn_event.c
+++ b/net/rxrpc/conn_event.c
@@ -393,6 +393,8 @@ static void rxrpc_do_process_connection(struct rxrpc_connection *conn)
while ((skb = skb_dequeue(&conn->rx_queue))) {
rxrpc_see_skb(skb, rxrpc_skb_see_conn_work);
rxrpc_process_event(conn, skb);
+ if (skb->mark != RXRPC_SKB_MARK_SERVICE_CONN_SECURED)
+ rxrpc_uncharge_async_queue(conn->local, skb);
rxrpc_free_skb(skb, rxrpc_skb_put_conn_work);
}
}
@@ -414,15 +416,22 @@ void rxrpc_process_connection(struct work_struct *work)
* post connection-level events to the connection
* - this includes challenges, responses, some aborts and call terminal packet
* retransmission.
+ * - bound the queued memory independently of the UDP rmem charge.
*/
-static void rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
+static bool rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
struct sk_buff *skb)
{
+ struct rxrpc_local *local = conn->local;
+
+ if (!rxrpc_charge_async_queue(local, skb))
+ return false;
+
_enter("%p,%p", conn, skb);
rxrpc_get_skb(skb, rxrpc_skb_get_conn_work);
skb_queue_tail(&conn->rx_queue, skb);
rxrpc_queue_conn(conn, rxrpc_conn_queue_rx_work);
+ return true;
}
/*
@@ -442,7 +451,10 @@ static bool rxrpc_post_challenge(struct rxrpc_connection *conn,
rxrpc_get_connection(conn, rxrpc_conn_get_challenge_input);
if (!conn->security->challenge_to_recvmsg) {
- rxrpc_post_packet_to_conn(conn, skb);
+ if (!rxrpc_post_packet_to_conn(conn, skb)) {
+ rxrpc_put_connection(conn, rxrpc_conn_put_challenge_input);
+ sp->chall.conn = NULL;
+ }
return true;
}
@@ -480,8 +492,10 @@ static bool rxrpc_post_challenge(struct rxrpc_connection *conn,
return false;
}
- if (!call)
- rxrpc_post_packet_to_conn(conn, skb);
+ if (!call && !rxrpc_post_packet_to_conn(conn, skb)) {
+ rxrpc_put_connection(conn, rxrpc_conn_put_challenge_input);
+ sp->chall.conn = NULL;
+ }
return true;
}
diff --git a/net/rxrpc/conn_object.c b/net/rxrpc/conn_object.c
index 0ece717db0f85..d85345e014a9f 100644
--- a/net/rxrpc/conn_object.c
+++ b/net/rxrpc/conn_object.c
@@ -300,6 +300,19 @@ static void rxrpc_rcu_free_connection(struct rcu_head *rcu)
wake_up_var(&rxnet->nr_conns);
}
+/*
+ * Purge a connection's receive queue, updating the local queue accounting.
+ */
+static void rxrpc_purge_conn_queue(struct rxrpc_connection *conn)
+{
+ struct sk_buff *skb;
+
+ while ((skb = skb_dequeue(&conn->rx_queue))) {
+ rxrpc_uncharge_async_queue(conn->local, skb);
+ rxrpc_free_skb(skb, rxrpc_skb_put_purge);
+ }
+}
+
/*
* Clean up a dead connection.
*/
@@ -329,7 +342,7 @@ static void rxrpc_clean_up_connection(struct work_struct *work)
conn->peer->pmtud_pending = true;
}
- rxrpc_purge_queue(&conn->rx_queue);
+ rxrpc_purge_conn_queue(conn);
rxrpc_free_skb(conn->tx_response, rxrpc_skb_put_response);
rxrpc_kill_client_conn(conn);
@@ -338,15 +351,15 @@ static void rxrpc_clean_up_connection(struct work_struct *work)
key_put(conn->key);
rxrpc_put_bundle(conn->bundle, rxrpc_bundle_put_conn);
rxrpc_put_peer(conn->peer, rxrpc_peer_put_conn);
- rxrpc_put_local(conn->local, rxrpc_local_put_kill_conn);
/* Drain the Rx queue. Note that even though we've unpublished, an
* incoming packet could still be being added to our Rx queue, so we
* will need to drain it again in the RCU cleanup handler.
*/
- rxrpc_purge_queue(&conn->rx_queue);
+ rxrpc_purge_conn_queue(conn);
page_frag_cache_drain(&conn->tx_data_alloc);
+ rxrpc_put_local(conn->local, rxrpc_local_put_kill_conn);
call_rcu(&conn->rcu, rxrpc_rcu_free_connection);
}
diff --git a/net/rxrpc/io_thread.c b/net/rxrpc/io_thread.c
index 634a6e1cab87a..8679402c03e1b 100644
--- a/net/rxrpc/io_thread.c
+++ b/net/rxrpc/io_thread.c
@@ -528,6 +528,8 @@ int rxrpc_io_thread(void *data)
if (skb->mark == RXRPC_SKB_MARK_PACKET)
rxrpc_skb_orphan_udp(skb);
+ else if (skb->mark == RXRPC_SKB_MARK_SERVICE_CONN_SECURED)
+ rxrpc_uncharge_async_queue(sp->poke_conn->local, 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 f519a93059244..8e43f77d43683 100644
--- a/net/rxrpc/local_object.c
+++ b/net/rxrpc/local_object.c
@@ -137,6 +137,7 @@ static struct rxrpc_local *rxrpc_alloc_local(struct net *net,
skb_queue_head_init(&local->rx_delay_queue);
#endif
skb_queue_head_init(&local->rx_queue);
+ atomic_long_set(&local->rx_async_queue_bytes, 0);
INIT_LIST_HEAD(&local->conn_attend_q);
INIT_LIST_HEAD(&local->call_attend_q);
diff --git a/net/rxrpc/oob.c b/net/rxrpc/oob.c
index c80ee2487d094..5ab4ec5ad8c31 100644
--- a/net/rxrpc/oob.c
+++ b/net/rxrpc/oob.c
@@ -46,7 +46,8 @@ bool rxrpc_notify_socket_oob(struct rxrpc_call *call, struct sk_buff *skb)
sk = &rx->sk;
spin_lock_irq(&rx->recvmsg_lock);
- if (sk->sk_state < RXRPC_CLOSE) {
+ if (sk->sk_state < RXRPC_CLOSE &&
+ rxrpc_charge_async_queue(sp->chall.conn->local, skb)) {
skb->skb_mstamp_ns = rx->oob_id_counter++;
rxrpc_get_skb(skb, rxrpc_skb_get_post_oob);
skb_queue_tail(&rx->recvmsg_oobq, skb);
@@ -215,6 +216,7 @@ static int rxrpc_respond_to_oob(struct rxrpc_sock *rx,
switch (skb->mark) {
case RXRPC_OOB_CHALLENGE:
+ rxrpc_uncharge_async_queue(sp->chall.conn->local, skb);
rxrpc_put_connection(sp->chall.conn, rxrpc_conn_put_oob);
break;
}
@@ -329,6 +331,7 @@ void rxrpc_kernel_free_oob(struct sk_buff *oob)
switch (oob->mark) {
case RXRPC_OOB_CHALLENGE:
+ rxrpc_uncharge_async_queue(sp->chall.conn->local, oob);
rxrpc_put_connection(sp->chall.conn, rxrpc_conn_put_oob);
break;
}
diff --git a/net/rxrpc/skbuff.c b/net/rxrpc/skbuff.c
index e2169d1a14b5f..a5113a284456e 100644
--- a/net/rxrpc/skbuff.c
+++ b/net/rxrpc/skbuff.c
@@ -58,6 +58,39 @@ void rxrpc_free_skb(struct sk_buff *skb, enum rxrpc_skb_trace why)
}
}
+/*
+ * Charge an skb retained outside UDP receive-memory accounting.
+ */
+bool rxrpc_charge_async_queue(struct rxrpc_local *local, struct sk_buff *skb)
+{
+ struct sock *sk = local->socket->sk;
+ long rcvbuf;
+ long queued;
+
+ rcvbuf = READ_ONCE(sk->sk_rcvbuf);
+
+ for (;;) {
+ queued = atomic_long_read(&local->rx_async_queue_bytes);
+ if (queued &&
+ (queued >= rcvbuf ||
+ (long)skb->truesize >= rcvbuf - queued)) {
+ sk_drops_inc(sk);
+ return false;
+ }
+ if (atomic_long_cmpxchg(&local->rx_async_queue_bytes, queued,
+ queued + (long)skb->truesize) == queued)
+ return true;
+ }
+}
+
+/*
+ * Release the asynchronous receive charge on an skb.
+ */
+void rxrpc_uncharge_async_queue(struct rxrpc_local *local, struct sk_buff *skb)
+{
+ atomic_long_sub(skb->truesize, &local->rx_async_queue_bytes);
+}
+
/*
* Clear a queue of socket buffers.
*/
@@ -66,7 +99,15 @@ void rxrpc_purge_queue(struct sk_buff_head *list)
struct sk_buff *skb;
while ((skb = skb_dequeue((list))) != NULL) {
- int n = atomic_dec_return(select_skb_count(skb));
+ int n;
+
+ if (skb->mark == RXRPC_SKB_MARK_SERVICE_CONN_SECURED) {
+ struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+
+ rxrpc_uncharge_async_queue(sp->poke_conn->local, skb);
+ rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke);
+ }
+ n = atomic_dec_return(select_skb_count(skb));
trace_rxrpc_skb(skb, refcount_read(&skb->users), n,
rxrpc_skb_put_purge);
consume_skb(skb);
--
2.43.0
More information about the linux-afs
mailing list