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

[Liu Kui]

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>
---
 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 0ef33b730204..ee88bc17e6d9 100644
--- a/fs/fuse/kio/pcs/pcs_krpc.c
+++ b/fs/fuse/kio/pcs/pcs_krpc.c
@@ -474,6 +474,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;
 	}
@@ -723,6 +725,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.
  */
@@ -732,6 +766,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)
@@ -741,18 +777,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);
@@ -764,13 +808,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 8100dfb2629d..db13b71f6357 100644
--- a/fs/fuse/kio/pcs/pcs_krpc.h
+++ b/fs/fuse/kio/pcs/pcs_krpc.h
@@ -37,11 +37,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;
-- 
2.39.3 (Apple Git-146)