net/sunrpc/xprtrdma/svc_rdma_recvfrom.c

   1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
   2 /*
   3  * Copyright (c) 2016-2018 Oracle. All rights reserved.
   4  * Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved.
   5  * Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved.
   6  *
   7  * This software is available to you under a choice of one of two
   8  * licenses.  You may choose to be licensed under the terms of the GNU
   9  * General Public License (GPL) Version 2, available from the file
  10  * COPYING in the main directory of this source tree, or the BSD-type
  11  * license below:
  12  *
  13  * Redistribution and use in source and binary forms, with or without
  14  * modification, are permitted provided that the following conditions
  15  * are met:
  16  *
  17  *      Redistributions of source code must retain the above copyright
  18  *      notice, this list of conditions and the following disclaimer.
  19  *
  20  *      Redistributions in binary form must reproduce the above
  21  *      copyright notice, this list of conditions and the following
  22  *      disclaimer in the documentation and/or other materials provided
  23  *      with the distribution.
  24  *
  25  *      Neither the name of the Network Appliance, Inc. nor the names of
  26  *      its contributors may be used to endorse or promote products
  27  *      derived from this software without specific prior written
  28  *      permission.
  29  *
  30  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  31  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  32  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  33  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  34  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  35  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  36  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  37  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  38  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  39  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  40  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  41  *
  42  * Author: Tom Tucker <tom@opengridcomputing.com>
  43  */
  44
  45 /* Operation
  46  *
  47  * The main entry point is svc_rdma_recvfrom. This is called from
  48  * svc_recv when the transport indicates there is incoming data to
  49  * be read. "Data Ready" is signaled when an RDMA Receive completes,
  50  * or when a set of RDMA Reads complete.
  51  *
  52  * An svc_rqst is passed in. This structure contains an array of
  53  * free pages (rq_pages) that will contain the incoming RPC message.
  54  *
  55  * Short messages are moved directly into svc_rqst::rq_arg, and
  56  * the RPC Call is ready to be processed by the Upper Layer.
  57  * svc_rdma_recvfrom returns the length of the RPC Call message,
  58  * completing the reception of the RPC Call.
  59  *
  60  * However, when an incoming message has Read chunks,
  61  * svc_rdma_recvfrom must post RDMA Reads to pull the RPC Call's
  62  * data payload from the client. svc_rdma_recvfrom sets up the
  63  * RDMA Reads using pages in svc_rqst::rq_pages, which are
  64  * transferred to an svc_rdma_recv_ctxt for the duration of the
  65  * I/O. svc_rdma_recvfrom then returns zero, since the RPC message
  66  * is still not yet ready.
  67  *
  68  * When the Read chunk payloads have become available on the
  69  * server, "Data Ready" is raised again, and svc_recv calls
  70  * svc_rdma_recvfrom again. This second call may use a different
  71  * svc_rqst than the first one, thus any information that needs
  72  * to be preserved across these two calls is kept in an
  73  * svc_rdma_recv_ctxt.
  74  *
  75  * The second call to svc_rdma_recvfrom performs final assembly
  76  * of the RPC Call message, using the RDMA Read sink pages kept in
  77  * the svc_rdma_recv_ctxt. The xdr_buf is copied from the
  78  * svc_rdma_recv_ctxt to the second svc_rqst. The second call returns
  79  * the length of the completed RPC Call message.
  80  *
  81  * Page Management
  82  *
  83  * Pages under I/O must be transferred from the first svc_rqst to an
  84  * svc_rdma_recv_ctxt before the first svc_rdma_recvfrom call returns.
  85  *
  86  * The first svc_rqst supplies pages for RDMA Reads. These are moved
  87  * from rqstp::rq_pages into ctxt::pages. The consumed elements of
  88  * the rq_pages array are set to NULL and refilled with the first
  89  * svc_rdma_recvfrom call returns.
  90  *
  91  * During the second svc_rdma_recvfrom call, RDMA Read sink pages
  92  * are transferred from the svc_rdma_recv_ctxt to the second svc_rqst
  93  * (see rdma_read_complete() below).
  94  */
  95
  96 #include <linux/slab.h>
  97 #include <linux/spinlock.h>
  98 #include <asm/unaligned.h>
  99 #include <rdma/ib_verbs.h>
 100 #include <rdma/rdma_cm.h>
 101
 102 #include <linux/sunrpc/xdr.h>
 103 #include <linux/sunrpc/debug.h>
 104 #include <linux/sunrpc/rpc_rdma.h>
 105 #include <linux/sunrpc/svc_rdma.h>
 106
 107 #include "xprt_rdma.h"
 108 #include <trace/events/rpcrdma.h>
 109
 110 #define RPCDBG_FACILITY RPCDBG_SVCXPRT
 111
 112 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc);
 113
 114 static inline struct svc_rdma_recv_ctxt *
 115 svc_rdma_next_recv_ctxt(struct list_head *list)
 116 {
 117         return list_first_entry_or_null(list, struct svc_rdma_recv_ctxt,
 118                                         rc_list);
 119 }
 120
 121 static void svc_rdma_recv_cid_init(struct svcxprt_rdma *rdma,
 122                                    struct rpc_rdma_cid *cid)
 123 {
 124         cid->ci_queue_id = rdma->sc_rq_cq->res.id;
 125         cid->ci_completion_id = atomic_inc_return(&rdma->sc_completion_ids);
 126 }
 127
 128 static struct svc_rdma_recv_ctxt *
 129 svc_rdma_recv_ctxt_alloc(struct svcxprt_rdma *rdma)
 130 {
 131         struct svc_rdma_recv_ctxt *ctxt;
 132         dma_addr_t addr;
 133         void *buffer;
 134
 135         ctxt = kmalloc(sizeof(*ctxt), GFP_KERNEL);
 136         if (!ctxt)
 137                 goto fail0;
 138         buffer = kmalloc(rdma->sc_max_req_size, GFP_KERNEL);
 139         if (!buffer)
 140                 goto fail1;
 141         addr = ib_dma_map_single(rdma->sc_pd->device, buffer,
 142                                  rdma->sc_max_req_size, DMA_FROM_DEVICE);
 143         if (ib_dma_mapping_error(rdma->sc_pd->device, addr))
 144                 goto fail2;
 145
 146         svc_rdma_recv_cid_init(rdma, &ctxt->rc_cid);
 147         pcl_init(&ctxt->rc_call_pcl);
 148         pcl_init(&ctxt->rc_read_pcl);
 149         pcl_init(&ctxt->rc_write_pcl);
 150         pcl_init(&ctxt->rc_reply_pcl);
 151
 152         ctxt->rc_recv_wr.next = NULL;
 153         ctxt->rc_recv_wr.wr_cqe = &ctxt->rc_cqe;
 154         ctxt->rc_recv_wr.sg_list = &ctxt->rc_recv_sge;
 155         ctxt->rc_recv_wr.num_sge = 1;
 156         ctxt->rc_cqe.done = svc_rdma_wc_receive;
 157         ctxt->rc_recv_sge.addr = addr;
 158         ctxt->rc_recv_sge.length = rdma->sc_max_req_size;
 159         ctxt->rc_recv_sge.lkey = rdma->sc_pd->local_dma_lkey;
 160         ctxt->rc_recv_buf = buffer;
 161         ctxt->rc_temp = false;
 162         return ctxt;
 163
 164 fail2:
 165         kfree(buffer);
 166 fail1:
 167         kfree(ctxt);
 168 fail0:
 169         return NULL;
 170 }
 171
 172 static void svc_rdma_recv_ctxt_destroy(struct svcxprt_rdma *rdma,
 173                                        struct svc_rdma_recv_ctxt *ctxt)
 174 {
 175         ib_dma_unmap_single(rdma->sc_pd->device, ctxt->rc_recv_sge.addr,
 176                             ctxt->rc_recv_sge.length, DMA_FROM_DEVICE);
 177         kfree(ctxt->rc_recv_buf);
 178         kfree(ctxt);
 179 }
 180
 181 /**
 182  * svc_rdma_recv_ctxts_destroy - Release all recv_ctxt's for an xprt
 183  * @rdma: svcxprt_rdma being torn down
 184  *
 185  */
 186 void svc_rdma_recv_ctxts_destroy(struct svcxprt_rdma *rdma)
 187 {
 188         struct svc_rdma_recv_ctxt *ctxt;
 189         struct llist_node *node;
 190
 191         while ((node = llist_del_first(&rdma->sc_recv_ctxts))) {
 192                 ctxt = llist_entry(node, struct svc_rdma_recv_ctxt, rc_node);
 193                 svc_rdma_recv_ctxt_destroy(rdma, ctxt);
 194         }
 195 }
 196
 197 /**
 198  * svc_rdma_recv_ctxt_get - Allocate a recv_ctxt
 199  * @rdma: controlling svcxprt_rdma
 200  *
 201  * Returns a recv_ctxt or (rarely) NULL if none are available.
 202  */
 203 struct svc_rdma_recv_ctxt *svc_rdma_recv_ctxt_get(struct svcxprt_rdma *rdma)
 204 {
 205         struct svc_rdma_recv_ctxt *ctxt;
 206         struct llist_node *node;
 207
 208         node = llist_del_first(&rdma->sc_recv_ctxts);
 209         if (!node)
 210                 goto out_empty;
 211         ctxt = llist_entry(node, struct svc_rdma_recv_ctxt, rc_node);
 212
 213 out:
 214         ctxt->rc_page_count = 0;
 215         return ctxt;
 216
 217 out_empty:
 218         ctxt = svc_rdma_recv_ctxt_alloc(rdma);
 219         if (!ctxt)
 220                 return NULL;
 221         goto out;
 222 }
 223
 224 /**
 225  * svc_rdma_recv_ctxt_put - Return recv_ctxt to free list
 226  * @rdma: controlling svcxprt_rdma
 227  * @ctxt: object to return to the free list
 228  *
 229  */
 230 void svc_rdma_recv_ctxt_put(struct svcxprt_rdma *rdma,
 231                             struct svc_rdma_recv_ctxt *ctxt)
 232 {
 233         unsigned int i;
 234
 235         for (i = 0; i < ctxt->rc_page_count; i++)
 236                 put_page(ctxt->rc_pages[i]);
 237
 238         pcl_free(&ctxt->rc_call_pcl);
 239         pcl_free(&ctxt->rc_read_pcl);
 240         pcl_free(&ctxt->rc_write_pcl);
 241         pcl_free(&ctxt->rc_reply_pcl);
 242
 243         if (!ctxt->rc_temp)
 244                 llist_add(&ctxt->rc_node, &rdma->sc_recv_ctxts);
 245         else
 246                 svc_rdma_recv_ctxt_destroy(rdma, ctxt);
 247 }
 248
 249 /**
 250  * svc_rdma_release_rqst - Release transport-specific per-rqst resources
 251  * @rqstp: svc_rqst being released
 252  *
 253  * Ensure that the recv_ctxt is released whether or not a Reply
 254  * was sent. For example, the client could close the connection,
 255  * or svc_process could drop an RPC, before the Reply is sent.
 256  */
 257 void svc_rdma_release_rqst(struct svc_rqst *rqstp)
 258 {
 259         struct svc_rdma_recv_ctxt *ctxt = rqstp->rq_xprt_ctxt;
 260         struct svc_xprt *xprt = rqstp->rq_xprt;
 261         struct svcxprt_rdma *rdma =
 262                 container_of(xprt, struct svcxprt_rdma, sc_xprt);
 263
 264         rqstp->rq_xprt_ctxt = NULL;
 265         if (ctxt)
 266                 svc_rdma_recv_ctxt_put(rdma, ctxt);
 267 }
 268
 269 static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
 270                                 struct svc_rdma_recv_ctxt *ctxt)
 271 {
 272         int ret;
 273
 274         trace_svcrdma_post_recv(ctxt);
 275         ret = ib_post_recv(rdma->sc_qp, &ctxt->rc_recv_wr, NULL);
 276         if (ret)
 277                 goto err_post;
 278         return 0;
 279
 280 err_post:
 281         trace_svcrdma_rq_post_err(rdma, ret);
 282         svc_rdma_recv_ctxt_put(rdma, ctxt);
 283         return ret;
 284 }
 285
 286 static int svc_rdma_post_recv(struct svcxprt_rdma *rdma)
 287 {
 288         struct svc_rdma_recv_ctxt *ctxt;
 289
 290         if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))
 291                 return 0;
 292         ctxt = svc_rdma_recv_ctxt_get(rdma);
 293         if (!ctxt)
 294                 return -ENOMEM;
 295         return __svc_rdma_post_recv(rdma, ctxt);
 296 }
 297
 298 /**
 299  * svc_rdma_post_recvs - Post initial set of Recv WRs
 300  * @rdma: fresh svcxprt_rdma
 301  *
 302  * Returns true if successful, otherwise false.
 303  */
 304 bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma)
 305 {
 306         struct svc_rdma_recv_ctxt *ctxt;
 307         unsigned int i;
 308         int ret;
 309
 310         for (i = 0; i < rdma->sc_max_requests; i++) {
 311                 ctxt = svc_rdma_recv_ctxt_get(rdma);
 312                 if (!ctxt)
 313                         return false;
 314                 ctxt->rc_temp = true;
 315                 ret = __svc_rdma_post_recv(rdma, ctxt);
 316                 if (ret)
 317                         return false;
 318         }
 319         return true;
 320 }
 321
 322 /**
 323  * svc_rdma_wc_receive - Invoked by RDMA provider for each polled Receive WC
 324  * @cq: Completion Queue context
 325  * @wc: Work Completion object
 326  *
 327  * NB: The svc_xprt/svcxprt_rdma is pinned whenever it's possible that
 328  * the Receive completion handler could be running.
 329  */
 330 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
 331 {
 332         struct svcxprt_rdma *rdma = cq->cq_context;
 333         struct ib_cqe *cqe = wc->wr_cqe;
 334         struct svc_rdma_recv_ctxt *ctxt;
 335
 336         /* WARNING: Only wc->wr_cqe and wc->status are reliable */
 337         ctxt = container_of(cqe, struct svc_rdma_recv_ctxt, rc_cqe);
 338
 339         trace_svcrdma_wc_receive(wc, &ctxt->rc_cid);
 340         if (wc->status != IB_WC_SUCCESS)
 341                 goto flushed;
 342
 343         if (svc_rdma_post_recv(rdma))
 344                 goto post_err;
 345
 346         /* All wc fields are now known to be valid */
 347         ctxt->rc_byte_len = wc->byte_len;
 348
 349         spin_lock(&rdma->sc_rq_dto_lock);
 350         list_add_tail(&ctxt->rc_list, &rdma->sc_rq_dto_q);
 351         /* Note the unlock pairs with the smp_rmb in svc_xprt_ready: */
 352         set_bit(XPT_DATA, &rdma->sc_xprt.xpt_flags);
 353         spin_unlock(&rdma->sc_rq_dto_lock);
 354         if (!test_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags))
 355                 svc_xprt_enqueue(&rdma->sc_xprt);
 356         return;
 357
 358 flushed:
 359 post_err:
 360         svc_rdma_recv_ctxt_put(rdma, ctxt);
 361         set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
 362         svc_xprt_enqueue(&rdma->sc_xprt);
 363 }
 364
 365 /**
 366  * svc_rdma_flush_recv_queues - Drain pending Receive work
 367  * @rdma: svcxprt_rdma being shut down
 368  *
 369  */
 370 void svc_rdma_flush_recv_queues(struct svcxprt_rdma *rdma)
 371 {
 372         struct svc_rdma_recv_ctxt *ctxt;
 373
 374         while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_read_complete_q))) {
 375                 list_del(&ctxt->rc_list);
 376                 svc_rdma_recv_ctxt_put(rdma, ctxt);
 377         }
 378         while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_rq_dto_q))) {
 379                 list_del(&ctxt->rc_list);
 380                 svc_rdma_recv_ctxt_put(rdma, ctxt);
 381         }
 382 }
 383
 384 static void svc_rdma_build_arg_xdr(struct svc_rqst *rqstp,
 385                                    struct svc_rdma_recv_ctxt *ctxt)
 386 {
 387         struct xdr_buf *arg = &rqstp->rq_arg;
 388
 389         arg->head[0].iov_base = ctxt->rc_recv_buf;
 390         arg->head[0].iov_len = ctxt->rc_byte_len;
 391         arg->tail[0].iov_base = NULL;
 392         arg->tail[0].iov_len = 0;
 393         arg->page_len = 0;
 394         arg->page_base = 0;
 395         arg->buflen = ctxt->rc_byte_len;
 396         arg->len = ctxt->rc_byte_len;
 397 }
 398
 399 /**
 400  * xdr_count_read_segments - Count number of Read segments in Read list
 401  * @rctxt: Ingress receive context
 402  * @p: Start of an un-decoded Read list
 403  *
 404  * Before allocating anything, ensure the ingress Read list is safe
 405  * to use.
 406  *
 407  * The segment count is limited to how many segments can fit in the
 408  * transport header without overflowing the buffer. That's about 40
 409  * Read segments for a 1KB inline threshold.
 410  *
 411  * Return values:
 412  *   %true: Read list is valid. @rctxt's xdr_stream is updated to point
 413  *          to the first byte past the Read list. rc_read_pcl and
 414  *          rc_call_pcl cl_count fields are set to the number of
 415  *          Read segments in the list.
 416  *  %false: Read list is corrupt. @rctxt's xdr_stream is left in an
 417  *          unknown state.
 418  */
 419 static bool xdr_count_read_segments(struct svc_rdma_recv_ctxt *rctxt, __be32 *p)
 420 {
 421         rctxt->rc_call_pcl.cl_count = 0;
 422         rctxt->rc_read_pcl.cl_count = 0;
 423         while (xdr_item_is_present(p)) {
 424                 u32 position, handle, length;
 425                 u64 offset;
 426
 427                 p = xdr_inline_decode(&rctxt->rc_stream,
 428                                       rpcrdma_readseg_maxsz * sizeof(*p));
 429                 if (!p)
 430                         return false;
 431
 432                 xdr_decode_read_segment(p, &position, &handle,
 433                                             &length, &offset);
 434                 if (position) {
 435                         if (position & 3)
 436                                 return false;
 437                         ++rctxt->rc_read_pcl.cl_count;
 438                 } else {
 439                         ++rctxt->rc_call_pcl.cl_count;
 440                 }
 441
 442                 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
 443                 if (!p)
 444                         return false;
 445         }
 446         return true;
 447 }
 448
 449 /* Sanity check the Read list.
 450  *
 451  * Sanity checks:
 452  * - Read list does not overflow Receive buffer.
 453  * - Chunk size limited by largest NFS data payload.
 454  *
 455  * Return values:
 456  *   %true: Read list is valid. @rctxt's xdr_stream is updated
 457  *          to point to the first byte past the Read list.
 458  *  %false: Read list is corrupt. @rctxt's xdr_stream is left
 459  *          in an unknown state.
 460  */
 461 static bool xdr_check_read_list(struct svc_rdma_recv_ctxt *rctxt)
 462 {
 463         __be32 *p;
 464
 465         p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
 466         if (!p)
 467                 return false;
 468         if (!xdr_count_read_segments(rctxt, p))
 469                 return false;
 470         if (!pcl_alloc_call(rctxt, p))
 471                 return false;
 472         return pcl_alloc_read(rctxt, p);
 473 }
 474
 475 static bool xdr_check_write_chunk(struct svc_rdma_recv_ctxt *rctxt)
 476 {
 477         u32 segcount;
 478         __be32 *p;
 479
 480         if (xdr_stream_decode_u32(&rctxt->rc_stream, &segcount))
 481                 return false;
 482
 483         /* A bogus segcount causes this buffer overflow check to fail. */
 484         p = xdr_inline_decode(&rctxt->rc_stream,
 485                               segcount * rpcrdma_segment_maxsz * sizeof(*p));
 486         return p != NULL;
 487 }
 488
 489 /**
 490  * xdr_count_write_chunks - Count number of Write chunks in Write list
 491  * @rctxt: Received header and decoding state
 492  * @p: start of an un-decoded Write list
 493  *
 494  * Before allocating anything, ensure the ingress Write list is
 495  * safe to use.
 496  *
 497  * Return values:
 498  *       %true: Write list is valid. @rctxt's xdr_stream is updated
 499  *              to point to the first byte past the Write list, and
 500  *              the number of Write chunks is in rc_write_pcl.cl_count.
 501  *      %false: Write list is corrupt. @rctxt's xdr_stream is left
 502  *              in an indeterminate state.
 503  */
 504 static bool xdr_count_write_chunks(struct svc_rdma_recv_ctxt *rctxt, __be32 *p)
 505 {
 506         rctxt->rc_write_pcl.cl_count = 0;
 507         while (xdr_item_is_present(p)) {
 508                 if (!xdr_check_write_chunk(rctxt))
 509                         return false;
 510                 ++rctxt->rc_write_pcl.cl_count;
 511                 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
 512                 if (!p)
 513                         return false;
 514         }
 515         return true;
 516 }
 517
 518 /* Sanity check the Write list.
 519  *
 520  * Implementation limits:
 521  * - This implementation currently supports only one Write chunk.
 522  *
 523  * Sanity checks:
 524  * - Write list does not overflow Receive buffer.
 525  * - Chunk size limited by largest NFS data payload.
 526  *
 527  * Return values:
 528  *       %true: Write list is valid. @rctxt's xdr_stream is updated
 529  *              to point to the first byte past the Write list.
 530  *      %false: Write list is corrupt. @rctxt's xdr_stream is left
 531  *              in an unknown state.
 532  */
 533 static bool xdr_check_write_list(struct svc_rdma_recv_ctxt *rctxt)
 534 {
 535         __be32 *p;
 536
 537         p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
 538         if (!p)
 539                 return false;
 540         if (!xdr_count_write_chunks(rctxt, p))
 541                 return false;
 542         if (!pcl_alloc_write(rctxt, &rctxt->rc_write_pcl, p))
 543                 return false;
 544
 545         rctxt->rc_cur_result_payload = pcl_first_chunk(&rctxt->rc_write_pcl);
 546         return true;
 547 }
 548
 549 /* Sanity check the Reply chunk.
 550  *
 551  * Sanity checks:
 552  * - Reply chunk does not overflow Receive buffer.
 553  * - Chunk size limited by largest NFS data payload.
 554  *
 555  * Return values:
 556  *       %true: Reply chunk is valid. @rctxt's xdr_stream is updated
 557  *              to point to the first byte past the Reply chunk.
 558  *      %false: Reply chunk is corrupt. @rctxt's xdr_stream is left
 559  *              in an unknown state.
 560  */
 561 static bool xdr_check_reply_chunk(struct svc_rdma_recv_ctxt *rctxt)
 562 {
 563         __be32 *p;
 564
 565         p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
 566         if (!p)
 567                 return false;
 568
 569         if (!xdr_item_is_present(p))
 570                 return true;
 571         if (!xdr_check_write_chunk(rctxt))
 572                 return false;
 573
 574         rctxt->rc_reply_pcl.cl_count = 1;
 575         return pcl_alloc_write(rctxt, &rctxt->rc_reply_pcl, p);
 576 }
 577
 578 /* RPC-over-RDMA Version One private extension: Remote Invalidation.
 579  * Responder's choice: requester signals it can handle Send With
 580  * Invalidate, and responder chooses one R_key to invalidate.
 581  *
 582  * If there is exactly one distinct R_key in the received transport
 583  * header, set rc_inv_rkey to that R_key. Otherwise, set it to zero.
 584  */
 585 static void svc_rdma_get_inv_rkey(struct svcxprt_rdma *rdma,
 586                                   struct svc_rdma_recv_ctxt *ctxt)
 587 {
 588         struct svc_rdma_segment *segment;
 589         struct svc_rdma_chunk *chunk;
 590         u32 inv_rkey;
 591
 592         ctxt->rc_inv_rkey = 0;
 593
 594         if (!rdma->sc_snd_w_inv)
 595                 return;
 596
 597         inv_rkey = 0;
 598         pcl_for_each_chunk(chunk, &ctxt->rc_call_pcl) {
 599                 pcl_for_each_segment(segment, chunk) {
 600                         if (inv_rkey == 0)
 601                                 inv_rkey = segment->rs_handle;
 602                         else if (inv_rkey != segment->rs_handle)
 603                                 return;
 604                 }
 605         }
 606         pcl_for_each_chunk(chunk, &ctxt->rc_read_pcl) {
 607                 pcl_for_each_segment(segment, chunk) {
 608                         if (inv_rkey == 0)
 609                                 inv_rkey = segment->rs_handle;
 610                         else if (inv_rkey != segment->rs_handle)
 611                                 return;
 612                 }
 613         }
 614         pcl_for_each_chunk(chunk, &ctxt->rc_write_pcl) {
 615                 pcl_for_each_segment(segment, chunk) {
 616                         if (inv_rkey == 0)
 617                                 inv_rkey = segment->rs_handle;
 618                         else if (inv_rkey != segment->rs_handle)
 619                                 return;
 620                 }
 621         }
 622         pcl_for_each_chunk(chunk, &ctxt->rc_reply_pcl) {
 623                 pcl_for_each_segment(segment, chunk) {
 624                         if (inv_rkey == 0)
 625                                 inv_rkey = segment->rs_handle;
 626                         else if (inv_rkey != segment->rs_handle)
 627                                 return;
 628                 }
 629         }
 630         ctxt->rc_inv_rkey = inv_rkey;
 631 }
 632
 633 /**
 634  * svc_rdma_xdr_decode_req - Decode the transport header
 635  * @rq_arg: xdr_buf containing ingress RPC/RDMA message
 636  * @rctxt: state of decoding
 637  *
 638  * On entry, xdr->head[0].iov_base points to first byte of the
 639  * RPC-over-RDMA transport header.
 640  *
 641  * On successful exit, head[0] points to first byte past the
 642  * RPC-over-RDMA header. For RDMA_MSG, this is the RPC message.
 643  *
 644  * The length of the RPC-over-RDMA header is returned.
 645  *
 646  * Assumptions:
 647  * - The transport header is entirely contained in the head iovec.
 648  */
 649 static int svc_rdma_xdr_decode_req(struct xdr_buf *rq_arg,
 650                                    struct svc_rdma_recv_ctxt *rctxt)
 651 {
 652         __be32 *p, *rdma_argp;
 653         unsigned int hdr_len;
 654
 655         rdma_argp = rq_arg->head[0].iov_base;
 656         xdr_init_decode(&rctxt->rc_stream, rq_arg, rdma_argp, NULL);
 657
 658         p = xdr_inline_decode(&rctxt->rc_stream,
 659                               rpcrdma_fixed_maxsz * sizeof(*p));
 660         if (unlikely(!p))
 661                 goto out_short;
 662         p++;
 663         if (*p != rpcrdma_version)
 664                 goto out_version;
 665         p += 2;
 666         rctxt->rc_msgtype = *p;
 667         switch (rctxt->rc_msgtype) {
 668         case rdma_msg:
 669                 break;
 670         case rdma_nomsg:
 671                 break;
 672         case rdma_done:
 673                 goto out_drop;
 674         case rdma_error:
 675                 goto out_drop;
 676         default:
 677                 goto out_proc;
 678         }
 679
 680         if (!xdr_check_read_list(rctxt))
 681                 goto out_inval;
 682         if (!xdr_check_write_list(rctxt))
 683                 goto out_inval;
 684         if (!xdr_check_reply_chunk(rctxt))
 685                 goto out_inval;
 686
 687         rq_arg->head[0].iov_base = rctxt->rc_stream.p;
 688         hdr_len = xdr_stream_pos(&rctxt->rc_stream);
 689         rq_arg->head[0].iov_len -= hdr_len;
 690         rq_arg->len -= hdr_len;
 691         trace_svcrdma_decode_rqst(rctxt, rdma_argp, hdr_len);
 692         return hdr_len;
 693
 694 out_short:
 695         trace_svcrdma_decode_short_err(rctxt, rq_arg->len);
 696         return -EINVAL;
 697
 698 out_version:
 699         trace_svcrdma_decode_badvers_err(rctxt, rdma_argp);
 700         return -EPROTONOSUPPORT;
 701
 702 out_drop:
 703         trace_svcrdma_decode_drop_err(rctxt, rdma_argp);
 704         return 0;
 705
 706 out_proc:
 707         trace_svcrdma_decode_badproc_err(rctxt, rdma_argp);
 708         return -EINVAL;
 709
 710 out_inval:
 711         trace_svcrdma_decode_parse_err(rctxt, rdma_argp);
 712         return -EINVAL;
 713 }
 714
 715 static void rdma_read_complete(struct svc_rqst *rqstp,
 716                                struct svc_rdma_recv_ctxt *head)
 717 {
 718         int page_no;
 719
 720         /* Move Read chunk pages to rqstp so that they will be released
 721          * when svc_process is done with them.
 722          */
 723         for (page_no = 0; page_no < head->rc_page_count; page_no++) {
 724                 put_page(rqstp->rq_pages[page_no]);
 725                 rqstp->rq_pages[page_no] = head->rc_pages[page_no];
 726         }
 727         head->rc_page_count = 0;
 728
 729         /* Point rq_arg.pages past header */
 730         rqstp->rq_arg.pages = &rqstp->rq_pages[head->rc_hdr_count];
 731         rqstp->rq_arg.page_len = head->rc_arg.page_len;
 732
 733         /* rq_respages starts after the last arg page */
 734         rqstp->rq_respages = &rqstp->rq_pages[page_no];
 735         rqstp->rq_next_page = rqstp->rq_respages + 1;
 736
 737         /* Rebuild rq_arg head and tail. */
 738         rqstp->rq_arg.head[0] = head->rc_arg.head[0];
 739         rqstp->rq_arg.tail[0] = head->rc_arg.tail[0];
 740         rqstp->rq_arg.len = head->rc_arg.len;
 741         rqstp->rq_arg.buflen = head->rc_arg.buflen;
 742 }
 743
 744 static void svc_rdma_send_error(struct svcxprt_rdma *rdma,
 745                                 struct svc_rdma_recv_ctxt *rctxt,
 746                                 int status)
 747 {
 748         struct svc_rdma_send_ctxt *sctxt;
 749
 750         sctxt = svc_rdma_send_ctxt_get(rdma);
 751         if (!sctxt)
 752                 return;
 753         svc_rdma_send_error_msg(rdma, sctxt, rctxt, status);
 754 }
 755
 756 /* By convention, backchannel calls arrive via rdma_msg type
 757  * messages, and never populate the chunk lists. This makes
 758  * the RPC/RDMA header small and fixed in size, so it is
 759  * straightforward to check the RPC header's direction field.
 760  */
 761 static bool svc_rdma_is_reverse_direction_reply(struct svc_xprt *xprt,
 762                                                 struct svc_rdma_recv_ctxt *rctxt)
 763 {
 764         __be32 *p = rctxt->rc_recv_buf;
 765
 766         if (!xprt->xpt_bc_xprt)
 767                 return false;
 768
 769         if (rctxt->rc_msgtype != rdma_msg)
 770                 return false;
 771
 772         if (!pcl_is_empty(&rctxt->rc_call_pcl))
 773                 return false;
 774         if (!pcl_is_empty(&rctxt->rc_read_pcl))
 775                 return false;
 776         if (!pcl_is_empty(&rctxt->rc_write_pcl))
 777                 return false;
 778         if (!pcl_is_empty(&rctxt->rc_reply_pcl))
 779                 return false;
 780
 781         /* RPC call direction */
 782         if (*(p + 8) == cpu_to_be32(RPC_CALL))
 783                 return false;
 784
 785         return true;
 786 }
 787
 788 /**
 789  * svc_rdma_recvfrom - Receive an RPC call
 790  * @rqstp: request structure into which to receive an RPC Call
 791  *
 792  * Returns:
 793  *      The positive number of bytes in the RPC Call message,
 794  *      %0 if there were no Calls ready to return,
 795  *      %-EINVAL if the Read chunk data is too large,
 796  *      %-ENOMEM if rdma_rw context pool was exhausted,
 797  *      %-ENOTCONN if posting failed (connection is lost),
 798  *      %-EIO if rdma_rw initialization failed (DMA mapping, etc).
 799  *
 800  * Called in a loop when XPT_DATA is set. XPT_DATA is cleared only
 801  * when there are no remaining ctxt's to process.
 802  *
 803  * The next ctxt is removed from the "receive" lists.
 804  *
 805  * - If the ctxt completes a Read, then finish assembling the Call
 806  *   message and return the number of bytes in the message.
 807  *
 808  * - If the ctxt completes a Receive, then construct the Call
 809  *   message from the contents of the Receive buffer.
 810  *
 811  *   - If there are no Read chunks in this message, then finish
 812  *     assembling the Call message and return the number of bytes
 813  *     in the message.
 814  *
 815  *   - If there are Read chunks in this message, post Read WRs to
 816  *     pull that payload and return 0.
 817  */
 818 int svc_rdma_recvfrom(struct svc_rqst *rqstp)
 819 {
 820         struct svc_xprt *xprt = rqstp->rq_xprt;
 821         struct svcxprt_rdma *rdma_xprt =
 822                 container_of(xprt, struct svcxprt_rdma, sc_xprt);
 823         struct svc_rdma_recv_ctxt *ctxt;
 824         int ret;
 825
 826         rqstp->rq_xprt_ctxt = NULL;
 827
 828         spin_lock(&rdma_xprt->sc_rq_dto_lock);
 829         ctxt = svc_rdma_next_recv_ctxt(&rdma_xprt->sc_read_complete_q);
 830         if (ctxt) {
 831                 list_del(&ctxt->rc_list);
 832                 spin_unlock(&rdma_xprt->sc_rq_dto_lock);
 833                 rdma_read_complete(rqstp, ctxt);
 834                 goto complete;
 835         }
 836         ctxt = svc_rdma_next_recv_ctxt(&rdma_xprt->sc_rq_dto_q);
 837         if (!ctxt) {
 838                 /* No new incoming requests, terminate the loop */
 839                 clear_bit(XPT_DATA, &xprt->xpt_flags);
 840                 spin_unlock(&rdma_xprt->sc_rq_dto_lock);
 841                 return 0;
 842         }
 843         list_del(&ctxt->rc_list);
 844         spin_unlock(&rdma_xprt->sc_rq_dto_lock);
 845         percpu_counter_inc(&svcrdma_stat_recv);
 846
 847         ib_dma_sync_single_for_cpu(rdma_xprt->sc_pd->device,
 848                                    ctxt->rc_recv_sge.addr, ctxt->rc_byte_len,
 849                                    DMA_FROM_DEVICE);
 850         svc_rdma_build_arg_xdr(rqstp, ctxt);
 851
 852         /* Prevent svc_xprt_release from releasing pages in rq_pages
 853          * if we return 0 or an error.
 854          */
 855         rqstp->rq_respages = rqstp->rq_pages;
 856         rqstp->rq_next_page = rqstp->rq_respages;
 857
 858         ret = svc_rdma_xdr_decode_req(&rqstp->rq_arg, ctxt);
 859         if (ret < 0)
 860                 goto out_err;
 861         if (ret == 0)
 862                 goto out_drop;
 863         rqstp->rq_xprt_hlen = ret;
 864
 865         if (svc_rdma_is_reverse_direction_reply(xprt, ctxt))
 866                 goto out_backchannel;
 867
 868         svc_rdma_get_inv_rkey(rdma_xprt, ctxt);
 869
 870         if (!pcl_is_empty(&ctxt->rc_read_pcl) ||
 871             !pcl_is_empty(&ctxt->rc_call_pcl))
 872                 goto out_readlist;
 873
 874 complete:
 875         rqstp->rq_xprt_ctxt = ctxt;
 876         rqstp->rq_prot = IPPROTO_MAX;
 877         svc_xprt_copy_addrs(rqstp, xprt);
 878         return rqstp->rq_arg.len;
 879
 880 out_readlist:
 881         ret = svc_rdma_process_read_list(rdma_xprt, rqstp, ctxt);
 882         if (ret < 0)
 883                 goto out_readfail;
 884         return 0;
 885
 886 out_err:
 887         svc_rdma_send_error(rdma_xprt, ctxt, ret);
 888         svc_rdma_recv_ctxt_put(rdma_xprt, ctxt);
 889         return 0;
 890
 891 out_readfail:
 892         if (ret == -EINVAL)
 893                 svc_rdma_send_error(rdma_xprt, ctxt, ret);
 894         svc_rdma_recv_ctxt_put(rdma_xprt, ctxt);
 895         return ret;
 896
 897 out_backchannel:
 898         svc_rdma_handle_bc_reply(rqstp, ctxt);
 899 out_drop:
 900         svc_rdma_recv_ctxt_put(rdma_xprt, ctxt);
 901         return 0;
 902 }