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 17static void svc_rdma_write_done(struct ib_cq *cq, struct ib_wc *wc); 18static 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 */ 37struct svc_rdma_rw_ctxt { 38 struct list_head rw_list; 39 struct rdma_rw_ctx rw_ctx; 40 unsigned int rw_nents; 41 struct sg_table rw_sg_table; 42 struct scatterlist rw_first_sgl[]; 43}; 44 45static inline struct svc_rdma_rw_ctxt * 46svc_rdma_next_ctxt(struct list_head *list) 47{ 48 return list_first_entry_or_null(list, struct svc_rdma_rw_ctxt, 49 rw_list); 50} 51 52static struct svc_rdma_rw_ctxt * 53svc_rdma_get_rw_ctxt(struct svcxprt_rdma *rdma, unsigned int sges) 54{ 55 struct svc_rdma_rw_ctxt *ctxt; 56 57 spin_lock(&rdma->sc_rw_ctxt_lock); 58 59 ctxt = svc_rdma_next_ctxt(&rdma->sc_rw_ctxts); 60 if (ctxt) { 61 list_del(&ctxt->rw_list); 62 spin_unlock(&rdma->sc_rw_ctxt_lock); 63 } else { 64 spin_unlock(&rdma->sc_rw_ctxt_lock); 65 ctxt = kmalloc(struct_size(ctxt, rw_first_sgl, SG_CHUNK_SIZE), 66 GFP_KERNEL); 67 if (!ctxt) 68 goto out_noctx; 69 INIT_LIST_HEAD(&ctxt->rw_list); 70 } 71 72 ctxt->rw_sg_table.sgl = ctxt->rw_first_sgl; 73 if (sg_alloc_table_chained(&ctxt->rw_sg_table, sges, 74 ctxt->rw_sg_table.sgl, 75 SG_CHUNK_SIZE)) 76 goto out_free; 77 return ctxt; 78 79out_free: 80 kfree(ctxt); 81out_noctx: 82 trace_svcrdma_no_rwctx_err(rdma, sges); 83 return NULL; 84} 85 86static void svc_rdma_put_rw_ctxt(struct svcxprt_rdma *rdma, 87 struct svc_rdma_rw_ctxt *ctxt) 88{ 89 sg_free_table_chained(&ctxt->rw_sg_table, SG_CHUNK_SIZE); 90 91 spin_lock(&rdma->sc_rw_ctxt_lock); 92 list_add(&ctxt->rw_list, &rdma->sc_rw_ctxts); 93 spin_unlock(&rdma->sc_rw_ctxt_lock); 94} 95 96/** 97 * svc_rdma_destroy_rw_ctxts - Free accumulated R/W contexts 98 * @rdma: transport about to be destroyed 99 * 100 */ 101void svc_rdma_destroy_rw_ctxts(struct svcxprt_rdma *rdma) 102{ 103 struct svc_rdma_rw_ctxt *ctxt; 104 105 while ((ctxt = svc_rdma_next_ctxt(&rdma->sc_rw_ctxts)) != NULL) { 106 list_del(&ctxt->rw_list); 107 kfree(ctxt); 108 } 109} 110 111/** 112 * svc_rdma_rw_ctx_init - Prepare a R/W context for I/O 113 * @rdma: controlling transport instance 114 * @ctxt: R/W context to prepare 115 * @offset: RDMA offset 116 * @handle: RDMA tag/handle 117 * @direction: I/O direction 118 * 119 * Returns on success, the number of WQEs that will be needed 120 * on the workqueue, or a negative errno. 121 */ 122static int svc_rdma_rw_ctx_init(struct svcxprt_rdma *rdma, 123 struct svc_rdma_rw_ctxt *ctxt, 124 u64 offset, u32 handle, 125 enum dma_data_direction direction) 126{ 127 int ret; 128 129 ret = rdma_rw_ctx_init(&ctxt->rw_ctx, rdma->sc_qp, rdma->sc_port_num, 130 ctxt->rw_sg_table.sgl, ctxt->rw_nents, 131 0, offset, handle, direction); 132 if (unlikely(ret < 0)) { 133 svc_rdma_put_rw_ctxt(rdma, ctxt); 134 trace_svcrdma_dma_map_rw_err(rdma, ctxt->rw_nents, ret); 135 } 136 return ret; 137} 138 139/* A chunk context tracks all I/O for moving one Read or Write 140 * chunk. This is a set of rdma_rw's that handle data movement 141 * for all segments of one chunk. 142 * 143 * These are small, acquired with a single allocator call, and 144 * no more than one is needed per chunk. They are allocated on 145 * demand, and not cached. 146 */ 147struct svc_rdma_chunk_ctxt { 148 struct rpc_rdma_cid cc_cid; 149 struct ib_cqe cc_cqe; 150 struct svcxprt_rdma *cc_rdma; 151 struct list_head cc_rwctxts; 152 int cc_sqecount; 153}; 154 155static void svc_rdma_cc_cid_init(struct svcxprt_rdma *rdma, 156 struct rpc_rdma_cid *cid) 157{ 158 cid->ci_queue_id = rdma->sc_sq_cq->res.id; 159 cid->ci_completion_id = atomic_inc_return(&rdma->sc_completion_ids); 160} 161 162static void svc_rdma_cc_init(struct svcxprt_rdma *rdma, 163 struct svc_rdma_chunk_ctxt *cc) 164{ 165 svc_rdma_cc_cid_init(rdma, &cc->cc_cid); 166 cc->cc_rdma = rdma; 167 168 INIT_LIST_HEAD(&cc->cc_rwctxts); 169 cc->cc_sqecount = 0; 170} 171 172static void svc_rdma_cc_release(struct svc_rdma_chunk_ctxt *cc, 173 enum dma_data_direction dir) 174{ 175 struct svcxprt_rdma *rdma = cc->cc_rdma; 176 struct svc_rdma_rw_ctxt *ctxt; 177 178 while ((ctxt = svc_rdma_next_ctxt(&cc->cc_rwctxts)) != NULL) { 179 list_del(&ctxt->rw_list); 180 181 rdma_rw_ctx_destroy(&ctxt->rw_ctx, rdma->sc_qp, 182 rdma->sc_port_num, ctxt->rw_sg_table.sgl, 183 ctxt->rw_nents, dir); 184 svc_rdma_put_rw_ctxt(rdma, ctxt); 185 } 186} 187 188/* State for sending a Write or Reply chunk. 189 * - Tracks progress of writing one chunk over all its segments 190 * - Stores arguments for the SGL constructor functions 191 */ 192struct svc_rdma_write_info { 193 /* write state of this chunk */ 194 unsigned int wi_seg_off; 195 unsigned int wi_seg_no; 196 unsigned int wi_nsegs; 197 __be32 *wi_segs; 198 199 /* SGL constructor arguments */ 200 struct xdr_buf *wi_xdr; 201 unsigned char *wi_base; 202 unsigned int wi_next_off; 203 204 struct svc_rdma_chunk_ctxt wi_cc; 205}; 206 207static struct svc_rdma_write_info * 208svc_rdma_write_info_alloc(struct svcxprt_rdma *rdma, __be32 *chunk) 209{ 210 struct svc_rdma_write_info *info; 211 212 info = kmalloc(sizeof(*info), GFP_KERNEL); 213 if (!info) 214 return info; 215 216 info->wi_seg_off = 0; 217 info->wi_seg_no = 0; 218 info->wi_nsegs = be32_to_cpup(++chunk); 219 info->wi_segs = ++chunk; 220 svc_rdma_cc_init(rdma, &info->wi_cc); 221 info->wi_cc.cc_cqe.done = svc_rdma_write_done; 222 return info; 223} 224 225static void svc_rdma_write_info_free(struct svc_rdma_write_info *info) 226{ 227 svc_rdma_cc_release(&info->wi_cc, DMA_TO_DEVICE); 228 kfree(info); 229} 230 231/** 232 * svc_rdma_write_done - Write chunk completion 233 * @cq: controlling Completion Queue 234 * @wc: Work Completion 235 * 236 * Pages under I/O are freed by a subsequent Send completion. 237 */ 238static void svc_rdma_write_done(struct ib_cq *cq, struct ib_wc *wc) 239{ 240 struct ib_cqe *cqe = wc->wr_cqe; 241 struct svc_rdma_chunk_ctxt *cc = 242 container_of(cqe, struct svc_rdma_chunk_ctxt, cc_cqe); 243 struct svcxprt_rdma *rdma = cc->cc_rdma; 244 struct svc_rdma_write_info *info = 245 container_of(cc, struct svc_rdma_write_info, wi_cc); 246 247 trace_svcrdma_wc_write(wc, &cc->cc_cid); 248 249 atomic_add(cc->cc_sqecount, &rdma->sc_sq_avail); 250 wake_up(&rdma->sc_send_wait); 251 252 if (unlikely(wc->status != IB_WC_SUCCESS)) 253 set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags); 254 255 svc_rdma_write_info_free(info); 256} 257 258/* State for pulling a Read chunk. 259 */ 260struct svc_rdma_read_info { 261 struct svc_rdma_recv_ctxt *ri_readctxt; 262 unsigned int ri_position; 263 unsigned int ri_pageno; 264 unsigned int ri_pageoff; 265 unsigned int ri_chunklen; 266 267 struct svc_rdma_chunk_ctxt ri_cc; 268}; 269 270static struct svc_rdma_read_info * 271svc_rdma_read_info_alloc(struct svcxprt_rdma *rdma) 272{ 273 struct svc_rdma_read_info *info; 274 275 info = kmalloc(sizeof(*info), GFP_KERNEL); 276 if (!info) 277 return info; 278 279 svc_rdma_cc_init(rdma, &info->ri_cc); 280 info->ri_cc.cc_cqe.done = svc_rdma_wc_read_done; 281 return info; 282} 283 284static void svc_rdma_read_info_free(struct svc_rdma_read_info *info) 285{ 286 svc_rdma_cc_release(&info->ri_cc, DMA_FROM_DEVICE); 287 kfree(info); 288} 289 290/** 291 * svc_rdma_wc_read_done - Handle completion of an RDMA Read ctx 292 * @cq: controlling Completion Queue 293 * @wc: Work Completion 294 * 295 */ 296static void svc_rdma_wc_read_done(struct ib_cq *cq, struct ib_wc *wc) 297{ 298 struct ib_cqe *cqe = wc->wr_cqe; 299 struct svc_rdma_chunk_ctxt *cc = 300 container_of(cqe, struct svc_rdma_chunk_ctxt, cc_cqe); 301 struct svcxprt_rdma *rdma = cc->cc_rdma; 302 struct svc_rdma_read_info *info = 303 container_of(cc, struct svc_rdma_read_info, ri_cc); 304 305 trace_svcrdma_wc_read(wc, &cc->cc_cid); 306 307 atomic_add(cc->cc_sqecount, &rdma->sc_sq_avail); 308 wake_up(&rdma->sc_send_wait); 309 310 if (unlikely(wc->status != IB_WC_SUCCESS)) { 311 set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags); 312 svc_rdma_recv_ctxt_put(rdma, info->ri_readctxt); 313 } else { 314 spin_lock(&rdma->sc_rq_dto_lock); 315 list_add_tail(&info->ri_readctxt->rc_list, 316 &rdma->sc_read_complete_q); 317 /* Note the unlock pairs with the smp_rmb in svc_xprt_ready: */ 318 set_bit(XPT_DATA, &rdma->sc_xprt.xpt_flags); 319 spin_unlock(&rdma->sc_rq_dto_lock); 320 321 svc_xprt_enqueue(&rdma->sc_xprt); 322 } 323 324 svc_rdma_read_info_free(info); 325} 326 327/* This function sleeps when the transport's Send Queue is congested. 328 * 329 * Assumptions: 330 * - If ib_post_send() succeeds, only one completion is expected, 331 * even if one or more WRs are flushed. This is true when posting 332 * an rdma_rw_ctx or when posting a single signaled WR. 333 */ 334static int svc_rdma_post_chunk_ctxt(struct svc_rdma_chunk_ctxt *cc) 335{ 336 struct svcxprt_rdma *rdma = cc->cc_rdma; 337 struct svc_xprt *xprt = &rdma->sc_xprt; 338 struct ib_send_wr *first_wr; 339 const struct ib_send_wr *bad_wr; 340 struct list_head *tmp; 341 struct ib_cqe *cqe; 342 int ret; 343 344 if (cc->cc_sqecount > rdma->sc_sq_depth) 345 return -EINVAL; 346 347 first_wr = NULL; 348 cqe = &cc->cc_cqe; 349 list_for_each(tmp, &cc->cc_rwctxts) { 350 struct svc_rdma_rw_ctxt *ctxt; 351 352 ctxt = list_entry(tmp, struct svc_rdma_rw_ctxt, rw_list); 353 first_wr = rdma_rw_ctx_wrs(&ctxt->rw_ctx, rdma->sc_qp, 354 rdma->sc_port_num, cqe, first_wr); 355 cqe = NULL; 356 } 357 358 do { 359 if (atomic_sub_return(cc->cc_sqecount, 360 &rdma->sc_sq_avail) > 0) { 361 trace_svcrdma_post_chunk(&cc->cc_cid, cc->cc_sqecount); 362 ret = ib_post_send(rdma->sc_qp, first_wr, &bad_wr); 363 if (ret) 364 break; 365 return 0; 366 } 367 368 trace_svcrdma_sq_full(rdma); 369 atomic_add(cc->cc_sqecount, &rdma->sc_sq_avail); 370 wait_event(rdma->sc_send_wait, 371 atomic_read(&rdma->sc_sq_avail) > cc->cc_sqecount); 372 trace_svcrdma_sq_retry(rdma); 373 } while (1); 374 375 trace_svcrdma_sq_post_err(rdma, ret); 376 set_bit(XPT_CLOSE, &xprt->xpt_flags); 377 378 /* If even one was posted, there will be a completion. */ 379 if (bad_wr != first_wr) 380 return 0; 381 382 atomic_add(cc->cc_sqecount, &rdma->sc_sq_avail); 383 wake_up(&rdma->sc_send_wait); 384 return -ENOTCONN; 385} 386 387/* Build and DMA-map an SGL that covers one kvec in an xdr_buf 388 */ 389static void svc_rdma_vec_to_sg(struct svc_rdma_write_info *info, 390 unsigned int len, 391 struct svc_rdma_rw_ctxt *ctxt) 392{ 393 struct scatterlist *sg = ctxt->rw_sg_table.sgl; 394 395 sg_set_buf(&sg[0], info->wi_base, len); 396 info->wi_base += len; 397 398 ctxt->rw_nents = 1; 399} 400 401/* Build and DMA-map an SGL that covers part of an xdr_buf's pagelist. 402 */ 403static void svc_rdma_pagelist_to_sg(struct svc_rdma_write_info *info, 404 unsigned int remaining, 405 struct svc_rdma_rw_ctxt *ctxt) 406{ 407 unsigned int sge_no, sge_bytes, page_off, page_no; 408 struct xdr_buf *xdr = info->wi_xdr; 409 struct scatterlist *sg; 410 struct page **page; 411 412 page_off = info->wi_next_off + xdr->page_base; 413 page_no = page_off >> PAGE_SHIFT; 414 page_off = offset_in_page(page_off); 415 page = xdr->pages + page_no; 416 info->wi_next_off += remaining; 417 sg = ctxt->rw_sg_table.sgl; 418 sge_no = 0; 419 do { 420 sge_bytes = min_t(unsigned int, remaining, 421 PAGE_SIZE - page_off); 422 sg_set_page(sg, *page, sge_bytes, page_off); 423 424 remaining -= sge_bytes; 425 sg = sg_next(sg); 426 page_off = 0; 427 sge_no++; 428 page++; 429 } while (remaining); 430 431 ctxt->rw_nents = sge_no; 432} 433 434/* Construct RDMA Write WRs to send a portion of an xdr_buf containing 435 * an RPC Reply. 436 */ 437static int 438svc_rdma_build_writes(struct svc_rdma_write_info *info, 439 void (*constructor)(struct svc_rdma_write_info *info, 440 unsigned int len, 441 struct svc_rdma_rw_ctxt *ctxt), 442 unsigned int remaining) 443{ 444 struct svc_rdma_chunk_ctxt *cc = &info->wi_cc; 445 struct svcxprt_rdma *rdma = cc->cc_rdma; 446 struct svc_rdma_rw_ctxt *ctxt; 447 __be32 *seg; 448 int ret; 449 450 seg = info->wi_segs + info->wi_seg_no * rpcrdma_segment_maxsz; 451 do { 452 unsigned int write_len; 453 u32 handle, length; 454 u64 offset; 455 456 if (info->wi_seg_no >= info->wi_nsegs) 457 goto out_overflow; 458 459 xdr_decode_rdma_segment(seg, &handle, &length, &offset); 460 offset += info->wi_seg_off; 461 462 write_len = min(remaining, length - info->wi_seg_off); 463 ctxt = svc_rdma_get_rw_ctxt(rdma, 464 (write_len >> PAGE_SHIFT) + 2); 465 if (!ctxt) 466 return -ENOMEM; 467 468 constructor(info, write_len, ctxt); 469 ret = svc_rdma_rw_ctx_init(rdma, ctxt, offset, handle, 470 DMA_TO_DEVICE); 471 if (ret < 0) 472 return -EIO; 473 474 trace_svcrdma_send_wseg(handle, write_len, offset); 475 476 list_add(&ctxt->rw_list, &cc->cc_rwctxts); 477 cc->cc_sqecount += ret; 478 if (write_len == length - info->wi_seg_off) { 479 seg += 4; 480 info->wi_seg_no++; 481 info->wi_seg_off = 0; 482 } else { 483 info->wi_seg_off += write_len; 484 } 485 remaining -= write_len; 486 } while (remaining); 487 488 return 0; 489 490out_overflow: 491 trace_svcrdma_small_wrch_err(rdma, remaining, info->wi_seg_no, 492 info->wi_nsegs); 493 return -E2BIG; 494} 495 496/* Send one of an xdr_buf's kvecs by itself. To send a Reply 497 * chunk, the whole RPC Reply is written back to the client. 498 * This function writes either the head or tail of the xdr_buf 499 * containing the Reply. 500 */ 501static int svc_rdma_send_xdr_kvec(struct svc_rdma_write_info *info, 502 struct kvec *vec) 503{ 504 info->wi_base = vec->iov_base; 505 return svc_rdma_build_writes(info, svc_rdma_vec_to_sg, 506 vec->iov_len); 507} 508 509/* Send an xdr_buf's page list by itself. A Write chunk is just 510 * the page list. A Reply chunk is @xdr's head, page list, and 511 * tail. This function is shared between the two types of chunk. 512 */ 513static int svc_rdma_send_xdr_pagelist(struct svc_rdma_write_info *info, 514 struct xdr_buf *xdr, 515 unsigned int offset, 516 unsigned long length) 517{ 518 info->wi_xdr = xdr; 519 info->wi_next_off = offset - xdr->head[0].iov_len; 520 return svc_rdma_build_writes(info, svc_rdma_pagelist_to_sg, 521 length); 522} 523 524/** 525 * svc_rdma_send_write_chunk - Write all segments in a Write chunk 526 * @rdma: controlling RDMA transport 527 * @wr_ch: Write chunk provided by client 528 * @xdr: xdr_buf containing the data payload 529 * @offset: payload's byte offset in @xdr 530 * @length: size of payload, in bytes 531 * 532 * Returns a non-negative number of bytes the chunk consumed, or 533 * %-E2BIG if the payload was larger than the Write chunk, 534 * %-EINVAL if client provided too many segments, 535 * %-ENOMEM if rdma_rw context pool was exhausted, 536 * %-ENOTCONN if posting failed (connection is lost), 537 * %-EIO if rdma_rw initialization failed (DMA mapping, etc). 538 */ 539int svc_rdma_send_write_chunk(struct svcxprt_rdma *rdma, __be32 *wr_ch, 540 struct xdr_buf *xdr, 541 unsigned int offset, unsigned long length) 542{ 543 struct svc_rdma_write_info *info; 544 int ret; 545 546 if (!length) 547 return 0; 548 549 info = svc_rdma_write_info_alloc(rdma, wr_ch); 550 if (!info) 551 return -ENOMEM; 552 553 ret = svc_rdma_send_xdr_pagelist(info, xdr, offset, length); 554 if (ret < 0) 555 goto out_err; 556 557 ret = svc_rdma_post_chunk_ctxt(&info->wi_cc); 558 if (ret < 0) 559 goto out_err; 560 561 trace_svcrdma_send_write_chunk(xdr->page_len); 562 return length; 563 564out_err: 565 svc_rdma_write_info_free(info); 566 return ret; 567} 568 569/** 570 * svc_rdma_send_reply_chunk - Write all segments in the Reply chunk 571 * @rdma: controlling RDMA transport 572 * @rctxt: Write and Reply chunks from client 573 * @xdr: xdr_buf containing an RPC Reply 574 * 575 * Returns a non-negative number of bytes the chunk consumed, or 576 * %-E2BIG if the payload was larger than the Reply chunk, 577 * %-EINVAL if client provided too many segments, 578 * %-ENOMEM if rdma_rw context pool was exhausted, 579 * %-ENOTCONN if posting failed (connection is lost), 580 * %-EIO if rdma_rw initialization failed (DMA mapping, etc). 581 */ 582int svc_rdma_send_reply_chunk(struct svcxprt_rdma *rdma, 583 const struct svc_rdma_recv_ctxt *rctxt, 584 struct xdr_buf *xdr) 585{ 586 struct svc_rdma_write_info *info; 587 int consumed, ret; 588 589 info = svc_rdma_write_info_alloc(rdma, rctxt->rc_reply_chunk); 590 if (!info) 591 return -ENOMEM; 592 593 ret = svc_rdma_send_xdr_kvec(info, &xdr->head[0]); 594 if (ret < 0) 595 goto out_err; 596 consumed = xdr->head[0].iov_len; 597 598 /* Send the page list in the Reply chunk only if the 599 * client did not provide Write chunks. 600 */ 601 if (!rctxt->rc_write_list && xdr->page_len) { 602 ret = svc_rdma_send_xdr_pagelist(info, xdr, 603 xdr->head[0].iov_len, 604 xdr->page_len); 605 if (ret < 0) 606 goto out_err; 607 consumed += xdr->page_len; 608 } 609 610 if (xdr->tail[0].iov_len) { 611 ret = svc_rdma_send_xdr_kvec(info, &xdr->tail[0]); 612 if (ret < 0) 613 goto out_err; 614 consumed += xdr->tail[0].iov_len; 615 } 616 617 ret = svc_rdma_post_chunk_ctxt(&info->wi_cc); 618 if (ret < 0) 619 goto out_err; 620 621 trace_svcrdma_send_reply_chunk(consumed); 622 return consumed; 623 624out_err: 625 svc_rdma_write_info_free(info); 626 return ret; 627} 628 629static int svc_rdma_build_read_segment(struct svc_rdma_read_info *info, 630 struct svc_rqst *rqstp, 631 u32 rkey, u32 len, u64 offset) 632{ 633 struct svc_rdma_recv_ctxt *head = info->ri_readctxt; 634 struct svc_rdma_chunk_ctxt *cc = &info->ri_cc; 635 struct svc_rdma_rw_ctxt *ctxt; 636 unsigned int sge_no, seg_len; 637 struct scatterlist *sg; 638 int ret; 639 640 sge_no = PAGE_ALIGN(info->ri_pageoff + len) >> PAGE_SHIFT; 641 ctxt = svc_rdma_get_rw_ctxt(cc->cc_rdma, sge_no); 642 if (!ctxt) 643 return -ENOMEM; 644 ctxt->rw_nents = sge_no; 645 646 sg = ctxt->rw_sg_table.sgl; 647 for (sge_no = 0; sge_no < ctxt->rw_nents; sge_no++) { 648 seg_len = min_t(unsigned int, len, 649 PAGE_SIZE - info->ri_pageoff); 650 651 head->rc_arg.pages[info->ri_pageno] = 652 rqstp->rq_pages[info->ri_pageno]; 653 if (!info->ri_pageoff) 654 head->rc_page_count++; 655 656 sg_set_page(sg, rqstp->rq_pages[info->ri_pageno], 657 seg_len, info->ri_pageoff); 658 sg = sg_next(sg); 659 660 info->ri_pageoff += seg_len; 661 if (info->ri_pageoff == PAGE_SIZE) { 662 info->ri_pageno++; 663 info->ri_pageoff = 0; 664 } 665 len -= seg_len; 666 667 /* Safety check */ 668 if (len && 669 &rqstp->rq_pages[info->ri_pageno + 1] > rqstp->rq_page_end) 670 goto out_overrun; 671 } 672 673 ret = svc_rdma_rw_ctx_init(cc->cc_rdma, ctxt, offset, rkey, 674 DMA_FROM_DEVICE); 675 if (ret < 0) 676 return -EIO; 677 678 list_add(&ctxt->rw_list, &cc->cc_rwctxts); 679 cc->cc_sqecount += ret; 680 return 0; 681 682out_overrun: 683 trace_svcrdma_page_overrun_err(cc->cc_rdma, rqstp, info->ri_pageno); 684 return -EINVAL; 685} 686 687/* Walk the segments in the Read chunk starting at @p and construct 688 * RDMA Read operations to pull the chunk to the server. 689 */ 690static int svc_rdma_build_read_chunk(struct svc_rqst *rqstp, 691 struct svc_rdma_read_info *info, 692 __be32 *p) 693{ 694 int ret; 695 696 ret = -EINVAL; 697 info->ri_chunklen = 0; 698 while (*p++ != xdr_zero && be32_to_cpup(p++) == info->ri_position) { 699 u32 handle, length; 700 u64 offset; 701 702 p = xdr_decode_rdma_segment(p, &handle, &length, &offset); 703 ret = svc_rdma_build_read_segment(info, rqstp, handle, length, 704 offset); 705 if (ret < 0) 706 break; 707 708 trace_svcrdma_send_rseg(handle, length, offset); 709 info->ri_chunklen += length; 710 } 711 712 return ret; 713} 714 715/* Construct RDMA Reads to pull over a normal Read chunk. The chunk 716 * data lands in the page list of head->rc_arg.pages. 717 * 718 * Currently NFSD does not look at the head->rc_arg.tail[0] iovec. 719 * Therefore, XDR round-up of the Read chunk and trailing 720 * inline content must both be added at the end of the pagelist. 721 */ 722static int svc_rdma_build_normal_read_chunk(struct svc_rqst *rqstp, 723 struct svc_rdma_read_info *info, 724 __be32 *p) 725{ 726 struct svc_rdma_recv_ctxt *head = info->ri_readctxt; 727 int ret; 728 729 ret = svc_rdma_build_read_chunk(rqstp, info, p); 730 if (ret < 0) 731 goto out; 732 733 trace_svcrdma_send_read_chunk(info->ri_chunklen, info->ri_position); 734 735 head->rc_hdr_count = 0; 736 737 /* Split the Receive buffer between the head and tail 738 * buffers at Read chunk's position. XDR roundup of the 739 * chunk is not included in either the pagelist or in 740 * the tail. 741 */ 742 head->rc_arg.tail[0].iov_base = 743 head->rc_arg.head[0].iov_base + info->ri_position; 744 head->rc_arg.tail[0].iov_len = 745 head->rc_arg.head[0].iov_len - info->ri_position; 746 head->rc_arg.head[0].iov_len = info->ri_position; 747 748 /* Read chunk may need XDR roundup (see RFC 8166, s. 3.4.5.2). 749 * 750 * If the client already rounded up the chunk length, the 751 * length does not change. Otherwise, the length of the page 752 * list is increased to include XDR round-up. 753 * 754 * Currently these chunks always start at page offset 0, 755 * thus the rounded-up length never crosses a page boundary. 756 */ 757 info->ri_chunklen = XDR_QUADLEN(info->ri_chunklen) << 2; 758 759 head->rc_arg.page_len = info->ri_chunklen; 760 head->rc_arg.len += info->ri_chunklen; 761 head->rc_arg.buflen += info->ri_chunklen; 762 763out: 764 return ret; 765} 766 767/* Construct RDMA Reads to pull over a Position Zero Read chunk. 768 * The start of the data lands in the first page just after 769 * the Transport header, and the rest lands in the page list of 770 * head->rc_arg.pages. 771 * 772 * Assumptions: 773 * - A PZRC has an XDR-aligned length (no implicit round-up). 774 * - There can be no trailing inline content (IOW, we assume 775 * a PZRC is never sent in an RDMA_MSG message, though it's 776 * allowed by spec). 777 */ 778static int svc_rdma_build_pz_read_chunk(struct svc_rqst *rqstp, 779 struct svc_rdma_read_info *info, 780 __be32 *p) 781{ 782 struct svc_rdma_recv_ctxt *head = info->ri_readctxt; 783 int ret; 784 785 ret = svc_rdma_build_read_chunk(rqstp, info, p); 786 if (ret < 0) 787 goto out; 788 789 trace_svcrdma_send_pzr(info->ri_chunklen); 790 791 head->rc_arg.len += info->ri_chunklen; 792 head->rc_arg.buflen += info->ri_chunklen; 793 794 head->rc_hdr_count = 1; 795 head->rc_arg.head[0].iov_base = page_address(head->rc_pages[0]); 796 head->rc_arg.head[0].iov_len = min_t(size_t, PAGE_SIZE, 797 info->ri_chunklen); 798 799 head->rc_arg.page_len = info->ri_chunklen - 800 head->rc_arg.head[0].iov_len; 801 802out: 803 return ret; 804} 805 806/* Pages under I/O have been copied to head->rc_pages. Ensure they 807 * are not released by svc_xprt_release() until the I/O is complete. 808 * 809 * This has to be done after all Read WRs are constructed to properly 810 * handle a page that is part of I/O on behalf of two different RDMA 811 * segments. 812 * 813 * Do this only if I/O has been posted. Otherwise, we do indeed want 814 * svc_xprt_release() to clean things up properly. 815 */ 816static void svc_rdma_save_io_pages(struct svc_rqst *rqstp, 817 const unsigned int start, 818 const unsigned int num_pages) 819{ 820 unsigned int i; 821 822 for (i = start; i < num_pages + start; i++) 823 rqstp->rq_pages[i] = NULL; 824} 825 826/** 827 * svc_rdma_recv_read_chunk - Pull a Read chunk from the client 828 * @rdma: controlling RDMA transport 829 * @rqstp: set of pages to use as Read sink buffers 830 * @head: pages under I/O collect here 831 * @p: pointer to start of Read chunk 832 * 833 * Returns: 834 * %0 if all needed RDMA Reads were posted successfully, 835 * %-EINVAL if client provided too many segments, 836 * %-ENOMEM if rdma_rw context pool was exhausted, 837 * %-ENOTCONN if posting failed (connection is lost), 838 * %-EIO if rdma_rw initialization failed (DMA mapping, etc). 839 * 840 * Assumptions: 841 * - All Read segments in @p have the same Position value. 842 */ 843int svc_rdma_recv_read_chunk(struct svcxprt_rdma *rdma, struct svc_rqst *rqstp, 844 struct svc_rdma_recv_ctxt *head, __be32 *p) 845{ 846 struct svc_rdma_read_info *info; 847 int ret; 848 849 /* The request (with page list) is constructed in 850 * head->rc_arg. Pages involved with RDMA Read I/O are 851 * transferred there. 852 */ 853 head->rc_arg.head[0] = rqstp->rq_arg.head[0]; 854 head->rc_arg.tail[0] = rqstp->rq_arg.tail[0]; 855 head->rc_arg.pages = head->rc_pages; 856 head->rc_arg.page_base = 0; 857 head->rc_arg.page_len = 0; 858 head->rc_arg.len = rqstp->rq_arg.len; 859 head->rc_arg.buflen = rqstp->rq_arg.buflen; 860 861 info = svc_rdma_read_info_alloc(rdma); 862 if (!info) 863 return -ENOMEM; 864 info->ri_readctxt = head; 865 info->ri_pageno = 0; 866 info->ri_pageoff = 0; 867 868 info->ri_position = be32_to_cpup(p + 1); 869 if (info->ri_position) 870 ret = svc_rdma_build_normal_read_chunk(rqstp, info, p); 871 else 872 ret = svc_rdma_build_pz_read_chunk(rqstp, info, p); 873 if (ret < 0) 874 goto out_err; 875 876 ret = svc_rdma_post_chunk_ctxt(&info->ri_cc); 877 if (ret < 0) 878 goto out_err; 879 svc_rdma_save_io_pages(rqstp, 0, head->rc_page_count); 880 return 0; 881 882out_err: 883 svc_rdma_read_info_free(info); 884 return ret; 885} 886