xref: /openbmc/linux/net/sunrpc/xprtrdma/rpc_rdma.c (revision 11270e7c)
1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
2 /*
3  * Copyright (c) 2014-2020, Oracle and/or its affiliates.
4  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
5  *
6  * This software is available to you under a choice of one of two
7  * licenses.  You may choose to be licensed under the terms of the GNU
8  * General Public License (GPL) Version 2, available from the file
9  * COPYING in the main directory of this source tree, or the BSD-type
10  * license below:
11  *
12  * Redistribution and use in source and binary forms, with or without
13  * modification, are permitted provided that the following conditions
14  * are met:
15  *
16  *      Redistributions of source code must retain the above copyright
17  *      notice, this list of conditions and the following disclaimer.
18  *
19  *      Redistributions in binary form must reproduce the above
20  *      copyright notice, this list of conditions and the following
21  *      disclaimer in the documentation and/or other materials provided
22  *      with the distribution.
23  *
24  *      Neither the name of the Network Appliance, Inc. nor the names of
25  *      its contributors may be used to endorse or promote products
26  *      derived from this software without specific prior written
27  *      permission.
28  *
29  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
30  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
31  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
32  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
33  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
34  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
35  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
36  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
37  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
38  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
39  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
40  */
41 
42 /*
43  * rpc_rdma.c
44  *
45  * This file contains the guts of the RPC RDMA protocol, and
46  * does marshaling/unmarshaling, etc. It is also where interfacing
47  * to the Linux RPC framework lives.
48  */
49 
50 #include <linux/highmem.h>
51 
52 #include <linux/sunrpc/svc_rdma.h>
53 
54 #include "xprt_rdma.h"
55 #include <trace/events/rpcrdma.h>
56 
57 /* Returns size of largest RPC-over-RDMA header in a Call message
58  *
59  * The largest Call header contains a full-size Read list and a
60  * minimal Reply chunk.
61  */
rpcrdma_max_call_header_size(unsigned int maxsegs)62 static unsigned int rpcrdma_max_call_header_size(unsigned int maxsegs)
63 {
64 	unsigned int size;
65 
66 	/* Fixed header fields and list discriminators */
67 	size = RPCRDMA_HDRLEN_MIN;
68 
69 	/* Maximum Read list size */
70 	size += maxsegs * rpcrdma_readchunk_maxsz * sizeof(__be32);
71 
72 	/* Minimal Read chunk size */
73 	size += sizeof(__be32);	/* segment count */
74 	size += rpcrdma_segment_maxsz * sizeof(__be32);
75 	size += sizeof(__be32);	/* list discriminator */
76 
77 	return size;
78 }
79 
80 /* Returns size of largest RPC-over-RDMA header in a Reply message
81  *
82  * There is only one Write list or one Reply chunk per Reply
83  * message.  The larger list is the Write list.
84  */
rpcrdma_max_reply_header_size(unsigned int maxsegs)85 static unsigned int rpcrdma_max_reply_header_size(unsigned int maxsegs)
86 {
87 	unsigned int size;
88 
89 	/* Fixed header fields and list discriminators */
90 	size = RPCRDMA_HDRLEN_MIN;
91 
92 	/* Maximum Write list size */
93 	size += sizeof(__be32);		/* segment count */
94 	size += maxsegs * rpcrdma_segment_maxsz * sizeof(__be32);
95 	size += sizeof(__be32);	/* list discriminator */
96 
97 	return size;
98 }
99 
100 /**
101  * rpcrdma_set_max_header_sizes - Initialize inline payload sizes
102  * @ep: endpoint to initialize
103  *
104  * The max_inline fields contain the maximum size of an RPC message
105  * so the marshaling code doesn't have to repeat this calculation
106  * for every RPC.
107  */
rpcrdma_set_max_header_sizes(struct rpcrdma_ep * ep)108 void rpcrdma_set_max_header_sizes(struct rpcrdma_ep *ep)
109 {
110 	unsigned int maxsegs = ep->re_max_rdma_segs;
111 
112 	ep->re_max_inline_send =
113 		ep->re_inline_send - rpcrdma_max_call_header_size(maxsegs);
114 	ep->re_max_inline_recv =
115 		ep->re_inline_recv - rpcrdma_max_reply_header_size(maxsegs);
116 }
117 
118 /* The client can send a request inline as long as the RPCRDMA header
119  * plus the RPC call fit under the transport's inline limit. If the
120  * combined call message size exceeds that limit, the client must use
121  * a Read chunk for this operation.
122  *
123  * A Read chunk is also required if sending the RPC call inline would
124  * exceed this device's max_sge limit.
125  */
rpcrdma_args_inline(struct rpcrdma_xprt * r_xprt,struct rpc_rqst * rqst)126 static bool rpcrdma_args_inline(struct rpcrdma_xprt *r_xprt,
127 				struct rpc_rqst *rqst)
128 {
129 	struct xdr_buf *xdr = &rqst->rq_snd_buf;
130 	struct rpcrdma_ep *ep = r_xprt->rx_ep;
131 	unsigned int count, remaining, offset;
132 
133 	if (xdr->len > ep->re_max_inline_send)
134 		return false;
135 
136 	if (xdr->page_len) {
137 		remaining = xdr->page_len;
138 		offset = offset_in_page(xdr->page_base);
139 		count = RPCRDMA_MIN_SEND_SGES;
140 		while (remaining) {
141 			remaining -= min_t(unsigned int,
142 					   PAGE_SIZE - offset, remaining);
143 			offset = 0;
144 			if (++count > ep->re_attr.cap.max_send_sge)
145 				return false;
146 		}
147 	}
148 
149 	return true;
150 }
151 
152 /* The client can't know how large the actual reply will be. Thus it
153  * plans for the largest possible reply for that particular ULP
154  * operation. If the maximum combined reply message size exceeds that
155  * limit, the client must provide a write list or a reply chunk for
156  * this request.
157  */
rpcrdma_results_inline(struct rpcrdma_xprt * r_xprt,struct rpc_rqst * rqst)158 static bool rpcrdma_results_inline(struct rpcrdma_xprt *r_xprt,
159 				   struct rpc_rqst *rqst)
160 {
161 	return rqst->rq_rcv_buf.buflen <= r_xprt->rx_ep->re_max_inline_recv;
162 }
163 
164 /* The client is required to provide a Reply chunk if the maximum
165  * size of the non-payload part of the RPC Reply is larger than
166  * the inline threshold.
167  */
168 static bool
rpcrdma_nonpayload_inline(const struct rpcrdma_xprt * r_xprt,const struct rpc_rqst * rqst)169 rpcrdma_nonpayload_inline(const struct rpcrdma_xprt *r_xprt,
170 			  const struct rpc_rqst *rqst)
171 {
172 	const struct xdr_buf *buf = &rqst->rq_rcv_buf;
173 
174 	return (buf->head[0].iov_len + buf->tail[0].iov_len) <
175 		r_xprt->rx_ep->re_max_inline_recv;
176 }
177 
178 /* ACL likes to be lazy in allocating pages. For TCP, these
179  * pages can be allocated during receive processing. Not true
180  * for RDMA, which must always provision receive buffers
181  * up front.
182  */
183 static noinline int
rpcrdma_alloc_sparse_pages(struct xdr_buf * buf)184 rpcrdma_alloc_sparse_pages(struct xdr_buf *buf)
185 {
186 	struct page **ppages;
187 	int len;
188 
189 	len = buf->page_len;
190 	ppages = buf->pages + (buf->page_base >> PAGE_SHIFT);
191 	while (len > 0) {
192 		if (!*ppages)
193 			*ppages = alloc_page(GFP_NOWAIT | __GFP_NOWARN);
194 		if (!*ppages)
195 			return -ENOBUFS;
196 		ppages++;
197 		len -= PAGE_SIZE;
198 	}
199 
200 	return 0;
201 }
202 
203 /* Convert @vec to a single SGL element.
204  *
205  * Returns pointer to next available SGE, and bumps the total number
206  * of SGEs consumed.
207  */
208 static struct rpcrdma_mr_seg *
rpcrdma_convert_kvec(struct kvec * vec,struct rpcrdma_mr_seg * seg,unsigned int * n)209 rpcrdma_convert_kvec(struct kvec *vec, struct rpcrdma_mr_seg *seg,
210 		     unsigned int *n)
211 {
212 	seg->mr_page = virt_to_page(vec->iov_base);
213 	seg->mr_offset = offset_in_page(vec->iov_base);
214 	seg->mr_len = vec->iov_len;
215 	++seg;
216 	++(*n);
217 	return seg;
218 }
219 
220 /* Convert @xdrbuf into SGEs no larger than a page each. As they
221  * are registered, these SGEs are then coalesced into RDMA segments
222  * when the selected memreg mode supports it.
223  *
224  * Returns positive number of SGEs consumed, or a negative errno.
225  */
226 
227 static int
rpcrdma_convert_iovs(struct rpcrdma_xprt * r_xprt,struct xdr_buf * xdrbuf,unsigned int pos,enum rpcrdma_chunktype type,struct rpcrdma_mr_seg * seg)228 rpcrdma_convert_iovs(struct rpcrdma_xprt *r_xprt, struct xdr_buf *xdrbuf,
229 		     unsigned int pos, enum rpcrdma_chunktype type,
230 		     struct rpcrdma_mr_seg *seg)
231 {
232 	unsigned long page_base;
233 	unsigned int len, n;
234 	struct page **ppages;
235 
236 	n = 0;
237 	if (pos == 0)
238 		seg = rpcrdma_convert_kvec(&xdrbuf->head[0], seg, &n);
239 
240 	len = xdrbuf->page_len;
241 	ppages = xdrbuf->pages + (xdrbuf->page_base >> PAGE_SHIFT);
242 	page_base = offset_in_page(xdrbuf->page_base);
243 	while (len) {
244 		seg->mr_page = *ppages;
245 		seg->mr_offset = page_base;
246 		seg->mr_len = min_t(u32, PAGE_SIZE - page_base, len);
247 		len -= seg->mr_len;
248 		++ppages;
249 		++seg;
250 		++n;
251 		page_base = 0;
252 	}
253 
254 	if (type == rpcrdma_readch || type == rpcrdma_writech)
255 		goto out;
256 
257 	if (xdrbuf->tail[0].iov_len)
258 		rpcrdma_convert_kvec(&xdrbuf->tail[0], seg, &n);
259 
260 out:
261 	if (unlikely(n > RPCRDMA_MAX_SEGS))
262 		return -EIO;
263 	return n;
264 }
265 
266 static int
encode_rdma_segment(struct xdr_stream * xdr,struct rpcrdma_mr * mr)267 encode_rdma_segment(struct xdr_stream *xdr, struct rpcrdma_mr *mr)
268 {
269 	__be32 *p;
270 
271 	p = xdr_reserve_space(xdr, 4 * sizeof(*p));
272 	if (unlikely(!p))
273 		return -EMSGSIZE;
274 
275 	xdr_encode_rdma_segment(p, mr->mr_handle, mr->mr_length, mr->mr_offset);
276 	return 0;
277 }
278 
279 static int
encode_read_segment(struct xdr_stream * xdr,struct rpcrdma_mr * mr,u32 position)280 encode_read_segment(struct xdr_stream *xdr, struct rpcrdma_mr *mr,
281 		    u32 position)
282 {
283 	__be32 *p;
284 
285 	p = xdr_reserve_space(xdr, 6 * sizeof(*p));
286 	if (unlikely(!p))
287 		return -EMSGSIZE;
288 
289 	*p++ = xdr_one;			/* Item present */
290 	xdr_encode_read_segment(p, position, mr->mr_handle, mr->mr_length,
291 				mr->mr_offset);
292 	return 0;
293 }
294 
rpcrdma_mr_prepare(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct rpcrdma_mr_seg * seg,int nsegs,bool writing,struct rpcrdma_mr ** mr)295 static struct rpcrdma_mr_seg *rpcrdma_mr_prepare(struct rpcrdma_xprt *r_xprt,
296 						 struct rpcrdma_req *req,
297 						 struct rpcrdma_mr_seg *seg,
298 						 int nsegs, bool writing,
299 						 struct rpcrdma_mr **mr)
300 {
301 	*mr = rpcrdma_mr_pop(&req->rl_free_mrs);
302 	if (!*mr) {
303 		*mr = rpcrdma_mr_get(r_xprt);
304 		if (!*mr)
305 			goto out_getmr_err;
306 		(*mr)->mr_req = req;
307 	}
308 
309 	rpcrdma_mr_push(*mr, &req->rl_registered);
310 	return frwr_map(r_xprt, seg, nsegs, writing, req->rl_slot.rq_xid, *mr);
311 
312 out_getmr_err:
313 	trace_xprtrdma_nomrs_err(r_xprt, req);
314 	xprt_wait_for_buffer_space(&r_xprt->rx_xprt);
315 	rpcrdma_mrs_refresh(r_xprt);
316 	return ERR_PTR(-EAGAIN);
317 }
318 
319 /* Register and XDR encode the Read list. Supports encoding a list of read
320  * segments that belong to a single read chunk.
321  *
322  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
323  *
324  *  Read chunklist (a linked list):
325  *   N elements, position P (same P for all chunks of same arg!):
326  *    1 - PHLOO - 1 - PHLOO - ... - 1 - PHLOO - 0
327  *
328  * Returns zero on success, or a negative errno if a failure occurred.
329  * @xdr is advanced to the next position in the stream.
330  *
331  * Only a single @pos value is currently supported.
332  */
rpcrdma_encode_read_list(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct rpc_rqst * rqst,enum rpcrdma_chunktype rtype)333 static int rpcrdma_encode_read_list(struct rpcrdma_xprt *r_xprt,
334 				    struct rpcrdma_req *req,
335 				    struct rpc_rqst *rqst,
336 				    enum rpcrdma_chunktype rtype)
337 {
338 	struct xdr_stream *xdr = &req->rl_stream;
339 	struct rpcrdma_mr_seg *seg;
340 	struct rpcrdma_mr *mr;
341 	unsigned int pos;
342 	int nsegs;
343 
344 	if (rtype == rpcrdma_noch_pullup || rtype == rpcrdma_noch_mapped)
345 		goto done;
346 
347 	pos = rqst->rq_snd_buf.head[0].iov_len;
348 	if (rtype == rpcrdma_areadch)
349 		pos = 0;
350 	seg = req->rl_segments;
351 	nsegs = rpcrdma_convert_iovs(r_xprt, &rqst->rq_snd_buf, pos,
352 				     rtype, seg);
353 	if (nsegs < 0)
354 		return nsegs;
355 
356 	do {
357 		seg = rpcrdma_mr_prepare(r_xprt, req, seg, nsegs, false, &mr);
358 		if (IS_ERR(seg))
359 			return PTR_ERR(seg);
360 
361 		if (encode_read_segment(xdr, mr, pos) < 0)
362 			return -EMSGSIZE;
363 
364 		trace_xprtrdma_chunk_read(rqst->rq_task, pos, mr, nsegs);
365 		r_xprt->rx_stats.read_chunk_count++;
366 		nsegs -= mr->mr_nents;
367 	} while (nsegs);
368 
369 done:
370 	if (xdr_stream_encode_item_absent(xdr) < 0)
371 		return -EMSGSIZE;
372 	return 0;
373 }
374 
375 /* Register and XDR encode the Write list. Supports encoding a list
376  * containing one array of plain segments that belong to a single
377  * write chunk.
378  *
379  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
380  *
381  *  Write chunklist (a list of (one) counted array):
382  *   N elements:
383  *    1 - N - HLOO - HLOO - ... - HLOO - 0
384  *
385  * Returns zero on success, or a negative errno if a failure occurred.
386  * @xdr is advanced to the next position in the stream.
387  *
388  * Only a single Write chunk is currently supported.
389  */
rpcrdma_encode_write_list(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct rpc_rqst * rqst,enum rpcrdma_chunktype wtype)390 static int rpcrdma_encode_write_list(struct rpcrdma_xprt *r_xprt,
391 				     struct rpcrdma_req *req,
392 				     struct rpc_rqst *rqst,
393 				     enum rpcrdma_chunktype wtype)
394 {
395 	struct xdr_stream *xdr = &req->rl_stream;
396 	struct rpcrdma_ep *ep = r_xprt->rx_ep;
397 	struct rpcrdma_mr_seg *seg;
398 	struct rpcrdma_mr *mr;
399 	int nsegs, nchunks;
400 	__be32 *segcount;
401 
402 	if (wtype != rpcrdma_writech)
403 		goto done;
404 
405 	seg = req->rl_segments;
406 	nsegs = rpcrdma_convert_iovs(r_xprt, &rqst->rq_rcv_buf,
407 				     rqst->rq_rcv_buf.head[0].iov_len,
408 				     wtype, seg);
409 	if (nsegs < 0)
410 		return nsegs;
411 
412 	if (xdr_stream_encode_item_present(xdr) < 0)
413 		return -EMSGSIZE;
414 	segcount = xdr_reserve_space(xdr, sizeof(*segcount));
415 	if (unlikely(!segcount))
416 		return -EMSGSIZE;
417 	/* Actual value encoded below */
418 
419 	nchunks = 0;
420 	do {
421 		seg = rpcrdma_mr_prepare(r_xprt, req, seg, nsegs, true, &mr);
422 		if (IS_ERR(seg))
423 			return PTR_ERR(seg);
424 
425 		if (encode_rdma_segment(xdr, mr) < 0)
426 			return -EMSGSIZE;
427 
428 		trace_xprtrdma_chunk_write(rqst->rq_task, mr, nsegs);
429 		r_xprt->rx_stats.write_chunk_count++;
430 		r_xprt->rx_stats.total_rdma_request += mr->mr_length;
431 		nchunks++;
432 		nsegs -= mr->mr_nents;
433 	} while (nsegs);
434 
435 	if (xdr_pad_size(rqst->rq_rcv_buf.page_len)) {
436 		if (encode_rdma_segment(xdr, ep->re_write_pad_mr) < 0)
437 			return -EMSGSIZE;
438 
439 		trace_xprtrdma_chunk_wp(rqst->rq_task, ep->re_write_pad_mr,
440 					nsegs);
441 		r_xprt->rx_stats.write_chunk_count++;
442 		r_xprt->rx_stats.total_rdma_request += mr->mr_length;
443 		nchunks++;
444 		nsegs -= mr->mr_nents;
445 	}
446 
447 	/* Update count of segments in this Write chunk */
448 	*segcount = cpu_to_be32(nchunks);
449 
450 done:
451 	if (xdr_stream_encode_item_absent(xdr) < 0)
452 		return -EMSGSIZE;
453 	return 0;
454 }
455 
456 /* Register and XDR encode the Reply chunk. Supports encoding an array
457  * of plain segments that belong to a single write (reply) chunk.
458  *
459  * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
460  *
461  *  Reply chunk (a counted array):
462  *   N elements:
463  *    1 - N - HLOO - HLOO - ... - HLOO
464  *
465  * Returns zero on success, or a negative errno if a failure occurred.
466  * @xdr is advanced to the next position in the stream.
467  */
rpcrdma_encode_reply_chunk(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct rpc_rqst * rqst,enum rpcrdma_chunktype wtype)468 static int rpcrdma_encode_reply_chunk(struct rpcrdma_xprt *r_xprt,
469 				      struct rpcrdma_req *req,
470 				      struct rpc_rqst *rqst,
471 				      enum rpcrdma_chunktype wtype)
472 {
473 	struct xdr_stream *xdr = &req->rl_stream;
474 	struct rpcrdma_mr_seg *seg;
475 	struct rpcrdma_mr *mr;
476 	int nsegs, nchunks;
477 	__be32 *segcount;
478 
479 	if (wtype != rpcrdma_replych) {
480 		if (xdr_stream_encode_item_absent(xdr) < 0)
481 			return -EMSGSIZE;
482 		return 0;
483 	}
484 
485 	seg = req->rl_segments;
486 	nsegs = rpcrdma_convert_iovs(r_xprt, &rqst->rq_rcv_buf, 0, wtype, seg);
487 	if (nsegs < 0)
488 		return nsegs;
489 
490 	if (xdr_stream_encode_item_present(xdr) < 0)
491 		return -EMSGSIZE;
492 	segcount = xdr_reserve_space(xdr, sizeof(*segcount));
493 	if (unlikely(!segcount))
494 		return -EMSGSIZE;
495 	/* Actual value encoded below */
496 
497 	nchunks = 0;
498 	do {
499 		seg = rpcrdma_mr_prepare(r_xprt, req, seg, nsegs, true, &mr);
500 		if (IS_ERR(seg))
501 			return PTR_ERR(seg);
502 
503 		if (encode_rdma_segment(xdr, mr) < 0)
504 			return -EMSGSIZE;
505 
506 		trace_xprtrdma_chunk_reply(rqst->rq_task, mr, nsegs);
507 		r_xprt->rx_stats.reply_chunk_count++;
508 		r_xprt->rx_stats.total_rdma_request += mr->mr_length;
509 		nchunks++;
510 		nsegs -= mr->mr_nents;
511 	} while (nsegs);
512 
513 	/* Update count of segments in the Reply chunk */
514 	*segcount = cpu_to_be32(nchunks);
515 
516 	return 0;
517 }
518 
rpcrdma_sendctx_done(struct kref * kref)519 static void rpcrdma_sendctx_done(struct kref *kref)
520 {
521 	struct rpcrdma_req *req =
522 		container_of(kref, struct rpcrdma_req, rl_kref);
523 	struct rpcrdma_rep *rep = req->rl_reply;
524 
525 	rpcrdma_complete_rqst(rep);
526 	rep->rr_rxprt->rx_stats.reply_waits_for_send++;
527 }
528 
529 /**
530  * rpcrdma_sendctx_unmap - DMA-unmap Send buffer
531  * @sc: sendctx containing SGEs to unmap
532  *
533  */
rpcrdma_sendctx_unmap(struct rpcrdma_sendctx * sc)534 void rpcrdma_sendctx_unmap(struct rpcrdma_sendctx *sc)
535 {
536 	struct rpcrdma_regbuf *rb = sc->sc_req->rl_sendbuf;
537 	struct ib_sge *sge;
538 
539 	if (!sc->sc_unmap_count)
540 		return;
541 
542 	/* The first two SGEs contain the transport header and
543 	 * the inline buffer. These are always left mapped so
544 	 * they can be cheaply re-used.
545 	 */
546 	for (sge = &sc->sc_sges[2]; sc->sc_unmap_count;
547 	     ++sge, --sc->sc_unmap_count)
548 		ib_dma_unmap_page(rdmab_device(rb), sge->addr, sge->length,
549 				  DMA_TO_DEVICE);
550 
551 	kref_put(&sc->sc_req->rl_kref, rpcrdma_sendctx_done);
552 }
553 
554 /* Prepare an SGE for the RPC-over-RDMA transport header.
555  */
rpcrdma_prepare_hdr_sge(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,u32 len)556 static void rpcrdma_prepare_hdr_sge(struct rpcrdma_xprt *r_xprt,
557 				    struct rpcrdma_req *req, u32 len)
558 {
559 	struct rpcrdma_sendctx *sc = req->rl_sendctx;
560 	struct rpcrdma_regbuf *rb = req->rl_rdmabuf;
561 	struct ib_sge *sge = &sc->sc_sges[req->rl_wr.num_sge++];
562 
563 	sge->addr = rdmab_addr(rb);
564 	sge->length = len;
565 	sge->lkey = rdmab_lkey(rb);
566 
567 	ib_dma_sync_single_for_device(rdmab_device(rb), sge->addr, sge->length,
568 				      DMA_TO_DEVICE);
569 }
570 
571 /* The head iovec is straightforward, as it is usually already
572  * DMA-mapped. Sync the content that has changed.
573  */
rpcrdma_prepare_head_iov(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,unsigned int len)574 static bool rpcrdma_prepare_head_iov(struct rpcrdma_xprt *r_xprt,
575 				     struct rpcrdma_req *req, unsigned int len)
576 {
577 	struct rpcrdma_sendctx *sc = req->rl_sendctx;
578 	struct ib_sge *sge = &sc->sc_sges[req->rl_wr.num_sge++];
579 	struct rpcrdma_regbuf *rb = req->rl_sendbuf;
580 
581 	if (!rpcrdma_regbuf_dma_map(r_xprt, rb))
582 		return false;
583 
584 	sge->addr = rdmab_addr(rb);
585 	sge->length = len;
586 	sge->lkey = rdmab_lkey(rb);
587 
588 	ib_dma_sync_single_for_device(rdmab_device(rb), sge->addr, sge->length,
589 				      DMA_TO_DEVICE);
590 	return true;
591 }
592 
593 /* If there is a page list present, DMA map and prepare an
594  * SGE for each page to be sent.
595  */
rpcrdma_prepare_pagelist(struct rpcrdma_req * req,struct xdr_buf * xdr)596 static bool rpcrdma_prepare_pagelist(struct rpcrdma_req *req,
597 				     struct xdr_buf *xdr)
598 {
599 	struct rpcrdma_sendctx *sc = req->rl_sendctx;
600 	struct rpcrdma_regbuf *rb = req->rl_sendbuf;
601 	unsigned int page_base, len, remaining;
602 	struct page **ppages;
603 	struct ib_sge *sge;
604 
605 	ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT);
606 	page_base = offset_in_page(xdr->page_base);
607 	remaining = xdr->page_len;
608 	while (remaining) {
609 		sge = &sc->sc_sges[req->rl_wr.num_sge++];
610 		len = min_t(unsigned int, PAGE_SIZE - page_base, remaining);
611 		sge->addr = ib_dma_map_page(rdmab_device(rb), *ppages,
612 					    page_base, len, DMA_TO_DEVICE);
613 		if (ib_dma_mapping_error(rdmab_device(rb), sge->addr))
614 			goto out_mapping_err;
615 
616 		sge->length = len;
617 		sge->lkey = rdmab_lkey(rb);
618 
619 		sc->sc_unmap_count++;
620 		ppages++;
621 		remaining -= len;
622 		page_base = 0;
623 	}
624 
625 	return true;
626 
627 out_mapping_err:
628 	trace_xprtrdma_dma_maperr(sge->addr);
629 	return false;
630 }
631 
632 /* The tail iovec may include an XDR pad for the page list,
633  * as well as additional content, and may not reside in the
634  * same page as the head iovec.
635  */
rpcrdma_prepare_tail_iov(struct rpcrdma_req * req,struct xdr_buf * xdr,unsigned int page_base,unsigned int len)636 static bool rpcrdma_prepare_tail_iov(struct rpcrdma_req *req,
637 				     struct xdr_buf *xdr,
638 				     unsigned int page_base, unsigned int len)
639 {
640 	struct rpcrdma_sendctx *sc = req->rl_sendctx;
641 	struct ib_sge *sge = &sc->sc_sges[req->rl_wr.num_sge++];
642 	struct rpcrdma_regbuf *rb = req->rl_sendbuf;
643 	struct page *page = virt_to_page(xdr->tail[0].iov_base);
644 
645 	sge->addr = ib_dma_map_page(rdmab_device(rb), page, page_base, len,
646 				    DMA_TO_DEVICE);
647 	if (ib_dma_mapping_error(rdmab_device(rb), sge->addr))
648 		goto out_mapping_err;
649 
650 	sge->length = len;
651 	sge->lkey = rdmab_lkey(rb);
652 	++sc->sc_unmap_count;
653 	return true;
654 
655 out_mapping_err:
656 	trace_xprtrdma_dma_maperr(sge->addr);
657 	return false;
658 }
659 
660 /* Copy the tail to the end of the head buffer.
661  */
rpcrdma_pullup_tail_iov(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct xdr_buf * xdr)662 static void rpcrdma_pullup_tail_iov(struct rpcrdma_xprt *r_xprt,
663 				    struct rpcrdma_req *req,
664 				    struct xdr_buf *xdr)
665 {
666 	unsigned char *dst;
667 
668 	dst = (unsigned char *)xdr->head[0].iov_base;
669 	dst += xdr->head[0].iov_len + xdr->page_len;
670 	memmove(dst, xdr->tail[0].iov_base, xdr->tail[0].iov_len);
671 	r_xprt->rx_stats.pullup_copy_count += xdr->tail[0].iov_len;
672 }
673 
674 /* Copy pagelist content into the head buffer.
675  */
rpcrdma_pullup_pagelist(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct xdr_buf * xdr)676 static void rpcrdma_pullup_pagelist(struct rpcrdma_xprt *r_xprt,
677 				    struct rpcrdma_req *req,
678 				    struct xdr_buf *xdr)
679 {
680 	unsigned int len, page_base, remaining;
681 	struct page **ppages;
682 	unsigned char *src, *dst;
683 
684 	dst = (unsigned char *)xdr->head[0].iov_base;
685 	dst += xdr->head[0].iov_len;
686 	ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT);
687 	page_base = offset_in_page(xdr->page_base);
688 	remaining = xdr->page_len;
689 	while (remaining) {
690 		src = page_address(*ppages);
691 		src += page_base;
692 		len = min_t(unsigned int, PAGE_SIZE - page_base, remaining);
693 		memcpy(dst, src, len);
694 		r_xprt->rx_stats.pullup_copy_count += len;
695 
696 		ppages++;
697 		dst += len;
698 		remaining -= len;
699 		page_base = 0;
700 	}
701 }
702 
703 /* Copy the contents of @xdr into @rl_sendbuf and DMA sync it.
704  * When the head, pagelist, and tail are small, a pull-up copy
705  * is considerably less costly than DMA mapping the components
706  * of @xdr.
707  *
708  * Assumptions:
709  *  - the caller has already verified that the total length
710  *    of the RPC Call body will fit into @rl_sendbuf.
711  */
rpcrdma_prepare_noch_pullup(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct xdr_buf * xdr)712 static bool rpcrdma_prepare_noch_pullup(struct rpcrdma_xprt *r_xprt,
713 					struct rpcrdma_req *req,
714 					struct xdr_buf *xdr)
715 {
716 	if (unlikely(xdr->tail[0].iov_len))
717 		rpcrdma_pullup_tail_iov(r_xprt, req, xdr);
718 
719 	if (unlikely(xdr->page_len))
720 		rpcrdma_pullup_pagelist(r_xprt, req, xdr);
721 
722 	/* The whole RPC message resides in the head iovec now */
723 	return rpcrdma_prepare_head_iov(r_xprt, req, xdr->len);
724 }
725 
rpcrdma_prepare_noch_mapped(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct xdr_buf * xdr)726 static bool rpcrdma_prepare_noch_mapped(struct rpcrdma_xprt *r_xprt,
727 					struct rpcrdma_req *req,
728 					struct xdr_buf *xdr)
729 {
730 	struct kvec *tail = &xdr->tail[0];
731 
732 	if (!rpcrdma_prepare_head_iov(r_xprt, req, xdr->head[0].iov_len))
733 		return false;
734 	if (xdr->page_len)
735 		if (!rpcrdma_prepare_pagelist(req, xdr))
736 			return false;
737 	if (tail->iov_len)
738 		if (!rpcrdma_prepare_tail_iov(req, xdr,
739 					      offset_in_page(tail->iov_base),
740 					      tail->iov_len))
741 			return false;
742 
743 	if (req->rl_sendctx->sc_unmap_count)
744 		kref_get(&req->rl_kref);
745 	return true;
746 }
747 
rpcrdma_prepare_readch(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,struct xdr_buf * xdr)748 static bool rpcrdma_prepare_readch(struct rpcrdma_xprt *r_xprt,
749 				   struct rpcrdma_req *req,
750 				   struct xdr_buf *xdr)
751 {
752 	if (!rpcrdma_prepare_head_iov(r_xprt, req, xdr->head[0].iov_len))
753 		return false;
754 
755 	/* If there is a Read chunk, the page list is being handled
756 	 * via explicit RDMA, and thus is skipped here.
757 	 */
758 
759 	/* Do not include the tail if it is only an XDR pad */
760 	if (xdr->tail[0].iov_len > 3) {
761 		unsigned int page_base, len;
762 
763 		/* If the content in the page list is an odd length,
764 		 * xdr_write_pages() adds a pad at the beginning of
765 		 * the tail iovec. Force the tail's non-pad content to
766 		 * land at the next XDR position in the Send message.
767 		 */
768 		page_base = offset_in_page(xdr->tail[0].iov_base);
769 		len = xdr->tail[0].iov_len;
770 		page_base += len & 3;
771 		len -= len & 3;
772 		if (!rpcrdma_prepare_tail_iov(req, xdr, page_base, len))
773 			return false;
774 		kref_get(&req->rl_kref);
775 	}
776 
777 	return true;
778 }
779 
780 /**
781  * rpcrdma_prepare_send_sges - Construct SGEs for a Send WR
782  * @r_xprt: controlling transport
783  * @req: context of RPC Call being marshalled
784  * @hdrlen: size of transport header, in bytes
785  * @xdr: xdr_buf containing RPC Call
786  * @rtype: chunk type being encoded
787  *
788  * Returns 0 on success; otherwise a negative errno is returned.
789  */
rpcrdma_prepare_send_sges(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req,u32 hdrlen,struct xdr_buf * xdr,enum rpcrdma_chunktype rtype)790 inline int rpcrdma_prepare_send_sges(struct rpcrdma_xprt *r_xprt,
791 				     struct rpcrdma_req *req, u32 hdrlen,
792 				     struct xdr_buf *xdr,
793 				     enum rpcrdma_chunktype rtype)
794 {
795 	int ret;
796 
797 	ret = -EAGAIN;
798 	req->rl_sendctx = rpcrdma_sendctx_get_locked(r_xprt);
799 	if (!req->rl_sendctx)
800 		goto out_nosc;
801 	req->rl_sendctx->sc_unmap_count = 0;
802 	req->rl_sendctx->sc_req = req;
803 	kref_init(&req->rl_kref);
804 	req->rl_wr.wr_cqe = &req->rl_sendctx->sc_cqe;
805 	req->rl_wr.sg_list = req->rl_sendctx->sc_sges;
806 	req->rl_wr.num_sge = 0;
807 	req->rl_wr.opcode = IB_WR_SEND;
808 
809 	rpcrdma_prepare_hdr_sge(r_xprt, req, hdrlen);
810 
811 	ret = -EIO;
812 	switch (rtype) {
813 	case rpcrdma_noch_pullup:
814 		if (!rpcrdma_prepare_noch_pullup(r_xprt, req, xdr))
815 			goto out_unmap;
816 		break;
817 	case rpcrdma_noch_mapped:
818 		if (!rpcrdma_prepare_noch_mapped(r_xprt, req, xdr))
819 			goto out_unmap;
820 		break;
821 	case rpcrdma_readch:
822 		if (!rpcrdma_prepare_readch(r_xprt, req, xdr))
823 			goto out_unmap;
824 		break;
825 	case rpcrdma_areadch:
826 		break;
827 	default:
828 		goto out_unmap;
829 	}
830 
831 	return 0;
832 
833 out_unmap:
834 	rpcrdma_sendctx_unmap(req->rl_sendctx);
835 out_nosc:
836 	trace_xprtrdma_prepsend_failed(&req->rl_slot, ret);
837 	return ret;
838 }
839 
840 /**
841  * rpcrdma_marshal_req - Marshal and send one RPC request
842  * @r_xprt: controlling transport
843  * @rqst: RPC request to be marshaled
844  *
845  * For the RPC in "rqst", this function:
846  *  - Chooses the transfer mode (eg., RDMA_MSG or RDMA_NOMSG)
847  *  - Registers Read, Write, and Reply chunks
848  *  - Constructs the transport header
849  *  - Posts a Send WR to send the transport header and request
850  *
851  * Returns:
852  *	%0 if the RPC was sent successfully,
853  *	%-ENOTCONN if the connection was lost,
854  *	%-EAGAIN if the caller should call again with the same arguments,
855  *	%-ENOBUFS if the caller should call again after a delay,
856  *	%-EMSGSIZE if the transport header is too small,
857  *	%-EIO if a permanent problem occurred while marshaling.
858  */
859 int
rpcrdma_marshal_req(struct rpcrdma_xprt * r_xprt,struct rpc_rqst * rqst)860 rpcrdma_marshal_req(struct rpcrdma_xprt *r_xprt, struct rpc_rqst *rqst)
861 {
862 	struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
863 	struct xdr_stream *xdr = &req->rl_stream;
864 	enum rpcrdma_chunktype rtype, wtype;
865 	struct xdr_buf *buf = &rqst->rq_snd_buf;
866 	bool ddp_allowed;
867 	__be32 *p;
868 	int ret;
869 
870 	if (unlikely(rqst->rq_rcv_buf.flags & XDRBUF_SPARSE_PAGES)) {
871 		ret = rpcrdma_alloc_sparse_pages(&rqst->rq_rcv_buf);
872 		if (ret)
873 			return ret;
874 	}
875 
876 	rpcrdma_set_xdrlen(&req->rl_hdrbuf, 0);
877 	xdr_init_encode(xdr, &req->rl_hdrbuf, rdmab_data(req->rl_rdmabuf),
878 			rqst);
879 
880 	/* Fixed header fields */
881 	ret = -EMSGSIZE;
882 	p = xdr_reserve_space(xdr, 4 * sizeof(*p));
883 	if (!p)
884 		goto out_err;
885 	*p++ = rqst->rq_xid;
886 	*p++ = rpcrdma_version;
887 	*p++ = r_xprt->rx_buf.rb_max_requests;
888 
889 	/* When the ULP employs a GSS flavor that guarantees integrity
890 	 * or privacy, direct data placement of individual data items
891 	 * is not allowed.
892 	 */
893 	ddp_allowed = !test_bit(RPCAUTH_AUTH_DATATOUCH,
894 				&rqst->rq_cred->cr_auth->au_flags);
895 
896 	/*
897 	 * Chunks needed for results?
898 	 *
899 	 * o If the expected result is under the inline threshold, all ops
900 	 *   return as inline.
901 	 * o Large read ops return data as write chunk(s), header as
902 	 *   inline.
903 	 * o Large non-read ops return as a single reply chunk.
904 	 */
905 	if (rpcrdma_results_inline(r_xprt, rqst))
906 		wtype = rpcrdma_noch;
907 	else if ((ddp_allowed && rqst->rq_rcv_buf.flags & XDRBUF_READ) &&
908 		 rpcrdma_nonpayload_inline(r_xprt, rqst))
909 		wtype = rpcrdma_writech;
910 	else
911 		wtype = rpcrdma_replych;
912 
913 	/*
914 	 * Chunks needed for arguments?
915 	 *
916 	 * o If the total request is under the inline threshold, all ops
917 	 *   are sent as inline.
918 	 * o Large write ops transmit data as read chunk(s), header as
919 	 *   inline.
920 	 * o Large non-write ops are sent with the entire message as a
921 	 *   single read chunk (protocol 0-position special case).
922 	 *
923 	 * This assumes that the upper layer does not present a request
924 	 * that both has a data payload, and whose non-data arguments
925 	 * by themselves are larger than the inline threshold.
926 	 */
927 	if (rpcrdma_args_inline(r_xprt, rqst)) {
928 		*p++ = rdma_msg;
929 		rtype = buf->len < rdmab_length(req->rl_sendbuf) ?
930 			rpcrdma_noch_pullup : rpcrdma_noch_mapped;
931 	} else if (ddp_allowed && buf->flags & XDRBUF_WRITE) {
932 		*p++ = rdma_msg;
933 		rtype = rpcrdma_readch;
934 	} else {
935 		r_xprt->rx_stats.nomsg_call_count++;
936 		*p++ = rdma_nomsg;
937 		rtype = rpcrdma_areadch;
938 	}
939 
940 	/* This implementation supports the following combinations
941 	 * of chunk lists in one RPC-over-RDMA Call message:
942 	 *
943 	 *   - Read list
944 	 *   - Write list
945 	 *   - Reply chunk
946 	 *   - Read list + Reply chunk
947 	 *
948 	 * It might not yet support the following combinations:
949 	 *
950 	 *   - Read list + Write list
951 	 *
952 	 * It does not support the following combinations:
953 	 *
954 	 *   - Write list + Reply chunk
955 	 *   - Read list + Write list + Reply chunk
956 	 *
957 	 * This implementation supports only a single chunk in each
958 	 * Read or Write list. Thus for example the client cannot
959 	 * send a Call message with a Position Zero Read chunk and a
960 	 * regular Read chunk at the same time.
961 	 */
962 	ret = rpcrdma_encode_read_list(r_xprt, req, rqst, rtype);
963 	if (ret)
964 		goto out_err;
965 	ret = rpcrdma_encode_write_list(r_xprt, req, rqst, wtype);
966 	if (ret)
967 		goto out_err;
968 	ret = rpcrdma_encode_reply_chunk(r_xprt, req, rqst, wtype);
969 	if (ret)
970 		goto out_err;
971 
972 	ret = rpcrdma_prepare_send_sges(r_xprt, req, req->rl_hdrbuf.len,
973 					buf, rtype);
974 	if (ret)
975 		goto out_err;
976 
977 	trace_xprtrdma_marshal(req, rtype, wtype);
978 	return 0;
979 
980 out_err:
981 	trace_xprtrdma_marshal_failed(rqst, ret);
982 	r_xprt->rx_stats.failed_marshal_count++;
983 	frwr_reset(req);
984 	return ret;
985 }
986 
__rpcrdma_update_cwnd_locked(struct rpc_xprt * xprt,struct rpcrdma_buffer * buf,u32 grant)987 static void __rpcrdma_update_cwnd_locked(struct rpc_xprt *xprt,
988 					 struct rpcrdma_buffer *buf,
989 					 u32 grant)
990 {
991 	buf->rb_credits = grant;
992 	xprt->cwnd = grant << RPC_CWNDSHIFT;
993 }
994 
rpcrdma_update_cwnd(struct rpcrdma_xprt * r_xprt,u32 grant)995 static void rpcrdma_update_cwnd(struct rpcrdma_xprt *r_xprt, u32 grant)
996 {
997 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
998 
999 	spin_lock(&xprt->transport_lock);
1000 	__rpcrdma_update_cwnd_locked(xprt, &r_xprt->rx_buf, grant);
1001 	spin_unlock(&xprt->transport_lock);
1002 }
1003 
1004 /**
1005  * rpcrdma_reset_cwnd - Reset the xprt's congestion window
1006  * @r_xprt: controlling transport instance
1007  *
1008  * Prepare @r_xprt for the next connection by reinitializing
1009  * its credit grant to one (see RFC 8166, Section 3.3.3).
1010  */
rpcrdma_reset_cwnd(struct rpcrdma_xprt * r_xprt)1011 void rpcrdma_reset_cwnd(struct rpcrdma_xprt *r_xprt)
1012 {
1013 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1014 
1015 	spin_lock(&xprt->transport_lock);
1016 	xprt->cong = 0;
1017 	__rpcrdma_update_cwnd_locked(xprt, &r_xprt->rx_buf, 1);
1018 	spin_unlock(&xprt->transport_lock);
1019 }
1020 
1021 /**
1022  * rpcrdma_inline_fixup - Scatter inline received data into rqst's iovecs
1023  * @rqst: controlling RPC request
1024  * @srcp: points to RPC message payload in receive buffer
1025  * @copy_len: remaining length of receive buffer content
1026  * @pad: Write chunk pad bytes needed (zero for pure inline)
1027  *
1028  * The upper layer has set the maximum number of bytes it can
1029  * receive in each component of rq_rcv_buf. These values are set in
1030  * the head.iov_len, page_len, tail.iov_len, and buflen fields.
1031  *
1032  * Unlike the TCP equivalent (xdr_partial_copy_from_skb), in
1033  * many cases this function simply updates iov_base pointers in
1034  * rq_rcv_buf to point directly to the received reply data, to
1035  * avoid copying reply data.
1036  *
1037  * Returns the count of bytes which had to be memcopied.
1038  */
1039 static unsigned long
rpcrdma_inline_fixup(struct rpc_rqst * rqst,char * srcp,int copy_len,int pad)1040 rpcrdma_inline_fixup(struct rpc_rqst *rqst, char *srcp, int copy_len, int pad)
1041 {
1042 	unsigned long fixup_copy_count;
1043 	int i, npages, curlen;
1044 	char *destp;
1045 	struct page **ppages;
1046 	int page_base;
1047 
1048 	/* The head iovec is redirected to the RPC reply message
1049 	 * in the receive buffer, to avoid a memcopy.
1050 	 */
1051 	rqst->rq_rcv_buf.head[0].iov_base = srcp;
1052 	rqst->rq_private_buf.head[0].iov_base = srcp;
1053 
1054 	/* The contents of the receive buffer that follow
1055 	 * head.iov_len bytes are copied into the page list.
1056 	 */
1057 	curlen = rqst->rq_rcv_buf.head[0].iov_len;
1058 	if (curlen > copy_len)
1059 		curlen = copy_len;
1060 	srcp += curlen;
1061 	copy_len -= curlen;
1062 
1063 	ppages = rqst->rq_rcv_buf.pages +
1064 		(rqst->rq_rcv_buf.page_base >> PAGE_SHIFT);
1065 	page_base = offset_in_page(rqst->rq_rcv_buf.page_base);
1066 	fixup_copy_count = 0;
1067 	if (copy_len && rqst->rq_rcv_buf.page_len) {
1068 		int pagelist_len;
1069 
1070 		pagelist_len = rqst->rq_rcv_buf.page_len;
1071 		if (pagelist_len > copy_len)
1072 			pagelist_len = copy_len;
1073 		npages = PAGE_ALIGN(page_base + pagelist_len) >> PAGE_SHIFT;
1074 		for (i = 0; i < npages; i++) {
1075 			curlen = PAGE_SIZE - page_base;
1076 			if (curlen > pagelist_len)
1077 				curlen = pagelist_len;
1078 
1079 			destp = kmap_atomic(ppages[i]);
1080 			memcpy(destp + page_base, srcp, curlen);
1081 			flush_dcache_page(ppages[i]);
1082 			kunmap_atomic(destp);
1083 			srcp += curlen;
1084 			copy_len -= curlen;
1085 			fixup_copy_count += curlen;
1086 			pagelist_len -= curlen;
1087 			if (!pagelist_len)
1088 				break;
1089 			page_base = 0;
1090 		}
1091 
1092 		/* Implicit padding for the last segment in a Write
1093 		 * chunk is inserted inline at the front of the tail
1094 		 * iovec. The upper layer ignores the content of
1095 		 * the pad. Simply ensure inline content in the tail
1096 		 * that follows the Write chunk is properly aligned.
1097 		 */
1098 		if (pad)
1099 			srcp -= pad;
1100 	}
1101 
1102 	/* The tail iovec is redirected to the remaining data
1103 	 * in the receive buffer, to avoid a memcopy.
1104 	 */
1105 	if (copy_len || pad) {
1106 		rqst->rq_rcv_buf.tail[0].iov_base = srcp;
1107 		rqst->rq_private_buf.tail[0].iov_base = srcp;
1108 	}
1109 
1110 	if (fixup_copy_count)
1111 		trace_xprtrdma_fixup(rqst, fixup_copy_count);
1112 	return fixup_copy_count;
1113 }
1114 
1115 /* By convention, backchannel calls arrive via rdma_msg type
1116  * messages, and never populate the chunk lists. This makes
1117  * the RPC/RDMA header small and fixed in size, so it is
1118  * straightforward to check the RPC header's direction field.
1119  */
1120 static bool
rpcrdma_is_bcall(struct rpcrdma_xprt * r_xprt,struct rpcrdma_rep * rep)1121 rpcrdma_is_bcall(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep)
1122 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1123 {
1124 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1125 	struct xdr_stream *xdr = &rep->rr_stream;
1126 	__be32 *p;
1127 
1128 	if (rep->rr_proc != rdma_msg)
1129 		return false;
1130 
1131 	/* Peek at stream contents without advancing. */
1132 	p = xdr_inline_decode(xdr, 0);
1133 
1134 	/* Chunk lists */
1135 	if (xdr_item_is_present(p++))
1136 		return false;
1137 	if (xdr_item_is_present(p++))
1138 		return false;
1139 	if (xdr_item_is_present(p++))
1140 		return false;
1141 
1142 	/* RPC header */
1143 	if (*p++ != rep->rr_xid)
1144 		return false;
1145 	if (*p != cpu_to_be32(RPC_CALL))
1146 		return false;
1147 
1148 	/* No bc service. */
1149 	if (xprt->bc_serv == NULL)
1150 		return false;
1151 
1152 	/* Now that we are sure this is a backchannel call,
1153 	 * advance to the RPC header.
1154 	 */
1155 	p = xdr_inline_decode(xdr, 3 * sizeof(*p));
1156 	if (unlikely(!p))
1157 		return true;
1158 
1159 	rpcrdma_bc_receive_call(r_xprt, rep);
1160 	return true;
1161 }
1162 #else	/* CONFIG_SUNRPC_BACKCHANNEL */
1163 {
1164 	return false;
1165 }
1166 #endif	/* CONFIG_SUNRPC_BACKCHANNEL */
1167 
decode_rdma_segment(struct xdr_stream * xdr,u32 * length)1168 static int decode_rdma_segment(struct xdr_stream *xdr, u32 *length)
1169 {
1170 	u32 handle;
1171 	u64 offset;
1172 	__be32 *p;
1173 
1174 	p = xdr_inline_decode(xdr, 4 * sizeof(*p));
1175 	if (unlikely(!p))
1176 		return -EIO;
1177 
1178 	xdr_decode_rdma_segment(p, &handle, length, &offset);
1179 	trace_xprtrdma_decode_seg(handle, *length, offset);
1180 	return 0;
1181 }
1182 
decode_write_chunk(struct xdr_stream * xdr,u32 * length)1183 static int decode_write_chunk(struct xdr_stream *xdr, u32 *length)
1184 {
1185 	u32 segcount, seglength;
1186 	__be32 *p;
1187 
1188 	p = xdr_inline_decode(xdr, sizeof(*p));
1189 	if (unlikely(!p))
1190 		return -EIO;
1191 
1192 	*length = 0;
1193 	segcount = be32_to_cpup(p);
1194 	while (segcount--) {
1195 		if (decode_rdma_segment(xdr, &seglength))
1196 			return -EIO;
1197 		*length += seglength;
1198 	}
1199 
1200 	return 0;
1201 }
1202 
1203 /* In RPC-over-RDMA Version One replies, a Read list is never
1204  * expected. This decoder is a stub that returns an error if
1205  * a Read list is present.
1206  */
decode_read_list(struct xdr_stream * xdr)1207 static int decode_read_list(struct xdr_stream *xdr)
1208 {
1209 	__be32 *p;
1210 
1211 	p = xdr_inline_decode(xdr, sizeof(*p));
1212 	if (unlikely(!p))
1213 		return -EIO;
1214 	if (unlikely(xdr_item_is_present(p)))
1215 		return -EIO;
1216 	return 0;
1217 }
1218 
1219 /* Supports only one Write chunk in the Write list
1220  */
decode_write_list(struct xdr_stream * xdr,u32 * length)1221 static int decode_write_list(struct xdr_stream *xdr, u32 *length)
1222 {
1223 	u32 chunklen;
1224 	bool first;
1225 	__be32 *p;
1226 
1227 	*length = 0;
1228 	first = true;
1229 	do {
1230 		p = xdr_inline_decode(xdr, sizeof(*p));
1231 		if (unlikely(!p))
1232 			return -EIO;
1233 		if (xdr_item_is_absent(p))
1234 			break;
1235 		if (!first)
1236 			return -EIO;
1237 
1238 		if (decode_write_chunk(xdr, &chunklen))
1239 			return -EIO;
1240 		*length += chunklen;
1241 		first = false;
1242 	} while (true);
1243 	return 0;
1244 }
1245 
decode_reply_chunk(struct xdr_stream * xdr,u32 * length)1246 static int decode_reply_chunk(struct xdr_stream *xdr, u32 *length)
1247 {
1248 	__be32 *p;
1249 
1250 	p = xdr_inline_decode(xdr, sizeof(*p));
1251 	if (unlikely(!p))
1252 		return -EIO;
1253 
1254 	*length = 0;
1255 	if (xdr_item_is_present(p))
1256 		if (decode_write_chunk(xdr, length))
1257 			return -EIO;
1258 	return 0;
1259 }
1260 
1261 static int
rpcrdma_decode_msg(struct rpcrdma_xprt * r_xprt,struct rpcrdma_rep * rep,struct rpc_rqst * rqst)1262 rpcrdma_decode_msg(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep,
1263 		   struct rpc_rqst *rqst)
1264 {
1265 	struct xdr_stream *xdr = &rep->rr_stream;
1266 	u32 writelist, replychunk, rpclen;
1267 	char *base;
1268 
1269 	/* Decode the chunk lists */
1270 	if (decode_read_list(xdr))
1271 		return -EIO;
1272 	if (decode_write_list(xdr, &writelist))
1273 		return -EIO;
1274 	if (decode_reply_chunk(xdr, &replychunk))
1275 		return -EIO;
1276 
1277 	/* RDMA_MSG sanity checks */
1278 	if (unlikely(replychunk))
1279 		return -EIO;
1280 
1281 	/* Build the RPC reply's Payload stream in rqst->rq_rcv_buf */
1282 	base = (char *)xdr_inline_decode(xdr, 0);
1283 	rpclen = xdr_stream_remaining(xdr);
1284 	r_xprt->rx_stats.fixup_copy_count +=
1285 		rpcrdma_inline_fixup(rqst, base, rpclen, writelist & 3);
1286 
1287 	r_xprt->rx_stats.total_rdma_reply += writelist;
1288 	return rpclen + xdr_align_size(writelist);
1289 }
1290 
1291 static noinline int
rpcrdma_decode_nomsg(struct rpcrdma_xprt * r_xprt,struct rpcrdma_rep * rep)1292 rpcrdma_decode_nomsg(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep)
1293 {
1294 	struct xdr_stream *xdr = &rep->rr_stream;
1295 	u32 writelist, replychunk;
1296 
1297 	/* Decode the chunk lists */
1298 	if (decode_read_list(xdr))
1299 		return -EIO;
1300 	if (decode_write_list(xdr, &writelist))
1301 		return -EIO;
1302 	if (decode_reply_chunk(xdr, &replychunk))
1303 		return -EIO;
1304 
1305 	/* RDMA_NOMSG sanity checks */
1306 	if (unlikely(writelist))
1307 		return -EIO;
1308 	if (unlikely(!replychunk))
1309 		return -EIO;
1310 
1311 	/* Reply chunk buffer already is the reply vector */
1312 	r_xprt->rx_stats.total_rdma_reply += replychunk;
1313 	return replychunk;
1314 }
1315 
1316 static noinline int
rpcrdma_decode_error(struct rpcrdma_xprt * r_xprt,struct rpcrdma_rep * rep,struct rpc_rqst * rqst)1317 rpcrdma_decode_error(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep,
1318 		     struct rpc_rqst *rqst)
1319 {
1320 	struct xdr_stream *xdr = &rep->rr_stream;
1321 	__be32 *p;
1322 
1323 	p = xdr_inline_decode(xdr, sizeof(*p));
1324 	if (unlikely(!p))
1325 		return -EIO;
1326 
1327 	switch (*p) {
1328 	case err_vers:
1329 		p = xdr_inline_decode(xdr, 2 * sizeof(*p));
1330 		if (!p)
1331 			break;
1332 		trace_xprtrdma_err_vers(rqst, p, p + 1);
1333 		break;
1334 	case err_chunk:
1335 		trace_xprtrdma_err_chunk(rqst);
1336 		break;
1337 	default:
1338 		trace_xprtrdma_err_unrecognized(rqst, p);
1339 	}
1340 
1341 	return -EIO;
1342 }
1343 
1344 /**
1345  * rpcrdma_unpin_rqst - Release rqst without completing it
1346  * @rep: RPC/RDMA Receive context
1347  *
1348  * This is done when a connection is lost so that a Reply
1349  * can be dropped and its matching Call can be subsequently
1350  * retransmitted on a new connection.
1351  */
rpcrdma_unpin_rqst(struct rpcrdma_rep * rep)1352 void rpcrdma_unpin_rqst(struct rpcrdma_rep *rep)
1353 {
1354 	struct rpc_xprt *xprt = &rep->rr_rxprt->rx_xprt;
1355 	struct rpc_rqst *rqst = rep->rr_rqst;
1356 	struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
1357 
1358 	req->rl_reply = NULL;
1359 	rep->rr_rqst = NULL;
1360 
1361 	spin_lock(&xprt->queue_lock);
1362 	xprt_unpin_rqst(rqst);
1363 	spin_unlock(&xprt->queue_lock);
1364 }
1365 
1366 /**
1367  * rpcrdma_complete_rqst - Pass completed rqst back to RPC
1368  * @rep: RPC/RDMA Receive context
1369  *
1370  * Reconstruct the RPC reply and complete the transaction
1371  * while @rqst is still pinned to ensure the rep, rqst, and
1372  * rq_task pointers remain stable.
1373  */
rpcrdma_complete_rqst(struct rpcrdma_rep * rep)1374 void rpcrdma_complete_rqst(struct rpcrdma_rep *rep)
1375 {
1376 	struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;
1377 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1378 	struct rpc_rqst *rqst = rep->rr_rqst;
1379 	int status;
1380 
1381 	switch (rep->rr_proc) {
1382 	case rdma_msg:
1383 		status = rpcrdma_decode_msg(r_xprt, rep, rqst);
1384 		break;
1385 	case rdma_nomsg:
1386 		status = rpcrdma_decode_nomsg(r_xprt, rep);
1387 		break;
1388 	case rdma_error:
1389 		status = rpcrdma_decode_error(r_xprt, rep, rqst);
1390 		break;
1391 	default:
1392 		status = -EIO;
1393 	}
1394 	if (status < 0)
1395 		goto out_badheader;
1396 
1397 out:
1398 	spin_lock(&xprt->queue_lock);
1399 	xprt_complete_rqst(rqst->rq_task, status);
1400 	xprt_unpin_rqst(rqst);
1401 	spin_unlock(&xprt->queue_lock);
1402 	return;
1403 
1404 out_badheader:
1405 	trace_xprtrdma_reply_hdr_err(rep);
1406 	r_xprt->rx_stats.bad_reply_count++;
1407 	rqst->rq_task->tk_status = status;
1408 	status = 0;
1409 	goto out;
1410 }
1411 
rpcrdma_reply_done(struct kref * kref)1412 static void rpcrdma_reply_done(struct kref *kref)
1413 {
1414 	struct rpcrdma_req *req =
1415 		container_of(kref, struct rpcrdma_req, rl_kref);
1416 
1417 	rpcrdma_complete_rqst(req->rl_reply);
1418 }
1419 
1420 /**
1421  * rpcrdma_reply_handler - Process received RPC/RDMA messages
1422  * @rep: Incoming rpcrdma_rep object to process
1423  *
1424  * Errors must result in the RPC task either being awakened, or
1425  * allowed to timeout, to discover the errors at that time.
1426  */
rpcrdma_reply_handler(struct rpcrdma_rep * rep)1427 void rpcrdma_reply_handler(struct rpcrdma_rep *rep)
1428 {
1429 	struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;
1430 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
1431 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1432 	struct rpcrdma_req *req;
1433 	struct rpc_rqst *rqst;
1434 	u32 credits;
1435 	__be32 *p;
1436 
1437 	/* Any data means we had a useful conversation, so
1438 	 * then we don't need to delay the next reconnect.
1439 	 */
1440 	if (xprt->reestablish_timeout)
1441 		xprt->reestablish_timeout = 0;
1442 
1443 	/* Fixed transport header fields */
1444 	xdr_init_decode(&rep->rr_stream, &rep->rr_hdrbuf,
1445 			rep->rr_hdrbuf.head[0].iov_base, NULL);
1446 	p = xdr_inline_decode(&rep->rr_stream, 4 * sizeof(*p));
1447 	if (unlikely(!p))
1448 		goto out_shortreply;
1449 	rep->rr_xid = *p++;
1450 	rep->rr_vers = *p++;
1451 	credits = be32_to_cpu(*p++);
1452 	rep->rr_proc = *p++;
1453 
1454 	if (rep->rr_vers != rpcrdma_version)
1455 		goto out_badversion;
1456 
1457 	if (rpcrdma_is_bcall(r_xprt, rep))
1458 		return;
1459 
1460 	/* Match incoming rpcrdma_rep to an rpcrdma_req to
1461 	 * get context for handling any incoming chunks.
1462 	 */
1463 	spin_lock(&xprt->queue_lock);
1464 	rqst = xprt_lookup_rqst(xprt, rep->rr_xid);
1465 	if (!rqst)
1466 		goto out_norqst;
1467 	xprt_pin_rqst(rqst);
1468 	spin_unlock(&xprt->queue_lock);
1469 
1470 	if (credits == 0)
1471 		credits = 1;	/* don't deadlock */
1472 	else if (credits > r_xprt->rx_ep->re_max_requests)
1473 		credits = r_xprt->rx_ep->re_max_requests;
1474 	rpcrdma_post_recvs(r_xprt, credits + (buf->rb_bc_srv_max_requests << 1),
1475 			   false);
1476 	if (buf->rb_credits != credits)
1477 		rpcrdma_update_cwnd(r_xprt, credits);
1478 
1479 	req = rpcr_to_rdmar(rqst);
1480 	if (unlikely(req->rl_reply))
1481 		rpcrdma_rep_put(buf, req->rl_reply);
1482 	req->rl_reply = rep;
1483 	rep->rr_rqst = rqst;
1484 
1485 	trace_xprtrdma_reply(rqst->rq_task, rep, credits);
1486 
1487 	if (rep->rr_wc_flags & IB_WC_WITH_INVALIDATE)
1488 		frwr_reminv(rep, &req->rl_registered);
1489 	if (!list_empty(&req->rl_registered))
1490 		frwr_unmap_async(r_xprt, req);
1491 		/* LocalInv completion will complete the RPC */
1492 	else
1493 		kref_put(&req->rl_kref, rpcrdma_reply_done);
1494 	return;
1495 
1496 out_badversion:
1497 	trace_xprtrdma_reply_vers_err(rep);
1498 	goto out;
1499 
1500 out_norqst:
1501 	spin_unlock(&xprt->queue_lock);
1502 	trace_xprtrdma_reply_rqst_err(rep);
1503 	goto out;
1504 
1505 out_shortreply:
1506 	trace_xprtrdma_reply_short_err(rep);
1507 
1508 out:
1509 	rpcrdma_rep_put(buf, rep);
1510 }
1511