[PATCH net-next v8 10/13] net/rds: take cp_lock to purge cp_send_queue in the quiesce

rds_conn_path_quiesce() empties cp_send_queue by walking it with no
lock held, while every path that adds to that queue -
rds_send_queue_rm(), rds_send_probe(), the retransmit requeue in
rds_send_path_reset() - does so under cp_lock.  The unlocked walk was
justified by the destroy running with nothing else alive: at every
destroy trigger there is - netns teardown and module unload - no
socket can still be sending on the connection, since a bound socket
pins its transport module and a namespace closes its sockets before
its RDS connections are torn down.

That argument still holds, but it is an argument about the callers,
not a property of the code.  Splice the queue away under cp_lock and
drop the message references outside it, so that the one walker of the
list follows the same lock discipline as its adders.  This does not
by itself make a sender that is still running safe - nothing here
tells such a sender that the queue is closed - and it does not need
to, since no such sender exists at any destroy trigger.

While at it, make the purge coherent with rds_send_drop_to(), the
other path that removes messages from a connection queue.  drop_to
decides whether it owns the queue's reference by test_and_clear on
RDS_MSG_ON_CONN, and unlinks m_conn_item from whatever list the
message is on.  The purge used to BUG_ON() a message that a socket
still had queued, and that assertion is also what kept a drop_to
racing it from putting the queue's reference a second time or
unlinking from the purge list; the lock above is what closes the
stale-next hazard of the old walk, not the assertion.  Clear the bit
under cp_lock as part of the splice, so that drop_to leaves a purged
message alone, and turn the BUG_ON() into a WARN_ON_ONCE(): a socket
with messages queued at destroy is still a condition worth reporting -
no sender or closer can be running at any destroy trigger today - but
not one worth a panic, since the socket side keeps its own reference
and retires the message on close.

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

diff --git a/net/rds/connection.c b/net/rds/connection.c
index c7655e9339ca..9c8c4b28d2b2 100644
--- a/net/rds/connection.c
+++ b/net/rds/connection.c

@@ -540,6 +540,8 @@ void rds_conn_shutdown(struct rds_conn_path *cp)
 static void rds_conn_path_quiesce(struct rds_conn_path *cp)
 {
 	struct rds_message *rm, *rtmp;
+	unsigned long flags;
+	LIST_HEAD(purge);
 
 	if (!cp->cp_transport_data)
 		return;

@@ -551,12 +553,29 @@ static void rds_conn_path_quiesce(struct rds_conn_path *cp)
 	rds_conn_path_drop(cp, true);
 	flush_work(&cp->cp_down_w);
 
-	/* tear down queued messages */
-	list_for_each_entry_safe(rm, rtmp,
-				 &cp->cp_send_queue,
-				 m_conn_item) {
+	/* Tear down queued messages.  Every path that adds to
+	 * cp_send_queue does so under cp_lock; take it here too.  No
+	 * sender can still be running at any destroy trigger, so this is
+	 * lock discipline rather than a race fix: nothing here tells a
+	 * sender that the queue is closed.
+	 */
+	spin_lock_irqsave(&cp->cp_lock, flags);
+	list_splice_init(&cp->cp_send_queue, &purge);
+	/* Give up the queue's claim on each message while still under
+	 * the lock, so that rds_send_drop_to(), which decides ownership
+	 * of the connection-queue reference by this bit, neither drops
+	 * it a second time nor unlinks the message from our list.
+	 */
+	list_for_each_entry(rm, &purge, m_conn_item)
+		clear_bit(RDS_MSG_ON_CONN, &rm->m_flags);
+	spin_unlock_irqrestore(&cp->cp_lock, flags);
+	list_for_each_entry_safe(rm, rtmp, &purge, m_conn_item) {
+		/* No socket can still have messages queued on a connection
+		 * at any destroy trigger; say so if one does, since the
+		 * socket side then retires the message on its own.
+		 */
+		WARN_ON_ONCE(!list_empty(&rm->m_sock_item));
 		list_del_init(&rm->m_conn_item);
-		BUG_ON(!list_empty(&rm->m_sock_item));
 		rds_message_put(rm);
 	}
 	if (cp->cp_xmit_rm)

-- 
2.25.1