net/sunrpc/xprtrdma/rpc_rdma.c

   1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
   2 /*
   3  * Copyright (c) 2014-2017 Oracle.  All rights reserved.
   4  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
   5  *
   6  * This software is available to you under a choice of one of two
   7  * licenses.  You may choose to be licensed under the terms of the GNU
   8  * General Public License (GPL) Version 2, available from the file
   9  * COPYING in the main directory of this source tree, or the BSD-type
  10  * license below:
  11  *
  12  * Redistribution and use in source and binary forms, with or without
  13  * modification, are permitted provided that the following conditions
  14  * are met:
  15  *
  16  *      Redistributions of source code must retain the above copyright
  17  *      notice, this list of conditions and the following disclaimer.
  18  *
  19  *      Redistributions in binary form must reproduce the above
  20  *      copyright notice, this list of conditions and the following
  21  *      disclaimer in the documentation and/or other materials provided
  22  *      with the distribution.
  23  *
  24  *      Neither the name of the Network Appliance, Inc. nor the names of
  25  *      its contributors may be used to endorse or promote products
  26  *      derived from this software without specific prior written
  27  *      permission.
  28  *
  29  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  30  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  31  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  32  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  33  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  34  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  35  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  36  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  37  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  38  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  39  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  40  */
  41
  42 /*
  43  * rpc_rdma.c
  44  *
  45  * This file contains the guts of the RPC RDMA protocol, and
  46  * does marshaling/unmarshaling, etc. It is also where interfacing
  47  * to the Linux RPC framework lives.
  48  */
  49
  50 #include <linux/highmem.h>
  51
  52 #include <linux/sunrpc/svc_rdma.h>
  53
  54 #include "xprt_rdma.h"
  55 #include <trace/events/rpcrdma.h>
  56
  57 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  58 # define RPCDBG_FACILITY        RPCDBG_TRANS
  59 #endif
  60
  61 /* Returns size of largest RPC-over-RDMA header in a Call message
  62  *
  63  * The largest Call header contains a full-size Read list and a
  64  * minimal Reply chunk.
  65  */
  66 static unsigned int rpcrdma_max_call_header_size(unsigned int maxsegs)
  67 {
  68         unsigned int size;
  69
  70         /* Fixed header fields and list discriminators */
  71         size = RPCRDMA_HDRLEN_MIN;
  72
  73         /* Maximum Read list size */
  74         size = maxsegs * rpcrdma_readchunk_maxsz * sizeof(__be32);
  75
  76         /* Minimal Read chunk size */
  77         size += sizeof(__be32); /* segment count */
  78         size += rpcrdma_segment_maxsz * sizeof(__be32);
  79         size += sizeof(__be32); /* list discriminator */
  80
  81         dprintk("RPC:       %s: max call header size = %u\n",
  82                 __func__, size);
  83         return size;
  84 }
  85
  86 /* Returns size of largest RPC-over-RDMA header in a Reply message
  87  *
  88  * There is only one Write list or one Reply chunk per Reply
  89  * message.  The larger list is the Write list.
  90  */
  91 static unsigned int rpcrdma_max_reply_header_size(unsigned int maxsegs)
  92 {
  93         unsigned int size;
  94
  95         /* Fixed header fields and list discriminators */
  96         size = RPCRDMA_HDRLEN_MIN;
  97
  98         /* Maximum Write list size */
  99         size = sizeof(__be32);          /* segment count */
 100         size += maxsegs * rpcrdma_segment_maxsz * sizeof(__be32);
 101         size += sizeof(__be32); /* list discriminator */
 102
 103         dprintk("RPC:       %s: max reply header size = %u\n",
 104                 __func__, size);
 105         return size;
 106 }
 107
 108 /**
 109  * rpcrdma_set_max_header_sizes - Initialize inline payload sizes
 110  * @r_xprt: transport instance to initialize
 111  *
 112  * The max_inline fields contain the maximum size of an RPC message
 113  * so the marshaling code doesn't have to repeat this calculation
 114  * for every RPC.
 115  */
 116 void rpcrdma_set_max_header_sizes(struct rpcrdma_xprt *r_xprt)
 117 {
 118         unsigned int maxsegs = r_xprt->rx_ia.ri_max_segs;
 119         struct rpcrdma_ep *ep = &r_xprt->rx_ep;
 120
 121         ep->rep_max_inline_send =
 122                 ep->rep_inline_send - rpcrdma_max_call_header_size(maxsegs);
 123         ep->rep_max_inline_recv =
 124                 ep->rep_inline_recv - rpcrdma_max_reply_header_size(maxsegs);
 125 }
 126
 127 /* The client can send a request inline as long as the RPCRDMA header
 128  * plus the RPC call fit under the transport's inline limit. If the
 129  * combined call message size exceeds that limit, the client must use
 130  * a Read chunk for this operation.
 131  *
 132  * A Read chunk is also required if sending the RPC call inline would
 133  * exceed this device's max_sge limit.
 134  */
 135 static bool rpcrdma_args_inline(struct rpcrdma_xprt *r_xprt,
 136                                 struct rpc_rqst *rqst)
 137 {
 138         struct xdr_buf *xdr = &rqst->rq_snd_buf;
 139         unsigned int count, remaining, offset;
 140
 141         if (xdr->len > r_xprt->rx_ep.rep_max_inline_send)
 142                 return false;
 143
 144         if (xdr->page_len) {
 145                 remaining = xdr->page_len;
 146                 offset = offset_in_page(xdr->page_base);
 147                 count = RPCRDMA_MIN_SEND_SGES;
 148                 while (remaining) {
 149                         remaining -= min_t(unsigned int,
 150                                            PAGE_SIZE - offset, remaining);
 151                         offset = 0;
 152                         if (++count > r_xprt->rx_ia.ri_max_send_sges)
 153                                 return false;
 154                 }
 155         }
 156
 157         return true;
 158 }
 159
 160 /* The client can't know how large the actual reply will be. Thus it
 161  * plans for the largest possible reply for that particular ULP
 162  * operation. If the maximum combined reply message size exceeds that
 163  * limit, the client must provide a write list or a reply chunk for
 164  * this request.
 165  */
 166 static bool rpcrdma_results_inline(struct rpcrdma_xprt *r_xprt,
 167                                    struct rpc_rqst *rqst)
 168 {
 169         return rqst->rq_rcv_buf.buflen <= r_xprt->rx_ep.rep_max_inline_recv;
 170 }
 171
 172 /* The client is required to provide a Reply chunk if the maximum
 173  * size of the non-payload part of the RPC Reply is larger than
 174  * the inline threshold.
 175  */
 176 static bool
 177 rpcrdma_nonpayload_inline(const struct rpcrdma_xprt *r_xprt,
 178                           const struct rpc_rqst *rqst)
 179 {
 180         const struct xdr_buf *buf = &rqst->rq_rcv_buf;
 181
 182         return (buf->head[0].iov_len + buf->tail[0].iov_len) <
 183                 r_xprt->rx_ep.rep_max_inline_recv;
 184 }
 185
 186 /* Split @vec on page boundaries into SGEs. FMR registers pages, not
 187  * a byte range. Other modes coalesce these SGEs into a single MR
 188  * when they can.
 189  *
 190  * Returns pointer to next available SGE, and bumps the total number
 191  * of SGEs consumed.
 192  */
 193 static struct rpcrdma_mr_seg *
 194 rpcrdma_convert_kvec(struct kvec *vec, struct rpcrdma_mr_seg *seg,
 195                      unsigned int *n)
 196 {
 197         u32 remaining, page_offset;
 198         char *base;
 199
 200         base = vec->iov_base;
 201         page_offset = offset_in_page(base);
 202         remaining = vec->iov_len;
 203         while (remaining) {
 204                 seg->mr_page = NULL;
 205                 seg->mr_offset = base;
 206                 seg->mr_len = min_t(u32, PAGE_SIZE - page_offset, remaining);
 207                 remaining -= seg->mr_len;
 208                 base += seg->mr_len;
 209                 ++seg;
 210                 ++(*n);
 211                 page_offset = 0;
 212         }
 213         return seg;
 214 }
 215
 216 /* Convert @xdrbuf into SGEs no larger than a page each. As they
 217  * are registered, these SGEs are then coalesced into RDMA segments
 218  * when the selected memreg mode supports it.
 219  *
 220  * Returns positive number of SGEs consumed, or a negative errno.
 221  */
 222
 223 static int
 224 rpcrdma_convert_iovs(struct rpcrdma_xprt *r_xprt, struct xdr_buf *xdrbuf,
 225                      unsigned int pos, enum rpcrdma_chunktype type,
 226                      struct rpcrdma_mr_seg *seg)
 227 {
 228         unsigned long page_base;
 229         unsigned int len, n;
 230         struct page **ppages;
 231
 232         n = 0;
 233         if (pos == 0)
 234                 seg = rpcrdma_convert_kvec(&xdrbuf->head[0], seg, &n);
 235
 236         len = xdrbuf->page_len;
 237         ppages = xdrbuf->pages + (xdrbuf->page_base >> PAGE_SHIFT);
 238         page_base = offset_in_page(xdrbuf->page_base);
 239         while (len) {
 240                 /* ACL likes to be lazy in allocating pages - ACLs
 241                  * are small by default but can get huge.
 242                  */
 243                 if (unlikely(xdrbuf->flags & XDRBUF_SPARSE_PAGES)) {
 244                         if (!*ppages)
 245                                 *ppages = alloc_page(GFP_NOWAIT | __GFP_NOWARN);
 246                         if (!*ppages)
 247                                 return -ENOBUFS;
 248                 }
 249                 seg->mr_page = *ppages;
 250                 seg->mr_offset = (char *)page_base;
 251                 seg->mr_len = min_t(u32, PAGE_SIZE - page_base, len);
 252                 len -= seg->mr_len;
 253                 ++ppages;
 254                 ++seg;
 255                 ++n;
 256                 page_base = 0;
 257         }
 258
 259         /* When encoding a Read chunk, the tail iovec contains an
 260          * XDR pad and may be omitted.
 261          */
 262         if (type == rpcrdma_readch && r_xprt->rx_ia.ri_implicit_roundup)
 263                 goto out;
 264
 265         /* When encoding a Write chunk, some servers need to see an
 266          * extra segment for non-XDR-aligned Write chunks. The upper
 267          * layer provides space in the tail iovec that may be used
 268          * for this purpose.
 269          */
 270         if (type == rpcrdma_writech && r_xprt->rx_ia.ri_implicit_roundup)
 271                 goto out;
 272
 273         if (xdrbuf->tail[0].iov_len)
 274                 seg = rpcrdma_convert_kvec(&xdrbuf->tail[0], seg, &n);
 275
 276 out:
 277         if (unlikely(n > RPCRDMA_MAX_SEGS))
 278                 return -EIO;
 279         return n;
 280 }
 281
 282 static inline int
 283 encode_item_present(struct xdr_stream *xdr)
 284 {
 285         __be32 *p;
 286
 287         p = xdr_reserve_space(xdr, sizeof(*p));
 288         if (unlikely(!p))
 289                 return -EMSGSIZE;
 290
 291         *p = xdr_one;
 292         return 0;
 293 }
 294
 295 static inline int
 296 encode_item_not_present(struct xdr_stream *xdr)
 297 {
 298         __be32 *p;
 299
 300         p = xdr_reserve_space(xdr, sizeof(*p));
 301         if (unlikely(!p))
 302                 return -EMSGSIZE;
 303
 304         *p = xdr_zero;
 305         return 0;
 306 }
 307
 308 static void
 309 xdr_encode_rdma_segment(__be32 *iptr, struct rpcrdma_mr *mr)
 310 {
 311         *iptr++ = cpu_to_be32(mr->mr_handle);
 312         *iptr++ = cpu_to_be32(mr->mr_length);
 313         xdr_encode_hyper(iptr, mr->mr_offset);
 314 }
 315
 316 static int
 317 encode_rdma_segment(struct xdr_stream *xdr, struct rpcrdma_mr *mr)
 318 {
 319         __be32 *p;
 320
 321         p = xdr_reserve_space(xdr, 4 * sizeof(*p));
 322         if (unlikely(!p))
 323                 return -EMSGSIZE;
 324
 325         xdr_encode_rdma_segment(p, mr);
 326         return 0;
 327 }
 328
 329 static int
 330 encode_read_segment(struct xdr_stream *xdr, struct rpcrdma_mr *mr,
 331                     u32 position)
 332 {
 333         __be32 *p;
 334
 335         p = xdr_reserve_space(xdr, 6 * sizeof(*p));
 336         if (unlikely(!p))
 337                 return -EMSGSIZE;
 338
 339         *p++ = xdr_one;                 /* Item present */
 340         *p++ = cpu_to_be32(position);
 341         xdr_encode_rdma_segment(p, mr);
 342         return 0;
 343 }
 344
 345 static struct rpcrdma_mr_seg *rpcrdma_mr_prepare(struct rpcrdma_xprt *r_xprt,
 346                                                  struct rpcrdma_req *req,
 347                                                  struct rpcrdma_mr_seg *seg,
 348                                                  int nsegs, bool writing,
 349                                                  struct rpcrdma_mr **mr)
 350 {
 351         *mr = rpcrdma_mr_pop(&req->rl_free_mrs);
 352         if (!*mr) {
 353                 *mr = rpcrdma_mr_get(r_xprt);
 354                 if (!*mr)
 355                         goto out_getmr_err;
 356                 trace_xprtrdma_mr_get(req);
 357                 (*mr)->mr_req = req;
 358         }
 359
 360         rpcrdma_mr_push(*mr, &req->rl_registered);
 361         return frwr_map(r_xprt, seg, nsegs, writing, req->rl_slot.rq_xid, *mr);
 362
 363 out_getmr_err:
 364         trace_xprtrdma_nomrs(req);
 365         xprt_wait_for_buffer_space(&r_xprt->rx_xprt);
 366         rpcrdma_mrs_refresh(r_xprt);
 367         return ERR_PTR(-EAGAIN);
 368 }
 369
 370 /* Register and XDR encode the Read list. Supports encoding a list of read
 371  * segments that belong to a single read chunk.
 372  *
 373  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
 374  *
 375  *  Read chunklist (a linked list):
 376  *   N elements, position P (same P for all chunks of same arg!):
 377  *    1 - PHLOO - 1 - PHLOO - ... - 1 - PHLOO - 0
 378  *
 379  * Returns zero on success, or a negative errno if a failure occurred.
 380  * @xdr is advanced to the next position in the stream.
 381  *
 382  * Only a single @pos value is currently supported.
 383  */
 384 static int rpcrdma_encode_read_list(struct rpcrdma_xprt *r_xprt,
 385                                     struct rpcrdma_req *req,
 386                                     struct rpc_rqst *rqst,
 387                                     enum rpcrdma_chunktype rtype)
 388 {
 389         struct xdr_stream *xdr = &req->rl_stream;
 390         struct rpcrdma_mr_seg *seg;
 391         struct rpcrdma_mr *mr;
 392         unsigned int pos;
 393         int nsegs;
 394
 395         if (rtype == rpcrdma_noch_pullup || rtype == rpcrdma_noch_mapped)
 396                 goto done;
 397
 398         pos = rqst->rq_snd_buf.head[0].iov_len;
 399         if (rtype == rpcrdma_areadch)
 400                 pos = 0;
 401         seg = req->rl_segments;
 402         nsegs = rpcrdma_convert_iovs(r_xprt, &rqst->rq_snd_buf, pos,
 403                                      rtype, seg);
 404         if (nsegs < 0)
 405                 return nsegs;
 406
 407         do {
 408                 seg = rpcrdma_mr_prepare(r_xprt, req, seg, nsegs, false, &mr);
 409                 if (IS_ERR(seg))
 410                         return PTR_ERR(seg);
 411
 412                 if (encode_read_segment(xdr, mr, pos) < 0)
 413                         return -EMSGSIZE;
 414
 415                 trace_xprtrdma_chunk_read(rqst->rq_task, pos, mr, nsegs);
 416                 r_xprt->rx_stats.read_chunk_count++;
 417                 nsegs -= mr->mr_nents;
 418         } while (nsegs);
 419
 420 done:
 421         return encode_item_not_present(xdr);
 422 }
 423
 424 /* Register and XDR encode the Write list. Supports encoding a list
 425  * containing one array of plain segments that belong to a single
 426  * write chunk.
 427  *
 428  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
 429  *
 430  *  Write chunklist (a list of (one) counted array):
 431  *   N elements:
 432  *    1 - N - HLOO - HLOO - ... - HLOO - 0
 433  *
 434  * Returns zero on success, or a negative errno if a failure occurred.
 435  * @xdr is advanced to the next position in the stream.
 436  *
 437  * Only a single Write chunk is currently supported.
 438  */
 439 static int rpcrdma_encode_write_list(struct rpcrdma_xprt *r_xprt,
 440                                      struct rpcrdma_req *req,
 441                                      struct rpc_rqst *rqst,
 442                                      enum rpcrdma_chunktype wtype)
 443 {
 444         struct xdr_stream *xdr = &req->rl_stream;
 445         struct rpcrdma_mr_seg *seg;
 446         struct rpcrdma_mr *mr;
 447         int nsegs, nchunks;
 448         __be32 *segcount;
 449
 450         if (wtype != rpcrdma_writech)
 451                 goto done;
 452
 453         seg = req->rl_segments;
 454         nsegs = rpcrdma_convert_iovs(r_xprt, &rqst->rq_rcv_buf,
 455                                      rqst->rq_rcv_buf.head[0].iov_len,
 456                                      wtype, seg);
 457         if (nsegs < 0)
 458                 return nsegs;
 459
 460         if (encode_item_present(xdr) < 0)
 461                 return -EMSGSIZE;
 462         segcount = xdr_reserve_space(xdr, sizeof(*segcount));
 463         if (unlikely(!segcount))
 464                 return -EMSGSIZE;
 465         /* Actual value encoded below */
 466
 467         nchunks = 0;
 468         do {
 469                 seg = rpcrdma_mr_prepare(r_xprt, req, seg, nsegs, true, &mr);
 470                 if (IS_ERR(seg))
 471                         return PTR_ERR(seg);
 472
 473                 if (encode_rdma_segment(xdr, mr) < 0)
 474                         return -EMSGSIZE;
 475
 476                 trace_xprtrdma_chunk_write(rqst->rq_task, mr, nsegs);
 477                 r_xprt->rx_stats.write_chunk_count++;
 478                 r_xprt->rx_stats.total_rdma_request += mr->mr_length;
 479                 nchunks++;
 480                 nsegs -= mr->mr_nents;
 481         } while (nsegs);
 482
 483         /* Update count of segments in this Write chunk */
 484         *segcount = cpu_to_be32(nchunks);
 485
 486 done:
 487         return encode_item_not_present(xdr);
 488 }
 489
 490 /* Register and XDR encode the Reply chunk. Supports encoding an array
 491  * of plain segments that belong to a single write (reply) chunk.
 492  *
 493  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
 494  *
 495  *  Reply chunk (a counted array):
 496  *   N elements:
 497  *    1 - N - HLOO - HLOO - ... - HLOO
 498  *
 499  * Returns zero on success, or a negative errno if a failure occurred.
 500  * @xdr is advanced to the next position in the stream.
 501  */
 502 static int rpcrdma_encode_reply_chunk(struct rpcrdma_xprt *r_xprt,
 503                                       struct rpcrdma_req *req,
 504                                       struct rpc_rqst *rqst,
 505                                       enum rpcrdma_chunktype wtype)
 506 {
 507         struct xdr_stream *xdr = &req->rl_stream;
 508         struct rpcrdma_mr_seg *seg;
 509         struct rpcrdma_mr *mr;
 510         int nsegs, nchunks;
 511         __be32 *segcount;
 512
 513         if (wtype != rpcrdma_replych)
 514                 return encode_item_not_present(xdr);
 515
 516         seg = req->rl_segments;
 517         nsegs = rpcrdma_convert_iovs(r_xprt, &rqst->rq_rcv_buf, 0, wtype, seg);
 518         if (nsegs < 0)
 519                 return nsegs;
 520
 521         if (encode_item_present(xdr) < 0)
 522                 return -EMSGSIZE;
 523         segcount = xdr_reserve_space(xdr, sizeof(*segcount));
 524         if (unlikely(!segcount))
 525                 return -EMSGSIZE;
 526         /* Actual value encoded below */
 527
 528         nchunks = 0;
 529         do {
 530                 seg = rpcrdma_mr_prepare(r_xprt, req, seg, nsegs, true, &mr);
 531                 if (IS_ERR(seg))
 532                         return PTR_ERR(seg);
 533
 534                 if (encode_rdma_segment(xdr, mr) < 0)
 535                         return -EMSGSIZE;
 536
 537                 trace_xprtrdma_chunk_reply(rqst->rq_task, mr, nsegs);
 538                 r_xprt->rx_stats.reply_chunk_count++;
 539                 r_xprt->rx_stats.total_rdma_request += mr->mr_length;
 540                 nchunks++;
 541                 nsegs -= mr->mr_nents;
 542         } while (nsegs);
 543
 544         /* Update count of segments in the Reply chunk */
 545         *segcount = cpu_to_be32(nchunks);
 546
 547         return 0;
 548 }
 549
 550 static void rpcrdma_sendctx_done(struct kref *kref)
 551 {
 552         struct rpcrdma_req *req =
 553                 container_of(kref, struct rpcrdma_req, rl_kref);
 554         struct rpcrdma_rep *rep = req->rl_reply;
 555
 556         rpcrdma_complete_rqst(rep);
 557         rep->rr_rxprt->rx_stats.reply_waits_for_send++;
 558 }
 559
 560 /**
 561  * rpcrdma_sendctx_unmap - DMA-unmap Send buffer
 562  * @sc: sendctx containing SGEs to unmap
 563  *
 564  */
 565 void rpcrdma_sendctx_unmap(struct rpcrdma_sendctx *sc)
 566 {
 567         struct rpcrdma_regbuf *rb = sc->sc_req->rl_sendbuf;
 568         struct ib_sge *sge;
 569
 570         if (!sc->sc_unmap_count)
 571                 return;
 572
 573         /* The first two SGEs contain the transport header and
 574          * the inline buffer. These are always left mapped so
 575          * they can be cheaply re-used.
 576          */
 577         for (sge = &sc->sc_sges[2]; sc->sc_unmap_count;
 578              ++sge, --sc->sc_unmap_count)
 579                 ib_dma_unmap_page(rdmab_device(rb), sge->addr, sge->length,
 580                                   DMA_TO_DEVICE);
 581
 582         kref_put(&sc->sc_req->rl_kref, rpcrdma_sendctx_done);
 583 }
 584
 585 /* Prepare an SGE for the RPC-over-RDMA transport header.
 586  */
 587 static bool rpcrdma_prepare_hdr_sge(struct rpcrdma_xprt *r_xprt,
 588                                     struct rpcrdma_req *req, u32 len)
 589 {
 590         struct rpcrdma_sendctx *sc = req->rl_sendctx;
 591         struct rpcrdma_regbuf *rb = req->rl_rdmabuf;
 592         struct ib_sge *sge = &sc->sc_sges[req->rl_wr.num_sge++];
 593
 594         if (!rpcrdma_regbuf_dma_map(r_xprt, rb))
 595                 return false;
 596         sge->addr = rdmab_addr(rb);
 597         sge->length = len;
 598         sge->lkey = rdmab_lkey(rb);
 599
 600         ib_dma_sync_single_for_device(rdmab_device(rb), sge->addr, sge->length,
 601                                       DMA_TO_DEVICE);
 602         return true;
 603 }
 604
 605 /* The head iovec is straightforward, as it is usually already
 606  * DMA-mapped. Sync the content that has changed.
 607  */
 608 static bool rpcrdma_prepare_head_iov(struct rpcrdma_xprt *r_xprt,
 609                                      struct rpcrdma_req *req, unsigned int len)
 610 {
 611         struct rpcrdma_sendctx *sc = req->rl_sendctx;
 612         struct ib_sge *sge = &sc->sc_sges[req->rl_wr.num_sge++];
 613         struct rpcrdma_regbuf *rb = req->rl_sendbuf;
 614
 615         if (!rpcrdma_regbuf_dma_map(r_xprt, rb))
 616                 return false;
 617
 618         sge->addr = rdmab_addr(rb);
 619         sge->length = len;
 620         sge->lkey = rdmab_lkey(rb);
 621
 622         ib_dma_sync_single_for_device(rdmab_device(rb), sge->addr, sge->length,
 623                                       DMA_TO_DEVICE);
 624         return true;
 625 }
 626
 627 /* If there is a page list present, DMA map and prepare an
 628  * SGE for each page to be sent.
 629  */
 630 static bool rpcrdma_prepare_pagelist(struct rpcrdma_req *req,
 631                                      struct xdr_buf *xdr)
 632 {
 633         struct rpcrdma_sendctx *sc = req->rl_sendctx;
 634         struct rpcrdma_regbuf *rb = req->rl_sendbuf;
 635         unsigned int page_base, len, remaining;
 636         struct page **ppages;
 637         struct ib_sge *sge;
 638
 639         ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT);
 640         page_base = offset_in_page(xdr->page_base);
 641         remaining = xdr->page_len;
 642         while (remaining) {
 643                 sge = &sc->sc_sges[req->rl_wr.num_sge++];
 644                 len = min_t(unsigned int, PAGE_SIZE - page_base, remaining);
 645                 sge->addr = ib_dma_map_page(rdmab_device(rb), *ppages,
 646                                             page_base, len, DMA_TO_DEVICE);
 647                 if (ib_dma_mapping_error(rdmab_device(rb), sge->addr))
 648                         goto out_mapping_err;
 649
 650                 sge->length = len;
 651                 sge->lkey = rdmab_lkey(rb);
 652
 653                 sc->sc_unmap_count++;
 654                 ppages++;
 655                 remaining -= len;
 656                 page_base = 0;
 657         }
 658
 659         return true;
 660
 661 out_mapping_err:
 662         trace_xprtrdma_dma_maperr(sge->addr);
 663         return false;
 664 }
 665
 666 /* The tail iovec may include an XDR pad for the page list,
 667  * as well as additional content, and may not reside in the
 668  * same page as the head iovec.
 669  */
 670 static bool rpcrdma_prepare_tail_iov(struct rpcrdma_req *req,
 671                                      struct xdr_buf *xdr,
 672                                      unsigned int page_base, unsigned int len)
 673 {
 674         struct rpcrdma_sendctx *sc = req->rl_sendctx;
 675         struct ib_sge *sge = &sc->sc_sges[req->rl_wr.num_sge++];
 676         struct rpcrdma_regbuf *rb = req->rl_sendbuf;
 677         struct page *page = virt_to_page(xdr->tail[0].iov_base);
 678
 679         sge->addr = ib_dma_map_page(rdmab_device(rb), page, page_base, len,
 680                                     DMA_TO_DEVICE);
 681         if (ib_dma_mapping_error(rdmab_device(rb), sge->addr))
 682                 goto out_mapping_err;
 683
 684         sge->length = len;
 685         sge->lkey = rdmab_lkey(rb);
 686         ++sc->sc_unmap_count;
 687         return true;
 688
 689 out_mapping_err:
 690         trace_xprtrdma_dma_maperr(sge->addr);
 691         return false;
 692 }
 693
 694 /* Copy the tail to the end of the head buffer.
 695  */
 696 static void rpcrdma_pullup_tail_iov(struct rpcrdma_xprt *r_xprt,
 697                                     struct rpcrdma_req *req,
 698                                     struct xdr_buf *xdr)
 699 {
 700         unsigned char *dst;
 701
 702         dst = (unsigned char *)xdr->head[0].iov_base;
 703         dst += xdr->head[0].iov_len + xdr->page_len;
 704         memmove(dst, xdr->tail[0].iov_base, xdr->tail[0].iov_len);
 705         r_xprt->rx_stats.pullup_copy_count += xdr->tail[0].iov_len;
 706 }
 707
 708 /* Copy pagelist content into the head buffer.
 709  */
 710 static void rpcrdma_pullup_pagelist(struct rpcrdma_xprt *r_xprt,
 711                                     struct rpcrdma_req *req,
 712                                     struct xdr_buf *xdr)
 713 {
 714         unsigned int len, page_base, remaining;
 715         struct page **ppages;
 716         unsigned char *src, *dst;
 717
 718         dst = (unsigned char *)xdr->head[0].iov_base;
 719         dst += xdr->head[0].iov_len;
 720         ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT);
 721         page_base = offset_in_page(xdr->page_base);
 722         remaining = xdr->page_len;
 723         while (remaining) {
 724                 src = page_address(*ppages);
 725                 src += page_base;
 726                 len = min_t(unsigned int, PAGE_SIZE - page_base, remaining);
 727                 memcpy(dst, src, len);
 728                 r_xprt->rx_stats.pullup_copy_count += len;
 729
 730                 ppages++;
 731                 dst += len;
 732                 remaining -= len;
 733                 page_base = 0;
 734         }
 735 }
 736
 737 /* Copy the contents of @xdr into @rl_sendbuf and DMA sync it.
 738  * When the head, pagelist, and tail are small, a pull-up copy
 739  * is considerably less costly than DMA mapping the components
 740  * of @xdr.
 741  *
 742  * Assumptions:
 743  *  - the caller has already verified that the total length
 744  *    of the RPC Call body will fit into @rl_sendbuf.
 745  */
 746 static bool rpcrdma_prepare_noch_pullup(struct rpcrdma_xprt *r_xprt,
 747                                         struct rpcrdma_req *req,
 748                                         struct xdr_buf *xdr)
 749 {
 750         if (unlikely(xdr->tail[0].iov_len))
 751                 rpcrdma_pullup_tail_iov(r_xprt, req, xdr);
 752
 753         if (unlikely(xdr->page_len))
 754                 rpcrdma_pullup_pagelist(r_xprt, req, xdr);
 755
 756         /* The whole RPC message resides in the head iovec now */
 757         return rpcrdma_prepare_head_iov(r_xprt, req, xdr->len);
 758 }
 759
 760 static bool rpcrdma_prepare_noch_mapped(struct rpcrdma_xprt *r_xprt,
 761                                         struct rpcrdma_req *req,
 762                                         struct xdr_buf *xdr)
 763 {
 764         struct kvec *tail = &xdr->tail[0];
 765
 766         if (!rpcrdma_prepare_head_iov(r_xprt, req, xdr->head[0].iov_len))
 767                 return false;
 768         if (xdr->page_len)
 769                 if (!rpcrdma_prepare_pagelist(req, xdr))
 770                         return false;
 771         if (tail->iov_len)
 772                 if (!rpcrdma_prepare_tail_iov(req, xdr,
 773                                               offset_in_page(tail->iov_base),
 774                                               tail->iov_len))
 775                         return false;
 776
 777         if (req->rl_sendctx->sc_unmap_count)
 778                 kref_get(&req->rl_kref);
 779         return true;
 780 }
 781
 782 static bool rpcrdma_prepare_readch(struct rpcrdma_xprt *r_xprt,
 783                                    struct rpcrdma_req *req,
 784                                    struct xdr_buf *xdr)
 785 {
 786         if (!rpcrdma_prepare_head_iov(r_xprt, req, xdr->head[0].iov_len))
 787                 return false;
 788
 789         /* If there is a Read chunk, the page list is being handled
 790          * via explicit RDMA, and thus is skipped here.
 791          */
 792
 793         /* Do not include the tail if it is only an XDR pad */
 794         if (xdr->tail[0].iov_len > 3) {
 795                 unsigned int page_base, len;
 796
 797                 /* If the content in the page list is an odd length,
 798                  * xdr_write_pages() adds a pad at the beginning of
 799                  * the tail iovec. Force the tail's non-pad content to
 800                  * land at the next XDR position in the Send message.
 801                  */
 802                 page_base = offset_in_page(xdr->tail[0].iov_base);
 803                 len = xdr->tail[0].iov_len;
 804                 page_base += len & 3;
 805                 len -= len & 3;
 806                 if (!rpcrdma_prepare_tail_iov(req, xdr, page_base, len))
 807                         return false;
 808                 kref_get(&req->rl_kref);
 809         }
 810
 811         return true;
 812 }
 813
 814 /**
 815  * rpcrdma_prepare_send_sges - Construct SGEs for a Send WR
 816  * @r_xprt: controlling transport
 817  * @req: context of RPC Call being marshalled
 818  * @hdrlen: size of transport header, in bytes
 819  * @xdr: xdr_buf containing RPC Call
 820  * @rtype: chunk type being encoded
 821  *
 822  * Returns 0 on success; otherwise a negative errno is returned.
 823  */
 824 inline int rpcrdma_prepare_send_sges(struct rpcrdma_xprt *r_xprt,
 825                                      struct rpcrdma_req *req, u32 hdrlen,
 826                                      struct xdr_buf *xdr,
 827                                      enum rpcrdma_chunktype rtype)
 828 {
 829         int ret;
 830
 831         ret = -EAGAIN;
 832         req->rl_sendctx = rpcrdma_sendctx_get_locked(r_xprt);
 833         if (!req->rl_sendctx)
 834                 goto out_nosc;
 835         req->rl_sendctx->sc_unmap_count = 0;
 836         req->rl_sendctx->sc_req = req;
 837         kref_init(&req->rl_kref);
 838         req->rl_wr.wr_cqe = &req->rl_sendctx->sc_cqe;
 839         req->rl_wr.sg_list = req->rl_sendctx->sc_sges;
 840         req->rl_wr.num_sge = 0;
 841         req->rl_wr.opcode = IB_WR_SEND;
 842
 843         ret = -EIO;
 844         if (!rpcrdma_prepare_hdr_sge(r_xprt, req, hdrlen))
 845                 goto out_unmap;
 846
 847         switch (rtype) {
 848         case rpcrdma_noch_pullup:
 849                 if (!rpcrdma_prepare_noch_pullup(r_xprt, req, xdr))
 850                         goto out_unmap;
 851                 break;
 852         case rpcrdma_noch_mapped:
 853                 if (!rpcrdma_prepare_noch_mapped(r_xprt, req, xdr))
 854                         goto out_unmap;
 855                 break;
 856         case rpcrdma_readch:
 857                 if (!rpcrdma_prepare_readch(r_xprt, req, xdr))
 858                         goto out_unmap;
 859                 break;
 860         case rpcrdma_areadch:
 861                 break;
 862         default:
 863                 goto out_unmap;
 864         }
 865
 866         return 0;
 867
 868 out_unmap:
 869         rpcrdma_sendctx_unmap(req->rl_sendctx);
 870 out_nosc:
 871         trace_xprtrdma_prepsend_failed(&req->rl_slot, ret);
 872         return ret;
 873 }
 874
 875 /**
 876  * rpcrdma_marshal_req - Marshal and send one RPC request
 877  * @r_xprt: controlling transport
 878  * @rqst: RPC request to be marshaled
 879  *
 880  * For the RPC in "rqst", this function:
 881  *  - Chooses the transfer mode (eg., RDMA_MSG or RDMA_NOMSG)
 882  *  - Registers Read, Write, and Reply chunks
 883  *  - Constructs the transport header
 884  *  - Posts a Send WR to send the transport header and request
 885  *
 886  * Returns:
 887  *      %0 if the RPC was sent successfully,
 888  *      %-ENOTCONN if the connection was lost,
 889  *      %-EAGAIN if the caller should call again with the same arguments,
 890  *      %-ENOBUFS if the caller should call again after a delay,
 891  *      %-EMSGSIZE if the transport header is too small,
 892  *      %-EIO if a permanent problem occurred while marshaling.
 893  */
 894 int
 895 rpcrdma_marshal_req(struct rpcrdma_xprt *r_xprt, struct rpc_rqst *rqst)
 896 {
 897         struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
 898         struct xdr_stream *xdr = &req->rl_stream;
 899         enum rpcrdma_chunktype rtype, wtype;
 900         struct xdr_buf *buf = &rqst->rq_snd_buf;
 901         bool ddp_allowed;
 902         __be32 *p;
 903         int ret;
 904
 905         rpcrdma_set_xdrlen(&req->rl_hdrbuf, 0);
 906         xdr_init_encode(xdr, &req->rl_hdrbuf, rdmab_data(req->rl_rdmabuf),
 907                         rqst);
 908
 909         /* Fixed header fields */
 910         ret = -EMSGSIZE;
 911         p = xdr_reserve_space(xdr, 4 * sizeof(*p));
 912         if (!p)
 913                 goto out_err;
 914         *p++ = rqst->rq_xid;
 915         *p++ = rpcrdma_version;
 916         *p++ = cpu_to_be32(r_xprt->rx_buf.rb_max_requests);
 917
 918         /* When the ULP employs a GSS flavor that guarantees integrity
 919          * or privacy, direct data placement of individual data items
 920          * is not allowed.
 921          */
 922         ddp_allowed = !(rqst->rq_cred->cr_auth->au_flags &
 923                                                 RPCAUTH_AUTH_DATATOUCH);
 924
 925         /*
 926          * Chunks needed for results?
 927          *
 928          * o If the expected result is under the inline threshold, all ops
 929          *   return as inline.
 930          * o Large read ops return data as write chunk(s), header as
 931          *   inline.
 932          * o Large non-read ops return as a single reply chunk.
 933          */
 934         if (rpcrdma_results_inline(r_xprt, rqst))
 935                 wtype = rpcrdma_noch;
 936         else if ((ddp_allowed && rqst->rq_rcv_buf.flags & XDRBUF_READ) &&
 937                  rpcrdma_nonpayload_inline(r_xprt, rqst))
 938                 wtype = rpcrdma_writech;
 939         else
 940                 wtype = rpcrdma_replych;
 941
 942         /*
 943          * Chunks needed for arguments?
 944          *
 945          * o If the total request is under the inline threshold, all ops
 946          *   are sent as inline.
 947          * o Large write ops transmit data as read chunk(s), header as
 948          *   inline.
 949          * o Large non-write ops are sent with the entire message as a
 950          *   single read chunk (protocol 0-position special case).
 951          *
 952          * This assumes that the upper layer does not present a request
 953          * that both has a data payload, and whose non-data arguments
 954          * by themselves are larger than the inline threshold.
 955          */
 956         if (rpcrdma_args_inline(r_xprt, rqst)) {
 957                 *p++ = rdma_msg;
 958                 rtype = buf->len < rdmab_length(req->rl_sendbuf) ?
 959                         rpcrdma_noch_pullup : rpcrdma_noch_mapped;
 960         } else if (ddp_allowed && buf->flags & XDRBUF_WRITE) {
 961                 *p++ = rdma_msg;
 962                 rtype = rpcrdma_readch;
 963         } else {
 964                 r_xprt->rx_stats.nomsg_call_count++;
 965                 *p++ = rdma_nomsg;
 966                 rtype = rpcrdma_areadch;
 967         }
 968
 969         /* This implementation supports the following combinations
 970          * of chunk lists in one RPC-over-RDMA Call message:
 971          *
 972          *   - Read list
 973          *   - Write list
 974          *   - Reply chunk
 975          *   - Read list + Reply chunk
 976          *
 977          * It might not yet support the following combinations:
 978          *
 979          *   - Read list + Write list
 980          *
 981          * It does not support the following combinations:
 982          *
 983          *   - Write list + Reply chunk
 984          *   - Read list + Write list + Reply chunk
 985          *
 986          * This implementation supports only a single chunk in each
 987          * Read or Write list. Thus for example the client cannot
 988          * send a Call message with a Position Zero Read chunk and a
 989          * regular Read chunk at the same time.
 990          */
 991         ret = rpcrdma_encode_read_list(r_xprt, req, rqst, rtype);
 992         if (ret)
 993                 goto out_err;
 994         ret = rpcrdma_encode_write_list(r_xprt, req, rqst, wtype);
 995         if (ret)
 996                 goto out_err;
 997         ret = rpcrdma_encode_reply_chunk(r_xprt, req, rqst, wtype);
 998         if (ret)
 999                 goto out_err;
1000
1001         ret = rpcrdma_prepare_send_sges(r_xprt, req, req->rl_hdrbuf.len,
1002                                         buf, rtype);
1003         if (ret)
1004                 goto out_err;
1005
1006         trace_xprtrdma_marshal(req, rtype, wtype);
1007         return 0;
1008
1009 out_err:
1010         trace_xprtrdma_marshal_failed(rqst, ret);
1011         r_xprt->rx_stats.failed_marshal_count++;
1012         frwr_reset(req);
1013         return ret;
1014 }
1015
1016 static void __rpcrdma_update_cwnd_locked(struct rpc_xprt *xprt,
1017                                          struct rpcrdma_buffer *buf,
1018                                          u32 grant)
1019 {
1020         buf->rb_credits = grant;
1021         xprt->cwnd = grant << RPC_CWNDSHIFT;
1022 }
1023
1024 static void rpcrdma_update_cwnd(struct rpcrdma_xprt *r_xprt, u32 grant)
1025 {
1026         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1027
1028         spin_lock(&xprt->transport_lock);
1029         __rpcrdma_update_cwnd_locked(xprt, &r_xprt->rx_buf, grant);
1030         spin_unlock(&xprt->transport_lock);
1031 }
1032
1033 /**
1034  * rpcrdma_reset_cwnd - Reset the xprt's congestion window
1035  * @r_xprt: controlling transport instance
1036  *
1037  * Prepare @r_xprt for the next connection by reinitializing
1038  * its credit grant to one (see RFC 8166, Section 3.3.3).
1039  */
1040 void rpcrdma_reset_cwnd(struct rpcrdma_xprt *r_xprt)
1041 {
1042         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1043
1044         spin_lock(&xprt->transport_lock);
1045         xprt->cong = 0;
1046         __rpcrdma_update_cwnd_locked(xprt, &r_xprt->rx_buf, 1);
1047         spin_unlock(&xprt->transport_lock);
1048 }
1049
1050 /**
1051  * rpcrdma_inline_fixup - Scatter inline received data into rqst's iovecs
1052  * @rqst: controlling RPC request
1053  * @srcp: points to RPC message payload in receive buffer
1054  * @copy_len: remaining length of receive buffer content
1055  * @pad: Write chunk pad bytes needed (zero for pure inline)
1056  *
1057  * The upper layer has set the maximum number of bytes it can
1058  * receive in each component of rq_rcv_buf. These values are set in
1059  * the head.iov_len, page_len, tail.iov_len, and buflen fields.
1060  *
1061  * Unlike the TCP equivalent (xdr_partial_copy_from_skb), in
1062  * many cases this function simply updates iov_base pointers in
1063  * rq_rcv_buf to point directly to the received reply data, to
1064  * avoid copying reply data.
1065  *
1066  * Returns the count of bytes which had to be memcopied.
1067  */
1068 static unsigned long
1069 rpcrdma_inline_fixup(struct rpc_rqst *rqst, char *srcp, int copy_len, int pad)
1070 {
1071         unsigned long fixup_copy_count;
1072         int i, npages, curlen;
1073         char *destp;
1074         struct page **ppages;
1075         int page_base;
1076
1077         /* The head iovec is redirected to the RPC reply message
1078          * in the receive buffer, to avoid a memcopy.
1079          */
1080         rqst->rq_rcv_buf.head[0].iov_base = srcp;
1081         rqst->rq_private_buf.head[0].iov_base = srcp;
1082
1083         /* The contents of the receive buffer that follow
1084          * head.iov_len bytes are copied into the page list.
1085          */
1086         curlen = rqst->rq_rcv_buf.head[0].iov_len;
1087         if (curlen > copy_len)
1088                 curlen = copy_len;
1089         srcp += curlen;
1090         copy_len -= curlen;
1091
1092         ppages = rqst->rq_rcv_buf.pages +
1093                 (rqst->rq_rcv_buf.page_base >> PAGE_SHIFT);
1094         page_base = offset_in_page(rqst->rq_rcv_buf.page_base);
1095         fixup_copy_count = 0;
1096         if (copy_len && rqst->rq_rcv_buf.page_len) {
1097                 int pagelist_len;
1098
1099                 pagelist_len = rqst->rq_rcv_buf.page_len;
1100                 if (pagelist_len > copy_len)
1101                         pagelist_len = copy_len;
1102                 npages = PAGE_ALIGN(page_base + pagelist_len) >> PAGE_SHIFT;
1103                 for (i = 0; i < npages; i++) {
1104                         curlen = PAGE_SIZE - page_base;
1105                         if (curlen > pagelist_len)
1106                                 curlen = pagelist_len;
1107
1108                         destp = kmap_atomic(ppages[i]);
1109                         memcpy(destp + page_base, srcp, curlen);
1110                         flush_dcache_page(ppages[i]);
1111                         kunmap_atomic(destp);
1112                         srcp += curlen;
1113                         copy_len -= curlen;
1114                         fixup_copy_count += curlen;
1115                         pagelist_len -= curlen;
1116                         if (!pagelist_len)
1117                                 break;
1118                         page_base = 0;
1119                 }
1120
1121                 /* Implicit padding for the last segment in a Write
1122                  * chunk is inserted inline at the front of the tail
1123                  * iovec. The upper layer ignores the content of
1124                  * the pad. Simply ensure inline content in the tail
1125                  * that follows the Write chunk is properly aligned.
1126                  */
1127                 if (pad)
1128                         srcp -= pad;
1129         }
1130
1131         /* The tail iovec is redirected to the remaining data
1132          * in the receive buffer, to avoid a memcopy.
1133          */
1134         if (copy_len || pad) {
1135                 rqst->rq_rcv_buf.tail[0].iov_base = srcp;
1136                 rqst->rq_private_buf.tail[0].iov_base = srcp;
1137         }
1138
1139         if (fixup_copy_count)
1140                 trace_xprtrdma_fixup(rqst, fixup_copy_count);
1141         return fixup_copy_count;
1142 }
1143
1144 /* By convention, backchannel calls arrive via rdma_msg type
1145  * messages, and never populate the chunk lists. This makes
1146  * the RPC/RDMA header small and fixed in size, so it is
1147  * straightforward to check the RPC header's direction field.
1148  */
1149 static bool
1150 rpcrdma_is_bcall(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep)
1151 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1152 {
1153         struct xdr_stream *xdr = &rep->rr_stream;
1154         __be32 *p;
1155
1156         if (rep->rr_proc != rdma_msg)
1157                 return false;
1158
1159         /* Peek at stream contents without advancing. */
1160         p = xdr_inline_decode(xdr, 0);
1161
1162         /* Chunk lists */
1163         if (*p++ != xdr_zero)
1164                 return false;
1165         if (*p++ != xdr_zero)
1166                 return false;
1167         if (*p++ != xdr_zero)
1168                 return false;
1169
1170         /* RPC header */
1171         if (*p++ != rep->rr_xid)
1172                 return false;
1173         if (*p != cpu_to_be32(RPC_CALL))
1174                 return false;
1175
1176         /* Now that we are sure this is a backchannel call,
1177          * advance to the RPC header.
1178          */
1179         p = xdr_inline_decode(xdr, 3 * sizeof(*p));
1180         if (unlikely(!p))
1181                 goto out_short;
1182
1183         rpcrdma_bc_receive_call(r_xprt, rep);
1184         return true;
1185
1186 out_short:
1187         pr_warn("RPC/RDMA short backward direction call\n");
1188         return true;
1189 }
1190 #else   /* CONFIG_SUNRPC_BACKCHANNEL */
1191 {
1192         return false;
1193 }
1194 #endif  /* CONFIG_SUNRPC_BACKCHANNEL */
1195
1196 static int decode_rdma_segment(struct xdr_stream *xdr, u32 *length)
1197 {
1198         u32 handle;
1199         u64 offset;
1200         __be32 *p;
1201
1202         p = xdr_inline_decode(xdr, 4 * sizeof(*p));
1203         if (unlikely(!p))
1204                 return -EIO;
1205
1206         handle = be32_to_cpup(p++);
1207         *length = be32_to_cpup(p++);
1208         xdr_decode_hyper(p, &offset);
1209
1210         trace_xprtrdma_decode_seg(handle, *length, offset);
1211         return 0;
1212 }
1213
1214 static int decode_write_chunk(struct xdr_stream *xdr, u32 *length)
1215 {
1216         u32 segcount, seglength;
1217         __be32 *p;
1218
1219         p = xdr_inline_decode(xdr, sizeof(*p));
1220         if (unlikely(!p))
1221                 return -EIO;
1222
1223         *length = 0;
1224         segcount = be32_to_cpup(p);
1225         while (segcount--) {
1226                 if (decode_rdma_segment(xdr, &seglength))
1227                         return -EIO;
1228                 *length += seglength;
1229         }
1230
1231         return 0;
1232 }
1233
1234 /* In RPC-over-RDMA Version One replies, a Read list is never
1235  * expected. This decoder is a stub that returns an error if
1236  * a Read list is present.
1237  */
1238 static int decode_read_list(struct xdr_stream *xdr)
1239 {
1240         __be32 *p;
1241
1242         p = xdr_inline_decode(xdr, sizeof(*p));
1243         if (unlikely(!p))
1244                 return -EIO;
1245         if (unlikely(*p != xdr_zero))
1246                 return -EIO;
1247         return 0;
1248 }
1249
1250 /* Supports only one Write chunk in the Write list
1251  */
1252 static int decode_write_list(struct xdr_stream *xdr, u32 *length)
1253 {
1254         u32 chunklen;
1255         bool first;
1256         __be32 *p;
1257
1258         *length = 0;
1259         first = true;
1260         do {
1261                 p = xdr_inline_decode(xdr, sizeof(*p));
1262                 if (unlikely(!p))
1263                         return -EIO;
1264                 if (*p == xdr_zero)
1265                         break;
1266                 if (!first)
1267                         return -EIO;
1268
1269                 if (decode_write_chunk(xdr, &chunklen))
1270                         return -EIO;
1271                 *length += chunklen;
1272                 first = false;
1273         } while (true);
1274         return 0;
1275 }
1276
1277 static int decode_reply_chunk(struct xdr_stream *xdr, u32 *length)
1278 {
1279         __be32 *p;
1280
1281         p = xdr_inline_decode(xdr, sizeof(*p));
1282         if (unlikely(!p))
1283                 return -EIO;
1284
1285         *length = 0;
1286         if (*p != xdr_zero)
1287                 if (decode_write_chunk(xdr, length))
1288                         return -EIO;
1289         return 0;
1290 }
1291
1292 static int
1293 rpcrdma_decode_msg(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep,
1294                    struct rpc_rqst *rqst)
1295 {
1296         struct xdr_stream *xdr = &rep->rr_stream;
1297         u32 writelist, replychunk, rpclen;
1298         char *base;
1299
1300         /* Decode the chunk lists */
1301         if (decode_read_list(xdr))
1302                 return -EIO;
1303         if (decode_write_list(xdr, &writelist))
1304                 return -EIO;
1305         if (decode_reply_chunk(xdr, &replychunk))
1306                 return -EIO;
1307
1308         /* RDMA_MSG sanity checks */
1309         if (unlikely(replychunk))
1310                 return -EIO;
1311
1312         /* Build the RPC reply's Payload stream in rqst->rq_rcv_buf */
1313         base = (char *)xdr_inline_decode(xdr, 0);
1314         rpclen = xdr_stream_remaining(xdr);
1315         r_xprt->rx_stats.fixup_copy_count +=
1316                 rpcrdma_inline_fixup(rqst, base, rpclen, writelist & 3);
1317
1318         r_xprt->rx_stats.total_rdma_reply += writelist;
1319         return rpclen + xdr_align_size(writelist);
1320 }
1321
1322 static noinline int
1323 rpcrdma_decode_nomsg(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep)
1324 {
1325         struct xdr_stream *xdr = &rep->rr_stream;
1326         u32 writelist, replychunk;
1327
1328         /* Decode the chunk lists */
1329         if (decode_read_list(xdr))
1330                 return -EIO;
1331         if (decode_write_list(xdr, &writelist))
1332                 return -EIO;
1333         if (decode_reply_chunk(xdr, &replychunk))
1334                 return -EIO;
1335
1336         /* RDMA_NOMSG sanity checks */
1337         if (unlikely(writelist))
1338                 return -EIO;
1339         if (unlikely(!replychunk))
1340                 return -EIO;
1341
1342         /* Reply chunk buffer already is the reply vector */
1343         r_xprt->rx_stats.total_rdma_reply += replychunk;
1344         return replychunk;
1345 }
1346
1347 static noinline int
1348 rpcrdma_decode_error(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep,
1349                      struct rpc_rqst *rqst)
1350 {
1351         struct xdr_stream *xdr = &rep->rr_stream;
1352         __be32 *p;
1353
1354         p = xdr_inline_decode(xdr, sizeof(*p));
1355         if (unlikely(!p))
1356                 return -EIO;
1357
1358         switch (*p) {
1359         case err_vers:
1360                 p = xdr_inline_decode(xdr, 2 * sizeof(*p));
1361                 if (!p)
1362                         break;
1363                 dprintk("RPC:       %s: server reports "
1364                         "version error (%u-%u), xid %08x\n", __func__,
1365                         be32_to_cpup(p), be32_to_cpu(*(p + 1)),
1366                         be32_to_cpu(rep->rr_xid));
1367                 break;
1368         case err_chunk:
1369                 dprintk("RPC:       %s: server reports "
1370                         "header decoding error, xid %08x\n", __func__,
1371                         be32_to_cpu(rep->rr_xid));
1372                 break;
1373         default:
1374                 dprintk("RPC:       %s: server reports "
1375                         "unrecognized error %d, xid %08x\n", __func__,
1376                         be32_to_cpup(p), be32_to_cpu(rep->rr_xid));
1377         }
1378
1379         r_xprt->rx_stats.bad_reply_count++;
1380         return -EREMOTEIO;
1381 }
1382
1383 /* Perform XID lookup, reconstruction of the RPC reply, and
1384  * RPC completion while holding the transport lock to ensure
1385  * the rep, rqst, and rq_task pointers remain stable.
1386  */
1387 void rpcrdma_complete_rqst(struct rpcrdma_rep *rep)
1388 {
1389         struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;
1390         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1391         struct rpc_rqst *rqst = rep->rr_rqst;
1392         int status;
1393
1394         switch (rep->rr_proc) {
1395         case rdma_msg:
1396                 status = rpcrdma_decode_msg(r_xprt, rep, rqst);
1397                 break;
1398         case rdma_nomsg:
1399                 status = rpcrdma_decode_nomsg(r_xprt, rep);
1400                 break;
1401         case rdma_error:
1402                 status = rpcrdma_decode_error(r_xprt, rep, rqst);
1403                 break;
1404         default:
1405                 status = -EIO;
1406         }
1407         if (status < 0)
1408                 goto out_badheader;
1409
1410 out:
1411         spin_lock(&xprt->queue_lock);
1412         xprt_complete_rqst(rqst->rq_task, status);
1413         xprt_unpin_rqst(rqst);
1414         spin_unlock(&xprt->queue_lock);
1415         return;
1416
1417 /* If the incoming reply terminated a pending RPC, the next
1418  * RPC call will post a replacement receive buffer as it is
1419  * being marshaled.
1420  */
1421 out_badheader:
1422         trace_xprtrdma_reply_hdr(rep);
1423         r_xprt->rx_stats.bad_reply_count++;
1424         goto out;
1425 }
1426
1427 static void rpcrdma_reply_done(struct kref *kref)
1428 {
1429         struct rpcrdma_req *req =
1430                 container_of(kref, struct rpcrdma_req, rl_kref);
1431
1432         rpcrdma_complete_rqst(req->rl_reply);
1433 }
1434
1435 /**
1436  * rpcrdma_reply_handler - Process received RPC/RDMA messages
1437  * @rep: Incoming rpcrdma_rep object to process
1438  *
1439  * Errors must result in the RPC task either being awakened, or
1440  * allowed to timeout, to discover the errors at that time.
1441  */
1442 void rpcrdma_reply_handler(struct rpcrdma_rep *rep)
1443 {
1444         struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;
1445         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1446         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1447         struct rpcrdma_req *req;
1448         struct rpc_rqst *rqst;
1449         u32 credits;
1450         __be32 *p;
1451
1452         /* Any data means we had a useful conversation, so
1453          * then we don't need to delay the next reconnect.
1454          */
1455         if (xprt->reestablish_timeout)
1456                 xprt->reestablish_timeout = 0;
1457
1458         /* Fixed transport header fields */
1459         xdr_init_decode(&rep->rr_stream, &rep->rr_hdrbuf,
1460                         rep->rr_hdrbuf.head[0].iov_base, NULL);
1461         p = xdr_inline_decode(&rep->rr_stream, 4 * sizeof(*p));
1462         if (unlikely(!p))
1463                 goto out_shortreply;
1464         rep->rr_xid = *p++;
1465         rep->rr_vers = *p++;
1466         credits = be32_to_cpu(*p++);
1467         rep->rr_proc = *p++;
1468
1469         if (rep->rr_vers != rpcrdma_version)
1470                 goto out_badversion;
1471
1472         if (rpcrdma_is_bcall(r_xprt, rep))
1473                 return;
1474
1475         /* Match incoming rpcrdma_rep to an rpcrdma_req to
1476          * get context for handling any incoming chunks.
1477          */
1478         spin_lock(&xprt->queue_lock);
1479         rqst = xprt_lookup_rqst(xprt, rep->rr_xid);
1480         if (!rqst)
1481                 goto out_norqst;
1482         xprt_pin_rqst(rqst);
1483         spin_unlock(&xprt->queue_lock);
1484
1485         if (credits == 0)
1486                 credits = 1;    /* don't deadlock */
1487         else if (credits > buf->rb_max_requests)
1488                 credits = buf->rb_max_requests;
1489         if (buf->rb_credits != credits)
1490                 rpcrdma_update_cwnd(r_xprt, credits);
1491         rpcrdma_post_recvs(r_xprt, false);
1492
1493         req = rpcr_to_rdmar(rqst);
1494         if (req->rl_reply) {
1495                 trace_xprtrdma_leaked_rep(rqst, req->rl_reply);
1496                 rpcrdma_recv_buffer_put(req->rl_reply);
1497         }
1498         req->rl_reply = rep;
1499         rep->rr_rqst = rqst;
1500
1501         trace_xprtrdma_reply(rqst->rq_task, rep, req, credits);
1502
1503         if (rep->rr_wc_flags & IB_WC_WITH_INVALIDATE)
1504                 frwr_reminv(rep, &req->rl_registered);
1505         if (!list_empty(&req->rl_registered))
1506                 frwr_unmap_async(r_xprt, req);
1507                 /* LocalInv completion will complete the RPC */
1508         else
1509                 kref_put(&req->rl_kref, rpcrdma_reply_done);
1510         return;
1511
1512 out_badversion:
1513         trace_xprtrdma_reply_vers(rep);
1514         goto out;
1515
1516 out_norqst:
1517         spin_unlock(&xprt->queue_lock);
1518         trace_xprtrdma_reply_rqst(rep);
1519         goto out;
1520
1521 out_shortreply:
1522         trace_xprtrdma_reply_short(rep);
1523
1524 out:
1525         rpcrdma_recv_buffer_put(rep);
1526 }