xref: /openbmc/linux/net/sunrpc/xprtrdma/backchannel.c (revision 7c8d9e7c8863905951d4eaa7a8d277150f3a37f7)
1b2441318SGreg Kroah-Hartman // SPDX-License-Identifier: GPL-2.0
2f531a5dbSChuck Lever /*
3f531a5dbSChuck Lever  * Copyright (c) 2015 Oracle.  All rights reserved.
4f531a5dbSChuck Lever  *
5f531a5dbSChuck Lever  * Support for backward direction RPCs on RPC/RDMA.
6f531a5dbSChuck Lever  */
7f531a5dbSChuck Lever 
8f531a5dbSChuck Lever #include <linux/module.h>
963cae470SChuck Lever #include <linux/sunrpc/xprt.h>
1063cae470SChuck Lever #include <linux/sunrpc/svc.h>
1176566773SChuck Lever #include <linux/sunrpc/svc_xprt.h>
12f531a5dbSChuck Lever 
13f531a5dbSChuck Lever #include "xprt_rdma.h"
14f531a5dbSChuck Lever 
15f531a5dbSChuck Lever #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
16f531a5dbSChuck Lever # define RPCDBG_FACILITY	RPCDBG_TRANS
17f531a5dbSChuck Lever #endif
18f531a5dbSChuck Lever 
19c8bbe0c7SChuck Lever #undef RPCRDMA_BACKCHANNEL_DEBUG
2063cae470SChuck Lever 
21f531a5dbSChuck Lever static void rpcrdma_bc_free_rqst(struct rpcrdma_xprt *r_xprt,
22f531a5dbSChuck Lever 				 struct rpc_rqst *rqst)
23f531a5dbSChuck Lever {
24f531a5dbSChuck Lever 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
25f531a5dbSChuck Lever 	struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
26f531a5dbSChuck Lever 
27f531a5dbSChuck Lever 	spin_lock(&buf->rb_reqslock);
28f531a5dbSChuck Lever 	list_del(&req->rl_all);
29f531a5dbSChuck Lever 	spin_unlock(&buf->rb_reqslock);
30f531a5dbSChuck Lever 
3113650c23SChuck Lever 	rpcrdma_destroy_req(req);
32f531a5dbSChuck Lever }
33f531a5dbSChuck Lever 
34edb41e61SChuck Lever static int rpcrdma_bc_setup_reqs(struct rpcrdma_xprt *r_xprt,
35edb41e61SChuck Lever 				 unsigned int count)
36f531a5dbSChuck Lever {
37edb41e61SChuck Lever 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
38edb41e61SChuck Lever 	struct rpc_rqst *rqst;
39edb41e61SChuck Lever 	unsigned int i;
40edb41e61SChuck Lever 
41edb41e61SChuck Lever 	for (i = 0; i < (count << 1); i++) {
42f531a5dbSChuck Lever 		struct rpcrdma_regbuf *rb;
43f531a5dbSChuck Lever 		struct rpcrdma_req *req;
44f531a5dbSChuck Lever 		size_t size;
45f531a5dbSChuck Lever 
46f531a5dbSChuck Lever 		req = rpcrdma_create_req(r_xprt);
47abfb6897SDan Carpenter 		if (IS_ERR(req))
48abfb6897SDan Carpenter 			return PTR_ERR(req);
49edb41e61SChuck Lever 		rqst = &req->rl_slot;
50edb41e61SChuck Lever 
51edb41e61SChuck Lever 		rqst->rq_xprt = xprt;
52edb41e61SChuck Lever 		INIT_LIST_HEAD(&rqst->rq_list);
53edb41e61SChuck Lever 		INIT_LIST_HEAD(&rqst->rq_bc_list);
54edb41e61SChuck Lever 		__set_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state);
55edb41e61SChuck Lever 		spin_lock_bh(&xprt->bc_pa_lock);
56edb41e61SChuck Lever 		list_add(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
57edb41e61SChuck Lever 		spin_unlock_bh(&xprt->bc_pa_lock);
58f531a5dbSChuck Lever 
5908cf2efdSChuck Lever 		size = r_xprt->rx_data.inline_rsize;
6013650c23SChuck Lever 		rb = rpcrdma_alloc_regbuf(size, DMA_TO_DEVICE, GFP_KERNEL);
61f531a5dbSChuck Lever 		if (IS_ERR(rb))
62f531a5dbSChuck Lever 			goto out_fail;
63f531a5dbSChuck Lever 		req->rl_sendbuf = rb;
6462aee0e3SChuck Lever 		xdr_buf_init(&rqst->rq_snd_buf, rb->rg_base,
6562aee0e3SChuck Lever 			     min_t(size_t, size, PAGE_SIZE));
66edb41e61SChuck Lever 	}
67f531a5dbSChuck Lever 	return 0;
68f531a5dbSChuck Lever 
69f531a5dbSChuck Lever out_fail:
70f531a5dbSChuck Lever 	rpcrdma_bc_free_rqst(r_xprt, rqst);
71f531a5dbSChuck Lever 	return -ENOMEM;
72f531a5dbSChuck Lever }
73f531a5dbSChuck Lever 
74f531a5dbSChuck Lever /**
75f531a5dbSChuck Lever  * xprt_rdma_bc_setup - Pre-allocate resources for handling backchannel requests
76f531a5dbSChuck Lever  * @xprt: transport associated with these backchannel resources
77f531a5dbSChuck Lever  * @reqs: number of concurrent incoming requests to expect
78f531a5dbSChuck Lever  *
79f531a5dbSChuck Lever  * Returns 0 on success; otherwise a negative errno
80f531a5dbSChuck Lever  */
81f531a5dbSChuck Lever int xprt_rdma_bc_setup(struct rpc_xprt *xprt, unsigned int reqs)
82f531a5dbSChuck Lever {
83f531a5dbSChuck Lever 	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
84f531a5dbSChuck Lever 	int rc;
85f531a5dbSChuck Lever 
86f531a5dbSChuck Lever 	/* The backchannel reply path returns each rpc_rqst to the
87f531a5dbSChuck Lever 	 * bc_pa_list _after_ the reply is sent. If the server is
88f531a5dbSChuck Lever 	 * faster than the client, it can send another backward
89f531a5dbSChuck Lever 	 * direction request before the rpc_rqst is returned to the
90f531a5dbSChuck Lever 	 * list. The client rejects the request in this case.
91f531a5dbSChuck Lever 	 *
92f531a5dbSChuck Lever 	 * Twice as many rpc_rqsts are prepared to ensure there is
93f531a5dbSChuck Lever 	 * always an rpc_rqst available as soon as a reply is sent.
94f531a5dbSChuck Lever 	 */
95124fa17dSChuck Lever 	if (reqs > RPCRDMA_BACKWARD_WRS >> 1)
96124fa17dSChuck Lever 		goto out_err;
97124fa17dSChuck Lever 
98edb41e61SChuck Lever 	rc = rpcrdma_bc_setup_reqs(r_xprt, reqs);
99edb41e61SChuck Lever 	if (rc)
100f531a5dbSChuck Lever 		goto out_free;
101d2c23c00SMarkus Elfring 
102edb41e61SChuck Lever 	r_xprt->rx_buf.rb_bc_srv_max_requests = reqs;
103f531a5dbSChuck Lever 	request_module("svcrdma");
104fc1eb807SChuck Lever 	trace_xprtrdma_cb_setup(r_xprt, reqs);
105f531a5dbSChuck Lever 	return 0;
106f531a5dbSChuck Lever 
107f531a5dbSChuck Lever out_free:
108f531a5dbSChuck Lever 	xprt_rdma_bc_destroy(xprt, reqs);
109f531a5dbSChuck Lever 
110124fa17dSChuck Lever out_err:
111f531a5dbSChuck Lever 	pr_err("RPC:       %s: setup backchannel transport failed\n", __func__);
112f531a5dbSChuck Lever 	return -ENOMEM;
113f531a5dbSChuck Lever }
114f531a5dbSChuck Lever 
115f531a5dbSChuck Lever /**
11676566773SChuck Lever  * xprt_rdma_bc_up - Create transport endpoint for backchannel service
11776566773SChuck Lever  * @serv: server endpoint
11876566773SChuck Lever  * @net: network namespace
11976566773SChuck Lever  *
12076566773SChuck Lever  * The "xprt" is an implied argument: it supplies the name of the
12176566773SChuck Lever  * backchannel transport class.
12276566773SChuck Lever  *
12376566773SChuck Lever  * Returns zero on success, negative errno on failure
12476566773SChuck Lever  */
12576566773SChuck Lever int xprt_rdma_bc_up(struct svc_serv *serv, struct net *net)
12676566773SChuck Lever {
12776566773SChuck Lever 	int ret;
12876566773SChuck Lever 
12976566773SChuck Lever 	ret = svc_create_xprt(serv, "rdma-bc", net, PF_INET, 0, 0);
13076566773SChuck Lever 	if (ret < 0)
13176566773SChuck Lever 		return ret;
13276566773SChuck Lever 	return 0;
13376566773SChuck Lever }
13476566773SChuck Lever 
13576566773SChuck Lever /**
1366b26cc8cSChuck Lever  * xprt_rdma_bc_maxpayload - Return maximum backchannel message size
1376b26cc8cSChuck Lever  * @xprt: transport
1386b26cc8cSChuck Lever  *
1396b26cc8cSChuck Lever  * Returns maximum size, in bytes, of a backchannel message
1406b26cc8cSChuck Lever  */
1416b26cc8cSChuck Lever size_t xprt_rdma_bc_maxpayload(struct rpc_xprt *xprt)
1426b26cc8cSChuck Lever {
1436b26cc8cSChuck Lever 	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
1446b26cc8cSChuck Lever 	struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
1456b26cc8cSChuck Lever 	size_t maxmsg;
1466b26cc8cSChuck Lever 
1476b26cc8cSChuck Lever 	maxmsg = min_t(unsigned int, cdata->inline_rsize, cdata->inline_wsize);
14862aee0e3SChuck Lever 	maxmsg = min_t(unsigned int, maxmsg, PAGE_SIZE);
1496b26cc8cSChuck Lever 	return maxmsg - RPCRDMA_HDRLEN_MIN;
1506b26cc8cSChuck Lever }
1516b26cc8cSChuck Lever 
152cf73daf5SChuck Lever static int rpcrdma_bc_marshal_reply(struct rpc_rqst *rqst)
15383128a60SChuck Lever {
1547ec910e7SChuck Lever 	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt);
15583128a60SChuck Lever 	struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
1567ec910e7SChuck Lever 	__be32 *p;
15783128a60SChuck Lever 
1587ec910e7SChuck Lever 	rpcrdma_set_xdrlen(&req->rl_hdrbuf, 0);
1597ec910e7SChuck Lever 	xdr_init_encode(&req->rl_stream, &req->rl_hdrbuf,
1607ec910e7SChuck Lever 			req->rl_rdmabuf->rg_base);
1617ec910e7SChuck Lever 
1627ec910e7SChuck Lever 	p = xdr_reserve_space(&req->rl_stream, 28);
1637ec910e7SChuck Lever 	if (unlikely(!p))
1647ec910e7SChuck Lever 		return -EIO;
1657ec910e7SChuck Lever 	*p++ = rqst->rq_xid;
1667ec910e7SChuck Lever 	*p++ = rpcrdma_version;
1677ec910e7SChuck Lever 	*p++ = cpu_to_be32(r_xprt->rx_buf.rb_bc_srv_max_requests);
1687ec910e7SChuck Lever 	*p++ = rdma_msg;
1697ec910e7SChuck Lever 	*p++ = xdr_zero;
1707ec910e7SChuck Lever 	*p++ = xdr_zero;
1717ec910e7SChuck Lever 	*p = xdr_zero;
17283128a60SChuck Lever 
173857f9acaSChuck Lever 	if (rpcrdma_prepare_send_sges(r_xprt, req, RPCRDMA_HDRLEN_MIN,
174655fec69SChuck Lever 				      &rqst->rq_snd_buf, rpcrdma_noch))
17554cbd6b0SChuck Lever 		return -EIO;
176fc1eb807SChuck Lever 
177fc1eb807SChuck Lever 	trace_xprtrdma_cb_reply(rqst);
178655fec69SChuck Lever 	return 0;
17983128a60SChuck Lever }
18083128a60SChuck Lever 
18183128a60SChuck Lever /**
182cf73daf5SChuck Lever  * xprt_rdma_bc_send_reply - marshal and send a backchannel reply
183cf73daf5SChuck Lever  * @rqst: RPC rqst with a backchannel RPC reply in rq_snd_buf
184cf73daf5SChuck Lever  *
185cf73daf5SChuck Lever  * Caller holds the transport's write lock.
186cf73daf5SChuck Lever  *
187cf73daf5SChuck Lever  * Returns:
188cf73daf5SChuck Lever  *	%0 if the RPC message has been sent
189cf73daf5SChuck Lever  *	%-ENOTCONN if the caller should reconnect and call again
190cf73daf5SChuck Lever  *	%-EIO if a permanent error occurred and the request was not
191cf73daf5SChuck Lever  *		sent. Do not try to send this message again.
192cf73daf5SChuck Lever  */
193cf73daf5SChuck Lever int xprt_rdma_bc_send_reply(struct rpc_rqst *rqst)
194cf73daf5SChuck Lever {
195cf73daf5SChuck Lever 	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt);
196cf73daf5SChuck Lever 	struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
197cf73daf5SChuck Lever 	int rc;
198cf73daf5SChuck Lever 
199cf73daf5SChuck Lever 	if (!xprt_connected(rqst->rq_xprt))
200cf73daf5SChuck Lever 		goto drop_connection;
201cf73daf5SChuck Lever 
202cf73daf5SChuck Lever 	rc = rpcrdma_bc_marshal_reply(rqst);
203cf73daf5SChuck Lever 	if (rc < 0)
204cf73daf5SChuck Lever 		goto failed_marshal;
205cf73daf5SChuck Lever 
206*7c8d9e7cSChuck Lever 	rpcrdma_post_recvs(r_xprt, true);
207cf73daf5SChuck Lever 	if (rpcrdma_ep_post(&r_xprt->rx_ia, &r_xprt->rx_ep, req))
208cf73daf5SChuck Lever 		goto drop_connection;
209cf73daf5SChuck Lever 	return 0;
210cf73daf5SChuck Lever 
211cf73daf5SChuck Lever failed_marshal:
212cf73daf5SChuck Lever 	if (rc != -ENOTCONN)
213cf73daf5SChuck Lever 		return rc;
214cf73daf5SChuck Lever drop_connection:
215cf73daf5SChuck Lever 	xprt_disconnect_done(rqst->rq_xprt);
216cf73daf5SChuck Lever 	return -ENOTCONN;
217cf73daf5SChuck Lever }
218cf73daf5SChuck Lever 
219cf73daf5SChuck Lever /**
220f531a5dbSChuck Lever  * xprt_rdma_bc_destroy - Release resources for handling backchannel requests
221f531a5dbSChuck Lever  * @xprt: transport associated with these backchannel resources
222f531a5dbSChuck Lever  * @reqs: number of incoming requests to destroy; ignored
223f531a5dbSChuck Lever  */
224f531a5dbSChuck Lever void xprt_rdma_bc_destroy(struct rpc_xprt *xprt, unsigned int reqs)
225f531a5dbSChuck Lever {
226f531a5dbSChuck Lever 	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
227f531a5dbSChuck Lever 	struct rpc_rqst *rqst, *tmp;
228f531a5dbSChuck Lever 
229f531a5dbSChuck Lever 	spin_lock_bh(&xprt->bc_pa_lock);
230f531a5dbSChuck Lever 	list_for_each_entry_safe(rqst, tmp, &xprt->bc_pa_list, rq_bc_pa_list) {
231f531a5dbSChuck Lever 		list_del(&rqst->rq_bc_pa_list);
232f531a5dbSChuck Lever 		spin_unlock_bh(&xprt->bc_pa_lock);
233f531a5dbSChuck Lever 
234f531a5dbSChuck Lever 		rpcrdma_bc_free_rqst(r_xprt, rqst);
235f531a5dbSChuck Lever 
236f531a5dbSChuck Lever 		spin_lock_bh(&xprt->bc_pa_lock);
237f531a5dbSChuck Lever 	}
238f531a5dbSChuck Lever 	spin_unlock_bh(&xprt->bc_pa_lock);
239f531a5dbSChuck Lever }
240f531a5dbSChuck Lever 
241f531a5dbSChuck Lever /**
242f531a5dbSChuck Lever  * xprt_rdma_bc_free_rqst - Release a backchannel rqst
243f531a5dbSChuck Lever  * @rqst: request to release
244f531a5dbSChuck Lever  */
245f531a5dbSChuck Lever void xprt_rdma_bc_free_rqst(struct rpc_rqst *rqst)
246f531a5dbSChuck Lever {
247*7c8d9e7cSChuck Lever 	struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
248f531a5dbSChuck Lever 	struct rpc_xprt *xprt = rqst->rq_xprt;
249f531a5dbSChuck Lever 
250c8bbe0c7SChuck Lever 	dprintk("RPC:       %s: freeing rqst %p (req %p)\n",
251*7c8d9e7cSChuck Lever 		__func__, rqst, req);
252*7c8d9e7cSChuck Lever 
253*7c8d9e7cSChuck Lever 	rpcrdma_recv_buffer_put(req->rl_reply);
254*7c8d9e7cSChuck Lever 	req->rl_reply = NULL;
255c8bbe0c7SChuck Lever 
256f531a5dbSChuck Lever 	spin_lock_bh(&xprt->bc_pa_lock);
257f531a5dbSChuck Lever 	list_add_tail(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
258f531a5dbSChuck Lever 	spin_unlock_bh(&xprt->bc_pa_lock);
259f531a5dbSChuck Lever }
26063cae470SChuck Lever 
26163cae470SChuck Lever /**
26263cae470SChuck Lever  * rpcrdma_bc_receive_call - Handle a backward direction call
2639ab6d89eSChuck Lever  * @r_xprt: transport receiving the call
26463cae470SChuck Lever  * @rep: receive buffer containing the call
26563cae470SChuck Lever  *
26663cae470SChuck Lever  * Operational assumptions:
26763cae470SChuck Lever  *    o Backchannel credits are ignored, just as the NFS server
26863cae470SChuck Lever  *      forechannel currently does
26963cae470SChuck Lever  *    o The ULP manages a replay cache (eg, NFSv4.1 sessions).
27063cae470SChuck Lever  *      No replay detection is done at the transport level
27163cae470SChuck Lever  */
27263cae470SChuck Lever void rpcrdma_bc_receive_call(struct rpcrdma_xprt *r_xprt,
27363cae470SChuck Lever 			     struct rpcrdma_rep *rep)
27463cae470SChuck Lever {
27563cae470SChuck Lever 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
27663cae470SChuck Lever 	struct svc_serv *bc_serv;
27763cae470SChuck Lever 	struct rpcrdma_req *req;
27863cae470SChuck Lever 	struct rpc_rqst *rqst;
27963cae470SChuck Lever 	struct xdr_buf *buf;
28063cae470SChuck Lever 	size_t size;
28163cae470SChuck Lever 	__be32 *p;
28263cae470SChuck Lever 
28341c8f70fSChuck Lever 	p = xdr_inline_decode(&rep->rr_stream, 0);
28441c8f70fSChuck Lever 	size = xdr_stream_remaining(&rep->rr_stream);
28541c8f70fSChuck Lever 
28663cae470SChuck Lever #ifdef RPCRDMA_BACKCHANNEL_DEBUG
28763cae470SChuck Lever 	pr_info("RPC:       %s: callback XID %08x, length=%u\n",
28841c8f70fSChuck Lever 		__func__, be32_to_cpup(p), size);
28941c8f70fSChuck Lever 	pr_info("RPC:       %s: %*ph\n", __func__, size, p);
29063cae470SChuck Lever #endif
29163cae470SChuck Lever 
29263cae470SChuck Lever 	/* Grab a free bc rqst */
29363cae470SChuck Lever 	spin_lock(&xprt->bc_pa_lock);
29463cae470SChuck Lever 	if (list_empty(&xprt->bc_pa_list)) {
29563cae470SChuck Lever 		spin_unlock(&xprt->bc_pa_lock);
29663cae470SChuck Lever 		goto out_overflow;
29763cae470SChuck Lever 	}
29863cae470SChuck Lever 	rqst = list_first_entry(&xprt->bc_pa_list,
29963cae470SChuck Lever 				struct rpc_rqst, rq_bc_pa_list);
30063cae470SChuck Lever 	list_del(&rqst->rq_bc_pa_list);
30163cae470SChuck Lever 	spin_unlock(&xprt->bc_pa_lock);
30263cae470SChuck Lever 
30363cae470SChuck Lever 	/* Prepare rqst */
30463cae470SChuck Lever 	rqst->rq_reply_bytes_recvd = 0;
30563cae470SChuck Lever 	rqst->rq_bytes_sent = 0;
30641c8f70fSChuck Lever 	rqst->rq_xid = *p;
3079f74660bSChuck Lever 
3089f74660bSChuck Lever 	rqst->rq_private_buf.len = size;
30963cae470SChuck Lever 
31063cae470SChuck Lever 	buf = &rqst->rq_rcv_buf;
31163cae470SChuck Lever 	memset(buf, 0, sizeof(*buf));
31263cae470SChuck Lever 	buf->head[0].iov_base = p;
31363cae470SChuck Lever 	buf->head[0].iov_len = size;
31463cae470SChuck Lever 	buf->len = size;
31563cae470SChuck Lever 
31663cae470SChuck Lever 	/* The receive buffer has to be hooked to the rpcrdma_req
31741c8f70fSChuck Lever 	 * so that it is not released while the req is pointing
31841c8f70fSChuck Lever 	 * to its buffer, and so that it can be reposted after
31941c8f70fSChuck Lever 	 * the Upper Layer is done decoding it.
32063cae470SChuck Lever 	 */
32163cae470SChuck Lever 	req = rpcr_to_rdmar(rqst);
32263cae470SChuck Lever 	req->rl_reply = rep;
323fc1eb807SChuck Lever 	trace_xprtrdma_cb_call(rqst);
32463cae470SChuck Lever 
32563cae470SChuck Lever 	/* Queue rqst for ULP's callback service */
32663cae470SChuck Lever 	bc_serv = xprt->bc_serv;
32763cae470SChuck Lever 	spin_lock(&bc_serv->sv_cb_lock);
32863cae470SChuck Lever 	list_add(&rqst->rq_bc_list, &bc_serv->sv_cb_list);
32963cae470SChuck Lever 	spin_unlock(&bc_serv->sv_cb_lock);
33063cae470SChuck Lever 
33163cae470SChuck Lever 	wake_up(&bc_serv->sv_cb_waitq);
33263cae470SChuck Lever 
33363cae470SChuck Lever 	r_xprt->rx_stats.bcall_count++;
33463cae470SChuck Lever 	return;
33563cae470SChuck Lever 
33663cae470SChuck Lever out_overflow:
33763cae470SChuck Lever 	pr_warn("RPC/RDMA backchannel overflow\n");
33863cae470SChuck Lever 	xprt_disconnect_done(xprt);
33963cae470SChuck Lever 	/* This receive buffer gets reposted automatically
34063cae470SChuck Lever 	 * when the connection is re-established.
34163cae470SChuck Lever 	 */
34263cae470SChuck Lever 	return;
34363cae470SChuck Lever }
344