1 // SPDX-License-Identifier: GPL-2.0 2 /* 3 * Copyright (c) 2016-2018 Oracle. All rights reserved. 4 * 5 * Use the core R/W API to move RPC-over-RDMA Read and Write chunks. 6 */ 7 8 #include <rdma/rw.h> 9 10 #include <linux/sunrpc/xdr.h> 11 #include <linux/sunrpc/rpc_rdma.h> 12 #include <linux/sunrpc/svc_rdma.h> 13 14 #include "xprt_rdma.h" 15 #include <trace/events/rpcrdma.h> 16 17 static void svc_rdma_write_done(struct ib_cq *cq, struct ib_wc *wc); 18 static void svc_rdma_wc_read_done(struct ib_cq *cq, struct ib_wc *wc); 19 20 /* Each R/W context contains state for one chain of RDMA Read or 21 * Write Work Requests. 22 * 23 * Each WR chain handles a single contiguous server-side buffer, 24 * because scatterlist entries after the first have to start on 25 * page alignment. xdr_buf iovecs cannot guarantee alignment. 26 * 27 * Each WR chain handles only one R_key. Each RPC-over-RDMA segment 28 * from a client may contain a unique R_key, so each WR chain moves 29 * up to one segment at a time. 30 * 31 * The scatterlist makes this data structure over 4KB in size. To 32 * make it less likely to fail, and to handle the allocation for 33 * smaller I/O requests without disabling bottom-halves, these 34 * contexts are created on demand, but cached and reused until the 35 * controlling svcxprt_rdma is destroyed. 36 */ 37 struct svc_rdma_rw_ctxt { 38 struct llist_node rw_node; 39 struct list_head rw_list; 40 struct rdma_rw_ctx rw_ctx; 41 unsigned int rw_nents; 42 struct sg_table rw_sg_table; 43 struct scatterlist rw_first_sgl[]; 44 }; 45 46 static inline struct svc_rdma_rw_ctxt * 47 svc_rdma_next_ctxt(struct list_head *list) 48 { 49 return list_first_entry_or_null(list, struct svc_rdma_rw_ctxt, 50 rw_list); 51 } 52 53 static struct svc_rdma_rw_ctxt * 54 svc_rdma_get_rw_ctxt(struct svcxprt_rdma *rdma, unsigned int sges) 55 { 56 struct svc_rdma_rw_ctxt *ctxt; 57 struct llist_node *node; 58 59 spin_lock(&rdma->sc_rw_ctxt_lock); 60 node = llist_del_first(&rdma->sc_rw_ctxts); 61 spin_unlock(&rdma->sc_rw_ctxt_lock); 62 if (node) { 63 ctxt = llist_entry(node, struct svc_rdma_rw_ctxt, rw_node); 64 } else { 65 ctxt = kmalloc(struct_size(ctxt, rw_first_sgl, SG_CHUNK_SIZE), 66 GFP_KERNEL); 67 if (!ctxt) 68 goto out_noctx; 69 70 INIT_LIST_HEAD(&ctxt->rw_list); 71 } 72 73 ctxt->rw_sg_table.sgl = ctxt->rw_first_sgl; 74 if (sg_alloc_table_chained(&ctxt->rw_sg_table, sges, 75 ctxt->rw_sg_table.sgl, 76 SG_CHUNK_SIZE)) 77 goto out_free; 78 return ctxt; 79 80 out_free: 81 kfree(ctxt); 82 out_noctx: 83 trace_svcrdma_no_rwctx_err(rdma, sges); 84 return NULL; 85 } 86 87 static void __svc_rdma_put_rw_ctxt(struct svcxprt_rdma *rdma, 88 struct svc_rdma_rw_ctxt *ctxt, 89 struct llist_head *list) 90 { 91 sg_free_table_chained(&ctxt->rw_sg_table, SG_CHUNK_SIZE); 92 llist_add(&ctxt->rw_node, list); 93 } 94 95 static void svc_rdma_put_rw_ctxt(struct svcxprt_rdma *rdma, 96 struct svc_rdma_rw_ctxt *ctxt) 97 { 98 __svc_rdma_put_rw_ctxt(rdma, ctxt, &rdma->sc_rw_ctxts); 99 } 100 101 /** 102 * svc_rdma_destroy_rw_ctxts - Free accumulated R/W contexts 103 * @rdma: transport about to be destroyed 104 * 105 */ 106 void svc_rdma_destroy_rw_ctxts(struct svcxprt_rdma *rdma) 107 { 108 struct svc_rdma_rw_ctxt *ctxt; 109 struct llist_node *node; 110 111 while ((node = llist_del_first(&rdma->sc_rw_ctxts)) != NULL) { 112 ctxt = llist_entry(node, struct svc_rdma_rw_ctxt, rw_node); 113 kfree(ctxt); 114 } 115 } 116 117 /** 118 * svc_rdma_rw_ctx_init - Prepare a R/W context for I/O 119 * @rdma: controlling transport instance 120 * @ctxt: R/W context to prepare 121 * @offset: RDMA offset 122 * @handle: RDMA tag/handle 123 * @direction: I/O direction 124 * 125 * Returns on success, the number of WQEs that will be needed 126 * on the workqueue, or a negative errno. 127 */ 128 static int svc_rdma_rw_ctx_init(struct svcxprt_rdma *rdma, 129 struct svc_rdma_rw_ctxt *ctxt, 130 u64 offset, u32 handle, 131 enum dma_data_direction direction) 132 { 133 int ret; 134 135 ret = rdma_rw_ctx_init(&ctxt->rw_ctx, rdma->sc_qp, rdma->sc_port_num, 136 ctxt->rw_sg_table.sgl, ctxt->rw_nents, 137 0, offset, handle, direction); 138 if (unlikely(ret < 0)) { 139 svc_rdma_put_rw_ctxt(rdma, ctxt); 140 trace_svcrdma_dma_map_rw_err(rdma, ctxt->rw_nents, ret); 141 } 142 return ret; 143 } 144 145 /* A chunk context tracks all I/O for moving one Read or Write 146 * chunk. This is a set of rdma_rw's that handle data movement 147 * for all segments of one chunk. 148 * 149 * These are small, acquired with a single allocator call, and 150 * no more than one is needed per chunk. They are allocated on 151 * demand, and not cached. 152 */ 153 struct svc_rdma_chunk_ctxt { 154 struct rpc_rdma_cid cc_cid; 155 struct ib_cqe cc_cqe; 156 struct svcxprt_rdma *cc_rdma; 157 struct list_head cc_rwctxts; 158 ktime_t cc_posttime; 159 int cc_sqecount; 160 enum ib_wc_status cc_status; 161 struct completion cc_done; 162 }; 163 164 static void svc_rdma_cc_cid_init(struct svcxprt_rdma *rdma, 165 struct rpc_rdma_cid *cid) 166 { 167 cid->ci_queue_id = rdma->sc_sq_cq->res.id; 168 cid->ci_completion_id = atomic_inc_return(&rdma->sc_completion_ids); 169 } 170 171 static void svc_rdma_cc_init(struct svcxprt_rdma *rdma, 172 struct svc_rdma_chunk_ctxt *cc) 173 { 174 svc_rdma_cc_cid_init(rdma, &cc->cc_cid); 175 cc->cc_rdma = rdma; 176 177 INIT_LIST_HEAD(&cc->cc_rwctxts); 178 cc->cc_sqecount = 0; 179 } 180 181 /* 182 * The consumed rw_ctx's are cleaned and placed on a local llist so 183 * that only one atomic llist operation is needed to put them all 184 * back on the free list. 185 */ 186 static void svc_rdma_cc_release(struct svc_rdma_chunk_ctxt *cc, 187 enum dma_data_direction dir) 188 { 189 struct svcxprt_rdma *rdma = cc->cc_rdma; 190 struct llist_node *first, *last; 191 struct svc_rdma_rw_ctxt *ctxt; 192 LLIST_HEAD(free); 193 194 first = last = NULL; 195 while ((ctxt = svc_rdma_next_ctxt(&cc->cc_rwctxts)) != NULL) { 196 list_del(&ctxt->rw_list); 197 198 rdma_rw_ctx_destroy(&ctxt->rw_ctx, rdma->sc_qp, 199 rdma->sc_port_num, ctxt->rw_sg_table.sgl, 200 ctxt->rw_nents, dir); 201 __svc_rdma_put_rw_ctxt(rdma, ctxt, &free); 202 203 ctxt->rw_node.next = first; 204 first = &ctxt->rw_node; 205 if (!last) 206 last = first; 207 } 208 if (first) 209 llist_add_batch(first, last, &rdma->sc_rw_ctxts); 210 } 211 212 /* State for sending a Write or Reply chunk. 213 * - Tracks progress of writing one chunk over all its segments 214 * - Stores arguments for the SGL constructor functions 215 */ 216 struct svc_rdma_write_info { 217 const struct svc_rdma_chunk *wi_chunk; 218 219 /* write state of this chunk */ 220 unsigned int wi_seg_off; 221 unsigned int wi_seg_no; 222 223 /* SGL constructor arguments */ 224 const struct xdr_buf *wi_xdr; 225 unsigned char *wi_base; 226 unsigned int wi_next_off; 227 228 struct svc_rdma_chunk_ctxt wi_cc; 229 }; 230 231 static struct svc_rdma_write_info * 232 svc_rdma_write_info_alloc(struct svcxprt_rdma *rdma, 233 const struct svc_rdma_chunk *chunk) 234 { 235 struct svc_rdma_write_info *info; 236 237 info = kmalloc(sizeof(*info), GFP_KERNEL); 238 if (!info) 239 return info; 240 241 info->wi_chunk = chunk; 242 info->wi_seg_off = 0; 243 info->wi_seg_no = 0; 244 svc_rdma_cc_init(rdma, &info->wi_cc); 245 info->wi_cc.cc_cqe.done = svc_rdma_write_done; 246 return info; 247 } 248 249 static void svc_rdma_write_info_free(struct svc_rdma_write_info *info) 250 { 251 svc_rdma_cc_release(&info->wi_cc, DMA_TO_DEVICE); 252 kfree(info); 253 } 254 255 /** 256 * svc_rdma_write_done - Write chunk completion 257 * @cq: controlling Completion Queue 258 * @wc: Work Completion 259 * 260 * Pages under I/O are freed by a subsequent Send completion. 261 */ 262 static void svc_rdma_write_done(struct ib_cq *cq, struct ib_wc *wc) 263 { 264 struct ib_cqe *cqe = wc->wr_cqe; 265 struct svc_rdma_chunk_ctxt *cc = 266 container_of(cqe, struct svc_rdma_chunk_ctxt, cc_cqe); 267 struct svcxprt_rdma *rdma = cc->cc_rdma; 268 struct svc_rdma_write_info *info = 269 container_of(cc, struct svc_rdma_write_info, wi_cc); 270 271 switch (wc->status) { 272 case IB_WC_SUCCESS: 273 trace_svcrdma_wc_write(wc, &cc->cc_cid); 274 break; 275 case IB_WC_WR_FLUSH_ERR: 276 trace_svcrdma_wc_write_flush(wc, &cc->cc_cid); 277 break; 278 default: 279 trace_svcrdma_wc_write_err(wc, &cc->cc_cid); 280 } 281 282 svc_rdma_wake_send_waiters(rdma, cc->cc_sqecount); 283 284 if (unlikely(wc->status != IB_WC_SUCCESS)) 285 svc_xprt_deferred_close(&rdma->sc_xprt); 286 287 svc_rdma_write_info_free(info); 288 } 289 290 /* State for pulling a Read chunk. 291 */ 292 struct svc_rdma_read_info { 293 struct svc_rqst *ri_rqst; 294 struct svc_rdma_recv_ctxt *ri_readctxt; 295 unsigned int ri_pageno; 296 unsigned int ri_pageoff; 297 unsigned int ri_totalbytes; 298 299 struct svc_rdma_chunk_ctxt ri_cc; 300 }; 301 302 static struct svc_rdma_read_info * 303 svc_rdma_read_info_alloc(struct svcxprt_rdma *rdma) 304 { 305 struct svc_rdma_read_info *info; 306 307 info = kmalloc(sizeof(*info), GFP_KERNEL); 308 if (!info) 309 return info; 310 311 svc_rdma_cc_init(rdma, &info->ri_cc); 312 info->ri_cc.cc_cqe.done = svc_rdma_wc_read_done; 313 return info; 314 } 315 316 static void svc_rdma_read_info_free(struct svc_rdma_read_info *info) 317 { 318 svc_rdma_cc_release(&info->ri_cc, DMA_FROM_DEVICE); 319 kfree(info); 320 } 321 322 /** 323 * svc_rdma_wc_read_done - Handle completion of an RDMA Read ctx 324 * @cq: controlling Completion Queue 325 * @wc: Work Completion 326 * 327 */ 328 static void svc_rdma_wc_read_done(struct ib_cq *cq, struct ib_wc *wc) 329 { 330 struct ib_cqe *cqe = wc->wr_cqe; 331 struct svc_rdma_chunk_ctxt *cc = 332 container_of(cqe, struct svc_rdma_chunk_ctxt, cc_cqe); 333 struct svc_rdma_read_info *info; 334 335 switch (wc->status) { 336 case IB_WC_SUCCESS: 337 info = container_of(cc, struct svc_rdma_read_info, ri_cc); 338 trace_svcrdma_wc_read(wc, &cc->cc_cid, info->ri_totalbytes, 339 cc->cc_posttime); 340 break; 341 case IB_WC_WR_FLUSH_ERR: 342 trace_svcrdma_wc_read_flush(wc, &cc->cc_cid); 343 break; 344 default: 345 trace_svcrdma_wc_read_err(wc, &cc->cc_cid); 346 } 347 348 svc_rdma_wake_send_waiters(cc->cc_rdma, cc->cc_sqecount); 349 cc->cc_status = wc->status; 350 complete(&cc->cc_done); 351 return; 352 } 353 354 /* This function sleeps when the transport's Send Queue is congested. 355 * 356 * Assumptions: 357 * - If ib_post_send() succeeds, only one completion is expected, 358 * even if one or more WRs are flushed. This is true when posting 359 * an rdma_rw_ctx or when posting a single signaled WR. 360 */ 361 static int svc_rdma_post_chunk_ctxt(struct svc_rdma_chunk_ctxt *cc) 362 { 363 struct svcxprt_rdma *rdma = cc->cc_rdma; 364 struct ib_send_wr *first_wr; 365 const struct ib_send_wr *bad_wr; 366 struct list_head *tmp; 367 struct ib_cqe *cqe; 368 int ret; 369 370 if (cc->cc_sqecount > rdma->sc_sq_depth) 371 return -EINVAL; 372 373 first_wr = NULL; 374 cqe = &cc->cc_cqe; 375 list_for_each(tmp, &cc->cc_rwctxts) { 376 struct svc_rdma_rw_ctxt *ctxt; 377 378 ctxt = list_entry(tmp, struct svc_rdma_rw_ctxt, rw_list); 379 first_wr = rdma_rw_ctx_wrs(&ctxt->rw_ctx, rdma->sc_qp, 380 rdma->sc_port_num, cqe, first_wr); 381 cqe = NULL; 382 } 383 384 do { 385 if (atomic_sub_return(cc->cc_sqecount, 386 &rdma->sc_sq_avail) > 0) { 387 cc->cc_posttime = ktime_get(); 388 ret = ib_post_send(rdma->sc_qp, first_wr, &bad_wr); 389 if (ret) 390 break; 391 return 0; 392 } 393 394 percpu_counter_inc(&svcrdma_stat_sq_starve); 395 trace_svcrdma_sq_full(rdma); 396 atomic_add(cc->cc_sqecount, &rdma->sc_sq_avail); 397 wait_event(rdma->sc_send_wait, 398 atomic_read(&rdma->sc_sq_avail) > cc->cc_sqecount); 399 trace_svcrdma_sq_retry(rdma); 400 } while (1); 401 402 trace_svcrdma_sq_post_err(rdma, ret); 403 svc_xprt_deferred_close(&rdma->sc_xprt); 404 405 /* If even one was posted, there will be a completion. */ 406 if (bad_wr != first_wr) 407 return 0; 408 409 atomic_add(cc->cc_sqecount, &rdma->sc_sq_avail); 410 wake_up(&rdma->sc_send_wait); 411 return -ENOTCONN; 412 } 413 414 /* Build and DMA-map an SGL that covers one kvec in an xdr_buf 415 */ 416 static void svc_rdma_vec_to_sg(struct svc_rdma_write_info *info, 417 unsigned int len, 418 struct svc_rdma_rw_ctxt *ctxt) 419 { 420 struct scatterlist *sg = ctxt->rw_sg_table.sgl; 421 422 sg_set_buf(&sg[0], info->wi_base, len); 423 info->wi_base += len; 424 425 ctxt->rw_nents = 1; 426 } 427 428 /* Build and DMA-map an SGL that covers part of an xdr_buf's pagelist. 429 */ 430 static void svc_rdma_pagelist_to_sg(struct svc_rdma_write_info *info, 431 unsigned int remaining, 432 struct svc_rdma_rw_ctxt *ctxt) 433 { 434 unsigned int sge_no, sge_bytes, page_off, page_no; 435 const struct xdr_buf *xdr = info->wi_xdr; 436 struct scatterlist *sg; 437 struct page **page; 438 439 page_off = info->wi_next_off + xdr->page_base; 440 page_no = page_off >> PAGE_SHIFT; 441 page_off = offset_in_page(page_off); 442 page = xdr->pages + page_no; 443 info->wi_next_off += remaining; 444 sg = ctxt->rw_sg_table.sgl; 445 sge_no = 0; 446 do { 447 sge_bytes = min_t(unsigned int, remaining, 448 PAGE_SIZE - page_off); 449 sg_set_page(sg, *page, sge_bytes, page_off); 450 451 remaining -= sge_bytes; 452 sg = sg_next(sg); 453 page_off = 0; 454 sge_no++; 455 page++; 456 } while (remaining); 457 458 ctxt->rw_nents = sge_no; 459 } 460 461 /* Construct RDMA Write WRs to send a portion of an xdr_buf containing 462 * an RPC Reply. 463 */ 464 static int 465 svc_rdma_build_writes(struct svc_rdma_write_info *info, 466 void (*constructor)(struct svc_rdma_write_info *info, 467 unsigned int len, 468 struct svc_rdma_rw_ctxt *ctxt), 469 unsigned int remaining) 470 { 471 struct svc_rdma_chunk_ctxt *cc = &info->wi_cc; 472 struct svcxprt_rdma *rdma = cc->cc_rdma; 473 const struct svc_rdma_segment *seg; 474 struct svc_rdma_rw_ctxt *ctxt; 475 int ret; 476 477 do { 478 unsigned int write_len; 479 u64 offset; 480 481 seg = &info->wi_chunk->ch_segments[info->wi_seg_no]; 482 if (!seg) 483 goto out_overflow; 484 485 write_len = min(remaining, seg->rs_length - info->wi_seg_off); 486 if (!write_len) 487 goto out_overflow; 488 ctxt = svc_rdma_get_rw_ctxt(rdma, 489 (write_len >> PAGE_SHIFT) + 2); 490 if (!ctxt) 491 return -ENOMEM; 492 493 constructor(info, write_len, ctxt); 494 offset = seg->rs_offset + info->wi_seg_off; 495 ret = svc_rdma_rw_ctx_init(rdma, ctxt, offset, seg->rs_handle, 496 DMA_TO_DEVICE); 497 if (ret < 0) 498 return -EIO; 499 percpu_counter_inc(&svcrdma_stat_write); 500 501 list_add(&ctxt->rw_list, &cc->cc_rwctxts); 502 cc->cc_sqecount += ret; 503 if (write_len == seg->rs_length - info->wi_seg_off) { 504 info->wi_seg_no++; 505 info->wi_seg_off = 0; 506 } else { 507 info->wi_seg_off += write_len; 508 } 509 remaining -= write_len; 510 } while (remaining); 511 512 return 0; 513 514 out_overflow: 515 trace_svcrdma_small_wrch_err(rdma, remaining, info->wi_seg_no, 516 info->wi_chunk->ch_segcount); 517 return -E2BIG; 518 } 519 520 /** 521 * svc_rdma_iov_write - Construct RDMA Writes from an iov 522 * @info: pointer to write arguments 523 * @iov: kvec to write 524 * 525 * Returns: 526 * On success, returns zero 527 * %-E2BIG if the client-provided Write chunk is too small 528 * %-ENOMEM if a resource has been exhausted 529 * %-EIO if an rdma-rw error occurred 530 */ 531 static int svc_rdma_iov_write(struct svc_rdma_write_info *info, 532 const struct kvec *iov) 533 { 534 info->wi_base = iov->iov_base; 535 return svc_rdma_build_writes(info, svc_rdma_vec_to_sg, 536 iov->iov_len); 537 } 538 539 /** 540 * svc_rdma_pages_write - Construct RDMA Writes from pages 541 * @info: pointer to write arguments 542 * @xdr: xdr_buf with pages to write 543 * @offset: offset into the content of @xdr 544 * @length: number of bytes to write 545 * 546 * Returns: 547 * On success, returns zero 548 * %-E2BIG if the client-provided Write chunk is too small 549 * %-ENOMEM if a resource has been exhausted 550 * %-EIO if an rdma-rw error occurred 551 */ 552 static int svc_rdma_pages_write(struct svc_rdma_write_info *info, 553 const struct xdr_buf *xdr, 554 unsigned int offset, 555 unsigned long length) 556 { 557 info->wi_xdr = xdr; 558 info->wi_next_off = offset - xdr->head[0].iov_len; 559 return svc_rdma_build_writes(info, svc_rdma_pagelist_to_sg, 560 length); 561 } 562 563 /** 564 * svc_rdma_xb_write - Construct RDMA Writes to write an xdr_buf 565 * @xdr: xdr_buf to write 566 * @data: pointer to write arguments 567 * 568 * Returns: 569 * On success, returns zero 570 * %-E2BIG if the client-provided Write chunk is too small 571 * %-ENOMEM if a resource has been exhausted 572 * %-EIO if an rdma-rw error occurred 573 */ 574 static int svc_rdma_xb_write(const struct xdr_buf *xdr, void *data) 575 { 576 struct svc_rdma_write_info *info = data; 577 int ret; 578 579 if (xdr->head[0].iov_len) { 580 ret = svc_rdma_iov_write(info, &xdr->head[0]); 581 if (ret < 0) 582 return ret; 583 } 584 585 if (xdr->page_len) { 586 ret = svc_rdma_pages_write(info, xdr, xdr->head[0].iov_len, 587 xdr->page_len); 588 if (ret < 0) 589 return ret; 590 } 591 592 if (xdr->tail[0].iov_len) { 593 ret = svc_rdma_iov_write(info, &xdr->tail[0]); 594 if (ret < 0) 595 return ret; 596 } 597 598 return xdr->len; 599 } 600 601 /** 602 * svc_rdma_send_write_chunk - Write all segments in a Write chunk 603 * @rdma: controlling RDMA transport 604 * @chunk: Write chunk provided by the client 605 * @xdr: xdr_buf containing the data payload 606 * 607 * Returns a non-negative number of bytes the chunk consumed, or 608 * %-E2BIG if the payload was larger than the Write chunk, 609 * %-EINVAL if client provided too many segments, 610 * %-ENOMEM if rdma_rw context pool was exhausted, 611 * %-ENOTCONN if posting failed (connection is lost), 612 * %-EIO if rdma_rw initialization failed (DMA mapping, etc). 613 */ 614 int svc_rdma_send_write_chunk(struct svcxprt_rdma *rdma, 615 const struct svc_rdma_chunk *chunk, 616 const struct xdr_buf *xdr) 617 { 618 struct svc_rdma_write_info *info; 619 struct svc_rdma_chunk_ctxt *cc; 620 int ret; 621 622 info = svc_rdma_write_info_alloc(rdma, chunk); 623 if (!info) 624 return -ENOMEM; 625 cc = &info->wi_cc; 626 627 ret = svc_rdma_xb_write(xdr, info); 628 if (ret != xdr->len) 629 goto out_err; 630 631 trace_svcrdma_post_write_chunk(&cc->cc_cid, cc->cc_sqecount); 632 ret = svc_rdma_post_chunk_ctxt(cc); 633 if (ret < 0) 634 goto out_err; 635 return xdr->len; 636 637 out_err: 638 svc_rdma_write_info_free(info); 639 return ret; 640 } 641 642 /** 643 * svc_rdma_send_reply_chunk - Write all segments in the Reply chunk 644 * @rdma: controlling RDMA transport 645 * @rctxt: Write and Reply chunks from client 646 * @xdr: xdr_buf containing an RPC Reply 647 * 648 * Returns a non-negative number of bytes the chunk consumed, or 649 * %-E2BIG if the payload was larger than the Reply chunk, 650 * %-EINVAL if client provided too many segments, 651 * %-ENOMEM if rdma_rw context pool was exhausted, 652 * %-ENOTCONN if posting failed (connection is lost), 653 * %-EIO if rdma_rw initialization failed (DMA mapping, etc). 654 */ 655 int svc_rdma_send_reply_chunk(struct svcxprt_rdma *rdma, 656 const struct svc_rdma_recv_ctxt *rctxt, 657 const struct xdr_buf *xdr) 658 { 659 struct svc_rdma_write_info *info; 660 struct svc_rdma_chunk_ctxt *cc; 661 struct svc_rdma_chunk *chunk; 662 int ret; 663 664 if (pcl_is_empty(&rctxt->rc_reply_pcl)) 665 return 0; 666 667 chunk = pcl_first_chunk(&rctxt->rc_reply_pcl); 668 info = svc_rdma_write_info_alloc(rdma, chunk); 669 if (!info) 670 return -ENOMEM; 671 cc = &info->wi_cc; 672 673 ret = pcl_process_nonpayloads(&rctxt->rc_write_pcl, xdr, 674 svc_rdma_xb_write, info); 675 if (ret < 0) 676 goto out_err; 677 678 trace_svcrdma_post_reply_chunk(&cc->cc_cid, cc->cc_sqecount); 679 ret = svc_rdma_post_chunk_ctxt(cc); 680 if (ret < 0) 681 goto out_err; 682 683 return xdr->len; 684 685 out_err: 686 svc_rdma_write_info_free(info); 687 return ret; 688 } 689 690 /** 691 * svc_rdma_build_read_segment - Build RDMA Read WQEs to pull one RDMA segment 692 * @info: context for ongoing I/O 693 * @segment: co-ordinates of remote memory to be read 694 * 695 * Returns: 696 * %0: the Read WR chain was constructed successfully 697 * %-EINVAL: there were not enough rq_pages to finish 698 * %-ENOMEM: allocating a local resources failed 699 * %-EIO: a DMA mapping error occurred 700 */ 701 static int svc_rdma_build_read_segment(struct svc_rdma_read_info *info, 702 const struct svc_rdma_segment *segment) 703 { 704 struct svc_rdma_recv_ctxt *head = info->ri_readctxt; 705 struct svc_rdma_chunk_ctxt *cc = &info->ri_cc; 706 struct svc_rqst *rqstp = info->ri_rqst; 707 unsigned int sge_no, seg_len, len; 708 struct svc_rdma_rw_ctxt *ctxt; 709 struct scatterlist *sg; 710 int ret; 711 712 len = segment->rs_length; 713 sge_no = PAGE_ALIGN(info->ri_pageoff + len) >> PAGE_SHIFT; 714 ctxt = svc_rdma_get_rw_ctxt(cc->cc_rdma, sge_no); 715 if (!ctxt) 716 return -ENOMEM; 717 ctxt->rw_nents = sge_no; 718 719 sg = ctxt->rw_sg_table.sgl; 720 for (sge_no = 0; sge_no < ctxt->rw_nents; sge_no++) { 721 seg_len = min_t(unsigned int, len, 722 PAGE_SIZE - info->ri_pageoff); 723 724 if (!info->ri_pageoff) 725 head->rc_page_count++; 726 727 sg_set_page(sg, rqstp->rq_pages[info->ri_pageno], 728 seg_len, info->ri_pageoff); 729 sg = sg_next(sg); 730 731 info->ri_pageoff += seg_len; 732 if (info->ri_pageoff == PAGE_SIZE) { 733 info->ri_pageno++; 734 info->ri_pageoff = 0; 735 } 736 len -= seg_len; 737 738 /* Safety check */ 739 if (len && 740 &rqstp->rq_pages[info->ri_pageno + 1] > rqstp->rq_page_end) 741 goto out_overrun; 742 } 743 744 ret = svc_rdma_rw_ctx_init(cc->cc_rdma, ctxt, segment->rs_offset, 745 segment->rs_handle, DMA_FROM_DEVICE); 746 if (ret < 0) 747 return -EIO; 748 percpu_counter_inc(&svcrdma_stat_read); 749 750 list_add(&ctxt->rw_list, &cc->cc_rwctxts); 751 cc->cc_sqecount += ret; 752 return 0; 753 754 out_overrun: 755 trace_svcrdma_page_overrun_err(cc->cc_rdma, rqstp, info->ri_pageno); 756 return -EINVAL; 757 } 758 759 /** 760 * svc_rdma_build_read_chunk - Build RDMA Read WQEs to pull one RDMA chunk 761 * @info: context for ongoing I/O 762 * @chunk: Read chunk to pull 763 * 764 * Return values: 765 * %0: the Read WR chain was constructed successfully 766 * %-EINVAL: there were not enough resources to finish 767 * %-ENOMEM: allocating a local resources failed 768 * %-EIO: a DMA mapping error occurred 769 */ 770 static int svc_rdma_build_read_chunk(struct svc_rdma_read_info *info, 771 const struct svc_rdma_chunk *chunk) 772 { 773 const struct svc_rdma_segment *segment; 774 int ret; 775 776 ret = -EINVAL; 777 pcl_for_each_segment(segment, chunk) { 778 ret = svc_rdma_build_read_segment(info, segment); 779 if (ret < 0) 780 break; 781 info->ri_totalbytes += segment->rs_length; 782 } 783 return ret; 784 } 785 786 /** 787 * svc_rdma_copy_inline_range - Copy part of the inline content into pages 788 * @info: context for RDMA Reads 789 * @offset: offset into the Receive buffer of region to copy 790 * @remaining: length of region to copy 791 * 792 * Take a page at a time from rqstp->rq_pages and copy the inline 793 * content from the Receive buffer into that page. Update 794 * info->ri_pageno and info->ri_pageoff so that the next RDMA Read 795 * result will land contiguously with the copied content. 796 * 797 * Return values: 798 * %0: Inline content was successfully copied 799 * %-EINVAL: offset or length was incorrect 800 */ 801 static int svc_rdma_copy_inline_range(struct svc_rdma_read_info *info, 802 unsigned int offset, 803 unsigned int remaining) 804 { 805 struct svc_rdma_recv_ctxt *head = info->ri_readctxt; 806 unsigned char *dst, *src = head->rc_recv_buf; 807 struct svc_rqst *rqstp = info->ri_rqst; 808 unsigned int page_no, numpages; 809 810 numpages = PAGE_ALIGN(info->ri_pageoff + remaining) >> PAGE_SHIFT; 811 for (page_no = 0; page_no < numpages; page_no++) { 812 unsigned int page_len; 813 814 page_len = min_t(unsigned int, remaining, 815 PAGE_SIZE - info->ri_pageoff); 816 817 if (!info->ri_pageoff) 818 head->rc_page_count++; 819 820 dst = page_address(rqstp->rq_pages[info->ri_pageno]); 821 memcpy(dst + info->ri_pageno, src + offset, page_len); 822 823 info->ri_totalbytes += page_len; 824 info->ri_pageoff += page_len; 825 if (info->ri_pageoff == PAGE_SIZE) { 826 info->ri_pageno++; 827 info->ri_pageoff = 0; 828 } 829 remaining -= page_len; 830 offset += page_len; 831 } 832 833 return -EINVAL; 834 } 835 836 /** 837 * svc_rdma_read_multiple_chunks - Construct RDMA Reads to pull data item Read chunks 838 * @info: context for RDMA Reads 839 * 840 * The chunk data lands in rqstp->rq_arg as a series of contiguous pages, 841 * like an incoming TCP call. 842 * 843 * Return values: 844 * %0: RDMA Read WQEs were successfully built 845 * %-EINVAL: client provided too many chunks or segments, 846 * %-ENOMEM: rdma_rw context pool was exhausted, 847 * %-ENOTCONN: posting failed (connection is lost), 848 * %-EIO: rdma_rw initialization failed (DMA mapping, etc). 849 */ 850 static noinline int svc_rdma_read_multiple_chunks(struct svc_rdma_read_info *info) 851 { 852 struct svc_rdma_recv_ctxt *head = info->ri_readctxt; 853 const struct svc_rdma_pcl *pcl = &head->rc_read_pcl; 854 struct xdr_buf *buf = &info->ri_rqst->rq_arg; 855 struct svc_rdma_chunk *chunk, *next; 856 unsigned int start, length; 857 int ret; 858 859 start = 0; 860 chunk = pcl_first_chunk(pcl); 861 length = chunk->ch_position; 862 ret = svc_rdma_copy_inline_range(info, start, length); 863 if (ret < 0) 864 return ret; 865 866 pcl_for_each_chunk(chunk, pcl) { 867 ret = svc_rdma_build_read_chunk(info, chunk); 868 if (ret < 0) 869 return ret; 870 871 next = pcl_next_chunk(pcl, chunk); 872 if (!next) 873 break; 874 875 start += length; 876 length = next->ch_position - info->ri_totalbytes; 877 ret = svc_rdma_copy_inline_range(info, start, length); 878 if (ret < 0) 879 return ret; 880 } 881 882 start += length; 883 length = head->rc_byte_len - start; 884 ret = svc_rdma_copy_inline_range(info, start, length); 885 if (ret < 0) 886 return ret; 887 888 buf->len += info->ri_totalbytes; 889 buf->buflen += info->ri_totalbytes; 890 891 buf->head[0].iov_base = page_address(info->ri_rqst->rq_pages[0]); 892 buf->head[0].iov_len = min_t(size_t, PAGE_SIZE, info->ri_totalbytes); 893 buf->pages = &info->ri_rqst->rq_pages[1]; 894 buf->page_len = info->ri_totalbytes - buf->head[0].iov_len; 895 return 0; 896 } 897 898 /** 899 * svc_rdma_read_data_item - Construct RDMA Reads to pull data item Read chunks 900 * @info: context for RDMA Reads 901 * 902 * The chunk data lands in the page list of rqstp->rq_arg.pages. 903 * 904 * Currently NFSD does not look at the rqstp->rq_arg.tail[0] kvec. 905 * Therefore, XDR round-up of the Read chunk and trailing 906 * inline content must both be added at the end of the pagelist. 907 * 908 * Return values: 909 * %0: RDMA Read WQEs were successfully built 910 * %-EINVAL: client provided too many chunks or segments, 911 * %-ENOMEM: rdma_rw context pool was exhausted, 912 * %-ENOTCONN: posting failed (connection is lost), 913 * %-EIO: rdma_rw initialization failed (DMA mapping, etc). 914 */ 915 static int svc_rdma_read_data_item(struct svc_rdma_read_info *info) 916 { 917 struct svc_rdma_recv_ctxt *head = info->ri_readctxt; 918 struct xdr_buf *buf = &info->ri_rqst->rq_arg; 919 struct svc_rdma_chunk *chunk; 920 unsigned int length; 921 int ret; 922 923 chunk = pcl_first_chunk(&head->rc_read_pcl); 924 ret = svc_rdma_build_read_chunk(info, chunk); 925 if (ret < 0) 926 goto out; 927 928 /* Split the Receive buffer between the head and tail 929 * buffers at Read chunk's position. XDR roundup of the 930 * chunk is not included in either the pagelist or in 931 * the tail. 932 */ 933 buf->tail[0].iov_base = buf->head[0].iov_base + chunk->ch_position; 934 buf->tail[0].iov_len = buf->head[0].iov_len - chunk->ch_position; 935 buf->head[0].iov_len = chunk->ch_position; 936 937 /* Read chunk may need XDR roundup (see RFC 8166, s. 3.4.5.2). 938 * 939 * If the client already rounded up the chunk length, the 940 * length does not change. Otherwise, the length of the page 941 * list is increased to include XDR round-up. 942 * 943 * Currently these chunks always start at page offset 0, 944 * thus the rounded-up length never crosses a page boundary. 945 */ 946 buf->pages = &info->ri_rqst->rq_pages[0]; 947 length = xdr_align_size(chunk->ch_length); 948 buf->page_len = length; 949 buf->len += length; 950 buf->buflen += length; 951 952 out: 953 return ret; 954 } 955 956 /** 957 * svc_rdma_read_chunk_range - Build RDMA Read WQEs for portion of a chunk 958 * @info: context for RDMA Reads 959 * @chunk: parsed Call chunk to pull 960 * @offset: offset of region to pull 961 * @length: length of region to pull 962 * 963 * Return values: 964 * %0: RDMA Read WQEs were successfully built 965 * %-EINVAL: there were not enough resources to finish 966 * %-ENOMEM: rdma_rw context pool was exhausted, 967 * %-ENOTCONN: posting failed (connection is lost), 968 * %-EIO: rdma_rw initialization failed (DMA mapping, etc). 969 */ 970 static int svc_rdma_read_chunk_range(struct svc_rdma_read_info *info, 971 const struct svc_rdma_chunk *chunk, 972 unsigned int offset, unsigned int length) 973 { 974 const struct svc_rdma_segment *segment; 975 int ret; 976 977 ret = -EINVAL; 978 pcl_for_each_segment(segment, chunk) { 979 struct svc_rdma_segment dummy; 980 981 if (offset > segment->rs_length) { 982 offset -= segment->rs_length; 983 continue; 984 } 985 986 dummy.rs_handle = segment->rs_handle; 987 dummy.rs_length = min_t(u32, length, segment->rs_length) - offset; 988 dummy.rs_offset = segment->rs_offset + offset; 989 990 ret = svc_rdma_build_read_segment(info, &dummy); 991 if (ret < 0) 992 break; 993 994 info->ri_totalbytes += dummy.rs_length; 995 length -= dummy.rs_length; 996 offset = 0; 997 } 998 return ret; 999 } 1000 1001 /** 1002 * svc_rdma_read_call_chunk - Build RDMA Read WQEs to pull a Long Message 1003 * @info: context for RDMA Reads 1004 * 1005 * Return values: 1006 * %0: RDMA Read WQEs were successfully built 1007 * %-EINVAL: there were not enough resources to finish 1008 * %-ENOMEM: rdma_rw context pool was exhausted, 1009 * %-ENOTCONN: posting failed (connection is lost), 1010 * %-EIO: rdma_rw initialization failed (DMA mapping, etc). 1011 */ 1012 static int svc_rdma_read_call_chunk(struct svc_rdma_read_info *info) 1013 { 1014 struct svc_rdma_recv_ctxt *head = info->ri_readctxt; 1015 const struct svc_rdma_chunk *call_chunk = 1016 pcl_first_chunk(&head->rc_call_pcl); 1017 const struct svc_rdma_pcl *pcl = &head->rc_read_pcl; 1018 struct svc_rdma_chunk *chunk, *next; 1019 unsigned int start, length; 1020 int ret; 1021 1022 if (pcl_is_empty(pcl)) 1023 return svc_rdma_build_read_chunk(info, call_chunk); 1024 1025 start = 0; 1026 chunk = pcl_first_chunk(pcl); 1027 length = chunk->ch_position; 1028 ret = svc_rdma_read_chunk_range(info, call_chunk, start, length); 1029 if (ret < 0) 1030 return ret; 1031 1032 pcl_for_each_chunk(chunk, pcl) { 1033 ret = svc_rdma_build_read_chunk(info, chunk); 1034 if (ret < 0) 1035 return ret; 1036 1037 next = pcl_next_chunk(pcl, chunk); 1038 if (!next) 1039 break; 1040 1041 start += length; 1042 length = next->ch_position - info->ri_totalbytes; 1043 ret = svc_rdma_read_chunk_range(info, call_chunk, 1044 start, length); 1045 if (ret < 0) 1046 return ret; 1047 } 1048 1049 start += length; 1050 length = call_chunk->ch_length - start; 1051 return svc_rdma_read_chunk_range(info, call_chunk, start, length); 1052 } 1053 1054 /** 1055 * svc_rdma_read_special - Build RDMA Read WQEs to pull a Long Message 1056 * @info: context for RDMA Reads 1057 * 1058 * The start of the data lands in the first page just after the 1059 * Transport header, and the rest lands in rqstp->rq_arg.pages. 1060 * 1061 * Assumptions: 1062 * - A PZRC is never sent in an RDMA_MSG message, though it's 1063 * allowed by spec. 1064 * 1065 * Return values: 1066 * %0: RDMA Read WQEs were successfully built 1067 * %-EINVAL: client provided too many chunks or segments, 1068 * %-ENOMEM: rdma_rw context pool was exhausted, 1069 * %-ENOTCONN: posting failed (connection is lost), 1070 * %-EIO: rdma_rw initialization failed (DMA mapping, etc). 1071 */ 1072 static noinline int svc_rdma_read_special(struct svc_rdma_read_info *info) 1073 { 1074 struct xdr_buf *buf = &info->ri_rqst->rq_arg; 1075 int ret; 1076 1077 ret = svc_rdma_read_call_chunk(info); 1078 if (ret < 0) 1079 goto out; 1080 1081 buf->len += info->ri_totalbytes; 1082 buf->buflen += info->ri_totalbytes; 1083 1084 buf->head[0].iov_base = page_address(info->ri_rqst->rq_pages[0]); 1085 buf->head[0].iov_len = min_t(size_t, PAGE_SIZE, info->ri_totalbytes); 1086 buf->pages = &info->ri_rqst->rq_pages[1]; 1087 buf->page_len = info->ri_totalbytes - buf->head[0].iov_len; 1088 1089 out: 1090 return ret; 1091 } 1092 1093 /** 1094 * svc_rdma_process_read_list - Pull list of Read chunks from the client 1095 * @rdma: controlling RDMA transport 1096 * @rqstp: set of pages to use as Read sink buffers 1097 * @head: pages under I/O collect here 1098 * 1099 * The RPC/RDMA protocol assumes that the upper layer's XDR decoders 1100 * pull each Read chunk as they decode an incoming RPC message. 1101 * 1102 * On Linux, however, the server needs to have a fully-constructed RPC 1103 * message in rqstp->rq_arg when there is a positive return code from 1104 * ->xpo_recvfrom. So the Read list is safety-checked immediately when 1105 * it is received, then here the whole Read list is pulled all at once. 1106 * The ingress RPC message is fully reconstructed once all associated 1107 * RDMA Reads have completed. 1108 * 1109 * Return values: 1110 * %1: all needed RDMA Reads were posted successfully, 1111 * %-EINVAL: client provided too many chunks or segments, 1112 * %-ENOMEM: rdma_rw context pool was exhausted, 1113 * %-ENOTCONN: posting failed (connection is lost), 1114 * %-EIO: rdma_rw initialization failed (DMA mapping, etc). 1115 */ 1116 int svc_rdma_process_read_list(struct svcxprt_rdma *rdma, 1117 struct svc_rqst *rqstp, 1118 struct svc_rdma_recv_ctxt *head) 1119 { 1120 struct svc_rdma_read_info *info; 1121 struct svc_rdma_chunk_ctxt *cc; 1122 int ret; 1123 1124 info = svc_rdma_read_info_alloc(rdma); 1125 if (!info) 1126 return -ENOMEM; 1127 cc = &info->ri_cc; 1128 info->ri_rqst = rqstp; 1129 info->ri_readctxt = head; 1130 info->ri_pageno = 0; 1131 info->ri_pageoff = 0; 1132 info->ri_totalbytes = 0; 1133 1134 if (pcl_is_empty(&head->rc_call_pcl)) { 1135 if (head->rc_read_pcl.cl_count == 1) 1136 ret = svc_rdma_read_data_item(info); 1137 else 1138 ret = svc_rdma_read_multiple_chunks(info); 1139 } else 1140 ret = svc_rdma_read_special(info); 1141 if (ret < 0) 1142 goto out_err; 1143 1144 trace_svcrdma_post_read_chunk(&cc->cc_cid, cc->cc_sqecount); 1145 init_completion(&cc->cc_done); 1146 ret = svc_rdma_post_chunk_ctxt(cc); 1147 if (ret < 0) 1148 goto out_err; 1149 1150 ret = 1; 1151 wait_for_completion(&cc->cc_done); 1152 if (cc->cc_status != IB_WC_SUCCESS) 1153 ret = -EIO; 1154 1155 /* rq_respages starts after the last arg page */ 1156 rqstp->rq_respages = &rqstp->rq_pages[head->rc_page_count]; 1157 rqstp->rq_next_page = rqstp->rq_respages + 1; 1158 1159 /* Ensure svc_rdma_recv_ctxt_put() does not try to release pages */ 1160 head->rc_page_count = 0; 1161 1162 out_err: 1163 svc_rdma_read_info_free(info); 1164 return ret; 1165 } 1166