[PATCH net v11 5/5] rxrpc: bound DATA receive queues
Zihan Xi
zihanx at nebusec.ai
Wed Oct 21 00:37:16 PDT 2026
DATA in per-call buffers is not covered by common socket rmem accounting.
Multiple calls can therefore retain more DATA than the shared limit. The
receive window advances as packets arrive, while rx_consumed advances only
as the application reads them.
Charge accepted DATA to per-call and per-local asynchronous budgets, and
use rx_consumed as the receive-window baseline. Limit each call's
contribution to the shared DATA budget, while reserving room for one
maximum-sized hole-filling packet beyond ordinary DATA and OOS charges.
Keep UDP rmem accounting until the asynchronous charge is acquired,
including call-queue handoff and jumbo-packet splitting. Release DATA
charges when consumed or purged, and purge call input and OOS queues on
completion.
The per-local DATA budget remains shared across calls. Aggregate pressure
can reject an otherwise in-window DATA packet with RXRPC_ACK_NOSPACE, even
though each call's receive window is still based on application
consumption. Check the window before accepting RXRPC_LAST_PACKET, so an
out-of-window LAST packet cannot set RXRPC_CALL_RX_LAST.
Refine managed OOB accounting with per-socket and per-peer sublimits.
When a per-owner cap is smaller than the local OOB allowance, it prevents
one socket or peer from consuming it all. With a small allowance, the
minimum per-owner cap can equal or exceed the local allowance, so a
socket or peer may still consume it entirely. An OOB skb larger than the
local allowance is rejected even when the OOB queue is empty. Preserve
the received skb mark for rx_done tracing and classify OOB challenges by
the RxRPC packet type; the public OOB API continues to report
RXRPC_OOB_CHALLENGE.
The issue was found during development and reproduced with a PoC against
an unpatched RxRPC baseline.
Fixes: 248f219cb8bc ("rxrpc: Rewrite the data and ack handling code")
Fixes: 5d7edbc9231e ("rxrpc: Get rid of the Rx ring")
Cc: stable at vger.kernel.org
Link: https://lore.kernel.org/all/cover.1791994057.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 v11:
- Use kzalloc_obj() for the asynchronous charge object.
- Keep UDP rmem accounting until accepted DATA is charged, including
call-queue processing and jumbo-packet splitting.
- Bound DATA against per-call and per-local limits; reserve headroom
for a maximum-sized hole-filling packet and keep capacity for
connection-control and OOB traffic.
- Release out-of-sequence DATA charges when a call completes.
- Document the shared-budget RXRPC_ACK_NOSPACE and OOB-drop behavior.
- Purge call input and OOS queues on completion; check the window
before accepting LAST_PACKET.
- Correct the DATA-accounting Fixes attribution to 248f219cb8bc,
retaining 5d7edbc9231e for the receive-window issue.
- Record discovery and measured baseline/patched test results.
- v10 Link: https://lore.kernel.org/all/cover.1792168462.git.zihanx@nebusec.ai/
changes in v10:
- Correct the aggregate DATA-accounting Fixes tag to 17926a79320a;
retain 5d7edbc9231e for the receive-window issue.
- v9 Link: https://lore.kernel.org/all/cover.1791994057.git.zihanx@nebusec.ai/
changes in v9:
- Split DATA receive-window and per-local aggregate queue accounting
from connection/OOB queue budgeting.
- Attribute the aggregate DATA accounting gap to 248f219cb8bc and retain
5d7edbc9231e for the receive-window accounting issue.
- v8 Link: https://lore.kernel.org/all/cover.1791428529.git.zihanx@nebusec.ai/
---
net/rxrpc/af_rxrpc.c | 1 +
net/rxrpc/ar-internal.h | 51 +++++-
net/rxrpc/call_event.c | 2 +
net/rxrpc/call_object.c | 6 +-
net/rxrpc/conn_event.c | 17 +-
net/rxrpc/conn_object.c | 2 +-
net/rxrpc/input.c | 100 +++++++---
net/rxrpc/io_thread.c | 2 +-
net/rxrpc/local_object.c | 8 +-
net/rxrpc/oob.c | 22 +--
net/rxrpc/peer_object.c | 1 +
net/rxrpc/recvmsg.c | 11 +-
net/rxrpc/skbuff.c | 382 +++++++++++++++++++++++++++++++++++++--
13 files changed, 529 insertions(+), 76 deletions(-)
diff --git a/net/rxrpc/af_rxrpc.c b/net/rxrpc/af_rxrpc.c
index 9ab0f22c881ec..9ddaed3dfde47 100644
--- a/net/rxrpc/af_rxrpc.c
+++ b/net/rxrpc/af_rxrpc.c
@@ -833,6 +833,7 @@ static int rxrpc_create(struct net *net, struct socket *sock, int protocol,
rx = rxrpc_sk(sk);
rx->family = protocol;
rx->calls = RB_ROOT;
+ atomic_long_set(&rx->rx_async_oob_queue_bytes, 0);
spin_lock_init(&rx->incoming_lock);
skb_queue_head_init(&rx->recvmsg_oobq);
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index e606145664e06..dfde4b409f1a1 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -59,7 +59,7 @@ struct rxgk_context;
enum rxrpc_skb_mark {
RXRPC_SKB_MARK_PACKET, /* Received packet */
RXRPC_SKB_MARK_ERROR, /* Error notification */
- RXRPC_SKB_MARK_CHALLENGE, /* Challenge notification */
+ RXRPC_SKB_MARK_CHALLENGE, /* Reserved to preserve mark values */
RXRPC_SKB_MARK_SERVICE_CONN_SECURED, /* Service connection response has been verified */
RXRPC_SKB_MARK_REJECT_BUSY, /* Reject with BUSY */
RXRPC_SKB_MARK_REJECT_ABORT, /* Reject with ABORT (code in skb->priority) */
@@ -173,6 +173,7 @@ struct rxrpc_sock {
struct rxrpc_backlog *backlog; /* Preallocation for services */
struct sk_buff_head recvmsg_oobq; /* OOB messages for recvmsg to pick up */
struct rb_root pending_oobq; /* OOB messages awaiting userspace to respond to */
+ atomic_long_t rx_async_oob_queue_bytes; /* OOB queue charge */
u64 oob_id_counter; /* OOB message ID counter */
spinlock_t incoming_lock; /* Incoming call vs service shutdown lock */
struct list_head sock_calls; /* List of calls owned by this socket */
@@ -222,6 +223,25 @@ struct rxrpc_host_header {
u16 serviceId; /* service ID */
} __packed;
+enum rxrpc_async_queue_class {
+ RXRPC_ASYNC_QUEUE_CONTROL,
+ RXRPC_ASYNC_QUEUE_OOB,
+ RXRPC_ASYNC_QUEUE_DATA,
+ RXRPC_ASYNC_QUEUE_DATA_RESERVE,
+ RXRPC_ASYNC_QUEUE_DATA_OOS,
+ RXRPC_ASYNC_QUEUE_DATA_HEAD,
+};
+
+/*
+ * Shared asynchronous receive-memory charge for split jumbo DATA packets.
+ */
+struct rxrpc_async_queue_charge {
+ refcount_t ref;
+ long size;
+ bool charged;
+ u8 class;
+};
+
/*
* RxRPC socket buffer private variables
* - max 48 bytes (struct sk_buff::cb)
@@ -233,6 +253,11 @@ struct rxrpc_skb_priv {
u16 offset; /* Offset of data */
u16 len; /* Length of data */
};
+ struct {
+ u16 offset;
+ u16 len;
+ struct rxrpc_async_queue_charge *charge;
+ } data;
struct {
rxrpc_seq_t first_ack; /* First packet in acks table */
rxrpc_seq_t prev_ack; /* Highest seq seen */
@@ -244,6 +269,7 @@ struct rxrpc_skb_priv {
struct rxrpc_connection *conn; /* Connection referred to */
union {
u32 rxkad_nonce;
+ struct rxrpc_sock *oob_owner;
};
} chall;
struct {
@@ -254,6 +280,7 @@ struct rxrpc_skb_priv {
u16 ticket_len;
} resp;
};
+ u8 async_queue_class;
struct rxrpc_host_header hdr; /* RxRPC packet header from this packet */
};
@@ -361,6 +388,10 @@ struct rxrpc_local {
#endif
struct sk_buff_head rx_queue; /* Received packets */
atomic_long_t rx_async_queue_bytes; /* Outstanding async Rx bytes */
+ atomic_long_t rx_async_data_queue_bytes; /* DATA queue charge */
+ atomic_long_t rx_async_data_oos_queue_bytes; /* OOS DATA charge */
+ atomic_long_t rx_async_control_queue_bytes;
+ atomic_long_t rx_async_oob_queue_bytes;
struct list_head conn_attend_q; /* Conns requiring immediate attention */
struct list_head call_attend_q; /* Calls requiring immediate attention */
@@ -404,6 +435,7 @@ struct rxrpc_peer {
struct rxrpc_local *local;
struct hlist_head error_targets; /* targets for net error distribution */
struct rb_root service_conns; /* Service connections */
+ atomic_long_t rx_async_oob_queue_bytes; /* OOB queue charge */
struct list_head keepalive_link; /* Link in net->peer_keepalive[] */
unsigned long app_data; /* Application data (e.g. afs_server) */
unsigned int last_tx_at; /* Last time packet sent here (time64_t LSW) */
@@ -795,6 +827,8 @@ struct rxrpc_call {
struct sk_buff_head recvmsg_queue; /* Queue of packets ready for recvmsg() */
struct sk_buff_head rx_queue; /* Queue of packets for this call to receive */
struct sk_buff_head rx_oos_queue; /* Queue of out of sequence packets */
+ atomic_long_t rx_async_data_queue_bytes;
+ bool rx_async_data_headroom_reserved; /* OOS hole-fill slot */
void *rx_dec_buffer; /* Decryption buffer */
unsigned short rx_dec_bsize; /* rx_dec_buffer size */
unsigned short rx_dec_offset; /* Decrypted packet data offset */
@@ -1518,14 +1552,23 @@ 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);
void rxrpc_skb_orphan_udp(struct sk_buff *skb);
-bool rxrpc_charge_async_queue(struct rxrpc_local *local, struct sk_buff *skb);
-void rxrpc_uncharge_async_queue(struct rxrpc_local *local, struct sk_buff *skb);
+struct rxrpc_async_queue_charge *rxrpc_alloc_async_queue_charge(struct sk_buff *skb);
+void rxrpc_put_async_queue_charge(struct rxrpc_local *local,
+ struct rxrpc_call *call,
+ struct rxrpc_async_queue_charge *charge);
+bool rxrpc_charge_async_queue(struct rxrpc_local *local, struct rxrpc_call *call,
+ struct rxrpc_sock *rx, struct sk_buff *skb,
+ enum rxrpc_async_queue_class class);
+void rxrpc_uncharge_async_queue(struct rxrpc_local *local,
+ struct rxrpc_call *call, struct sk_buff *skb);
+void rxrpc_release_async_queue_headroom(struct rxrpc_call *call);
+void rxrpc_purge_data_queue(struct rxrpc_call *call, struct sk_buff_head *list);
void rxrpc_purge_queue(struct sk_buff_head *);
+void rxrpc_purge_conn_queue(struct rxrpc_connection *);
/*
* stats.c
*/
-void rxrpc_purge_conn_queue(struct rxrpc_connection *);
int rxrpc_stats_show(struct seq_file *seq, void *v);
int rxrpc_stats_clear(struct file *file, char *buf, size_t size);
diff --git a/net/rxrpc/call_event.c b/net/rxrpc/call_event.c
index 17c3a2606216f..6265d113090bf 100644
--- a/net/rxrpc/call_event.c
+++ b/net/rxrpc/call_event.c
@@ -475,6 +475,8 @@ bool rxrpc_input_call_event(struct rxrpc_call *call)
out:
if (__rxrpc_call_is_complete(call)) {
timer_delete_sync(&call->timer);
+ rxrpc_purge_queue(&call->rx_queue);
+ rxrpc_purge_data_queue(call, &call->rx_oos_queue);
if (!test_bit(RXRPC_CALL_DISCONNECTED, &call->flags))
rxrpc_disconnect_call(call);
if (call->security)
diff --git a/net/rxrpc/call_object.c b/net/rxrpc/call_object.c
index 817ed9acb91e6..a57fdbd5972a1 100644
--- a/net/rxrpc/call_object.c
+++ b/net/rxrpc/call_object.c
@@ -150,6 +150,8 @@ struct rxrpc_call *rxrpc_alloc_call(struct rxrpc_sock *rx, gfp_t gfp,
skb_queue_head_init(&call->recvmsg_queue);
skb_queue_head_init(&call->rx_queue);
skb_queue_head_init(&call->rx_oos_queue);
+ atomic_long_set(&call->rx_async_data_queue_bytes, 0);
+ call->rx_async_data_headroom_reserved = false;
init_waitqueue_head(&call->waitq);
spin_lock_init(&call->notify_lock);
refcount_set(&call->ref, 1);
@@ -553,9 +555,9 @@ static void rxrpc_cleanup_tx_buffers(struct rxrpc_call *call)
*/
static void rxrpc_cleanup_rx_buffers(struct rxrpc_call *call)
{
- rxrpc_purge_queue(&call->recvmsg_queue);
+ rxrpc_purge_data_queue(call, &call->recvmsg_queue);
rxrpc_purge_queue(&call->rx_queue);
- rxrpc_purge_queue(&call->rx_oos_queue);
+ rxrpc_purge_data_queue(call, &call->rx_oos_queue);
kfree(call->rx_dec_buffer);
}
diff --git a/net/rxrpc/conn_event.c b/net/rxrpc/conn_event.c
index 7d31d7fa9cbd3..9e240d7969b2f 100644
--- a/net/rxrpc/conn_event.c
+++ b/net/rxrpc/conn_event.c
@@ -395,7 +395,7 @@ static void rxrpc_do_process_connection(struct rxrpc_connection *conn)
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_uncharge_async_queue(conn->local, NULL, skb);
if (sp->hdr.type == RXRPC_PACKET_TYPE_CHALLENGE && sp->chall.conn) {
rxrpc_put_connection(sp->chall.conn,
rxrpc_conn_put_challenge_input);
@@ -431,23 +431,26 @@ static bool rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
{
struct rxrpc_local *local = conn->local;
- if (!rxrpc_charge_async_queue(local, skb))
+ if (unlikely(!atomic_read(&local->active_users)))
+ return false;
+ if (!rxrpc_charge_async_queue(local, NULL, NULL, skb,
+ RXRPC_ASYNC_QUEUE_CONTROL))
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);
spin_lock_bh(&local->rx_queue.lock);
if (unlikely(!atomic_read(&local->active_users))) {
spin_unlock_bh(&local->rx_queue.lock);
- rxrpc_uncharge_async_queue(local, skb);
+ rxrpc_uncharge_async_queue(local, NULL, skb);
return false;
}
+ rxrpc_get_skb(skb, rxrpc_skb_get_conn_work);
+ skb_queue_tail(&conn->rx_queue, skb);
+ spin_unlock_bh(&local->rx_queue.lock);
+ rxrpc_queue_conn(conn, rxrpc_conn_queue_rx_work);
return true;
}
- spin_unlock_bh(&local->rx_queue.lock);
/*
* Post a CHALLENGE packet to the socket of one of a connection's calls so that
diff --git a/net/rxrpc/conn_object.c b/net/rxrpc/conn_object.c
index 73e60cbc8a138..15bd7ed301dee 100644
--- a/net/rxrpc/conn_object.c
+++ b/net/rxrpc/conn_object.c
@@ -310,7 +310,7 @@ void rxrpc_purge_conn_queue(struct rxrpc_connection *conn)
while ((skb = skb_dequeue(&conn->rx_queue))) {
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
- rxrpc_uncharge_async_queue(conn->local, skb);
+ rxrpc_uncharge_async_queue(conn->local, NULL, skb);
if (sp->hdr.type == RXRPC_PACKET_TYPE_CHALLENGE && sp->chall.conn) {
rxrpc_put_connection(sp->chall.conn,
rxrpc_conn_put_challenge_input);
diff --git a/net/rxrpc/input.c b/net/rxrpc/input.c
index 73cafe6bfa9f5..ae783661e4038 100644
--- a/net/rxrpc/input.c
+++ b/net/rxrpc/input.c
@@ -479,18 +479,25 @@ static void rxrpc_input_queue_data(struct rxrpc_call *call, struct sk_buff *skb,
* Process a DATA packet.
*/
static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,
- bool *_notify, rxrpc_serial_t *_ack_serial, int *_ack_reason)
+ bool *_notify, rxrpc_serial_t *_ack_serial,
+ int *_ack_reason,
+ struct rxrpc_async_queue_charge **_charge)
{
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
struct sk_buff *oos;
rxrpc_serial_t serial = sp->hdr.serial;
- unsigned int sack = call->ackr_sack_base;
rxrpc_seq_t window = call->ackr_window;
rxrpc_seq_t wtop = call->ackr_wtop;
- rxrpc_seq_t wlimit = window + call->rx_winsize - 1;
+ /* Bound queued DATA by the application's receive window. */
+ rxrpc_seq_t wlimit = smp_load_acquire(&call->rx_consumed) +
+ call->rx_winsize;
rxrpc_seq_t seq = sp->hdr.seq;
- bool last = sp->hdr.flags & RXRPC_LAST_PACKET;
+ unsigned int sack = call->ackr_sack_base;
+ unsigned int slot = 0;
+ enum rxrpc_async_queue_class queue_class;
int ack_reason = -1;
+ bool last = sp->hdr.flags & RXRPC_LAST_PACKET;
+ bool set_last = false;
rxrpc_inc_stat(call->rxnet, stat_rx_data);
if (sp->hdr.flags & RXRPC_REQUEST_ACK)
@@ -498,10 +505,24 @@ static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,
if (sp->hdr.flags & RXRPC_JUMBO_PACKET)
rxrpc_inc_stat(call->rxnet, stat_rx_data_jumbo);
+ if (after(seq, call->rx_highest_seq))
+ call->rx_highest_seq = seq;
+
+ trace_rxrpc_rx_data(call->debug_id, seq, serial, sp->hdr.flags);
+
+ if (after(seq, wlimit)) {
+ ack_reason = RXRPC_ACK_EXCEEDS_WINDOW;
+ goto send_ack;
+ }
+
if (last) {
- if (test_and_set_bit(RXRPC_CALL_RX_LAST, &call->flags) &&
- seq + 1 != wtop)
- return rxrpc_proto_abort(call, seq, rxrpc_eproto_different_last);
+ if (test_and_set_bit(RXRPC_CALL_RX_LAST, &call->flags)) {
+ if (seq + 1 != wtop)
+ return rxrpc_proto_abort(call, seq,
+ rxrpc_eproto_different_last);
+ } else {
+ set_last = true;
+ }
} else {
if (test_bit(RXRPC_CALL_RX_LAST, &call->flags) &&
after_eq(seq, wtop)) {
@@ -511,17 +532,40 @@ static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,
}
}
- if (after(seq, call->rx_highest_seq))
- call->rx_highest_seq = seq;
-
- trace_rxrpc_rx_data(call->debug_id, seq, serial, sp->hdr.flags);
-
if (before(seq, window)) {
ack_reason = RXRPC_ACK_DUPLICATE;
goto send_ack;
}
- if (after(seq, wlimit)) {
- ack_reason = RXRPC_ACK_EXCEEDS_WINDOW;
+
+ if (seq != window) {
+ slot = seq - window;
+ sack = (sack + slot) % RXRPC_SACK_SIZE;
+
+ if (call->ackr_sack_table[sack]) {
+ ack_reason = RXRPC_ACK_DUPLICATE;
+ goto send_ack;
+ }
+ }
+
+ if (seq != window)
+ queue_class = RXRPC_ASYNC_QUEUE_DATA_OOS;
+ else if (!skb_queue_empty(&call->rx_oos_queue))
+ queue_class = RXRPC_ASYNC_QUEUE_DATA_HEAD;
+ else
+ queue_class = RXRPC_ASYNC_QUEUE_DATA;
+
+ sp->data.charge = NULL;
+ if (sp->offset != sizeof(struct rxrpc_wire_header) ||
+ (sp->hdr.flags & RXRPC_JUMBO_PACKET)) {
+ if (!*_charge)
+ *_charge = rxrpc_alloc_async_queue_charge(skb);
+ sp->data.charge = *_charge;
+ }
+ if (!rxrpc_charge_async_queue(call->local, call, NULL, skb, queue_class)) {
+ sp->data.charge = NULL;
+ if (set_last)
+ clear_bit(RXRPC_CALL_RX_LAST, &call->flags);
+ ack_reason = RXRPC_ACK_NOSPACE;
goto send_ack;
}
@@ -565,22 +609,14 @@ static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,
rxrpc_input_queue_data(call, oos, window, wtop,
rxrpc_receive_queue_oos);
}
+ if (queue_class == RXRPC_ASYNC_QUEUE_DATA_HEAD &&
+ skb_queue_empty(&call->rx_oos_queue))
+ rxrpc_release_async_queue_headroom(call);
call->ackr_sack_base = sack;
} else {
- unsigned int slot;
-
ack_reason = RXRPC_ACK_OUT_OF_SEQUENCE;
-
- slot = seq - window;
- sack = (sack + slot) % RXRPC_SACK_SIZE;
-
- if (call->ackr_sack_table[sack % RXRPC_SACK_SIZE]) {
- ack_reason = RXRPC_ACK_DUPLICATE;
- goto send_ack;
- }
-
- call->ackr_sack_table[sack % RXRPC_SACK_SIZE] |= 1;
+ call->ackr_sack_table[sack] |= 1;
trace_rxrpc_sack(call, seq, sack, rxrpc_sack_oos);
if (after(seq + 1, wtop)) {
@@ -623,6 +659,7 @@ static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,
*/
static bool rxrpc_input_split_jumbo(struct rxrpc_call *call, struct sk_buff *skb)
{
+ struct rxrpc_async_queue_charge *charge = NULL;
struct rxrpc_jumbo_header jhdr;
struct rxrpc_skb_priv *sp = rxrpc_skb(skb), *jsp;
struct sk_buff *jskb;
@@ -644,13 +681,17 @@ static bool rxrpc_input_split_jumbo(struct rxrpc_call *call, struct sk_buff *skb
jskb = skb_clone(skb, GFP_NOFS);
if (!jskb) {
kdebug("couldn't clone");
+ rxrpc_put_async_queue_charge(call->local, call, charge);
return false;
}
rxrpc_new_skb(jskb, rxrpc_skb_new_jumbo_subpacket);
jsp = rxrpc_skb(jskb);
jsp->offset = offset;
jsp->len = RXRPC_JUMBO_DATALEN;
- rxrpc_input_data_one(call, jskb, ¬ify, &ack_serial, &ack_reason);
+ rxrpc_input_data_one(call, jskb, ¬ify, &ack_serial, &ack_reason,
+ &charge);
+ if (charge && charge->charged)
+ rxrpc_skb_orphan_udp(skb);
rxrpc_free_skb(jskb, rxrpc_skb_put_jumbo_subpacket);
sp->hdr.flags = jhdr.flags;
@@ -664,7 +705,8 @@ static bool rxrpc_input_split_jumbo(struct rxrpc_call *call, struct sk_buff *skb
sp->offset = offset;
sp->len = len;
- rxrpc_input_data_one(call, skb, ¬ify, &ack_serial, &ack_reason);
+ rxrpc_input_data_one(call, skb, ¬ify, &ack_serial, &ack_reason,
+ &charge);
stat_ix = umin(count, ARRAY_SIZE(call->rxnet->stat_rx_jumbo)) - 1;
atomic_inc(&call->rxnet->stat_rx_jumbo[stat_ix]);
@@ -681,9 +723,11 @@ static bool rxrpc_input_split_jumbo(struct rxrpc_call *call, struct sk_buff *skb
trace_rxrpc_notify_socket(call->debug_id, sp->hdr.serial);
rxrpc_notify_socket(call);
}
+ rxrpc_put_async_queue_charge(call->local, call, charge);
return true;
protocol_error:
+ rxrpc_put_async_queue_charge(call->local, call, charge);
return false;
}
diff --git a/net/rxrpc/io_thread.c b/net/rxrpc/io_thread.c
index 4699f4100ee4d..cd5fee95aea7f 100644
--- a/net/rxrpc/io_thread.c
+++ b/net/rxrpc/io_thread.c
@@ -527,7 +527,7 @@ int rxrpc_io_thread(void *data)
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
if (skb->mark == RXRPC_SKB_MARK_SERVICE_CONN_SECURED)
- rxrpc_uncharge_async_queue(sp->poke_conn->local, skb);
+ rxrpc_uncharge_async_queue(sp->poke_conn->local, NULL, 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 df1b4711082a1..bf4c16de29039 100644
--- a/net/rxrpc/local_object.c
+++ b/net/rxrpc/local_object.c
@@ -138,6 +138,10 @@ static struct rxrpc_local *rxrpc_alloc_local(struct net *net,
#endif
skb_queue_head_init(&local->rx_queue);
atomic_long_set(&local->rx_async_queue_bytes, 0);
+ atomic_long_set(&local->rx_async_data_queue_bytes, 0);
+ atomic_long_set(&local->rx_async_data_oos_queue_bytes, 0);
+ atomic_long_set(&local->rx_async_control_queue_bytes, 0);
+ atomic_long_set(&local->rx_async_oob_queue_bytes, 0);
INIT_LIST_HEAD(&local->conn_attend_q);
INIT_LIST_HEAD(&local->call_attend_q);
@@ -420,10 +424,10 @@ void rxrpc_unuse_local(struct rxrpc_local *local, enum rxrpc_local_trace why)
if (local) {
debug_id = local->debug_id;
r = refcount_read(&local->ref);
- u = atomic_dec_return(&local->active_users);
spin_lock_bh(&local->rx_queue.lock);
- trace_rxrpc_local(debug_id, why, r, u);
+ u = atomic_dec_return(&local->active_users);
spin_unlock_bh(&local->rx_queue.lock);
+ trace_rxrpc_local(debug_id, why, r, u);
if (u == 0)
kthread_stop(local->io_thread);
}
diff --git a/net/rxrpc/oob.c b/net/rxrpc/oob.c
index 0890b0d29112c..0d2c7208c7623 100644
--- a/net/rxrpc/oob.c
+++ b/net/rxrpc/oob.c
@@ -47,8 +47,8 @@ bool rxrpc_notify_socket_oob(struct rxrpc_call *call, struct sk_buff *skb)
spin_lock_irq(&rx->recvmsg_lock);
if (sk->sk_state < RXRPC_CLOSE &&
- rxrpc_charge_async_queue(sp->chall.conn->local, skb)) {
- skb->mark = RXRPC_SKB_MARK_CHALLENGE;
+ rxrpc_charge_async_queue(sp->chall.conn->local, NULL, rx, skb,
+ RXRPC_ASYNC_QUEUE_OOB)) {
skb->skb_mstamp_ns = rx->oob_id_counter++;
rxrpc_get_skb(skb, rxrpc_skb_get_post_oob);
skb_queue_tail(&rx->recvmsg_oobq, skb);
@@ -196,7 +196,7 @@ static int rxrpc_respond_to_oob(struct rxrpc_sock *rx,
switch (p->command) {
case RXRPC_OOB_CMD_RESPOND:
ret = -EPROTO;
- if (skb->mark != RXRPC_SKB_MARK_CHALLENGE)
+ if (sp->hdr.type != RXRPC_PACKET_TYPE_CHALLENGE)
break;
conn = sp->chall.conn;
ret = -EOPNOTSUPP;
@@ -215,9 +215,9 @@ static int rxrpc_respond_to_oob(struct rxrpc_sock *rx,
break;
}
- switch (skb->mark) {
- case RXRPC_SKB_MARK_CHALLENGE:
- rxrpc_uncharge_async_queue(sp->chall.conn->local, skb);
+ switch (sp->hdr.type) {
+ case RXRPC_PACKET_TYPE_CHALLENGE:
+ rxrpc_uncharge_async_queue(sp->chall.conn->local, NULL, skb);
rxrpc_put_connection(sp->chall.conn, rxrpc_conn_put_oob);
break;
}
@@ -279,8 +279,8 @@ enum rxrpc_oob_type rxrpc_kernel_query_oob(struct sk_buff *oob,
{
struct rxrpc_skb_priv *sp = rxrpc_skb(oob);
- switch (oob->mark) {
- case RXRPC_SKB_MARK_CHALLENGE:
+ switch (sp->hdr.type) {
+ case RXRPC_PACKET_TYPE_CHALLENGE:
*_peer = sp->chall.conn->peer;
*_peer_appdata = sp->chall.conn->peer->app_data;
break;
@@ -329,9 +329,9 @@ void rxrpc_kernel_free_oob(struct sk_buff *oob)
{
struct rxrpc_skb_priv *sp = rxrpc_skb(oob);
- switch (oob->mark) {
- case RXRPC_SKB_MARK_CHALLENGE:
- rxrpc_uncharge_async_queue(sp->chall.conn->local, oob);
+ switch (sp->hdr.type) {
+ case RXRPC_PACKET_TYPE_CHALLENGE:
+ rxrpc_uncharge_async_queue(sp->chall.conn->local, NULL, oob);
rxrpc_put_connection(sp->chall.conn, rxrpc_conn_put_oob);
break;
}
diff --git a/net/rxrpc/peer_object.c b/net/rxrpc/peer_object.c
index fa9a406e11683..74d549c53f106 100644
--- a/net/rxrpc/peer_object.c
+++ b/net/rxrpc/peer_object.c
@@ -229,6 +229,7 @@ struct rxrpc_peer *rxrpc_alloc_peer(struct rxrpc_local *local, gfp_t gfp,
peer = kzalloc_obj(struct rxrpc_peer, gfp);
if (peer) {
refcount_set(&peer->ref, 1);
+ atomic_long_set(&peer->rx_async_oob_queue_bytes, 0);
peer->local = rxrpc_get_local(local, rxrpc_local_get_peer);
INIT_HLIST_HEAD(&peer->error_targets);
peer->service_conns = RB_ROOT;
diff --git a/net/rxrpc/recvmsg.c b/net/rxrpc/recvmsg.c
index 172c72944cddd..5cd0d1801b69e 100644
--- a/net/rxrpc/recvmsg.c
+++ b/net/rxrpc/recvmsg.c
@@ -118,6 +118,7 @@ static void rxrpc_rotate_rx_window(struct rxrpc_call *call)
skb = skb_dequeue(&call->recvmsg_queue);
rxrpc_see_skb(skb, rxrpc_skb_see_rotate);
+ rxrpc_uncharge_async_queue(call->local, call, skb);
sp = rxrpc_skb(skb);
tseq = sp->hdr.seq;
@@ -234,6 +235,7 @@ static int rxrpc_recvmsg_oob(struct socket *sock, struct msghdr *msg,
unsigned int flags)
{
struct rxrpc_sock *rx = rxrpc_sk(sock->sk);
+ struct rxrpc_skb_priv *sp;
struct sk_buff *skb;
bool need_response = false;
int ret;
@@ -241,6 +243,7 @@ static int rxrpc_recvmsg_oob(struct socket *sock, struct msghdr *msg,
skb = skb_peek(&rx->recvmsg_oobq);
if (!skb)
return -EAGAIN;
+ sp = rxrpc_skb(skb);
rxrpc_see_skb(skb, rxrpc_skb_see_recvmsg);
ret = put_cmsg(msg, SOL_RXRPC, RXRPC_OOB_ID, sizeof(u64),
@@ -248,14 +251,14 @@ static int rxrpc_recvmsg_oob(struct socket *sock, struct msghdr *msg,
if (ret < 0)
return ret;
- switch (skb->mark) {
- case RXRPC_SKB_MARK_CHALLENGE:
+ switch (sp->hdr.type) {
+ case RXRPC_PACKET_TYPE_CHALLENGE:
need_response = true;
ret = rxrpc_recvmsg_challenge(sock, msg, skb, flags);
break;
default:
WARN_ONCE(1, "recvmsg() can't process unknown OOB type %u\n",
- skb->mark);
+ sp->hdr.type);
ret = -EIO;
break;
}
@@ -565,7 +568,7 @@ int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
goto not_yet_complete;
call_failed:
- rxrpc_purge_queue(&call->recvmsg_queue);
+ rxrpc_purge_data_queue(call, &call->recvmsg_queue);
call_complete:
ret = rxrpc_recvmsg_term(call, msg);
if (ret < 0)
diff --git a/net/rxrpc/skbuff.c b/net/rxrpc/skbuff.c
index 357226741e97e..11a9b1f14478f 100644
--- a/net/rxrpc/skbuff.c
+++ b/net/rxrpc/skbuff.c
@@ -8,8 +8,10 @@
#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
#include <linux/module.h>
+#include <linux/limits.h>
#include <linux/net.h>
#include <linux/skbuff.h>
+#include <linux/slab.h>
#include <net/sock.h>
#include <net/af_rxrpc.h>
#include "ar-internal.h"
@@ -58,39 +60,387 @@ void rxrpc_free_skb(struct sk_buff *skb, enum rxrpc_skb_trace why)
}
}
+/* Keep DATA from consuming the connection-control and OOB reserves. */
+static long rxrpc_async_queue_control_limit(long rcvbuf)
+{
+ long minimum = 2L * SKB_TRUESIZE(RXRPC_JUMBO(1));
+
+ return min_t(long, max_t(long, rcvbuf / 4, minimum), rcvbuf / 2);
+}
+
+/* Reserve a maximum-size packet, including split-jumbo skb overhead. */
+static long rxrpc_async_queue_data_headroom(void)
+{
+ /* rxrpc_rx_mtu can be set as high as the 16-bit packet size. */
+ unsigned int mtu = 65535;
+ unsigned int nr = DIV_ROUND_UP(mtu, RXRPC_JUMBO_SUBPKTLEN);
+
+ return SKB_TRUESIZE(mtu) + sizeof(struct rxrpc_async_queue_charge) +
+ nr * SKB_DATA_ALIGN(sizeof(struct sk_buff));
+}
+
+static enum rxrpc_async_queue_class
+rxrpc_async_queue_account_class(enum rxrpc_async_queue_class class)
+{
+ switch (class) {
+ case RXRPC_ASYNC_QUEUE_DATA_RESERVE:
+ case RXRPC_ASYNC_QUEUE_DATA_HEAD:
+ case RXRPC_ASYNC_QUEUE_DATA_OOS:
+ return RXRPC_ASYNC_QUEUE_DATA;
+ default:
+ return class;
+ }
+}
+
+static bool rxrpc_async_queue_charge_counter(atomic_long_t *counter, long size,
+ long limit)
+{
+ long queued;
+
+ for (;;) {
+ queued = atomic_long_read(counter);
+ if (queued >= limit || size > limit - queued)
+ return false;
+ if (atomic_long_cmpxchg(counter, queued, queued + size) == queued)
+ return true;
+ }
+}
+
/*
- * Transfer an skb to asynchronous receive-memory accounting.
+ * Calculate the charge corresponding to the call's advertised receive window.
*/
-bool rxrpc_charge_async_queue(struct rxrpc_local *local, struct sk_buff *skb)
+static long rxrpc_async_queue_call_data_limit(struct rxrpc_call *call)
+{
+ return (long)call->rx_winsize * SKB_TRUESIZE(RXRPC_JUMBO(1)) * 2;
+}
+
+/* Release the reserved space when a call's OOS queue has been drained. */
+void rxrpc_release_async_queue_headroom(struct rxrpc_call *call)
+{
+ long headroom;
+
+ if (!call->rx_async_data_headroom_reserved)
+ return;
+
+ headroom = rxrpc_async_queue_data_headroom();
+ atomic_long_sub(headroom, &call->local->rx_async_queue_bytes);
+ atomic_long_sub(headroom, &call->local->rx_async_data_queue_bytes);
+ atomic_long_sub(headroom, &call->local->rx_async_data_oos_queue_bytes);
+ atomic_long_sub(headroom, &call->rx_async_data_queue_bytes);
+ call->rx_async_data_headroom_reserved = false;
+}
+
+static bool rxrpc_charge_async_queue_bytes(struct rxrpc_local *local,
+ struct rxrpc_call *call,
+ struct rxrpc_sock *rx,
+ struct rxrpc_peer *peer,
+ long size,
+ enum rxrpc_async_queue_class class)
{
struct sock *sk = local->socket->sk;
- long rcvbuf;
- long queued;
+ atomic_long_t *counter;
+ long rcvbuf = READ_ONCE(sk->sk_rcvbuf);
+ long control_limit = rxrpc_async_queue_control_limit(rcvbuf);
+ long headroom = rxrpc_async_queue_data_headroom();
+ /* Allow an oversized packet and retain a maximum-size hole-filler. */
+ long queue_limit = min_t(long, rcvbuf, LONG_MAX - 2 * headroom) +
+ 2 * headroom;
+ long data_base_limit = rcvbuf - control_limit;
+ long data_limit = queue_limit - control_limit;
+ long call_data_base_limit = 0;
+ long limit, queued;
+ bool call_charged = false, charged = false;
+ bool reserve_headroom = false;
+
+ if (class >= RXRPC_ASYNC_QUEUE_DATA) {
+ if (WARN_ON_ONCE(!call))
+ goto drop;
+ if (class == RXRPC_ASYNC_QUEUE_DATA_OOS &&
+ !call->rx_async_data_headroom_reserved) {
+ if (!rxrpc_charge_async_queue_bytes(
+ local, call, NULL, NULL, headroom,
+ RXRPC_ASYNC_QUEUE_DATA_RESERVE))
+ return false;
+ call->rx_async_data_headroom_reserved = true;
+ reserve_headroom = true;
+ }
- rcvbuf = READ_ONCE(sk->sk_rcvbuf);
+ /* Keep one slow call from consuming the endpoint's DATA budget. */
+ call_data_base_limit = min_t(long,
+ rxrpc_async_queue_call_data_limit(call),
+ data_base_limit / 2);
+ if (class == RXRPC_ASYNC_QUEUE_DATA) {
+ limit = max_t(long, call_data_base_limit, headroom);
+ } else {
+ limit = call_data_base_limit + headroom;
+ if (class == RXRPC_ASYNC_QUEUE_DATA_HEAD)
+ limit += headroom;
+ }
+
+ if (!rxrpc_async_queue_charge_counter(
+ &call->rx_async_data_queue_bytes, size, limit))
+ goto drop;
+ call_charged = true;
+ }
for (;;) {
queued = atomic_long_read(&local->rx_async_queue_bytes);
- if (queued &&
- (queued >= rcvbuf ||
- (long)skb->truesize >= rcvbuf - queued)) {
- sk_drops_inc(sk);
+ if (queued >= queue_limit || size > queue_limit - queued)
+ goto drop;
+ if (atomic_long_cmpxchg(&local->rx_async_queue_bytes, queued,
+ queued + size) == queued)
+ break;
+ }
+
+ switch (class) {
+ case RXRPC_ASYNC_QUEUE_CONTROL:
+ counter = &local->rx_async_control_queue_bytes;
+ limit = control_limit - control_limit / 2;
+ charged = rxrpc_async_queue_charge_counter(counter, size, limit);
+ break;
+ case RXRPC_ASYNC_QUEUE_OOB:
+ counter = &local->rx_async_oob_queue_bytes;
+ limit = control_limit / 2;
+ charged = rxrpc_async_queue_charge_counter(counter, size, limit);
+ if (charged) {
+ /* Keep the per-owner cap at least one jumbo-sized skb. */
+ long owner_limit = max_t(long, control_limit / 4,
+ SKB_TRUESIZE(RXRPC_JUMBO(1)));
+
+ if (!rx || !peer) {
+ atomic_long_sub(size, counter);
+ charged = false;
+ } else if (!rxrpc_async_queue_charge_counter(
+ &peer->rx_async_oob_queue_bytes,
+ size, owner_limit)) {
+ atomic_long_sub(size, counter);
+ charged = false;
+ } else if (!rxrpc_async_queue_charge_counter(
+ &rx->rx_async_oob_queue_bytes,
+ size, owner_limit)) {
+ atomic_long_sub(size, &peer->rx_async_oob_queue_bytes);
+ atomic_long_sub(size, counter);
+ charged = false;
+ }
+ }
+ break;
+ case RXRPC_ASYNC_QUEUE_DATA:
+ counter = &local->rx_async_data_queue_bytes;
+ /* Leave one maximum-size packet for an OOS hole-filler. */
+ limit = data_limit - headroom;
+ charged = rxrpc_async_queue_charge_counter(counter, size, limit);
+ break;
+ case RXRPC_ASYNC_QUEUE_DATA_HEAD:
+ counter = &local->rx_async_data_queue_bytes;
+ limit = data_limit;
+ charged = rxrpc_async_queue_charge_counter(counter, size, limit);
+ break;
+ case RXRPC_ASYNC_QUEUE_DATA_RESERVE:
+ case RXRPC_ASYNC_QUEUE_DATA_OOS:
+ counter = &local->rx_async_data_oos_queue_bytes;
+ limit = data_base_limit + headroom;
+ charged = rxrpc_async_queue_charge_counter(counter, size, limit);
+ if (charged &&
+ !rxrpc_async_queue_charge_counter(
+ &local->rx_async_data_queue_bytes, size,
+ data_limit - headroom)) {
+ atomic_long_sub(size, counter);
+ charged = false;
+ }
+ break;
+ default:
+ break;
+ }
+
+ if (charged)
+ return true;
+ atomic_long_sub(size, &local->rx_async_queue_bytes);
+drop:
+ if (call_charged)
+ atomic_long_sub(size, &call->rx_async_data_queue_bytes);
+ if (reserve_headroom)
+ rxrpc_release_async_queue_headroom(call);
+ sk_drops_inc(sk);
+ return false;
+}
+
+struct rxrpc_async_queue_charge *rxrpc_alloc_async_queue_charge(struct sk_buff *skb)
+{
+ struct rxrpc_async_queue_charge *charge;
+
+ charge = kzalloc_obj(*charge, GFP_NOFS);
+ if (!charge)
+ return NULL;
+
+ refcount_set(&charge->ref, 1);
+ charge->size = skb->truesize + sizeof(*charge);
+ return charge;
+}
+
+static void rxrpc_uncharge_async_queue_bytes(struct rxrpc_local *local, long size,
+ enum rxrpc_async_queue_class class)
+{
+ if (class == RXRPC_ASYNC_QUEUE_DATA_RESERVE ||
+ class == RXRPC_ASYNC_QUEUE_DATA_OOS)
+ atomic_long_sub(size, &local->rx_async_data_oos_queue_bytes);
+
+ switch (rxrpc_async_queue_account_class(class)) {
+ case RXRPC_ASYNC_QUEUE_CONTROL:
+ atomic_long_sub(size, &local->rx_async_control_queue_bytes);
+ break;
+ case RXRPC_ASYNC_QUEUE_OOB:
+ atomic_long_sub(size, &local->rx_async_oob_queue_bytes);
+ break;
+ case RXRPC_ASYNC_QUEUE_DATA:
+ atomic_long_sub(size, &local->rx_async_data_queue_bytes);
+ break;
+ default:
+ WARN_ON_ONCE(1);
+ break;
+ }
+ atomic_long_sub(size, &local->rx_async_queue_bytes);
+}
+
+void rxrpc_put_async_queue_charge(struct rxrpc_local *local,
+ struct rxrpc_call *call,
+ struct rxrpc_async_queue_charge *charge)
+{
+ if (charge && refcount_dec_and_test(&charge->ref)) {
+ if (charge->charged) {
+ rxrpc_uncharge_async_queue_bytes(local, charge->size,
+ charge->class);
+ if (WARN_ON_ONCE(!call)) {
+ kfree(charge);
+ return;
+ }
+ atomic_long_sub(charge->size,
+ &call->rx_async_data_queue_bytes);
+ }
+ kfree(charge);
+ }
+}
+
+/*
+ * Transfer an skb to asynchronous receive-memory accounting.
+ */
+bool rxrpc_charge_async_queue(struct rxrpc_local *local, struct rxrpc_call *call,
+ struct rxrpc_sock *rx, struct sk_buff *skb,
+ enum rxrpc_async_queue_class class)
+{
+ struct rxrpc_async_queue_charge *charge = NULL;
+ struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+ struct rxrpc_peer *peer = NULL;
+ long size = 0;
+
+ if (class == RXRPC_ASYNC_QUEUE_OOB) {
+ if (WARN_ON_ONCE(!rx || !sp->chall.conn))
return false;
+ peer = sp->chall.conn->peer;
+ }
+
+ if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
+ charge = sp->data.charge;
+
+ if (charge) {
+ bool clone = sp->hdr.flags & RXRPC_JUMBO_PACKET;
+
+ /* Account all clones of a jumbo against the first subpacket's slot. */
+ if (charge->charged)
+ class = charge->class;
+
+ if (!charge->charged) {
+ size = charge->size;
+ if (clone)
+ size += SKB_DATA_ALIGN(sizeof(struct sk_buff));
+ } else if (clone) {
+ size = SKB_DATA_ALIGN(sizeof(struct sk_buff));
}
- if (atomic_long_cmpxchg(&local->rx_async_queue_bytes, queued,
- queued + (long)skb->truesize) == queued) {
- rxrpc_skb_orphan_udp(skb);
- return true;
+ if (size &&
+ !rxrpc_charge_async_queue_bytes(local, call, rx, peer,
+ size, class))
+ return false;
+ if (!charge->charged) {
+ charge->class = class;
+ charge->charged = true;
}
+ refcount_inc(&charge->ref);
+ } else if (!rxrpc_charge_async_queue_bytes(local, call, rx, peer,
+ skb->truesize, class)) {
+ return false;
+ }
+
+ if (class == RXRPC_ASYNC_QUEUE_OOB) {
+ sp->chall.oob_owner = rx;
+ sock_hold(&rx->sk);
}
+ sp->async_queue_class = class;
+ rxrpc_skb_orphan_udp(skb);
+ return true;
}
/*
* Release the asynchronous receive charge on an skb.
*/
-void rxrpc_uncharge_async_queue(struct rxrpc_local *local, struct sk_buff *skb)
+void rxrpc_uncharge_async_queue(struct rxrpc_local *local,
+ struct rxrpc_call *call, struct sk_buff *skb)
{
- atomic_long_sub(skb->truesize, &local->rx_async_queue_bytes);
+ struct rxrpc_async_queue_charge *charge;
+ struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+ enum rxrpc_async_queue_class class = sp->async_queue_class;
+ long size;
+
+ if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA && WARN_ON_ONCE(!call))
+ return;
+
+ if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA) {
+ charge = sp->data.charge;
+ if (charge) {
+ sp->data.charge = NULL;
+ if (sp->hdr.flags & RXRPC_JUMBO_PACKET) {
+ size = SKB_DATA_ALIGN(sizeof(struct sk_buff));
+ rxrpc_uncharge_async_queue_bytes(local, size, class);
+ atomic_long_sub(size,
+ &call->rx_async_data_queue_bytes);
+ }
+ rxrpc_put_async_queue_charge(local, call, charge);
+ return;
+ }
+
+ size = skb->truesize;
+ rxrpc_uncharge_async_queue_bytes(local, size, class);
+ atomic_long_sub(size, &call->rx_async_data_queue_bytes);
+ return;
+ }
+
+ size = skb->truesize;
+ rxrpc_uncharge_async_queue_bytes(local, size, class);
+ if (class == RXRPC_ASYNC_QUEUE_OOB) {
+ struct rxrpc_sock *rx = sp->chall.oob_owner;
+ struct rxrpc_peer *peer;
+
+ if (WARN_ON_ONCE(!rx || !sp->chall.conn))
+ return;
+ peer = sp->chall.conn->peer;
+ atomic_long_sub(size, &peer->rx_async_oob_queue_bytes);
+ atomic_long_sub(size, &rx->rx_async_oob_queue_bytes);
+ sp->chall.oob_owner = NULL;
+ sock_put(&rx->sk);
+ }
+}
+
+/*
+ * Purge DATA from a call's receive queue and release its per-local charge.
+ */
+void rxrpc_purge_data_queue(struct rxrpc_call *call, struct sk_buff_head *list)
+{
+ struct sk_buff *skb;
+
+ while ((skb = skb_dequeue(list)) != NULL) {
+ rxrpc_uncharge_async_queue(call->local, call, skb);
+ rxrpc_free_skb(skb, rxrpc_skb_put_purge);
+ }
+ if (list == &call->rx_oos_queue)
+ rxrpc_release_async_queue_headroom(call);
}
/*
@@ -106,7 +456,7 @@ void rxrpc_purge_queue(struct sk_buff_head *list)
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_uncharge_async_queue(sp->poke_conn->local, NULL, skb);
rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke);
}
rxrpc_skb_orphan_udp(skb);
--
2.43.0
More information about the linux-afs
mailing list