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> 10998895edbSChuck Lever #include <linux/sunrpc/rpc_rdma.h> 110c06b540aSTom Tucker #include <linux/sunrpc/svc_rdma.h> 111c06b540aSTom Tucker 11298895edbSChuck Lever #include "xprt_rdma.h" 11398895edbSChuck Lever #include <trace/events/rpcrdma.h> 11498895edbSChuck Lever 115c06b540aSTom Tucker #define RPCDBG_FACILITY RPCDBG_SVCXPRT 116c06b540aSTom Tucker 1174201c746SChuck Lever static void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc); 1184201c746SChuck Lever 1194201c746SChuck Lever static inline struct svc_rdma_send_ctxt * 1204201c746SChuck Lever svc_rdma_next_send_ctxt(struct list_head *list) 1214201c746SChuck Lever { 1224201c746SChuck Lever return list_first_entry_or_null(list, struct svc_rdma_send_ctxt, 1234201c746SChuck Lever sc_list); 1244201c746SChuck Lever } 1254201c746SChuck Lever 1264201c746SChuck Lever static struct svc_rdma_send_ctxt * 1274201c746SChuck Lever svc_rdma_send_ctxt_alloc(struct svcxprt_rdma *rdma) 1284201c746SChuck Lever { 1294201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt; 13099722fe4SChuck Lever dma_addr_t addr; 13199722fe4SChuck Lever void *buffer; 13225fd86ecSChuck Lever size_t size; 1334201c746SChuck Lever int i; 1344201c746SChuck Lever 13525fd86ecSChuck Lever size = sizeof(*ctxt); 13625fd86ecSChuck Lever size += rdma->sc_max_send_sges * sizeof(struct ib_sge); 13725fd86ecSChuck Lever ctxt = kmalloc(size, GFP_KERNEL); 1384201c746SChuck Lever if (!ctxt) 13999722fe4SChuck Lever goto fail0; 14099722fe4SChuck Lever buffer = kmalloc(rdma->sc_max_req_size, GFP_KERNEL); 14199722fe4SChuck Lever if (!buffer) 14299722fe4SChuck Lever goto fail1; 14399722fe4SChuck Lever addr = ib_dma_map_single(rdma->sc_pd->device, buffer, 14499722fe4SChuck Lever rdma->sc_max_req_size, DMA_TO_DEVICE); 14599722fe4SChuck Lever if (ib_dma_mapping_error(rdma->sc_pd->device, addr)) 14699722fe4SChuck Lever goto fail2; 1474201c746SChuck Lever 1484201c746SChuck Lever ctxt->sc_send_wr.next = NULL; 1494201c746SChuck Lever ctxt->sc_send_wr.wr_cqe = &ctxt->sc_cqe; 1504201c746SChuck Lever ctxt->sc_send_wr.sg_list = ctxt->sc_sges; 1514201c746SChuck Lever ctxt->sc_send_wr.send_flags = IB_SEND_SIGNALED; 15299722fe4SChuck Lever ctxt->sc_cqe.done = svc_rdma_wc_send; 15399722fe4SChuck Lever ctxt->sc_xprt_buf = buffer; 1546fd5034dSChuck Lever xdr_buf_init(&ctxt->sc_hdrbuf, ctxt->sc_xprt_buf, 1556fd5034dSChuck Lever rdma->sc_max_req_size); 15699722fe4SChuck Lever ctxt->sc_sges[0].addr = addr; 15799722fe4SChuck Lever 15825fd86ecSChuck Lever for (i = 0; i < rdma->sc_max_send_sges; i++) 1594201c746SChuck Lever ctxt->sc_sges[i].lkey = rdma->sc_pd->local_dma_lkey; 1604201c746SChuck Lever return ctxt; 16199722fe4SChuck Lever 16299722fe4SChuck Lever fail2: 16399722fe4SChuck Lever kfree(buffer); 16499722fe4SChuck Lever fail1: 16599722fe4SChuck Lever kfree(ctxt); 16699722fe4SChuck Lever fail0: 16799722fe4SChuck Lever return NULL; 1684201c746SChuck Lever } 1694201c746SChuck Lever 1704201c746SChuck Lever /** 1714201c746SChuck Lever * svc_rdma_send_ctxts_destroy - Release all send_ctxt's for an xprt 1724201c746SChuck Lever * @rdma: svcxprt_rdma being torn down 1734201c746SChuck Lever * 1744201c746SChuck Lever */ 1754201c746SChuck Lever void svc_rdma_send_ctxts_destroy(struct svcxprt_rdma *rdma) 1764201c746SChuck Lever { 1774201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt; 1784201c746SChuck Lever 1794201c746SChuck Lever while ((ctxt = svc_rdma_next_send_ctxt(&rdma->sc_send_ctxts))) { 1804201c746SChuck Lever list_del(&ctxt->sc_list); 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; 2004201c746SChuck Lever 2014201c746SChuck Lever spin_lock(&rdma->sc_send_lock); 2024201c746SChuck Lever ctxt = svc_rdma_next_send_ctxt(&rdma->sc_send_ctxts); 2034201c746SChuck Lever if (!ctxt) 2044201c746SChuck Lever goto out_empty; 2054201c746SChuck Lever list_del(&ctxt->sc_list); 2064201c746SChuck Lever spin_unlock(&rdma->sc_send_lock); 2074201c746SChuck Lever 2084201c746SChuck Lever out: 2096fd5034dSChuck Lever rpcrdma_set_xdrlen(&ctxt->sc_hdrbuf, 0); 2106fd5034dSChuck Lever xdr_init_encode(&ctxt->sc_stream, &ctxt->sc_hdrbuf, 2116fd5034dSChuck Lever ctxt->sc_xprt_buf, NULL); 2126fd5034dSChuck Lever 2134201c746SChuck Lever ctxt->sc_send_wr.num_sge = 0; 21499722fe4SChuck Lever ctxt->sc_cur_sge_no = 0; 2154201c746SChuck Lever ctxt->sc_page_count = 0; 2164201c746SChuck Lever return ctxt; 2174201c746SChuck Lever 2184201c746SChuck Lever out_empty: 2194201c746SChuck Lever spin_unlock(&rdma->sc_send_lock); 2204201c746SChuck Lever ctxt = svc_rdma_send_ctxt_alloc(rdma); 2214201c746SChuck Lever if (!ctxt) 2224201c746SChuck Lever return NULL; 2234201c746SChuck Lever goto out; 2244201c746SChuck Lever } 2254201c746SChuck Lever 2264201c746SChuck Lever /** 2274201c746SChuck Lever * svc_rdma_send_ctxt_put - Return send_ctxt to free list 2284201c746SChuck Lever * @rdma: controlling svcxprt_rdma 2294201c746SChuck Lever * @ctxt: object to return to the free list 2304201c746SChuck Lever * 2314201c746SChuck Lever * Pages left in sc_pages are DMA unmapped and released. 2324201c746SChuck Lever */ 2334201c746SChuck Lever void svc_rdma_send_ctxt_put(struct svcxprt_rdma *rdma, 2344201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt) 2354201c746SChuck Lever { 2364201c746SChuck Lever struct ib_device *device = rdma->sc_cm_id->device; 2374201c746SChuck Lever unsigned int i; 2384201c746SChuck Lever 23999722fe4SChuck Lever /* The first SGE contains the transport header, which 24099722fe4SChuck Lever * remains mapped until @ctxt is destroyed. 24199722fe4SChuck Lever */ 242832b2cb9SChuck Lever for (i = 1; i < ctxt->sc_send_wr.num_sge; i++) { 2434201c746SChuck Lever ib_dma_unmap_page(device, 2444201c746SChuck Lever ctxt->sc_sges[i].addr, 2454201c746SChuck Lever ctxt->sc_sges[i].length, 2464201c746SChuck Lever DMA_TO_DEVICE); 247832b2cb9SChuck Lever trace_svcrdma_dma_unmap_page(rdma, 248832b2cb9SChuck Lever ctxt->sc_sges[i].addr, 249832b2cb9SChuck Lever ctxt->sc_sges[i].length); 250832b2cb9SChuck Lever } 2514201c746SChuck Lever 2524201c746SChuck Lever for (i = 0; i < ctxt->sc_page_count; ++i) 2534201c746SChuck Lever put_page(ctxt->sc_pages[i]); 2544201c746SChuck Lever 2554201c746SChuck Lever spin_lock(&rdma->sc_send_lock); 2564201c746SChuck Lever list_add(&ctxt->sc_list, &rdma->sc_send_ctxts); 2574201c746SChuck Lever spin_unlock(&rdma->sc_send_lock); 2584201c746SChuck Lever } 2594201c746SChuck Lever 2604201c746SChuck Lever /** 2614201c746SChuck Lever * svc_rdma_wc_send - Invoked by RDMA provider for each polled Send WC 2624201c746SChuck Lever * @cq: Completion Queue context 2634201c746SChuck Lever * @wc: Work Completion object 2644201c746SChuck Lever * 2654201c746SChuck Lever * NB: The svc_xprt/svcxprt_rdma is pinned whenever it's possible that 2664201c746SChuck Lever * the Send completion handler could be running. 2674201c746SChuck Lever */ 2684201c746SChuck Lever static void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc) 2694201c746SChuck Lever { 2704201c746SChuck Lever struct svcxprt_rdma *rdma = cq->cq_context; 2714201c746SChuck Lever struct ib_cqe *cqe = wc->wr_cqe; 2724201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt; 2734201c746SChuck Lever 2744201c746SChuck Lever trace_svcrdma_wc_send(wc); 2754201c746SChuck Lever 2764201c746SChuck Lever atomic_inc(&rdma->sc_sq_avail); 2774201c746SChuck Lever wake_up(&rdma->sc_send_wait); 2784201c746SChuck Lever 2794201c746SChuck Lever ctxt = container_of(cqe, struct svc_rdma_send_ctxt, sc_cqe); 2804201c746SChuck Lever svc_rdma_send_ctxt_put(rdma, ctxt); 2814201c746SChuck Lever 2824201c746SChuck Lever if (unlikely(wc->status != IB_WC_SUCCESS)) { 2834201c746SChuck Lever set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags); 2844201c746SChuck Lever svc_xprt_enqueue(&rdma->sc_xprt); 2854201c746SChuck Lever } 2864201c746SChuck Lever 2874201c746SChuck Lever svc_xprt_put(&rdma->sc_xprt); 2884201c746SChuck Lever } 2894201c746SChuck Lever 2903abb03faSChuck Lever /** 2913abb03faSChuck Lever * svc_rdma_send - Post a single Send WR 2923abb03faSChuck Lever * @rdma: transport on which to post the WR 2933abb03faSChuck Lever * @wr: prepared Send WR to post 2943abb03faSChuck Lever * 2953abb03faSChuck Lever * Returns zero the Send WR was posted successfully. Otherwise, a 2963abb03faSChuck Lever * negative errno is returned. 2973abb03faSChuck Lever */ 2984201c746SChuck Lever int svc_rdma_send(struct svcxprt_rdma *rdma, struct ib_send_wr *wr) 2994201c746SChuck Lever { 3004201c746SChuck Lever int ret; 3014201c746SChuck Lever 3023abb03faSChuck Lever might_sleep(); 3034201c746SChuck Lever 304aee4b74aSChuck Lever /* Sync the transport header buffer */ 305aee4b74aSChuck Lever ib_dma_sync_single_for_device(rdma->sc_pd->device, 306aee4b74aSChuck Lever wr->sg_list[0].addr, 307aee4b74aSChuck Lever wr->sg_list[0].length, 308aee4b74aSChuck Lever DMA_TO_DEVICE); 309aee4b74aSChuck Lever 3104201c746SChuck Lever /* If the SQ is full, wait until an SQ entry is available */ 3114201c746SChuck Lever while (1) { 3123abb03faSChuck Lever if ((atomic_dec_return(&rdma->sc_sq_avail) < 0)) { 3134201c746SChuck Lever atomic_inc(&rdma_stat_sq_starve); 3144201c746SChuck Lever trace_svcrdma_sq_full(rdma); 3153abb03faSChuck Lever atomic_inc(&rdma->sc_sq_avail); 3164201c746SChuck Lever wait_event(rdma->sc_send_wait, 3173abb03faSChuck Lever atomic_read(&rdma->sc_sq_avail) > 1); 3184201c746SChuck Lever if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags)) 3194201c746SChuck Lever return -ENOTCONN; 3204201c746SChuck Lever trace_svcrdma_sq_retry(rdma); 3214201c746SChuck Lever continue; 3224201c746SChuck Lever } 3234201c746SChuck Lever 3243abb03faSChuck Lever svc_xprt_get(&rdma->sc_xprt); 325e28b4fc6SChuck Lever trace_svcrdma_post_send(wr); 326ed288d74SBart Van Assche ret = ib_post_send(rdma->sc_qp, wr, NULL); 327e28b4fc6SChuck Lever if (ret) 328e28b4fc6SChuck Lever break; 329e28b4fc6SChuck Lever return 0; 330e28b4fc6SChuck Lever } 331e28b4fc6SChuck Lever 332e28b4fc6SChuck Lever trace_svcrdma_sq_post_err(rdma, ret); 3334201c746SChuck Lever set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags); 3344201c746SChuck Lever svc_xprt_put(&rdma->sc_xprt); 3354201c746SChuck Lever wake_up(&rdma->sc_send_wait); 3364201c746SChuck Lever return ret; 3374201c746SChuck Lever } 3384201c746SChuck Lever 3396fd5034dSChuck Lever /** 3406fd5034dSChuck Lever * svc_rdma_encode_read_list - Encode RPC Reply's Read chunk list 3416fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 3429a6a180bSChuck Lever * 3436fd5034dSChuck Lever * Return values: 3446fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 3456fd5034dSChuck Lever * that was consumed by the Reply Read list 3466fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 3479a6a180bSChuck Lever */ 3486fd5034dSChuck Lever static ssize_t svc_rdma_encode_read_list(struct svc_rdma_send_ctxt *sctxt) 3496fd5034dSChuck Lever { 3506fd5034dSChuck Lever /* RPC-over-RDMA version 1 replies never have a Read list. */ 3516fd5034dSChuck Lever return xdr_stream_encode_item_absent(&sctxt->sc_stream); 3526fd5034dSChuck Lever } 3536fd5034dSChuck Lever 3546fd5034dSChuck Lever /** 3556fd5034dSChuck Lever * svc_rdma_encode_write_segment - Encode one Write segment 3566fd5034dSChuck Lever * @src: matching Write chunk in the RPC Call header 3576fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 3586fd5034dSChuck Lever * @remaining: remaining bytes of the payload left in the Write chunk 3596fd5034dSChuck Lever * 3606fd5034dSChuck Lever * Return values: 3616fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 3626fd5034dSChuck Lever * that was consumed by the Write segment 3636fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 3646fd5034dSChuck Lever */ 3656fd5034dSChuck Lever static ssize_t svc_rdma_encode_write_segment(__be32 *src, 3666fd5034dSChuck Lever struct svc_rdma_send_ctxt *sctxt, 3676fd5034dSChuck Lever unsigned int *remaining) 3686fd5034dSChuck Lever { 3696fd5034dSChuck Lever __be32 *p; 3706fd5034dSChuck Lever const size_t len = rpcrdma_segment_maxsz * sizeof(*p); 3716fd5034dSChuck Lever u32 handle, length; 3726fd5034dSChuck Lever u64 offset; 3736fd5034dSChuck Lever 3746fd5034dSChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, len); 3756fd5034dSChuck Lever if (!p) 3766fd5034dSChuck Lever return -EMSGSIZE; 3776fd5034dSChuck Lever 3786fd5034dSChuck Lever handle = be32_to_cpup(src++); 3796fd5034dSChuck Lever length = be32_to_cpup(src++); 3806fd5034dSChuck Lever xdr_decode_hyper(src, &offset); 3816fd5034dSChuck Lever 3826fd5034dSChuck Lever *p++ = cpu_to_be32(handle); 3836fd5034dSChuck Lever if (*remaining < length) { 3846fd5034dSChuck Lever /* segment only partly filled */ 3856fd5034dSChuck Lever length = *remaining; 3866fd5034dSChuck Lever *remaining = 0; 3876fd5034dSChuck Lever } else { 3886fd5034dSChuck Lever /* entire segment was consumed */ 3896fd5034dSChuck Lever *remaining -= length; 3906fd5034dSChuck Lever } 3916fd5034dSChuck Lever *p++ = cpu_to_be32(length); 3926fd5034dSChuck Lever xdr_encode_hyper(p, offset); 3936fd5034dSChuck Lever 3946fd5034dSChuck Lever trace_svcrdma_encode_wseg(handle, length, offset); 3956fd5034dSChuck Lever return len; 3966fd5034dSChuck Lever } 3976fd5034dSChuck Lever 3986fd5034dSChuck Lever /** 3996fd5034dSChuck Lever * svc_rdma_encode_write_chunk - Encode one Write chunk 4006fd5034dSChuck Lever * @src: matching Write chunk in the RPC Call header 4016fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 4026fd5034dSChuck Lever * @remaining: size in bytes of the payload in the Write chunk 4036fd5034dSChuck Lever * 4046fd5034dSChuck Lever * Copy a Write chunk from the Call transport header to the 4056fd5034dSChuck Lever * Reply transport header. Update each segment's length field 4066fd5034dSChuck Lever * to reflect the number of bytes written in that segment. 4076fd5034dSChuck Lever * 4086fd5034dSChuck Lever * Return values: 4096fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 4106fd5034dSChuck Lever * that was consumed by the Write chunk 4116fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 4126fd5034dSChuck Lever */ 4136fd5034dSChuck Lever static ssize_t svc_rdma_encode_write_chunk(__be32 *src, 4146fd5034dSChuck Lever struct svc_rdma_send_ctxt *sctxt, 4159a6a180bSChuck Lever unsigned int remaining) 4169a6a180bSChuck Lever { 4179a6a180bSChuck Lever unsigned int i, nsegs; 4186fd5034dSChuck Lever ssize_t len, ret; 4199a6a180bSChuck Lever 4206fd5034dSChuck Lever len = 0; 4216fd5034dSChuck Lever trace_svcrdma_encode_write_chunk(remaining); 4229a6a180bSChuck Lever 4236fd5034dSChuck Lever src++; 4246fd5034dSChuck Lever ret = xdr_stream_encode_item_present(&sctxt->sc_stream); 4256fd5034dSChuck Lever if (ret < 0) 4266fd5034dSChuck Lever return -EMSGSIZE; 4276fd5034dSChuck Lever len += ret; 4286fd5034dSChuck Lever 4296fd5034dSChuck Lever nsegs = be32_to_cpup(src++); 4306fd5034dSChuck Lever ret = xdr_stream_encode_u32(&sctxt->sc_stream, nsegs); 4316fd5034dSChuck Lever if (ret < 0) 4326fd5034dSChuck Lever return -EMSGSIZE; 4336fd5034dSChuck Lever len += ret; 4349a6a180bSChuck Lever 4359a6a180bSChuck Lever for (i = nsegs; i; i--) { 4366fd5034dSChuck Lever ret = svc_rdma_encode_write_segment(src, sctxt, &remaining); 4376fd5034dSChuck Lever if (ret < 0) 4386fd5034dSChuck Lever return -EMSGSIZE; 4396fd5034dSChuck Lever src += rpcrdma_segment_maxsz; 4406fd5034dSChuck Lever len += ret; 4419a6a180bSChuck Lever } 4429a6a180bSChuck Lever 4436fd5034dSChuck Lever return len; 4449a6a180bSChuck Lever } 4459a6a180bSChuck Lever 4466fd5034dSChuck Lever /** 4476fd5034dSChuck Lever * svc_rdma_encode_write_list - Encode RPC Reply's Write chunk list 4486fd5034dSChuck Lever * @rctxt: Reply context with information about the RPC Call 4496fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 4506fd5034dSChuck Lever * @length: size in bytes of the payload in the first Write chunk 4516fd5034dSChuck Lever * 4526fd5034dSChuck Lever * The client provides a Write chunk list in the Call message. Fill 4536fd5034dSChuck Lever * in the segments in the first Write chunk in the Reply's transport 4549a6a180bSChuck Lever * header with the number of bytes consumed in each segment. 4559a6a180bSChuck Lever * Remaining chunks are returned unused. 4569a6a180bSChuck Lever * 4579a6a180bSChuck Lever * Assumptions: 4589a6a180bSChuck Lever * - Client has provided only one Write chunk 4596fd5034dSChuck Lever * 4606fd5034dSChuck Lever * Return values: 4616fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 4626fd5034dSChuck Lever * that was consumed by the Reply's Write list 4636fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 4649a6a180bSChuck Lever */ 4656fd5034dSChuck Lever static ssize_t 4666fd5034dSChuck Lever svc_rdma_encode_write_list(const struct svc_rdma_recv_ctxt *rctxt, 4676fd5034dSChuck Lever struct svc_rdma_send_ctxt *sctxt, 4686fd5034dSChuck Lever unsigned int length) 4699a6a180bSChuck Lever { 4706fd5034dSChuck Lever ssize_t len, ret; 4719a6a180bSChuck Lever 4726fd5034dSChuck Lever ret = svc_rdma_encode_write_chunk(rctxt->rc_write_list, sctxt, length); 4736fd5034dSChuck Lever if (ret < 0) 4746fd5034dSChuck Lever return ret; 4756fd5034dSChuck Lever len = ret; 4769a6a180bSChuck Lever 4776fd5034dSChuck Lever /* Terminate the Write list */ 4786fd5034dSChuck Lever ret = xdr_stream_encode_item_absent(&sctxt->sc_stream); 4796fd5034dSChuck Lever if (ret < 0) 4806fd5034dSChuck Lever return ret; 4816fd5034dSChuck Lever 4826fd5034dSChuck Lever return len + ret; 4839a6a180bSChuck Lever } 4849a6a180bSChuck Lever 4856fd5034dSChuck Lever /** 4866fd5034dSChuck Lever * svc_rdma_encode_reply_chunk - Encode RPC Reply's Reply chunk 4876fd5034dSChuck Lever * @rctxt: Reply context with information about the RPC Call 4886fd5034dSChuck Lever * @sctxt: Send context for the RPC Reply 4896fd5034dSChuck Lever * @length: size in bytes of the payload in the Reply chunk 4909a6a180bSChuck Lever * 4919a6a180bSChuck Lever * Assumptions: 4926fd5034dSChuck Lever * - Reply can always fit in the client-provided Reply chunk 4936fd5034dSChuck Lever * 4946fd5034dSChuck Lever * Return values: 4956fd5034dSChuck Lever * On success, returns length in bytes of the Reply XDR buffer 4966fd5034dSChuck Lever * that was consumed by the Reply's Reply chunk 4976fd5034dSChuck Lever * %-EMSGSIZE on XDR buffer overflow 4989a6a180bSChuck Lever */ 4996fd5034dSChuck Lever static ssize_t 5006fd5034dSChuck Lever svc_rdma_encode_reply_chunk(const struct svc_rdma_recv_ctxt *rctxt, 5016fd5034dSChuck Lever struct svc_rdma_send_ctxt *sctxt, 5026fd5034dSChuck Lever unsigned int length) 5039a6a180bSChuck Lever { 5046fd5034dSChuck Lever return svc_rdma_encode_write_chunk(rctxt->rc_reply_chunk, sctxt, 5056fd5034dSChuck Lever length); 5069a6a180bSChuck Lever } 5079a6a180bSChuck Lever 5086e6092caSChuck Lever static int svc_rdma_dma_map_page(struct svcxprt_rdma *rdma, 5094201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt, 5106e6092caSChuck Lever struct page *page, 511f016f305SChuck Lever unsigned long offset, 5126e6092caSChuck Lever unsigned int len) 5136e6092caSChuck Lever { 5146e6092caSChuck Lever struct ib_device *dev = rdma->sc_cm_id->device; 5156e6092caSChuck Lever dma_addr_t dma_addr; 5166e6092caSChuck Lever 5176e6092caSChuck Lever dma_addr = ib_dma_map_page(dev, page, offset, len, DMA_TO_DEVICE); 518832b2cb9SChuck Lever trace_svcrdma_dma_map_page(rdma, dma_addr, len); 5196e6092caSChuck Lever if (ib_dma_mapping_error(dev, dma_addr)) 52091a08eaeSChuck Lever goto out_maperr; 5216e6092caSChuck Lever 52225fd86ecSChuck Lever ctxt->sc_sges[ctxt->sc_cur_sge_no].addr = dma_addr; 52325fd86ecSChuck Lever ctxt->sc_sges[ctxt->sc_cur_sge_no].length = len; 5244201c746SChuck Lever ctxt->sc_send_wr.num_sge++; 5256e6092caSChuck Lever return 0; 52691a08eaeSChuck Lever 52791a08eaeSChuck Lever out_maperr: 52891a08eaeSChuck Lever return -EIO; 5296e6092caSChuck Lever } 5306e6092caSChuck Lever 531f016f305SChuck Lever /* ib_dma_map_page() is used here because svc_rdma_dma_unmap() 532f016f305SChuck Lever * handles DMA-unmap and it uses ib_dma_unmap_page() exclusively. 533f016f305SChuck Lever */ 534f016f305SChuck Lever static int svc_rdma_dma_map_buf(struct svcxprt_rdma *rdma, 5354201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt, 536f016f305SChuck Lever unsigned char *base, 537f016f305SChuck Lever unsigned int len) 538f016f305SChuck Lever { 53925fd86ecSChuck Lever return svc_rdma_dma_map_page(rdma, ctxt, virt_to_page(base), 540f016f305SChuck Lever offset_in_page(base), len); 541f016f305SChuck Lever } 542f016f305SChuck Lever 5436e6092caSChuck Lever /** 5444554755eSChuck Lever * svc_rdma_pull_up_needed - Determine whether to use pull-up 5454554755eSChuck Lever * @rdma: controlling transport 5460dabe948SChuck Lever * @sctxt: send_ctxt for the Send WR 5474554755eSChuck Lever * @rctxt: Write and Reply chunks provided by client 5484554755eSChuck Lever * @xdr: xdr_buf containing RPC message to transmit 5494554755eSChuck Lever * 5504554755eSChuck Lever * Returns: 5514554755eSChuck Lever * %true if pull-up must be used 5524554755eSChuck Lever * %false otherwise 553e248aa7bSChuck Lever */ 554e248aa7bSChuck Lever static bool svc_rdma_pull_up_needed(struct svcxprt_rdma *rdma, 5550dabe948SChuck Lever struct svc_rdma_send_ctxt *sctxt, 5564554755eSChuck Lever const struct svc_rdma_recv_ctxt *rctxt, 5574554755eSChuck Lever struct xdr_buf *xdr) 558e248aa7bSChuck Lever { 559e248aa7bSChuck Lever int elements; 560e248aa7bSChuck Lever 5610dabe948SChuck Lever /* For small messages, copying bytes is cheaper than DMA mapping. 5620dabe948SChuck Lever */ 5630dabe948SChuck Lever if (sctxt->sc_hdrbuf.len + xdr->len < RPCRDMA_PULLUP_THRESH) 5640dabe948SChuck Lever return true; 5650dabe948SChuck Lever 5660dabe948SChuck Lever /* Check whether the xdr_buf has more elements than can 5670dabe948SChuck Lever * fit in a single RDMA Send. 5680dabe948SChuck Lever */ 569e248aa7bSChuck Lever /* xdr->head */ 570e248aa7bSChuck Lever elements = 1; 571e248aa7bSChuck Lever 572e248aa7bSChuck Lever /* xdr->pages */ 5734554755eSChuck Lever if (!rctxt || !rctxt->rc_write_list) { 574e248aa7bSChuck Lever unsigned int remaining; 575e248aa7bSChuck Lever unsigned long pageoff; 576e248aa7bSChuck Lever 577e248aa7bSChuck Lever pageoff = xdr->page_base & ~PAGE_MASK; 578e248aa7bSChuck Lever remaining = xdr->page_len; 579e248aa7bSChuck Lever while (remaining) { 580e248aa7bSChuck Lever ++elements; 581e248aa7bSChuck Lever remaining -= min_t(u32, PAGE_SIZE - pageoff, 582e248aa7bSChuck Lever remaining); 583e248aa7bSChuck Lever pageoff = 0; 584e248aa7bSChuck Lever } 585e248aa7bSChuck Lever } 586e248aa7bSChuck Lever 587e248aa7bSChuck Lever /* xdr->tail */ 588e248aa7bSChuck Lever if (xdr->tail[0].iov_len) 589e248aa7bSChuck Lever ++elements; 590e248aa7bSChuck Lever 591e248aa7bSChuck Lever /* assume 1 SGE is needed for the transport header */ 592e248aa7bSChuck Lever return elements >= rdma->sc_max_send_sges; 593e248aa7bSChuck Lever } 594e248aa7bSChuck Lever 5954554755eSChuck Lever /** 5964554755eSChuck Lever * svc_rdma_pull_up_reply_msg - Copy Reply into a single buffer 5974554755eSChuck Lever * @rdma: controlling transport 5984554755eSChuck Lever * @sctxt: send_ctxt for the Send WR; xprt hdr is already prepared 5994554755eSChuck Lever * @rctxt: Write and Reply chunks provided by client 6004554755eSChuck Lever * @xdr: prepared xdr_buf containing RPC message 6014554755eSChuck Lever * 6024554755eSChuck Lever * The device is not capable of sending the reply directly. 6034554755eSChuck Lever * Assemble the elements of @xdr into the transport header buffer. 6044554755eSChuck Lever * 6054554755eSChuck Lever * Returns zero on success, or a negative errno on failure. 606e248aa7bSChuck Lever */ 607e248aa7bSChuck Lever static int svc_rdma_pull_up_reply_msg(struct svcxprt_rdma *rdma, 6084554755eSChuck Lever struct svc_rdma_send_ctxt *sctxt, 6094554755eSChuck Lever const struct svc_rdma_recv_ctxt *rctxt, 6104554755eSChuck Lever const struct xdr_buf *xdr) 611e248aa7bSChuck Lever { 612e248aa7bSChuck Lever unsigned char *dst, *tailbase; 613e248aa7bSChuck Lever unsigned int taillen; 614e248aa7bSChuck Lever 615aee4b74aSChuck Lever dst = sctxt->sc_xprt_buf + sctxt->sc_hdrbuf.len; 616e248aa7bSChuck Lever memcpy(dst, xdr->head[0].iov_base, xdr->head[0].iov_len); 617e248aa7bSChuck Lever dst += xdr->head[0].iov_len; 618e248aa7bSChuck Lever 619e248aa7bSChuck Lever tailbase = xdr->tail[0].iov_base; 620e248aa7bSChuck Lever taillen = xdr->tail[0].iov_len; 6214554755eSChuck Lever if (rctxt && rctxt->rc_write_list) { 622e248aa7bSChuck Lever u32 xdrpad; 623e248aa7bSChuck Lever 62496f194b7SChuck Lever xdrpad = xdr_pad_size(xdr->page_len); 625e248aa7bSChuck Lever if (taillen && xdrpad) { 626e248aa7bSChuck Lever tailbase += xdrpad; 627e248aa7bSChuck Lever taillen -= xdrpad; 628e248aa7bSChuck Lever } 629e248aa7bSChuck Lever } else { 630e248aa7bSChuck Lever unsigned int len, remaining; 631e248aa7bSChuck Lever unsigned long pageoff; 632e248aa7bSChuck Lever struct page **ppages; 633e248aa7bSChuck Lever 634e248aa7bSChuck Lever ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT); 635e248aa7bSChuck Lever pageoff = xdr->page_base & ~PAGE_MASK; 636e248aa7bSChuck Lever remaining = xdr->page_len; 637e248aa7bSChuck Lever while (remaining) { 638e248aa7bSChuck Lever len = min_t(u32, PAGE_SIZE - pageoff, remaining); 639e248aa7bSChuck Lever 640e248aa7bSChuck Lever memcpy(dst, page_address(*ppages), len); 641e248aa7bSChuck Lever remaining -= len; 642e248aa7bSChuck Lever dst += len; 643e248aa7bSChuck Lever pageoff = 0; 644e248aa7bSChuck Lever } 645e248aa7bSChuck Lever } 646e248aa7bSChuck Lever 647e248aa7bSChuck Lever if (taillen) 648e248aa7bSChuck Lever memcpy(dst, tailbase, taillen); 649e248aa7bSChuck Lever 6504554755eSChuck Lever sctxt->sc_sges[0].length += xdr->len; 6510dabe948SChuck Lever trace_svcrdma_send_pullup(sctxt->sc_sges[0].length); 652e248aa7bSChuck Lever return 0; 653e248aa7bSChuck Lever } 654e248aa7bSChuck Lever 6554554755eSChuck Lever /* svc_rdma_map_reply_msg - DMA map the buffer holding RPC message 65699722fe4SChuck Lever * @rdma: controlling transport 6574554755eSChuck Lever * @sctxt: send_ctxt for the Send WR 6584554755eSChuck Lever * @rctxt: Write and Reply chunks provided by client 65999722fe4SChuck Lever * @xdr: prepared xdr_buf containing RPC message 66099722fe4SChuck Lever * 66199722fe4SChuck Lever * Load the xdr_buf into the ctxt's sge array, and DMA map each 662aee4b74aSChuck Lever * element as it is added. The Send WR's num_sge field is set. 6639a6a180bSChuck Lever * 66423262790SChuck Lever * Returns zero on success, or a negative errno on failure. 665c06b540aSTom Tucker */ 66699722fe4SChuck Lever int svc_rdma_map_reply_msg(struct svcxprt_rdma *rdma, 6674554755eSChuck Lever struct svc_rdma_send_ctxt *sctxt, 6684554755eSChuck Lever const struct svc_rdma_recv_ctxt *rctxt, 6694554755eSChuck Lever struct xdr_buf *xdr) 670c06b540aSTom Tucker { 67125fd86ecSChuck Lever unsigned int len, remaining; 672f016f305SChuck Lever unsigned long page_off; 6739a6a180bSChuck Lever struct page **ppages; 6749a6a180bSChuck Lever unsigned char *base; 6759a6a180bSChuck Lever u32 xdr_pad; 676c06b540aSTom Tucker int ret; 677c06b540aSTom Tucker 678aee4b74aSChuck Lever /* Set up the (persistently-mapped) transport header SGE. */ 679aee4b74aSChuck Lever sctxt->sc_send_wr.num_sge = 1; 680aee4b74aSChuck Lever sctxt->sc_sges[0].length = sctxt->sc_hdrbuf.len; 681aee4b74aSChuck Lever 682aee4b74aSChuck Lever /* If there is a Reply chunk, nothing follows the transport 683aee4b74aSChuck Lever * header, and we're done here. 684aee4b74aSChuck Lever */ 685aee4b74aSChuck Lever if (rctxt && rctxt->rc_reply_chunk) 686aee4b74aSChuck Lever return 0; 687aee4b74aSChuck Lever 688aee4b74aSChuck Lever /* For pull-up, svc_rdma_send() will sync the transport header. 689aee4b74aSChuck Lever * No additional DMA mapping is necessary. 690aee4b74aSChuck Lever */ 6910dabe948SChuck Lever if (svc_rdma_pull_up_needed(rdma, sctxt, rctxt, xdr)) 6924554755eSChuck Lever return svc_rdma_pull_up_reply_msg(rdma, sctxt, rctxt, xdr); 693e248aa7bSChuck Lever 6944554755eSChuck Lever ++sctxt->sc_cur_sge_no; 6954554755eSChuck Lever ret = svc_rdma_dma_map_buf(rdma, sctxt, 6969a6a180bSChuck Lever xdr->head[0].iov_base, 6979a6a180bSChuck Lever xdr->head[0].iov_len); 6989a6a180bSChuck Lever if (ret < 0) 6999a6a180bSChuck Lever return ret; 700c06b540aSTom Tucker 7019a6a180bSChuck Lever /* If a Write chunk is present, the xdr_buf's page list 7029a6a180bSChuck Lever * is not included inline. However the Upper Layer may 7039a6a180bSChuck Lever * have added XDR padding in the tail buffer, and that 7049a6a180bSChuck Lever * should not be included inline. 7059a6a180bSChuck Lever */ 7064554755eSChuck Lever if (rctxt && rctxt->rc_write_list) { 7079a6a180bSChuck Lever base = xdr->tail[0].iov_base; 7089a6a180bSChuck Lever len = xdr->tail[0].iov_len; 70996f194b7SChuck Lever xdr_pad = xdr_pad_size(xdr->page_len); 710c06b540aSTom Tucker 7119a6a180bSChuck Lever if (len && xdr_pad) { 7129a6a180bSChuck Lever base += xdr_pad; 7139a6a180bSChuck Lever len -= xdr_pad; 714c06b540aSTom Tucker } 715c06b540aSTom Tucker 7169a6a180bSChuck Lever goto tail; 717c06b540aSTom Tucker } 7189a6a180bSChuck Lever 7199a6a180bSChuck Lever ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT); 7209a6a180bSChuck Lever page_off = xdr->page_base & ~PAGE_MASK; 7219a6a180bSChuck Lever remaining = xdr->page_len; 7229a6a180bSChuck Lever while (remaining) { 7239a6a180bSChuck Lever len = min_t(u32, PAGE_SIZE - page_off, remaining); 7249a6a180bSChuck Lever 7254554755eSChuck Lever ++sctxt->sc_cur_sge_no; 7264554755eSChuck Lever ret = svc_rdma_dma_map_page(rdma, sctxt, *ppages++, 72725fd86ecSChuck Lever page_off, len); 7289a6a180bSChuck Lever if (ret < 0) 7299a6a180bSChuck Lever return ret; 7309a6a180bSChuck Lever 7319a6a180bSChuck Lever remaining -= len; 7329a6a180bSChuck Lever page_off = 0; 733c06b540aSTom Tucker } 734c06b540aSTom Tucker 7359a6a180bSChuck Lever base = xdr->tail[0].iov_base; 7369a6a180bSChuck Lever len = xdr->tail[0].iov_len; 7379a6a180bSChuck Lever tail: 7389a6a180bSChuck Lever if (len) { 7394554755eSChuck Lever ++sctxt->sc_cur_sge_no; 7404554755eSChuck Lever ret = svc_rdma_dma_map_buf(rdma, sctxt, base, len); 7419a6a180bSChuck Lever if (ret < 0) 7429a6a180bSChuck Lever return ret; 7439a6a180bSChuck Lever } 74408ae4e7fSChuck Lever 74523262790SChuck Lever return 0; 746c06b540aSTom Tucker } 747c06b540aSTom Tucker 748c55ab070SChuck Lever /* The svc_rqst and all resources it owns are released as soon as 749c55ab070SChuck Lever * svc_rdma_sendto returns. Transfer pages under I/O to the ctxt 750c55ab070SChuck Lever * so they are released by the Send completion handler. 751c55ab070SChuck Lever */ 752c55ab070SChuck Lever static void svc_rdma_save_io_pages(struct svc_rqst *rqstp, 7534201c746SChuck Lever struct svc_rdma_send_ctxt *ctxt) 754c55ab070SChuck Lever { 755c55ab070SChuck Lever int i, pages = rqstp->rq_next_page - rqstp->rq_respages; 756c55ab070SChuck Lever 7574201c746SChuck Lever ctxt->sc_page_count += pages; 758c55ab070SChuck Lever for (i = 0; i < pages; i++) { 75999722fe4SChuck Lever ctxt->sc_pages[i] = rqstp->rq_respages[i]; 760c55ab070SChuck Lever rqstp->rq_respages[i] = NULL; 761c55ab070SChuck Lever } 762a53d5cb0SChuck Lever 763a53d5cb0SChuck Lever /* Prevent svc_xprt_release from releasing pages in rq_pages */ 764a53d5cb0SChuck Lever rqstp->rq_next_page = rqstp->rq_respages; 765c55ab070SChuck Lever } 766c55ab070SChuck Lever 7679a6a180bSChuck Lever /* Prepare the portion of the RPC Reply that will be transmitted 7689a6a180bSChuck Lever * via RDMA Send. The RPC-over-RDMA transport header is prepared 7694201c746SChuck Lever * in sc_sges[0], and the RPC xdr_buf is prepared in following sges. 7709a6a180bSChuck Lever * 7719a6a180bSChuck Lever * Depending on whether a Write list or Reply chunk is present, 7729a6a180bSChuck Lever * the server may send all, a portion of, or none of the xdr_buf. 7734201c746SChuck Lever * In the latter case, only the transport header (sc_sges[0]) is 7749a6a180bSChuck Lever * transmitted. 7759a6a180bSChuck Lever * 7769a6a180bSChuck Lever * RDMA Send is the last step of transmitting an RPC reply. Pages 7779a6a180bSChuck Lever * involved in the earlier RDMA Writes are here transferred out 77897bce634SChuck Lever * of the rqstp and into the sctxt's page array. These pages are 7799a6a180bSChuck Lever * DMA unmapped by each Write completion, but the subsequent Send 7809a6a180bSChuck Lever * completion finally releases these pages. 7819a6a180bSChuck Lever * 7829a6a180bSChuck Lever * Assumptions: 7839a6a180bSChuck Lever * - The Reply's transport header will never be larger than a page. 784c06b540aSTom Tucker */ 7859a6a180bSChuck Lever static int svc_rdma_send_reply_msg(struct svcxprt_rdma *rdma, 78697bce634SChuck Lever struct svc_rdma_send_ctxt *sctxt, 787db9602e4SChuck Lever const struct svc_rdma_recv_ctxt *rctxt, 788db9602e4SChuck Lever struct svc_rqst *rqstp) 789c06b540aSTom Tucker { 7909a6a180bSChuck Lever int ret; 7910e7f011aSTom Tucker 792aee4b74aSChuck Lever ret = svc_rdma_map_reply_msg(rdma, sctxt, rctxt, &rqstp->rq_res); 7939a6a180bSChuck Lever if (ret < 0) 79499722fe4SChuck Lever return ret; 795c06b540aSTom Tucker 79697bce634SChuck Lever svc_rdma_save_io_pages(rqstp, sctxt); 7970bf48289SSteve Wise 79897bce634SChuck Lever if (rctxt->rc_inv_rkey) { 79997bce634SChuck Lever sctxt->sc_send_wr.opcode = IB_WR_SEND_WITH_INV; 80097bce634SChuck Lever sctxt->sc_send_wr.ex.invalidate_rkey = rctxt->rc_inv_rkey; 80197bce634SChuck Lever } else { 80297bce634SChuck Lever sctxt->sc_send_wr.opcode = IB_WR_SEND; 803986b7889SChuck Lever } 80497bce634SChuck Lever return svc_rdma_send(rdma, &sctxt->sc_send_wr); 805c06b540aSTom Tucker } 806c06b540aSTom Tucker 8074757d90bSChuck Lever /* Given the client-provided Write and Reply chunks, the server was not 8084757d90bSChuck Lever * able to form a complete reply. Return an RDMA_ERROR message so the 8096e9fab70SChuck Lever * client can retire this RPC transaction. 8104757d90bSChuck Lever * 8114757d90bSChuck Lever * Remote Invalidation is skipped for simplicity. 8124757d90bSChuck Lever */ 8134757d90bSChuck Lever static int svc_rdma_send_error_msg(struct svcxprt_rdma *rdma, 814d1f6e236SChuck Lever struct svc_rdma_send_ctxt *sctxt, 8154f200bd8SChuck Lever struct svc_rdma_recv_ctxt *rctxt, 8164f200bd8SChuck Lever int status) 8174757d90bSChuck Lever { 8186fd5034dSChuck Lever __be32 *rdma_argp = rctxt->rc_recv_buf; 8194757d90bSChuck Lever __be32 *p; 8204757d90bSChuck Lever 821d1f6e236SChuck Lever rpcrdma_set_xdrlen(&sctxt->sc_hdrbuf, 0); 822d1f6e236SChuck Lever xdr_init_encode(&sctxt->sc_stream, &sctxt->sc_hdrbuf, 823d1f6e236SChuck Lever sctxt->sc_xprt_buf, NULL); 8246fd5034dSChuck Lever 8254f200bd8SChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, 8264f200bd8SChuck Lever rpcrdma_fixed_maxsz * sizeof(*p)); 8276fd5034dSChuck Lever if (!p) 8286fd5034dSChuck Lever return -ENOMSG; 8296fd5034dSChuck Lever 8306fd5034dSChuck Lever *p++ = *rdma_argp; 8316fd5034dSChuck Lever *p++ = *(rdma_argp + 1); 8326fd5034dSChuck Lever *p++ = rdma->sc_fc_credits; 8334f200bd8SChuck Lever *p = rdma_error; 8344f200bd8SChuck Lever 8354f200bd8SChuck Lever switch (status) { 8364f200bd8SChuck Lever case -EPROTONOSUPPORT: 8374f200bd8SChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, 3 * sizeof(*p)); 8384f200bd8SChuck Lever if (!p) 8394f200bd8SChuck Lever return -ENOMSG; 8404f200bd8SChuck Lever 8414f200bd8SChuck Lever *p++ = err_vers; 8424f200bd8SChuck Lever *p++ = rpcrdma_version; 8434f200bd8SChuck Lever *p = rpcrdma_version; 8444f200bd8SChuck Lever trace_svcrdma_err_vers(*rdma_argp); 8454f200bd8SChuck Lever break; 8464f200bd8SChuck Lever default: 8474f200bd8SChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, sizeof(*p)); 8484f200bd8SChuck Lever if (!p) 8494f200bd8SChuck Lever return -ENOMSG; 8504f200bd8SChuck Lever 8514757d90bSChuck Lever *p = err_chunk; 8526fd5034dSChuck Lever trace_svcrdma_err_chunk(*rdma_argp); 8534f200bd8SChuck Lever } 8546fd5034dSChuck Lever 855d1f6e236SChuck Lever sctxt->sc_send_wr.num_sge = 1; 856d1f6e236SChuck Lever sctxt->sc_send_wr.opcode = IB_WR_SEND; 857d1f6e236SChuck Lever sctxt->sc_sges[0].length = sctxt->sc_hdrbuf.len; 858d1f6e236SChuck Lever return svc_rdma_send(rdma, &sctxt->sc_send_wr); 85999722fe4SChuck Lever } 86099722fe4SChuck Lever 8619a6a180bSChuck Lever /** 8629a6a180bSChuck Lever * svc_rdma_sendto - Transmit an RPC reply 8639a6a180bSChuck Lever * @rqstp: processed RPC request, reply XDR already in ::rq_res 8649a6a180bSChuck Lever * 8659a6a180bSChuck Lever * Any resources still associated with @rqstp are released upon return. 8669a6a180bSChuck Lever * If no reply message was possible, the connection is closed. 8679a6a180bSChuck Lever * 8689a6a180bSChuck Lever * Returns: 8699a6a180bSChuck Lever * %0 if an RPC reply has been successfully posted, 8709a6a180bSChuck Lever * %-ENOMEM if a resource shortage occurred (connection is lost), 8719a6a180bSChuck Lever * %-ENOTCONN if posting failed (connection is lost). 8729a6a180bSChuck Lever */ 873c06b540aSTom Tucker int svc_rdma_sendto(struct svc_rqst *rqstp) 874c06b540aSTom Tucker { 875c06b540aSTom Tucker struct svc_xprt *xprt = rqstp->rq_xprt; 876c06b540aSTom Tucker struct svcxprt_rdma *rdma = 877c06b540aSTom Tucker container_of(xprt, struct svcxprt_rdma, sc_xprt); 8783a88092eSChuck Lever struct svc_rdma_recv_ctxt *rctxt = rqstp->rq_xprt_ctxt; 8792fe8c446SChuck Lever __be32 *rdma_argp = rctxt->rc_recv_buf; 8802fe8c446SChuck Lever __be32 *wr_lst = rctxt->rc_write_list; 8812fe8c446SChuck Lever __be32 *rp_ch = rctxt->rc_reply_chunk; 8829a6a180bSChuck Lever struct xdr_buf *xdr = &rqstp->rq_res; 88399722fe4SChuck Lever struct svc_rdma_send_ctxt *sctxt; 8846fd5034dSChuck Lever __be32 *p; 8859a6a180bSChuck Lever int ret; 886c06b540aSTom Tucker 887ca4faf54SChuck Lever ret = -ENOTCONN; 888ca4faf54SChuck Lever if (svc_xprt_is_dead(xprt)) 889ca4faf54SChuck Lever goto err0; 890ca4faf54SChuck Lever 89178da2b3cSChuck Lever ret = -ENOMEM; 89299722fe4SChuck Lever sctxt = svc_rdma_send_ctxt_get(rdma); 89399722fe4SChuck Lever if (!sctxt) 89478da2b3cSChuck Lever goto err0; 89598fc21d3SChuck Lever 8966fd5034dSChuck Lever p = xdr_reserve_space(&sctxt->sc_stream, 8976fd5034dSChuck Lever rpcrdma_fixed_maxsz * sizeof(*p)); 8986fd5034dSChuck Lever if (!p) 8996fd5034dSChuck Lever goto err0; 9009a6a180bSChuck Lever *p++ = *rdma_argp; 9019a6a180bSChuck Lever *p++ = *(rdma_argp + 1); 90298fc21d3SChuck Lever *p++ = rdma->sc_fc_credits; 9036fd5034dSChuck Lever *p = rp_ch ? rdma_nomsg : rdma_msg; 90498fc21d3SChuck Lever 9056fd5034dSChuck Lever if (svc_rdma_encode_read_list(sctxt) < 0) 9066fd5034dSChuck Lever goto err0; 9079a6a180bSChuck Lever if (wr_lst) { 9089a6a180bSChuck Lever /* XXX: Presume the client sent only one Write chunk */ 90941205539SChuck Lever unsigned long offset; 91041205539SChuck Lever unsigned int length; 91141205539SChuck Lever 91241205539SChuck Lever if (rctxt->rc_read_payload_length) { 91341205539SChuck Lever offset = rctxt->rc_read_payload_offset; 91441205539SChuck Lever length = rctxt->rc_read_payload_length; 91541205539SChuck Lever } else { 91641205539SChuck Lever offset = xdr->head[0].iov_len; 91741205539SChuck Lever length = xdr->page_len; 91841205539SChuck Lever } 91941205539SChuck Lever ret = svc_rdma_send_write_chunk(rdma, wr_lst, xdr, offset, 92041205539SChuck Lever length); 92108ae4e7fSChuck Lever if (ret < 0) 9224757d90bSChuck Lever goto err2; 9236fd5034dSChuck Lever if (svc_rdma_encode_write_list(rctxt, sctxt, length) < 0) 9246fd5034dSChuck Lever goto err0; 9256fd5034dSChuck Lever } else { 9266fd5034dSChuck Lever if (xdr_stream_encode_item_absent(&sctxt->sc_stream) < 0) 9276fd5034dSChuck Lever goto err0; 92808ae4e7fSChuck Lever } 9299a6a180bSChuck Lever if (rp_ch) { 9306fa5785eSChuck Lever ret = svc_rdma_send_reply_chunk(rdma, rctxt, &rqstp->rq_res); 93108ae4e7fSChuck Lever if (ret < 0) 9324757d90bSChuck Lever goto err2; 9336fd5034dSChuck Lever if (svc_rdma_encode_reply_chunk(rctxt, sctxt, ret) < 0) 9346fd5034dSChuck Lever goto err0; 9356fd5034dSChuck Lever } else { 9366fd5034dSChuck Lever if (xdr_stream_encode_item_absent(&sctxt->sc_stream) < 0) 9376fd5034dSChuck Lever goto err0; 93808ae4e7fSChuck Lever } 939c06b540aSTom Tucker 940db9602e4SChuck Lever ret = svc_rdma_send_reply_msg(rdma, sctxt, rctxt, rqstp); 9413e1eeb98SChuck Lever if (ret < 0) 94299722fe4SChuck Lever goto err1; 94323cf1ee1SChuck Lever return 0; 944afd566eaSTom Tucker 9454757d90bSChuck Lever err2: 946b20dae70SColin Ian King if (ret != -E2BIG && ret != -EINVAL) 9474757d90bSChuck Lever goto err1; 9484757d90bSChuck Lever 9496e9fab70SChuck Lever /* Send completion releases payload pages that were part 9506e9fab70SChuck Lever * of previously posted RDMA Writes. 9516e9fab70SChuck Lever */ 9526e9fab70SChuck Lever svc_rdma_save_io_pages(rqstp, sctxt); 9534f200bd8SChuck Lever ret = svc_rdma_send_error_msg(rdma, sctxt, rctxt, ret); 9544757d90bSChuck Lever if (ret < 0) 95599722fe4SChuck Lever goto err1; 95623cf1ee1SChuck Lever return 0; 9574757d90bSChuck Lever 958afd566eaSTom Tucker err1: 95999722fe4SChuck Lever svc_rdma_send_ctxt_put(rdma, sctxt); 960afd566eaSTom Tucker err0: 961bd2abef3SChuck Lever trace_svcrdma_send_failed(rqstp, ret); 9629a6a180bSChuck Lever set_bit(XPT_CLOSE, &xprt->xpt_flags); 96323cf1ee1SChuck Lever return -ENOTCONN; 964c06b540aSTom Tucker } 96541205539SChuck Lever 96641205539SChuck Lever /** 96741205539SChuck Lever * svc_rdma_read_payload - special processing for a READ payload 96841205539SChuck Lever * @rqstp: svc_rqst to operate on 96941205539SChuck Lever * @offset: payload's byte offset in @xdr 97041205539SChuck Lever * @length: size of payload, in bytes 97141205539SChuck Lever * 97241205539SChuck Lever * Returns zero on success. 97341205539SChuck Lever * 97441205539SChuck Lever * For the moment, just record the xdr_buf location of the READ 97541205539SChuck Lever * payload. svc_rdma_sendto will use that location later when 97641205539SChuck Lever * we actually send the payload. 97741205539SChuck Lever */ 97841205539SChuck Lever int svc_rdma_read_payload(struct svc_rqst *rqstp, unsigned int offset, 97941205539SChuck Lever unsigned int length) 98041205539SChuck Lever { 98141205539SChuck Lever struct svc_rdma_recv_ctxt *rctxt = rqstp->rq_xprt_ctxt; 98241205539SChuck Lever 98341205539SChuck Lever /* XXX: Just one READ payload slot for now, since our 98441205539SChuck Lever * transport implementation currently supports only one 98541205539SChuck Lever * Write chunk. 98641205539SChuck Lever */ 98741205539SChuck Lever rctxt->rc_read_payload_offset = offset; 98841205539SChuck Lever rctxt->rc_read_payload_length = length; 98941205539SChuck Lever 99041205539SChuck Lever return 0; 99141205539SChuck Lever } 992