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)
 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 ib_sge *sge;
 568
 569         if (!sc->sc_unmap_count)
 570                 return;
 571
 572         /* The first two SGEs contain the transport header and
 573          * the inline buffer. These are always left mapped so
 574          * they can be cheaply re-used.
 575          */
 576         for (sge = &sc->sc_sges[2]; sc->sc_unmap_count;
 577              ++sge, --sc->sc_unmap_count)
 578                 ib_dma_unmap_page(sc->sc_device, sge->addr, sge->length,
 579                                   DMA_TO_DEVICE);
 580
 581         kref_put(&sc->sc_req->rl_kref, rpcrdma_sendctx_done);
 582 }
 583
 584 /* Prepare an SGE for the RPC-over-RDMA transport header.
 585  */
 586 static bool rpcrdma_prepare_hdr_sge(struct rpcrdma_xprt *r_xprt,
 587                                     struct rpcrdma_req *req, u32 len)
 588 {
 589         struct rpcrdma_sendctx *sc = req->rl_sendctx;
 590         struct rpcrdma_regbuf *rb = req->rl_rdmabuf;
 591         struct ib_sge *sge = sc->sc_sges;
 592
 593         if (!rpcrdma_regbuf_dma_map(r_xprt, rb))
 594                 goto out_regbuf;
 595         sge->addr = rdmab_addr(rb);
 596         sge->length = len;
 597         sge->lkey = rdmab_lkey(rb);
 598
 599         ib_dma_sync_single_for_device(rdmab_device(rb), sge->addr, sge->length,
 600                                       DMA_TO_DEVICE);
 601         sc->sc_wr.num_sge++;
 602         return true;
 603
 604 out_regbuf:
 605         pr_err("rpcrdma: failed to DMA map a Send buffer\n");
 606         return false;
 607 }
 608
 609 /* Prepare the Send SGEs. The head and tail iovec, and each entry
 610  * in the page list, gets its own SGE.
 611  */
 612 static bool rpcrdma_prepare_msg_sges(struct rpcrdma_xprt *r_xprt,
 613                                      struct rpcrdma_req *req,
 614                                      struct xdr_buf *xdr,
 615                                      enum rpcrdma_chunktype rtype)
 616 {
 617         struct rpcrdma_sendctx *sc = req->rl_sendctx;
 618         unsigned int sge_no, page_base, len, remaining;
 619         struct rpcrdma_regbuf *rb = req->rl_sendbuf;
 620         struct ib_sge *sge = sc->sc_sges;
 621         struct page *page, **ppages;
 622
 623         /* The head iovec is straightforward, as it is already
 624          * DMA-mapped. Sync the content that has changed.
 625          */
 626         if (!rpcrdma_regbuf_dma_map(r_xprt, rb))
 627                 goto out_regbuf;
 628         sc->sc_device = rdmab_device(rb);
 629         sge_no = 1;
 630         sge[sge_no].addr = rdmab_addr(rb);
 631         sge[sge_no].length = xdr->head[0].iov_len;
 632         sge[sge_no].lkey = rdmab_lkey(rb);
 633         ib_dma_sync_single_for_device(rdmab_device(rb), sge[sge_no].addr,
 634                                       sge[sge_no].length, DMA_TO_DEVICE);
 635
 636         /* If there is a Read chunk, the page list is being handled
 637          * via explicit RDMA, and thus is skipped here. However, the
 638          * tail iovec may include an XDR pad for the page list, as
 639          * well as additional content, and may not reside in the
 640          * same page as the head iovec.
 641          */
 642         if (rtype == rpcrdma_readch) {
 643                 len = xdr->tail[0].iov_len;
 644
 645                 /* Do not include the tail if it is only an XDR pad */
 646                 if (len < 4)
 647                         goto out;
 648
 649                 page = virt_to_page(xdr->tail[0].iov_base);
 650                 page_base = offset_in_page(xdr->tail[0].iov_base);
 651
 652                 /* If the content in the page list is an odd length,
 653                  * xdr_write_pages() has added a pad at the beginning
 654                  * of the tail iovec. Force the tail's non-pad content
 655                  * to land at the next XDR position in the Send message.
 656                  */
 657                 page_base += len & 3;
 658                 len -= len & 3;
 659                 goto map_tail;
 660         }
 661
 662         /* If there is a page list present, temporarily DMA map
 663          * and prepare an SGE for each page to be sent.
 664          */
 665         if (xdr->page_len) {
 666                 ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT);
 667                 page_base = offset_in_page(xdr->page_base);
 668                 remaining = xdr->page_len;
 669                 while (remaining) {
 670                         sge_no++;
 671                         if (sge_no > RPCRDMA_MAX_SEND_SGES - 2)
 672                                 goto out_mapping_overflow;
 673
 674                         len = min_t(u32, PAGE_SIZE - page_base, remaining);
 675                         sge[sge_no].addr =
 676                                 ib_dma_map_page(rdmab_device(rb), *ppages,
 677                                                 page_base, len, DMA_TO_DEVICE);
 678                         if (ib_dma_mapping_error(rdmab_device(rb),
 679                                                  sge[sge_no].addr))
 680                                 goto out_mapping_err;
 681                         sge[sge_no].length = len;
 682                         sge[sge_no].lkey = rdmab_lkey(rb);
 683
 684                         sc->sc_unmap_count++;
 685                         ppages++;
 686                         remaining -= len;
 687                         page_base = 0;
 688                 }
 689         }
 690
 691         /* The tail iovec is not always constructed in the same
 692          * page where the head iovec resides (see, for example,
 693          * gss_wrap_req_priv). To neatly accommodate that case,
 694          * DMA map it separately.
 695          */
 696         if (xdr->tail[0].iov_len) {
 697                 page = virt_to_page(xdr->tail[0].iov_base);
 698                 page_base = offset_in_page(xdr->tail[0].iov_base);
 699                 len = xdr->tail[0].iov_len;
 700
 701 map_tail:
 702                 sge_no++;
 703                 sge[sge_no].addr =
 704                         ib_dma_map_page(rdmab_device(rb), page, page_base, len,
 705                                         DMA_TO_DEVICE);
 706                 if (ib_dma_mapping_error(rdmab_device(rb), sge[sge_no].addr))
 707                         goto out_mapping_err;
 708                 sge[sge_no].length = len;
 709                 sge[sge_no].lkey = rdmab_lkey(rb);
 710                 sc->sc_unmap_count++;
 711         }
 712
 713 out:
 714         sc->sc_wr.num_sge += sge_no;
 715         if (sc->sc_unmap_count)
 716                 kref_get(&req->rl_kref);
 717         return true;
 718
 719 out_regbuf:
 720         pr_err("rpcrdma: failed to DMA map a Send buffer\n");
 721         return false;
 722
 723 out_mapping_overflow:
 724         rpcrdma_sendctx_unmap(sc);
 725         pr_err("rpcrdma: too many Send SGEs (%u)\n", sge_no);
 726         return false;
 727
 728 out_mapping_err:
 729         rpcrdma_sendctx_unmap(sc);
 730         trace_xprtrdma_dma_maperr(sge[sge_no].addr);
 731         return false;
 732 }
 733
 734 /**
 735  * rpcrdma_prepare_send_sges - Construct SGEs for a Send WR
 736  * @r_xprt: controlling transport
 737  * @req: context of RPC Call being marshalled
 738  * @hdrlen: size of transport header, in bytes
 739  * @xdr: xdr_buf containing RPC Call
 740  * @rtype: chunk type being encoded
 741  *
 742  * Returns 0 on success; otherwise a negative errno is returned.
 743  */
 744 int
 745 rpcrdma_prepare_send_sges(struct rpcrdma_xprt *r_xprt,
 746                           struct rpcrdma_req *req, u32 hdrlen,
 747                           struct xdr_buf *xdr, enum rpcrdma_chunktype rtype)
 748 {
 749         int ret;
 750
 751         ret = -EAGAIN;
 752         req->rl_sendctx = rpcrdma_sendctx_get_locked(r_xprt);
 753         if (!req->rl_sendctx)
 754                 goto err;
 755         req->rl_sendctx->sc_wr.num_sge = 0;
 756         req->rl_sendctx->sc_unmap_count = 0;
 757         req->rl_sendctx->sc_req = req;
 758         kref_init(&req->rl_kref);
 759
 760         ret = -EIO;
 761         if (!rpcrdma_prepare_hdr_sge(r_xprt, req, hdrlen))
 762                 goto err;
 763         if (rtype != rpcrdma_areadch)
 764                 if (!rpcrdma_prepare_msg_sges(r_xprt, req, xdr, rtype))
 765                         goto err;
 766         return 0;
 767
 768 err:
 769         trace_xprtrdma_prepsend_failed(&req->rl_slot, ret);
 770         return ret;
 771 }
 772
 773 /**
 774  * rpcrdma_marshal_req - Marshal and send one RPC request
 775  * @r_xprt: controlling transport
 776  * @rqst: RPC request to be marshaled
 777  *
 778  * For the RPC in "rqst", this function:
 779  *  - Chooses the transfer mode (eg., RDMA_MSG or RDMA_NOMSG)
 780  *  - Registers Read, Write, and Reply chunks
 781  *  - Constructs the transport header
 782  *  - Posts a Send WR to send the transport header and request
 783  *
 784  * Returns:
 785  *      %0 if the RPC was sent successfully,
 786  *      %-ENOTCONN if the connection was lost,
 787  *      %-EAGAIN if the caller should call again with the same arguments,
 788  *      %-ENOBUFS if the caller should call again after a delay,
 789  *      %-EMSGSIZE if the transport header is too small,
 790  *      %-EIO if a permanent problem occurred while marshaling.
 791  */
 792 int
 793 rpcrdma_marshal_req(struct rpcrdma_xprt *r_xprt, struct rpc_rqst *rqst)
 794 {
 795         struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
 796         struct xdr_stream *xdr = &req->rl_stream;
 797         enum rpcrdma_chunktype rtype, wtype;
 798         bool ddp_allowed;
 799         __be32 *p;
 800         int ret;
 801
 802         rpcrdma_set_xdrlen(&req->rl_hdrbuf, 0);
 803         xdr_init_encode(xdr, &req->rl_hdrbuf, rdmab_data(req->rl_rdmabuf),
 804                         rqst);
 805
 806         /* Fixed header fields */
 807         ret = -EMSGSIZE;
 808         p = xdr_reserve_space(xdr, 4 * sizeof(*p));
 809         if (!p)
 810                 goto out_err;
 811         *p++ = rqst->rq_xid;
 812         *p++ = rpcrdma_version;
 813         *p++ = cpu_to_be32(r_xprt->rx_buf.rb_max_requests);
 814
 815         /* When the ULP employs a GSS flavor that guarantees integrity
 816          * or privacy, direct data placement of individual data items
 817          * is not allowed.
 818          */
 819         ddp_allowed = !(rqst->rq_cred->cr_auth->au_flags &
 820                                                 RPCAUTH_AUTH_DATATOUCH);
 821
 822         /*
 823          * Chunks needed for results?
 824          *
 825          * o If the expected result is under the inline threshold, all ops
 826          *   return as inline.
 827          * o Large read ops return data as write chunk(s), header as
 828          *   inline.
 829          * o Large non-read ops return as a single reply chunk.
 830          */
 831         if (rpcrdma_results_inline(r_xprt, rqst))
 832                 wtype = rpcrdma_noch;
 833         else if ((ddp_allowed && rqst->rq_rcv_buf.flags & XDRBUF_READ) &&
 834                  rpcrdma_nonpayload_inline(r_xprt, rqst))
 835                 wtype = rpcrdma_writech;
 836         else
 837                 wtype = rpcrdma_replych;
 838
 839         /*
 840          * Chunks needed for arguments?
 841          *
 842          * o If the total request is under the inline threshold, all ops
 843          *   are sent as inline.
 844          * o Large write ops transmit data as read chunk(s), header as
 845          *   inline.
 846          * o Large non-write ops are sent with the entire message as a
 847          *   single read chunk (protocol 0-position special case).
 848          *
 849          * This assumes that the upper layer does not present a request
 850          * that both has a data payload, and whose non-data arguments
 851          * by themselves are larger than the inline threshold.
 852          */
 853         if (rpcrdma_args_inline(r_xprt, rqst)) {
 854                 *p++ = rdma_msg;
 855                 rtype = rpcrdma_noch;
 856         } else if (ddp_allowed && rqst->rq_snd_buf.flags & XDRBUF_WRITE) {
 857                 *p++ = rdma_msg;
 858                 rtype = rpcrdma_readch;
 859         } else {
 860                 r_xprt->rx_stats.nomsg_call_count++;
 861                 *p++ = rdma_nomsg;
 862                 rtype = rpcrdma_areadch;
 863         }
 864
 865         /* This implementation supports the following combinations
 866          * of chunk lists in one RPC-over-RDMA Call message:
 867          *
 868          *   - Read list
 869          *   - Write list
 870          *   - Reply chunk
 871          *   - Read list + Reply chunk
 872          *
 873          * It might not yet support the following combinations:
 874          *
 875          *   - Read list + Write list
 876          *
 877          * It does not support the following combinations:
 878          *
 879          *   - Write list + Reply chunk
 880          *   - Read list + Write list + Reply chunk
 881          *
 882          * This implementation supports only a single chunk in each
 883          * Read or Write list. Thus for example the client cannot
 884          * send a Call message with a Position Zero Read chunk and a
 885          * regular Read chunk at the same time.
 886          */
 887         ret = rpcrdma_encode_read_list(r_xprt, req, rqst, rtype);
 888         if (ret)
 889                 goto out_err;
 890         ret = rpcrdma_encode_write_list(r_xprt, req, rqst, wtype);
 891         if (ret)
 892                 goto out_err;
 893         ret = rpcrdma_encode_reply_chunk(r_xprt, req, rqst, wtype);
 894         if (ret)
 895                 goto out_err;
 896
 897         ret = rpcrdma_prepare_send_sges(r_xprt, req, req->rl_hdrbuf.len,
 898                                         &rqst->rq_snd_buf, rtype);
 899         if (ret)
 900                 goto out_err;
 901
 902         trace_xprtrdma_marshal(req, rtype, wtype);
 903         return 0;
 904
 905 out_err:
 906         trace_xprtrdma_marshal_failed(rqst, ret);
 907         r_xprt->rx_stats.failed_marshal_count++;
 908         frwr_reset(req);
 909         return ret;
 910 }
 911
 912 static void __rpcrdma_update_cwnd_locked(struct rpc_xprt *xprt,
 913                                          struct rpcrdma_buffer *buf,
 914                                          u32 grant)
 915 {
 916         buf->rb_credits = grant;
 917         xprt->cwnd = grant << RPC_CWNDSHIFT;
 918 }
 919
 920 static void rpcrdma_update_cwnd(struct rpcrdma_xprt *r_xprt, u32 grant)
 921 {
 922         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
 923
 924         spin_lock(&xprt->transport_lock);
 925         __rpcrdma_update_cwnd_locked(xprt, &r_xprt->rx_buf, grant);
 926         spin_unlock(&xprt->transport_lock);
 927 }
 928
 929 /**
 930  * rpcrdma_reset_cwnd - Reset the xprt's congestion window
 931  * @r_xprt: controlling transport instance
 932  *
 933  * Prepare @r_xprt for the next connection by reinitializing
 934  * its credit grant to one (see RFC 8166, Section 3.3.3).
 935  */
 936 void rpcrdma_reset_cwnd(struct rpcrdma_xprt *r_xprt)
 937 {
 938         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
 939
 940         spin_lock(&xprt->transport_lock);
 941         xprt->cong = 0;
 942         __rpcrdma_update_cwnd_locked(xprt, &r_xprt->rx_buf, 1);
 943         spin_unlock(&xprt->transport_lock);
 944 }
 945
 946 /**
 947  * rpcrdma_inline_fixup - Scatter inline received data into rqst's iovecs
 948  * @rqst: controlling RPC request
 949  * @srcp: points to RPC message payload in receive buffer
 950  * @copy_len: remaining length of receive buffer content
 951  * @pad: Write chunk pad bytes needed (zero for pure inline)
 952  *
 953  * The upper layer has set the maximum number of bytes it can
 954  * receive in each component of rq_rcv_buf. These values are set in
 955  * the head.iov_len, page_len, tail.iov_len, and buflen fields.
 956  *
 957  * Unlike the TCP equivalent (xdr_partial_copy_from_skb), in
 958  * many cases this function simply updates iov_base pointers in
 959  * rq_rcv_buf to point directly to the received reply data, to
 960  * avoid copying reply data.
 961  *
 962  * Returns the count of bytes which had to be memcopied.
 963  */
 964 static unsigned long
 965 rpcrdma_inline_fixup(struct rpc_rqst *rqst, char *srcp, int copy_len, int pad)
 966 {
 967         unsigned long fixup_copy_count;
 968         int i, npages, curlen;
 969         char *destp;
 970         struct page **ppages;
 971         int page_base;
 972
 973         /* The head iovec is redirected to the RPC reply message
 974          * in the receive buffer, to avoid a memcopy.
 975          */
 976         rqst->rq_rcv_buf.head[0].iov_base = srcp;
 977         rqst->rq_private_buf.head[0].iov_base = srcp;
 978
 979         /* The contents of the receive buffer that follow
 980          * head.iov_len bytes are copied into the page list.
 981          */
 982         curlen = rqst->rq_rcv_buf.head[0].iov_len;
 983         if (curlen > copy_len)
 984                 curlen = copy_len;
 985         trace_xprtrdma_fixup(rqst, copy_len, curlen);
 986         srcp += curlen;
 987         copy_len -= curlen;
 988
 989         ppages = rqst->rq_rcv_buf.pages +
 990                 (rqst->rq_rcv_buf.page_base >> PAGE_SHIFT);
 991         page_base = offset_in_page(rqst->rq_rcv_buf.page_base);
 992         fixup_copy_count = 0;
 993         if (copy_len && rqst->rq_rcv_buf.page_len) {
 994                 int pagelist_len;
 995
 996                 pagelist_len = rqst->rq_rcv_buf.page_len;
 997                 if (pagelist_len > copy_len)
 998                         pagelist_len = copy_len;
 999                 npages = PAGE_ALIGN(page_base + pagelist_len) >> PAGE_SHIFT;
1000                 for (i = 0; i < npages; i++) {
1001                         curlen = PAGE_SIZE - page_base;
1002                         if (curlen > pagelist_len)
1003                                 curlen = pagelist_len;
1004
1005                         trace_xprtrdma_fixup_pg(rqst, i, srcp,
1006                                                 copy_len, curlen);
1007                         destp = kmap_atomic(ppages[i]);
1008                         memcpy(destp + page_base, srcp, curlen);
1009                         flush_dcache_page(ppages[i]);
1010                         kunmap_atomic(destp);
1011                         srcp += curlen;
1012                         copy_len -= curlen;
1013                         fixup_copy_count += curlen;
1014                         pagelist_len -= curlen;
1015                         if (!pagelist_len)
1016                                 break;
1017                         page_base = 0;
1018                 }
1019
1020                 /* Implicit padding for the last segment in a Write
1021                  * chunk is inserted inline at the front of the tail
1022                  * iovec. The upper layer ignores the content of
1023                  * the pad. Simply ensure inline content in the tail
1024                  * that follows the Write chunk is properly aligned.
1025                  */
1026                 if (pad)
1027                         srcp -= pad;
1028         }
1029
1030         /* The tail iovec is redirected to the remaining data
1031          * in the receive buffer, to avoid a memcopy.
1032          */
1033         if (copy_len || pad) {
1034                 rqst->rq_rcv_buf.tail[0].iov_base = srcp;
1035                 rqst->rq_private_buf.tail[0].iov_base = srcp;
1036         }
1037
1038         return fixup_copy_count;
1039 }
1040
1041 /* By convention, backchannel calls arrive via rdma_msg type
1042  * messages, and never populate the chunk lists. This makes
1043  * the RPC/RDMA header small and fixed in size, so it is
1044  * straightforward to check the RPC header's direction field.
1045  */
1046 static bool
1047 rpcrdma_is_bcall(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep)
1048 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1049 {
1050         struct xdr_stream *xdr = &rep->rr_stream;
1051         __be32 *p;
1052
1053         if (rep->rr_proc != rdma_msg)
1054                 return false;
1055
1056         /* Peek at stream contents without advancing. */
1057         p = xdr_inline_decode(xdr, 0);
1058
1059         /* Chunk lists */
1060         if (*p++ != xdr_zero)
1061                 return false;
1062         if (*p++ != xdr_zero)
1063                 return false;
1064         if (*p++ != xdr_zero)
1065                 return false;
1066
1067         /* RPC header */
1068         if (*p++ != rep->rr_xid)
1069                 return false;
1070         if (*p != cpu_to_be32(RPC_CALL))
1071                 return false;
1072
1073         /* Now that we are sure this is a backchannel call,
1074          * advance to the RPC header.
1075          */
1076         p = xdr_inline_decode(xdr, 3 * sizeof(*p));
1077         if (unlikely(!p))
1078                 goto out_short;
1079
1080         rpcrdma_bc_receive_call(r_xprt, rep);
1081         return true;
1082
1083 out_short:
1084         pr_warn("RPC/RDMA short backward direction call\n");
1085         return true;
1086 }
1087 #else   /* CONFIG_SUNRPC_BACKCHANNEL */
1088 {
1089         return false;
1090 }
1091 #endif  /* CONFIG_SUNRPC_BACKCHANNEL */
1092
1093 static int decode_rdma_segment(struct xdr_stream *xdr, u32 *length)
1094 {
1095         u32 handle;
1096         u64 offset;
1097         __be32 *p;
1098
1099         p = xdr_inline_decode(xdr, 4 * sizeof(*p));
1100         if (unlikely(!p))
1101                 return -EIO;
1102
1103         handle = be32_to_cpup(p++);
1104         *length = be32_to_cpup(p++);
1105         xdr_decode_hyper(p, &offset);
1106
1107         trace_xprtrdma_decode_seg(handle, *length, offset);
1108         return 0;
1109 }
1110
1111 static int decode_write_chunk(struct xdr_stream *xdr, u32 *length)
1112 {
1113         u32 segcount, seglength;
1114         __be32 *p;
1115
1116         p = xdr_inline_decode(xdr, sizeof(*p));
1117         if (unlikely(!p))
1118                 return -EIO;
1119
1120         *length = 0;
1121         segcount = be32_to_cpup(p);
1122         while (segcount--) {
1123                 if (decode_rdma_segment(xdr, &seglength))
1124                         return -EIO;
1125                 *length += seglength;
1126         }
1127
1128         return 0;
1129 }
1130
1131 /* In RPC-over-RDMA Version One replies, a Read list is never
1132  * expected. This decoder is a stub that returns an error if
1133  * a Read list is present.
1134  */
1135 static int decode_read_list(struct xdr_stream *xdr)
1136 {
1137         __be32 *p;
1138
1139         p = xdr_inline_decode(xdr, sizeof(*p));
1140         if (unlikely(!p))
1141                 return -EIO;
1142         if (unlikely(*p != xdr_zero))
1143                 return -EIO;
1144         return 0;
1145 }
1146
1147 /* Supports only one Write chunk in the Write list
1148  */
1149 static int decode_write_list(struct xdr_stream *xdr, u32 *length)
1150 {
1151         u32 chunklen;
1152         bool first;
1153         __be32 *p;
1154
1155         *length = 0;
1156         first = true;
1157         do {
1158                 p = xdr_inline_decode(xdr, sizeof(*p));
1159                 if (unlikely(!p))
1160                         return -EIO;
1161                 if (*p == xdr_zero)
1162                         break;
1163                 if (!first)
1164                         return -EIO;
1165
1166                 if (decode_write_chunk(xdr, &chunklen))
1167                         return -EIO;
1168                 *length += chunklen;
1169                 first = false;
1170         } while (true);
1171         return 0;
1172 }
1173
1174 static int decode_reply_chunk(struct xdr_stream *xdr, u32 *length)
1175 {
1176         __be32 *p;
1177
1178         p = xdr_inline_decode(xdr, sizeof(*p));
1179         if (unlikely(!p))
1180                 return -EIO;
1181
1182         *length = 0;
1183         if (*p != xdr_zero)
1184                 if (decode_write_chunk(xdr, length))
1185                         return -EIO;
1186         return 0;
1187 }
1188
1189 static int
1190 rpcrdma_decode_msg(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep,
1191                    struct rpc_rqst *rqst)
1192 {
1193         struct xdr_stream *xdr = &rep->rr_stream;
1194         u32 writelist, replychunk, rpclen;
1195         char *base;
1196
1197         /* Decode the chunk lists */
1198         if (decode_read_list(xdr))
1199                 return -EIO;
1200         if (decode_write_list(xdr, &writelist))
1201                 return -EIO;
1202         if (decode_reply_chunk(xdr, &replychunk))
1203                 return -EIO;
1204
1205         /* RDMA_MSG sanity checks */
1206         if (unlikely(replychunk))
1207                 return -EIO;
1208
1209         /* Build the RPC reply's Payload stream in rqst->rq_rcv_buf */
1210         base = (char *)xdr_inline_decode(xdr, 0);
1211         rpclen = xdr_stream_remaining(xdr);
1212         r_xprt->rx_stats.fixup_copy_count +=
1213                 rpcrdma_inline_fixup(rqst, base, rpclen, writelist & 3);
1214
1215         r_xprt->rx_stats.total_rdma_reply += writelist;
1216         return rpclen + xdr_align_size(writelist);
1217 }
1218
1219 static noinline int
1220 rpcrdma_decode_nomsg(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep)
1221 {
1222         struct xdr_stream *xdr = &rep->rr_stream;
1223         u32 writelist, replychunk;
1224
1225         /* Decode the chunk lists */
1226         if (decode_read_list(xdr))
1227                 return -EIO;
1228         if (decode_write_list(xdr, &writelist))
1229                 return -EIO;
1230         if (decode_reply_chunk(xdr, &replychunk))
1231                 return -EIO;
1232
1233         /* RDMA_NOMSG sanity checks */
1234         if (unlikely(writelist))
1235                 return -EIO;
1236         if (unlikely(!replychunk))
1237                 return -EIO;
1238
1239         /* Reply chunk buffer already is the reply vector */
1240         r_xprt->rx_stats.total_rdma_reply += replychunk;
1241         return replychunk;
1242 }
1243
1244 static noinline int
1245 rpcrdma_decode_error(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep,
1246                      struct rpc_rqst *rqst)
1247 {
1248         struct xdr_stream *xdr = &rep->rr_stream;
1249         __be32 *p;
1250
1251         p = xdr_inline_decode(xdr, sizeof(*p));
1252         if (unlikely(!p))
1253                 return -EIO;
1254
1255         switch (*p) {
1256         case err_vers:
1257                 p = xdr_inline_decode(xdr, 2 * sizeof(*p));
1258                 if (!p)
1259                         break;
1260                 dprintk("RPC:       %s: server reports "
1261                         "version error (%u-%u), xid %08x\n", __func__,
1262                         be32_to_cpup(p), be32_to_cpu(*(p + 1)),
1263                         be32_to_cpu(rep->rr_xid));
1264                 break;
1265         case err_chunk:
1266                 dprintk("RPC:       %s: server reports "
1267                         "header decoding error, xid %08x\n", __func__,
1268                         be32_to_cpu(rep->rr_xid));
1269                 break;
1270         default:
1271                 dprintk("RPC:       %s: server reports "
1272                         "unrecognized error %d, xid %08x\n", __func__,
1273                         be32_to_cpup(p), be32_to_cpu(rep->rr_xid));
1274         }
1275
1276         r_xprt->rx_stats.bad_reply_count++;
1277         return -EREMOTEIO;
1278 }
1279
1280 /* Perform XID lookup, reconstruction of the RPC reply, and
1281  * RPC completion while holding the transport lock to ensure
1282  * the rep, rqst, and rq_task pointers remain stable.
1283  */
1284 void rpcrdma_complete_rqst(struct rpcrdma_rep *rep)
1285 {
1286         struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;
1287         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1288         struct rpc_rqst *rqst = rep->rr_rqst;
1289         int status;
1290
1291         switch (rep->rr_proc) {
1292         case rdma_msg:
1293                 status = rpcrdma_decode_msg(r_xprt, rep, rqst);
1294                 break;
1295         case rdma_nomsg:
1296                 status = rpcrdma_decode_nomsg(r_xprt, rep);
1297                 break;
1298         case rdma_error:
1299                 status = rpcrdma_decode_error(r_xprt, rep, rqst);
1300                 break;
1301         default:
1302                 status = -EIO;
1303         }
1304         if (status < 0)
1305                 goto out_badheader;
1306
1307 out:
1308         spin_lock(&xprt->queue_lock);
1309         xprt_complete_rqst(rqst->rq_task, status);
1310         xprt_unpin_rqst(rqst);
1311         spin_unlock(&xprt->queue_lock);
1312         return;
1313
1314 /* If the incoming reply terminated a pending RPC, the next
1315  * RPC call will post a replacement receive buffer as it is
1316  * being marshaled.
1317  */
1318 out_badheader:
1319         trace_xprtrdma_reply_hdr(rep);
1320         r_xprt->rx_stats.bad_reply_count++;
1321         goto out;
1322 }
1323
1324 static void rpcrdma_reply_done(struct kref *kref)
1325 {
1326         struct rpcrdma_req *req =
1327                 container_of(kref, struct rpcrdma_req, rl_kref);
1328
1329         rpcrdma_complete_rqst(req->rl_reply);
1330 }
1331
1332 /**
1333  * rpcrdma_reply_handler - Process received RPC/RDMA messages
1334  * @rep: Incoming rpcrdma_rep object to process
1335  *
1336  * Errors must result in the RPC task either being awakened, or
1337  * allowed to timeout, to discover the errors at that time.
1338  */
1339 void rpcrdma_reply_handler(struct rpcrdma_rep *rep)
1340 {
1341         struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;
1342         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1343         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1344         struct rpcrdma_req *req;
1345         struct rpc_rqst *rqst;
1346         u32 credits;
1347         __be32 *p;
1348
1349         /* Any data means we had a useful conversation, so
1350          * then we don't need to delay the next reconnect.
1351          */
1352         if (xprt->reestablish_timeout)
1353                 xprt->reestablish_timeout = 0;
1354
1355         /* Fixed transport header fields */
1356         xdr_init_decode(&rep->rr_stream, &rep->rr_hdrbuf,
1357                         rep->rr_hdrbuf.head[0].iov_base, NULL);
1358         p = xdr_inline_decode(&rep->rr_stream, 4 * sizeof(*p));
1359         if (unlikely(!p))
1360                 goto out_shortreply;
1361         rep->rr_xid = *p++;
1362         rep->rr_vers = *p++;
1363         credits = be32_to_cpu(*p++);
1364         rep->rr_proc = *p++;
1365
1366         if (rep->rr_vers != rpcrdma_version)
1367                 goto out_badversion;
1368
1369         if (rpcrdma_is_bcall(r_xprt, rep))
1370                 return;
1371
1372         /* Match incoming rpcrdma_rep to an rpcrdma_req to
1373          * get context for handling any incoming chunks.
1374          */
1375         spin_lock(&xprt->queue_lock);
1376         rqst = xprt_lookup_rqst(xprt, rep->rr_xid);
1377         if (!rqst)
1378                 goto out_norqst;
1379         xprt_pin_rqst(rqst);
1380         spin_unlock(&xprt->queue_lock);
1381
1382         if (credits == 0)
1383                 credits = 1;    /* don't deadlock */
1384         else if (credits > buf->rb_max_requests)
1385                 credits = buf->rb_max_requests;
1386         if (buf->rb_credits != credits)
1387                 rpcrdma_update_cwnd(r_xprt, credits);
1388         rpcrdma_post_recvs(r_xprt, false);
1389
1390         req = rpcr_to_rdmar(rqst);
1391         if (req->rl_reply) {
1392                 trace_xprtrdma_leaked_rep(rqst, req->rl_reply);
1393                 rpcrdma_recv_buffer_put(req->rl_reply);
1394         }
1395         req->rl_reply = rep;
1396         rep->rr_rqst = rqst;
1397
1398         trace_xprtrdma_reply(rqst->rq_task, rep, req, credits);
1399
1400         if (rep->rr_wc_flags & IB_WC_WITH_INVALIDATE)
1401                 frwr_reminv(rep, &req->rl_registered);
1402         if (!list_empty(&req->rl_registered))
1403                 frwr_unmap_async(r_xprt, req);
1404                 /* LocalInv completion will complete the RPC */
1405         else
1406                 kref_put(&req->rl_kref, rpcrdma_reply_done);
1407         return;
1408
1409 out_badversion:
1410         trace_xprtrdma_reply_vers(rep);
1411         goto out;
1412
1413 out_norqst:
1414         spin_unlock(&xprt->queue_lock);
1415         trace_xprtrdma_reply_rqst(rep);
1416         goto out;
1417
1418 out_shortreply:
1419         trace_xprtrdma_reply_short(rep);
1420
1421 out:
1422         rpcrdma_recv_buffer_put(rep);
1423 }