[Devel] [PATCH RK hci-7.4 2/2] fs/fuse kio: repoint RDMA handlers and restrack names on RK load

[Denis V. Lunev]

Up to v8.53 the RK carried its own copies of pcs_rdma_io_event_handler(),
pcs_rdma_cq_event_handler() and pcs_rdma_qp_event_handler(), and every
RDMA connection opened while such an RK was active points at them.
"kpatch replace" unloads the replaced RK right after the new one is
enabled, so these connections are left with dangling handlers and crash
the node on their next CM, QP or CQ event.

The same connections carry that RK's KBUILD_MODNAME as restrack name of
their PD and CQ, and the MRs allocated from the PD inherit it. Any
"rdma resource show pd|cq|mr", which needs no privileges, reads it.
The PD and CQ hang off the connection. MRs in use are on no pcs list,
so they are found through the device's restrack MR table.

>From a post-patch callback, rewrite the handlers of every live
connection to the fuse_kio_pcs originals, and the PD, CQ and MR names to
the name of the fuse_kio_pcs module itself. Afterwards nothing points
into any RK module, this one included, so neither a later replace nor
an unload depends on callbacks of the outgoing RK, which "kpatch replace"
does not run anyway.

At that point the transition is complete, so the old pcs_rdma_create()
can no longer run, while the outgoing RK text is still mapped, so a
racing caller still lands in valid code. The values are written
unconditionally, which also repairs connections already pointing into an
RK that has been unloaded.

All endpoints are reached through pcs_client_list and the engine hash.
Each one is fixed under ep->mutex: pcs_rdmaconnect_start() holds it
from pcs_rdma_create() until ep->conn is set, every RDMA netio op runs
under it, so no MR can be allocated from the PD meanwhile, and a
connection is detached from ep->conn and queued on rio_destroy_list
under the same mutex, so flushing rio_destroy_work afterwards covers the
rest. synchronize_rcu_tasks() then waits for anyone preempted between
loading an old handler and calling it.

https://virtuozzo.atlassian.net/browse/VSTOR-151314
Signed-off-by: Denis V. Lunev <den at openvz.org>
---
 fs/fuse/kio/pcs/pcs_fuse_kdirect.c | 88 ++++++++++++++++++++++++++++++
 fs/fuse/kio/pcs/pcs_rdma_io.c      | 44 +++++++++++++++
 2 files changed, 132 insertions(+)

diff --git a/fs/fuse/kio/pcs/pcs_fuse_kdirect.c b/fs/fuse/kio/pcs/pcs_fuse_kdirect.c
index 1031c43..5c7d461 100644
--- a/fs/fuse/kio/pcs/pcs_fuse_kdirect.c
+++ b/fs/fuse/kio/pcs/pcs_fuse_kdirect.c
@@ -2121,3 +2121,91 @@ module_exit(kpcs_mod_exit);
 
 MODULE_LICENSE("GPL v2");
 MODULE_AUTHOR("Virtuozzo <devel at openvz.org>");
