1bcf3ffd4SChuck Lever // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause 2c06b540aSTom Tucker /* 3ecf85b23SChuck Lever * Copyright (c) 2016-2018 Oracle. All rights reserved. 40bf48289SSteve Wise * Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved. 5c06b540aSTom Tucker * Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved. 6c06b540aSTom Tucker * 7c06b540aSTom Tucker * This software is available to you under a choice of one of two 8c06b540aSTom Tucker * licenses. You may choose to be licensed under the terms of the GNU 9c06b540aSTom Tucker * General Public License (GPL) Version 2, available from the file 10c06b540aSTom Tucker * COPYING in the main directory of this source tree, or the BSD-type 11c06b540aSTom Tucker * license below: 12c06b540aSTom Tucker * 13c06b540aSTom Tucker * Redistribution and use in source and binary forms, with or without 14c06b540aSTom Tucker * modification, are permitted provided that the following conditions 15c06b540aSTom Tucker * are met: 16c06b540aSTom Tucker * 17c06b540aSTom Tucker * Redistributions of source code must retain the above copyright 18c06b540aSTom Tucker * notice, this list of conditions and the following disclaimer. 19c06b540aSTom Tucker * 20c06b540aSTom Tucker * Redistributions in binary form must reproduce the above 21c06b540aSTom Tucker * copyright notice, this list of conditions and the following 22c06b540aSTom Tucker * disclaimer in the documentation and/or other materials provided 23c06b540aSTom Tucker * with the distribution. 24c06b540aSTom Tucker * 25c06b540aSTom Tucker * Neither the name of the Network Appliance, Inc. nor the names of 26c06b540aSTom Tucker * its contributors may be used to endorse or promote products 27c06b540aSTom Tucker * derived from this software without specific prior written 28c06b540aSTom Tucker * permission. 29c06b540aSTom Tucker * 30c06b540aSTom Tucker * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 31c06b540aSTom Tucker * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 32c06b540aSTom Tucker * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 33c06b540aSTom Tucker * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 34c06b540aSTom Tucker * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 35c06b540aSTom Tucker * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 36c06b540aSTom Tucker * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 37c06b540aSTom Tucker * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 38c06b540aSTom Tucker * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 39c06b540aSTom Tucker * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 40c06b540aSTom Tucker * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 41c06b540aSTom Tucker * 42c06b540aSTom Tucker * Author: Tom Tucker <tom@opengridcomputing.com> 43c06b540aSTom Tucker */ 44c06b540aSTom Tucker 459a6a180bSChuck Lever /* Operation 469a6a180bSChuck Lever * 479a6a180bSChuck Lever * The main entry point is svc_rdma_sendto. This is called by the 489a6a180bSChuck Lever * RPC server when an RPC Reply is ready to be transmitted to a client. 499a6a180bSChuck Lever * 509a6a180bSChuck Lever * The passed-in svc_rqst contains a struct xdr_buf which holds an 519a6a180bSChuck Lever * XDR-encoded RPC Reply message. sendto must construct the RPC-over-RDMA 529a6a180bSChuck Lever * transport header, post all Write WRs needed for this Reply, then post 539a6a180bSChuck Lever * a Send WR conveying the transport header and the RPC message itself to 549a6a180bSChuck Lever * the client. 559a6a180bSChuck Lever * 569a6a180bSChuck Lever * svc_rdma_sendto must fully transmit the Reply before returning, as 579a6a180bSChuck Lever * the svc_rqst will be recycled as soon as sendto returns. Remaining 589a6a180bSChuck Lever * resources referred to by the svc_rqst are also recycled at that time. 599a6a180bSChuck Lever * Therefore any resources that must remain longer must be detached 609a6a180bSChuck Lever * from the svc_rqst and released later. 619a6a180bSChuck Lever * 629a6a180bSChuck Lever * Page Management 639a6a180bSChuck Lever * 649a6a180bSChuck Lever * The I/O that performs Reply transmission is asynchronous, and may 659a6a180bSChuck Lever * complete well after sendto returns. Thus pages under I/O must be 669a6a180bSChuck Lever * removed from the svc_rqst before sendto returns. 679a6a180bSChuck Lever * 689a6a180bSChuck Lever * The logic here depends on Send Queue and completion ordering. Since 699a6a180bSChuck Lever * the Send WR is always posted last, it will always complete last. Thus 709a6a180bSChuck Lever * when it completes, it is guaranteed that all previous Write WRs have 719a6a180bSChuck Lever * also completed. 729a6a180bSChuck Lever * 739a6a180bSChuck Lever * Write WRs are constructed and posted. Each Write segment gets its own 749a6a180bSChuck Lever * svc_rdma_rw_ctxt, allowing the Write completion handler to find and 759a6a180bSChuck Lever * DMA-unmap the pages under I/O for that Write segment. The Write 769a6a180bSChuck Lever * completion handler does not release any pages. 779a6a180bSChuck Lever * 784201c746SChuck Lever * When the Send WR is constructed, it also gets its own svc_rdma_send_ctxt. 799a6a180bSChuck Lever * The ownership of all of the Reply's pages are transferred into that 809a6a180bSChuck Lever * ctxt, the Send WR is posted, and sendto returns. 819a6a180bSChuck Lever * 824201c746SChuck Lever * The svc_rdma_send_ctxt is presented when the Send WR completes. The 839a6a180bSChuck Lever * Send completion handler finally releases the Reply's pages. 849a6a180bSChuck Lever * 859a6a180bSChuck Lever * This mechanism also assumes that completions on the transport's Send 869a6a180bSChuck Lever * Completion Queue do not run in parallel. Otherwise a Write completion 879a6a180bSChuck Lever * and Send completion running at the same time could release pages that 889a6a180bSChuck Lever * are still DMA-mapped. 899a6a180bSChuck Lever * 909a6a180bSChuck Lever * Error Handling 919a6a180bSChuck Lever * 929a6a180bSChuck Lever * - If the Send WR is posted successfully, it will either complete 939a6a180bSChuck Lever * successfully, or get flushed. Either way, the Send completion 949a6a180bSChuck Lever * handler releases the Reply's pages. 959a6a180bSChuck Lever * - If the Send WR cannot be not posted, the forward path releases 969a6a180bSChuck Lever * the Reply's pages. 979a6a180bSChuck Lever * 989a6a180bSChuck Lever * This handles the case, without the use of page reference counting, 999a6a180bSChuck Lever * where two different Write segments send portions of the same page. 1009a6a180bSChuck Lever */ 1019a6a180bSChuck Lever 102c06b540aSTom Tucker #include <linux/spinlock.h> 103c06b540aSTom Tucker #include <asm/unaligned.h> 10498895edbSChuck Lever 105c06b540aSTom Tucker #include <rdma/ib_verbs.h> 106c06b540aSTom Tucker #include <rdma/rdma_cm.h> 10798895edbSChuck Lever 10898895edbSChuck Lever #include <linux/sunrpc/debug.h> 109c06b540aSTom Tucker #include <linux/sunrpc/svc_rdma.h> 110c06b540aSTom Tucker 11198895edbSChuck Lever #include "xprt_rdma.h" 11298895edbSChuck Lever #include <trace/events/rpcrdma.h> 11398895edbSChuck Lever 1144201c746SChuck Lever static void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc); 1154201c746SChuck Lever 1163ac56c2fSChuck Lever static void svc_rdma_send_cid_init(struct svcxprt_rdma *rdma, 1173ac56c2fSChuck Lever struct rpc_rdma_cid *cid) 1183ac56c2fSChuck Lever { 1193ac56c2fSChuck Lever cid->ci_queue_id = rdma->sc_sq_cq->res.id; 1203ac56c2fSChuck Lever cid->ci_completion_id = atomic_inc_return(&rdma->sc_completion_ids); 1213ac56c2fSChuck Lever } 1223ac56c2fSChuck Lever 1234201c746SChuck Lever static struct svc_rdma_send_ctxt * 1244201c746SChuck Lever svc_rdma_send_ctxt_alloc(struct svcxprt_rdma *rdma) 1254201c746SChuck Lever { 126ed51b426SChuck Lever int node = ibdev_to_node(rdma->sc_cm_id->device); 1274201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt; 12899722fe4SChuck Lever dma_addr_t addr; 12999722fe4SChuck Lever void *buffer; 1304201c746SChuck Lever int i; 1314201c746SChuck Lever 132ed51b426SChuck Lever ctxt = kmalloc_node(struct_size(ctxt, sc_sges, rdma->sc_max_send_sges), 133ed51b426SChuck Lever GFP_KERNEL, node); 1344201c746SChuck Lever if (!ctxt) 13599722fe4SChuck Lever goto fail0; 136ed51b426SChuck Lever buffer = kmalloc_node(rdma->sc_max_req_size, GFP_KERNEL, node); 13799722fe4SChuck Lever if (!buffer) 13899722fe4SChuck Lever goto fail1; 13999722fe4SChuck Lever addr = ib_dma_map_single(rdma->sc_pd->device, buffer, 14099722fe4SChuck Lever rdma->sc_max_req_size, DMA_TO_DEVICE); 14199722fe4SChuck Lever if (ib_dma_mapping_error(rdma->sc_pd->device, addr)) 14299722fe4SChuck Lever goto fail2; 1434201c746SChuck Lever 1443ac56c2fSChuck Lever svc_rdma_send_cid_init(rdma, &ctxt->sc_cid); 1453ac56c2fSChuck Lever 1464201c746SChuck Lever ctxt->sc_send_wr.next = NULL; 1474201c746SChuck Lever ctxt->sc_send_wr.wr_cqe = &ctxt->sc_cqe; 1484201c746SChuck Lever ctxt->sc_send_wr.sg_list = ctxt->sc_sges; 1494201c746SChuck Lever ctxt->sc_send_wr.send_flags = IB_SEND_SIGNALED; 1502a1e4f21SChuck Lever init_completion(&ctxt->sc_done); 15199722fe4SChuck Lever ctxt->sc_cqe.done = svc_rdma_wc_send; 15299722fe4SChuck Lever ctxt->sc_xprt_buf = buffer; 1536fd5034dSChuck Lever xdr_buf_init(&ctxt->sc_hdrbuf, ctxt->sc_xprt_buf, 1546fd5034dSChuck Lever rdma->sc_max_req_size); 15599722fe4SChuck Lever ctxt->sc_sges[0].addr = addr; 15699722fe4SChuck Lever 15725fd86ecSChuck Lever for (i = 0; i < rdma->sc_max_send_sges; i++) 1584201c746SChuck Lever ctxt->sc_sges[i].lkey = rdma->sc_pd->local_dma_lkey; 1594201c746SChuck Lever return ctxt; 16099722fe4SChuck Lever 16199722fe4SChuck Lever fail2: 16299722fe4SChuck Lever kfree(buffer); 16399722fe4SChuck Lever fail1: 16499722fe4SChuck Lever kfree(ctxt); 16599722fe4SChuck Lever fail0: 16699722fe4SChuck Lever return NULL; 1674201c746SChuck Lever } 1684201c746SChuck Lever 1694201c746SChuck Lever /** 1704201c746SChuck Lever * svc_rdma_send_ctxts_destroy - Release all send_ctxt's for an xprt 1714201c746SChuck Lever * @rdma: svcxprt_rdma being torn down 1724201c746SChuck Lever * 1734201c746SChuck Lever */ 1744201c746SChuck Lever void svc_rdma_send_ctxts_destroy(struct svcxprt_rdma *rdma) 1754201c746SChuck Lever { 1764201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt; 177b6c2bfeaSChuck Lever struct llist_node *node; 1784201c746SChuck Lever 179b6c2bfeaSChuck Lever while ((node = llist_del_first(&rdma->sc_send_ctxts)) != NULL) { 180b6c2bfeaSChuck Lever ctxt = llist_entry(node, struct svc_rdma_send_ctxt, sc_node); 18199722fe4SChuck Lever ib_dma_unmap_single(rdma->sc_pd->device, 18299722fe4SChuck Lever ctxt->sc_sges[0].addr, 18399722fe4SChuck Lever rdma->sc_max_req_size, 18499722fe4SChuck Lever DMA_TO_DEVICE); 18599722fe4SChuck Lever kfree(ctxt->sc_xprt_buf); 1864201c746SChuck Lever kfree(ctxt); 1874201c746SChuck Lever } 1884201c746SChuck Lever } 1894201c746SChuck Lever 1904201c746SChuck Lever /** 1914201c746SChuck Lever * svc_rdma_send_ctxt_get - Get a free send_ctxt 1924201c746SChuck Lever * @rdma: controlling svcxprt_rdma 1934201c746SChuck Lever * 1944201c746SChuck Lever * Returns a ready-to-use send_ctxt, or NULL if none are 1954201c746SChuck Lever * available and a fresh one cannot be allocated. 1964201c746SChuck Lever */ 1974201c746SChuck Lever struct svc_rdma_send_ctxt *svc_rdma_send_ctxt_get(struct svcxprt_rdma *rdma) 1984201c746SChuck Lever { 1994201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt; 200b6c2bfeaSChuck Lever struct llist_node *node; 2014201c746SChuck Lever 2024201c746SChuck Lever spin_lock(&rdma->sc_send_lock); 203b6c2bfeaSChuck Lever node = llist_del_first(&rdma->sc_send_ctxts); 204b6c2bfeaSChuck Lever if (!node) 2054201c746SChuck Lever goto out_empty; 206b6c2bfeaSChuck Lever ctxt = llist_entry(node, struct svc_rdma_send_ctxt, sc_node); 2074201c746SChuck Lever spin_unlock(&rdma->sc_send_lock); 2084201c746SChuck Lever 2094201c746SChuck Lever out: 2106fd5034dSChuck Lever rpcrdma_set_xdrlen(&ctxt->sc_hdrbuf, 0); 2116fd5034dSChuck Lever xdr_init_encode(&ctxt->sc_stream, &ctxt->sc_hdrbuf, 2126fd5034dSChuck Lever ctxt->sc_xprt_buf, NULL); 2136fd5034dSChuck Lever 2144201c746SChuck Lever ctxt->sc_send_wr.num_sge = 0; 21599722fe4SChuck Lever ctxt->sc_cur_sge_no = 0; 216*a944209cSChuck Lever ctxt->sc_page_count = 0; 2174201c746SChuck Lever return ctxt; 2184201c746SChuck Lever 2194201c746SChuck Lever out_empty: 2204201c746SChuck Lever spin_unlock(&rdma->sc_send_lock); 2214201c746SChuck Lever ctxt = svc_rdma_send_ctxt_alloc(rdma); 2224201c746SChuck Lever if (!ctxt) 2234201c746SChuck Lever return NULL; 2244201c746SChuck Lever goto out; 2254201c746SChuck Lever } 2264201c746SChuck Lever 2274201c746SChuck Lever /** 2284201c746SChuck Lever * svc_rdma_send_ctxt_put - Return send_ctxt to free list 2294201c746SChuck Lever * @rdma: controlling svcxprt_rdma 2304201c746SChuck Lever * @ctxt: object to return to the free list 231*a944209cSChuck Lever * 232*a944209cSChuck Lever * Pages left in sc_pages are DMA unmapped and released. 2334201c746SChuck Lever */ 2344201c746SChuck Lever void svc_rdma_send_ctxt_put(struct svcxprt_rdma *rdma, 2354201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt) 2364201c746SChuck Lever { 2374201c746SChuck Lever struct ib_device *device = rdma->sc_cm_id->device; 2384201c746SChuck Lever unsigned int i; 2394201c746SChuck Lever 240*a944209cSChuck Lever for (i = 0; i < ctxt->sc_page_count; ++i) 241*a944209cSChuck Lever put_page(ctxt->sc_pages[i]); 242*a944209cSChuck Lever 24399722fe4SChuck Lever /* The first SGE contains the transport header, which 24499722fe4SChuck Lever * remains mapped until @ctxt is destroyed. 24599722fe4SChuck Lever */ 246832b2cb9SChuck Lever for (i = 1; i < ctxt->sc_send_wr.num_sge; i++) { 2474201c746SChuck Lever ib_dma_unmap_page(device, 2484201c746SChuck Lever ctxt->sc_sges[i].addr, 2494201c746SChuck Lever ctxt->sc_sges[i].length, 2504201c746SChuck Lever DMA_TO_DEVICE); 251832b2cb9SChuck Lever trace_svcrdma_dma_unmap_page(rdma, 252832b2cb9SChuck Lever ctxt->sc_sges[i].addr, 253832b2cb9SChuck Lever ctxt->sc_sges[i].length); 254832b2cb9SChuck Lever } 2554201c746SChuck Lever 256b6c2bfeaSChuck Lever llist_add(&ctxt->sc_node, &rdma->sc_send_ctxts); 2574201c746SChuck Lever } 2584201c746SChuck Lever 2594201c746SChuck Lever /** 2606c8c84f5SChuck Lever * svc_rdma_wake_send_waiters - manage Send Queue accounting 2616c8c84f5SChuck Lever * @rdma: controlling transport 2626c8c84f5SChuck Lever * @avail: Number of additional SQEs that are now available 2636c8c84f5SChuck Lever * 2646c8c84f5SChuck Lever */ 2656c8c84f5SChuck Lever void svc_rdma_wake_send_waiters(struct svcxprt_rdma *rdma, int avail) 2666c8c84f5SChuck Lever { 2676c8c84f5SChuck Lever atomic_add(avail, &rdma->sc_sq_avail); 2686c8c84f5SChuck Lever smp_mb__after_atomic(); 2696c8c84f5SChuck Lever if (unlikely(waitqueue_active(&rdma->sc_send_wait))) 2706c8c84f5SChuck Lever wake_up(&rdma->sc_send_wait); 2716c8c84f5SChuck Lever } 2726c8c84f5SChuck Lever 2736c8c84f5SChuck Lever /** 2744201c746SChuck Lever * svc_rdma_wc_send - Invoked by RDMA provider for each polled Send WC 2754201c746SChuck Lever * @cq: Completion Queue context 2764201c746SChuck Lever * @wc: Work Completion object 2774201c746SChuck Lever * 2784201c746SChuck Lever * NB: The svc_xprt/svcxprt_rdma is pinned whenever it's possible that 2794201c746SChuck Lever * the Send completion handler could be running. 2804201c746SChuck Lever */ 2814201c746SChuck Lever static void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc) 2824201c746SChuck Lever { 2834201c746SChuck Lever struct svcxprt_rdma *rdma = cq->cq_context; 2844201c746SChuck Lever struct ib_cqe *cqe = wc->wr_cqe; 2853ac56c2fSChuck Lever struct svc_rdma_send_ctxt *ctxt = 2863ac56c2fSChuck Lever container_of(cqe, struct svc_rdma_send_ctxt, sc_cqe); 2874201c746SChuck Lever 2886c8c84f5SChuck Lever svc_rdma_wake_send_waiters(rdma, 1); 2892a1e4f21SChuck Lever complete(&ctxt->sc_done); 2902a1e4f21SChuck Lever 291e844d307SChuck Lever if (unlikely(wc->status != IB_WC_SUCCESS)) 292eef2d8d4SChuck Lever goto flushed; 293eef2d8d4SChuck Lever 294eef2d8d4SChuck Lever trace_svcrdma_wc_send(wc, &ctxt->sc_cid); 295eef2d8d4SChuck Lever return; 296eef2d8d4SChuck Lever 297eef2d8d4SChuck Lever flushed: 298eef2d8d4SChuck Lever if (wc->status != IB_WC_WR_FLUSH_ERR) 299eef2d8d4SChuck Lever trace_svcrdma_wc_send_err(wc, &ctxt->sc_cid); 300eef2d8d4SChuck Lever else 301eef2d8d4SChuck Lever trace_svcrdma_wc_send_flush(wc, &ctxt->sc_cid); 302e844d307SChuck Lever svc_xprt_deferred_close(&rdma->sc_xprt); 3034201c746SChuck Lever } 3044201c746SChuck Lever 3053abb03faSChuck Lever /** 3063abb03faSChuck Lever * svc_rdma_send - Post a single Send WR 3073abb03faSChuck Lever * @rdma: transport on which to post the WR 30817f70f8dSChuck Lever * @ctxt: send ctxt with a Send WR ready to post 3093abb03faSChuck Lever * 3102a1e4f21SChuck Lever * Returns zero if the Send WR was posted successfully. Otherwise, a 3113abb03faSChuck Lever * negative errno is returned. 3123abb03faSChuck Lever */ 31317f70f8dSChuck Lever int svc_rdma_send(struct svcxprt_rdma *rdma, struct svc_rdma_send_ctxt *ctxt) 3144201c746SChuck Lever { 31517f70f8dSChuck Lever struct ib_send_wr *wr = &ctxt->sc_send_wr; 3164201c746SChuck Lever int ret; 3174201c746SChuck Lever 3182a1e4f21SChuck Lever reinit_completion(&ctxt->sc_done); 3194201c746SChuck Lever 320aee4b74aSChuck Lever /* Sync the transport header buffer */ 321aee4b74aSChuck Lever ib_dma_sync_single_for_device(rdma->sc_pd->device, 322aee4b74aSChuck Lever wr->sg_list[0].addr, 323aee4b74aSChuck Lever wr->sg_list[0].length, 324aee4b74aSChuck Lever DMA_TO_DEVICE); 325aee4b74aSChuck Lever 3264201c746SChuck Lever /* If the SQ is full, wait until an SQ entry is available */ 3274201c746SChuck Lever while (1) { 3283abb03faSChuck Lever if ((atomic_dec_return(&rdma->sc_sq_avail) < 0)) { 32922df5a22SChuck Lever percpu_counter_inc(&svcrdma_stat_sq_starve); 3304201c746SChuck Lever trace_svcrdma_sq_full(rdma); 3313abb03faSChuck Lever atomic_inc(&rdma->sc_sq_avail); 3324201c746SChuck Lever wait_event(rdma->sc_send_wait, 3333abb03faSChuck Lever atomic_read(&rdma->sc_sq_avail) > 1); 3344201c746SChuck Lever if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags)) 3354201c746SChuck Lever return -ENOTCONN; 3364201c746SChuck Lever trace_svcrdma_sq_retry(rdma); 3374201c746SChuck Lever continue; 3384201c746SChuck Lever } 3394201c746SChuck Lever 34017f70f8dSChuck Lever trace_svcrdma_post_send(ctxt); 341ed288d74SBart Van Assche ret = ib_post_send(rdma->sc_qp, wr, NULL); 342e28b4fc6SChuck Lever if (ret) 343e28b4fc6SChuck Lever break; 344e28b4fc6SChuck Lever return 0; 345e28b4fc6SChuck Lever } 346e28b4fc6SChuck Lever 347e28b4fc6SChuck Lever trace_svcrdma_sq_post_err(rdma, ret); 348e844d307SChuck Lever svc_xprt_deferred_close(&rdma->sc_xprt); 3494201c746SChuck Lever wake_up(&rdma->sc_send_wait); 3504201c746SChuck Lever return ret; 3514201c746SChuck Lever } 3524201c746SChuck Lever 3536fd5034dSChuck Lever /** 3546fd5034dSChuck Lever * svc_rdma_encode_read_list - Encode RPC Reply's Read chunk list 3556fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 3569a6a180bSChuck Lever * 3576fd5034dSChuck Lever * Return values: 3586fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 3596fd5034dSChuck Lever * that was consumed by the Reply Read list 3606fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 3619a6a180bSChuck Lever */ 3626fd5034dSChuck Lever static ssize_t svc_rdma_encode_read_list(struct svc_rdma_send_ctxt *sctxt) 3636fd5034dSChuck Lever { 3646fd5034dSChuck Lever /* RPC-over-RDMA version 1 replies never have a Read list. */ 3656fd5034dSChuck Lever return xdr_stream_encode_item_absent(&sctxt->sc_stream); 3666fd5034dSChuck Lever } 3676fd5034dSChuck Lever 3686fd5034dSChuck Lever /** 3696fd5034dSChuck Lever * svc_rdma_encode_write_segment - Encode one Write segment 3706fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 3716911f3e1SChuck Lever * @chunk: Write chunk to push 3726fd5034dSChuck Lever * @remaining: remaining bytes of the payload left in the Write chunk 3736911f3e1SChuck Lever * @segno: which segment in the chunk 3746fd5034dSChuck Lever * 3756fd5034dSChuck Lever * Return values: 3766fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 3776911f3e1SChuck Lever * that was consumed by the Write segment, and updates @remaining 3786fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 3796fd5034dSChuck Lever */ 3806911f3e1SChuck Lever static ssize_t svc_rdma_encode_write_segment(struct svc_rdma_send_ctxt *sctxt, 3816911f3e1SChuck Lever const struct svc_rdma_chunk *chunk, 3826911f3e1SChuck Lever u32 *remaining, unsigned int segno) 3836fd5034dSChuck Lever { 3846911f3e1SChuck Lever const struct svc_rdma_segment *segment = &chunk->ch_segments[segno]; 3856911f3e1SChuck Lever const size_t len = rpcrdma_segment_maxsz * sizeof(__be32); 3866911f3e1SChuck Lever u32 length; 3876fd5034dSChuck Lever __be32 *p; 3886fd5034dSChuck Lever 3896fd5034dSChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, len); 3906fd5034dSChuck Lever if (!p) 3916fd5034dSChuck Lever return -EMSGSIZE; 3926fd5034dSChuck Lever 3936911f3e1SChuck Lever length = min_t(u32, *remaining, segment->rs_length); 3946fd5034dSChuck Lever *remaining -= length; 3956911f3e1SChuck Lever xdr_encode_rdma_segment(p, segment->rs_handle, length, 3966911f3e1SChuck Lever segment->rs_offset); 3976911f3e1SChuck Lever trace_svcrdma_encode_wseg(sctxt, segno, segment->rs_handle, length, 3986911f3e1SChuck Lever segment->rs_offset); 3996fd5034dSChuck Lever return len; 4006fd5034dSChuck Lever } 4016fd5034dSChuck Lever 4026fd5034dSChuck Lever /** 4036fd5034dSChuck Lever * svc_rdma_encode_write_chunk - Encode one Write chunk 4046fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 4056911f3e1SChuck Lever * @chunk: Write chunk to push 4066fd5034dSChuck Lever * 4076fd5034dSChuck Lever * Copy a Write chunk from the Call transport header to the 4086fd5034dSChuck Lever * Reply transport header. Update each segment's length field 4096fd5034dSChuck Lever * to reflect the number of bytes written in that segment. 4106fd5034dSChuck Lever * 4116fd5034dSChuck Lever * Return values: 4126fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 4136fd5034dSChuck Lever * that was consumed by the Write chunk 4146fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 4156fd5034dSChuck Lever */ 4166911f3e1SChuck Lever static ssize_t svc_rdma_encode_write_chunk(struct svc_rdma_send_ctxt *sctxt, 4176911f3e1SChuck Lever const struct svc_rdma_chunk *chunk) 4189a6a180bSChuck Lever { 4196911f3e1SChuck Lever u32 remaining = chunk->ch_payload_length; 4206911f3e1SChuck Lever unsigned int segno; 4216fd5034dSChuck Lever ssize_t len, ret; 4229a6a180bSChuck Lever 4236911f3e1SChuck Lever len = 0; 4246fd5034dSChuck Lever ret = xdr_stream_encode_item_present(&sctxt->sc_stream); 4256fd5034dSChuck Lever if (ret < 0) 4266911f3e1SChuck Lever return ret; 4276fd5034dSChuck Lever len += ret; 4286fd5034dSChuck Lever 4296911f3e1SChuck Lever ret = xdr_stream_encode_u32(&sctxt->sc_stream, chunk->ch_segcount); 4306fd5034dSChuck Lever if (ret < 0) 4316911f3e1SChuck Lever return ret; 4326fd5034dSChuck Lever len += ret; 4339a6a180bSChuck Lever 4346911f3e1SChuck Lever for (segno = 0; segno < chunk->ch_segcount; segno++) { 4356911f3e1SChuck Lever ret = svc_rdma_encode_write_segment(sctxt, chunk, &remaining, segno); 4366fd5034dSChuck Lever if (ret < 0) 4376911f3e1SChuck Lever return ret; 4386fd5034dSChuck Lever len += ret; 4399a6a180bSChuck Lever } 4409a6a180bSChuck Lever 4416fd5034dSChuck Lever return len; 4429a6a180bSChuck Lever } 4439a6a180bSChuck Lever 4446fd5034dSChuck Lever /** 4456fd5034dSChuck Lever * svc_rdma_encode_write_list - Encode RPC Reply's Write chunk list 4466fd5034dSChuck Lever * @rctxt: Reply context with information about the RPC Call 4476fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 4486fd5034dSChuck Lever * 4496fd5034dSChuck Lever * Return values: 4506fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 4516fd5034dSChuck Lever * that was consumed by the Reply's Write list 4526fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 4539a6a180bSChuck Lever */ 4546911f3e1SChuck Lever static ssize_t svc_rdma_encode_write_list(struct svc_rdma_recv_ctxt *rctxt, 45576e5492bSChuck Lever struct svc_rdma_send_ctxt *sctxt) 4569a6a180bSChuck Lever { 4577a1cbfa1SChuck Lever struct svc_rdma_chunk *chunk; 4586fd5034dSChuck Lever ssize_t len, ret; 4599a6a180bSChuck Lever 460f6ad7759SChuck Lever len = 0; 4616911f3e1SChuck Lever pcl_for_each_chunk(chunk, &rctxt->rc_write_pcl) { 4626911f3e1SChuck Lever ret = svc_rdma_encode_write_chunk(sctxt, chunk); 4636fd5034dSChuck Lever if (ret < 0) 4646fd5034dSChuck Lever return ret; 4656911f3e1SChuck Lever len += ret; 466f6ad7759SChuck Lever } 4679a6a180bSChuck Lever 4686fd5034dSChuck Lever /* Terminate the Write list */ 4696fd5034dSChuck Lever ret = xdr_stream_encode_item_absent(&sctxt->sc_stream); 4706fd5034dSChuck Lever if (ret < 0) 4716fd5034dSChuck Lever return ret; 4726fd5034dSChuck Lever 4736fd5034dSChuck Lever return len + ret; 4749a6a180bSChuck Lever } 4759a6a180bSChuck Lever 4766fd5034dSChuck Lever /** 4776fd5034dSChuck Lever * svc_rdma_encode_reply_chunk - Encode RPC Reply's Reply chunk 4786fd5034dSChuck Lever * @rctxt: Reply context with information about the RPC Call 4796fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 4806fd5034dSChuck Lever * @length: size in bytes of the payload in the Reply chunk 4819a6a180bSChuck Lever * 4826fd5034dSChuck Lever * Return values: 4836fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 4846fd5034dSChuck Lever * that was consumed by the Reply's Reply chunk 4856fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 4866911f3e1SChuck Lever * %-E2BIG if the RPC message is larger than the Reply chunk 4879a6a180bSChuck Lever */ 4886fd5034dSChuck Lever static ssize_t 4896911f3e1SChuck Lever svc_rdma_encode_reply_chunk(struct svc_rdma_recv_ctxt *rctxt, 4906fd5034dSChuck Lever struct svc_rdma_send_ctxt *sctxt, 4916fd5034dSChuck Lever unsigned int length) 4929a6a180bSChuck Lever { 4936911f3e1SChuck Lever struct svc_rdma_chunk *chunk; 4946911f3e1SChuck Lever 4956911f3e1SChuck Lever if (pcl_is_empty(&rctxt->rc_reply_pcl)) 496ded380f1SChuck Lever return xdr_stream_encode_item_absent(&sctxt->sc_stream); 497ded380f1SChuck Lever 4986911f3e1SChuck Lever chunk = pcl_first_chunk(&rctxt->rc_reply_pcl); 4996911f3e1SChuck Lever if (length > chunk->ch_length) 5006911f3e1SChuck Lever return -E2BIG; 5016911f3e1SChuck Lever 5026911f3e1SChuck Lever chunk->ch_payload_length = length; 5036911f3e1SChuck Lever return svc_rdma_encode_write_chunk(sctxt, chunk); 5049a6a180bSChuck Lever } 5059a6a180bSChuck Lever 5062371bcc0SChuck Lever struct svc_rdma_map_data { 5072371bcc0SChuck Lever struct svcxprt_rdma *md_rdma; 5082371bcc0SChuck Lever struct svc_rdma_send_ctxt *md_ctxt; 5092371bcc0SChuck Lever }; 5102371bcc0SChuck Lever 5112371bcc0SChuck Lever /** 5122371bcc0SChuck Lever * svc_rdma_page_dma_map - DMA map one page 5132371bcc0SChuck Lever * @data: pointer to arguments 5142371bcc0SChuck Lever * @page: struct page to DMA map 5152371bcc0SChuck Lever * @offset: offset into the page 5162371bcc0SChuck Lever * @len: number of bytes to map 5172371bcc0SChuck Lever * 5182371bcc0SChuck Lever * Returns: 5192371bcc0SChuck Lever * %0 if DMA mapping was successful 5202371bcc0SChuck Lever * %-EIO if the page cannot be DMA mapped 5212371bcc0SChuck Lever */ 5222371bcc0SChuck Lever static int svc_rdma_page_dma_map(void *data, struct page *page, 5232371bcc0SChuck Lever unsigned long offset, unsigned int len) 5246e6092caSChuck Lever { 5252371bcc0SChuck Lever struct svc_rdma_map_data *args = data; 5262371bcc0SChuck Lever struct svcxprt_rdma *rdma = args->md_rdma; 5272371bcc0SChuck Lever struct svc_rdma_send_ctxt *ctxt = args->md_ctxt; 5286e6092caSChuck Lever struct ib_device *dev = rdma->sc_cm_id->device; 5296e6092caSChuck Lever dma_addr_t dma_addr; 5306e6092caSChuck Lever 5312371bcc0SChuck Lever ++ctxt->sc_cur_sge_no; 5322371bcc0SChuck Lever 5336e6092caSChuck Lever dma_addr = ib_dma_map_page(dev, page, offset, len, DMA_TO_DEVICE); 5346e6092caSChuck Lever if (ib_dma_mapping_error(dev, dma_addr)) 53591a08eaeSChuck Lever goto out_maperr; 5366e6092caSChuck Lever 5372371bcc0SChuck Lever trace_svcrdma_dma_map_page(rdma, dma_addr, len); 53825fd86ecSChuck Lever ctxt->sc_sges[ctxt->sc_cur_sge_no].addr = dma_addr; 53925fd86ecSChuck Lever ctxt->sc_sges[ctxt->sc_cur_sge_no].length = len; 5404201c746SChuck Lever ctxt->sc_send_wr.num_sge++; 5416e6092caSChuck Lever return 0; 54291a08eaeSChuck Lever 54391a08eaeSChuck Lever out_maperr: 5442371bcc0SChuck Lever trace_svcrdma_dma_map_err(rdma, dma_addr, len); 54591a08eaeSChuck Lever return -EIO; 5466e6092caSChuck Lever } 5476e6092caSChuck Lever 5482371bcc0SChuck Lever /** 5492371bcc0SChuck Lever * svc_rdma_iov_dma_map - DMA map an iovec 5502371bcc0SChuck Lever * @data: pointer to arguments 5512371bcc0SChuck Lever * @iov: kvec to DMA map 5522371bcc0SChuck Lever * 5532371bcc0SChuck Lever * ib_dma_map_page() is used here because svc_rdma_dma_unmap() 554f016f305SChuck Lever * handles DMA-unmap and it uses ib_dma_unmap_page() exclusively. 5552371bcc0SChuck Lever * 5562371bcc0SChuck Lever * Returns: 5572371bcc0SChuck Lever * %0 if DMA mapping was successful 5582371bcc0SChuck Lever * %-EIO if the iovec cannot be DMA mapped 559f016f305SChuck Lever */ 5602371bcc0SChuck Lever static int svc_rdma_iov_dma_map(void *data, const struct kvec *iov) 561f016f305SChuck Lever { 5622371bcc0SChuck Lever if (!iov->iov_len) 5632371bcc0SChuck Lever return 0; 5642371bcc0SChuck Lever return svc_rdma_page_dma_map(data, virt_to_page(iov->iov_base), 5652371bcc0SChuck Lever offset_in_page(iov->iov_base), 5662371bcc0SChuck Lever iov->iov_len); 5672371bcc0SChuck Lever } 5682371bcc0SChuck Lever 5692371bcc0SChuck Lever /** 5702371bcc0SChuck Lever * svc_rdma_xb_dma_map - DMA map all segments of an xdr_buf 5712371bcc0SChuck Lever * @xdr: xdr_buf containing portion of an RPC message to transmit 5722371bcc0SChuck Lever * @data: pointer to arguments 5732371bcc0SChuck Lever * 5742371bcc0SChuck Lever * Returns: 5752371bcc0SChuck Lever * %0 if DMA mapping was successful 5762371bcc0SChuck Lever * %-EIO if DMA mapping failed 5772371bcc0SChuck Lever * 5782371bcc0SChuck Lever * On failure, any DMA mappings that have been already done must be 5792371bcc0SChuck Lever * unmapped by the caller. 5802371bcc0SChuck Lever */ 5812371bcc0SChuck Lever static int svc_rdma_xb_dma_map(const struct xdr_buf *xdr, void *data) 5822371bcc0SChuck Lever { 5832371bcc0SChuck Lever unsigned int len, remaining; 5842371bcc0SChuck Lever unsigned long pageoff; 5852371bcc0SChuck Lever struct page **ppages; 5862371bcc0SChuck Lever int ret; 5872371bcc0SChuck Lever 5882371bcc0SChuck Lever ret = svc_rdma_iov_dma_map(data, &xdr->head[0]); 5892371bcc0SChuck Lever if (ret < 0) 5902371bcc0SChuck Lever return ret; 5912371bcc0SChuck Lever 5922371bcc0SChuck Lever ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT); 5932371bcc0SChuck Lever pageoff = offset_in_page(xdr->page_base); 5942371bcc0SChuck Lever remaining = xdr->page_len; 5952371bcc0SChuck Lever while (remaining) { 5962371bcc0SChuck Lever len = min_t(u32, PAGE_SIZE - pageoff, remaining); 5972371bcc0SChuck Lever 5982371bcc0SChuck Lever ret = svc_rdma_page_dma_map(data, *ppages++, pageoff, len); 5992371bcc0SChuck Lever if (ret < 0) 6002371bcc0SChuck Lever return ret; 6012371bcc0SChuck Lever 6022371bcc0SChuck Lever remaining -= len; 6032371bcc0SChuck Lever pageoff = 0; 6042371bcc0SChuck Lever } 6052371bcc0SChuck Lever 6062371bcc0SChuck Lever ret = svc_rdma_iov_dma_map(data, &xdr->tail[0]); 6072371bcc0SChuck Lever if (ret < 0) 6082371bcc0SChuck Lever return ret; 6092371bcc0SChuck Lever 6102371bcc0SChuck Lever return xdr->len; 611f016f305SChuck Lever } 612f016f305SChuck Lever 6139d0b09d5SChuck Lever struct svc_rdma_pullup_data { 6149d0b09d5SChuck Lever u8 *pd_dest; 6159d0b09d5SChuck Lever unsigned int pd_length; 6169d0b09d5SChuck Lever unsigned int pd_num_sges; 6179d0b09d5SChuck Lever }; 6189d0b09d5SChuck Lever 6199d0b09d5SChuck Lever /** 6209d0b09d5SChuck Lever * svc_rdma_xb_count_sges - Count how many SGEs will be needed 6219d0b09d5SChuck Lever * @xdr: xdr_buf containing portion of an RPC message to transmit 6229d0b09d5SChuck Lever * @data: pointer to arguments 6239d0b09d5SChuck Lever * 6249d0b09d5SChuck Lever * Returns: 6259d0b09d5SChuck Lever * Number of SGEs needed to Send the contents of @xdr inline 6269d0b09d5SChuck Lever */ 6279d0b09d5SChuck Lever static int svc_rdma_xb_count_sges(const struct xdr_buf *xdr, 6289d0b09d5SChuck Lever void *data) 6299d0b09d5SChuck Lever { 6309d0b09d5SChuck Lever struct svc_rdma_pullup_data *args = data; 6319d0b09d5SChuck Lever unsigned int remaining; 6329d0b09d5SChuck Lever unsigned long offset; 6339d0b09d5SChuck Lever 6349d0b09d5SChuck Lever if (xdr->head[0].iov_len) 6359d0b09d5SChuck Lever ++args->pd_num_sges; 6369d0b09d5SChuck Lever 6379d0b09d5SChuck Lever offset = offset_in_page(xdr->page_base); 6389d0b09d5SChuck Lever remaining = xdr->page_len; 6399d0b09d5SChuck Lever while (remaining) { 6409d0b09d5SChuck Lever ++args->pd_num_sges; 6419d0b09d5SChuck Lever remaining -= min_t(u32, PAGE_SIZE - offset, remaining); 6429d0b09d5SChuck Lever offset = 0; 6439d0b09d5SChuck Lever } 6449d0b09d5SChuck Lever 6459d0b09d5SChuck Lever if (xdr->tail[0].iov_len) 6469d0b09d5SChuck Lever ++args->pd_num_sges; 6479d0b09d5SChuck Lever 6489d0b09d5SChuck Lever args->pd_length += xdr->len; 6499d0b09d5SChuck Lever return 0; 6509d0b09d5SChuck Lever } 6519d0b09d5SChuck Lever 6526e6092caSChuck Lever /** 6534554755eSChuck Lever * svc_rdma_pull_up_needed - Determine whether to use pull-up 6544554755eSChuck Lever * @rdma: controlling transport 6550dabe948SChuck Lever * @sctxt: send_ctxt for the Send WR 6564554755eSChuck Lever * @rctxt: Write and Reply chunks provided by client 6574554755eSChuck Lever * @xdr: xdr_buf containing RPC message to transmit 6584554755eSChuck Lever * 6594554755eSChuck Lever * Returns: 6604554755eSChuck Lever * %true if pull-up must be used 6614554755eSChuck Lever * %false otherwise 662e248aa7bSChuck Lever */ 6639d0b09d5SChuck Lever static bool svc_rdma_pull_up_needed(const struct svcxprt_rdma *rdma, 6649d0b09d5SChuck Lever const struct svc_rdma_send_ctxt *sctxt, 6654554755eSChuck Lever const struct svc_rdma_recv_ctxt *rctxt, 6669d0b09d5SChuck Lever const struct xdr_buf *xdr) 667e248aa7bSChuck Lever { 6689d0b09d5SChuck Lever /* Resources needed for the transport header */ 6699d0b09d5SChuck Lever struct svc_rdma_pullup_data args = { 6709d0b09d5SChuck Lever .pd_length = sctxt->sc_hdrbuf.len, 6719d0b09d5SChuck Lever .pd_num_sges = 1, 6729d0b09d5SChuck Lever }; 6739d0b09d5SChuck Lever int ret; 674e248aa7bSChuck Lever 6759d0b09d5SChuck Lever ret = pcl_process_nonpayloads(&rctxt->rc_write_pcl, xdr, 6769d0b09d5SChuck Lever svc_rdma_xb_count_sges, &args); 6779d0b09d5SChuck Lever if (ret < 0) 6789d0b09d5SChuck Lever return false; 6799d0b09d5SChuck Lever 6809d0b09d5SChuck Lever if (args.pd_length < RPCRDMA_PULLUP_THRESH) 6810dabe948SChuck Lever return true; 6829d0b09d5SChuck Lever return args.pd_num_sges >= rdma->sc_max_send_sges; 6839d0b09d5SChuck Lever } 6840dabe948SChuck Lever 6859d0b09d5SChuck Lever /** 6869d0b09d5SChuck Lever * svc_rdma_xb_linearize - Copy region of xdr_buf to flat buffer 6879d0b09d5SChuck Lever * @xdr: xdr_buf containing portion of an RPC message to copy 6889d0b09d5SChuck Lever * @data: pointer to arguments 6899d0b09d5SChuck Lever * 6909d0b09d5SChuck Lever * Returns: 6919d0b09d5SChuck Lever * Always zero. 6920dabe948SChuck Lever */ 6939d0b09d5SChuck Lever static int svc_rdma_xb_linearize(const struct xdr_buf *xdr, 6949d0b09d5SChuck Lever void *data) 6959d0b09d5SChuck Lever { 6969d0b09d5SChuck Lever struct svc_rdma_pullup_data *args = data; 6979d0b09d5SChuck Lever unsigned int len, remaining; 698e248aa7bSChuck Lever unsigned long pageoff; 6999d0b09d5SChuck Lever struct page **ppages; 700e248aa7bSChuck Lever 7019d0b09d5SChuck Lever if (xdr->head[0].iov_len) { 7029d0b09d5SChuck Lever memcpy(args->pd_dest, xdr->head[0].iov_base, xdr->head[0].iov_len); 7039d0b09d5SChuck Lever args->pd_dest += xdr->head[0].iov_len; 7049d0b09d5SChuck Lever } 7059d0b09d5SChuck Lever 7069d0b09d5SChuck Lever ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT); 7079d0b09d5SChuck Lever pageoff = offset_in_page(xdr->page_base); 708e248aa7bSChuck Lever remaining = xdr->page_len; 709e248aa7bSChuck Lever while (remaining) { 7109d0b09d5SChuck Lever len = min_t(u32, PAGE_SIZE - pageoff, remaining); 7119d0b09d5SChuck Lever memcpy(args->pd_dest, page_address(*ppages) + pageoff, len); 7129d0b09d5SChuck Lever remaining -= len; 7139d0b09d5SChuck Lever args->pd_dest += len; 714e248aa7bSChuck Lever pageoff = 0; 7159d0b09d5SChuck Lever ppages++; 716e248aa7bSChuck Lever } 717e248aa7bSChuck Lever 7189d0b09d5SChuck Lever if (xdr->tail[0].iov_len) { 7199d0b09d5SChuck Lever memcpy(args->pd_dest, xdr->tail[0].iov_base, xdr->tail[0].iov_len); 7209d0b09d5SChuck Lever args->pd_dest += xdr->tail[0].iov_len; 7219d0b09d5SChuck Lever } 722e248aa7bSChuck Lever 7239d0b09d5SChuck Lever args->pd_length += xdr->len; 7249d0b09d5SChuck Lever return 0; 725e248aa7bSChuck Lever } 726e248aa7bSChuck Lever 7274554755eSChuck Lever /** 7284554755eSChuck Lever * svc_rdma_pull_up_reply_msg - Copy Reply into a single buffer 7294554755eSChuck Lever * @rdma: controlling transport 7304554755eSChuck Lever * @sctxt: send_ctxt for the Send WR; xprt hdr is already prepared 7314554755eSChuck Lever * @rctxt: Write and Reply chunks provided by client 7324554755eSChuck Lever * @xdr: prepared xdr_buf containing RPC message 7334554755eSChuck Lever * 7344554755eSChuck Lever * The device is not capable of sending the reply directly. 7354554755eSChuck Lever * Assemble the elements of @xdr into the transport header buffer. 7364554755eSChuck Lever * 7379d0b09d5SChuck Lever * Assumptions: 7389d0b09d5SChuck Lever * pull_up_needed has determined that @xdr will fit in the buffer. 7399d0b09d5SChuck Lever * 7409d0b09d5SChuck Lever * Returns: 7419d0b09d5SChuck Lever * %0 if pull-up was successful 7429d0b09d5SChuck Lever * %-EMSGSIZE if a buffer manipulation problem occurred 743e248aa7bSChuck Lever */ 7449d0b09d5SChuck Lever static int svc_rdma_pull_up_reply_msg(const struct svcxprt_rdma *rdma, 7454554755eSChuck Lever struct svc_rdma_send_ctxt *sctxt, 7464554755eSChuck Lever const struct svc_rdma_recv_ctxt *rctxt, 7474554755eSChuck Lever const struct xdr_buf *xdr) 748e248aa7bSChuck Lever { 7499d0b09d5SChuck Lever struct svc_rdma_pullup_data args = { 7509d0b09d5SChuck Lever .pd_dest = sctxt->sc_xprt_buf + sctxt->sc_hdrbuf.len, 7519d0b09d5SChuck Lever }; 7529d0b09d5SChuck Lever int ret; 753e248aa7bSChuck Lever 7549d0b09d5SChuck Lever ret = pcl_process_nonpayloads(&rctxt->rc_write_pcl, xdr, 7559d0b09d5SChuck Lever svc_rdma_xb_linearize, &args); 7569d0b09d5SChuck Lever if (ret < 0) 7579d0b09d5SChuck Lever return ret; 758e248aa7bSChuck Lever 7599d0b09d5SChuck Lever sctxt->sc_sges[0].length = sctxt->sc_hdrbuf.len + args.pd_length; 7609d0b09d5SChuck Lever trace_svcrdma_send_pullup(sctxt, args.pd_length); 761e248aa7bSChuck Lever return 0; 762e248aa7bSChuck Lever } 763e248aa7bSChuck Lever 7644554755eSChuck Lever /* svc_rdma_map_reply_msg - DMA map the buffer holding RPC message 76599722fe4SChuck Lever * @rdma: controlling transport 7664554755eSChuck Lever * @sctxt: send_ctxt for the Send WR 7674554755eSChuck Lever * @rctxt: Write and Reply chunks provided by client 76899722fe4SChuck Lever * @xdr: prepared xdr_buf containing RPC message 76999722fe4SChuck Lever * 7702371bcc0SChuck Lever * Returns: 7712371bcc0SChuck Lever * %0 if DMA mapping was successful. 7722371bcc0SChuck Lever * %-EMSGSIZE if a buffer manipulation problem occurred 7732371bcc0SChuck Lever * %-EIO if DMA mapping failed 7749a6a180bSChuck Lever * 7752371bcc0SChuck Lever * The Send WR's num_sge field is set in all cases. 776c06b540aSTom Tucker */ 77799722fe4SChuck Lever int svc_rdma_map_reply_msg(struct svcxprt_rdma *rdma, 7784554755eSChuck Lever struct svc_rdma_send_ctxt *sctxt, 7794554755eSChuck Lever const struct svc_rdma_recv_ctxt *rctxt, 7802371bcc0SChuck Lever const struct xdr_buf *xdr) 781c06b540aSTom Tucker { 7822371bcc0SChuck Lever struct svc_rdma_map_data args = { 7832371bcc0SChuck Lever .md_rdma = rdma, 7842371bcc0SChuck Lever .md_ctxt = sctxt, 7852371bcc0SChuck Lever }; 786c06b540aSTom Tucker 787aee4b74aSChuck Lever /* Set up the (persistently-mapped) transport header SGE. */ 788aee4b74aSChuck Lever sctxt->sc_send_wr.num_sge = 1; 789aee4b74aSChuck Lever sctxt->sc_sges[0].length = sctxt->sc_hdrbuf.len; 790aee4b74aSChuck Lever 791aee4b74aSChuck Lever /* If there is a Reply chunk, nothing follows the transport 792aee4b74aSChuck Lever * header, and we're done here. 793aee4b74aSChuck Lever */ 7942371bcc0SChuck Lever if (!pcl_is_empty(&rctxt->rc_reply_pcl)) 795aee4b74aSChuck Lever return 0; 796aee4b74aSChuck Lever 797aee4b74aSChuck Lever /* For pull-up, svc_rdma_send() will sync the transport header. 798aee4b74aSChuck Lever * No additional DMA mapping is necessary. 799aee4b74aSChuck Lever */ 8000dabe948SChuck Lever if (svc_rdma_pull_up_needed(rdma, sctxt, rctxt, xdr)) 8014554755eSChuck Lever return svc_rdma_pull_up_reply_msg(rdma, sctxt, rctxt, xdr); 802e248aa7bSChuck Lever 8032371bcc0SChuck Lever return pcl_process_nonpayloads(&rctxt->rc_write_pcl, xdr, 8042371bcc0SChuck Lever svc_rdma_xb_dma_map, &args); 805c06b540aSTom Tucker } 806c06b540aSTom Tucker 807*a944209cSChuck Lever /* The svc_rqst and all resources it owns are released as soon as 808*a944209cSChuck Lever * svc_rdma_sendto returns. Transfer pages under I/O to the ctxt 809*a944209cSChuck Lever * so they are released by the Send completion handler. 810*a944209cSChuck Lever */ 811*a944209cSChuck Lever static inline void svc_rdma_save_io_pages(struct svc_rqst *rqstp, 812*a944209cSChuck Lever struct svc_rdma_send_ctxt *ctxt) 813*a944209cSChuck Lever { 814*a944209cSChuck Lever int i, pages = rqstp->rq_next_page - rqstp->rq_respages; 815*a944209cSChuck Lever 816*a944209cSChuck Lever ctxt->sc_page_count += pages; 817*a944209cSChuck Lever for (i = 0; i < pages; i++) { 818*a944209cSChuck Lever ctxt->sc_pages[i] = rqstp->rq_respages[i]; 819*a944209cSChuck Lever rqstp->rq_respages[i] = NULL; 820*a944209cSChuck Lever } 821*a944209cSChuck Lever 822*a944209cSChuck Lever /* Prevent svc_xprt_release from releasing pages in rq_pages */ 823*a944209cSChuck Lever rqstp->rq_next_page = rqstp->rq_respages; 824*a944209cSChuck Lever } 825*a944209cSChuck Lever 8269a6a180bSChuck Lever /* Prepare the portion of the RPC Reply that will be transmitted 8279a6a180bSChuck Lever * via RDMA Send. The RPC-over-RDMA transport header is prepared 8284201c746SChuck Lever * in sc_sges[0], and the RPC xdr_buf is prepared in following sges. 8299a6a180bSChuck Lever * 8309a6a180bSChuck Lever * Depending on whether a Write list or Reply chunk is present, 8319a6a180bSChuck Lever * the server may send all, a portion of, or none of the xdr_buf. 8324201c746SChuck Lever * In the latter case, only the transport header (sc_sges[0]) is 8339a6a180bSChuck Lever * transmitted. 8349a6a180bSChuck Lever * 8359a6a180bSChuck Lever * RDMA Send is the last step of transmitting an RPC reply. Pages 8369a6a180bSChuck Lever * involved in the earlier RDMA Writes are here transferred out 83797bce634SChuck Lever * of the rqstp and into the sctxt's page array. These pages are 8389a6a180bSChuck Lever * DMA unmapped by each Write completion, but the subsequent Send 8399a6a180bSChuck Lever * completion finally releases these pages. 8409a6a180bSChuck Lever * 8419a6a180bSChuck Lever * Assumptions: 8429a6a180bSChuck Lever * - The Reply's transport header will never be larger than a page. 843c06b540aSTom Tucker */ 8449a6a180bSChuck Lever static int svc_rdma_send_reply_msg(struct svcxprt_rdma *rdma, 84597bce634SChuck Lever struct svc_rdma_send_ctxt *sctxt, 846db9602e4SChuck Lever const struct svc_rdma_recv_ctxt *rctxt, 847db9602e4SChuck Lever struct svc_rqst *rqstp) 848c06b540aSTom Tucker { 8499a6a180bSChuck Lever int ret; 8500e7f011aSTom Tucker 851aee4b74aSChuck Lever ret = svc_rdma_map_reply_msg(rdma, sctxt, rctxt, &rqstp->rq_res); 8529a6a180bSChuck Lever if (ret < 0) 85399722fe4SChuck Lever return ret; 854c06b540aSTom Tucker 85597bce634SChuck Lever if (rctxt->rc_inv_rkey) { 85697bce634SChuck Lever sctxt->sc_send_wr.opcode = IB_WR_SEND_WITH_INV; 85797bce634SChuck Lever sctxt->sc_send_wr.ex.invalidate_rkey = rctxt->rc_inv_rkey; 85897bce634SChuck Lever } else { 85997bce634SChuck Lever sctxt->sc_send_wr.opcode = IB_WR_SEND; 860986b7889SChuck Lever } 8612a1e4f21SChuck Lever 8622a1e4f21SChuck Lever ret = svc_rdma_send(rdma, sctxt); 8632a1e4f21SChuck Lever if (ret < 0) 8642a1e4f21SChuck Lever return ret; 8652a1e4f21SChuck Lever 8662a1e4f21SChuck Lever ret = wait_for_completion_killable(&sctxt->sc_done); 8672a1e4f21SChuck Lever svc_rdma_send_ctxt_put(rdma, sctxt); 8682a1e4f21SChuck Lever return ret; 869c06b540aSTom Tucker } 870c06b540aSTom Tucker 871c65b326bSChuck Lever /** 872c65b326bSChuck Lever * svc_rdma_send_error_msg - Send an RPC/RDMA v1 error response 873c65b326bSChuck Lever * @rdma: controlling transport context 874c65b326bSChuck Lever * @sctxt: Send context for the response 875c65b326bSChuck Lever * @rctxt: Receive context for incoming bad message 876c65b326bSChuck Lever * @status: negative errno indicating error that occurred 8774757d90bSChuck Lever * 878c65b326bSChuck Lever * Given the client-provided Read, Write, and Reply chunks, the 879c65b326bSChuck Lever * server was not able to parse the Call or form a complete Reply. 880c65b326bSChuck Lever * Return an RDMA_ERROR message so the client can retire the RPC 881c65b326bSChuck Lever * transaction. 882c65b326bSChuck Lever * 883c65b326bSChuck Lever * The caller does not have to release @sctxt. It is released by 884c65b326bSChuck Lever * Send completion, or by this function on error. 8854757d90bSChuck Lever */ 886c65b326bSChuck Lever void svc_rdma_send_error_msg(struct svcxprt_rdma *rdma, 887d1f6e236SChuck Lever struct svc_rdma_send_ctxt *sctxt, 8884f200bd8SChuck Lever struct svc_rdma_recv_ctxt *rctxt, 8894f200bd8SChuck Lever int status) 8904757d90bSChuck Lever { 8916fd5034dSChuck Lever __be32 *rdma_argp = rctxt->rc_recv_buf; 8924757d90bSChuck Lever __be32 *p; 8934757d90bSChuck Lever 894d1f6e236SChuck Lever rpcrdma_set_xdrlen(&sctxt->sc_hdrbuf, 0); 895d1f6e236SChuck Lever xdr_init_encode(&sctxt->sc_stream, &sctxt->sc_hdrbuf, 896d1f6e236SChuck Lever sctxt->sc_xprt_buf, NULL); 8976fd5034dSChuck Lever 8984f200bd8SChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, 8994f200bd8SChuck Lever rpcrdma_fixed_maxsz * sizeof(*p)); 9006fd5034dSChuck Lever if (!p) 901605c61beSChuck Lever goto put_ctxt; 9026fd5034dSChuck Lever 9036fd5034dSChuck Lever *p++ = *rdma_argp; 9046fd5034dSChuck Lever *p++ = *(rdma_argp + 1); 9056fd5034dSChuck Lever *p++ = rdma->sc_fc_credits; 9064f200bd8SChuck Lever *p = rdma_error; 9074f200bd8SChuck Lever 9084f200bd8SChuck Lever switch (status) { 9094f200bd8SChuck Lever case -EPROTONOSUPPORT: 9104f200bd8SChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, 3 * sizeof(*p)); 9114f200bd8SChuck Lever if (!p) 912605c61beSChuck Lever goto put_ctxt; 9134f200bd8SChuck Lever 9144f200bd8SChuck Lever *p++ = err_vers; 9154f200bd8SChuck Lever *p++ = rpcrdma_version; 9164f200bd8SChuck Lever *p = rpcrdma_version; 9174f200bd8SChuck Lever trace_svcrdma_err_vers(*rdma_argp); 9184f200bd8SChuck Lever break; 9194f200bd8SChuck Lever default: 9204f200bd8SChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, sizeof(*p)); 9214f200bd8SChuck Lever if (!p) 922605c61beSChuck Lever goto put_ctxt; 9234f200bd8SChuck Lever 9244757d90bSChuck Lever *p = err_chunk; 9256fd5034dSChuck Lever trace_svcrdma_err_chunk(*rdma_argp); 9264f200bd8SChuck Lever } 9276fd5034dSChuck Lever 928c65b326bSChuck Lever /* Remote Invalidation is skipped for simplicity. */ 929d1f6e236SChuck Lever sctxt->sc_send_wr.num_sge = 1; 930d1f6e236SChuck Lever sctxt->sc_send_wr.opcode = IB_WR_SEND; 931d1f6e236SChuck Lever sctxt->sc_sges[0].length = sctxt->sc_hdrbuf.len; 93217f70f8dSChuck Lever if (svc_rdma_send(rdma, sctxt)) 933605c61beSChuck Lever goto put_ctxt; 9342a1e4f21SChuck Lever 9352a1e4f21SChuck Lever wait_for_completion_killable(&sctxt->sc_done); 936605c61beSChuck Lever 937605c61beSChuck Lever put_ctxt: 938605c61beSChuck Lever svc_rdma_send_ctxt_put(rdma, sctxt); 93999722fe4SChuck Lever } 94099722fe4SChuck Lever 9419a6a180bSChuck Lever /** 9429a6a180bSChuck Lever * svc_rdma_sendto - Transmit an RPC reply 9439a6a180bSChuck Lever * @rqstp: processed RPC request, reply XDR already in ::rq_res 9449a6a180bSChuck Lever * 9459a6a180bSChuck Lever * Any resources still associated with @rqstp are released upon return. 9469a6a180bSChuck Lever * If no reply message was possible, the connection is closed. 9479a6a180bSChuck Lever * 9489a6a180bSChuck Lever * Returns: 9499a6a180bSChuck Lever * %0 if an RPC reply has been successfully posted, 9509a6a180bSChuck Lever * %-ENOMEM if a resource shortage occurred (connection is lost), 9519a6a180bSChuck Lever * %-ENOTCONN if posting failed (connection is lost). 9529a6a180bSChuck Lever */ 953c06b540aSTom Tucker int svc_rdma_sendto(struct svc_rqst *rqstp) 954c06b540aSTom Tucker { 955c06b540aSTom Tucker struct svc_xprt *xprt = rqstp->rq_xprt; 956c06b540aSTom Tucker struct svcxprt_rdma *rdma = 957c06b540aSTom Tucker container_of(xprt, struct svcxprt_rdma, sc_xprt); 9583a88092eSChuck Lever struct svc_rdma_recv_ctxt *rctxt = rqstp->rq_xprt_ctxt; 9592fe8c446SChuck Lever __be32 *rdma_argp = rctxt->rc_recv_buf; 96099722fe4SChuck Lever struct svc_rdma_send_ctxt *sctxt; 9618727f788SChuck Lever unsigned int rc_size; 9626fd5034dSChuck Lever __be32 *p; 9639a6a180bSChuck Lever int ret; 964c06b540aSTom Tucker 965ca4faf54SChuck Lever ret = -ENOTCONN; 966ca4faf54SChuck Lever if (svc_xprt_is_dead(xprt)) 967c7731d5eSChuck Lever goto drop_connection; 968ca4faf54SChuck Lever 96978da2b3cSChuck Lever ret = -ENOMEM; 97099722fe4SChuck Lever sctxt = svc_rdma_send_ctxt_get(rdma); 97199722fe4SChuck Lever if (!sctxt) 972c7731d5eSChuck Lever goto drop_connection; 97398fc21d3SChuck Lever 9748727f788SChuck Lever ret = -EMSGSIZE; 9756fd5034dSChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, 9766fd5034dSChuck Lever rpcrdma_fixed_maxsz * sizeof(*p)); 9776fd5034dSChuck Lever if (!p) 978c7731d5eSChuck Lever goto put_ctxt; 979ded380f1SChuck Lever 980ded380f1SChuck Lever ret = svc_rdma_send_reply_chunk(rdma, rctxt, &rqstp->rq_res); 981ded380f1SChuck Lever if (ret < 0) 982c7731d5eSChuck Lever goto reply_chunk; 9838727f788SChuck Lever rc_size = ret; 984ded380f1SChuck Lever 9859a6a180bSChuck Lever *p++ = *rdma_argp; 9869a6a180bSChuck Lever *p++ = *(rdma_argp + 1); 98798fc21d3SChuck Lever *p++ = rdma->sc_fc_credits; 9886911f3e1SChuck Lever *p = pcl_is_empty(&rctxt->rc_reply_pcl) ? rdma_msg : rdma_nomsg; 98998fc21d3SChuck Lever 9908727f788SChuck Lever ret = svc_rdma_encode_read_list(sctxt); 9918727f788SChuck Lever if (ret < 0) 992c7731d5eSChuck Lever goto put_ctxt; 9938727f788SChuck Lever ret = svc_rdma_encode_write_list(rctxt, sctxt); 9948727f788SChuck Lever if (ret < 0) 995c7731d5eSChuck Lever goto put_ctxt; 9968727f788SChuck Lever ret = svc_rdma_encode_reply_chunk(rctxt, sctxt, rc_size); 9978727f788SChuck Lever if (ret < 0) 998c7731d5eSChuck Lever goto put_ctxt; 999c06b540aSTom Tucker 1000db9602e4SChuck Lever ret = svc_rdma_send_reply_msg(rdma, sctxt, rctxt, rqstp); 10013e1eeb98SChuck Lever if (ret < 0) 1002c7731d5eSChuck Lever goto put_ctxt; 100323cf1ee1SChuck Lever return 0; 1004afd566eaSTom Tucker 1005c7731d5eSChuck Lever reply_chunk: 1006b20dae70SColin Ian King if (ret != -E2BIG && ret != -EINVAL) 1007c7731d5eSChuck Lever goto put_ctxt; 10084757d90bSChuck Lever 1009605c61beSChuck Lever svc_rdma_send_error_msg(rdma, sctxt, rctxt, ret); 101023cf1ee1SChuck Lever return 0; 10114757d90bSChuck Lever 1012c7731d5eSChuck Lever put_ctxt: 101399722fe4SChuck Lever svc_rdma_send_ctxt_put(rdma, sctxt); 1014c7731d5eSChuck Lever drop_connection: 10153f8f25c6SChuck Lever trace_svcrdma_send_err(rqstp, ret); 1016e844d307SChuck Lever svc_xprt_deferred_close(&rdma->sc_xprt); 101723cf1ee1SChuck Lever return -ENOTCONN; 1018c06b540aSTom Tucker } 101941205539SChuck Lever 102041205539SChuck Lever /** 102103493bcaSChuck Lever * svc_rdma_result_payload - special processing for a result payload 102241205539SChuck Lever * @rqstp: svc_rqst to operate on 102341205539SChuck Lever * @offset: payload's byte offset in @xdr 102441205539SChuck Lever * @length: size of payload, in bytes 102541205539SChuck Lever * 1026f6ad7759SChuck Lever * Return values: 1027f6ad7759SChuck Lever * %0 if successful or nothing needed to be done 1028f6ad7759SChuck Lever * %-EMSGSIZE on XDR buffer overflow 1029f6ad7759SChuck Lever * %-E2BIG if the payload was larger than the Write chunk 1030f6ad7759SChuck Lever * %-EINVAL if client provided too many segments 1031f6ad7759SChuck Lever * %-ENOMEM if rdma_rw context pool was exhausted 1032f6ad7759SChuck Lever * %-ENOTCONN if posting failed (connection is lost) 1033f6ad7759SChuck Lever * %-EIO if rdma_rw initialization failed (DMA mapping, etc) 103441205539SChuck Lever */ 103503493bcaSChuck Lever int svc_rdma_result_payload(struct svc_rqst *rqstp, unsigned int offset, 103641205539SChuck Lever unsigned int length) 103741205539SChuck Lever { 103841205539SChuck Lever struct svc_rdma_recv_ctxt *rctxt = rqstp->rq_xprt_ctxt; 10397a1cbfa1SChuck Lever struct svc_rdma_chunk *chunk; 1040f6ad7759SChuck Lever struct svcxprt_rdma *rdma; 1041f6ad7759SChuck Lever struct xdr_buf subbuf; 1042f6ad7759SChuck Lever int ret; 1043f6ad7759SChuck Lever 10447a1cbfa1SChuck Lever chunk = rctxt->rc_cur_result_payload; 10457a1cbfa1SChuck Lever if (!length || !chunk) 1046f6ad7759SChuck Lever return 0; 10477a1cbfa1SChuck Lever rctxt->rc_cur_result_payload = 10487a1cbfa1SChuck Lever pcl_next_chunk(&rctxt->rc_write_pcl, chunk); 10497a1cbfa1SChuck Lever if (length > chunk->ch_length) 10507a1cbfa1SChuck Lever return -E2BIG; 105141205539SChuck Lever 10527a1cbfa1SChuck Lever chunk->ch_position = offset; 10537a1cbfa1SChuck Lever chunk->ch_payload_length = length; 105441205539SChuck Lever 1055f6ad7759SChuck Lever if (xdr_buf_subsegment(&rqstp->rq_res, &subbuf, offset, length)) 1056f6ad7759SChuck Lever return -EMSGSIZE; 1057f6ad7759SChuck Lever 1058f6ad7759SChuck Lever rdma = container_of(rqstp->rq_xprt, struct svcxprt_rdma, sc_xprt); 10597a1cbfa1SChuck Lever ret = svc_rdma_send_write_chunk(rdma, chunk, &subbuf); 1060f6ad7759SChuck Lever if (ret < 0) 1061f6ad7759SChuck Lever return ret; 106241205539SChuck Lever return 0; 106341205539SChuck Lever } 1064