[Devel] [PATCH VZ9 2/4] fs/fuse kio: implement pcs_krpc - export kernel RPC to userspace
Liu Kui
kui.liu at virtuozzo.com
Thu Jun 27 13:32:25 MSK 2024
Currently there are 2 connections for every RPC, one in userspace,
one in kernel. This wastes a lot of resources on client hosts from
a huge cluster. It's therefore desirable to eliminate the connection
in userspace by using kernel RPC directly from userspace.
The newly implemented pcs_krpc serves as the glue layer between
userspace RPC and kernel RPC. It provides execution context for
forwarding messages from/to userspace RPC to/from kernel RPC.
It also providee APIs for userspace to send/receive RPC messages.
https://pmc.acronis.work/browse/VSTOR-82613
Signed-off-by: Liu Kui <kui.liu at virtuozzo.com>
---
fs/fuse/kio/pcs/pcs_krpc.c | 817 ++++++++++++++++++++++++++++++++
fs/fuse/kio/pcs/pcs_krpc.h | 144 ++++++
fs/fuse/kio/pcs/pcs_krpc_prot.h | 44 ++
3 files changed, 1005 insertions(+)
create mode 100644 fs/fuse/kio/pcs/pcs_krpc.c
create mode 100644 fs/fuse/kio/pcs/pcs_krpc.h
create mode 100644 fs/fuse/kio/pcs/pcs_krpc_prot.h
diff --git a/fs/fuse/kio/pcs/pcs_krpc.c b/fs/fuse/kio/pcs/pcs_krpc.c
new file mode 100644
index 000000000000..567d8a0c884f
--- /dev/null
+++ b/fs/fuse/kio/pcs/pcs_krpc.c
@@ -0,0 +1,817 @@
+/*
+ * Copyright (c) 2018-2021 Virtuozzo International GmbH. All rights reserved.
+ */
+
+#include <linux/types.h>
+#include <linux/list.h>
+#include <linux/rbtree.h>
+#include <linux/refcount.h>
+#include <linux/file.h>
+#include <linux/anon_inodes.h>
+
+#include "pcs_types.h"
+#include "pcs_cluster.h"
+#include "pcs_cs_prot.h"
+#include "pcs_req.h"
+#include "pcs_krpc.h"
+
+struct kmem_cache *krpc_req_cachep;
+
+static int krpc_completion_post(struct pcs_krpc *krpc, struct krpc_completion *comp)
+__releases(krpc->lock)
+{
+ /* post complete only when connected */
+ if (krpc->state != PCS_KRPC_STATE_CONNECTED) {
+ spin_unlock(&krpc->lock);
+ return -1;
+ }
+ list_add_tail(&comp->link, &krpc->completion_queue);
+ krpc->nr_completion++;
+ spin_unlock(&krpc->lock);
+
+ wake_up_poll(&krpc->poll_wait, EPOLLIN);
+ return 0;
+}
+
+static struct krpc_req *krpc_req_alloc(void)
+{
+ struct krpc_req *kreq;
+
+ kreq = kmem_cache_alloc(krpc_req_cachep, GFP_NOIO);
+ if (!kreq)
+ return NULL;
+
+ spin_lock_init(&kreq->lock);
+ kreq->flags = 0;
+ memset(&kreq->completion, 0, sizeof(struct krpc_completion));
+
+ return kreq;
+}
+
+static void krpc_req_free(struct krpc_req *kreq)
+{
+ kmem_cache_free(krpc_req_cachep, kreq);
+}
+
+static void krpc_completion_free(struct krpc_completion *comp)
+{
+ if (comp->private)
+ krpc_req_free((struct krpc_req *)comp->private);
+ else
+ kfree(comp);
+}
+
+static void krpc_req_complete(struct krpc_req *kreq, int error)
+{
+ struct krpc_completion *comp = &kreq->completion;
+ struct pcs_krpc *krpc = kreq->krpc;
+ int i;
+
+ BUG_ON(!comp->xid);
+
+ comp->result = error;
+
+ pcs_mr_put(kreq->hdr_chunk.mr);
+
+ for (i = 0; i < kreq->nr_data_chunks; i++) {
+ struct krpc_chunk *chunk;
+
+ chunk = &kreq->data_chunks[i];
+ if (chunk->type == KRPC_CHUNK_TYPE_UMEM)
+ pcs_umem_release(chunk->umem);
+ else
+ pcs_mr_put(chunk->mr);
+ }
+
+ if (kreq->data_chunks != &kreq->inline_data_chunks[0])
+ kfree(kreq->data_chunks);
+
+ spin_lock(&krpc->lock);
+ list_del(&kreq->link);
+
+ if (kreq->flags & KRPC_REQ_F_ABORTED) {
+ krpc_req_free(kreq);
+ spin_unlock(&krpc->lock);
+ } else if (krpc_completion_post(krpc, comp)) {
+ krpc_req_free(kreq);
+ }
+
+ pcs_krpc_put(krpc);
+}
+
+static void krpc_msg_get_response_iter(struct pcs_msg *msg, int offset,
+ struct iov_iter *it, unsigned int direction)
+{
+ struct pcs_msg *req = msg->private;
+ struct krpc_req *kreq = req->private2;
+
+ if (!(kreq->flags & KRPC_REQ_F_RESP_BUFF)) {
+ /* No data payload */
+ BUG_ON(msg->size > PAGE_SIZE);
+
+ kreq->hdr_kv.iov_base = (void *)kreq->hdr_buf;
+ kreq->hdr_kv.iov_len = msg->size;
+
+ iov_iter_kvec(it, direction, &kreq->hdr_kv, 1, msg->size);
+ iov_iter_advance(it, offset);
+ } else {
+ /* With data payload */
+ int hdr_size = sizeof(struct pcs_cs_iohdr);
+
+ if (offset < hdr_size) {
+ kreq->hdr_kv.iov_base = (void *)kreq->hdr_buf;
+ kreq->hdr_kv.iov_len = hdr_size;
+ iov_iter_kvec(it, direction, &kreq->hdr_kv, 1, hdr_size);
+ iov_iter_advance(it, offset);
+ } else {
+ BUG_ON((offset - hdr_size) > kreq->data_len);
+ BUG_ON(msg->size - hdr_size > kreq->data_len);
+
+ iov_iter_bvec(it, direction, &kreq->data_bvecs[0],
+ kreq->nr_data_bvecs, kreq->data_len);
+ iov_iter_truncate(it, msg->size - hdr_size);
+ iov_iter_advance(it, offset - hdr_size);
+ }
+ }
+}
+
+struct pcs_msg *krpc_get_hdr(struct pcs_rpc *ep, struct pcs_rpc_hdr *h)
+{
+ struct pcs_msg *msg, *resp;
+ struct pcs_rpc_hdr *req_h;
+ struct krpc_req *kreq;
+
+ if (!RPC_IS_RESPONSE(h->type))
+ return NULL;
+
+ msg = pcs_rpc_lookup_xid(ep, &h->xid);
+ if (msg == NULL)
+ return NULL;
+
+ req_h = (struct pcs_rpc_hdr *)msg_inline_head(msg);
+ if (req_h->type != (h->type & ~PCS_RPC_DIRECTION))
+ return NULL;
+
+ kreq = msg->private2;
+
+ resp = pcs_rpc_alloc_input_msg(ep, sizeof(struct pcs_rpc_hdr));
+ if (!resp)
+ return NULL;
+
+ memcpy(resp->_inline_buffer, h, sizeof(struct pcs_rpc_hdr));
+ memcpy(kreq->hdr_buf, h, sizeof(struct pcs_rpc_hdr));
+ resp->size = h->len;
+ resp->private = msg;
+ resp->get_iter = krpc_msg_get_response_iter;
+ resp->done = rpc_work_input;
+ pcs_msg_del_calendar(msg);
+
+ return resp;
+}
+
+/*
+ * All request type message from cs are forwarded to userspace directly. The PCS KRPC
+ * layer cannot process these request on its own.
+ */
+static void krpc_handle_request(struct pcs_krpc *krpc, struct pcs_msg *msg)
+{
+ struct krpc_completion *comp;
+
+ if (krpc->state != PCS_KRPC_STATE_CONNECTED)
+ return;
+
+ comp = kzalloc(sizeof(*comp) + msg->size, GFP_NOIO);
+ if (!comp)
+ return;
+
+ memcpy(comp->_data_buf, msg->_inline_buffer, msg->size);
+ comp->data_len = msg->size;
+
+ spin_lock(&krpc->lock);
+ if (krpc_completion_post(krpc, comp))
+ kfree(comp);
+}
+
+/*
+ * Let userspace to handle cs_congestion_notify request.
+ */
+void krpc_handle_congestion(struct pcs_rpc *ep, struct pcs_msg *msg)
+{
+ krpc_handle_request(ep->clnt_krpc, msg);
+}
+
+/*
+ * Let userspace to handle the keep_waiting request.
+ */
+void krpc_keep_waiting(struct pcs_rpc *ep, struct pcs_msg *req, struct pcs_msg *msg)
+{
+ krpc_handle_request(ep->clnt_krpc, msg);
+}
+
+unsigned int pcs_krpc_hash(PCS_NODE_ID_T *id)
+{
+ return *(unsigned int *)id % PCS_KRPC_HASH_SIZE;
+}
+
+static void krpc_msg_get_data(struct pcs_msg *msg, int offset,
+ struct iov_iter *it, unsigned int direction)
+{
+ struct krpc_req *kreq = msg->private2;
+ int hdr_size = pcs_krpc_msg_size(kreq->hdr_chunk.len, kreq->flags);
+ int data_size = pcs_krpc_msg_size(kreq->data_len, kreq->flags);
+ struct pcs_krpc *krpc = kreq->krpc;
+
+ if (offset < kreq->hdr_chunk.len) {
+ iov_iter_kvec(it, direction, &kreq->hdr_kv, 1, kreq->hdr_chunk.len);
+ iov_iter_advance(it, offset);
+ return;
+ }
+
+ if (offset < hdr_size) {
+ iov_iter_kvec(it, direction, &(cc_from_krpcset(krpc->krpcs))->nilbuffer_kv,
+ 1, hdr_size - offset);
+ return;
+ }
+
+ if (offset < hdr_size + kreq->data_len) {
+ iov_iter_bvec(it, direction, &kreq->data_bvecs[0],
+ kreq->nr_data_bvecs, kreq->data_len);
+ iov_iter_advance(it, offset - hdr_size);
+ return;
+ }
+
+ BUG_ON(offset >= hdr_size + data_size);
+
+ iov_iter_kvec(it, direction, &(cc_from_krpcset(krpc->krpcs))->nilbuffer_kv,
+ 1, (hdr_size + data_size - offset));
+}
+
+static void pcs_krpc_response_done(struct pcs_msg *msg)
+{
+ struct krpc_req *kreq = msg->private2;
+
+ if (msg->rpc) {
+ pcs_rpc_put(msg->rpc);
+ msg->rpc = NULL;
+ }
+
+ krpc_req_complete(kreq, msg->error.value);
+}
+
+static void pcs_krpc_msg_sent(struct pcs_msg *msg)
+{
+ msg->done = pcs_krpc_response_done;
+ if (pcs_if_error(&msg->error)) {
+ msg->done(msg);
+ return;
+ }
+ pcs_rpc_sent(msg);
+}
+
+static int pcs_krpc_ioctl_recv_msg(struct pcs_krpc *krpc, struct pcs_krpc_ioc_recvmsg *iocmsg)
+{
+ struct krpc_completion *comp;
+ int res = 0;
+
+ spin_lock(&krpc->lock);
+ if (list_empty(&krpc->completion_queue)) {
+ spin_unlock(&krpc->lock);
+ return 0;
+ }
+
+ comp = list_first_entry(&krpc->completion_queue, struct krpc_completion, link);
+ list_del(&comp->link);
+ krpc->nr_completion--;
+ spin_unlock(&krpc->lock);
+
+ if (comp->result) {
+ res -= comp->result;
+ goto out;
+ }
+
+ res = 1;
+ iocmsg->xid = comp->xid;
+ if (comp->xid == 0) {
+ BUG_ON(!comp->data_len);
+ BUG_ON(iocmsg->buf.len < comp->data_len);
+ if (copy_to_user((void __user *)iocmsg->buf.addr, comp->_data_buf, comp->data_len))
+ res = -EFAULT;
+ }
+
+out:
+ krpc_completion_free(comp);
+ return res;
+}
+
+static int pcs_krpc_ioctl_send_msg(struct pcs_krpc *krpc, struct pcs_krpc_ioc_sendmsg *iocmsg)
+{
+ struct krpc_req *kreq;
+ struct pcs_msg *msg;
+ struct pcs_krpc_buf_desc *chunk_bd;
+ struct krpc_chunk *chunk;
+ int res, i;
+ struct bio_vec *bvec;
+
+ kreq = krpc_req_alloc();
+ if (!kreq)
+ return -ENOMEM;
+
+ if (iocmsg->nr_data_chunks > NR_KRPC_DATA_CHUNKS_INLINE) {
+ kreq->data_chunks = kzalloc(iocmsg->nr_data_chunks, GFP_NOIO);
+ if (!kreq->data_chunks) {
+ res = -ENOMEM;
+ goto err_free_kreq;
+ }
+ } else {
+ kreq->data_chunks = &kreq->inline_data_chunks[0];
+ }
+
+ /*
+ * Header buff is exactly one page, used for staging message header. Will be used for
+ * receiving header of response message as well.
+ */
+ BUG_ON(iocmsg->hdr_chunk.addr & (PAGE_SIZE - 1));
+
+ chunk_bd = &iocmsg->hdr_chunk;
+ chunk = &kreq->hdr_chunk;
+
+ chunk->addr = chunk_bd->addr;
+ chunk->len = chunk_bd->len;
+ chunk->type = KRPC_CHUNK_TYPE_MR;
+
+ chunk->mr = pcs_mr_get(&cc_from_krpc(krpc)->mrs, iocmsg->hdr_chunk.mr_id);
+ BUG_ON(!chunk->mr);
+
+ kreq->hdr_buf = (char *) kmap(pcs_umem_page(chunk->mr->umem, chunk->addr));
+ kreq->hdr_kv.iov_base = kreq->hdr_buf;
+ kreq->hdr_kv.iov_len = chunk->len;
+
+ /* data chunk buf descriptors are placed at end of header buf, grow backwards */
+ kreq->data_len = 0;
+ kreq->nr_data_chunks = 0;
+ kreq->nr_data_bvecs = 0;
+
+ chunk_bd = (struct pcs_krpc_buf_desc *)(kreq->hdr_buf + PAGE_SIZE) - 1;
+ chunk = kreq->data_chunks;
+ bvec = &kreq->data_bvecs[0];
+
+ for (i = 0; i < iocmsg->nr_data_chunks; i++) {
+ struct page *page;
+ u64 addr;
+ int offset, len;
+
+ chunk->addr = chunk_bd->addr;
+ chunk->len = chunk_bd->len;
+ if (chunk_bd->mr_id) {
+ chunk->mr = pcs_mr_get(&cc_from_krpc(krpc)->mrs, chunk_bd->mr_id);
+ chunk->type = KRPC_CHUNK_TYPE_MR;
+ if (!chunk->mr) {
+ res = -ENXIO;
+ goto err_free_data_chunk;
+ }
+ } else {
+ /* unregistered memory buf */
+ chunk->umem = pcs_umem_get(chunk->addr, chunk->len);
+ if (IS_ERR(chunk->umem)) {
+ res = PTR_ERR(chunk->umem);
+ goto err_free_data_chunk;
+ }
+
+ chunk->type = KRPC_CHUNK_TYPE_UMEM;
+ }
+
+ /* build the bvecs for the data chunk*/
+ addr = chunk->addr;
+ len = chunk->len;
+
+ while (len) {
+ /* data bvec array overflow? */
+ BUG_ON(kreq->nr_data_bvecs >= KRPC_MAX_DATA_PAGES);
+
+ if (chunk->type == KRPC_CHUNK_TYPE_MR)
+ page = pcs_umem_page(chunk->mr->umem, addr);
+ else
+ page = pcs_umem_page(chunk->umem, addr);
+
+ BUG_ON(!page);
+
+ offset = addr & (PAGE_SIZE - 1);
+
+ bvec->bv_page = page;
+ bvec->bv_offset = offset;
+
+ bvec->bv_len = len < (PAGE_SIZE - offset) ? len : (PAGE_SIZE - offset);
+
+ addr += bvec->bv_len;
+ len -= bvec->bv_len;
+ bvec++;
+ kreq->nr_data_bvecs++;
+ }
+ BUG_ON(len != 0);
+
+ kreq->data_len += chunk->len;
+ chunk++;
+ chunk_bd--;
+ kreq->nr_data_chunks++;
+ }
+
+ kreq->completion.xid = iocmsg->xid;
+ kreq->completion.private = kreq;
+
+ kreq->flags = iocmsg->flags;
+
+ msg = &kreq->msg;
+ msg->private = krpc;
+ msg->private2 = kreq;
+
+ INIT_HLIST_NODE(&msg->kill_link);
+ pcs_clear_error(&msg->error);
+
+ msg->size = iocmsg->msg_size;
+ msg->timeout = iocmsg->timeout;
+
+ msg->rpc = NULL;
+ msg->done = pcs_krpc_msg_sent;
+ msg->get_iter = krpc_msg_get_data;
+
+ spin_lock(&krpc->lock);
+ kreq->krpc = pcs_krpc_get(krpc);
+ list_add_tail(&kreq->link, &krpc->pending_queue);
+ spin_unlock(&krpc->lock);
+ /* DTRACE to be added */
+ pcs_rpc_queue(krpc->rpc, msg);
+
+ return 0;
+
+err_free_data_chunk:
+ for (i = 0; i < kreq->nr_data_chunks; i++) {
+ chunk = &kreq->data_chunks[i];
+ if (chunk->type == KRPC_CHUNK_TYPE_UMEM)
+ pcs_umem_release(chunk->umem);
+ else
+ pcs_mr_put(chunk->mr);
+ }
+ pcs_mr_put(kreq->hdr_chunk.mr);
+
+ if (kreq->data_chunks != &kreq->inline_data_chunks[0])
+ kfree(kreq->data_chunks);
+
+err_free_kreq:
+ krpc_req_free(kreq);
+ return res;
+}
+
+static int pcs_krpc_abort(struct pcs_krpc *krpc)
+{
+ struct list_head dispose_list;
+ struct krpc_req *kreq;
+ struct krpc_completion *comp;
+
+ INIT_LIST_HEAD(&dispose_list);
+
+ spin_lock(&krpc->lock);
+
+ if (krpc->state != PCS_KRPC_STATE_CONNECTED) {
+ spin_unlock(&krpc->lock);
+ return 0;
+ }
+
+ krpc->state = PCS_KRPC_STATE_ABORTED;
+
+ /* abort incompleted requests */
+ list_splice_tail_init(&krpc->pending_queue, &krpc->dispose_queue);
+ while (!list_empty(&krpc->dispose_queue)) {
+ kreq = list_first_entry(&krpc->dispose_queue, struct krpc_req, link);
+ kreq->flags |= KRPC_REQ_F_ABORTED;
+ }
+
+ list_splice_tail_init(&krpc->completion_queue, &dispose_list);
+ krpc->nr_completion = 0;
+
+ /* dispose all unprocessed completions */
+ while (!list_empty(&dispose_list)) {
+ comp = list_first_entry(&dispose_list, struct krpc_completion, link);
+ list_del(&comp->link);
+ krpc_completion_free(comp);
+ }
+
+ spin_unlock(&krpc->lock);
+
+ return 0;
+}
+
+static long pcs_krpc_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
+{
+ struct pcs_krpc_context *ctx = file->private_data;
+ struct pcs_krpc *krpc = ctx->krpc;
+ int res = 0;
+
+ if (unlikely(ctx->gen != krpc->gen))
+ return -EFAULT;
+
+ switch (cmd) {
+ case PCS_KRPC_IOC_SEND_MSG: {
+ struct pcs_krpc_ioc_sendmsg req;
+
+ if (copy_from_user(&req, (void __user *)arg, sizeof(req)))
+ return -EFAULT;
+
+ res = pcs_krpc_ioctl_send_msg(krpc, &req);
+ break;
+ }
+ case PCS_KRPC_IOC_RECV_MSG: {
+ struct pcs_krpc_ioc_recvmsg req;
+
+ res = pcs_krpc_ioctl_recv_msg(krpc, &req);
+
+ if (copy_to_user((void __user *)arg, &req, sizeof(req)))
+ return -EFAULT;
+ break;
+ }
+ case PCS_KRPC_IOC_ABORT:
+ res = pcs_krpc_abort(krpc);
+ break;
+ default:
+ res = -ENOTTY;
+ break;
+ }
+
+ return res;
+}
+
+static __poll_t pcs_krpc_poll(struct file *file, poll_table *wait)
+{
+ struct pcs_krpc_context *ctx = file->private_data;
+ struct pcs_krpc *krpc = ctx->krpc;
+ __poll_t pollflags = 0;
+
+ poll_wait(file, &krpc->poll_wait, wait);
+
+ if (unlikely(ctx->gen != krpc->gen)) {
+ pollflags |= EPOLLERR;
+ return pollflags;
+ }
+
+ spin_lock(&krpc->lock);
+
+ if (krpc->state == PCS_KRPC_STATE_ABORTED)
+ pollflags |= EPOLLERR;
+ else if (krpc->state == PCS_KRPC_STATE_CONNECTED) {
+ pollflags |= EPOLLOUT;
+ if (!list_empty(&krpc->completion_queue))
+ pollflags |= EPOLLIN;
+ }
+
+ spin_unlock(&krpc->lock);
+
+ return pollflags;
+}
+
+/*
+ * Close connection
+ */
+static int pcs_krpc_release(struct inode *inode, struct file *file)
+{
+ struct pcs_krpc_context *ctx = file->private_data;
+ struct pcs_krpc *krpc = ctx->krpc;
+
+ /* User may have started a new connection before close the old one */
+ if (ctx->gen == krpc->gen)
+ pcs_krpc_abort(krpc);
+
+ pcs_krpc_put(krpc);
+ kfree(ctx);
+
+ return 0;
+}
+
+static const struct file_operations pcs_krpc_fops = {
+ .owner = THIS_MODULE,
+ .release = pcs_krpc_release,
+ .poll = pcs_krpc_poll,
+ .unlocked_ioctl = pcs_krpc_ioctl,
+ .llseek = no_llseek,
+};
+
+struct pcs_krpc *pcs_krpc_lookup(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
+{
+ struct pcs_krpc *krpc;
+
+ hlist_for_each_entry(krpc, &krpcs->ht[pcs_krpc_hash(id)], hlist) {
+ if (memcmp(&krpc->id, id, sizeof(krpc->id)) == 0)
+ return krpc;
+ }
+ return NULL;
+}
+
+struct pcs_krpc *pcs_krpc_get(struct pcs_krpc *krpc)
+{
+ refcount_inc(&krpc->refcnt);
+ return krpc;
+}
+
+void pcs_krpc_put(struct pcs_krpc *krpc)
+{
+ if (refcount_dec_and_test(&krpc->refcnt))
+ kfree(krpc);
+}
+
+/*
+ * Create a new pcs_krpc
+ */
+int pcs_krpc_create(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id,
+ PCS_NET_ADDR_T *addr, int cs_flags)
+{
+ struct pcs_krpc *krpc;
+
+ krpc = kzalloc(sizeof(struct pcs_krpc), GFP_NOIO);
+ if (!krpc)
+ return -ENOMEM;
+
+ INIT_HLIST_NODE(&krpc->hlist);
+ INIT_LIST_HEAD(&krpc->link);
+ spin_lock_init(&krpc->lock);
+ refcount_set(&krpc->refcnt, 1);
+ INIT_LIST_HEAD(&krpc->pending_queue);
+ INIT_LIST_HEAD(&krpc->dispose_queue);
+ INIT_LIST_HEAD(&krpc->completion_queue);
+ init_waitqueue_head(&krpc->poll_wait);
+
+ krpc->krpcs = krpcs;
+ krpc->id = *id;
+ krpc->gen = 0;
+ krpc->state = PCS_KRPC_STATE_UNCONN;
+
+ krpc->rpc = pcs_rpc_clnt_create(&cc_from_krpcset(krpcs)->eng, id, addr, cs_flags);
+ if (!krpc->rpc) {
+ kfree(krpc);
+ return -ENOMEM;
+ }
+
+ krpc->rpc->clnt_krpc = krpc;
+
+ spin_lock(&krpc->lock);
+ spin_lock(&krpcs->lock);
+ hlist_add_head(&krpc->hlist, &krpcs->ht[pcs_krpc_hash(&krpc->id)]);
+ list_add_tail(&krpc->link, &krpcs->list);
+ krpcs->nkrpc++;
+ spin_unlock(&krpcs->lock);
+ spin_unlock(&krpc->lock);
+
+ return 0;
+}
+
+int pcs_krpc_update_addr(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id,
+ PCS_NET_ADDR_T *addr, int flags)
+{
+ struct pcs_krpc *krpc;
+
+ krpc = pcs_krpc_lookup(krpcs, id);
+ if (!krpc)
+ return -ENXIO;
+
+ pcs_rpc_set_address(krpc->rpc, addr);
+
+ if (flags & CS_FL_LOCAL_SOCK)
+ krpc->rpc->flags |= PCS_RPC_F_LOCAL;
+ else
+ krpc->rpc->flags &= ~PCS_RPC_F_LOCAL;
+
+ return 0;
+}
+/*
+ * Connect to a pcs_krpc, return a valid fd on success.
+ */
+int pcs_krpc_connect(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
+{
+ struct pcs_krpc *krpc;
+ int fd;
+ struct file *file;
+ struct pcs_krpc_context *ctx;
+
+ krpc = pcs_krpc_lookup(krpcs, id);
+ if (!krpc)
+ return -ENXIO;
+
+ if (krpc->state == PCS_KRPC_STATE_CONNECTED ||
+ krpc->state == PCS_KRPC_STATE_DESTROYED)
+ return -EPERM;
+
+ ctx = kmalloc(sizeof(*ctx), GFP_KERNEL);
+ if (!ctx)
+ return -ENOMEM;
+
+ fd = get_unused_fd_flags(O_CLOEXEC);
+ if (fd < 0) {
+ kfree(ctx);
+ return fd;
+ }
+
+ file = anon_inode_getfile("[pcs_krpc]", &pcs_krpc_fops, ctx, 0);
+ if (IS_ERR(file)) {
+ kfree(ctx);
+ put_unused_fd(fd);
+ fd = PTR_ERR(file);
+ return fd;
+ }
+
+ fd_install(fd, file);
+
+ 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;
+ spin_unlock(&krpc->lock);
+
+ return fd;
+}
+
+static void __pcs_krpc_destroy(struct pcs_krpc *krpc)
+{
+ spin_lock(&krpc->lock);
+
+ krpc->state = PCS_KRPC_STATE_DESTROYED;
+
+ /*Remove from krpc set*/
+ spin_lock(&krpc->krpcs->lock);
+ hlist_del(&krpc->hlist);
+ list_del(&krpc->link);
+ krpc->krpcs->nkrpc--;
+ spin_unlock(&krpc->krpcs->lock);
+
+ spin_unlock(&krpc->lock);
+
+ if (krpc->rpc) {
+ krpc->rpc->clnt_krpc = NULL;
+ pcs_rpc_clnt_close(krpc->rpc);
+ krpc->rpc = NULL;
+ }
+ pcs_krpc_put(krpc);
+}
+
+int pcs_krpc_destroy(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id)
+{
+ struct pcs_krpc *krpc;
+
+ krpc = pcs_krpc_lookup(krpcs, id);
+ BUG_ON(!krpc);
+
+ /* Abort current connection */
+ pcs_krpc_abort(krpc);
+
+ __pcs_krpc_destroy(krpc);
+
+ return 0;
+}
+
+void pcs_krpcset_init(struct pcs_krpc_set *krpcs)
+{
+ unsigned int i;
+
+ for (i = 0; i < PCS_KRPC_HASH_SIZE; i++)
+ INIT_HLIST_HEAD(&krpcs->ht[i]);
+
+ INIT_LIST_HEAD(&krpcs->list);
+ krpcs->nkrpc = 0;
+
+ spin_lock_init(&krpcs->lock);
+}
+
+void pcs_krpcset_fini(struct pcs_krpc_set *krpcs)
+{
+ spin_lock(&krpcs->lock);
+ while (!list_empty(&krpcs->list)) {
+ struct pcs_krpc *krpc;
+
+ krpc = list_first_entry(&krpcs->list, struct pcs_krpc, link);
+ spin_unlock(&krpcs->lock);
+ pcs_krpc_abort(krpc);
+ __pcs_krpc_destroy(krpc);
+ spin_lock(&krpcs->lock);
+ }
+ spin_unlock(&krpcs->lock);
+
+ BUG_ON(!list_empty(&krpcs->list));
+ BUG_ON(krpcs->nkrpc != 0);
+}
+
+int __init pcs_krpc_init(void)
+{
+ krpc_req_cachep = kmem_cache_create("pcs_krpc_req",
+ sizeof(struct krpc_req), 0,
+ SLAB_RECLAIM_ACCOUNT|SLAB_ACCOUNT, NULL);
+
+ if (!krpc_req_cachep)
+ return -ENOMEM;
+
+ return 0;
+}
+
+void pcs_krpc_fini(void)
+{
+ kmem_cache_destroy(krpc_req_cachep);
+}
diff --git a/fs/fuse/kio/pcs/pcs_krpc.h b/fs/fuse/kio/pcs/pcs_krpc.h
new file mode 100644
index 000000000000..6c0ef20ebc99
--- /dev/null
+++ b/fs/fuse/kio/pcs/pcs_krpc.h
@@ -0,0 +1,144 @@
+/*
+ * Copyright (c) 2018-2021 Virtuozzo International GmbH. All rights reserved.
+ */
+
+#ifndef _PCS_KRPC_H_
+#define _PCS_KRPC_H_ 1
+#include <linux/types.h>
+#include <linux/bvec.h>
+
+#include "pcs_prot_types.h"
+#include "pcs_perfcounters.h"
+#include "pcs_rpc_clnt.h"
+#include "pcs_krpc_prot.h"
+#include "pcs_mr.h"
+
+struct krpc_chunk {
+ u64 addr;
+ u32 len;
+ u32 type;
+#define KRPC_CHUNK_TYPE_MR 0
+#define KRPC_CHUNK_TYPE_UMEM 1
+ union {
+ struct pcs_mr *mr;
+ struct pcs_umem *umem;
+ };
+};
+
+#define PCS_KRPC_HASH_SIZE 1024
+struct pcs_krpc_set {
+ /* Set of krpcs */
+ struct hlist_head ht[PCS_KRPC_HASH_SIZE];
+ struct list_head list;
+ unsigned int nkrpc;
+
+ spinlock_t lock;
+};
+
+enum {
+ PCS_KRPC_STATE_UNCONN,
+ PCS_KRPC_STATE_CONNECTED,
+ PCS_KRPC_STATE_ABORTED,
+ PCS_KRPC_STATE_DESTROYED,
+};
+
+struct pcs_krpc {
+ struct hlist_node hlist;
+ struct list_head link;
+ spinlock_t lock;
+ refcount_t refcnt;
+
+ struct pcs_rpc *rpc;
+
+ struct pcs_krpc_set *krpcs;
+
+ PCS_NODE_ID_T id;
+
+ u32 state;
+ u32 gen;
+
+ struct list_head pending_queue;
+ struct list_head dispose_queue;
+ struct list_head completion_queue;
+ int nr_completion;
+
+ /** Wait queue head for poll */
+ wait_queue_head_t poll_wait;
+};
+
+struct pcs_krpc_context {
+ struct pcs_krpc *krpc;
+ u32 gen;
+};
+/*
+ * Completion message to be received by userspace
+ */
+struct krpc_completion {
+ struct list_head link; /* in krpc->completion_queue */
+
+ u64 xid;
+ int result;
+
+ void *private;
+ int data_len;
+ u8 _data_buf[0];
+};
+
+#define KRPC_MAX_DATA_PAGES 256
+#define NR_KRPC_DATA_CHUNKS_INLINE 4
+struct krpc_req {
+ struct list_head link;
+ spinlock_t lock;
+ struct pcs_krpc *krpc;
+
+#define KRPC_REQ_F_ALIGNMENT PCS_KRPC_MSG_F_ALIGNMENT
+#define KRPC_REQ_F_RESP_BUFF PCS_KRPC_MSG_F_RESP_BUFF /* data buff is for read response */
+#define KRPC_REQ_F_ABORTED 0x10000
+ int flags;
+
+ struct pcs_msg msg;
+
+ char *hdr_buf;
+ struct kvec hdr_kv;
+ struct krpc_chunk hdr_chunk;
+
+ int nr_data_chunks;
+ struct krpc_chunk *data_chunks;
+ struct krpc_chunk inline_data_chunks[NR_KRPC_DATA_CHUNKS_INLINE];
+
+ int data_len;
+ int nr_data_bvecs;
+ struct bio_vec data_bvecs[KRPC_MAX_DATA_PAGES];
+
+ struct krpc_completion completion;
+};
+
+static inline u32 pcs_krpc_msg_size(u32 size, u8 flags)
+{
+ if (flags & PCS_KRPC_MSG_F_ALIGNMENT)
+ size = ALIGN(size, PCS_KRPC_MSG_ALIGNMENT);
+
+ return size;
+}
+
+int pcs_krpc_init(void);
+void pcs_krpc_fini(void);
+
+void pcs_krpcset_init(struct pcs_krpc_set *krpcs);
+void pcs_krpcset_fini(struct pcs_krpc_set *krpcs);
+
+struct pcs_krpc *pcs_krpc_lookup(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id);
+int pcs_krpc_create(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id,
+ PCS_NET_ADDR_T *addr, int cs_flags);
+int pcs_krpc_update_addr(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id,
+ PCS_NET_ADDR_T *addr, int flags);
+int pcs_krpc_connect(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id);
+int pcs_krpc_destroy(struct pcs_krpc_set *krpcs, PCS_NODE_ID_T *id);
+
+void pcs_krpc_put(struct pcs_krpc *krpc);
+struct pcs_krpc *pcs_krpc_get(struct pcs_krpc *krpc);
+
+struct pcs_msg *krpc_get_hdr(struct pcs_rpc *ep, struct pcs_rpc_hdr *h);
+void krpc_keep_waiting(struct pcs_rpc *ep, struct pcs_msg *req, struct pcs_msg *msg);
+void krpc_handle_congestion(struct pcs_rpc *ep, struct pcs_msg *msg);
+#endif
diff --git a/fs/fuse/kio/pcs/pcs_krpc_prot.h b/fs/fuse/kio/pcs/pcs_krpc_prot.h
new file mode 100644
index 000000000000..4c5bbf4492d1
--- /dev/null
+++ b/fs/fuse/kio/pcs/pcs_krpc_prot.h
@@ -0,0 +1,44 @@
+#ifndef _PCS_KRPC_PROT_H_
+#define _PCS_KRPC_PROT_H_
+#include <linux/ioctl.h>
+
+#include "pcs_prot_types.h"
+
+/* Device ioctls: */
+#define PCS_KRPC_IOC_MAGIC 255
+
+/* ioctl cmds supported by the '[pcs_krpc]' anonymous inode */
+#define PCS_KRPC_IOC_SEND_MSG _IO(PCS_KRPC_IOC_MAGIC, 10)
+#define PCS_KRPC_IOC_RECV_MSG _IO(PCS_KRPC_IOC_MAGIC, 11)
+#define PCS_KRPC_IOC_ABORT _IO(PCS_KRPC_IOC_MAGIC, 12)
+
+struct pcs_krpc_buf_desc {
+ u64 addr; /* buf address in userspace. */
+ u32 len; /* size of the buf */
+ u32 mr_id; /* mr id */
+};
+
+#define PCS_KRPC_MSG_ALIGNMENT (512ULL)
+
+#define PCS_KRPC_MSG_F_ALIGNMENT 0x01
+#define PCS_KRPC_MSG_F_RESP_BUFF 0x02
+/*
+ * To avoid copying, a msg is sent to kernel in an array of buffer descriptor.
+ * Each buffer descriptor points to a buf contains a chunk of the msg. And all
+ * chunk buffers are from registered MRs
+ */
+struct pcs_krpc_ioc_sendmsg {
+ u64 xid; /* context id */
+ u32 msg_size; /* total size of the msg */
+ u16 timeout; /* timeout */
+ u8 flags; /* alignment, */
+ u8 nr_data_chunks; /* total number of data chunks */
+ struct pcs_krpc_buf_desc hdr_chunk; /* the buf holding msg header */
+};
+
+struct pcs_krpc_ioc_recvmsg {
+ u64 xid; /* context id */
+ struct pcs_krpc_buf_desc buf; /* for cs congestion notification and rpc keep waiting msg.*/
+};
+
+#endif
--
2.39.3 (Apple Git-146)
More information about the Devel
mailing list