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

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

[Allison Henderson]

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 <achender@kernel.org>
---
 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

[PATCH net-next 0/4] net/rds: own the fastpath locks across connection teardown

[Allison Henderson]

Hi all,

This is a follow-up set to the recent "net/rds: Bug fix ports, part 2"
series [1]. During review of v1, the later half of the set needed more
work than just a respin, so it was split off into a separate set here
along with a few other companion fixes identified in the reivew.

RDS connection teardown quiesces the transmit and receive-refill fast
paths by waiting for the RDS_IN_XMIT/RDS_RECV_REFILL bits to be
sampled clear.  Sampling a bit clear is not owning it: the fast path
can re-take its bit right after the wait returns and then run
concurrently with the transport shutdown and the send-state reset.
Oracle UEK closed this by making teardown acquire the bits as locks
("rds: Make sure transmit path and connection tear-down does not run
concurrently"); patches 3 and 4 do the same for the two
rds_send_path_reset() call sites upstream.  These pacthes are
effectively v2 of patches 3 and 4 from the aforementioned
"net/rds: Bug fix ports, part 2" [1] set.

Making teardown block on the bits as locks promotes two latent
ordering bugs from rare to load-bearing, so they are fixed first:

  Patch 1: release_in_xmit() checks waitqueue_active() after
  clear_bit_unlock(), which does not order that read; the wake-up of
  the (now uninterruptible, untimed) teardown wait can be lost.  Use
  wq_has_sleeper().

  Patch 2: rds_conn_path_reset() wipes the whole cp_flags word with a
  plain store.  Once teardown owns bits in that word across the
  reset, a blanket store would end lock ownership early - and it
  already races atomic RMWs on the same word today.  Clear the bits
  the reset is responsible for individually, as Oracle UEK also does.

With those in place, patch 3 has rds_conn_shutdown() hold both bit
locks across the transport shutdown and path reset, and patch 4
converts rds_tcp_reset_callbacks() from waiting on RDS_IN_XMIT to
acquiring it, which also serializes the duelling-SYN accept path
against a concurrent teardown of the same path (the old code could
resume with a socket pointer the teardown had already released).

[PATCH net-next 1/4] net/rds: use wq_has_sleeper() in release_in_xmit()
  Restore full barrier before wake-up checks in release_in_xmit()

[PATCH net-next 2/4] net/rds: clear cp_flags bits individually in rds_conn_path_reset()
  Partial port of commit d04896037223 ("net/rds: Preserve essential connection state flags")
  https://github.com/oracle/linux-uek/commit/d04896037223

[PATCH net 3/4] net/rds: acquire the fastpath locks in rds_conn_shutdown()
  Port commit 2b8aaa4f163b ("rds: Make sure transmit path and connection tear-down does not run concurrently")
  https://github.com/oracle/linux-uek/commit/2b8aaa4f163b 

[PATCH net 4/4] net/rds: acquire RDS_IN_XMIT in rds_tcp_reset_callbacks()
  Extend the last port to cover extra call sites in rds_tcp_reset_callbacks()

Questions and comments appreciated!

Thanks,
Allison

[1] https://lore.kernel.org/netdev/20260806072045.1092968-1-achender@kernel.org/

Allison Henderson (3):
  net/rds: use wq_has_sleeper() in release_in_xmit()
  net/rds: clear cp_flags bits individually in rds_conn_path_reset()
  net/rds: acquire RDS_IN_XMIT in rds_tcp_reset_callbacks()

Håkon Bugge (1):
  net/rds: acquire the fastpath locks in rds_conn_shutdown()

 net/rds/connection.c | 32 +++++++++++++++++++++++++---
 net/rds/send.c       | 12 ++++++++---
 net/rds/tcp.c        | 50 +++++++++++++++++++++++++++-----------------
 3 files changed, 69 insertions(+), 25 deletions(-)


base-commit: 3da8c3c8b8fa99505624b65ef590482f48e766b6
-- 
2.25.1

[PATCH net-next 1/4] net/rds: use wq_has_sleeper() in release_in_xmit()

[Allison Henderson]

release_in_xmit() clears RDS_IN_XMIT with clear_bit_unlock() and then
checks waitqueue_active() to decide whether anyone needs waking.
clear_bit_unlock() is only a release operation: it orders the
critical section before the bit clear, but does not order the
subsequent plain load of the wait queue head after it.  The waiter
side does the mirror image - it adds itself to the wait queue and
then tests the bit.  That is the classic store-buffering pattern: the
releasing CPU can read the wait queue as empty while the waiting CPU
still reads the bit as set, so the sleeper is never woken.

The waiters are rds_conn_shutdown() and rds_tcp_reset_callbacks(),
both in uninterruptible wait_event() with no timeout.  A lost wake-up
strands the shutdown worker on its single-threaded workqueue until
some other sender releases the bit again - and on a connection that
is being torn down precisely because it failed, there may never be
another sender.

The barrier used to be there: release_in_xmit() did clear_bit()
followed by smp_mb__after_atomic() until commit 1422f28826d2 ("rds:
introduce acquire/release ordering in acquire/release_in_xmit()")
folded both into clear_bit_unlock(), which strengthened the lock
hand-off but silently dropped the full barrier the wake-up check
depends on.  The refill counterpart, release_refill() in
net/rds/ib_recv.c, still carries its smp_mb__after_atomic() for
exactly this reason.

Use wq_has_sleeper(), which is waitqueue_active() preceded by the
required full barrier.

Fixes: 1422f28826d2 ("rds: introduce acquire/release ordering in acquire/release_in_xmit()")
Assisted-by: Claude-Code:claude-fable-5
Signed-off-by: Allison Henderson <achender@kernel.org>
---
 net/rds/send.c | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/net/rds/send.c b/net/rds/send.c
index 15a1b97f13e7..8aad185e4b1a 100644
--- a/net/rds/send.c
+++ b/net/rds/send.c
@@ -114,8 +114,13 @@ static void release_in_xmit(struct rds_conn_path *cp)
 	 * hot path and finding waiters is very rare.  We don't want to walk
 	 * the system-wide hashed waitqueue buckets in the fast path only to
 	 * almost never find waiters.
+	 *
+	 * wq_has_sleeper() supplies the full barrier that orders the wait
+	 * queue read after the bit clear; clear_bit_unlock() alone is only
+	 * a release and would let this check read a stale empty queue,
+	 * losing the wake-up.
 	 */
-	if (waitqueue_active(&cp->cp_waitq))
+	if (wq_has_sleeper(&cp->cp_waitq))
 		wake_up_all(&cp->cp_waitq);
 }
 
-- 
2.25.1

[PATCH net-next 2/4] net/rds: clear cp_flags bits individually in rds_conn_path_reset()

[Allison Henderson]

rds_conn_path_reset() wipes the whole flag word with a plain
cp->cp_flags = 0 store.  Every other accessor of that word uses
atomic bitops, and some of them can run concurrently with the reset:
RDS_LL_SEND_FULL is set from rds_send_xmit() and cleared from the
transport completion paths, neither of which holds anything that
excludes the shutdown worker.  A plain store racing an atomic
read-modify-write on the same word is a data race, and whichever
side loses has its update silently discarded.

Clear the two bits the reset is actually responsible for instead.
RDS_IN_XMIT and RDS_RECV_REFILL need no store at all here: the
caller, rds_conn_shutdown(), waits for both to be clear before
calling the transport shutdown and this reset.

This also gives every bit in cp_flags a single well-defined writer
discipline, which the following patches rely on when they turn
RDS_IN_XMIT and RDS_RECV_REFILL into bit locks held across the
teardown: a blanket store mid-teardown would destroy lock ownership
that an atomic clear preserves.

Oracle UEK carries the same conversion ("net/rds: Preserve essential
connection state flags"), motivated by its asynchronous shutdown
state machine, whose progress and destroy flags must survive the
reset.  UEK's variant also clears RDS_IN_XMIT and RDS_RECV_REFILL
because there the reset runs as the final step of a teardown that
owns both bits, making those clears its unlock; upstream the unlock
stays in rds_conn_shutdown(), which needs release semantics and a
wake-up that a plain clear inside the reset would not provide.

Based on Oracle UEK commit "net/rds: Preserve essential connection
state flags" by Gerd Rausch.

Assisted-by: Claude-Code:claude-fable-5
Signed-off-by: Allison Henderson <achender@kernel.org>
---
 net/rds/connection.c | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/net/rds/connection.c b/net/rds/connection.c
index 7c8ab8e973e1..ddd7e2291eea 100644
--- a/net/rds/connection.c
+++ b/net/rds/connection.c
@@ -120,7 +120,15 @@ static void rds_conn_path_reset(struct rds_conn_path *cp)
 
 	rds_stats_inc(s_conn_reset);
 	rds_send_path_reset(cp);
-	cp->cp_flags = 0;
+
+	/* Clear the bits the reset is responsible for individually: a
+	 * blanket cp_flags = 0 is a plain store that can clobber a
+	 * concurrent atomic read-modify-write on the same word.
+	 * RDS_IN_XMIT and RDS_RECV_REFILL are already clear here - the
+	 * caller waited for both before tearing the transport down.
+	 */
+	clear_bit(RDS_LL_SEND_FULL, &cp->cp_flags);
+	clear_bit(RDS_RECONNECT_PENDING, &cp->cp_flags);
 
 	/* Do not clear next_rx_seq here, else we cannot distinguish
 	 * retransmitted packets from new packets, and will hand all
-- 
2.25.1

[PATCH net-next 3/4] net/rds: acquire the fastpath locks in rds_conn_shutdown()

[Allison Henderson]

From: Håkon Bugge <haakon.bugge@oracle.com>

rds_conn_shutdown() quiesces the transmit and receive-refill paths by
waiting for RDS_IN_XMIT and RDS_RECV_REFILL to be sampled clear, and
then runs the transport shutdown and rds_conn_path_reset().  Sampling
the bits clear is not the same as owning them: the moment after the
wait_event() returns, rds_send_xmit() can re-acquire RDS_IN_XMIT (or
rds_ib_recv_refill() can re-acquire RDS_RECV_REFILL) and run
concurrently with the teardown.

The sender does recheck the connection state after taking the lock,
but that recheck is a classic store-buffering pattern: teardown writes
the state and reads the bit while the sender writes the bit and reads
the state.  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 while the transport zeroes its rings
(e.g. rds_ib_ring_init()) and rds_send_path_reset() rewrites the
transmit state under it.

Oracle UEK fixed the same class of crashes - a 14-year tail of
BUG_ON()s in rds_ib_sub_signaled(), unexpected op-codes and NULL
dereferences in rds_ib_send_cqe_handler() during failover testing -
by making the teardown path *acquire* the fastpath bit locks instead
of testing them ("rds: Make sure transmit path and connection
tear-down does not run concurrently").  Ownership of a single word is
decided by RMW atomicity, so no cross-variable ordering is needed.

Do the same here: take both locks before calling the transport
shutdown, hold them across rds_conn_path_reset(), and release them
explicitly with a wake-up afterwards.  The fastpath users of these
bits - rds_send_xmit() and rds_ib_recv_refill() - are trylock style
and back off while teardown owns the locks, so no new lock dependency
is introduced for them.  rds_tcp_reset_callbacks() is different: it
blocks waiting for RDS_IN_XMIT, so its wait now spans the teardown
instead of at most one send batch; it also still only samples the
bit, which the next patch converts to an acquisition as well.

One observable side effect: the SENDING flag reported by rds-info has
always mirrored RDS_IN_XMIT, so it now also covers the window where
teardown owns the bit.

The comment in rds_send_xmit() describing the old sample-based
handshake is updated to match.

Fixes: 0f4b1c7e89e6 ("rds: fix rds_send_xmit() serialization")
Signed-off-by: Håkon Bugge <haakon.bugge@oracle.com>
[achender: reimplement for net-next shutdown path: acquire the existing
 RDS_IN_XMIT/RDS_RECV_REFILL bit locks in rds_conn_shutdown() and release
 after teardown; update comments and commit message]
Assisted-by: Claude-Code:claude-fable-5
Signed-off-by: Allison Henderson <achender@kernel.org>
---
Previously patch 3/4 of "net/rds: Bug fix ports, part 2".
v2: rds_conn_path_reset() no longer touches the lock bits, so lock
ownership now spans the whole teardown and ends at the explicit
release here; correct the changelog inventory of the bits' users
(rds_tcp_reset_callbacks() blocks, the IB send path never takes the
bit); update the stale protocol comment in rds_send_xmit().
v1: https://lore.kernel.org/netdev/20260806072045.1092968-4-achender@kernel.org/
 net/rds/connection.c | 26 ++++++++++++++++++++++----
 net/rds/send.c       |  5 +++--
 2 files changed, 25 insertions(+), 6 deletions(-)

diff --git a/net/rds/connection.c b/net/rds/connection.c
index ddd7e2291eea..a10b667c06c8 100644
--- a/net/rds/connection.c
+++ b/net/rds/connection.c
@@ -124,8 +124,8 @@ static void rds_conn_path_reset(struct rds_conn_path *cp)
 	/* Clear the bits the reset is responsible for individually: a
 	 * blanket cp_flags = 0 is a plain store that can clobber a
 	 * concurrent atomic read-modify-write on the same word.
-	 * RDS_IN_XMIT and RDS_RECV_REFILL are already clear here - the
-	 * caller waited for both before tearing the transport down.
+	 * RDS_IN_XMIT and RDS_RECV_REFILL are owned by the caller,
+	 * which releases them once the teardown is complete.
 	 */
 	clear_bit(RDS_LL_SEND_FULL, &cp->cp_flags);
 	clear_bit(RDS_RECONNECT_PENDING, &cp->cp_flags);
@@ -414,14 +414,32 @@ void rds_conn_shutdown(struct rds_conn_path *cp)
 		}
 		mutex_unlock(&cp->cp_cm_lock);
 
+		/* Quiesce the transmit and receive-refill paths by
+		 * acquiring their bit locks, not merely waiting for
+		 * them to be released: with a plain wait, either path
+		 * can re-take its lock the instant after we sample it
+		 * clear and then run concurrently with the transport
+		 * shutdown and the path reset below.  Holding both
+		 * locks across the teardown makes that structurally
+		 * impossible.
+		 */
 		wait_event(cp->cp_waitq,
-			   !test_bit(RDS_IN_XMIT, &cp->cp_flags));
+			   !test_and_set_bit_lock(RDS_IN_XMIT, &cp->cp_flags));
 		wait_event(cp->cp_waitq,
-			   !test_bit(RDS_RECV_REFILL, &cp->cp_flags));
+			   !test_and_set_bit(RDS_RECV_REFILL, &cp->cp_flags));
 
 		conn->c_trans->conn_path_shutdown(cp);
 		rds_conn_path_reset(cp);
 
