net/sunrpc/xprtrdma/rpc_rdma.c

   1 /*
   2  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
   3  *
   4  * This software is available to you under a choice of one of two
   5  * licenses.  You may choose to be licensed under the terms of the GNU
   6  * General Public License (GPL) Version 2, available from the file
   7  * COPYING in the main directory of this source tree, or the BSD-type
   8  * license below:
   9  *
  10  * Redistribution and use in source and binary forms, with or without
  11  * modification, are permitted provided that the following conditions
  12  * are met:
  13  *
  14  *      Redistributions of source code must retain the above copyright
  15  *      notice, this list of conditions and the following disclaimer.
  16  *
  17  *      Redistributions in binary form must reproduce the above
  18  *      copyright notice, this list of conditions and the following
  19  *      disclaimer in the documentation and/or other materials provided
  20  *      with the distribution.
  21  *
  22  *      Neither the name of the Network Appliance, Inc. nor the names of
  23  *      its contributors may be used to endorse or promote products
  24  *      derived from this software without specific prior written
  25  *      permission.
  26  *
  27  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  28  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  29  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  30  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  31  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  32  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  33  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  34  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  35  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  36  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  37  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  38  */
  39
  40 /*
  41  * rpc_rdma.c
  42  *
  43  * This file contains the guts of the RPC RDMA protocol, and
  44  * does marshaling/unmarshaling, etc. It is also where interfacing
  45  * to the Linux RPC framework lives.
  46  */
  47
  48 #include "xprt_rdma.h"
  49
  50 #include <linux/highmem.h>
  51
  52 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  53 # define RPCDBG_FACILITY        RPCDBG_TRANS
  54 #endif
  55
  56 enum rpcrdma_chunktype {
  57         rpcrdma_noch = 0,
  58         rpcrdma_readch,
  59         rpcrdma_areadch,
  60         rpcrdma_writech,
  61         rpcrdma_replych
  62 };
  63
  64 static const char transfertypes[][12] = {
  65         "inline",       /* no chunks */
  66         "read list",    /* some argument via rdma read */
  67         "*read list",   /* entire request via rdma read */
  68         "write list",   /* some result via rdma write */
  69         "reply chunk"   /* entire reply via rdma write */
  70 };
  71
  72 /* Returns size of largest RPC-over-RDMA header in a Call message
  73  *
  74  * The largest Call header contains a full-size Read list and a
  75  * minimal Reply chunk.
  76  */
  77 static unsigned int rpcrdma_max_call_header_size(unsigned int maxsegs)
  78 {
  79         unsigned int size;
  80
  81         /* Fixed header fields and list discriminators */
  82         size = RPCRDMA_HDRLEN_MIN;
  83
  84         /* Maximum Read list size */
  85         maxsegs += 2;   /* segment for head and tail buffers */
  86         size = maxsegs * sizeof(struct rpcrdma_read_chunk);
  87
  88         /* Minimal Read chunk size */
  89         size += sizeof(__be32); /* segment count */
  90         size += sizeof(struct rpcrdma_segment);
  91         size += sizeof(__be32); /* list discriminator */
  92
  93         dprintk("RPC:       %s: max call header size = %u\n",
  94                 __func__, size);
  95         return size;
  96 }
  97
  98 /* Returns size of largest RPC-over-RDMA header in a Reply message
  99  *
 100  * There is only one Write list or one Reply chunk per Reply
 101  * message.  The larger list is the Write list.
 102  */
 103 static unsigned int rpcrdma_max_reply_header_size(unsigned int maxsegs)
 104 {
 105         unsigned int size;
 106
 107         /* Fixed header fields and list discriminators */
 108         size = RPCRDMA_HDRLEN_MIN;
 109
 110         /* Maximum Write list size */
 111         maxsegs += 2;   /* segment for head and tail buffers */
 112         size = sizeof(__be32);          /* segment count */
 113         size += maxsegs * sizeof(struct rpcrdma_segment);
 114         size += sizeof(__be32); /* list discriminator */
 115
 116         dprintk("RPC:       %s: max reply header size = %u\n",
 117                 __func__, size);
 118         return size;
 119 }
 120
 121 void rpcrdma_set_max_header_sizes(struct rpcrdma_ia *ia,
 122                                   struct rpcrdma_create_data_internal *cdata,
 123                                   unsigned int maxsegs)
 124 {
 125         ia->ri_max_inline_write = cdata->inline_wsize -
 126                                   rpcrdma_max_call_header_size(maxsegs);
 127         ia->ri_max_inline_read = cdata->inline_rsize -
 128                                  rpcrdma_max_reply_header_size(maxsegs);
 129 }
 130
 131 /* The client can send a request inline as long as the RPCRDMA header
 132  * plus the RPC call fit under the transport's inline limit. If the
 133  * combined call message size exceeds that limit, the client must use
 134  * the read chunk list for this operation.
 135  */
 136 static bool rpcrdma_args_inline(struct rpcrdma_xprt *r_xprt,
 137                                 struct rpc_rqst *rqst)
 138 {
 139         struct rpcrdma_ia *ia = &r_xprt->rx_ia;
 140
 141         return rqst->rq_snd_buf.len <= ia->ri_max_inline_write;
 142 }
 143
 144 /* The client can't know how large the actual reply will be. Thus it
 145  * plans for the largest possible reply for that particular ULP
 146  * operation. If the maximum combined reply message size exceeds that
 147  * limit, the client must provide a write list or a reply chunk for
 148  * this request.
 149  */
 150 static bool rpcrdma_results_inline(struct rpcrdma_xprt *r_xprt,
 151                                    struct rpc_rqst *rqst)
 152 {
 153         struct rpcrdma_ia *ia = &r_xprt->rx_ia;
 154
 155         return rqst->rq_rcv_buf.buflen <= ia->ri_max_inline_read;
 156 }
 157
 158 static int
 159 rpcrdma_tail_pullup(struct xdr_buf *buf)
 160 {
 161         size_t tlen = buf->tail[0].iov_len;
 162         size_t skip = tlen & 3;
 163
 164         /* Do not include the tail if it is only an XDR pad */
 165         if (tlen < 4)
 166                 return 0;
 167
 168         /* xdr_write_pages() adds a pad at the beginning of the tail
 169          * if the content in "buf->pages" is unaligned. Force the
 170          * tail's actual content to land at the next XDR position
 171          * after the head instead.
 172          */
 173         if (skip) {
 174                 unsigned char *src, *dst;
 175                 unsigned int count;
 176
 177                 src = buf->tail[0].iov_base;
 178                 dst = buf->head[0].iov_base;
 179                 dst += buf->head[0].iov_len;
 180
 181                 src += skip;
 182                 tlen -= skip;
 183
 184                 dprintk("RPC:       %s: skip=%zu, memmove(%p, %p, %zu)\n",
 185                         __func__, skip, dst, src, tlen);
 186
 187                 for (count = tlen; count; count--)
 188                         *dst++ = *src++;
 189         }
 190
 191         return tlen;
 192 }
 193
 194 /* Split "vec" on page boundaries into segments. FMR registers pages,
 195  * not a byte range. Other modes coalesce these segments into a single
 196  * MR when they can.
 197  */
 198 static int
 199 rpcrdma_convert_kvec(struct kvec *vec, struct rpcrdma_mr_seg *seg, int n)
 200 {
 201         size_t page_offset;
 202         u32 remaining;
 203         char *base;
 204
 205         base = vec->iov_base;
 206         page_offset = offset_in_page(base);
 207         remaining = vec->iov_len;
 208         while (remaining && n < RPCRDMA_MAX_SEGS) {
 209                 seg[n].mr_page = NULL;
 210                 seg[n].mr_offset = base;
 211                 seg[n].mr_len = min_t(u32, PAGE_SIZE - page_offset, remaining);
 212                 remaining -= seg[n].mr_len;
 213                 base += seg[n].mr_len;
 214                 ++n;
 215                 page_offset = 0;
 216         }
 217         return n;
 218 }
 219
 220 /*
 221  * Chunk assembly from upper layer xdr_buf.
 222  *
 223  * Prepare the passed-in xdr_buf into representation as RPC/RDMA chunk
 224  * elements. Segments are then coalesced when registered, if possible
 225  * within the selected memreg mode.
 226  *
 227  * Returns positive number of segments converted, or a negative errno.
 228  */
 229
 230 static int
 231 rpcrdma_convert_iovs(struct xdr_buf *xdrbuf, unsigned int pos,
 232         enum rpcrdma_chunktype type, struct rpcrdma_mr_seg *seg)
 233 {
 234         int len, n, p, page_base;
 235         struct page **ppages;
 236
 237         n = 0;
 238         if (pos == 0) {
 239                 n = rpcrdma_convert_kvec(&xdrbuf->head[0], seg, n);
 240                 if (n == RPCRDMA_MAX_SEGS)
 241                         goto out_overflow;
 242         }
 243
 244         len = xdrbuf->page_len;
 245         ppages = xdrbuf->pages + (xdrbuf->page_base >> PAGE_SHIFT);
 246         page_base = xdrbuf->page_base & ~PAGE_MASK;
 247         p = 0;
 248         while (len && n < RPCRDMA_MAX_SEGS) {
 249                 if (!ppages[p]) {
 250                         /* alloc the pagelist for receiving buffer */
 251                         ppages[p] = alloc_page(GFP_ATOMIC);
 252                         if (!ppages[p])
 253                                 return -EAGAIN;
 254                 }
 255                 seg[n].mr_page = ppages[p];
 256                 seg[n].mr_offset = (void *)(unsigned long) page_base;
 257                 seg[n].mr_len = min_t(u32, PAGE_SIZE - page_base, len);
 258                 if (seg[n].mr_len > PAGE_SIZE)
 259                         goto out_overflow;
 260                 len -= seg[n].mr_len;
 261                 ++n;
 262                 ++p;
 263                 page_base = 0;  /* page offset only applies to first page */
 264         }
 265
 266         /* Message overflows the seg array */
 267         if (len && n == RPCRDMA_MAX_SEGS)
 268                 goto out_overflow;
 269
 270         /* When encoding the read list, the tail is always sent inline */
 271         if (type == rpcrdma_readch)
 272                 return n;
 273
 274         if (xdrbuf->tail[0].iov_len) {
 275                 /* the rpcrdma protocol allows us to omit any trailing
 276                  * xdr pad bytes, saving the server an RDMA operation. */
 277                 if (xdrbuf->tail[0].iov_len < 4 && xprt_rdma_pad_optimize)
 278                         return n;
 279                 n = rpcrdma_convert_kvec(&xdrbuf->tail[0], seg, n);
 280                 if (n == RPCRDMA_MAX_SEGS)
 281                         goto out_overflow;
 282         }
 283
 284         return n;
 285
 286 out_overflow:
 287         pr_err("rpcrdma: segment array overflow\n");
 288         return -EIO;
 289 }
 290
 291 static inline __be32 *
 292 xdr_encode_rdma_segment(__be32 *iptr, struct rpcrdma_mw *mw)
 293 {
 294         *iptr++ = cpu_to_be32(mw->mw_handle);
 295         *iptr++ = cpu_to_be32(mw->mw_length);
 296         return xdr_encode_hyper(iptr, mw->mw_offset);
 297 }
 298
 299 /* XDR-encode the Read list. Supports encoding a list of read
 300  * segments that belong to a single read chunk.
 301  *
 302  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
 303  *
 304  *  Read chunklist (a linked list):
 305  *   N elements, position P (same P for all chunks of same arg!):
 306  *    1 - PHLOO - 1 - PHLOO - ... - 1 - PHLOO - 0
 307  *
 308  * Returns a pointer to the XDR word in the RDMA header following
 309  * the end of the Read list, or an error pointer.
 310  */
 311 static __be32 *
 312 rpcrdma_encode_read_list(struct rpcrdma_xprt *r_xprt,
 313                          struct rpcrdma_req *req, struct rpc_rqst *rqst,
 314                          __be32 *iptr, enum rpcrdma_chunktype rtype)
 315 {
 316         struct rpcrdma_mr_seg *seg;
 317         struct rpcrdma_mw *mw;
 318         unsigned int pos;
 319         int n, nsegs;
 320
 321         if (rtype == rpcrdma_noch) {
 322                 *iptr++ = xdr_zero;     /* item not present */
 323                 return iptr;
 324         }
 325
 326         pos = rqst->rq_snd_buf.head[0].iov_len;
 327         if (rtype == rpcrdma_areadch)
 328                 pos = 0;
 329         seg = req->rl_segments;
 330         nsegs = rpcrdma_convert_iovs(&rqst->rq_snd_buf, pos, rtype, seg);
 331         if (nsegs < 0)
 332                 return ERR_PTR(nsegs);
 333
 334         do {
 335                 n = r_xprt->rx_ia.ri_ops->ro_map(r_xprt, seg, nsegs,
 336                                                  false, &mw);
 337                 if (n < 0)
 338                         return ERR_PTR(n);
 339                 list_add(&mw->mw_list, &req->rl_registered);
 340
 341                 *iptr++ = xdr_one;      /* item present */
 342
 343                 /* All read segments in this chunk
 344                  * have the same "position".
 345                  */
 346                 *iptr++ = cpu_to_be32(pos);
 347                 iptr = xdr_encode_rdma_segment(iptr, mw);
 348
 349                 dprintk("RPC: %5u %s: pos %u %u@0x%016llx:0x%08x (%s)\n",
 350                         rqst->rq_task->tk_pid, __func__, pos,
 351                         mw->mw_length, (unsigned long long)mw->mw_offset,
 352                         mw->mw_handle, n < nsegs ? "more" : "last");
 353
 354                 r_xprt->rx_stats.read_chunk_count++;
 355                 seg += n;
 356                 nsegs -= n;
 357         } while (nsegs);
 358
 359         /* Finish Read list */
 360         *iptr++ = xdr_zero;     /* Next item not present */
 361         return iptr;
 362 }
 363
 364 /* XDR-encode the Write list. Supports encoding a list containing
 365  * one array of plain segments that belong to a single write chunk.
 366  *
 367  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
 368  *
 369  *  Write chunklist (a list of (one) counted array):
 370  *   N elements:
 371  *    1 - N - HLOO - HLOO - ... - HLOO - 0
 372  *
 373  * Returns a pointer to the XDR word in the RDMA header following
 374  * the end of the Write list, or an error pointer.
 375  */
 376 static __be32 *
 377 rpcrdma_encode_write_list(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req,
 378                           struct rpc_rqst *rqst, __be32 *iptr,
 379                           enum rpcrdma_chunktype wtype)
 380 {
 381         struct rpcrdma_mr_seg *seg;
 382         struct rpcrdma_mw *mw;
 383         int n, nsegs, nchunks;
 384         __be32 *segcount;
 385
 386         if (wtype != rpcrdma_writech) {
 387                 *iptr++ = xdr_zero;     /* no Write list present */
 388                 return iptr;
 389         }
 390
 391         seg = req->rl_segments;
 392         nsegs = rpcrdma_convert_iovs(&rqst->rq_rcv_buf,
 393                                      rqst->rq_rcv_buf.head[0].iov_len,
 394                                      wtype, seg);
 395         if (nsegs < 0)
 396                 return ERR_PTR(nsegs);
 397
 398         *iptr++ = xdr_one;      /* Write list present */
 399         segcount = iptr++;      /* save location of segment count */
 400
 401         nchunks = 0;
 402         do {
 403                 n = r_xprt->rx_ia.ri_ops->ro_map(r_xprt, seg, nsegs,
 404                                                  true, &mw);
 405                 if (n < 0)
 406                         return ERR_PTR(n);
 407                 list_add(&mw->mw_list, &req->rl_registered);
 408
 409                 iptr = xdr_encode_rdma_segment(iptr, mw);
 410
 411                 dprintk("RPC: %5u %s: %u@0x016%llx:0x%08x (%s)\n",
 412                         rqst->rq_task->tk_pid, __func__,
 413                         mw->mw_length, (unsigned long long)mw->mw_offset,
 414                         mw->mw_handle, n < nsegs ? "more" : "last");
 415
 416                 r_xprt->rx_stats.write_chunk_count++;
 417                 r_xprt->rx_stats.total_rdma_request += seg->mr_len;
 418                 nchunks++;
 419                 seg   += n;
 420                 nsegs -= n;
 421         } while (nsegs);
 422
 423         /* Update count of segments in this Write chunk */
 424         *segcount = cpu_to_be32(nchunks);
 425
 426         /* Finish Write list */
 427         *iptr++ = xdr_zero;     /* Next item not present */
 428         return iptr;
 429 }
 430
 431 /* XDR-encode the Reply chunk. Supports encoding an array of plain
 432  * segments that belong to a single write (reply) chunk.
 433  *
 434  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
 435  *
 436  *  Reply chunk (a counted array):
 437  *   N elements:
 438  *    1 - N - HLOO - HLOO - ... - HLOO
 439  *
 440  * Returns a pointer to the XDR word in the RDMA header following
 441  * the end of the Reply chunk, or an error pointer.
 442  */
 443 static __be32 *
 444 rpcrdma_encode_reply_chunk(struct rpcrdma_xprt *r_xprt,
 445                            struct rpcrdma_req *req, struct rpc_rqst *rqst,
 446                            __be32 *iptr, enum rpcrdma_chunktype wtype)
 447 {
 448         struct rpcrdma_mr_seg *seg;
 449         struct rpcrdma_mw *mw;
 450         int n, nsegs, nchunks;
 451         __be32 *segcount;
 452
 453         if (wtype != rpcrdma_replych) {
 454                 *iptr++ = xdr_zero;     /* no Reply chunk present */
 455                 return iptr;
 456         }
 457
 458         seg = req->rl_segments;
 459         nsegs = rpcrdma_convert_iovs(&rqst->rq_rcv_buf, 0, wtype, seg);
 460         if (nsegs < 0)
 461                 return ERR_PTR(nsegs);
 462
 463         *iptr++ = xdr_one;      /* Reply chunk present */
 464         segcount = iptr++;      /* save location of segment count */
 465
 466         nchunks = 0;
 467         do {
 468                 n = r_xprt->rx_ia.ri_ops->ro_map(r_xprt, seg, nsegs,
 469                                                  true, &mw);
 470                 if (n < 0)
 471                         return ERR_PTR(n);
 472                 list_add(&mw->mw_list, &req->rl_registered);
 473
 474                 iptr = xdr_encode_rdma_segment(iptr, mw);
 475
 476                 dprintk("RPC: %5u %s: %u@0x%016llx:0x%08x (%s)\n",
 477                         rqst->rq_task->tk_pid, __func__,
 478                         mw->mw_length, (unsigned long long)mw->mw_offset,
 479                         mw->mw_handle, n < nsegs ? "more" : "last");
 480
 481                 r_xprt->rx_stats.reply_chunk_count++;
 482                 r_xprt->rx_stats.total_rdma_request += seg->mr_len;
 483                 nchunks++;
 484                 seg   += n;
 485                 nsegs -= n;
 486         } while (nsegs);
 487
 488         /* Update count of segments in the Reply chunk */
 489         *segcount = cpu_to_be32(nchunks);
 490
 491         return iptr;
 492 }
 493
 494 /*
 495  * Copy write data inline.
 496  * This function is used for "small" requests. Data which is passed
 497  * to RPC via iovecs (or page list) is copied directly into the
 498  * pre-registered memory buffer for this request. For small amounts
 499  * of data, this is efficient. The cutoff value is tunable.
 500  */
 501 static void rpcrdma_inline_pullup(struct rpc_rqst *rqst)
 502 {
 503         int i, npages, curlen;
 504         int copy_len;
 505         unsigned char *srcp, *destp;
 506         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt);
 507         int page_base;
 508         struct page **ppages;
 509
 510         destp = rqst->rq_svec[0].iov_base;
 511         curlen = rqst->rq_svec[0].iov_len;
 512         destp += curlen;
 513
 514         dprintk("RPC:       %s: destp 0x%p len %d hdrlen %d\n",
 515                 __func__, destp, rqst->rq_slen, curlen);
 516
 517         copy_len = rqst->rq_snd_buf.page_len;
 518
 519         if (rqst->rq_snd_buf.tail[0].iov_len) {
 520                 curlen = rqst->rq_snd_buf.tail[0].iov_len;
 521                 if (destp + copy_len != rqst->rq_snd_buf.tail[0].iov_base) {
 522                         memmove(destp + copy_len,
 523                                 rqst->rq_snd_buf.tail[0].iov_base, curlen);
 524                         r_xprt->rx_stats.pullup_copy_count += curlen;
 525                 }
 526                 dprintk("RPC:       %s: tail destp 0x%p len %d\n",
 527                         __func__, destp + copy_len, curlen);
 528                 rqst->rq_svec[0].iov_len += curlen;
 529         }
 530         r_xprt->rx_stats.pullup_copy_count += copy_len;
 531
 532         page_base = rqst->rq_snd_buf.page_base;
 533         ppages = rqst->rq_snd_buf.pages + (page_base >> PAGE_SHIFT);
 534         page_base &= ~PAGE_MASK;
 535         npages = PAGE_ALIGN(page_base+copy_len) >> PAGE_SHIFT;
 536         for (i = 0; copy_len && i < npages; i++) {
 537                 curlen = PAGE_SIZE - page_base;
 538                 if (curlen > copy_len)
 539                         curlen = copy_len;
 540                 dprintk("RPC:       %s: page %d destp 0x%p len %d curlen %d\n",
 541                         __func__, i, destp, copy_len, curlen);
 542                 srcp = kmap_atomic(ppages[i]);
 543                 memcpy(destp, srcp+page_base, curlen);
 544                 kunmap_atomic(srcp);
 545                 rqst->rq_svec[0].iov_len += curlen;
 546                 destp += curlen;
 547                 copy_len -= curlen;
 548                 page_base = 0;
 549         }
 550         /* header now contains entire send message */
 551 }
 552
 553 /*
 554  * Marshal a request: the primary job of this routine is to choose
 555  * the transfer modes. See comments below.
 556  *
 557  * Prepares up to two IOVs per Call message:
 558  *
 559  *  [0] -- RPC RDMA header
 560  *  [1] -- the RPC header/data
 561  *
 562  * Returns zero on success, otherwise a negative errno.
 563  */
 564
 565 int
 566 rpcrdma_marshal_req(struct rpc_rqst *rqst)
 567 {
 568         struct rpc_xprt *xprt = rqst->rq_xprt;
 569         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 570         struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
 571         enum rpcrdma_chunktype rtype, wtype;
 572         struct rpcrdma_msg *headerp;
 573         ssize_t hdrlen;
 574         size_t rpclen;
 575         __be32 *iptr;
 576
 577 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
 578         if (test_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state))
 579                 return rpcrdma_bc_marshal_reply(rqst);
 580 #endif
 581
 582         headerp = rdmab_to_msg(req->rl_rdmabuf);
 583         /* don't byte-swap XID, it's already done in request */
 584         headerp->rm_xid = rqst->rq_xid;
 585         headerp->rm_vers = rpcrdma_version;
 586         headerp->rm_credit = cpu_to_be32(r_xprt->rx_buf.rb_max_requests);
 587         headerp->rm_type = rdma_msg;
 588
 589         /*
 590          * Chunks needed for results?
 591          *
 592          * o If the expected result is under the inline threshold, all ops
 593          *   return as inline.
 594          * o Large read ops return data as write chunk(s), header as
 595          *   inline.
 596          * o Large non-read ops return as a single reply chunk.
 597          */
 598         if (rpcrdma_results_inline(r_xprt, rqst))
 599                 wtype = rpcrdma_noch;
 600         else if (rqst->rq_rcv_buf.flags & XDRBUF_READ)
 601                 wtype = rpcrdma_writech;
 602         else
 603                 wtype = rpcrdma_replych;
 604
 605         /*
 606          * Chunks needed for arguments?
 607          *
 608          * o If the total request is under the inline threshold, all ops
 609          *   are sent as inline.
 610          * o Large write ops transmit data as read chunk(s), header as
 611          *   inline.
 612          * o Large non-write ops are sent with the entire message as a
 613          *   single read chunk (protocol 0-position special case).
 614          *
 615          * This assumes that the upper layer does not present a request
 616          * that both has a data payload, and whose non-data arguments
 617          * by themselves are larger than the inline threshold.
 618          */
 619         if (rpcrdma_args_inline(r_xprt, rqst)) {
 620                 rtype = rpcrdma_noch;
 621                 rpcrdma_inline_pullup(rqst);
 622                 rpclen = rqst->rq_svec[0].iov_len;
 623         } else if (rqst->rq_snd_buf.flags & XDRBUF_WRITE) {
 624                 rtype = rpcrdma_readch;
 625                 rpclen = rqst->rq_svec[0].iov_len;
 626                 rpclen += rpcrdma_tail_pullup(&rqst->rq_snd_buf);
 627         } else {
 628                 r_xprt->rx_stats.nomsg_call_count++;
 629                 headerp->rm_type = htonl(RDMA_NOMSG);
 630                 rtype = rpcrdma_areadch;
 631                 rpclen = 0;
 632         }
 633
 634         /* This implementation supports the following combinations
 635          * of chunk lists in one RPC-over-RDMA Call message:
 636          *
 637          *   - Read list
 638          *   - Write list
 639          *   - Reply chunk
 640          *   - Read list + Reply chunk
 641          *
 642          * It might not yet support the following combinations:
 643          *
 644          *   - Read list + Write list
 645          *
 646          * It does not support the following combinations:
 647          *
 648          *   - Write list + Reply chunk
 649          *   - Read list + Write list + Reply chunk
 650          *
 651          * This implementation supports only a single chunk in each
 652          * Read or Write list. Thus for example the client cannot
 653          * send a Call message with a Position Zero Read chunk and a
 654          * regular Read chunk at the same time.
 655          */
 656         iptr = headerp->rm_body.rm_chunks;
 657         iptr = rpcrdma_encode_read_list(r_xprt, req, rqst, iptr, rtype);
 658         if (IS_ERR(iptr))
 659                 goto out_unmap;
 660         iptr = rpcrdma_encode_write_list(r_xprt, req, rqst, iptr, wtype);
 661         if (IS_ERR(iptr))
 662                 goto out_unmap;
 663         iptr = rpcrdma_encode_reply_chunk(r_xprt, req, rqst, iptr, wtype);
 664         if (IS_ERR(iptr))
 665                 goto out_unmap;
 666         hdrlen = (unsigned char *)iptr - (unsigned char *)headerp;
 667
 668         if (hdrlen + rpclen > RPCRDMA_INLINE_WRITE_THRESHOLD(rqst))
 669                 goto out_overflow;
 670
 671         dprintk("RPC: %5u %s: %s/%s: hdrlen %zd rpclen %zd\n",
 672                 rqst->rq_task->tk_pid, __func__,
 673                 transfertypes[rtype], transfertypes[wtype],
 674                 hdrlen, rpclen);
 675
 676         req->rl_send_iov[0].addr = rdmab_addr(req->rl_rdmabuf);
 677         req->rl_send_iov[0].length = hdrlen;
 678         req->rl_send_iov[0].lkey = rdmab_lkey(req->rl_rdmabuf);
 679
 680         req->rl_niovs = 1;
 681         if (rtype == rpcrdma_areadch)
 682                 return 0;
 683
 684         req->rl_send_iov[1].addr = rdmab_addr(req->rl_sendbuf);
 685         req->rl_send_iov[1].length = rpclen;
 686         req->rl_send_iov[1].lkey = rdmab_lkey(req->rl_sendbuf);
 687
 688         req->rl_niovs = 2;
 689         return 0;
 690
 691 out_overflow:
 692         pr_err("rpcrdma: send overflow: hdrlen %zd rpclen %zu %s/%s\n",
 693                 hdrlen, rpclen, transfertypes[rtype], transfertypes[wtype]);
 694         iptr = ERR_PTR(-EIO);
 695
 696 out_unmap:
 697         r_xprt->rx_ia.ri_ops->ro_unmap_safe(r_xprt, req, false);
 698         return PTR_ERR(iptr);
 699 }
 700
 701 /*
 702  * Chase down a received write or reply chunklist to get length
 703  * RDMA'd by server. See map at rpcrdma_create_chunks()! :-)
 704  */
 705 static int
 706 rpcrdma_count_chunks(struct rpcrdma_rep *rep, int wrchunk, __be32 **iptrp)
 707 {
 708         unsigned int i, total_len;
 709         struct rpcrdma_write_chunk *cur_wchunk;
 710         char *base = (char *)rdmab_to_msg(rep->rr_rdmabuf);
 711
 712         i = be32_to_cpu(**iptrp);
 713         cur_wchunk = (struct rpcrdma_write_chunk *) (*iptrp + 1);
 714         total_len = 0;
 715         while (i--) {
 716                 struct rpcrdma_segment *seg = &cur_wchunk->wc_target;
 717                 ifdebug(FACILITY) {
 718                         u64 off;
 719                         xdr_decode_hyper((__be32 *)&seg->rs_offset, &off);
 720                         dprintk("RPC:       %s: chunk %d@0x%llx:0x%x\n",
 721                                 __func__,
 722                                 be32_to_cpu(seg->rs_length),
 723                                 (unsigned long long)off,
 724                                 be32_to_cpu(seg->rs_handle));
 725                 }
 726                 total_len += be32_to_cpu(seg->rs_length);
 727                 ++cur_wchunk;
 728         }
 729         /* check and adjust for properly terminated write chunk */
 730         if (wrchunk) {
 731                 __be32 *w = (__be32 *) cur_wchunk;
 732                 if (*w++ != xdr_zero)
 733                         return -1;
 734                 cur_wchunk = (struct rpcrdma_write_chunk *) w;
 735         }
 736         if ((char *)cur_wchunk > base + rep->rr_len)
 737                 return -1;
 738
 739         *iptrp = (__be32 *) cur_wchunk;
 740         return total_len;
 741 }
 742
 743 /*
 744  * Scatter inline received data back into provided iov's.
 745  */
 746 static void
 747 rpcrdma_inline_fixup(struct rpc_rqst *rqst, char *srcp, int copy_len, int pad)
 748 {
 749         int i, npages, curlen, olen;
 750         char *destp;
 751         struct page **ppages;
 752         int page_base;
 753
 754         curlen = rqst->rq_rcv_buf.head[0].iov_len;
 755         if (curlen > copy_len) {        /* write chunk header fixup */
 756                 curlen = copy_len;
 757                 rqst->rq_rcv_buf.head[0].iov_len = curlen;
 758         }
 759
 760         dprintk("RPC:       %s: srcp 0x%p len %d hdrlen %d\n",
 761                 __func__, srcp, copy_len, curlen);
 762
 763         /* Shift pointer for first receive segment only */
 764         rqst->rq_rcv_buf.head[0].iov_base = srcp;
 765         srcp += curlen;
 766         copy_len -= curlen;
 767
 768         olen = copy_len;
 769         i = 0;
 770         rpcx_to_rdmax(rqst->rq_xprt)->rx_stats.fixup_copy_count += olen;
 771         page_base = rqst->rq_rcv_buf.page_base;
 772         ppages = rqst->rq_rcv_buf.pages + (page_base >> PAGE_SHIFT);
 773         page_base &= ~PAGE_MASK;
 774
 775         if (copy_len && rqst->rq_rcv_buf.page_len) {
 776                 npages = PAGE_ALIGN(page_base +
 777                         rqst->rq_rcv_buf.page_len) >> PAGE_SHIFT;
 778                 for (; i < npages; i++) {
 779                         curlen = PAGE_SIZE - page_base;
 780                         if (curlen > copy_len)
 781                                 curlen = copy_len;
 782                         dprintk("RPC:       %s: page %d"
 783                                 " srcp 0x%p len %d curlen %d\n",
 784                                 __func__, i, srcp, copy_len, curlen);
 785                         destp = kmap_atomic(ppages[i]);
 786                         memcpy(destp + page_base, srcp, curlen);
 787                         flush_dcache_page(ppages[i]);
 788                         kunmap_atomic(destp);
 789                         srcp += curlen;
 790                         copy_len -= curlen;
 791                         if (copy_len == 0)
 792                                 break;
 793                         page_base = 0;
 794                 }
 795         }
 796
 797         if (copy_len && rqst->rq_rcv_buf.tail[0].iov_len) {
 798                 curlen = copy_len;
 799                 if (curlen > rqst->rq_rcv_buf.tail[0].iov_len)
 800                         curlen = rqst->rq_rcv_buf.tail[0].iov_len;
 801                 if (rqst->rq_rcv_buf.tail[0].iov_base != srcp)
 802                         memmove(rqst->rq_rcv_buf.tail[0].iov_base, srcp, curlen);
 803                 dprintk("RPC:       %s: tail srcp 0x%p len %d curlen %d\n",
 804                         __func__, srcp, copy_len, curlen);
 805                 rqst->rq_rcv_buf.tail[0].iov_len = curlen;
 806                 copy_len -= curlen; ++i;
 807         } else
 808                 rqst->rq_rcv_buf.tail[0].iov_len = 0;
 809
 810         if (pad) {
 811                 /* implicit padding on terminal chunk */
 812                 unsigned char *p = rqst->rq_rcv_buf.tail[0].iov_base;
 813                 while (pad--)
 814                         p[rqst->rq_rcv_buf.tail[0].iov_len++] = 0;
 815         }
 816
 817         if (copy_len)
 818                 dprintk("RPC:       %s: %d bytes in"
 819                         " %d extra segments (%d lost)\n",
 820                         __func__, olen, i, copy_len);
 821
 822         /* TBD avoid a warning from call_decode() */
 823         rqst->rq_private_buf = rqst->rq_rcv_buf;
 824 }
 825
 826 void
 827 rpcrdma_connect_worker(struct work_struct *work)
 828 {
 829         struct rpcrdma_ep *ep =
 830                 container_of(work, struct rpcrdma_ep, rep_connect_worker.work);
 831         struct rpcrdma_xprt *r_xprt =
 832                 container_of(ep, struct rpcrdma_xprt, rx_ep);
 833         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
 834
 835         spin_lock_bh(&xprt->transport_lock);
 836         if (++xprt->connect_cookie == 0)        /* maintain a reserved value */
 837                 ++xprt->connect_cookie;
 838         if (ep->rep_connected > 0) {
 839                 if (!xprt_test_and_set_connected(xprt))
 840                         xprt_wake_pending_tasks(xprt, 0);
 841         } else {
 842                 if (xprt_test_and_clear_connected(xprt))
 843                         xprt_wake_pending_tasks(xprt, -ENOTCONN);
 844         }
 845         spin_unlock_bh(&xprt->transport_lock);
 846 }
 847
 848 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
 849 /* By convention, backchannel calls arrive via rdma_msg type
 850  * messages, and never populate the chunk lists. This makes
 851  * the RPC/RDMA header small and fixed in size, so it is
 852  * straightforward to check the RPC header's direction field.
 853  */
 854 static bool
 855 rpcrdma_is_bcall(struct rpcrdma_msg *headerp)
 856 {
 857         __be32 *p = (__be32 *)headerp;
 858
 859         if (headerp->rm_type != rdma_msg)
 860                 return false;
 861         if (headerp->rm_body.rm_chunks[0] != xdr_zero)
 862                 return false;
 863         if (headerp->rm_body.rm_chunks[1] != xdr_zero)
 864                 return false;
 865         if (headerp->rm_body.rm_chunks[2] != xdr_zero)
 866                 return false;
 867
 868         /* sanity */
 869         if (p[7] != headerp->rm_xid)
 870                 return false;
 871         /* call direction */
 872         if (p[8] != cpu_to_be32(RPC_CALL))
 873                 return false;
 874
 875         return true;
 876 }
 877 #endif  /* CONFIG_SUNRPC_BACKCHANNEL */
 878
 879 /*
 880  * This function is called when an async event is posted to
 881  * the connection which changes the connection state. All it
 882  * does at this point is mark the connection up/down, the rpc
 883  * timers do the rest.
 884  */
 885 void
 886 rpcrdma_conn_func(struct rpcrdma_ep *ep)
 887 {
 888         schedule_delayed_work(&ep->rep_connect_worker, 0);
 889 }
 890
 891 /* Process received RPC/RDMA messages.
 892  *
 893  * Errors must result in the RPC task either being awakened, or
 894  * allowed to timeout, to discover the errors at that time.
 895  */
 896 void
 897 rpcrdma_reply_handler(struct rpcrdma_rep *rep)
 898 {
 899         struct rpcrdma_msg *headerp;
 900         struct rpcrdma_req *req;
 901         struct rpc_rqst *rqst;
 902         struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;
 903         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
 904         __be32 *iptr;
 905         int rdmalen, status, rmerr;
 906         unsigned long cwnd;
 907
 908         dprintk("RPC:       %s: incoming rep %p\n", __func__, rep);
 909
 910         if (rep->rr_len == RPCRDMA_BAD_LEN)
 911                 goto out_badstatus;
 912         if (rep->rr_len < RPCRDMA_HDRLEN_ERR)
 913                 goto out_shortreply;
 914
 915         headerp = rdmab_to_msg(rep->rr_rdmabuf);
 916 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
 917         if (rpcrdma_is_bcall(headerp))
 918                 goto out_bcall;
 919 #endif
 920
 921         /* Match incoming rpcrdma_rep to an rpcrdma_req to
 922          * get context for handling any incoming chunks.
 923          */
 924         spin_lock_bh(&xprt->transport_lock);
 925         rqst = xprt_lookup_rqst(xprt, headerp->rm_xid);
 926         if (!rqst)
 927                 goto out_nomatch;
 928
 929         req = rpcr_to_rdmar(rqst);
 930         if (req->rl_reply)
 931                 goto out_duplicate;
 932
 933         /* Sanity checking has passed. We are now committed
 934          * to complete this transaction.
 935          */
 936         list_del_init(&rqst->rq_list);
 937         spin_unlock_bh(&xprt->transport_lock);
 938         dprintk("RPC:       %s: reply %p completes request %p (xid 0x%08x)\n",
 939                 __func__, rep, req, be32_to_cpu(headerp->rm_xid));
 940
 941         /* from here on, the reply is no longer an orphan */
 942         req->rl_reply = rep;
 943         xprt->reestablish_timeout = 0;
 944
 945         if (headerp->rm_vers != rpcrdma_version)
 946                 goto out_badversion;
 947
 948         /* check for expected message types */
 949         /* The order of some of these tests is important. */
 950         switch (headerp->rm_type) {
 951         case rdma_msg:
 952                 /* never expect read chunks */
 953                 /* never expect reply chunks (two ways to check) */
 954                 /* never expect write chunks without having offered RDMA */
 955                 if (headerp->rm_body.rm_chunks[0] != xdr_zero ||
 956                     (headerp->rm_body.rm_chunks[1] == xdr_zero &&
 957                      headerp->rm_body.rm_chunks[2] != xdr_zero) ||
 958                     (headerp->rm_body.rm_chunks[1] != xdr_zero &&
 959                      list_empty(&req->rl_registered)))
 960                         goto badheader;
 961                 if (headerp->rm_body.rm_chunks[1] != xdr_zero) {
 962                         /* count any expected write chunks in read reply */
 963                         /* start at write chunk array count */
 964                         iptr = &headerp->rm_body.rm_chunks[2];
 965                         rdmalen = rpcrdma_count_chunks(rep, 1, &iptr);
 966                         /* check for validity, and no reply chunk after */
 967                         if (rdmalen < 0 || *iptr++ != xdr_zero)
 968                                 goto badheader;
 969                         rep->rr_len -=
 970                             ((unsigned char *)iptr - (unsigned char *)headerp);
 971                         status = rep->rr_len + rdmalen;
 972                         r_xprt->rx_stats.total_rdma_reply += rdmalen;
 973                         /* special case - last chunk may omit padding */
 974                         if (rdmalen &= 3) {
 975                                 rdmalen = 4 - rdmalen;
 976                                 status += rdmalen;
 977                         }
 978                 } else {
 979                         /* else ordinary inline */
 980                         rdmalen = 0;
 981                         iptr = (__be32 *)((unsigned char *)headerp +
 982                                                         RPCRDMA_HDRLEN_MIN);
 983                         rep->rr_len -= RPCRDMA_HDRLEN_MIN;
 984                         status = rep->rr_len;
 985                 }
 986                 /* Fix up the rpc results for upper layer */
 987                 rpcrdma_inline_fixup(rqst, (char *)iptr, rep->rr_len, rdmalen);
 988                 break;
 989
 990         case rdma_nomsg:
 991                 /* never expect read or write chunks, always reply chunks */
 992                 if (headerp->rm_body.rm_chunks[0] != xdr_zero ||
 993                     headerp->rm_body.rm_chunks[1] != xdr_zero ||
 994                     headerp->rm_body.rm_chunks[2] != xdr_one ||
 995                     list_empty(&req->rl_registered))
 996                         goto badheader;
 997                 iptr = (__be32 *)((unsigned char *)headerp +
 998                                                         RPCRDMA_HDRLEN_MIN);
 999                 rdmalen = rpcrdma_count_chunks(rep, 0, &iptr);
1000                 if (rdmalen < 0)
1001                         goto badheader;
1002                 r_xprt->rx_stats.total_rdma_reply += rdmalen;
1003                 /* Reply chunk buffer already is the reply vector - no fixup. */
1004                 status = rdmalen;
1005                 break;
1006
1007         case rdma_error:
1008                 goto out_rdmaerr;
1009
1010 badheader:
1011         default:
1012                 dprintk("RPC: %5u %s: invalid rpcrdma reply (type %u)\n",
1013                         rqst->rq_task->tk_pid, __func__,
1014                         be32_to_cpu(headerp->rm_type));
1015                 status = -EIO;
1016                 r_xprt->rx_stats.bad_reply_count++;
1017                 break;
1018         }
1019
1020 out:
1021         /* Invalidate and flush the data payloads before waking the
1022          * waiting application. This guarantees the memory region is
1023          * properly fenced from the server before the application
1024          * accesses the data. It also ensures proper send flow
1025          * control: waking the next RPC waits until this RPC has
1026          * relinquished all its Send Queue entries.
1027          */
1028         if (!list_empty(&req->rl_registered))
1029                 r_xprt->rx_ia.ri_ops->ro_unmap_sync(r_xprt, req);
1030
1031         spin_lock_bh(&xprt->transport_lock);
1032         cwnd = xprt->cwnd;
1033         xprt->cwnd = atomic_read(&r_xprt->rx_buf.rb_credits) << RPC_CWNDSHIFT;
1034         if (xprt->cwnd > cwnd)
1035                 xprt_release_rqst_cong(rqst->rq_task);
1036
1037         xprt_complete_rqst(rqst->rq_task, status);
1038         spin_unlock_bh(&xprt->transport_lock);
1039         dprintk("RPC:       %s: xprt_complete_rqst(0x%p, 0x%p, %d)\n",
1040                         __func__, xprt, rqst, status);
1041         return;
1042
1043 out_badstatus:
1044         rpcrdma_recv_buffer_put(rep);
1045         if (r_xprt->rx_ep.rep_connected == 1) {
1046                 r_xprt->rx_ep.rep_connected = -EIO;
1047                 rpcrdma_conn_func(&r_xprt->rx_ep);
1048         }
1049         return;
1050
1051 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1052 out_bcall:
1053         rpcrdma_bc_receive_call(r_xprt, rep);
1054         return;
1055 #endif
1056
1057 /* If the incoming reply terminated a pending RPC, the next
1058  * RPC call will post a replacement receive buffer as it is
1059  * being marshaled.
1060  */
1061 out_badversion:
1062         dprintk("RPC:       %s: invalid version %d\n",
1063                 __func__, be32_to_cpu(headerp->rm_vers));
1064         status = -EIO;
1065         r_xprt->rx_stats.bad_reply_count++;
1066         goto out;
1067
1068 out_rdmaerr:
1069         rmerr = be32_to_cpu(headerp->rm_body.rm_error.rm_err);
1070         switch (rmerr) {
1071         case ERR_VERS:
1072                 pr_err("%s: server reports header version error (%u-%u)\n",
1073                        __func__,
1074                        be32_to_cpu(headerp->rm_body.rm_error.rm_vers_low),
1075                        be32_to_cpu(headerp->rm_body.rm_error.rm_vers_high));
1076                 break;
1077         case ERR_CHUNK:
1078                 pr_err("%s: server reports header decoding error\n",
1079                        __func__);
1080                 break;
1081         default:
1082                 pr_err("%s: server reports unknown error %d\n",
1083                        __func__, rmerr);
1084         }
1085         status = -EREMOTEIO;
1086         r_xprt->rx_stats.bad_reply_count++;
1087         goto out;
1088
1089 /* If no pending RPC transaction was matched, post a replacement
1090  * receive buffer before returning.
1091  */
1092 out_shortreply:
1093         dprintk("RPC:       %s: short/invalid reply\n", __func__);
1094         goto repost;
1095
1096 out_nomatch:
1097         spin_unlock_bh(&xprt->transport_lock);
1098         dprintk("RPC:       %s: no match for incoming xid 0x%08x len %d\n",
1099                 __func__, be32_to_cpu(headerp->rm_xid),
1100                 rep->rr_len);
1101         goto repost;
1102
1103 out_duplicate:
1104         spin_unlock_bh(&xprt->transport_lock);
1105         dprintk("RPC:       %s: "
1106                 "duplicate reply %p to RPC request %p: xid 0x%08x\n",
1107                 __func__, rep, req, be32_to_cpu(headerp->rm_xid));
1108
1109 repost:
1110         r_xprt->rx_stats.bad_reply_count++;
1111         if (rpcrdma_ep_post_recv(&r_xprt->rx_ia, &r_xprt->rx_ep, rep))
1112                 rpcrdma_recv_buffer_put(rep);
1113 }