[PATCH net-next v7 09/12] net/rds: take cp_lock to purge cp_send_queue in the quiesce

[Allison Henderson <achender@kernel.org>]

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; the purge left that bit set, so a message it had
already put could be put a second time by a close() or
RDS_CANCEL_SENT_TO racing it.  Clear the bit under cp_lock as part of
the splice.  With that, a message that is still on its socket's send
queue is no longer a fatal condition for the purge - the socket side
keeps its own reference and retires it on close - so the
BUG_ON(!list_empty(&rm->m_sock_item)) goes as well.

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

diff --git a/net/rds/connection.c b/net/rds/connection.c
index 83e59fcaccea..1d7932cac035 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,24 @@ 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) {
 		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