net/sunrpc/xprtrdma/svc_rdma_rw.c

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