net/sunrpc/xprtrdma/rpc_rdma.c

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