[PATCH net-next 4/4] net/rds: acquire RDS_IN_XMIT in rds_tcp_reset_callbacks()

rds_tcp_reset_callbacks() quiesces the transmit path by setting the
path state to RDS_CONN_RESETTING and then waiting for RDS_IN_XMIT to
be sampled clear before swapping the underlying socket and calling
rds_send_path_reset().

As in rds_conn_shutdown(), sampling the bit clear is not the same as
owning it: rds_send_xmit() can re-acquire RDS_IN_XMIT right after the
wait_event() returns.  Its state recheck after taking the lock is a
store-buffering pattern (teardown writes the state and reads the bit,
the sender writes the bit and reads the state) and acquire_in_xmit()
is only an acquire operation, so on weakly ordered architectures both
sides can miss each other's write and the transmit path then runs
concurrently with rds_send_path_reset() rewriting cp_xmit_* state.

Take the lock instead, hold it across the socket swap and
rds_send_path_reset(), and release it with a wake-up at the end.  The
lock-ordering constraint documented above the wait still holds: the
lock is acquired before lock_sock(), so a sender inside tcp_sendmsg()
can never be waited on while we hold the socket lock.

Owning the lock also serializes this function against
rds_conn_shutdown(), which since the previous patch holds RDS_IN_XMIT
across the transport shutdown.  That closes two holes the old
sample-based wait left open when a teardown and a duelling-SYN accept
raced on the same path:

 - t_sock is now read only after the lock is acquired.  The old code
   cached it before waiting; a concurrent teardown could release that
   socket inside rds_tcp_conn_path_shutdown() while the accept path
   was still blocked, which would leave it locking and releasing a
   freed socket once it resumed.

 - The old !osock early path called rds_send_path_reset() with no
   serialization at all, while the teardown could be running
   rds_send_path_reset() on the same path concurrently.  Losing that
   race means two threads walking cp_send_queue and putting the same
   messages.  That path now also runs under the lock (and sets
   RDS_CONN_RESETTING like the normal path; the raced connection ends
   up dropped and reconnecting either way).

The in-function comment describing the old wait-based quiesce is
rewritten to describe the lock-based one.

Fixes: 335b48d980f6 ("RDS: TCP: Add/use rds_tcp_reset_callbacks to reset tcp socket safely")
Assisted-by: Claude-Code:claude-fable-5
Signed-off-by: Allison Henderson <[email protected]>
---
Previously patch 4/4 of "net/rds: Bug fix ports, part 2".
v2: read t_sock only after acquiring RDS_IN_XMIT and hold the lock on
the !osock path too, closing the teardown-vs-accept races described
in the commit message; rewrite the in-function quiesce comment.
v1: https://lore.kernel.org/netdev/[email protected]/
 net/rds/tcp.c | 50 +++++++++++++++++++++++++++++++-------------------
 1 file changed, 31 insertions(+), 19 deletions(-)

diff --git a/net/rds/tcp.c b/net/rds/tcp.c
index b263634ac750..4ba5d4858ec7 100644
--- a/net/rds/tcp.c
+++ b/net/rds/tcp.c
@@ -127,33 +127,42 @@ void rds_tcp_reset_callbacks(struct socket *sock,
 			     struct rds_conn_path *cp)
 {
 	struct rds_tcp_connection *tc = cp->cp_transport_data;
-	struct socket *osock = tc->t_sock;
-
-	if (!osock)
-		goto newsock;
+	struct socket *osock;
 
 	/* Need to resolve a duelling SYN between peers.
 	 * We have an outstanding SYN to this peer, which may
 	 * potentially have transitioned to the RDS_CONN_UP state,
 	 * so we must quiesce any send threads before resetting
-	 * cp_transport_data. We quiesce these threads by setting
-	 * cp_state to something other than RDS_CONN_UP, and then
-	 * waiting for any existing threads in rds_send_xmit to
-	 * complete release_in_xmit(). (Subsequent threads entering
-	 * rds_send_xmit() will bail on !rds_conn_up().
+	 * cp_transport_data.  Setting cp_state to something other
+	 * than RDS_CONN_UP stops new senders, and owning RDS_IN_XMIT
+	 * excludes any thread already inside rds_send_xmit() as well
+	 * as the teardown in rds_conn_shutdown(), which holds the same
+	 * lock across the transport shutdown.
 	 *
-	 * However an incoming syn-ack at this point would end up
-	 * marking the conn as RDS_CONN_UP, and would again permit
-	 * rds_send_xmi() threads through, so ideally we would
-	 * synchronize on RDS_CONN_UP after lock_sock(), but cannot
-	 * do that: waiting on !RDS_IN_XMIT after lock_sock() may
-	 * end up deadlocking with tcp_sendmsg(), and the RDS_IN_XMIT
-	 * would not get set. As a result, we set c_state to
-	 * RDS_CONN_RESETTTING, to ensure that rds_tcp_state_change
-	 * cannot mark rds_conn_path_up() in the window before lock_sock()
+	 * An incoming syn-ack at this point would end up marking the
+	 * conn as RDS_CONN_UP, and would again permit rds_send_xmit()
+	 * threads through, so ideally we would synchronize on
+	 * RDS_CONN_UP after lock_sock(), but cannot do that: acquiring
+	 * RDS_IN_XMIT after lock_sock() may end up deadlocking with
+	 * tcp_sendmsg(), which takes the socket lock while holding
+	 * RDS_IN_XMIT.  As a result, we set c_state to
+	 * RDS_CONN_RESETTING, to ensure that rds_tcp_state_change
+	 * cannot mark rds_conn_path_up() in the window before
+	 * lock_sock().
 	 */
 	atomic_set(&cp->cp_state, RDS_CONN_RESETTING);
-	wait_event(cp->cp_waitq, !test_bit(RDS_IN_XMIT, &cp->cp_flags));
+	wait_event(cp->cp_waitq,
+		   !test_and_set_bit_lock(RDS_IN_XMIT, &cp->cp_flags));
+
+	/* Only read t_sock while owning RDS_IN_XMIT: a concurrent
+	 * rds_conn_shutdown() releases the old socket and clears
+	 * t_sock under the same lock, so a pointer sampled before the
+	 * wait could be freed by the time we wake up.
+	 */
+	osock = tc->t_sock;
+	if (!osock)
+		goto newsock;
+
 	/* reset receive side state for rds_tcp_data_recv() for osock  */
 	cancel_delayed_work_sync(&cp->cp_send_w);
 	cancel_delayed_work_sync(&cp->cp_recv_w);
@@ -172,6 +181,9 @@ void rds_tcp_reset_callbacks(struct socket *sock,
 	lock_sock(sock->sk);
 	rds_tcp_set_callbacks(sock, cp);
 	release_sock(sock->sk);
+
+	clear_bit_unlock(RDS_IN_XMIT, &cp->cp_flags);
+	wake_up_all(&cp->cp_waitq);
 }
 
 /* Add tc to rds_tcp_tc_list and set tc->t_sock. See comments
-- 
2.25.1