[Devel] [PATCH hci-8.1] fs/fuse kio: start connect once per rpc_queue_work batch

[Liu Kui]

pcs_rpc_connect() is synchronous: it performs a full connect+auth
cycle while holding ep->mutex. rpc_queue_work() splices the entire
ep->input_queue onto an on-stack list and used to trigger the connect
from pcs_rpc_send() on the first staged message. When the connect
fails, pcs_rpc_reset() purges ep->input_queue and ep->state_queue,
but the messages still sitting on the on-stack list are invisible to
the purge - each of them re-triggers another blocking connect attempt
under the same mutex hold. With a large backlog and an unreachable
peer this keeps ep->mutex held for O(messages * connect timeout):
in one crashdump the rescuer of pcs_wq ground through such a batch
for days, starving calendar_work/rpc_close_work on the same ep and
wedging umount in flush_workqueue(pcs_wq).

Stage all messages from the batch on ep->state_queue first and start
the connect once afterwards. A failed connect now aborts the whole
batch in a single rpc_abort(), bounding the mutex hold to one connect
attempt. This also restores the meaning of state_queue, which
otherwise never held more than one message.

https://virtuozzo.atlassian.net/browse/VSTOR-135626

Signed-off-by: Liu Kui <kui.liu at virtuozzo.com>
---
Fix a pathological case seen in ACSSP-872 (and ASUP-3562) crashdumps:
a single rpc_queue_work() invocation could hold ep->mutex for days,
re-running a blocking connect+auth once per queued message to an
unreachable peer, wedging the pcs_wq rescuer, calendar/close works
and umount in flush_workqueue().

https://virtuozzo.atlassian.net/browse/ACSSP-872
---
 fs/fuse/kio/pcs/pcs_rpc.c | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/fs/fuse/kio/pcs/pcs_rpc.c b/fs/fuse/kio/pcs/pcs_rpc.c
index 540225006726..c50e31c08a31 100644
--- a/fs/fuse/kio/pcs/pcs_rpc.c
+++ b/fs/fuse/kio/pcs/pcs_rpc.c
@@ -702,9 +702,6 @@ static void pcs_rpc_send(struct pcs_rpc * ep, struct pcs_msg * msg, bool requeue
 
 	list_add_tail(&msg->list, &ep->state_queue);
 	msg->stage = PCS_MSG_STAGE_UNSENT;
-
-	if (ep->state == PCS_RPC_UNCONN)
-		pcs_rpc_connect(ep);
 }
 
 void pcs_rpc_kick_queue(struct pcs_rpc * ep)
@@ -930,6 +927,21 @@ static void rpc_queue_work(struct work_struct *w)
 		list_del_init(&msg->list);
 		pcs_rpc_send(ep, msg, 0);
 	}
+
+	/* pcs_rpc_connect() is synchronous: it can block for a full
+	 * connect+auth cycle while holding ep->mutex. Start it once, after
+	 * every message from input_q has been staged on ep->state_queue,
+	 * instead of from pcs_rpc_send() on the first message. This way a
+	 * failed connect aborts the whole batch at once: rpc_abort() purges
+	 * ep->state_queue, whereas messages still sitting on the on-stack
+	 * input_q are invisible to the purge and would each trigger another
+	 * blocking connect attempt under the same mutex hold. It also
+	 * restores the meaning of state_queue, which otherwise never holds
+	 * more than one message.
+	 */
+	if (ep->state == PCS_RPC_UNCONN && !list_empty(&ep->state_queue))
+		pcs_rpc_connect(ep);
+
 	list_splice_tail_init(&ep->state_queue, &state_q);
 	while (!list_empty(&state_q)) {
 		struct pcs_msg * msg = list_first_entry(&state_q, struct pcs_msg, list);

---
base-commit: 3851feefb321d7dee9e57cbe7926410aa6dd043f
change-id: 20260921-acssp872-batch-connect-bec05e9db04d

Best regards,
--  
Liu Kui <kui.liu at virtuozzo.com>