FAILED: patch "[PATCH] rxrpc: Fix socket notification race" failed to apply to 6.12-stable tree
Subsystems (3)
[email protected]
maintainerThe patch below does not apply to the 6.12-stable tree.
If someone wants it applied there, or to any other stable or longterm tree, then please email the backport, including the original git commit id to <[email protected]>.
To reproduce the conflict and resubmit, you may use the following commands:
git fetch https://git.kernel.org/pub/scm/linux/kernel/git/stable/linux.git/ linux-6.12.y
git checkout FETCH_HEAD
git cherry-pick -x e66f8f32f50116670dbbee5bc9e692cd2cd0c8f8
# <resolve conflicts, build, test, etc.>
git commit -s
git send-email --to '<[email protected]>' --in-reply-to '2026072023-cricket-devoutly-1a61@gregkh' --subject-prefix 'PATCH 6.12.y' 'HEAD^..'
Possible dependencies:
thanks,
greg k-h
------------------ original commit in Linus's tree ------------------
From e66f8f32f50116670dbbee5bc9e692cd2cd0c8f8 Mon Sep 17 00:00:00 2001
From: David Howells <[email protected]>
Date: Wed, 24 Jun 2026 17:38:16 +0100
Subject: [PATCH] rxrpc: Fix socket notification race
There's a race between rxrpc_recvmsg() and rxrpc_notify_socket(), whereby the latter's attempt to avoid disabling interrupts and taking the socket's recvmsg_lock if the call is already queued may happen simultaneously with the former's discarding of a call that has nothing queued.
Fix this by removing the shortcut. Note that this only affects userspace's use of AF_RXRPC; the AFS filesystem driver doesn't use the socket queue.
Fixes 248f219cb8bc ("rxrpc: Rewrite the data and ack handling code")
Signed-off-by David Howells <[email protected]>
cc Marc Dionne <[email protected]>
cc Jeffrey Altman <[email protected]>
cc Simon Horman <[email protected]>
Signed-off-by Jakub Kicinski <[email protected]>
net/rxrpc/recvmsg.c +0 -2
@@ -27,8 +27,6 @@ void rxrpc_notify_socket(struct rxrpc_call *call)_enter ("%d" , call->debug_id); -if (!list_empty (&call->recvmsg_link)) -return ;if (test_bit (RXRPC_CALL_RELEASED , &call->flags)) {rxrpc_see_call (call, rxrpc_call_see_notify_released);return ;
Sasha Levin
maintainerPATCH 1/5 rxrpc: Fix CPU time starvation in I/O thread
From: David Howells <[email protected]>
[ Upstream commit 9e3cccd176b5ec6ff78693287fb03097e453e69c ]
Starvation can happen in the rxrpc I/O thread because it goes back to the top of the I/O loop after it does any one thing without trying to give any other connection or call CPU time. Also, because it processes one call packet at a time, it tries to do the retransmission loop after each ACK without checking to see if there are other ACKs already in the queue that can update the SACK state.
Fix this by:
(1) Add a received-packet queue on each call.
(2) Distribute packets from the master Rx queue to the individual call, conn and error queues and 'poking' calls to add them to the attend queue first thing in the I/O thread.
(3) Go through all the attention-seeking connections and calls before going back to the top of the I/O thread. Each queue is extracted as a whole and then gone through so that new additions to insert themselves into the queue.
(4) Make the call event handler go through all the packets currently on the call's rx_queue before transmitting and retransmitting DATA packets.
(5) Drop the skb argument from the call event handler as this is now replaced with the rx_queue. Instead, keep track of whether we received a packet or an ACK for the tests that used to rely on that.
Signed-off-by David Howells <[email protected]>
cc Marc Dionne <[email protected]>
Signed-off-by Jakub Kicinski <[email protected]>
Stable-dep-of: e66f8f32f501 ("rxrpc: Fix socket notification race")
Signed-off-by Sasha Levin <[email protected]>
--- include/trace/events/rxrpc.h | 2 + net/rxrpc/ar-internal.h | 10 +++- net/rxrpc/call_accept.c | 2 +- net/rxrpc/call_event.c | 31 ++++++----- net/rxrpc/call_object.c | 2 + net/rxrpc/conn_client.c | 12 ++-- net/rxrpc/input.c | 2 +- net/rxrpc/io_thread.c | 104 ++++++++++++++++++----------------- net/rxrpc/peer_event.c | 2 +- 9 files changed, 93 insertions(+), 74 deletions(-)
include/trace/events/rxrpc.h +2 -0
@@ -121,6 +121,7 @@ EM(rxrpc_call_poke_conn_abort, "Conn-abort") \ EM(rxrpc_call_poke_error, "Error") \ EM(rxrpc_call_poke_idle, "Idle") \ + EM(rxrpc_call_poke_rx_packet, "Rx-packet") \ EM(rxrpc_call_poke_set_timeout, "Set-timo") \ EM(rxrpc_call_poke_start, "Start") \EM (rxrpc_call_poke_timer,"Timer" ) \
@@ -139,6 +140,7 @@ EM(rxrpc_skb_new_error_report, "NEW error-rpt") \ EM(rxrpc_skb_new_jumbo_subpacket, "NEW jumbo-sub") \ EM(rxrpc_skb_new_unshared, "NEW unshared ") \ + EM(rxrpc_skb_put_call_rx, "PUT call-rx ") \ EM(rxrpc_skb_put_conn_secured, "PUT conn-secd") \ EM(rxrpc_skb_put_conn_work, "PUT conn-work") \EM (rxrpc_skb_put_error_report,"PUT error-rep" ) \
net/rxrpc/ar-internal.h +9 -1
@@ -684,6 +684,7 @@ struct rxrpc_call {/* Received data tracking */ 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 */ void *rx_dec_buffer;/* Decryption buffer */ unsigned short rx_dec_bsize;/* rx_dec_buffer size */
@@ -872,7 +873,7 @@ void rxrpc_propose_delay_ACK(struct rxrpc_call *, rxrpc_serial_t,void rxrpc_shrink_call_tx_buffer (struct rxrpc_call *);void rxrpc_resend (struct rxrpc_call *call,struct sk_buff *ack_skb); -bool rxrpc_input_call_event (struct rxrpc_call *call,struct sk_buff *skb); +bool rxrpc_input_call_event (struct rxrpc_call *call);/* * call_object.c
@@ -1321,6 +1322,13 @@ static inline bool after_eq(u32 seq1, u32 seq2)return (s32)(seq1 - seq2) >=0 ; } +static inline void rxrpc_queue_rx_call_packet (struct rxrpc_call *call,struct sk_buff *skb) +{ +rxrpc_get_skb (skb, rxrpc_skb_get_call_rx); +__skb_queue_tail (&call->rx_queue, skb); +rxrpc_poke_call (call, rxrpc_call_poke_rx_packet); +} +/* * debug tracing */
net/rxrpc/call_accept.c +1 -1
@@ -413,7 +413,7 @@ bool rxrpc_new_incoming_call(struct rxrpc_local *local, }_leave (" = %p{%d}" , call, call->debug_id); -rxrpc_input_call_event (call, skb); +rxrpc_queue_rx_call_packet (call, skb);rxrpc_put_call (call, rxrpc_call_put_input);return true;
net/rxrpc/call_event.c +17 -14
@@ -315,10 +315,11 @@ static void rxrpc_send_initial_ping(struct rxrpc_call *call)/* * Handle retransmission and deferred ACK/abort generation. */ -bool rxrpc_input_call_event (struct rxrpc_call *call,struct sk_buff *skb) +bool rxrpc_input_call_event (struct rxrpc_call *call) { +struct sk_buff *skb;ktime_t now, t; -bool resend = false; +bool resend = false, saw_ack = false;s32 abort_code;rxrpc_see_call (call, rxrpc_call_see_input);
@@ -328,9 +329,6 @@ bool rxrpc_input_call_event(struct rxrpc_call *call, struct sk_buff *skb) call->debug_id, rxrpc_call_states[__rxrpc_call_state (call)], call->events); -if (__rxrpc_call_is_complete (call)) - goto out; -/* Handle abort request locklessly, vs rxrpc_propose_abort(). */ abort_code =smp_load_acquire (&call->send_abort);if (abort_code) {
@@ -339,11 +337,20 @@ bool rxrpc_input_call_event(struct rxrpc_call *call, struct sk_buff *skb) goto out; } -if (skb && skb->mark ==RXRPC_SKB_MARK_ERROR ) - goto out; +while ((skb =__skb_dequeue (&call->rx_queue))) { +struct rxrpc_skb_priv *sp =rxrpc_skb (skb); + +if (__rxrpc_call_is_complete (call) || + skb->mark ==RXRPC_SKB_MARK_ERROR ) { +rxrpc_free_skb (skb, rxrpc_skb_put_call_rx); + goto out; + } + + saw_ack |= sp->hdr.type ==RXRPC_PACKET_TYPE_ACK ; -if (skb)rxrpc_input_call_packet (call, skb); +rxrpc_free_skb (skb, rxrpc_skb_put_call_rx); + }/* If we see our async-event poke, check for timeout trippage. */ now =ktime_get_real ();
@@ -409,12 +416,8 @@ bool rxrpc_input_call_event(struct rxrpc_call *call, struct sk_buff *skb) rxrpc_propose_ack_ping_for_keepalive); } -if (skb) { -struct rxrpc_skb_priv *sp =rxrpc_skb (skb); - -if (sp->hdr.type ==RXRPC_PACKET_TYPE_ACK ) -rxrpc_congestion_degrade (call); - } +if (saw_ack) +rxrpc_congestion_degrade (call);if (test_and_clear_bit (RXRPC_CALL_EV_INITIAL_PING , &call->events))rxrpc_send_initial_ping (call);
net/rxrpc/call_object.c +2 -0
@@ -147,6 +147,7 @@ struct rxrpc_call *rxrpc_alloc_call(struct rxrpc_sock *rx, gfp_t gfp,INIT_LIST_HEAD (&call->attend_link);INIT_LIST_HEAD (&call->tx_sendmsg);INIT_LIST_HEAD (&call->tx_buffer); +skb_queue_head_init (&call->rx_queue);skb_queue_head_init (&call->recvmsg_queue);skb_queue_head_init (&call->rx_oos_queue);init_waitqueue_head (&call->waitq);
@@ -535,6 +536,7 @@ void rxrpc_get_call(struct rxrpc_call *call, enum rxrpc_call_trace why)static void rxrpc_cleanup_ring (struct rxrpc_call *call) {rxrpc_purge_queue (&call->recvmsg_queue); +rxrpc_purge_queue (&call->rx_queue);rxrpc_purge_queue (&call->rx_oos_queue);kfree (call->rx_dec_buffer); }
net/rxrpc/conn_client.c +7 -5
@@ -508,16 +508,18 @@ static void rxrpc_activate_channels(struct rxrpc_bundle *bundle)void rxrpc_connect_client_calls (struct rxrpc_local *local) {struct rxrpc_call *call; +LIST_HEAD (new_client_calls); -while ((call =list_first_entry_or_null (&local->new_client_calls, - struct rxrpc_call, wait_link)) - ) { +spin_lock (&local->client_call_lock); +list_splice_tail_init (&local->new_client_calls, &new_client_calls); +spin_unlock (&local->client_call_lock); + +while ((call =list_first_entry_or_null (&new_client_calls, + struct rxrpc_call, wait_link))) {struct rxrpc_bundle *bundle = call->bundle; -spin_lock (&local->client_call_lock);list_move_tail (&call->wait_link, &bundle->waiting_calls);rxrpc_see_call (call, rxrpc_call_see_waiting_call); -spin_unlock (&local->client_call_lock);if (rxrpc_bundle_has_space (bundle))rxrpc_activate_channels (bundle);
net/rxrpc/input.c +1 -1
@@ -1122,5 +1122,5 @@ void rxrpc_implicit_end_call(struct rxrpc_call *call, struct sk_buff *skb)break ; } -rxrpc_input_call_event (call, skb); +rxrpc_input_call_event (call); }
net/rxrpc/io_thread.c +53 -51
145 lines not on this page · raw message
net/rxrpc/peer_event.c +1 -1
@@ -216,7 +216,7 @@ static void rxrpc_distribute_error(struct rxrpc_peer *peer, struct sk_buff *skb,rxrpc_see_call (call, rxrpc_call_see_distribute_error);rxrpc_set_call_completion (call, compl,0 , -err); -rxrpc_input_call_event (call, skb); +rxrpc_input_call_event (call);spin_lock (&peer->lock); }
Signature · 1 line
2.53.0
Sasha Levin
maintainerPATCH 2/5 rxrpc: Don't need barrier for ->tx_bottom and ->acks_hard_ack
From: David Howells <[email protected]>
[ Upstream commit 6396b48ac0a77165f9c2c40ab03d6c8188c89739 ]
We don't need a barrier for the ->tx_bottom value (which indicates the lowest sequence still in the transmission queue) and the ->acks_hard_ack value (which tracks the DATA packets hard-ack'd by the latest ACK packet received and thus indicates which DATA packets can now be discarded) as the app thread doesn't use either value as a reference to memory to access. Rather, the app thread merely uses these as a guide to how much space is available in the transmission queue
Change the code to use READ/WRITE_ONCE() instead.
Also, change rxrpc_check_tx_space() to use the same value for tx_bottom throughout.
Signed-off-by David Howells <[email protected]>
cc Marc Dionne <[email protected]>
Signed-off-by Jakub Kicinski <[email protected]>
Stable-dep-of: e66f8f32f501 ("rxrpc: Fix socket notification race")
Signed-off-by Sasha Levin <[email protected]>
--- net/rxrpc/sendmsg.c | 8 +++++--- net/rxrpc/txbuf.c | 4 ++-- 2 files changed, 7 insertions(+), 5 deletions(-)
net/rxrpc/sendmsg.c +5 -3
@@ -93,9 +93,11 @@ static int rxrpc_wait_to_be_connected(struct rxrpc_call *call, long *timeo)*/ static bool rxrpc_check_tx_space (struct rxrpc_call *call,rxrpc_seq_t *_tx_win) { +rxrpc_seq_t tx_bottom =READ_ONCE (call->tx_bottom); +if (_tx_win) - *_tx_win = call->tx_bottom; -return call->tx_prepared - call->tx_bottom <256 ; + *_tx_win = tx_bottom; +return call->tx_prepared - tx_bottom <256 ; }/*
@@ -137,7 +139,7 @@ static int rxrpc_wait_for_tx_window_waitall(struct rxrpc_sock *rx, rtt =2 ; timeout = rtt; - tx_start =smp_load_acquire (&call->acks_hard_ack); + tx_start =READ_ONCE (call->acks_hard_ack);for (;;) {set_current_state (TASK_UNINTERRUPTIBLE );
net/rxrpc/txbuf.c +2 -2
@@ -216,14 +216,14 @@ void rxrpc_shrink_call_tx_buffer(struct rxrpc_call *call)while ((txb =list_first_entry_or_null (&call->tx_buffer, struct rxrpc_txbuf, call_link))) { - hard_ack =smp_load_acquire (&call->acks_hard_ack); + hard_ack = call->acks_hard_ack;if (before (hard_ack, txb->seq))break ;if (txb->seq != call->tx_bottom +1 )rxrpc_see_txbuf (txb, rxrpc_txbuf_see_out_of_step); ASSERTCMP(txb->seq, ==, call->tx_bottom + 1); -smp_store_release (&call->tx_bottom, call->tx_bottom +1 ); +WRITE_ONCE (call->tx_bottom, call->tx_bottom +1 );list_del_rcu (&txb->call_link);trace_rxrpc_txqueue (call, rxrpc_txqueue_dequeue);
Signature · 1 line
2.53.0
Sasha Levin
maintainerPATCH 3/5 rxrpc: Use irq-disabling spinlocks between app and I/O thread
In reply to Sasha LevinFrom: David Howells <[email protected]>
[ Upstream commit a2ea9a9072607c2fd6442bd1ffb4dbdbf882aed7 ]
Where a spinlock is used by both the application thread and the I/O thread, use irq-disabling locking so that an interrupt taken on the app thread doesn't also slow down the I/O thread.
Signed-off-by David Howells <[email protected]>
cc Marc Dionne <[email protected]>
Signed-off-by Jakub Kicinski <[email protected]>
Stable-dep-of: e66f8f32f501 ("rxrpc: Fix socket notification race")
Signed-off-by Sasha Levin <[email protected]>
--- net/rxrpc/af_rxrpc.c | 4 ++-- net/rxrpc/call_accept.c | 20 ++++++++++---------- net/rxrpc/call_object.c | 14 +++++++------- net/rxrpc/conn_client.c | 12 ++++++------ net/rxrpc/conn_event.c | 14 +++++++------- net/rxrpc/conn_object.c | 8 ++++---- net/rxrpc/input.c | 5 +---- net/rxrpc/io_thread.c | 8 ++++---- net/rxrpc/peer_event.c | 8 ++++---- net/rxrpc/peer_object.c | 1 + net/rxrpc/recvmsg.c | 22 +++++++++++----------- net/rxrpc/security.c | 4 ++-- 12 files changed, 59 insertions(+), 61 deletions(-)
net/rxrpc/af_rxrpc.c +2 -2
@@ -408,9 +408,9 @@ void rxrpc_kernel_shutdown_call(struct socket *sock, struct rxrpc_call *call)/* Make sure we're not going to call back into a kernel service */ if (call->notify_rx) { -spin_lock (&call->notify_lock); +spin_lock_irq (&call->notify_lock); call->notify_rx = rxrpc_dummy_notify_rx; -spin_unlock (&call->notify_lock); +spin_unlock_irq (&call->notify_lock); } }mutex_unlock (&call->user_mutex);
net/rxrpc/call_accept.c +10 -10
@@ -189,8 +189,8 @@ void rxrpc_discard_prealloc(struct rxrpc_sock *rx)/* Make sure that there aren't any incoming calls in progress before we * clear the preallocation buffers. */ -spin_lock (&rx->incoming_lock); -spin_unlock (&rx->incoming_lock); +spin_lock_irq (&rx->incoming_lock); +spin_unlock_irq (&rx->incoming_lock); head = b->peer_backlog_head; tail = b->peer_backlog_tail;
@@ -348,7 +348,7 @@ bool rxrpc_new_incoming_call(struct rxrpc_local *local,if (sp->hdr.type !=RXRPC_PACKET_TYPE_DATA )return rxrpc_protocol_error (skb, rxrpc_eproto_no_service_call); -read_lock (&local->services_lock); +read_lock_irq (&local->services_lock);/* Weed out packets to services we're not offering. Packets that would * begin a call are explicitly rejected and the rest are just
@@ -404,12 +404,12 @@ bool rxrpc_new_incoming_call(struct rxrpc_local *local,spin_unlock (&conn->state_lock);spin_unlock (&rx->incoming_lock); -read_unlock (&local->services_lock); +read_unlock_irq (&local->services_lock);if (hlist_unhashed (&call->error_link)) { -spin_lock (&call->peer->lock); +spin_lock_irq (&call->peer->lock);hlist_add_head (&call->error_link, &call->peer->error_targets); -spin_unlock (&call->peer->lock); +spin_unlock_irq (&call->peer->lock); }_leave (" = %p{%d}" , call, call->debug_id);
@@ -418,20 +418,20 @@ bool rxrpc_new_incoming_call(struct rxrpc_local *local,return true; unsupported_service: -read_unlock (&local->services_lock); +read_unlock_irq (&local->services_lock);return rxrpc_direct_abort (skb, rxrpc_abort_service_not_offered,RX_INVALID_OPERATION , -EOPNOTSUPP ); unsupported_security: -read_unlock (&local->services_lock); +read_unlock_irq (&local->services_lock);return rxrpc_direct_abort (skb, rxrpc_abort_service_not_offered,RX_INVALID_OPERATION , -EKEYREJECTED ); no_call:spin_unlock (&rx->incoming_lock); -read_unlock (&local->services_lock); +read_unlock_irq (&local->services_lock);_leave (" = f [%u]" , skb->mark);return false; discard: -read_unlock (&local->services_lock); +read_unlock_irq (&local->services_lock);return true; }
net/rxrpc/call_object.c +7 -7
@@ -48,7 +48,7 @@ void rxrpc_poke_call(struct rxrpc_call *call, enum rxrpc_call_poke_trace what)bool busy;if (!test_bit (RXRPC_CALL_DISCONNECTED , &call->flags)) { -spin_lock_bh (&local->lock); +spin_lock_irq (&local->lock); busy = !list_empty (&call->attend_link);trace_rxrpc_poke_call (call, busy, what);if (!busy && !rxrpc_try_get_call (call, rxrpc_call_get_poke))
@@ -56,7 +56,7 @@ void rxrpc_poke_call(struct rxrpc_call *call, enum rxrpc_call_poke_trace what)if (!busy) {list_add_tail (&call->attend_link, &local->call_attend_q); } -spin_unlock_bh (&local->lock); +spin_unlock_irq (&local->lock);if (!busy)rxrpc_wake_up_io_thread (local); }
@@ -303,9 +303,9 @@ static int rxrpc_connect_call(struct rxrpc_call *call, gfp_t gfp)trace_rxrpc_client (NULL ,-1 , rxrpc_client_queue_new_call);rxrpc_get_call (call, rxrpc_call_get_io_thread); -spin_lock (&local->client_call_lock); +spin_lock_irq (&local->client_call_lock);list_add_tail (&call->wait_link, &local->new_client_calls); -spin_unlock (&local->client_call_lock); +spin_unlock_irq (&local->client_call_lock);rxrpc_wake_up_io_thread (local);return 0 ;
@@ -435,7 +435,7 @@ struct rxrpc_call *rxrpc_new_client_call(struct rxrpc_sock *rx,/* * Set up an incoming call. call->conn points to the connection. -* This is called in BH context and isn't allowed to fail. +* This is called with interrupts disabled and isn't allowed to fail. */ void rxrpc_incoming_call(struct rxrpc_sock *rx, struct rxrpc_call *call,
@@ -560,7 +560,7 @@ void rxrpc_release_call(struct rxrpc_sock *rx, struct rxrpc_call *call)rxrpc_put_call_slot (call);/* Make sure we don't get any more notifications */ -spin_lock (&rx->recvmsg_lock); +spin_lock_irq (&rx->recvmsg_lock);if (!list_empty (&call->recvmsg_link)) { _debug("unlinking once-pending call %p { e=%lx f=%lx }",
@@ -573,7 +573,7 @@ void rxrpc_release_call(struct rxrpc_sock *rx, struct rxrpc_call *call) call->recvmsg_link.next =NULL ; call->recvmsg_link.prev =NULL ; -spin_unlock (&rx->recvmsg_lock); +spin_unlock_irq (&rx->recvmsg_lock);if (put)rxrpc_put_call (call, rxrpc_call_put_unnotify);
net/rxrpc/conn_client.c +6 -6
@@ -510,9 +510,9 @@ void rxrpc_connect_client_calls(struct rxrpc_local *local)struct rxrpc_call *call;LIST_HEAD (new_client_calls); -spin_lock (&local->client_call_lock); +spin_lock_irq (&local->client_call_lock);list_splice_tail_init (&local->new_client_calls, &new_client_calls); -spin_unlock (&local->client_call_lock); +spin_unlock_irq (&local->client_call_lock);while ((call =list_first_entry_or_null (&new_client_calls, struct rxrpc_call, wait_link))) {
@@ -547,9 +547,9 @@ void rxrpc_expose_client_call(struct rxrpc_call *call)set_bit (RXRPC_CONN_DONT_REUSE , &conn->flags);trace_rxrpc_client (conn, channel, rxrpc_client_exposed); -spin_lock (&call->peer->lock); +spin_lock_irq (&call->peer->lock);hlist_add_head (&call->error_link, &call->peer->error_targets); -spin_unlock (&call->peer->lock); +spin_unlock_irq (&call->peer->lock); } }
@@ -590,9 +590,9 @@ void rxrpc_disconnect_client_call(struct rxrpc_bundle *bundle, struct rxrpc_call ASSERTCMP(call->call_id, ==, 0);ASSERT (!test_bit (RXRPC_CALL_EXPOSED , &call->flags));/* May still be on ->new_client_calls. */ -spin_lock (&local->client_call_lock); +spin_lock_irq (&local->client_call_lock);list_del_init (&call->wait_link); -spin_unlock (&local->client_call_lock); +spin_unlock_irq (&local->client_call_lock);return ; }
net/rxrpc/conn_event.c +7 -7
@@ -26,7 +26,7 @@ static bool rxrpc_set_conn_aborted(struct rxrpc_connection *conn, struct sk_buffbool aborted = false;if (conn->state !=RXRPC_CONN_ABORTED ) { -spin_lock (&conn->state_lock); +spin_lock_irq (&conn->state_lock);if (conn->state !=RXRPC_CONN_ABORTED ) { conn->abort_code = abort_code; conn->error = err;
@@ -37,7 +37,7 @@ static bool rxrpc_set_conn_aborted(struct rxrpc_connection *conn, struct sk_buffset_bit (RXRPC_CONN_EV_ABORT_CALLS , &conn->events); aborted = true; } -spin_unlock (&conn->state_lock); +spin_unlock_irq (&conn->state_lock); }return aborted;
@@ -268,12 +268,12 @@ static int rxrpc_process_event(struct rxrpc_connection *conn,return conn->security->respond_to_challenge (conn, skb);case RXRPC_PACKET_TYPE_RESPONSE : -spin_lock (&conn->state_lock); +spin_lock_irq (&conn->state_lock);if (conn->state !=RXRPC_CONN_SERVICE_CHALLENGING ) { -spin_unlock (&conn->state_lock); +spin_unlock_irq (&conn->state_lock);return 0 ; } -spin_unlock (&conn->state_lock); +spin_unlock_irq (&conn->state_lock); ret =rxrpc_verify_response (conn, skb);if (ret <0 )
@@ -284,12 +284,12 @@ static int rxrpc_process_event(struct rxrpc_connection *conn,if (ret <0 )return ret; -spin_lock (&conn->state_lock); +spin_lock_irq (&conn->state_lock);if (conn->state ==RXRPC_CONN_SERVICE_CHALLENGING ) { conn->state =RXRPC_CONN_SERVICE ; secured = true; } -spin_unlock (&conn->state_lock); +spin_unlock_irq (&conn->state_lock);if (secured) {/* Offload call state flipping to the I/O thread. As
net/rxrpc/conn_object.c +4 -4
@@ -31,13 +31,13 @@ void rxrpc_poke_conn(struct rxrpc_connection *conn, enum rxrpc_conn_trace why)if (WARN_ON_ONCE (!local))return ; -spin_lock_bh (&local->lock); +spin_lock_irq (&local->lock); busy = !list_empty (&conn->attend_link);if (!busy) {rxrpc_get_connection (conn, why);list_add_tail (&conn->attend_link, &local->conn_attend_q); } -spin_unlock_bh (&local->lock); +spin_unlock_irq (&local->lock);rxrpc_wake_up_io_thread (local); }
@@ -197,9 +197,9 @@ void rxrpc_disconnect_call(struct rxrpc_call *call) call->peer->cong_ssthresh = call->cong_ssthresh;if (!hlist_unhashed (&call->error_link)) { -spin_lock (&call->peer->lock); +spin_lock_irq (&call->peer->lock);hlist_del_init (&call->error_link); -spin_unlock (&call->peer->lock); +spin_unlock_irq (&call->peer->lock); }if (rxrpc_is_client_call (call)) {
net/rxrpc/input.c +1 -4
@@ -365,7 +365,7 @@ static void rxrpc_input_queue_data(struct rxrpc_call *call, struct sk_buff *skb,struct rxrpc_skb_priv *sp =rxrpc_skb (skb);bool last = sp->hdr.flags &RXRPC_LAST_PACKET ; -__skb_queue_tail (&call->recvmsg_queue, skb); +skb_queue_tail (&call->recvmsg_queue, skb);rxrpc_input_update_ack_window (call, window, wtop);trace_rxrpc_receive (call, last ? why +1 : why, sp->hdr.serial, sp->hdr.seq);if (last)
@@ -442,7 +442,6 @@ static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb,rxrpc_get_skb (skb, rxrpc_skb_get_to_recvmsg); -spin_lock (&call->recvmsg_queue.lock);rxrpc_input_queue_data (call, skb, window, wtop, rxrpc_receive_queue); *_notify = true;
@@ -464,8 +463,6 @@ static void rxrpc_input_data_one(struct rxrpc_call *call, struct sk_buff *skb, rxrpc_receive_queue_oos); } -spin_unlock (&call->recvmsg_queue.lock); - call->ackr_sack_base = sack; }else {unsigned int slot;
net/rxrpc/io_thread.c +4 -4
@@ -475,9 +475,9 @@ int rxrpc_io_thread(void *data) }/* Deal with connections that want immediate attention. */ -spin_lock_bh (&local->lock); +spin_lock_irq (&local->lock);list_splice_tail_init (&local->conn_attend_q, &conn_attend_q); -spin_unlock_bh (&local->lock); +spin_unlock_irq (&local->lock); while ((conn = list_first_entry_or_null(&conn_attend_q, struct rxrpc_connection,
@@ -494,9 +494,9 @@ int rxrpc_io_thread(void *data)rxrpc_discard_expired_client_conns (local);/* Deal with calls that want immediate attention. */ -spin_lock_bh (&local->lock); +spin_lock_irq (&local->lock);list_splice_tail_init (&local->call_attend_q, &call_attend_q); -spin_unlock_bh (&local->lock); +spin_unlock_irq (&local->lock); while ((call = list_first_entry_or_null(&call_attend_q, struct rxrpc_call,
net/rxrpc/peer_event.c +4 -4
@@ -205,23 +205,23 @@ static void rxrpc_distribute_error(struct rxrpc_peer *peer, struct sk_buff *skb,struct rxrpc_call *call;HLIST_HEAD (error_targets); -spin_lock (&peer->lock); +spin_lock_irq (&peer->lock);hlist_move_list (&peer->error_targets, &error_targets);while (!hlist_empty (&error_targets)) { call =hlist_entry (error_targets.first, struct rxrpc_call, error_link);hlist_del_init (&call->error_link); -spin_unlock (&peer->lock); +spin_unlock_irq (&peer->lock);rxrpc_see_call (call, rxrpc_call_see_distribute_error);rxrpc_set_call_completion (call, compl,0 , -err);rxrpc_input_call_event (call); -spin_lock (&peer->lock); +spin_lock_irq (&peer->lock); } -spin_unlock (&peer->lock); +spin_unlock_irq (&peer->lock); }/*
net/rxrpc/peer_object.c +1 -0
@@ -304,6 +304,7 @@ static void rxrpc_free_peer(struct rxrpc_peer *peer)* Set up a new incoming peer. There shouldn't be any other matching peers * since we've already done a search in the list from the non-reentrant context * (the data_ready handler) that is the only place we can add new peers. +* Called with interrupts disabled. */ void rxrpc_new_incoming_peer (struct rxrpc_local *local,struct rxrpc_peer *peer) {
net/rxrpc/recvmsg.c +11 -11
@@ -36,16 +36,16 @@ void rxrpc_notify_socket(struct rxrpc_call *call) sk = &rx->sk;if (rx && sk->sk_state <RXRPC_CLOSE ) {if (call->notify_rx) { -spin_lock (&call->notify_lock); +spin_lock_irq (&call->notify_lock); call->notify_rx (sk, call, call->user_call_ID); -spin_unlock (&call->notify_lock); +spin_unlock_irq (&call->notify_lock); }else { -spin_lock (&rx->recvmsg_lock); +spin_lock_irq (&rx->recvmsg_lock);if (list_empty (&call->recvmsg_link)) {rxrpc_get_call (call, rxrpc_call_get_notify_socket);list_add_tail (&call->recvmsg_link, &rx->recvmsg_q); } -spin_unlock (&rx->recvmsg_lock); +spin_unlock_irq (&rx->recvmsg_lock);if (!sock_flag (sk,SOCK_DEAD )) {_debug ("call %ps" , sk->sk_data_ready);
@@ -379,14 +379,14 @@ int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,* We also want to weed out calls that got requeued whilst we were * shovelling data out. */ -spin_lock (&rx->recvmsg_lock); +spin_lock_irq (&rx->recvmsg_lock); l = rx->recvmsg_q.next; call = list_entry(l, struct rxrpc_call, recvmsg_link);if (!rxrpc_call_is_complete (call) &&skb_queue_empty (&call->recvmsg_queue)) {list_del_init (&call->recvmsg_link); -spin_unlock (&rx->recvmsg_lock); +spin_unlock_irq (&rx->recvmsg_lock);release_sock (&rx->sk);trace_rxrpc_recvmsg (call->debug_id, rxrpc_recvmsg_unqueue,0 );rxrpc_put_call (call, rxrpc_call_put_recvmsg);
@@ -407,7 +407,7 @@ int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,list_del_init (&call->recvmsg_link); elserxrpc_get_call (call ,rxrpc_call_get_recvmsg ); -spin_unlock (&rx->recvmsg_lock); +spin_unlock_irq (&rx->recvmsg_lock); call_debug_id = call->debug_id;trace_rxrpc_recvmsg (call_debug_id, rxrpc_recvmsg_dequeue,0 );
@@ -503,17 +503,17 @@ int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len, error_requeue_call:if (!(flags &MSG_PEEK )) { -spin_lock (&rx->recvmsg_lock); +spin_lock_irq (&rx->recvmsg_lock);if (list_empty (&call->recvmsg_link)) {list_add (&call->recvmsg_link, &rx->recvmsg_q);rxrpc_see_call (call, rxrpc_call_see_recvmsg_requeue); -spin_unlock (&rx->recvmsg_lock); +spin_unlock_irq (&rx->recvmsg_lock); }else if (list_is_first (&call->recvmsg_link, &rx->recvmsg_q)) { -spin_unlock (&rx->recvmsg_lock); +spin_unlock_irq (&rx->recvmsg_lock);rxrpc_put_call (call, rxrpc_call_see_recvmsg_requeue_first); }else {list_move (&call->recvmsg_link, &rx->recvmsg_q); -spin_unlock (&rx->recvmsg_lock); +spin_unlock_irq (&rx->recvmsg_lock);rxrpc_put_call (call, rxrpc_call_see_recvmsg_requeue_move); }trace_rxrpc_recvmsg (call_debug_id, rxrpc_recvmsg_requeue,0 );
net/rxrpc/security.c +2 -2
@@ -114,10 +114,10 @@ int rxrpc_init_client_conn_security(struct rxrpc_connection *conn)if (conn->state ==RXRPC_CONN_CLIENT_UNSECURED ) { ret = conn->security->init_connection_security (conn, token);if (ret ==0 ) { -spin_lock (&conn->state_lock); +spin_lock_irq (&conn->state_lock);if (conn->state ==RXRPC_CONN_CLIENT_UNSECURED ) conn->state =RXRPC_CONN_CLIENT ; -spin_unlock (&conn->state_lock); +spin_unlock_irq (&conn->state_lock); } }mutex_unlock (&conn->security_lock);
Signature · 1 line
2.53.0
Sasha Levin
maintainerPATCH 4/5 rxrpc: Fix notification vs call-release vs recvmsg
In reply to Sasha LevinFrom: David Howells <[email protected]>
[ Upstream commit 2fd895842d49c23137ae48252dd211e5d6d8a3ed ]
When a call is released, rxrpc takes the spinlock and removes it from ->recvmsg_q in an effort to prevent racing recvmsg() invocations from seeing the same call. Now, rxrpc_recvmsg() only takes the spinlock when actually removing a call from the queue; it doesn't, however, take it in the lead up to that when it checks to see if the queue is empty. It *does* hold the socket lock, which prevents a recvmsg/recvmsg race - but this doesn't prevent sendmsg from ending the call because sendmsg() drops the socket lock and relies on the call->user_mutex.
Fix this by firstly removing the bit in rxrpc_release_call() that dequeues the released call and, instead, rely on recvmsg() to simply discard released calls (done in a preceding fix).
Secondly, rxrpc_notify_socket() is abandoned if the call is already marked as released rather than trying to be clever by setting both pointers in call->recvmsg_link to NULL to trick list_empty(). This isn't perfect and can still race, resulting in a released call on the queue, but recvmsg() will now clean that up.
Fixes 17926a79320a ("[AF_RXRPC]: Provide secure RxRPC sockets for use by userspace and kernel both")
Signed-off-by David Howells <[email protected]>
Reviewed-by Jeffrey Altman <[email protected]>
cc Marc Dionne <[email protected]>
cc Junvyyang, Tencent Zhuque Lab <[email protected]>
cc LePremierHomme <[email protected]>
cc Simon Horman <[email protected]>
Signed-off-by Jakub Kicinski <[email protected]>
Stable-dep-of: e66f8f32f501 ("rxrpc: Fix socket notification race")
Signed-off-by Sasha Levin <[email protected]>
--- include/trace/events/rxrpc.h | 3 ++- net/rxrpc/call_object.c | 28 ++++++++++++---------------- net/rxrpc/recvmsg.c | 4 ++++ 3 files changed, 18 insertions(+), 17 deletions(-)
include/trace/events/rxrpc.h +2 -1
@@ -278,10 +278,10 @@ EM(rxrpc_call_put_poke, "PUT poke ") \ EM(rxrpc_call_put_recvmsg, "PUT recvmsg ") \ EM(rxrpc_call_put_recvmsg_peek_nowait, "PUT peek-nwt") \ + EM(rxrpc_call_put_release_recvmsg_q, "PUT rls-rcmq") \ EM(rxrpc_call_put_release_sock, "PUT rls-sock") \ EM(rxrpc_call_put_release_sock_tba, "PUT rls-sk-a") \ EM(rxrpc_call_put_sendmsg, "PUT sendmsg ") \ - EM(rxrpc_call_put_unnotify, "PUT unnotify") \ EM(rxrpc_call_put_userid_exists, "PUT u-exists") \ EM(rxrpc_call_put_userid, "PUT user-id ") \EM (rxrpc_call_see_accept,"SEE accept " ) \
@@ -294,6 +294,7 @@ EM(rxrpc_call_see_disconnected, "SEE disconn ") \ EM(rxrpc_call_see_distribute_error, "SEE dist-err") \ EM(rxrpc_call_see_input, "SEE input ") \ + EM(rxrpc_call_see_notify_released, "SEE nfy-rlsd") \ EM(rxrpc_call_see_recvmsg, "SEE recvmsg ") \ EM(rxrpc_call_see_recvmsg_requeue, "SEE recv-rqu") \EM (rxrpc_call_see_recvmsg_requeue_first,"SEE recv-rqF" ) \
net/rxrpc/call_object.c +12 -16
@@ -547,7 +547,7 @@ static void rxrpc_cleanup_ring(struct rxrpc_call *call)void rxrpc_release_call (struct rxrpc_sock *rx,struct rxrpc_call *call) {struct rxrpc_connection *conn = call->conn; -bool put = false, putu = false; +bool putu = false;_enter ("{%d,%d}" , call->debug_id,refcount_read (&call->ref));
@@ -559,23 +559,13 @@ void rxrpc_release_call(struct rxrpc_sock *rx, struct rxrpc_call *call)rxrpc_put_call_slot (call); -/* Make sure we don't get any more notifications */ +/* Note that at this point, the call may still be on or may have been +* added back on to the socket receive queue. recvmsg() must discard +* released calls. The CALL_RELEASED flag should prevent further +* notifications. +*/ spin_lock_irq (&rx->recvmsg_lock); - -if (!list_empty (&call->recvmsg_link)) { -_debug ("unlinking once-pending call %p { e=%lx f=%lx }" , - call, call->events, call->flags); -list_del (&call->recvmsg_link); - put = true; - } - -/* list_empty() must return false in rxrpc_notify_socket() */ - call->recvmsg_link.next =NULL ; - call->recvmsg_link.prev =NULL ; -spin_unlock_irq (&rx->recvmsg_lock); -if (put) -rxrpc_put_call (call, rxrpc_call_put_unnotify);write_lock (&rx->call_lock);
@@ -624,6 +614,12 @@ void rxrpc_release_calls_on_socket(struct rxrpc_sock *rx)rxrpc_put_call (call, rxrpc_call_put_release_sock); } +while ((call =list_first_entry_or_null (&rx->recvmsg_q, + struct rxrpc_call, recvmsg_link))) { +list_del_init (&call->recvmsg_link); +rxrpc_put_call (call, rxrpc_call_put_release_recvmsg_q); + } +_leave ("" ); }
net/rxrpc/recvmsg.c +4 -0
@@ -29,6 +29,10 @@ void rxrpc_notify_socket(struct rxrpc_call *call)if (!list_empty (&call->recvmsg_link))return ; +if (test_bit (RXRPC_CALL_RELEASED , &call->flags)) { +rxrpc_see_call (call, rxrpc_call_see_notify_released); +return ; + }rcu_read_lock ();
Signature · 1 line
2.53.0
Sasha Levin
maintainerPATCH 5/5 rxrpc: Fix socket notification race
In reply to Sasha LevinFrom: David Howells <[email protected]>
[ Upstream commit e66f8f32f50116670dbbee5bc9e692cd2cd0c8f8 ]
There's a race between rxrpc_recvmsg() and rxrpc_notify_socket(), whereby the latter's attempt to avoid disabling interrupts and taking the socket's recvmsg_lock if the call is already queued may happen simultaneously with the former's discarding of a call that has nothing queued.
Fix this by removing the shortcut. Note that this only affects userspace's use of AF_RXRPC; the AFS filesystem driver doesn't use the socket queue.
Fixes 248f219cb8bc ("rxrpc: Rewrite the data and ack handling code")
Signed-off-by David Howells <[email protected]>
cc Marc Dionne <[email protected]>
cc Jeffrey Altman <[email protected]>
cc Simon Horman <[email protected]>
Signed-off-by Jakub Kicinski <[email protected]>
Signed-off-by Sasha Levin <[email protected]>
--- net/rxrpc/recvmsg.c | 2 -- 1 file changed, 2 deletions(-)
net/rxrpc/recvmsg.c +0 -2
@@ -27,8 +27,6 @@ void rxrpc_notify_socket(struct rxrpc_call *call)_enter ("%d" , call->debug_id); -if (!list_empty (&call->recvmsg_link)) -return ;if (test_bit (RXRPC_CALL_RELEASED , &call->flags)) {rxrpc_see_call (call, rxrpc_call_see_notify_released);return ;
Signature · 1 line
2.53.0