+		/* Release the two locks and wake any waiter (e.g.
+		 * rds_tcp_reset_callbacks()) that blocked on them while
+		 * we held them.  rds_conn_path_reset() leaves both bits
+		 * alone: ownership ends here, not inside the reset.
+		 */
+		clear_bit_unlock(RDS_IN_XMIT, &cp->cp_flags);
+		clear_bit(RDS_RECV_REFILL, &cp->cp_flags);
+		wake_up_all(&cp->cp_waitq);
+
 		if (!rds_conn_path_transition(cp, RDS_CONN_DISCONNECTING,
 					      RDS_CONN_DOWN) &&
 		    !rds_conn_path_transition(cp, RDS_CONN_ERROR,
diff --git a/net/rds/send.c b/net/rds/send.c
index 8aad185e4b1a..b90e0586f818 100644
--- a/net/rds/send.c
+++ b/net/rds/send.c
@@ -244,8 +244,9 @@ int rds_send_xmit(struct rds_conn_path *cp)
 	WRITE_ONCE(cp->cp_send_gen, send_gen);
 
 	/*
-	 * rds_conn_shutdown() sets the conn state and then tests RDS_IN_XMIT,
-	 * we do the opposite to avoid races.
+	 * rds_conn_shutdown() sets the conn state and then acquires
+	 * RDS_IN_XMIT; we take the lock first and then check the state,
+	 * so one of us is guaranteed to see the other's update.
 	 */
 	if (!rds_conn_path_up(cp)) {
 		release_in_xmit(cp);
-- 
2.25.1

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

[Allison Henderson]

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 <achender@kernel.org>
---
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/20260806072045.1092968-5-achender@kernel.org/
 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