[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.  The !osock
early path is unchanged: it does not quiesce today and the connection
has never been RDS_CONN_UP at that point, so there is no sender to
serialize against.

This extends the previous change ("net/rds: acquire
the fastpath locks in rds_conn_shutdown()") to the only other
rds_send_path_reset() call site, mirroring Oracle UEK's "rds: Make sure
transmit path and connection tear-down does not run concurrently".

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]>
---
 net/rds/tcp.c | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/net/rds/tcp.c b/net/rds/tcp.c
index b263634ac750d..042d3fdbdf7fe 100644
--- a/net/rds/tcp.c
+++ b/net/rds/tcp.c
@@ -128,6 +128,7 @@ void rds_tcp_reset_callbacks(struct socket *sock,
 {
 	struct rds_tcp_connection *tc = cp->cp_transport_data;
 	struct socket *osock = tc->t_sock;
+	bool in_xmit_held = false;
 
 	if (!osock)
 		goto newsock;
@@ -153,7 +154,14 @@ void rds_tcp_reset_callbacks(struct socket *sock,
 	 * 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));
+	/* Acquire the send-path lock rather than waiting for it to be
+	 * released: a mere wait is racy, since rds_send_xmit() may take
+	 * the lock again right after we sample it clear and then run
+	 * concurrently with rds_send_path_reset() below.
+	 */
+	wait_event(cp->cp_waitq,
+		   !test_and_set_bit_lock(RDS_IN_XMIT, &cp->cp_flags));
+	in_xmit_held = true;
 	/* 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 +180,11 @@ void rds_tcp_reset_callbacks(struct socket *sock,
 	lock_sock(sock->sk);
 	rds_tcp_set_callbacks(sock, cp);
 	release_sock(sock->sk);
+
+	if (in_xmit_held) {
+		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