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

[Alexey Kuznetsov]

Ack

On Tue, Oct 1, 2024 at 1:11 PM Liu Kui <kui.liu at virtuozzo.com> wrote:
>
> 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)