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