svc_rdma_recvfrom.c 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881
  1. // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
  2. /*
  3. * Copyright (c) 2016-2018 Oracle. All rights reserved.
  4. * Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved.
  5. * Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved.
  6. *
  7. * This software is available to you under a choice of one of two
  8. * licenses. You may choose to be licensed under the terms of the GNU
  9. * General Public License (GPL) Version 2, available from the file
  10. * COPYING in the main directory of this source tree, or the BSD-type
  11. * license below:
  12. *
  13. * Redistribution and use in source and binary forms, with or without
  14. * modification, are permitted provided that the following conditions
  15. * are met:
  16. *
  17. * Redistributions of source code must retain the above copyright
  18. * notice, this list of conditions and the following disclaimer.
  19. *
  20. * Redistributions in binary form must reproduce the above
  21. * copyright notice, this list of conditions and the following
  22. * disclaimer in the documentation and/or other materials provided
  23. * with the distribution.
  24. *
  25. * Neither the name of the Network Appliance, Inc. nor the names of
  26. * its contributors may be used to endorse or promote products
  27. * derived from this software without specific prior written
  28. * permission.
  29. *
  30. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  31. * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  32. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  33. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  34. * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  35. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  36. * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  37. * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  38. * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  39. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  40. * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  41. *
  42. * Author: Tom Tucker <tom@opengridcomputing.com>
  43. */
  44. /* Operation
  45. *
  46. * The main entry point is svc_rdma_recvfrom. This is called from
  47. * svc_recv when the transport indicates there is incoming data to
  48. * be read. "Data Ready" is signaled when an RDMA Receive completes,
  49. * or when a set of RDMA Reads complete.
  50. *
  51. * An svc_rqst is passed in. This structure contains an array of
  52. * free pages (rq_pages) that will contain the incoming RPC message.
  53. *
  54. * Short messages are moved directly into svc_rqst::rq_arg, and
  55. * the RPC Call is ready to be processed by the Upper Layer.
  56. * svc_rdma_recvfrom returns the length of the RPC Call message,
  57. * completing the reception of the RPC Call.
  58. *
  59. * However, when an incoming message has Read chunks,
  60. * svc_rdma_recvfrom must post RDMA Reads to pull the RPC Call's
  61. * data payload from the client. svc_rdma_recvfrom sets up the
  62. * RDMA Reads using pages in svc_rqst::rq_pages, which are
  63. * transferred to an svc_rdma_recv_ctxt for the duration of the
  64. * I/O. svc_rdma_recvfrom then returns zero, since the RPC message
  65. * is still not yet ready.
  66. *
  67. * When the Read chunk payloads have become available on the
  68. * server, "Data Ready" is raised again, and svc_recv calls
  69. * svc_rdma_recvfrom again. This second call may use a different
  70. * svc_rqst than the first one, thus any information that needs
  71. * to be preserved across these two calls is kept in an
  72. * svc_rdma_recv_ctxt.
  73. *
  74. * The second call to svc_rdma_recvfrom performs final assembly
  75. * of the RPC Call message, using the RDMA Read sink pages kept in
  76. * the svc_rdma_recv_ctxt. The xdr_buf is copied from the
  77. * svc_rdma_recv_ctxt to the second svc_rqst. The second call returns
  78. * the length of the completed RPC Call message.
  79. *
  80. * Page Management
  81. *
  82. * Pages under I/O must be transferred from the first svc_rqst to an
  83. * svc_rdma_recv_ctxt before the first svc_rdma_recvfrom call returns.
  84. *
  85. * The first svc_rqst supplies pages for RDMA Reads. These are moved
  86. * from rqstp::rq_pages into ctxt::pages. The consumed elements of
  87. * the rq_pages array are set to NULL and refilled with the first
  88. * svc_rdma_recvfrom call returns.
  89. *
  90. * During the second svc_rdma_recvfrom call, RDMA Read sink pages
  91. * are transferred from the svc_rdma_recv_ctxt to the second svc_rqst
  92. * (see rdma_read_complete() below).
  93. */
  94. #include <linux/spinlock.h>
  95. #include <asm/unaligned.h>
  96. #include <rdma/ib_verbs.h>
  97. #include <rdma/rdma_cm.h>
  98. #include <linux/sunrpc/xdr.h>
  99. #include <linux/sunrpc/debug.h>
  100. #include <linux/sunrpc/rpc_rdma.h>
  101. #include <linux/sunrpc/svc_rdma.h>
  102. #include "xprt_rdma.h"
  103. #include <trace/events/rpcrdma.h>
  104. #define RPCDBG_FACILITY RPCDBG_SVCXPRT
  105. static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc);
  106. static inline struct svc_rdma_recv_ctxt *
  107. svc_rdma_next_recv_ctxt(struct list_head *list)
  108. {
  109. return list_first_entry_or_null(list, struct svc_rdma_recv_ctxt,
  110. rc_list);
  111. }
  112. static void svc_rdma_recv_cid_init(struct svcxprt_rdma *rdma,
  113. struct rpc_rdma_cid *cid)
  114. {
  115. cid->ci_queue_id = rdma->sc_rq_cq->res.id;
  116. cid->ci_completion_id = atomic_inc_return(&rdma->sc_completion_ids);
  117. }
  118. static struct svc_rdma_recv_ctxt *
  119. svc_rdma_recv_ctxt_alloc(struct svcxprt_rdma *rdma)
  120. {
  121. struct svc_rdma_recv_ctxt *ctxt;
  122. dma_addr_t addr;
  123. void *buffer;
  124. ctxt = kmalloc(sizeof(*ctxt), GFP_KERNEL);
  125. if (!ctxt)
  126. goto fail0;
  127. buffer = kmalloc(rdma->sc_max_req_size, GFP_KERNEL);
  128. if (!buffer)
  129. goto fail1;
  130. addr = ib_dma_map_single(rdma->sc_pd->device, buffer,
  131. rdma->sc_max_req_size, DMA_FROM_DEVICE);
  132. if (ib_dma_mapping_error(rdma->sc_pd->device, addr))
  133. goto fail2;
  134. svc_rdma_recv_cid_init(rdma, &ctxt->rc_cid);
  135. ctxt->rc_recv_wr.next = NULL;
  136. ctxt->rc_recv_wr.wr_cqe = &ctxt->rc_cqe;
  137. ctxt->rc_recv_wr.sg_list = &ctxt->rc_recv_sge;
  138. ctxt->rc_recv_wr.num_sge = 1;
  139. ctxt->rc_cqe.done = svc_rdma_wc_receive;
  140. ctxt->rc_recv_sge.addr = addr;
  141. ctxt->rc_recv_sge.length = rdma->sc_max_req_size;
  142. ctxt->rc_recv_sge.lkey = rdma->sc_pd->local_dma_lkey;
  143. ctxt->rc_recv_buf = buffer;
  144. ctxt->rc_temp = false;
  145. return ctxt;
  146. fail2:
  147. kfree(buffer);
  148. fail1:
  149. kfree(ctxt);
  150. fail0:
  151. return NULL;
  152. }
  153. static void svc_rdma_recv_ctxt_destroy(struct svcxprt_rdma *rdma,
  154. struct svc_rdma_recv_ctxt *ctxt)
  155. {
  156. ib_dma_unmap_single(rdma->sc_pd->device, ctxt->rc_recv_sge.addr,
  157. ctxt->rc_recv_sge.length, DMA_FROM_DEVICE);
  158. kfree(ctxt->rc_recv_buf);
  159. kfree(ctxt);
  160. }
  161. /**
  162. * svc_rdma_recv_ctxts_destroy - Release all recv_ctxt's for an xprt
  163. * @rdma: svcxprt_rdma being torn down
  164. *
  165. */
  166. void svc_rdma_recv_ctxts_destroy(struct svcxprt_rdma *rdma)
  167. {
  168. struct svc_rdma_recv_ctxt *ctxt;
  169. struct llist_node *node;
  170. while ((node = llist_del_first(&rdma->sc_recv_ctxts))) {
  171. ctxt = llist_entry(node, struct svc_rdma_recv_ctxt, rc_node);
  172. svc_rdma_recv_ctxt_destroy(rdma, ctxt);
  173. }
  174. }
  175. static struct svc_rdma_recv_ctxt *
  176. svc_rdma_recv_ctxt_get(struct svcxprt_rdma *rdma)
  177. {
  178. struct svc_rdma_recv_ctxt *ctxt;
  179. struct llist_node *node;
  180. node = llist_del_first(&rdma->sc_recv_ctxts);
  181. if (!node)
  182. goto out_empty;
  183. ctxt = llist_entry(node, struct svc_rdma_recv_ctxt, rc_node);
  184. out:
  185. ctxt->rc_page_count = 0;
  186. ctxt->rc_read_payload_length = 0;
  187. return ctxt;
  188. out_empty:
  189. ctxt = svc_rdma_recv_ctxt_alloc(rdma);
  190. if (!ctxt)
  191. return NULL;
  192. goto out;
  193. }
  194. /**
  195. * svc_rdma_recv_ctxt_put - Return recv_ctxt to free list
  196. * @rdma: controlling svcxprt_rdma
  197. * @ctxt: object to return to the free list
  198. *
  199. */
  200. void svc_rdma_recv_ctxt_put(struct svcxprt_rdma *rdma,
  201. struct svc_rdma_recv_ctxt *ctxt)
  202. {
  203. unsigned int i;
  204. for (i = 0; i < ctxt->rc_page_count; i++)
  205. put_page(ctxt->rc_pages[i]);
  206. if (!ctxt->rc_temp)
  207. llist_add(&ctxt->rc_node, &rdma->sc_recv_ctxts);
  208. else
  209. svc_rdma_recv_ctxt_destroy(rdma, ctxt);
  210. }
  211. /**
  212. * svc_rdma_release_rqst - Release transport-specific per-rqst resources
  213. * @rqstp: svc_rqst being released
  214. *
  215. * Ensure that the recv_ctxt is released whether or not a Reply
  216. * was sent. For example, the client could close the connection,
  217. * or svc_process could drop an RPC, before the Reply is sent.
  218. */
  219. void svc_rdma_release_rqst(struct svc_rqst *rqstp)
  220. {
  221. struct svc_rdma_recv_ctxt *ctxt = rqstp->rq_xprt_ctxt;
  222. struct svc_xprt *xprt = rqstp->rq_xprt;
  223. struct svcxprt_rdma *rdma =
  224. container_of(xprt, struct svcxprt_rdma, sc_xprt);
  225. rqstp->rq_xprt_ctxt = NULL;
  226. if (ctxt)
  227. svc_rdma_recv_ctxt_put(rdma, ctxt);
  228. }
  229. static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
  230. struct svc_rdma_recv_ctxt *ctxt)
  231. {
  232. int ret;
  233. trace_svcrdma_post_recv(ctxt);
  234. ret = ib_post_recv(rdma->sc_qp, &ctxt->rc_recv_wr, NULL);
  235. if (ret)
  236. goto err_post;
  237. return 0;
  238. err_post:
  239. trace_svcrdma_rq_post_err(rdma, ret);
  240. svc_rdma_recv_ctxt_put(rdma, ctxt);
  241. return ret;
  242. }
  243. static int svc_rdma_post_recv(struct svcxprt_rdma *rdma)
  244. {
  245. struct svc_rdma_recv_ctxt *ctxt;
  246. if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))
  247. return 0;
  248. ctxt = svc_rdma_recv_ctxt_get(rdma);
  249. if (!ctxt)
  250. return -ENOMEM;
  251. return __svc_rdma_post_recv(rdma, ctxt);
  252. }
  253. /**
  254. * svc_rdma_post_recvs - Post initial set of Recv WRs
  255. * @rdma: fresh svcxprt_rdma
  256. *
  257. * Returns true if successful, otherwise false.
  258. */
  259. bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma)
  260. {
  261. struct svc_rdma_recv_ctxt *ctxt;
  262. unsigned int i;
  263. int ret;
  264. for (i = 0; i < rdma->sc_max_requests; i++) {
  265. ctxt = svc_rdma_recv_ctxt_get(rdma);
  266. if (!ctxt)
  267. return false;
  268. ctxt->rc_temp = true;
  269. ret = __svc_rdma_post_recv(rdma, ctxt);
  270. if (ret)
  271. return false;
  272. }
  273. return true;
  274. }
  275. /**
  276. * svc_rdma_wc_receive - Invoked by RDMA provider for each polled Receive WC
  277. * @cq: Completion Queue context
  278. * @wc: Work Completion object
  279. *
  280. * NB: The svc_xprt/svcxprt_rdma is pinned whenever it's possible that
  281. * the Receive completion handler could be running.
  282. */
  283. static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
  284. {
  285. struct svcxprt_rdma *rdma = cq->cq_context;
  286. struct ib_cqe *cqe = wc->wr_cqe;
  287. struct svc_rdma_recv_ctxt *ctxt;
  288. /* WARNING: Only wc->wr_cqe and wc->status are reliable */
  289. ctxt = container_of(cqe, struct svc_rdma_recv_ctxt, rc_cqe);
  290. trace_svcrdma_wc_receive(wc, &ctxt->rc_cid);
  291. if (wc->status != IB_WC_SUCCESS)
  292. goto flushed;
  293. if (svc_rdma_post_recv(rdma))
  294. goto post_err;
  295. /* All wc fields are now known to be valid */
  296. ctxt->rc_byte_len = wc->byte_len;
  297. ib_dma_sync_single_for_cpu(rdma->sc_pd->device,
  298. ctxt->rc_recv_sge.addr,
  299. wc->byte_len, DMA_FROM_DEVICE);
  300. spin_lock(&rdma->sc_rq_dto_lock);
  301. list_add_tail(&ctxt->rc_list, &rdma->sc_rq_dto_q);
  302. /* Note the unlock pairs with the smp_rmb in svc_xprt_ready: */
  303. set_bit(XPT_DATA, &rdma->sc_xprt.xpt_flags);
  304. spin_unlock(&rdma->sc_rq_dto_lock);
  305. if (!test_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags))
  306. svc_xprt_enqueue(&rdma->sc_xprt);
  307. return;
  308. flushed:
  309. post_err:
  310. svc_rdma_recv_ctxt_put(rdma, ctxt);
  311. set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
  312. svc_xprt_enqueue(&rdma->sc_xprt);
  313. }
  314. /**
  315. * svc_rdma_flush_recv_queues - Drain pending Receive work
  316. * @rdma: svcxprt_rdma being shut down
  317. *
  318. */
  319. void svc_rdma_flush_recv_queues(struct svcxprt_rdma *rdma)
  320. {
  321. struct svc_rdma_recv_ctxt *ctxt;
  322. while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_read_complete_q))) {
  323. list_del(&ctxt->rc_list);
  324. svc_rdma_recv_ctxt_put(rdma, ctxt);
  325. }
  326. while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_rq_dto_q))) {
  327. list_del(&ctxt->rc_list);
  328. svc_rdma_recv_ctxt_put(rdma, ctxt);
  329. }
  330. }
  331. static void svc_rdma_build_arg_xdr(struct svc_rqst *rqstp,
  332. struct svc_rdma_recv_ctxt *ctxt)
  333. {
  334. struct xdr_buf *arg = &rqstp->rq_arg;
  335. arg->head[0].iov_base = ctxt->rc_recv_buf;
  336. arg->head[0].iov_len = ctxt->rc_byte_len;
  337. arg->tail[0].iov_base = NULL;
  338. arg->tail[0].iov_len = 0;
  339. arg->page_len = 0;
  340. arg->page_base = 0;
  341. arg->buflen = ctxt->rc_byte_len;
  342. arg->len = ctxt->rc_byte_len;
  343. }
  344. /* This accommodates the largest possible Write chunk.
  345. */
  346. #define MAX_BYTES_WRITE_CHUNK ((u32)(RPCSVC_MAXPAGES << PAGE_SHIFT))
  347. /* This accommodates the largest possible Position-Zero
  348. * Read chunk or Reply chunk.
  349. */
  350. #define MAX_BYTES_SPECIAL_CHUNK ((u32)((RPCSVC_MAXPAGES + 2) << PAGE_SHIFT))
  351. /* Sanity check the Read list.
  352. *
  353. * Implementation limits:
  354. * - This implementation supports only one Read chunk.
  355. *
  356. * Sanity checks:
  357. * - Read list does not overflow Receive buffer.
  358. * - Segment size limited by largest NFS data payload.
  359. *
  360. * The segment count is limited to how many segments can
  361. * fit in the transport header without overflowing the
  362. * buffer. That's about 40 Read segments for a 1KB inline
  363. * threshold.
  364. *
  365. * Return values:
  366. * %true: Read list is valid. @rctxt's xdr_stream is updated
  367. * to point to the first byte past the Read list.
  368. * %false: Read list is corrupt. @rctxt's xdr_stream is left
  369. * in an unknown state.
  370. */
  371. static bool xdr_check_read_list(struct svc_rdma_recv_ctxt *rctxt)
  372. {
  373. u32 position, len;
  374. bool first;
  375. __be32 *p;
  376. p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
  377. if (!p)
  378. return false;
  379. len = 0;
  380. first = true;
  381. while (xdr_item_is_present(p)) {
  382. p = xdr_inline_decode(&rctxt->rc_stream,
  383. rpcrdma_readseg_maxsz * sizeof(*p));
  384. if (!p)
  385. return false;
  386. if (first) {
  387. position = be32_to_cpup(p);
  388. first = false;
  389. } else if (be32_to_cpup(p) != position) {
  390. return false;
  391. }
  392. p += 2;
  393. len += be32_to_cpup(p);
  394. p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
  395. if (!p)
  396. return false;
  397. }
  398. return len <= MAX_BYTES_SPECIAL_CHUNK;
  399. }
  400. /* The segment count is limited to how many segments can
  401. * fit in the transport header without overflowing the
  402. * buffer. That's about 60 Write segments for a 1KB inline
  403. * threshold.
  404. */
  405. static bool xdr_check_write_chunk(struct svc_rdma_recv_ctxt *rctxt, u32 maxlen)
  406. {
  407. u32 i, segcount, total;
  408. __be32 *p;
  409. p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
  410. if (!p)
  411. return false;
  412. segcount = be32_to_cpup(p);
  413. total = 0;
  414. for (i = 0; i < segcount; i++) {
  415. u32 handle, length;
  416. u64 offset;
  417. p = xdr_inline_decode(&rctxt->rc_stream,
  418. rpcrdma_segment_maxsz * sizeof(*p));
  419. if (!p)
  420. return false;
  421. xdr_decode_rdma_segment(p, &handle, &length, &offset);
  422. trace_svcrdma_decode_wseg(handle, length, offset);
  423. total += length;
  424. }
  425. return total <= maxlen;
  426. }
  427. /* Sanity check the Write list.
  428. *
  429. * Implementation limits:
  430. * - This implementation currently supports only one Write chunk.
  431. *
  432. * Sanity checks:
  433. * - Write list does not overflow Receive buffer.
  434. * - Chunk size limited by largest NFS data payload.
  435. *
  436. * Return values:
  437. * %true: Write list is valid. @rctxt's xdr_stream is updated
  438. * to point to the first byte past the Write list.
  439. * %false: Write list is corrupt. @rctxt's xdr_stream is left
  440. * in an unknown state.
  441. */
  442. static bool xdr_check_write_list(struct svc_rdma_recv_ctxt *rctxt)
  443. {
  444. u32 chcount = 0;
  445. __be32 *p;
  446. p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
  447. if (!p)
  448. return false;
  449. rctxt->rc_write_list = p;
  450. while (xdr_item_is_present(p)) {
  451. if (!xdr_check_write_chunk(rctxt, MAX_BYTES_WRITE_CHUNK))
  452. return false;
  453. ++chcount;
  454. p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
  455. if (!p)
  456. return false;
  457. }
  458. if (!chcount)
  459. rctxt->rc_write_list = NULL;
  460. return chcount < 2;
  461. }
  462. /* Sanity check the Reply chunk.
  463. *
  464. * Sanity checks:
  465. * - Reply chunk does not overflow Receive buffer.
  466. * - Chunk size limited by largest NFS data payload.
  467. *
  468. * Return values:
  469. * %true: Reply chunk is valid. @rctxt's xdr_stream is updated
  470. * to point to the first byte past the Reply chunk.
  471. * %false: Reply chunk is corrupt. @rctxt's xdr_stream is left
  472. * in an unknown state.
  473. */
  474. static bool xdr_check_reply_chunk(struct svc_rdma_recv_ctxt *rctxt)
  475. {
  476. __be32 *p;
  477. p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
  478. if (!p)
  479. return false;
  480. rctxt->rc_reply_chunk = NULL;
  481. if (xdr_item_is_present(p)) {
  482. if (!xdr_check_write_chunk(rctxt, MAX_BYTES_SPECIAL_CHUNK))
  483. return false;
  484. rctxt->rc_reply_chunk = p;
  485. }
  486. return true;
  487. }
  488. /* RPC-over-RDMA Version One private extension: Remote Invalidation.
  489. * Responder's choice: requester signals it can handle Send With
  490. * Invalidate, and responder chooses one R_key to invalidate.
  491. *
  492. * If there is exactly one distinct R_key in the received transport
  493. * header, set rc_inv_rkey to that R_key. Otherwise, set it to zero.
  494. *
  495. * Perform this operation while the received transport header is
  496. * still in the CPU cache.
  497. */
  498. static void svc_rdma_get_inv_rkey(struct svcxprt_rdma *rdma,
  499. struct svc_rdma_recv_ctxt *ctxt)
  500. {
  501. __be32 inv_rkey, *p;
  502. u32 i, segcount;
  503. ctxt->rc_inv_rkey = 0;
  504. if (!rdma->sc_snd_w_inv)
  505. return;
  506. inv_rkey = xdr_zero;
  507. p = ctxt->rc_recv_buf;
  508. p += rpcrdma_fixed_maxsz;
  509. /* Read list */
  510. while (xdr_item_is_present(p++)) {
  511. p++; /* position */
  512. if (inv_rkey == xdr_zero)
  513. inv_rkey = *p;
  514. else if (inv_rkey != *p)
  515. return;
  516. p += 4;
  517. }
  518. /* Write list */
  519. while (xdr_item_is_present(p++)) {
  520. segcount = be32_to_cpup(p++);
  521. for (i = 0; i < segcount; i++) {
  522. if (inv_rkey == xdr_zero)
  523. inv_rkey = *p;
  524. else if (inv_rkey != *p)
  525. return;
  526. p += 4;
  527. }
  528. }
  529. /* Reply chunk */
  530. if (xdr_item_is_present(p++)) {
  531. segcount = be32_to_cpup(p++);
  532. for (i = 0; i < segcount; i++) {
  533. if (inv_rkey == xdr_zero)
  534. inv_rkey = *p;
  535. else if (inv_rkey != *p)
  536. return;
  537. p += 4;
  538. }
  539. }
  540. ctxt->rc_inv_rkey = be32_to_cpu(inv_rkey);
  541. }
  542. /**
  543. * svc_rdma_xdr_decode_req - Decode the transport header
  544. * @rq_arg: xdr_buf containing ingress RPC/RDMA message
  545. * @rctxt: state of decoding
  546. *
  547. * On entry, xdr->head[0].iov_base points to first byte of the
  548. * RPC-over-RDMA transport header.
  549. *
  550. * On successful exit, head[0] points to first byte past the
  551. * RPC-over-RDMA header. For RDMA_MSG, this is the RPC message.
  552. *
  553. * The length of the RPC-over-RDMA header is returned.
  554. *
  555. * Assumptions:
  556. * - The transport header is entirely contained in the head iovec.
  557. */
  558. static int svc_rdma_xdr_decode_req(struct xdr_buf *rq_arg,
  559. struct svc_rdma_recv_ctxt *rctxt)
  560. {
  561. __be32 *p, *rdma_argp;
  562. unsigned int hdr_len;
  563. rdma_argp = rq_arg->head[0].iov_base;
  564. xdr_init_decode(&rctxt->rc_stream, rq_arg, rdma_argp, NULL);
  565. p = xdr_inline_decode(&rctxt->rc_stream,
  566. rpcrdma_fixed_maxsz * sizeof(*p));
  567. if (unlikely(!p))
  568. goto out_short;
  569. p++;
  570. if (*p != rpcrdma_version)
  571. goto out_version;
  572. p += 2;
  573. switch (*p) {
  574. case rdma_msg:
  575. break;
  576. case rdma_nomsg:
  577. break;
  578. case rdma_done:
  579. goto out_drop;
  580. case rdma_error:
  581. goto out_drop;
  582. default:
  583. goto out_proc;
  584. }
  585. if (!xdr_check_read_list(rctxt))
  586. goto out_inval;
  587. if (!xdr_check_write_list(rctxt))
  588. goto out_inval;
  589. if (!xdr_check_reply_chunk(rctxt))
  590. goto out_inval;
  591. rq_arg->head[0].iov_base = rctxt->rc_stream.p;
  592. hdr_len = xdr_stream_pos(&rctxt->rc_stream);
  593. rq_arg->head[0].iov_len -= hdr_len;
  594. rq_arg->len -= hdr_len;
  595. trace_svcrdma_decode_rqst(rctxt, rdma_argp, hdr_len);
  596. return hdr_len;
  597. out_short:
  598. trace_svcrdma_decode_short_err(rctxt, rq_arg->len);
  599. return -EINVAL;
  600. out_version:
  601. trace_svcrdma_decode_badvers_err(rctxt, rdma_argp);
  602. return -EPROTONOSUPPORT;
  603. out_drop:
  604. trace_svcrdma_decode_drop_err(rctxt, rdma_argp);
  605. return 0;
  606. out_proc:
  607. trace_svcrdma_decode_badproc_err(rctxt, rdma_argp);
  608. return -EINVAL;
  609. out_inval:
  610. trace_svcrdma_decode_parse_err(rctxt, rdma_argp);
  611. return -EINVAL;
  612. }
  613. static void rdma_read_complete(struct svc_rqst *rqstp,
  614. struct svc_rdma_recv_ctxt *head)
  615. {
  616. int page_no;
  617. /* Move Read chunk pages to rqstp so that they will be released
  618. * when svc_process is done with them.
  619. */
  620. for (page_no = 0; page_no < head->rc_page_count; page_no++) {
  621. put_page(rqstp->rq_pages[page_no]);
  622. rqstp->rq_pages[page_no] = head->rc_pages[page_no];
  623. }
  624. head->rc_page_count = 0;
  625. /* Point rq_arg.pages past header */
  626. rqstp->rq_arg.pages = &rqstp->rq_pages[head->rc_hdr_count];
  627. rqstp->rq_arg.page_len = head->rc_arg.page_len;
  628. /* rq_respages starts after the last arg page */
  629. rqstp->rq_respages = &rqstp->rq_pages[page_no];
  630. rqstp->rq_next_page = rqstp->rq_respages + 1;
  631. /* Rebuild rq_arg head and tail. */
  632. rqstp->rq_arg.head[0] = head->rc_arg.head[0];
  633. rqstp->rq_arg.tail[0] = head->rc_arg.tail[0];
  634. rqstp->rq_arg.len = head->rc_arg.len;
  635. rqstp->rq_arg.buflen = head->rc_arg.buflen;
  636. }
  637. static void svc_rdma_send_error(struct svcxprt_rdma *rdma,
  638. struct svc_rdma_recv_ctxt *rctxt,
  639. int status)
  640. {
  641. struct svc_rdma_send_ctxt *sctxt;
  642. sctxt = svc_rdma_send_ctxt_get(rdma);
  643. if (!sctxt)
  644. return;
  645. svc_rdma_send_error_msg(rdma, sctxt, rctxt, status);
  646. }
  647. /* By convention, backchannel calls arrive via rdma_msg type
  648. * messages, and never populate the chunk lists. This makes
  649. * the RPC/RDMA header small and fixed in size, so it is
  650. * straightforward to check the RPC header's direction field.
  651. */
  652. static bool svc_rdma_is_backchannel_reply(struct svc_xprt *xprt,
  653. __be32 *rdma_resp)
  654. {
  655. __be32 *p;
  656. if (!xprt->xpt_bc_xprt)
  657. return false;
  658. p = rdma_resp + 3;
  659. if (*p++ != rdma_msg)
  660. return false;
  661. if (*p++ != xdr_zero)
  662. return false;
  663. if (*p++ != xdr_zero)
  664. return false;
  665. if (*p++ != xdr_zero)
  666. return false;
  667. /* XID sanity */
  668. if (*p++ != *rdma_resp)
  669. return false;
  670. /* call direction */
  671. if (*p == cpu_to_be32(RPC_CALL))
  672. return false;
  673. return true;
  674. }
  675. /**
  676. * svc_rdma_recvfrom - Receive an RPC call
  677. * @rqstp: request structure into which to receive an RPC Call
  678. *
  679. * Returns:
  680. * The positive number of bytes in the RPC Call message,
  681. * %0 if there were no Calls ready to return,
  682. * %-EINVAL if the Read chunk data is too large,
  683. * %-ENOMEM if rdma_rw context pool was exhausted,
  684. * %-ENOTCONN if posting failed (connection is lost),
  685. * %-EIO if rdma_rw initialization failed (DMA mapping, etc).
  686. *
  687. * Called in a loop when XPT_DATA is set. XPT_DATA is cleared only
  688. * when there are no remaining ctxt's to process.
  689. *
  690. * The next ctxt is removed from the "receive" lists.
  691. *
  692. * - If the ctxt completes a Read, then finish assembling the Call
  693. * message and return the number of bytes in the message.
  694. *
  695. * - If the ctxt completes a Receive, then construct the Call
  696. * message from the contents of the Receive buffer.
  697. *
  698. * - If there are no Read chunks in this message, then finish
  699. * assembling the Call message and return the number of bytes
  700. * in the message.
  701. *
  702. * - If there are Read chunks in this message, post Read WRs to
  703. * pull that payload and return 0.
  704. */
  705. int svc_rdma_recvfrom(struct svc_rqst *rqstp)
  706. {
  707. struct svc_xprt *xprt = rqstp->rq_xprt;
  708. struct svcxprt_rdma *rdma_xprt =
  709. container_of(xprt, struct svcxprt_rdma, sc_xprt);
  710. struct svc_rdma_recv_ctxt *ctxt;
  711. __be32 *p;
  712. int ret;
  713. rqstp->rq_xprt_ctxt = NULL;
  714. spin_lock(&rdma_xprt->sc_rq_dto_lock);
  715. ctxt = svc_rdma_next_recv_ctxt(&rdma_xprt->sc_read_complete_q);
  716. if (ctxt) {
  717. list_del(&ctxt->rc_list);
  718. spin_unlock(&rdma_xprt->sc_rq_dto_lock);
  719. rdma_read_complete(rqstp, ctxt);
  720. goto complete;
  721. }
  722. ctxt = svc_rdma_next_recv_ctxt(&rdma_xprt->sc_rq_dto_q);
  723. if (!ctxt) {
  724. /* No new incoming requests, terminate the loop */
  725. clear_bit(XPT_DATA, &xprt->xpt_flags);
  726. spin_unlock(&rdma_xprt->sc_rq_dto_lock);
  727. return 0;
  728. }
  729. list_del(&ctxt->rc_list);
  730. spin_unlock(&rdma_xprt->sc_rq_dto_lock);
  731. atomic_inc(&rdma_stat_recv);
  732. svc_rdma_build_arg_xdr(rqstp, ctxt);
  733. /* Prevent svc_xprt_release from releasing pages in rq_pages
  734. * if we return 0 or an error.
  735. */
  736. rqstp->rq_respages = rqstp->rq_pages;
  737. rqstp->rq_next_page = rqstp->rq_respages;
  738. p = (__be32 *)rqstp->rq_arg.head[0].iov_base;
  739. ret = svc_rdma_xdr_decode_req(&rqstp->rq_arg, ctxt);
  740. if (ret < 0)
  741. goto out_err;
  742. if (ret == 0)
  743. goto out_drop;
  744. rqstp->rq_xprt_hlen = ret;
  745. if (svc_rdma_is_backchannel_reply(xprt, p))
  746. goto out_backchannel;
  747. svc_rdma_get_inv_rkey(rdma_xprt, ctxt);
  748. p += rpcrdma_fixed_maxsz;
  749. if (*p != xdr_zero)
  750. goto out_readchunk;
  751. complete:
  752. rqstp->rq_xprt_ctxt = ctxt;
  753. rqstp->rq_prot = IPPROTO_MAX;
  754. svc_xprt_copy_addrs(rqstp, xprt);
  755. return rqstp->rq_arg.len;
  756. out_readchunk:
  757. ret = svc_rdma_recv_read_chunk(rdma_xprt, rqstp, ctxt, p);
  758. if (ret < 0)
  759. goto out_postfail;
  760. return 0;
  761. out_err:
  762. svc_rdma_send_error(rdma_xprt, ctxt, ret);
  763. svc_rdma_recv_ctxt_put(rdma_xprt, ctxt);
  764. return 0;
  765. out_postfail:
  766. if (ret == -EINVAL)
  767. svc_rdma_send_error(rdma_xprt, ctxt, ret);
  768. svc_rdma_recv_ctxt_put(rdma_xprt, ctxt);
  769. return ret;
  770. out_backchannel:
  771. svc_rdma_handle_bc_reply(rqstp, ctxt);
  772. out_drop:
  773. svc_rdma_recv_ctxt_put(rdma_xprt, ctxt);
  774. return 0;
  775. }