svc_rdma_backchannel.c 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281
  1. // SPDX-License-Identifier: GPL-2.0
  2. /*
  3. * Copyright (c) 2015-2018 Oracle. All rights reserved.
  4. *
  5. * Support for backward direction RPCs on RPC/RDMA (server-side).
  6. */
  7. #include <linux/sunrpc/svc_rdma.h>
  8. #include "xprt_rdma.h"
  9. #include <trace/events/rpcrdma.h>
  10. /**
  11. * svc_rdma_handle_bc_reply - Process incoming backchannel Reply
  12. * @rqstp: resources for handling the Reply
  13. * @rctxt: Received message
  14. *
  15. */
  16. void svc_rdma_handle_bc_reply(struct svc_rqst *rqstp,
  17. struct svc_rdma_recv_ctxt *rctxt)
  18. {
  19. struct svc_xprt *sxprt = rqstp->rq_xprt;
  20. struct rpc_xprt *xprt = sxprt->xpt_bc_xprt;
  21. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  22. struct xdr_buf *rcvbuf = &rqstp->rq_arg;
  23. struct kvec *dst, *src = &rcvbuf->head[0];
  24. __be32 *rdma_resp = rctxt->rc_recv_buf;
  25. struct rpc_rqst *req;
  26. u32 credits;
  27. spin_lock(&xprt->queue_lock);
  28. req = xprt_lookup_rqst(xprt, *rdma_resp);
  29. if (!req)
  30. goto out_unlock;
  31. dst = &req->rq_private_buf.head[0];
  32. memcpy(&req->rq_private_buf, &req->rq_rcv_buf, sizeof(struct xdr_buf));
  33. if (dst->iov_len < src->iov_len)
  34. goto out_unlock;
  35. memcpy(dst->iov_base, src->iov_base, src->iov_len);
  36. xprt_pin_rqst(req);
  37. spin_unlock(&xprt->queue_lock);
  38. credits = be32_to_cpup(rdma_resp + 2);
  39. if (credits == 0)
  40. credits = 1; /* don't deadlock */
  41. else if (credits > r_xprt->rx_buf.rb_bc_max_requests)
  42. credits = r_xprt->rx_buf.rb_bc_max_requests;
  43. spin_lock(&xprt->transport_lock);
  44. xprt->cwnd = credits << RPC_CWNDSHIFT;
  45. spin_unlock(&xprt->transport_lock);
  46. spin_lock(&xprt->queue_lock);
  47. xprt_complete_rqst(req->rq_task, rcvbuf->len);
  48. xprt_unpin_rqst(req);
  49. rcvbuf->len = 0;
  50. out_unlock:
  51. spin_unlock(&xprt->queue_lock);
  52. }
  53. /* Send a backwards direction RPC call.
  54. *
  55. * Caller holds the connection's mutex and has already marshaled
  56. * the RPC/RDMA request.
  57. *
  58. * This is similar to svc_rdma_send_reply_msg, but takes a struct
  59. * rpc_rqst instead, does not support chunks, and avoids blocking
  60. * memory allocation.
  61. *
  62. * XXX: There is still an opportunity to block in svc_rdma_send()
  63. * if there are no SQ entries to post the Send. This may occur if
  64. * the adapter has a small maximum SQ depth.
  65. */
  66. static int svc_rdma_bc_sendto(struct svcxprt_rdma *rdma,
  67. struct rpc_rqst *rqst,
  68. struct svc_rdma_send_ctxt *ctxt)
  69. {
  70. int ret;
  71. ret = svc_rdma_map_reply_msg(rdma, ctxt, NULL, &rqst->rq_snd_buf);
  72. if (ret < 0)
  73. return -EIO;
  74. /* Bump page refcnt so Send completion doesn't release
  75. * the rq_buffer before all retransmits are complete.
  76. */
  77. get_page(virt_to_page(rqst->rq_buffer));
  78. ctxt->sc_send_wr.opcode = IB_WR_SEND;
  79. return svc_rdma_send(rdma, ctxt);
  80. }
  81. /* Server-side transport endpoint wants a whole page for its send
  82. * buffer. The client RPC code constructs the RPC header in this
  83. * buffer before it invokes ->send_request.
  84. */
  85. static int
  86. xprt_rdma_bc_allocate(struct rpc_task *task)
  87. {
  88. struct rpc_rqst *rqst = task->tk_rqstp;
  89. size_t size = rqst->rq_callsize;
  90. struct page *page;
  91. if (size > PAGE_SIZE) {
  92. WARN_ONCE(1, "svcrdma: large bc buffer request (size %zu)\n",
  93. size);
  94. return -EINVAL;
  95. }
  96. page = alloc_page(RPCRDMA_DEF_GFP);
  97. if (!page)
  98. return -ENOMEM;
  99. rqst->rq_buffer = page_address(page);
  100. rqst->rq_rbuffer = kmalloc(rqst->rq_rcvsize, RPCRDMA_DEF_GFP);
  101. if (!rqst->rq_rbuffer) {
  102. put_page(page);
  103. return -ENOMEM;
  104. }
  105. return 0;
  106. }
  107. static void
  108. xprt_rdma_bc_free(struct rpc_task *task)
  109. {
  110. struct rpc_rqst *rqst = task->tk_rqstp;
  111. put_page(virt_to_page(rqst->rq_buffer));
  112. kfree(rqst->rq_rbuffer);
  113. }
  114. static int
  115. rpcrdma_bc_send_request(struct svcxprt_rdma *rdma, struct rpc_rqst *rqst)
  116. {
  117. struct rpc_xprt *xprt = rqst->rq_xprt;
  118. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  119. struct svc_rdma_send_ctxt *ctxt;
  120. __be32 *p;
  121. int rc;
  122. ctxt = svc_rdma_send_ctxt_get(rdma);
  123. if (!ctxt)
  124. goto drop_connection;
  125. p = xdr_reserve_space(&ctxt->sc_stream, RPCRDMA_HDRLEN_MIN);
  126. if (!p)
  127. goto put_ctxt;
  128. *p++ = rqst->rq_xid;
  129. *p++ = rpcrdma_version;
  130. *p++ = cpu_to_be32(r_xprt->rx_buf.rb_bc_max_requests);
  131. *p++ = rdma_msg;
  132. *p++ = xdr_zero;
  133. *p++ = xdr_zero;
  134. *p = xdr_zero;
  135. rqst->rq_xtime = ktime_get();
  136. rc = svc_rdma_bc_sendto(rdma, rqst, ctxt);
  137. if (rc)
  138. goto put_ctxt;
  139. return 0;
  140. put_ctxt:
  141. svc_rdma_send_ctxt_put(rdma, ctxt);
  142. drop_connection:
  143. return -ENOTCONN;
  144. }
  145. /**
  146. * xprt_rdma_bc_send_request - Send a reverse-direction Call
  147. * @rqst: rpc_rqst containing Call message to be sent
  148. *
  149. * Return values:
  150. * %0 if the message was sent successfully
  151. * %ENOTCONN if the message was not sent
  152. */
  153. static int xprt_rdma_bc_send_request(struct rpc_rqst *rqst)
  154. {
  155. struct svc_xprt *sxprt = rqst->rq_xprt->bc_xprt;
  156. struct svcxprt_rdma *rdma =
  157. container_of(sxprt, struct svcxprt_rdma, sc_xprt);
  158. int ret;
  159. if (test_bit(XPT_DEAD, &sxprt->xpt_flags))
  160. return -ENOTCONN;
  161. ret = rpcrdma_bc_send_request(rdma, rqst);
  162. if (ret == -ENOTCONN)
  163. svc_close_xprt(sxprt);
  164. return ret;
  165. }
  166. static void
  167. xprt_rdma_bc_close(struct rpc_xprt *xprt)
  168. {
  169. xprt_disconnect_done(xprt);
  170. xprt->cwnd = RPC_CWNDSHIFT;
  171. }
  172. static void
  173. xprt_rdma_bc_put(struct rpc_xprt *xprt)
  174. {
  175. xprt_rdma_free_addresses(xprt);
  176. xprt_free(xprt);
  177. }
  178. static const struct rpc_xprt_ops xprt_rdma_bc_procs = {
  179. .reserve_xprt = xprt_reserve_xprt_cong,
  180. .release_xprt = xprt_release_xprt_cong,
  181. .alloc_slot = xprt_alloc_slot,
  182. .free_slot = xprt_free_slot,
  183. .release_request = xprt_release_rqst_cong,
  184. .buf_alloc = xprt_rdma_bc_allocate,
  185. .buf_free = xprt_rdma_bc_free,
  186. .send_request = xprt_rdma_bc_send_request,
  187. .wait_for_reply_request = xprt_wait_for_reply_request_def,
  188. .close = xprt_rdma_bc_close,
  189. .destroy = xprt_rdma_bc_put,
  190. .print_stats = xprt_rdma_print_stats
  191. };
  192. static const struct rpc_timeout xprt_rdma_bc_timeout = {
  193. .to_initval = 60 * HZ,
  194. .to_maxval = 60 * HZ,
  195. };
  196. /* It shouldn't matter if the number of backchannel session slots
  197. * doesn't match the number of RPC/RDMA credits. That just means
  198. * one or the other will have extra slots that aren't used.
  199. */
  200. static struct rpc_xprt *
  201. xprt_setup_rdma_bc(struct xprt_create *args)
  202. {
  203. struct rpc_xprt *xprt;
  204. struct rpcrdma_xprt *new_xprt;
  205. if (args->addrlen > sizeof(xprt->addr))
  206. return ERR_PTR(-EBADF);
  207. xprt = xprt_alloc(args->net, sizeof(*new_xprt),
  208. RPCRDMA_MAX_BC_REQUESTS,
  209. RPCRDMA_MAX_BC_REQUESTS);
  210. if (!xprt)
  211. return ERR_PTR(-ENOMEM);
  212. xprt->timeout = &xprt_rdma_bc_timeout;
  213. xprt_set_bound(xprt);
  214. xprt_set_connected(xprt);
  215. xprt->bind_timeout = 0;
  216. xprt->reestablish_timeout = 0;
  217. xprt->idle_timeout = 0;
  218. xprt->prot = XPRT_TRANSPORT_BC_RDMA;
  219. xprt->ops = &xprt_rdma_bc_procs;
  220. memcpy(&xprt->addr, args->dstaddr, args->addrlen);
  221. xprt->addrlen = args->addrlen;
  222. xprt_rdma_format_addresses(xprt, (struct sockaddr *)&xprt->addr);
  223. xprt->resvport = 0;
  224. xprt->max_payload = xprt_rdma_max_inline_read;
  225. new_xprt = rpcx_to_rdmax(xprt);
  226. new_xprt->rx_buf.rb_bc_max_requests = xprt->max_reqs;
  227. xprt_get(xprt);
  228. args->bc_xprt->xpt_bc_xprt = xprt;
  229. xprt->bc_xprt = args->bc_xprt;
  230. /* Final put for backchannel xprt is in __svc_rdma_free */
  231. xprt_get(xprt);
  232. return xprt;
  233. }
  234. struct xprt_class xprt_rdma_bc = {
  235. .list = LIST_HEAD_INIT(xprt_rdma_bc.list),
  236. .name = "rdma backchannel",
  237. .owner = THIS_MODULE,
  238. .ident = XPRT_TRANSPORT_BC_RDMA,
  239. .setup = xprt_setup_rdma_bc,
  240. };