[Devel] [PATCH VZ10] fs/fuse kio: fix RPC response drop for large messages sent over RDMA

[Liu Kui]

When a large message is sent via RDMA READ, it is completed with
rio_msg_sent() - which moves the request onto the RPC pending queue
where the response lookup can find it - only after both the
RDMA_READ_REQ TX completion and the peer's RDMA_READ_ACK have been
observed. However these two events are independent and not ordered:
the ACK, and even the response to the request, can arrive before the
local TX completion is polled. When that happens, the response finds
no matching request on the RPC pending queue and is dropped, stalling
the request until the RPC timeout logic kicks in.

Fix it by completing the message in rio_handle_rx_read_ack() as soon
as the RDMA_READ_ACK arrives. Messages in a single RX queue are
guaranteed to be delivered in order, so the ACK is always processed
before the response, and the request is therefore on the RPC pending
queue by the time its response arrives. This is also what the comment
in rio_submit() has promised all along.

The tx buffer is still recycled only after both events have been seen.
A TX send failure is handled by the connection abort logic as before.
The same applies to an error status carried in the RDMA_READ_ACK:
since the request msg has already been added to the rpc's pending
queue, the rio abort triggers an rpc abort, which completes the msg
with the same PCS_ERR_NET_ABORT error as before.

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

Signed-off-by: Liu Kui <kui.liu at virtuozzo.com>
---
 fs/fuse/kio/pcs/pcs_rdma_io.c | 9 +--------
 1 file changed, 1 insertion(+), 8 deletions(-)

diff --git a/fs/fuse/kio/pcs/pcs_rdma_io.c b/fs/fuse/kio/pcs/pcs_rdma_io.c
index 1c464e0e60f3..14beab8fda8c 100644
--- a/fs/fuse/kio/pcs/pcs_rdma_io.c
+++ b/fs/fuse/kio/pcs/pcs_rdma_io.c
@@ -797,7 +797,6 @@ static inline void rio_kick_write_queue(struct pcs_rdmaio *rio)
 
 static void rio_handle_tx(struct pcs_rdmaio *rio, struct rio_tx *tx, int ok)
 {
-	struct pcs_msg *msg = tx->msg;
 	u64 xid = tx->xid;
 
 	/* override remote success if we already aborted */
@@ -825,13 +824,6 @@ static void rio_handle_tx(struct pcs_rdmaio *rio, struct rio_tx *tx, int ok)
 		default:
 			BUG();
 	}
-
-	if (msg) {
-		if (!ok)
-			pcs_set_local_error(&msg->error, PCS_ERR_NET_ABORT);
-
-		rio_msg_sent(rio, NULL, msg, 1);
-	}
 }
 
 /*
@@ -931,6 +923,7 @@ static int rio_handle_rx_read_ack(struct pcs_rdmaio *rio,
 
 	list_for_each_entry(tx, &rio->active_txs, list)
 		if (tx->xid == rack->xid) {
+			rio_msg_sent(rio, NULL, tx->msg, 1);
 			list_del(&tx->list);
 			rio_handle_tx(rio, tx, !rack->status);
 			return 0;
-- 
2.50.1 (Apple Git-155)