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 45 /* Operation 46 * 47 * The main entry point is svc_rdma_recvfrom. This is called from 48 * svc_recv when the transport indicates there is incoming data to 49 * be read. "Data Ready" is signaled when an RDMA Receive completes, 50 * or when a set of RDMA Reads complete. 51 * 52 * An svc_rqst is passed in. This structure contains an array of 53 * free pages (rq_pages) that will contain the incoming RPC message. 54 * 55 * Short messages are moved directly into svc_rqst::rq_arg, and 56 * the RPC Call is ready to be processed by the Upper Layer. 57 * svc_rdma_recvfrom returns the length of the RPC Call message, 58 * completing the reception of the RPC Call. 59 * 60 * However, when an incoming message has Read chunks, 61 * svc_rdma_recvfrom must post RDMA Reads to pull the RPC Call's 62 * data payload from the client. svc_rdma_recvfrom sets up the 63 * RDMA Reads using pages in svc_rqst::rq_pages, which are 64 * transferred to an svc_rdma_recv_ctxt for the duration of the 65 * I/O. svc_rdma_recvfrom then returns zero, since the RPC message 66 * is still not yet ready. 67 * 68 * When the Read chunk payloads have become available on the 69 * server, "Data Ready" is raised again, and svc_recv calls 70 * svc_rdma_recvfrom again. This second call may use a different 71 * svc_rqst than the first one, thus any information that needs 72 * to be preserved across these two calls is kept in an 73 * svc_rdma_recv_ctxt. 74 * 75 * The second call to svc_rdma_recvfrom performs final assembly 76 * of the RPC Call message, using the RDMA Read sink pages kept in 77 * the svc_rdma_recv_ctxt. The xdr_buf is copied from the 78 * svc_rdma_recv_ctxt to the second svc_rqst. The second call returns 79 * the length of the completed RPC Call message. 80 * 81 * Page Management 82 * 83 * Pages under I/O must be transferred from the first svc_rqst to an 84 * svc_rdma_recv_ctxt before the first svc_rdma_recvfrom call returns. 85 * 86 * The first svc_rqst supplies pages for RDMA Reads. These are moved 87 * from rqstp::rq_pages into ctxt::pages. The consumed elements of 88 * the rq_pages array are set to NULL and refilled with the first 89 * svc_rdma_recvfrom call returns. 90 * 91 * During the second svc_rdma_recvfrom call, RDMA Read sink pages 92 * are transferred from the svc_rdma_recv_ctxt to the second svc_rqst 93 * (see rdma_read_complete() below). 94 */ 95 96 #include <linux/slab.h> 97 #include <linux/spinlock.h> 98 #include <asm/unaligned.h> 99 #include <rdma/ib_verbs.h> 100 #include <rdma/rdma_cm.h> 101 102 #include <linux/sunrpc/xdr.h> 103 #include <linux/sunrpc/debug.h> 104 #include <linux/sunrpc/rpc_rdma.h> 105 #include <linux/sunrpc/svc_rdma.h> 106 107 #include "xprt_rdma.h" 108 #include <trace/events/rpcrdma.h> 109 110 #define RPCDBG_FACILITY RPCDBG_SVCXPRT 111 112 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc); 113 114 static inline struct svc_rdma_recv_ctxt * 115 svc_rdma_next_recv_ctxt(struct list_head *list) 116 { 117 return list_first_entry_or_null(list, struct svc_rdma_recv_ctxt, 118 rc_list); 119 } 120 121 static void svc_rdma_recv_cid_init(struct svcxprt_rdma *rdma, 122 struct rpc_rdma_cid *cid) 123 { 124 cid->ci_queue_id = rdma->sc_rq_cq->res.id; 125 cid->ci_completion_id = atomic_inc_return(&rdma->sc_completion_ids); 126 } 127 128 static struct svc_rdma_recv_ctxt * 129 svc_rdma_recv_ctxt_alloc(struct svcxprt_rdma *rdma) 130 { 131 struct svc_rdma_recv_ctxt *ctxt; 132 dma_addr_t addr; 133 void *buffer; 134 135 ctxt = kmalloc(sizeof(*ctxt), GFP_KERNEL); 136 if (!ctxt) 137 goto fail0; 138 buffer = kmalloc(rdma->sc_max_req_size, GFP_KERNEL); 139 if (!buffer) 140 goto fail1; 141 addr = ib_dma_map_single(rdma->sc_pd->device, buffer, 142 rdma->sc_max_req_size, DMA_FROM_DEVICE); 143 if (ib_dma_mapping_error(rdma->sc_pd->device, addr)) 144 goto fail2; 145 146 svc_rdma_recv_cid_init(rdma, &ctxt->rc_cid); 147 pcl_init(&ctxt->rc_call_pcl); 148 pcl_init(&ctxt->rc_read_pcl); 149 pcl_init(&ctxt->rc_write_pcl); 150 pcl_init(&ctxt->rc_reply_pcl); 151 152 ctxt->rc_recv_wr.next = NULL; 153 ctxt->rc_recv_wr.wr_cqe = &ctxt->rc_cqe; 154 ctxt->rc_recv_wr.sg_list = &ctxt->rc_recv_sge; 155 ctxt->rc_recv_wr.num_sge = 1; 156 ctxt->rc_cqe.done = svc_rdma_wc_receive; 157 ctxt->rc_recv_sge.addr = addr; 158 ctxt->rc_recv_sge.length = rdma->sc_max_req_size; 159 ctxt->rc_recv_sge.lkey = rdma->sc_pd->local_dma_lkey; 160 ctxt->rc_recv_buf = buffer; 161 ctxt->rc_temp = false; 162 return ctxt; 163 164 fail2: 165 kfree(buffer); 166 fail1: 167 kfree(ctxt); 168 fail0: 169 return NULL; 170 } 171 172 static void svc_rdma_recv_ctxt_destroy(struct svcxprt_rdma *rdma, 173 struct svc_rdma_recv_ctxt *ctxt) 174 { 175 ib_dma_unmap_single(rdma->sc_pd->device, ctxt->rc_recv_sge.addr, 176 ctxt->rc_recv_sge.length, DMA_FROM_DEVICE); 177 kfree(ctxt->rc_recv_buf); 178 kfree(ctxt); 179 } 180 181 /** 182 * svc_rdma_recv_ctxts_destroy - Release all recv_ctxt's for an xprt 183 * @rdma: svcxprt_rdma being torn down 184 * 185 */ 186 void svc_rdma_recv_ctxts_destroy(struct svcxprt_rdma *rdma) 187 { 188 struct svc_rdma_recv_ctxt *ctxt; 189 struct llist_node *node; 190 191 while ((node = llist_del_first(&rdma->sc_recv_ctxts))) { 192 ctxt = llist_entry(node, struct svc_rdma_recv_ctxt, rc_node); 193 svc_rdma_recv_ctxt_destroy(rdma, ctxt); 194 } 195 } 196 197 /** 198 * svc_rdma_recv_ctxt_get - Allocate a recv_ctxt 199 * @rdma: controlling svcxprt_rdma 200 * 201 * Returns a recv_ctxt or (rarely) NULL if none are available. 202 */ 203 struct svc_rdma_recv_ctxt *svc_rdma_recv_ctxt_get(struct svcxprt_rdma *rdma) 204 { 205 struct svc_rdma_recv_ctxt *ctxt; 206 struct llist_node *node; 207 208 node = llist_del_first(&rdma->sc_recv_ctxts); 209 if (!node) 210 goto out_empty; 211 ctxt = llist_entry(node, struct svc_rdma_recv_ctxt, rc_node); 212 213 out: 214 ctxt->rc_page_count = 0; 215 return ctxt; 216 217 out_empty: 218 ctxt = svc_rdma_recv_ctxt_alloc(rdma); 219 if (!ctxt) 220 return NULL; 221 goto out; 222 } 223 224 /** 225 * svc_rdma_recv_ctxt_put - Return recv_ctxt to free list 226 * @rdma: controlling svcxprt_rdma 227 * @ctxt: object to return to the free list 228 * 229 */ 230 void svc_rdma_recv_ctxt_put(struct svcxprt_rdma *rdma, 231 struct svc_rdma_recv_ctxt *ctxt) 232 { 233 unsigned int i; 234 235 for (i = 0; i < ctxt->rc_page_count; i++) 236 put_page(ctxt->rc_pages[i]); 237 238 pcl_free(&ctxt->rc_call_pcl); 239 pcl_free(&ctxt->rc_read_pcl); 240 pcl_free(&ctxt->rc_write_pcl); 241 pcl_free(&ctxt->rc_reply_pcl); 242 243 if (!ctxt->rc_temp) 244 llist_add(&ctxt->rc_node, &rdma->sc_recv_ctxts); 245 else 246 svc_rdma_recv_ctxt_destroy(rdma, ctxt); 247 } 248 249 /** 250 * svc_rdma_release_rqst - Release transport-specific per-rqst resources 251 * @rqstp: svc_rqst being released 252 * 253 * Ensure that the recv_ctxt is released whether or not a Reply 254 * was sent. For example, the client could close the connection, 255 * or svc_process could drop an RPC, before the Reply is sent. 256 */ 257 void svc_rdma_release_rqst(struct svc_rqst *rqstp) 258 { 259 struct svc_rdma_recv_ctxt *ctxt = rqstp->rq_xprt_ctxt; 260 struct svc_xprt *xprt = rqstp->rq_xprt; 261 struct svcxprt_rdma *rdma = 262 container_of(xprt, struct svcxprt_rdma, sc_xprt); 263 264 rqstp->rq_xprt_ctxt = NULL; 265 if (ctxt) 266 svc_rdma_recv_ctxt_put(rdma, ctxt); 267 } 268 269 static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma, 270 struct svc_rdma_recv_ctxt *ctxt) 271 { 272 int ret; 273 274 trace_svcrdma_post_recv(ctxt); 275 ret = ib_post_recv(rdma->sc_qp, &ctxt->rc_recv_wr, NULL); 276 if (ret) 277 goto err_post; 278 return 0; 279 280 err_post: 281 trace_svcrdma_rq_post_err(rdma, ret); 282 svc_rdma_recv_ctxt_put(rdma, ctxt); 283 return ret; 284 } 285 286 static int svc_rdma_post_recv(struct svcxprt_rdma *rdma) 287 { 288 struct svc_rdma_recv_ctxt *ctxt; 289 290 if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags)) 291 return 0; 292 ctxt = svc_rdma_recv_ctxt_get(rdma); 293 if (!ctxt) 294 return -ENOMEM; 295 return __svc_rdma_post_recv(rdma, ctxt); 296 } 297 298 /** 299 * svc_rdma_post_recvs - Post initial set of Recv WRs 300 * @rdma: fresh svcxprt_rdma 301 * 302 * Returns true if successful, otherwise false. 303 */ 304 bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma) 305 { 306 struct svc_rdma_recv_ctxt *ctxt; 307 unsigned int i; 308 int ret; 309 310 for (i = 0; i < rdma->sc_max_requests; i++) { 311 ctxt = svc_rdma_recv_ctxt_get(rdma); 312 if (!ctxt) 313 return false; 314 ctxt->rc_temp = true; 315 ret = __svc_rdma_post_recv(rdma, ctxt); 316 if (ret) 317 return false; 318 } 319 return true; 320 } 321 322 /** 323 * svc_rdma_wc_receive - Invoked by RDMA provider for each polled Receive WC 324 * @cq: Completion Queue context 325 * @wc: Work Completion object 326 * 327 * NB: The svc_xprt/svcxprt_rdma is pinned whenever it's possible that 328 * the Receive completion handler could be running. 329 */ 330 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc) 331 { 332 struct svcxprt_rdma *rdma = cq->cq_context; 333 struct ib_cqe *cqe = wc->wr_cqe; 334 struct svc_rdma_recv_ctxt *ctxt; 335 336 /* WARNING: Only wc->wr_cqe and wc->status are reliable */ 337 ctxt = container_of(cqe, struct svc_rdma_recv_ctxt, rc_cqe); 338 339 trace_svcrdma_wc_receive(wc, &ctxt->rc_cid); 340 if (wc->status != IB_WC_SUCCESS) 341 goto flushed; 342 343 if (svc_rdma_post_recv(rdma)) 344 goto post_err; 345 346 /* All wc fields are now known to be valid */ 347 ctxt->rc_byte_len = wc->byte_len; 348 349 spin_lock(&rdma->sc_rq_dto_lock); 350 list_add_tail(&ctxt->rc_list, &rdma->sc_rq_dto_q); 351 /* Note the unlock pairs with the smp_rmb in svc_xprt_ready: */ 352 set_bit(XPT_DATA, &rdma->sc_xprt.xpt_flags); 353 spin_unlock(&rdma->sc_rq_dto_lock); 354 if (!test_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags)) 355 svc_xprt_enqueue(&rdma->sc_xprt); 356 return; 357 358 flushed: 359 post_err: 360 svc_rdma_recv_ctxt_put(rdma, ctxt); 361 set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags); 362 svc_xprt_enqueue(&rdma->sc_xprt); 363 } 364 365 /** 366 * svc_rdma_flush_recv_queues - Drain pending Receive work 367 * @rdma: svcxprt_rdma being shut down 368 * 369 */ 370 void svc_rdma_flush_recv_queues(struct svcxprt_rdma *rdma) 371 { 372 struct svc_rdma_recv_ctxt *ctxt; 373 374 while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_read_complete_q))) { 375 list_del(&ctxt->rc_list); 376 svc_rdma_recv_ctxt_put(rdma, ctxt); 377 } 378 while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_rq_dto_q))) { 379 list_del(&ctxt->rc_list); 380 svc_rdma_recv_ctxt_put(rdma, ctxt); 381 } 382 } 383 384 static void svc_rdma_build_arg_xdr(struct svc_rqst *rqstp, 385 struct svc_rdma_recv_ctxt *ctxt) 386 { 387 struct xdr_buf *arg = &rqstp->rq_arg; 388 389 arg->head[0].iov_base = ctxt->rc_recv_buf; 390 arg->head[0].iov_len = ctxt->rc_byte_len; 391 arg->tail[0].iov_base = NULL; 392 arg->tail[0].iov_len = 0; 393 arg->page_len = 0; 394 arg->page_base = 0; 395 arg->buflen = ctxt->rc_byte_len; 396 arg->len = ctxt->rc_byte_len; 397 } 398 399 /** 400 * xdr_count_read_segments - Count number of Read segments in Read list 401 * @rctxt: Ingress receive context 402 * @p: Start of an un-decoded Read list 403 * 404 * Before allocating anything, ensure the ingress Read list is safe 405 * to use. 406 * 407 * The segment count is limited to how many segments can fit in the 408 * transport header without overflowing the buffer. That's about 40 409 * Read segments for a 1KB inline threshold. 410 * 411 * Return values: 412 * %true: Read list is valid. @rctxt's xdr_stream is updated to point 413 * to the first byte past the Read list. rc_read_pcl and 414 * rc_call_pcl cl_count fields are set to the number of 415 * Read segments in the list. 416 * %false: Read list is corrupt. @rctxt's xdr_stream is left in an 417 * unknown state. 418 */ 419 static bool xdr_count_read_segments(struct svc_rdma_recv_ctxt *rctxt, __be32 *p) 420 { 421 rctxt->rc_call_pcl.cl_count = 0; 422 rctxt->rc_read_pcl.cl_count = 0; 423 while (xdr_item_is_present(p)) { 424 u32 position, handle, length; 425 u64 offset; 426 427 p = xdr_inline_decode(&rctxt->rc_stream, 428 rpcrdma_readseg_maxsz * sizeof(*p)); 429 if (!p) 430 return false; 431 432 xdr_decode_read_segment(p, &position, &handle, 433 &length, &offset); 434 if (position) { 435 if (position & 3) 436 return false; 437 ++rctxt->rc_read_pcl.cl_count; 438 } else { 439 ++rctxt->rc_call_pcl.cl_count; 440 } 441 442 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p)); 443 if (!p) 444 return false; 445 } 446 return true; 447 } 448 449 /* Sanity check the Read list. 450 * 451 * Sanity checks: 452 * - Read list does not overflow Receive buffer. 453 * - Chunk size limited by largest NFS data payload. 454 * 455 * Return values: 456 * %true: Read list is valid. @rctxt's xdr_stream is updated 457 * to point to the first byte past the Read list. 458 * %false: Read list is corrupt. @rctxt's xdr_stream is left 459 * in an unknown state. 460 */ 461 static bool xdr_check_read_list(struct svc_rdma_recv_ctxt *rctxt) 462 { 463 __be32 *p; 464 465 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p)); 466 if (!p) 467 return false; 468 if (!xdr_count_read_segments(rctxt, p)) 469 return false; 470 if (!pcl_alloc_call(rctxt, p)) 471 return false; 472 return pcl_alloc_read(rctxt, p); 473 } 474 475 static bool xdr_check_write_chunk(struct svc_rdma_recv_ctxt *rctxt) 476 { 477 u32 segcount; 478 __be32 *p; 479 480 if (xdr_stream_decode_u32(&rctxt->rc_stream, &segcount)) 481 return false; 482 483 /* A bogus segcount causes this buffer overflow check to fail. */ 484 p = xdr_inline_decode(&rctxt->rc_stream, 485 segcount * rpcrdma_segment_maxsz * sizeof(*p)); 486 return p != NULL; 487 } 488 489 /** 490 * xdr_count_write_chunks - Count number of Write chunks in Write list 491 * @rctxt: Received header and decoding state 492 * @p: start of an un-decoded Write list 493 * 494 * Before allocating anything, ensure the ingress Write list is 495 * safe to use. 496 * 497 * Return values: 498 * %true: Write list is valid. @rctxt's xdr_stream is updated 499 * to point to the first byte past the Write list, and 500 * the number of Write chunks is in rc_write_pcl.cl_count. 501 * %false: Write list is corrupt. @rctxt's xdr_stream is left 502 * in an indeterminate state. 503 */ 504 static bool xdr_count_write_chunks(struct svc_rdma_recv_ctxt *rctxt, __be32 *p) 505 { 506 rctxt->rc_write_pcl.cl_count = 0; 507 while (xdr_item_is_present(p)) { 508 if (!xdr_check_write_chunk(rctxt)) 509 return false; 510 ++rctxt->rc_write_pcl.cl_count; 511 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p)); 512 if (!p) 513 return false; 514 } 515 return true; 516 } 517 518 /* Sanity check the Write list. 519 * 520 * Implementation limits: 521 * - This implementation currently supports only one Write chunk. 522 * 523 * Sanity checks: 524 * - Write list does not overflow Receive buffer. 525 * - Chunk size limited by largest NFS data payload. 526 * 527 * Return values: 528 * %true: Write list is valid. @rctxt's xdr_stream is updated 529 * to point to the first byte past the Write list. 530 * %false: Write list is corrupt. @rctxt's xdr_stream is left 531 * in an unknown state. 532 */ 533 static bool xdr_check_write_list(struct svc_rdma_recv_ctxt *rctxt) 534 { 535 __be32 *p; 536 537 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p)); 538 if (!p) 539 return false; 540 if (!xdr_count_write_chunks(rctxt, p)) 541 return false; 542 if (!pcl_alloc_write(rctxt, &rctxt->rc_write_pcl, p)) 543 return false; 544 545 rctxt->rc_cur_result_payload = pcl_first_chunk(&rctxt->rc_write_pcl); 546 return true; 547 } 548 549 /* Sanity check the Reply chunk. 550 * 551 * Sanity checks: 552 * - Reply chunk does not overflow Receive buffer. 553 * - Chunk size limited by largest NFS data payload. 554 * 555 * Return values: 556 * %true: Reply chunk is valid. @rctxt's xdr_stream is updated 557 * to point to the first byte past the Reply chunk. 558 * %false: Reply chunk is corrupt. @rctxt's xdr_stream is left 559 * in an unknown state. 560 */ 561 static bool xdr_check_reply_chunk(struct svc_rdma_recv_ctxt *rctxt) 562 { 563 __be32 *p; 564 565 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p)); 566 if (!p) 567 return false; 568 569 if (!xdr_item_is_present(p)) 570 return true; 571 if (!xdr_check_write_chunk(rctxt)) 572 return false; 573 574 rctxt->rc_reply_pcl.cl_count = 1; 575 return pcl_alloc_write(rctxt, &rctxt->rc_reply_pcl, p); 576 } 577 578 /* RPC-over-RDMA Version One private extension: Remote Invalidation. 579 * Responder's choice: requester signals it can handle Send With 580 * Invalidate, and responder chooses one R_key to invalidate. 581 * 582 * If there is exactly one distinct R_key in the received transport 583 * header, set rc_inv_rkey to that R_key. Otherwise, set it to zero. 584 */ 585 static void svc_rdma_get_inv_rkey(struct svcxprt_rdma *rdma, 586 struct svc_rdma_recv_ctxt *ctxt) 587 { 588 struct svc_rdma_segment *segment; 589 struct svc_rdma_chunk *chunk; 590 u32 inv_rkey; 591 592 ctxt->rc_inv_rkey = 0; 593 594 if (!rdma->sc_snd_w_inv) 595 return; 596 597 inv_rkey = 0; 598 pcl_for_each_chunk(chunk, &ctxt->rc_call_pcl) { 599 pcl_for_each_segment(segment, chunk) { 600 if (inv_rkey == 0) 601 inv_rkey = segment->rs_handle; 602 else if (inv_rkey != segment->rs_handle) 603 return; 604 } 605 } 606 pcl_for_each_chunk(chunk, &ctxt->rc_read_pcl) { 607 pcl_for_each_segment(segment, chunk) { 608 if (inv_rkey == 0) 609 inv_rkey = segment->rs_handle; 610 else if (inv_rkey != segment->rs_handle) 611 return; 612 } 613 } 614 pcl_for_each_chunk(chunk, &ctxt->rc_write_pcl) { 615 pcl_for_each_segment(segment, chunk) { 616 if (inv_rkey == 0) 617 inv_rkey = segment->rs_handle; 618 else if (inv_rkey != segment->rs_handle) 619 return; 620 } 621 } 622 pcl_for_each_chunk(chunk, &ctxt->rc_reply_pcl) { 623 pcl_for_each_segment(segment, chunk) { 624 if (inv_rkey == 0) 625 inv_rkey = segment->rs_handle; 626 else if (inv_rkey != segment->rs_handle) 627 return; 628 } 629 } 630 ctxt->rc_inv_rkey = inv_rkey; 631 } 632 633 /** 634 * svc_rdma_xdr_decode_req - Decode the transport header 635 * @rq_arg: xdr_buf containing ingress RPC/RDMA message 636 * @rctxt: state of decoding 637 * 638 * On entry, xdr->head[0].iov_base points to first byte of the 639 * RPC-over-RDMA transport header. 640 * 641 * On successful exit, head[0] points to first byte past the 642 * RPC-over-RDMA header. For RDMA_MSG, this is the RPC message. 643 * 644 * The length of the RPC-over-RDMA header is returned. 645 * 646 * Assumptions: 647 * - The transport header is entirely contained in the head iovec. 648 */ 649 static int svc_rdma_xdr_decode_req(struct xdr_buf *rq_arg, 650 struct svc_rdma_recv_ctxt *rctxt) 651 { 652 __be32 *p, *rdma_argp; 653 unsigned int hdr_len; 654 655 rdma_argp = rq_arg->head[0].iov_base; 656 xdr_init_decode(&rctxt->rc_stream, rq_arg, rdma_argp, NULL); 657 658 p = xdr_inline_decode(&rctxt->rc_stream, 659 rpcrdma_fixed_maxsz * sizeof(*p)); 660 if (unlikely(!p)) 661 goto out_short; 662 p++; 663 if (*p != rpcrdma_version) 664 goto out_version; 665 p += 2; 666 rctxt->rc_msgtype = *p; 667 switch (rctxt->rc_msgtype) { 668 case rdma_msg: 669 break; 670 case rdma_nomsg: 671 break; 672 case rdma_done: 673 goto out_drop; 674 case rdma_error: 675 goto out_drop; 676 default: 677 goto out_proc; 678 } 679 680 if (!xdr_check_read_list(rctxt)) 681 goto out_inval; 682 if (!xdr_check_write_list(rctxt)) 683 goto out_inval; 684 if (!xdr_check_reply_chunk(rctxt)) 685 goto out_inval; 686 687 rq_arg->head[0].iov_base = rctxt->rc_stream.p; 688 hdr_len = xdr_stream_pos(&rctxt->rc_stream); 689 rq_arg->head[0].iov_len -= hdr_len; 690 rq_arg->len -= hdr_len; 691 trace_svcrdma_decode_rqst(rctxt, rdma_argp, hdr_len); 692 return hdr_len; 693 694 out_short: 695 trace_svcrdma_decode_short_err(rctxt, rq_arg->len); 696 return -EINVAL; 697 698 out_version: 699 trace_svcrdma_decode_badvers_err(rctxt, rdma_argp); 700 return -EPROTONOSUPPORT; 701 702 out_drop: 703 trace_svcrdma_decode_drop_err(rctxt, rdma_argp); 704 return 0; 705 706 out_proc: 707 trace_svcrdma_decode_badproc_err(rctxt, rdma_argp); 708 return -EINVAL; 709 710 out_inval: 711 trace_svcrdma_decode_parse_err(rctxt, rdma_argp); 712 return -EINVAL; 713 } 714 715 static void rdma_read_complete(struct svc_rqst *rqstp, 716 struct svc_rdma_recv_ctxt *head) 717 { 718 int page_no; 719 720 /* Move Read chunk pages to rqstp so that they will be released 721 * when svc_process is done with them. 722 */ 723 for (page_no = 0; page_no < head->rc_page_count; page_no++) { 724 put_page(rqstp->rq_pages[page_no]); 725 rqstp->rq_pages[page_no] = head->rc_pages[page_no]; 726 } 727 head->rc_page_count = 0; 728 729 /* Point rq_arg.pages past header */ 730 rqstp->rq_arg.pages = &rqstp->rq_pages[head->rc_hdr_count]; 731 rqstp->rq_arg.page_len = head->rc_arg.page_len; 732 733 /* rq_respages starts after the last arg page */ 734 rqstp->rq_respages = &rqstp->rq_pages[page_no]; 735 rqstp->rq_next_page = rqstp->rq_respages + 1; 736 737 /* Rebuild rq_arg head and tail. */ 738 rqstp->rq_arg.head[0] = head->rc_arg.head[0]; 739 rqstp->rq_arg.tail[0] = head->rc_arg.tail[0]; 740 rqstp->rq_arg.len = head->rc_arg.len; 741 rqstp->rq_arg.buflen = head->rc_arg.buflen; 742 } 743 744 static void svc_rdma_send_error(struct svcxprt_rdma *rdma, 745 struct svc_rdma_recv_ctxt *rctxt, 746 int status) 747 { 748 struct svc_rdma_send_ctxt *sctxt; 749 750 sctxt = svc_rdma_send_ctxt_get(rdma); 751 if (!sctxt) 752 return; 753 svc_rdma_send_error_msg(rdma, sctxt, rctxt, status); 754 } 755 756 /* By convention, backchannel calls arrive via rdma_msg type 757 * messages, and never populate the chunk lists. This makes 758 * the RPC/RDMA header small and fixed in size, so it is 759 * straightforward to check the RPC header's direction field. 760 */ 761 static bool svc_rdma_is_reverse_direction_reply(struct svc_xprt *xprt, 762 struct svc_rdma_recv_ctxt *rctxt) 763 { 764 __be32 *p = rctxt->rc_recv_buf; 765 766 if (!xprt->xpt_bc_xprt) 767 return false; 768 769 if (rctxt->rc_msgtype != rdma_msg) 770 return false; 771 772 if (!pcl_is_empty(&rctxt->rc_call_pcl)) 773 return false; 774 if (!pcl_is_empty(&rctxt->rc_read_pcl)) 775 return false; 776 if (!pcl_is_empty(&rctxt->rc_write_pcl)) 777 return false; 778 if (!pcl_is_empty(&rctxt->rc_reply_pcl)) 779 return false; 780 781 /* RPC call direction */ 782 if (*(p + 8) == cpu_to_be32(RPC_CALL)) 783 return false; 784 785 return true; 786 } 787 788 /** 789 * svc_rdma_recvfrom - Receive an RPC call 790 * @rqstp: request structure into which to receive an RPC Call 791 * 792 * Returns: 793 * The positive number of bytes in the RPC Call message, 794 * %0 if there were no Calls ready to return, 795 * %-EINVAL if the Read chunk data is too large, 796 * %-ENOMEM if rdma_rw context pool was exhausted, 797 * %-ENOTCONN if posting failed (connection is lost), 798 * %-EIO if rdma_rw initialization failed (DMA mapping, etc). 799 * 800 * Called in a loop when XPT_DATA is set. XPT_DATA is cleared only 801 * when there are no remaining ctxt's to process. 802 * 803 * The next ctxt is removed from the "receive" lists. 804 * 805 * - If the ctxt completes a Read, then finish assembling the Call 806 * message and return the number of bytes in the message. 807 * 808 * - If the ctxt completes a Receive, then construct the Call 809 * message from the contents of the Receive buffer. 810 * 811 * - If there are no Read chunks in this message, then finish 812 * assembling the Call message and return the number of bytes 813 * in the message. 814 * 815 * - If there are Read chunks in this message, post Read WRs to 816 * pull that payload and return 0. 817 */ 818 int svc_rdma_recvfrom(struct svc_rqst *rqstp) 819 { 820 struct svc_xprt *xprt = rqstp->rq_xprt; 821 struct svcxprt_rdma *rdma_xprt = 822 container_of(xprt, struct svcxprt_rdma, sc_xprt); 823 struct svc_rdma_recv_ctxt *ctxt; 824 int ret; 825 826 rqstp->rq_xprt_ctxt = NULL; 827 828 spin_lock(&rdma_xprt->sc_rq_dto_lock); 829 ctxt = svc_rdma_next_recv_ctxt(&rdma_xprt->sc_read_complete_q); 830 if (ctxt) { 831 list_del(&ctxt->rc_list); 832 spin_unlock(&rdma_xprt->sc_rq_dto_lock); 833 rdma_read_complete(rqstp, ctxt); 834 goto complete; 835 } 836 ctxt = svc_rdma_next_recv_ctxt(&rdma_xprt->sc_rq_dto_q); 837 if (!ctxt) { 838 /* No new incoming requests, terminate the loop */ 839 clear_bit(XPT_DATA, &xprt->xpt_flags); 840 spin_unlock(&rdma_xprt->sc_rq_dto_lock); 841 return 0; 842 } 843 list_del(&ctxt->rc_list); 844 spin_unlock(&rdma_xprt->sc_rq_dto_lock); 845 percpu_counter_inc(&svcrdma_stat_recv); 846 847 ib_dma_sync_single_for_cpu(rdma_xprt->sc_pd->device, 848 ctxt->rc_recv_sge.addr, ctxt->rc_byte_len, 849 DMA_FROM_DEVICE); 850 svc_rdma_build_arg_xdr(rqstp, ctxt); 851 852 /* Prevent svc_xprt_release from releasing pages in rq_pages 853 * if we return 0 or an error. 854 */ 855 rqstp->rq_respages = rqstp->rq_pages; 856 rqstp->rq_next_page = rqstp->rq_respages; 857 858 ret = svc_rdma_xdr_decode_req(&rqstp->rq_arg, ctxt); 859 if (ret < 0) 860 goto out_err; 861 if (ret == 0) 862 goto out_drop; 863 rqstp->rq_xprt_hlen = ret; 864 865 if (svc_rdma_is_reverse_direction_reply(xprt, ctxt)) 866 goto out_backchannel; 867 868 svc_rdma_get_inv_rkey(rdma_xprt, ctxt); 869 870 if (!pcl_is_empty(&ctxt->rc_read_pcl) || 871 !pcl_is_empty(&ctxt->rc_call_pcl)) 872 goto out_readlist; 873 874 complete: 875 rqstp->rq_xprt_ctxt = ctxt; 876 rqstp->rq_prot = IPPROTO_MAX; 877 svc_xprt_copy_addrs(rqstp, xprt); 878 return rqstp->rq_arg.len; 879 880 out_readlist: 881 ret = svc_rdma_process_read_list(rdma_xprt, rqstp, ctxt); 882 if (ret < 0) 883 goto out_readfail; 884 return 0; 885 886 out_err: 887 svc_rdma_send_error(rdma_xprt, ctxt, ret); 888 svc_rdma_recv_ctxt_put(rdma_xprt, ctxt); 889 return 0; 890 891 out_readfail: 892 if (ret == -EINVAL) 893 svc_rdma_send_error(rdma_xprt, ctxt, ret); 894 svc_rdma_recv_ctxt_put(rdma_xprt, ctxt); 895 return ret; 896 897 out_backchannel: 898 svc_rdma_handle_bc_reply(rqstp, ctxt); 899 out_drop: 900 svc_rdma_recv_ctxt_put(rdma_xprt, ctxt); 901 return 0; 902 } 903