[Devel] [PATCH VZ10] fs/fuse kio: fix kRPC connect issues

[Liu Kui]

There are two problems with kRPC connect:

1) A kernel RPC connect failure is silently dropped instead of being
   reported as EPOLLERR. Userspace cannot detect the failure promptly
   and must fall back on its timeout mechanism to retry. Fix this by
   reporting EPOLLERR on connect failure.

2) kRPC connect requests can accumulate when the peer node is dead and
   the kernel RPC connect blocks for longer than the userspace connect
   timeout. Each new request restarts the kernel RPC connect with no
   hold-down time, effectively blocking the RPC work forever and
   preventing teardown of the fuse connection. Fix this by allowing
   only one pending connect request at a time: if userspace starts a
   new kRPC connect after a timeout, it fails immediately, forcing
   userspace to retry later.

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

Signed-off-by: Liu Kui <kui.liu at virtuozzo.com>
---
 fs/fuse/kio/pcs/pcs_krpc.c | 74 +++++++++++++++++++++++++-------------
 fs/fuse/kio/pcs/pcs_krpc.h |  2 ++
 2 files changed, 51 insertions(+), 25 deletions(-)

diff --git a/fs/fuse/kio/pcs/pcs_krpc.c b/fs/fuse/kio/pcs/pcs_krpc.c
index 3b42fd7d4489..0930fb4adf12 100644
--- a/fs/fuse/kio/pcs/pcs_krpc.c
+++ b/fs/fuse/kio/pcs/pcs_krpc.c
@@ -956,9 +956,10 @@ static __poll_t pcs_krpc_poll(struct file *file, poll_table *wait)
 
 	spin_lock(&krpc->lock);
 
-	if (krpc->state == PCS_KRPC_STATE_ABORTED)
+	if (krpc->state == PCS_KRPC_STATE_ABORTED ||
+	    (krpc->state == PCS_KRPC_STATE_CONNECT && !krpc->connect_req)) {
 		pollflags |= EPOLLERR;
-	else if (krpc->state == PCS_KRPC_STATE_CONNECTED) {
+	} else if (krpc->state == PCS_KRPC_STATE_CONNECTED) {
 		pollflags |= EPOLLOUT;
 		if (!list_empty(&krpc->completion_queue))
 			pollflags |= EPOLLIN;
@@ -1046,6 +1047,7 @@ int pcs_krpc_create(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id,
 	krpc->gen = 0;
 	krpc->state = PCS_KRPC_STATE_UNCONN;
 	krpc->cs = NULL;
+	krpc->connect_req = NULL;
 
 	krpc->rpc = pcs_rpc_clnt_create(&cc_from_krpcset(krpcs)->eng, id, addr, cs_flags);
 	if (!krpc->rpc) {
@@ -1089,7 +1091,7 @@ 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;
+	__poll_t pollflags = EPOLLHUP | EPOLLERR;
 
 	if (msg->rpc) {
 		pcs_rpc_put(msg->rpc);
@@ -1097,6 +1099,8 @@ static void krpc_connect_done(struct pcs_msg *msg)
 	}
 
 	spin_lock(&krpc->lock);
+	if (krpc->connect_req == req)
+		krpc->connect_req = NULL;
 	/* from a stale session, do nothing  */
 	if (req->gen != krpc->gen || krpc->state != PCS_KRPC_STATE_CONNECT) {
 		spin_unlock(&krpc->lock);
@@ -1122,7 +1126,7 @@ static void krpc_connect_done(struct pcs_msg *msg)
 int pcs_krpc_connect(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
 {
 	struct pcs_krpc *krpc;
-	int fd;
+	int fd, err;
 	struct file *file;
 	struct pcs_krpc_context *ctx;
 	struct krpc_connect_req *connect_req;
@@ -1132,46 +1136,57 @@ int pcs_krpc_connect(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
 	if (!krpc)
 		return -ENXIO;
 
-	if (krpc->state == PCS_KRPC_STATE_CONNECTED ||
-		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) {
-		kfree(connect_req);
-		return -ENOMEM;
-	}
-
 	fd = get_unused_fd_flags(O_CLOEXEC);
 	if (fd < 0) {
-		kfree(connect_req);
-		kfree(ctx);
-		return fd;
+		err = fd;
+		goto free_req;
 	}
 
+	ctx = kmalloc(sizeof(*ctx), GFP_KERNEL);
+	if (!ctx) {
+		err = -ENOMEM;
+		goto put_fd;
+	}
+	/*
+	 * Set up ctx before creating the file so that, if we have to bail out
+	 * below, fput() -> pcs_krpc_release() finds a consistent ctx.  gen 0 is
+	 * never a committed session (commit always pre-increments gen), so
+	 * release() will not mistake it for the live session and abort it.
+	 */
+	ctx->krpc = pcs_krpc_get(krpc);
+	ctx->gen = 0;
+
 	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);
-		return fd;
+		err = PTR_ERR(file);
+		goto put_ctx;
 	}
 
-	fd_install(fd, file);
-
 	spin_lock(&krpc->lock);
+	if (krpc->state == PCS_KRPC_STATE_CONNECTED ||
+	    krpc->state == PCS_KRPC_STATE_DESTROYED ||
+	    krpc->connect_req) {
+		spin_unlock(&krpc->lock);
+		err = -EPERM;
+		/* fput() drops ctx and its krpc reference via pcs_krpc_release() */
+		fput(file);
+		goto put_fd;
+	}
+
 	ctx->gen = ++krpc->gen;
-	ctx->krpc = pcs_krpc_get(krpc);
 	connect_req->gen = krpc->gen;
 	connect_req->krpc = pcs_krpc_get(krpc);
 	krpc->state = PCS_KRPC_STATE_CONNECT;
+	krpc->connect_req = connect_req;
 	spin_unlock(&krpc->lock);
 
+	/* publish the fd only after the connect is committed */
+	fd_install(fd, file);
+
 	/* Send a zero-size msg which should be completed after the rpc enters work state */
 	msg = &connect_req->msg;
 	msg->size = 0;
@@ -1183,6 +1198,15 @@ int pcs_krpc_connect(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
 	pcs_rpc_queue(krpc->rpc, msg);
 
 	return fd;
+
+put_ctx:
+	pcs_krpc_put(krpc);
+	kfree(ctx);
+put_fd:
+	put_unused_fd(fd);
+free_req:
+	kfree(connect_req);
+	return err;
 }
 
 static void __pcs_krpc_destroy(struct pcs_krpc *krpc)
diff --git a/fs/fuse/kio/pcs/pcs_krpc.h b/fs/fuse/kio/pcs/pcs_krpc.h
index 96b4815abf86..803376895cc5 100644
--- a/fs/fuse/kio/pcs/pcs_krpc.h
+++ b/fs/fuse/kio/pcs/pcs_krpc.h
@@ -84,6 +84,8 @@ struct pcs_krpc {
 	/** Wait queue head for poll */
 	wait_queue_head_t		poll_wait;
 	struct pcs_cs			*cs;
+
+	struct krpc_connect_req *connect_req;
 };
 
 struct pcs_krpc_context {
-- 
2.50.1 (Apple Git-155)