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

[Konstantin Khorenko]

From: Liu Kui <kui.liu at virtuozzo.com>

pcs_rpc_connect() is synchronous: it runs the whole connect+auth cycle
under ep->mutex.

rpc_queue_work() splices ep->input_queue onto an on-stack list and
feeds it to pcs_rpc_send() one message at a time. For an unconnected
ep this used to go as follows:

 1. pcs_rpc_send() stages the first message on ep->state_queue and
    calls pcs_rpc_connect(), which blocks for a connect timeout.
 2. The connect fails. pcs_rpc_reset() purges ep->input_queue and
    ep->state_queue, so that one message fails, and leaves ep
    unconnected.
 3. The rest of the batch is still on the on-stack list, invisible to
    the purge. The next message repeats step 1.

So with an unreachable peer the mutex stays held for a connect timeout
per message in the batch. Meanwhile calendar_work and rpc_close_work on
the same ep cannot run, and umount hangs in flush_workqueue(pcs_wq).

Stage the whole batch on ep->state_queue first and start the connect
once afterwards. A failed connect then fails the whole batch in a
single rpc_abort(), and the mutex is held for one connect attempt per
batch.

Fixes: 53e9def26e50 ("fuse kio: Add pcs engine combo v0.8")
https://virtuozzo.atlassian.net/browse/VSTOR-135626
Feature: vStorage

Signed-off-by: Liu Kui <kui.liu at virtuozzo.com>
Acked-by: Alexey Kuznetsov <kuznet at virtuozzo.com>
---
v1 from Liu Kui with edits to the commit message and the comment only.
The code change is identical to v1.

Signed-off-by: Konstantin Khorenko <khorenko at virtuozzo.com>
---
Changes in v2:
- Commit message: the failure chain is a numbered sequence now, and the
  claim that state_queue never held more than one message is dropped,
  since it accumulates messages during holddown.
- The comment in rpc_queue_work() keeps only the why, the history lives
  in the commit message.
- Added Fixes:, Feature: and Alexey's Acked-by.
- Link to v1: http://lore.virtuozzo.com/20260921-acssp872-batch-connect-v1-1-ded37d934f28@virtuozzo.com
---
 fs/fuse/kio/pcs/pcs_rpc.c | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/fs/fuse/kio/pcs/pcs_rpc.c b/fs/fuse/kio/pcs/pcs_rpc.c
index 540225006726d..6a3f63313d2e0 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,14 @@ static void rpc_queue_work(struct work_struct *w)
 		list_del_init(&msg->list);
 		pcs_rpc_send(ep, msg, 0);
 	}
+
+	/* pcs_rpc_connect() blocks under ep->mutex, so start it once per
+	 * batch, after every message is on ep->state_queue where a failed
+	 * connect can purge it via rpc_abort().
+	 */
+	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: c99164232e9dfa2f5b7615ab59c6e958a18c8caf
change-id: 20260921-batch-connect-287d60774a77

Best regards,
--  
Konstantin Khorenko <khorenko at virtuozzo.com>