[Devel] [PATCH RHEL9 COMMIT] fs/fuse kio: fix krpc state transition on connect

[Konstantin Khorenko]

The commit is pushed to "branch-rh9-5.14.0-427.35.1.vz9.76.x-ovz" and will appear at git at bitbucket.org:openvz/vzkernel.git
after rh9-5.14.0-427.35.1.vz9.76.4
------>
commit 06e44a360bf5bb5406a4be1172b26d6c195a2422
Author: Liu Kui <kui.liu at virtuozzo.com>
Date:   Tue Oct 1 13:07:39 2024 +0800

    fs/fuse kio: fix krpc state transition on connect
    
    Currently krpc would report to userspace to be in work state immediately
    after connect, even though kernel rpc may not be in work state. Userspace
    will start sending msg immediately after connect while kernel is still
    establishing the underlying connection. It's not uncommon that it may take
    very long time for a connection to be fully established. When this happens,
    userspace may abort the connection due to timeout. However krpc abort needs
    to acquire rpc's mutex lock, which could have been held for connection
    establishment, thus blocking the userspace evloop process.
    
    To avoid above problem, a proper krpc state transition on connect is
    implemented here. It now blocks userspace from transiting to work state
    on connect until the underlying kernel rpc transits to work state.
    Userspace can abort the krpc connection without being blocked while kernel
    rpc is still in establishment phrase.
    
    https://virtuozzo.atlassian.net/browse/VSTOR-93162
    
    Signed-off-by: Liu Kui <kui.liu at virtuozzo.com>
    Acked-by: Alexey Kuznetsov <kuznet at virtuozzo.com>
---
 fs/fuse/kio/pcs/pcs_krpc.c | 63 +++++++++++++++++++++++++++++++++++++++++-----
 fs/fuse/kio/pcs/pcs_krpc.h |  7 ++++++
 2 files changed, 64 insertions(+), 6 deletions(-)

diff --git a/fs/fuse/kio/pcs/pcs_krpc.c b/fs/fuse/kio/pcs/pcs_krpc.c
index 8f62f12782e5..3bc0d0376761 100644
--- a/fs/fuse/kio/pcs/pcs_krpc.c
+++ b/fs/fuse/kio/pcs/pcs_krpc.c
@@ -535,6 +535,8 @@ static int pcs_krpc_abort(struct pcs_krpc *krpc)
 	spin_lock(&krpc->lock);
 
 	if (krpc->state != PCS_KRPC_STATE_CONNECTED) {
+		if (krpc->state == PCS_KRPC_STATE_CONNECT)
+			krpc->state = PCS_KRPC_STATE_UNCONN;
 		spin_unlock(&krpc->lock);
 		return 0;
 	}
@@ -784,6 +786,38 @@ int pcs_krpc_update_addr(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id,
 
 	return 0;
 }
+
+static void krpc_connect_done(struct pcs_msg *msg)
+{
+	struct krpc_connect_req *req = container_of(msg, struct krpc_connect_req, msg);
+	struct pcs_krpc *krpc = req->krpc;
+	__poll_t pollflags = EPOLLHUP;
+
+	if (msg->rpc) {
+		pcs_rpc_put(msg->rpc);
+		msg->rpc = NULL;
+	}
+
+	spin_lock(&krpc->lock);
+	/* from a stale session, do nothing  */
+	if (req->gen != krpc->gen || krpc->state != PCS_KRPC_STATE_CONNECT) {
+		spin_unlock(&krpc->lock);
+		goto out;
+	}
+
+	if (!pcs_if_error(&msg->error)) {
+		krpc->state = PCS_KRPC_STATE_CONNECTED;
+		pollflags = EPOLLOUT;
+	}
+	spin_unlock(&krpc->lock);
+
+	wake_up_poll(&krpc->poll_wait, pollflags);
+
+out:
+	pcs_krpc_put(krpc);
+	kfree(req);
+}
+
 /*
  * Connect to a pcs_krpc, return a valid fd on success.
  */
@@ -793,6 +827,8 @@ int pcs_krpc_connect(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
 	int fd;
 	struct file *file;
 	struct pcs_krpc_context *ctx;
+	struct krpc_connect_req *connect_req;
+	struct pcs_msg *msg;
 
 	krpc = pcs_krpc_lookup(krpcs, id);
 	if (!krpc)
@@ -802,18 +838,26 @@ int pcs_krpc_connect(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
 		krpc->state == PCS_KRPC_STATE_DESTROYED)
 		return -EPERM;
 
+	connect_req = kzalloc(sizeof(*connect_req), GFP_KERNEL);
+	if (!connect_req)
+		return -ENOMEM;
+
 	ctx = kmalloc(sizeof(*ctx), GFP_KERNEL);
-	if (!ctx)
+	if (!ctx) {
+		kfree(connect_req);
 		return -ENOMEM;
+	}
 
 	fd = get_unused_fd_flags(O_CLOEXEC);
 	if (fd < 0) {
+		kfree(connect_req);
 		kfree(ctx);
 		return fd;
 	}
 
 	file = anon_inode_getfile("[pcs_krpc]", &pcs_krpc_fops, ctx, 0);
 	if (IS_ERR(file)) {
+		kfree(connect_req);
 		kfree(ctx);
 		put_unused_fd(fd);
 		fd = PTR_ERR(file);
@@ -825,13 +869,20 @@ int pcs_krpc_connect(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
 	spin_lock(&krpc->lock);
 	ctx->gen = ++krpc->gen;
 	ctx->krpc = pcs_krpc_get(krpc);
-	/*
-	 * the krpc should always be connected regardless state of
-	 * underlying RPC
-	 */
-	krpc->state = PCS_KRPC_STATE_CONNECTED;
+	connect_req->gen = krpc->gen;
+	connect_req->krpc = pcs_krpc_get(krpc);
+	krpc->state = PCS_KRPC_STATE_CONNECT;
 	spin_unlock(&krpc->lock);
 
+	/* Send a zero-size msg which should be completed after the rpc enters work state */
+	msg = &connect_req->msg;
+	msg->size = 0;
+	msg->timeout = 0;
+	msg->rpc = NULL;
+	msg->done = krpc_connect_done;
+	pcs_clear_error(&msg->error);
+	pcs_rpc_queue(krpc->rpc, msg);
+
 	return fd;
 }
 
diff --git a/fs/fuse/kio/pcs/pcs_krpc.h b/fs/fuse/kio/pcs/pcs_krpc.h
index 33d758e2503e..23b170c00d7a 100644
--- a/fs/fuse/kio/pcs/pcs_krpc.h
+++ b/fs/fuse/kio/pcs/pcs_krpc.h
@@ -39,11 +39,18 @@ struct pcs_krpc_set {
 
 enum {
 	PCS_KRPC_STATE_UNCONN,
+	PCS_KRPC_STATE_CONNECT,
 	PCS_KRPC_STATE_CONNECTED,
 	PCS_KRPC_STATE_ABORTED,
 	PCS_KRPC_STATE_DESTROYED,
 };
 
+struct krpc_connect_req {
+	struct pcs_msg	msg;
+	struct pcs_krpc *krpc;
+	u32	gen;
+};
+
 struct pcs_krpc {
 	struct hlist_node		hlist;
 	struct list_head		link;