backchannel.c 7.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284
  1. // SPDX-License-Identifier: GPL-2.0
  2. /*
  3. * Copyright (c) 2015 Oracle. All rights reserved.
  4. *
  5. * Support for backward direction RPCs on RPC/RDMA.
  6. */
  7. #include <linux/sunrpc/xprt.h>
  8. #include <linux/sunrpc/svc.h>
  9. #include <linux/sunrpc/svc_xprt.h>
  10. #include <linux/sunrpc/svc_rdma.h>
  11. #include "xprt_rdma.h"
  12. #include <trace/events/rpcrdma.h>
  13. #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  14. # define RPCDBG_FACILITY RPCDBG_TRANS
  15. #endif
  16. #undef RPCRDMA_BACKCHANNEL_DEBUG
  17. /**
  18. * xprt_rdma_bc_setup - Pre-allocate resources for handling backchannel requests
  19. * @xprt: transport associated with these backchannel resources
  20. * @reqs: number of concurrent incoming requests to expect
  21. *
  22. * Returns 0 on success; otherwise a negative errno
  23. */
  24. int xprt_rdma_bc_setup(struct rpc_xprt *xprt, unsigned int reqs)
  25. {
  26. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  27. r_xprt->rx_buf.rb_bc_srv_max_requests = RPCRDMA_BACKWARD_WRS >> 1;
  28. trace_xprtrdma_cb_setup(r_xprt, reqs);
  29. return 0;
  30. }
  31. /**
  32. * xprt_rdma_bc_maxpayload - Return maximum backchannel message size
  33. * @xprt: transport
  34. *
  35. * Returns maximum size, in bytes, of a backchannel message
  36. */
  37. size_t xprt_rdma_bc_maxpayload(struct rpc_xprt *xprt)
  38. {
  39. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  40. struct rpcrdma_ep *ep = r_xprt->rx_ep;
  41. size_t maxmsg;
  42. maxmsg = min_t(unsigned int, ep->re_inline_send, ep->re_inline_recv);
  43. maxmsg = min_t(unsigned int, maxmsg, PAGE_SIZE);
  44. return maxmsg - RPCRDMA_HDRLEN_MIN;
  45. }
  46. unsigned int xprt_rdma_bc_max_slots(struct rpc_xprt *xprt)
  47. {
  48. return RPCRDMA_BACKWARD_WRS >> 1;
  49. }
  50. static int rpcrdma_bc_marshal_reply(struct rpc_rqst *rqst)
  51. {
  52. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt);
  53. struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
  54. __be32 *p;
  55. rpcrdma_set_xdrlen(&req->rl_hdrbuf, 0);
  56. xdr_init_encode(&req->rl_stream, &req->rl_hdrbuf,
  57. rdmab_data(req->rl_rdmabuf), rqst);
  58. p = xdr_reserve_space(&req->rl_stream, 28);
  59. if (unlikely(!p))
  60. return -EIO;
  61. *p++ = rqst->rq_xid;
  62. *p++ = rpcrdma_version;
  63. *p++ = cpu_to_be32(r_xprt->rx_buf.rb_bc_srv_max_requests);
  64. *p++ = rdma_msg;
  65. *p++ = xdr_zero;
  66. *p++ = xdr_zero;
  67. *p = xdr_zero;
  68. if (rpcrdma_prepare_send_sges(r_xprt, req, RPCRDMA_HDRLEN_MIN,
  69. &rqst->rq_snd_buf, rpcrdma_noch_pullup))
  70. return -EIO;
  71. trace_xprtrdma_cb_reply(rqst);
  72. return 0;
  73. }
  74. /**
  75. * xprt_rdma_bc_send_reply - marshal and send a backchannel reply
  76. * @rqst: RPC rqst with a backchannel RPC reply in rq_snd_buf
  77. *
  78. * Caller holds the transport's write lock.
  79. *
  80. * Returns:
  81. * %0 if the RPC message has been sent
  82. * %-ENOTCONN if the caller should reconnect and call again
  83. * %-EIO if a permanent error occurred and the request was not
  84. * sent. Do not try to send this message again.
  85. */
  86. int xprt_rdma_bc_send_reply(struct rpc_rqst *rqst)
  87. {
  88. struct rpc_xprt *xprt = rqst->rq_xprt;
  89. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  90. struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
  91. int rc;
  92. if (!xprt_connected(xprt))
  93. return -ENOTCONN;
  94. if (!xprt_request_get_cong(xprt, rqst))
  95. return -EBADSLT;
  96. rc = rpcrdma_bc_marshal_reply(rqst);
  97. if (rc < 0)
  98. goto failed_marshal;
  99. if (rpcrdma_post_sends(r_xprt, req))
  100. goto drop_connection;
  101. return 0;
  102. failed_marshal:
  103. if (rc != -ENOTCONN)
  104. return rc;
  105. drop_connection:
  106. xprt_rdma_close(xprt);
  107. return -ENOTCONN;
  108. }
  109. /**
  110. * xprt_rdma_bc_destroy - Release resources for handling backchannel requests
  111. * @xprt: transport associated with these backchannel resources
  112. * @reqs: number of incoming requests to destroy; ignored
  113. */
  114. void xprt_rdma_bc_destroy(struct rpc_xprt *xprt, unsigned int reqs)
  115. {
  116. struct rpc_rqst *rqst, *tmp;
  117. spin_lock(&xprt->bc_pa_lock);
  118. list_for_each_entry_safe(rqst, tmp, &xprt->bc_pa_list, rq_bc_pa_list) {
  119. list_del(&rqst->rq_bc_pa_list);
  120. spin_unlock(&xprt->bc_pa_lock);
  121. rpcrdma_req_destroy(rpcr_to_rdmar(rqst));
  122. spin_lock(&xprt->bc_pa_lock);
  123. }
  124. spin_unlock(&xprt->bc_pa_lock);
  125. }
  126. /**
  127. * xprt_rdma_bc_free_rqst - Release a backchannel rqst
  128. * @rqst: request to release
  129. */
  130. void xprt_rdma_bc_free_rqst(struct rpc_rqst *rqst)
  131. {
  132. struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
  133. struct rpc_xprt *xprt = rqst->rq_xprt;
  134. rpcrdma_recv_buffer_put(req->rl_reply);
  135. req->rl_reply = NULL;
  136. spin_lock(&xprt->bc_pa_lock);
  137. list_add_tail(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
  138. spin_unlock(&xprt->bc_pa_lock);
  139. xprt_put(xprt);
  140. }
  141. static struct rpc_rqst *rpcrdma_bc_rqst_get(struct rpcrdma_xprt *r_xprt)
  142. {
  143. struct rpc_xprt *xprt = &r_xprt->rx_xprt;
  144. struct rpcrdma_req *req;
  145. struct rpc_rqst *rqst;
  146. size_t size;
  147. spin_lock(&xprt->bc_pa_lock);
  148. rqst = list_first_entry_or_null(&xprt->bc_pa_list, struct rpc_rqst,
  149. rq_bc_pa_list);
  150. if (!rqst)
  151. goto create_req;
  152. list_del(&rqst->rq_bc_pa_list);
  153. spin_unlock(&xprt->bc_pa_lock);
  154. return rqst;
  155. create_req:
  156. spin_unlock(&xprt->bc_pa_lock);
  157. /* Set a limit to prevent a remote from overrunning our resources.
  158. */
  159. if (xprt->bc_alloc_count >= RPCRDMA_BACKWARD_WRS)
  160. return NULL;
  161. size = min_t(size_t, r_xprt->rx_ep->re_inline_recv, PAGE_SIZE);
  162. req = rpcrdma_req_create(r_xprt, size, GFP_KERNEL);
  163. if (!req)
  164. return NULL;
  165. if (rpcrdma_req_setup(r_xprt, req)) {
  166. rpcrdma_req_destroy(req);
  167. return NULL;
  168. }
  169. xprt->bc_alloc_count++;
  170. rqst = &req->rl_slot;
  171. rqst->rq_xprt = xprt;
  172. __set_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state);
  173. xdr_buf_init(&rqst->rq_snd_buf, rdmab_data(req->rl_sendbuf), size);
  174. return rqst;
  175. }
  176. /**
  177. * rpcrdma_bc_receive_call - Handle a backward direction call
  178. * @r_xprt: transport receiving the call
  179. * @rep: receive buffer containing the call
  180. *
  181. * Operational assumptions:
  182. * o Backchannel credits are ignored, just as the NFS server
  183. * forechannel currently does
  184. * o The ULP manages a replay cache (eg, NFSv4.1 sessions).
  185. * No replay detection is done at the transport level
  186. */
  187. void rpcrdma_bc_receive_call(struct rpcrdma_xprt *r_xprt,
  188. struct rpcrdma_rep *rep)
  189. {
  190. struct rpc_xprt *xprt = &r_xprt->rx_xprt;
  191. struct svc_serv *bc_serv;
  192. struct rpcrdma_req *req;
  193. struct rpc_rqst *rqst;
  194. struct xdr_buf *buf;
  195. size_t size;
  196. __be32 *p;
  197. p = xdr_inline_decode(&rep->rr_stream, 0);
  198. size = xdr_stream_remaining(&rep->rr_stream);
  199. #ifdef RPCRDMA_BACKCHANNEL_DEBUG
  200. pr_info("RPC: %s: callback XID %08x, length=%u\n",
  201. __func__, be32_to_cpup(p), size);
  202. pr_info("RPC: %s: %*ph\n", __func__, size, p);
  203. #endif
  204. rqst = rpcrdma_bc_rqst_get(r_xprt);
  205. if (!rqst)
  206. goto out_overflow;
  207. rqst->rq_reply_bytes_recvd = 0;
  208. rqst->rq_xid = *p;
  209. rqst->rq_private_buf.len = size;
  210. buf = &rqst->rq_rcv_buf;
  211. memset(buf, 0, sizeof(*buf));
  212. buf->head[0].iov_base = p;
  213. buf->head[0].iov_len = size;
  214. buf->len = size;
  215. /* The receive buffer has to be hooked to the rpcrdma_req
  216. * so that it is not released while the req is pointing
  217. * to its buffer, and so that it can be reposted after
  218. * the Upper Layer is done decoding it.
  219. */
  220. req = rpcr_to_rdmar(rqst);
  221. req->rl_reply = rep;
  222. trace_xprtrdma_cb_call(rqst);
  223. /* Queue rqst for ULP's callback service */
  224. bc_serv = xprt->bc_serv;
  225. xprt_get(xprt);
  226. spin_lock(&bc_serv->sv_cb_lock);
  227. list_add(&rqst->rq_bc_list, &bc_serv->sv_cb_list);
  228. spin_unlock(&bc_serv->sv_cb_lock);
  229. wake_up(&bc_serv->sv_cb_waitq);
  230. r_xprt->rx_stats.bcall_count++;
  231. return;
  232. out_overflow:
  233. pr_warn("RPC/RDMA backchannel overflow\n");
  234. xprt_force_disconnect(xprt);
  235. /* This receive buffer gets reposted automatically
  236. * when the connection is re-established.
  237. */
  238. return;
  239. }