+
+#include <linux/rcupdate.h>
+#include "kpatch-macros.h"
+
+void pcs_rdma_rk_fixup_ep(struct pcs_rpc *ep, const char *name);
+void pcs_rdma_rk_fixup_flush(void);
+
+/*
+ * [RK] Grab a reference to every RPC endpoint. Returns false if more
+ * endpoints than @cap were found, the caller has to retry then.
+ */
+static bool rk_collect_eps(struct pcs_rpc **eps, unsigned int cap,
+			   unsigned int *nr)
+{
+	struct pcs_fuse_cluster *pfc;
+	struct pcs_rpc *ep;
+	unsigned int n = 0, i;
+	bool fit = true;
+
+	spin_lock(&pcs_clients_lock);
+	list_for_each_entry(pfc, &pcs_client_list, list) {
+		struct pcs_rpc_engine *eng = &pfc->cc.eng;
+
+		spin_lock(&eng->lock);
+		for (i = 0; i <= PCS_RPC_HASH_SIZE && fit; i++) {
+			struct hlist_head *head = i < PCS_RPC_HASH_SIZE ?
+						  &eng->ht[i] : &eng->unhashed;
+
+			hlist_for_each_entry(ep, head, link) {
+				if (n == cap) {
+					fit = false;
+					break;
+				}
+				if (atomic_inc_not_zero(&ep->refcnt))
+					eps[n++] = ep;
+			}
+		}
+		spin_unlock(&eng->lock);
+		if (!fit)
+			break;
+	}
+	spin_unlock(&pcs_clients_lock);
+
+	*nr = n;
+	return fit;
+}
+
+static unsigned int rk_count_eps(void)
+{
+	struct pcs_fuse_cluster *pfc;
+	unsigned int n = 0;
+
+	spin_lock(&pcs_clients_lock);
+	list_for_each_entry(pfc, &pcs_client_list, list)
+		n += READ_ONCE(pfc->cc.eng.nrpcs);
+	spin_unlock(&pcs_clients_lock);
+
+	return n;
+}
+
+static void rk_fixup_rdma(patch_object *obj)
+{
+	struct pcs_rpc **eps;
+	unsigned int cap, n, i;
+	bool done;
+
+	do {
+		cap = rk_count_eps() + 16;
+		while (!(eps = kvmalloc_array(cap, sizeof(*eps), GFP_KERNEL)))
+			msleep(100);
+
+		done = rk_collect_eps(eps, cap, &n);
+		for (i = 0; i < n; i++) {
+			mutex_lock(&eps[i]->mutex);
+			pcs_rdma_rk_fixup_ep(eps[i], obj->mod->name);
+			mutex_unlock(&eps[i]->mutex);
+			pcs_rpc_put(eps[i]);
+		}
+		kvfree(eps);
+	} while (!done);
+
+	/* rio detached from its ep under ep->mutex sit here until destroyed */
+	pcs_rdma_rk_fixup_flush();
+
+	/* Wait for anyone preempted between loading a handler and calling it */
+	synchronize_rcu_tasks();
+}
+KPATCH_POST_PATCH_CALLBACK(rk_fixup_rdma);
diff --git a/fs/fuse/kio/pcs/pcs_rdma_io.c b/fs/fuse/kio/pcs/pcs_rdma_io.c
index 67f3aad..cf12ede 100644
--- a/fs/fuse/kio/pcs/pcs_rdma_io.c
+++ b/fs/fuse/kio/pcs/pcs_rdma_io.c
@@ -1823,3 +1823,47 @@ struct pcs_netio_tops pcs_rdma_netio_tops = {
 	.sync_send		= pcs_rdma_sync_send,
 	.sync_recv		= pcs_rdma_sync_recv,
 };
+
+#include "../../../../drivers/infiniband/core/restrack.h"
+
+/*
+ * [RK] Connections created while an older RK was active point to that RK's
+ * copies of the handlers below, and their PD, CQ and MRs carry that RK's
+ * KBUILD_MODNAME as restrack name. Repoint them into fuse_kio_pcs before the
+ * RK module is unloaded.
+ */
+void pcs_rdma_rk_fixup_ep(struct pcs_rpc *ep, const char *name)
+{
+	struct pcs_rdmaio *rio;
+	struct rdma_restrack_entry *res;
+	struct ib_pd *pd;
+	struct xarray *xa;
+	unsigned long index;
+
+	BUG_ON(!mutex_is_locked(&ep->mutex));
+
+	if (!ep->conn || ((struct pcs_netio *)ep->conn)->tops != &pcs_rdma_netio_tops)
+		return;
+
+	rio = rio_from_ioconn(ep->conn);
+	WRITE_ONCE(rio->id.event_handler, pcs_rdma_io_event_handler);
+	WRITE_ONCE(rio->cq->event_handler, pcs_rdma_cq_event_handler);
+	WRITE_ONCE(rio->cmid->qp->event_handler, pcs_rdma_qp_event_handler);
+
+	pd = rio->dev->pd;
+	WRITE_ONCE(pd->res.kern_name, name);
+	WRITE_ONCE(rio->cq->res.kern_name, name);
+
+	xa = &pd->device->res[RDMA_RESTRACK_MR].xa;
+	xa_lock(xa);
+	xa_for_each(xa, index, res) {
+		if (container_of(res, struct ib_mr, res)->pd == pd)
+			WRITE_ONCE(res->kern_name, name);
+	}
+	xa_unlock(xa);
+}
+
+void pcs_rdma_rk_fixup_flush(void)
+{
+	flush_work(&rio_destroy_work);
+}
-- 
2.53.0