1 // SPDX-License-Identifier: GPL-2.0
2 /*
3  * Copyright (c) 2016-2018 Oracle.  All rights reserved.
4  *
5  * Use the core R/W API to move RPC-over-RDMA Read and Write chunks.
6  */
7 
8 #include <rdma/rw.h>
9 
10 #include <linux/sunrpc/xdr.h>
11 #include <linux/sunrpc/rpc_rdma.h>
12 #include <linux/sunrpc/svc_rdma.h>
13 
14 #include "xprt_rdma.h"
15 #include <trace/events/rpcrdma.h>
16 
17 static void svc_rdma_write_done(struct ib_cq *cq, struct ib_wc *wc);
18 static void svc_rdma_wc_read_done(struct ib_cq *cq, struct ib_wc *wc);
19 
20 /* Each R/W context contains state for one chain of RDMA Read or
21  * Write Work Requests.
22  *
23  * Each WR chain handles a single contiguous server-side buffer,
24  * because scatterlist entries after the first have to start on
25  * page alignment. xdr_buf iovecs cannot guarantee alignment.
26  *
27  * Each WR chain handles only one R_key. Each RPC-over-RDMA segment
28  * from a client may contain a unique R_key, so each WR chain moves
29  * up to one segment at a time.
30  *
31  * The scatterlist makes this data structure over 4KB in size. To
32  * make it less likely to fail, and to handle the allocation for
33  * smaller I/O requests without disabling bottom-halves, these
34  * contexts are created on demand, but cached and reused until the
35  * controlling svcxprt_rdma is destroyed.
36  */
37 struct svc_rdma_rw_ctxt {
38 	struct llist_node	rw_node;
39 	struct list_head	rw_list;
40 	struct rdma_rw_ctx	rw_ctx;
41 	unsigned int		rw_nents;
42 	struct sg_table		rw_sg_table;
43 	struct scatterlist	rw_first_sgl[];
44 };
45 
46 static inline struct svc_rdma_rw_ctxt *
47 svc_rdma_next_ctxt(struct list_head *list)
48 {
49 	return list_first_entry_or_null(list, struct svc_rdma_rw_ctxt,
50 					rw_list);
51 }
52 
53 static struct svc_rdma_rw_ctxt *
54 svc_rdma_get_rw_ctxt(struct svcxprt_rdma *rdma, unsigned int sges)
55 {
56 	struct svc_rdma_rw_ctxt *ctxt;
57 	struct llist_node *node;
58 
59 	spin_lock(&rdma->sc_rw_ctxt_lock);
60 	node = llist_del_first(&rdma->sc_rw_ctxts);
61 	spin_unlock(&rdma->sc_rw_ctxt_lock);
62 	if (node) {
63 		ctxt = llist_entry(node, struct svc_rdma_rw_ctxt, rw_node);
64 	} else {
65 		ctxt = kmalloc(struct_size(ctxt, rw_first_sgl, SG_CHUNK_SIZE),
66 			       GFP_KERNEL);
67 		if (!ctxt)
68 			goto out_noctx;
69 
70 		INIT_LIST_HEAD(&ctxt->rw_list);
71 	}
72 
73 	ctxt->rw_sg_table.sgl = ctxt->rw_first_sgl;
74 	if (sg_alloc_table_chained(&ctxt->rw_sg_table, sges,
75 				   ctxt->rw_sg_table.sgl,
76 				   SG_CHUNK_SIZE))
77 		goto out_free;
78 	return ctxt;
79 
80 out_free:
81 	kfree(ctxt);
82 out_noctx:
83 	trace_svcrdma_no_rwctx_err(rdma, sges);
84 	return NULL;
85 }
86 
87 static void __svc_rdma_put_rw_ctxt(struct svcxprt_rdma *rdma,
88 				   struct svc_rdma_rw_ctxt *ctxt,
89 				   struct llist_head *list)
90 {
91 	sg_free_table_chained(&ctxt->rw_sg_table, SG_CHUNK_SIZE);
92 	llist_add(&ctxt->rw_node, list);
93 }
94 
95 static void svc_rdma_put_rw_ctxt(struct svcxprt_rdma *rdma,
96 				 struct svc_rdma_rw_ctxt *ctxt)
97 {
98 	__svc_rdma_put_rw_ctxt(rdma, ctxt, &rdma->sc_rw_ctxts);
99 }
100 
101 /**
102  * svc_rdma_destroy_rw_ctxts - Free accumulated R/W contexts
103  * @rdma: transport about to be destroyed
104  *
105  */
106 void svc_rdma_destroy_rw_ctxts(struct svcxprt_rdma *rdma)
107 {
108 	struct svc_rdma_rw_ctxt *ctxt;
109 	struct llist_node *node;
110 
111 	while ((node = llist_del_first(&rdma->sc_rw_ctxts)) != NULL) {
112 		ctxt = llist_entry(node, struct svc_rdma_rw_ctxt, rw_node);
113 		kfree(ctxt);
114 	}
115 }
116 
117 /**
118  * svc_rdma_rw_ctx_init - Prepare a R/W context for I/O
119  * @rdma: controlling transport instance
120  * @ctxt: R/W context to prepare
121  * @offset: RDMA offset
122  * @handle: RDMA tag/handle
123  * @direction: I/O direction
124  *
125  * Returns on success, the number of WQEs that will be needed
126  * on the workqueue, or a negative errno.
127  */
128 static int svc_rdma_rw_ctx_init(struct svcxprt_rdma *rdma,
129 				struct svc_rdma_rw_ctxt *ctxt,
130 				u64 offset, u32 handle,
131 				enum dma_data_direction direction)
132 {
133 	int ret;
134 
135 	ret = rdma_rw_ctx_init(&ctxt->rw_ctx, rdma->sc_qp, rdma->sc_port_num,
136 			       ctxt->rw_sg_table.sgl, ctxt->rw_nents,
137 			       0, offset, handle, direction);
138 	if (unlikely(ret < 0)) {
139 		svc_rdma_put_rw_ctxt(rdma, ctxt);
140 		trace_svcrdma_dma_map_rw_err(rdma, ctxt->rw_nents, ret);
141 	}
142 	return ret;
143 }
144 
145 /* A chunk context tracks all I/O for moving one Read or Write
146  * chunk. This is a set of rdma_rw's that handle data movement
147  * for all segments of one chunk.
148  *
149  * These are small, acquired with a single allocator call, and
150  * no more than one is needed per chunk. They are allocated on
151  * demand, and not cached.
152  */
153 struct svc_rdma_chunk_ctxt {
154 	struct rpc_rdma_cid	cc_cid;
155 	struct ib_cqe		cc_cqe;
156 	struct svcxprt_rdma	*cc_rdma;
157 	struct list_head	cc_rwctxts;
158 	ktime_t			cc_posttime;
159 	int			cc_sqecount;
160 	enum ib_wc_status	cc_status;
161 	struct completion	cc_done;
162 };
163 
164 static void svc_rdma_cc_cid_init(struct svcxprt_rdma *rdma,
165 				 struct rpc_rdma_cid *cid)
166 {
167 	cid->ci_queue_id = rdma->sc_sq_cq->res.id;
168 	cid->ci_completion_id = atomic_inc_return(&rdma->sc_completion_ids);
169 }
170 
171 static void svc_rdma_cc_init(struct svcxprt_rdma *rdma,
172 			     struct svc_rdma_chunk_ctxt *cc)
173 {
174 	svc_rdma_cc_cid_init(rdma, &cc->cc_cid);
175 	cc->cc_rdma = rdma;
176 
177 	INIT_LIST_HEAD(&cc->cc_rwctxts);
178 	cc->cc_sqecount = 0;
179 }
180 
181 /*
182  * The consumed rw_ctx's are cleaned and placed on a local llist so
183  * that only one atomic llist operation is needed to put them all
184  * back on the free list.
185  */
186 static void svc_rdma_cc_release(struct svc_rdma_chunk_ctxt *cc,
187 				enum dma_data_direction dir)
188 {
189 	struct svcxprt_rdma *rdma = cc->cc_rdma;
190 	struct llist_node *first, *last;
191 	struct svc_rdma_rw_ctxt *ctxt;
192 	LLIST_HEAD(free);
193 
194 	first = last = NULL;
195 	while ((ctxt = svc_rdma_next_ctxt(&cc->cc_rwctxts)) != NULL) {
196 		list_del(&ctxt->rw_list);
197 
198 		rdma_rw_ctx_destroy(&ctxt->rw_ctx, rdma->sc_qp,
199 				    rdma->sc_port_num, ctxt->rw_sg_table.sgl,
200 				    ctxt->rw_nents, dir);
201 		__svc_rdma_put_rw_ctxt(rdma, ctxt, &free);
202 
203 		ctxt->rw_node.next = first;
204 		first = &ctxt->rw_node;
205 		if (!last)
206 			last = first;
207 	}
208 	if (first)
209 		llist_add_batch(first, last, &rdma->sc_rw_ctxts);
210 }
211 
212 /* State for sending a Write or Reply chunk.
213  *  - Tracks progress of writing one chunk over all its segments
214  *  - Stores arguments for the SGL constructor functions
215  */
216 struct svc_rdma_write_info {
217 	const struct svc_rdma_chunk	*wi_chunk;
218 
219 	/* write state of this chunk */
220 	unsigned int		wi_seg_off;
221 	unsigned int		wi_seg_no;
222 
223 	/* SGL constructor arguments */
224 	const struct xdr_buf	*wi_xdr;
225 	unsigned char		*wi_base;
226 	unsigned int		wi_next_off;
227 
228 	struct svc_rdma_chunk_ctxt	wi_cc;
229 };
230 
231 static struct svc_rdma_write_info *
232 svc_rdma_write_info_alloc(struct svcxprt_rdma *rdma,
233 			  const struct svc_rdma_chunk *chunk)
234 {
235 	struct svc_rdma_write_info *info;
236 
237 	info = kmalloc(sizeof(*info), GFP_KERNEL);
238 	if (!info)
239 		return info;
240 
241 	info->wi_chunk = chunk;
242 	info->wi_seg_off = 0;
243 	info->wi_seg_no = 0;
244 	svc_rdma_cc_init(rdma, &info->wi_cc);
245 	info->wi_cc.cc_cqe.done = svc_rdma_write_done;
246 	return info;
247 }
248 
249 static void svc_rdma_write_info_free(struct svc_rdma_write_info *info)
250 {
251 	svc_rdma_cc_release(&info->wi_cc, DMA_TO_DEVICE);
252 	kfree(info);
253 }
254 
255 /**
256  * svc_rdma_write_done - Write chunk completion
257  * @cq: controlling Completion Queue
258  * @wc: Work Completion
259  *
260  * Pages under I/O are freed by a subsequent Send completion.
261  */
262 static void svc_rdma_write_done(struct ib_cq *cq, struct ib_wc *wc)
263 {
264 	struct ib_cqe *cqe = wc->wr_cqe;
265 	struct svc_rdma_chunk_ctxt *cc =
266 			container_of(cqe, struct svc_rdma_chunk_ctxt, cc_cqe);
267 	struct svcxprt_rdma *rdma = cc->cc_rdma;
268 	struct svc_rdma_write_info *info =
269 			container_of(cc, struct svc_rdma_write_info, wi_cc);
270 
271 	switch (wc->status) {
272 	case IB_WC_SUCCESS:
273 		trace_svcrdma_wc_write(wc, &cc->cc_cid);
274 		break;
275 	case IB_WC_WR_FLUSH_ERR:
276 		trace_svcrdma_wc_write_flush(wc, &cc->cc_cid);
277 		break;
278 	default:
279 		trace_svcrdma_wc_write_err(wc, &cc->cc_cid);
280 	}
281 
282 	svc_rdma_wake_send_waiters(rdma, cc->cc_sqecount);
283 
284 	if (unlikely(wc->status != IB_WC_SUCCESS))
285 		svc_xprt_deferred_close(&rdma->sc_xprt);
286 
287 	svc_rdma_write_info_free(info);
288 }
289 
290 /* State for pulling a Read chunk.
291  */
292 struct svc_rdma_read_info {
293 	struct svc_rqst			*ri_rqst;
294 	struct svc_rdma_recv_ctxt	*ri_readctxt;
295 	unsigned int			ri_pageno;
296 	unsigned int			ri_pageoff;
297 	unsigned int			ri_totalbytes;
298 
299 	struct svc_rdma_chunk_ctxt	ri_cc;
300 };
301 
302 static struct svc_rdma_read_info *
303 svc_rdma_read_info_alloc(struct svcxprt_rdma *rdma)
304 {
305 	struct svc_rdma_read_info *info;
306 
307 	info = kmalloc(sizeof(*info), GFP_KERNEL);
308 	if (!info)
309 		return info;
310 
311 	svc_rdma_cc_init(rdma, &info->ri_cc);
312 	info->ri_cc.cc_cqe.done = svc_rdma_wc_read_done;
313 	return info;
314 }
315 
316 static void svc_rdma_read_info_free(struct svc_rdma_read_info *info)
317 {
318 	svc_rdma_cc_release(&info->ri_cc, DMA_FROM_DEVICE);
319 	kfree(info);
320 }
321 
322 /**
323  * svc_rdma_wc_read_done - Handle completion of an RDMA Read ctx
324  * @cq: controlling Completion Queue
325  * @wc: Work Completion
326  *
327  */
328 static void svc_rdma_wc_read_done(struct ib_cq *cq, struct ib_wc *wc)
329 {
330 	struct ib_cqe *cqe = wc->wr_cqe;
331 	struct svc_rdma_chunk_ctxt *cc =
332 			container_of(cqe, struct svc_rdma_chunk_ctxt, cc_cqe);
333 	struct svc_rdma_read_info *info;
334 
335 	switch (wc->status) {
336 	case IB_WC_SUCCESS:
337 		info = container_of(cc, struct svc_rdma_read_info, ri_cc);
338 		trace_svcrdma_wc_read(wc, &cc->cc_cid, info->ri_totalbytes,
339 				      cc->cc_posttime);
340 		break;
341 	case IB_WC_WR_FLUSH_ERR:
342 		trace_svcrdma_wc_read_flush(wc, &cc->cc_cid);
343 		break;
344 	default:
345 		trace_svcrdma_wc_read_err(wc, &cc->cc_cid);
346 	}
347 
348 	svc_rdma_wake_send_waiters(cc->cc_rdma, cc->cc_sqecount);
349 	cc->cc_status = wc->status;
350 	complete(&cc->cc_done);
351 	return;
352 }
353 
354 /* This function sleeps when the transport's Send Queue is congested.
355  *
356  * Assumptions:
357  * - If ib_post_send() succeeds, only one completion is expected,
358  *   even if one or more WRs are flushed. This is true when posting
359  *   an rdma_rw_ctx or when posting a single signaled WR.
360  */
361 static int svc_rdma_post_chunk_ctxt(struct svc_rdma_chunk_ctxt *cc)
362 {
363 	struct svcxprt_rdma *rdma = cc->cc_rdma;
364 	struct ib_send_wr *first_wr;
365 	const struct ib_send_wr *bad_wr;
366 	struct list_head *tmp;
367 	struct ib_cqe *cqe;
368 	int ret;
369 
370 	if (cc->cc_sqecount > rdma->sc_sq_depth)
371 		return -EINVAL;
372 
373 	first_wr = NULL;
374 	cqe = &cc->cc_cqe;
375 	list_for_each(tmp, &cc->cc_rwctxts) {
376 		struct svc_rdma_rw_ctxt *ctxt;
377 
378 		ctxt = list_entry(tmp, struct svc_rdma_rw_ctxt, rw_list);
379 		first_wr = rdma_rw_ctx_wrs(&ctxt->rw_ctx, rdma->sc_qp,
380 					   rdma->sc_port_num, cqe, first_wr);
381 		cqe = NULL;
382 	}
383 
384 	do {
385 		if (atomic_sub_return(cc->cc_sqecount,
386 				      &rdma->sc_sq_avail) > 0) {
387 			cc->cc_posttime = ktime_get();
388 			ret = ib_post_send(rdma->sc_qp, first_wr, &bad_wr);
389 			if (ret)
390 				break;
391 			return 0;
392 		}
393 
394 		percpu_counter_inc(&svcrdma_stat_sq_starve);
395 		trace_svcrdma_sq_full(rdma);
396 		atomic_add(cc->cc_sqecount, &rdma->sc_sq_avail);
397 		wait_event(rdma->sc_send_wait,
398 			   atomic_read(&rdma->sc_sq_avail) > cc->cc_sqecount);
399 		trace_svcrdma_sq_retry(rdma);
400 	} while (1);
401 
402 	trace_svcrdma_sq_post_err(rdma, ret);
403 	svc_xprt_deferred_close(&rdma->sc_xprt);
404 
405 	/* If even one was posted, there will be a completion. */
406 	if (bad_wr != first_wr)
407 		return 0;
408 
409 	atomic_add(cc->cc_sqecount, &rdma->sc_sq_avail);
410 	wake_up(&rdma->sc_send_wait);
411 	return -ENOTCONN;
412 }
413 
414 /* Build and DMA-map an SGL that covers one kvec in an xdr_buf
415  */
416 static void svc_rdma_vec_to_sg(struct svc_rdma_write_info *info,
417 			       unsigned int len,
418 			       struct svc_rdma_rw_ctxt *ctxt)
419 {
420 	struct scatterlist *sg = ctxt->rw_sg_table.sgl;
421 
422 	sg_set_buf(&sg[0], info->wi_base, len);
423 	info->wi_base += len;
424 
425 	ctxt->rw_nents = 1;
426 }
427 
428 /* Build and DMA-map an SGL that covers part of an xdr_buf's pagelist.
429  */
430 static void svc_rdma_pagelist_to_sg(struct svc_rdma_write_info *info,
431 				    unsigned int remaining,
432 				    struct svc_rdma_rw_ctxt *ctxt)
433 {
434 	unsigned int sge_no, sge_bytes, page_off, page_no;
435 	const struct xdr_buf *xdr = info->wi_xdr;
436 	struct scatterlist *sg;
437 	struct page **page;
438 
439 	page_off = info->wi_next_off + xdr->page_base;
440 	page_no = page_off >> PAGE_SHIFT;
441 	page_off = offset_in_page(page_off);
442 	page = xdr->pages + page_no;
443 	info->wi_next_off += remaining;
444 	sg = ctxt->rw_sg_table.sgl;
445 	sge_no = 0;
446 	do {
447 		sge_bytes = min_t(unsigned int, remaining,
448 				  PAGE_SIZE - page_off);
449 		sg_set_page(sg, *page, sge_bytes, page_off);
450 
451 		remaining -= sge_bytes;
452 		sg = sg_next(sg);
453 		page_off = 0;
454 		sge_no++;
455 		page++;
456 	} while (remaining);
457 
458 	ctxt->rw_nents = sge_no;
459 }
460 
461 /* Construct RDMA Write WRs to send a portion of an xdr_buf containing
462  * an RPC Reply.
463  */
464 static int
465 svc_rdma_build_writes(struct svc_rdma_write_info *info,
466 		      void (*constructor)(struct svc_rdma_write_info *info,
467 					  unsigned int len,
468 					  struct svc_rdma_rw_ctxt *ctxt),
469 		      unsigned int remaining)
470 {
471 	struct svc_rdma_chunk_ctxt *cc = &info->wi_cc;
472 	struct svcxprt_rdma *rdma = cc->cc_rdma;
473 	const struct svc_rdma_segment *seg;
474 	struct svc_rdma_rw_ctxt *ctxt;
475 	int ret;
476 
477 	do {
478 		unsigned int write_len;
479 		u64 offset;
480 
481 		seg = &info->wi_chunk->ch_segments[info->wi_seg_no];
482 		if (!seg)
483 			goto out_overflow;
484 
485 		write_len = min(remaining, seg->rs_length - info->wi_seg_off);
486 		if (!write_len)
487 			goto out_overflow;
488 		ctxt = svc_rdma_get_rw_ctxt(rdma,
489 					    (write_len >> PAGE_SHIFT) + 2);
490 		if (!ctxt)
491 			return -ENOMEM;
492 
493 		constructor(info, write_len, ctxt);
494 		offset = seg->rs_offset + info->wi_seg_off;
495 		ret = svc_rdma_rw_ctx_init(rdma, ctxt, offset, seg->rs_handle,
496 					   DMA_TO_DEVICE);
497 		if (ret < 0)
498 			return -EIO;
499 		percpu_counter_inc(&svcrdma_stat_write);
500 
501 		list_add(&ctxt->rw_list, &cc->cc_rwctxts);
502 		cc->cc_sqecount += ret;
503 		if (write_len == seg->rs_length - info->wi_seg_off) {
504 			info->wi_seg_no++;
505 			info->wi_seg_off = 0;
506 		} else {
507 			info->wi_seg_off += write_len;
508 		}
509 		remaining -= write_len;
510 	} while (remaining);
511 
512 	return 0;
513 
514 out_overflow:
515 	trace_svcrdma_small_wrch_err(rdma, remaining, info->wi_seg_no,
516 				     info->wi_chunk->ch_segcount);
517 	return -E2BIG;
518 }
519 
520 /**
521  * svc_rdma_iov_write - Construct RDMA Writes from an iov
522  * @info: pointer to write arguments
523  * @iov: kvec to write
524  *
525  * Returns:
526  *   On success, returns zero
527  *   %-E2BIG if the client-provided Write chunk is too small
528  *   %-ENOMEM if a resource has been exhausted
529  *   %-EIO if an rdma-rw error occurred
530  */
531 static int svc_rdma_iov_write(struct svc_rdma_write_info *info,
532 			      const struct kvec *iov)
533 {
534 	info->wi_base = iov->iov_base;
535 	return svc_rdma_build_writes(info, svc_rdma_vec_to_sg,
536 				     iov->iov_len);
537 }
538 
539 /**
540  * svc_rdma_pages_write - Construct RDMA Writes from pages
541  * @info: pointer to write arguments
542  * @xdr: xdr_buf with pages to write
543  * @offset: offset into the content of @xdr
544  * @length: number of bytes to write
545  *
546  * Returns:
547  *   On success, returns zero
548  *   %-E2BIG if the client-provided Write chunk is too small
549  *   %-ENOMEM if a resource has been exhausted
550  *   %-EIO if an rdma-rw error occurred
551  */
552 static int svc_rdma_pages_write(struct svc_rdma_write_info *info,
553 				const struct xdr_buf *xdr,
554 				unsigned int offset,
555 				unsigned long length)
556 {
557 	info->wi_xdr = xdr;
558 	info->wi_next_off = offset - xdr->head[0].iov_len;
559 	return svc_rdma_build_writes(info, svc_rdma_pagelist_to_sg,
560 				     length);
561 }
562 
563 /**
564  * svc_rdma_xb_write - Construct RDMA Writes to write an xdr_buf
565  * @xdr: xdr_buf to write
566  * @data: pointer to write arguments
567  *
568  * Returns:
569  *   On success, returns zero
570  *   %-E2BIG if the client-provided Write chunk is too small
571  *   %-ENOMEM if a resource has been exhausted
572  *   %-EIO if an rdma-rw error occurred
573  */
574 static int svc_rdma_xb_write(const struct xdr_buf *xdr, void *data)
575 {
576 	struct svc_rdma_write_info *info = data;
577 	int ret;
578 
579 	if (xdr->head[0].iov_len) {
580 		ret = svc_rdma_iov_write(info, &xdr->head[0]);
581 		if (ret < 0)
582 			return ret;
583 	}
584 
585 	if (xdr->page_len) {
586 		ret = svc_rdma_pages_write(info, xdr, xdr->head[0].iov_len,
587 					   xdr->page_len);
588 		if (ret < 0)
589 			return ret;
590 	}
591 
592 	if (xdr->tail[0].iov_len) {
593 		ret = svc_rdma_iov_write(info, &xdr->tail[0]);
594 		if (ret < 0)
595 			return ret;
596 	}
597 
598 	return xdr->len;
599 }
600 
601 /**
602  * svc_rdma_send_write_chunk - Write all segments in a Write chunk
603  * @rdma: controlling RDMA transport
604  * @chunk: Write chunk provided by the client
605  * @xdr: xdr_buf containing the data payload
606  *
607  * Returns a non-negative number of bytes the chunk consumed, or
608  *	%-E2BIG if the payload was larger than the Write chunk,
609  *	%-EINVAL if client provided too many segments,
610  *	%-ENOMEM if rdma_rw context pool was exhausted,
611  *	%-ENOTCONN if posting failed (connection is lost),
612  *	%-EIO if rdma_rw initialization failed (DMA mapping, etc).
613  */
614 int svc_rdma_send_write_chunk(struct svcxprt_rdma *rdma,
615 			      const struct svc_rdma_chunk *chunk,
616 			      const struct xdr_buf *xdr)
617 {
618 	struct svc_rdma_write_info *info;
619 	struct svc_rdma_chunk_ctxt *cc;
620 	int ret;
621 
622 	info = svc_rdma_write_info_alloc(rdma, chunk);
623 	if (!info)
624 		return -ENOMEM;
625 	cc = &info->wi_cc;
626 
627 	ret = svc_rdma_xb_write(xdr, info);
628 	if (ret != xdr->len)
629 		goto out_err;
630 
631 	trace_svcrdma_post_write_chunk(&cc->cc_cid, cc->cc_sqecount);
632 	ret = svc_rdma_post_chunk_ctxt(cc);
633 	if (ret < 0)
634 		goto out_err;
635 	return xdr->len;
636 
637 out_err:
638 	svc_rdma_write_info_free(info);
639 	return ret;
640 }
641 
642 /**
643  * svc_rdma_send_reply_chunk - Write all segments in the Reply chunk
644  * @rdma: controlling RDMA transport
645  * @rctxt: Write and Reply chunks from client
646  * @xdr: xdr_buf containing an RPC Reply
647  *
648  * Returns a non-negative number of bytes the chunk consumed, or
649  *	%-E2BIG if the payload was larger than the Reply chunk,
650  *	%-EINVAL if client provided too many segments,
651  *	%-ENOMEM if rdma_rw context pool was exhausted,
652  *	%-ENOTCONN if posting failed (connection is lost),
653  *	%-EIO if rdma_rw initialization failed (DMA mapping, etc).
654  */
655 int svc_rdma_send_reply_chunk(struct svcxprt_rdma *rdma,
656 			      const struct svc_rdma_recv_ctxt *rctxt,
657 			      const struct xdr_buf *xdr)
658 {
659 	struct svc_rdma_write_info *info;
660 	struct svc_rdma_chunk_ctxt *cc;
661 	struct svc_rdma_chunk *chunk;
662 	int ret;
663 
664 	if (pcl_is_empty(&rctxt->rc_reply_pcl))
665 		return 0;
666 
667 	chunk = pcl_first_chunk(&rctxt->rc_reply_pcl);
668 	info = svc_rdma_write_info_alloc(rdma, chunk);
669 	if (!info)
670 		return -ENOMEM;
671 	cc = &info->wi_cc;
672 
673 	ret = pcl_process_nonpayloads(&rctxt->rc_write_pcl, xdr,
674 				      svc_rdma_xb_write, info);
675 	if (ret < 0)
676 		goto out_err;
677 
678 	trace_svcrdma_post_reply_chunk(&cc->cc_cid, cc->cc_sqecount);
679 	ret = svc_rdma_post_chunk_ctxt(cc);
680 	if (ret < 0)
681 		goto out_err;
682 
683 	return xdr->len;
684 
685 out_err:
686 	svc_rdma_write_info_free(info);
687 	return ret;
688 }
689 
690 /**
691  * svc_rdma_build_read_segment - Build RDMA Read WQEs to pull one RDMA segment
692  * @info: context for ongoing I/O
693  * @segment: co-ordinates of remote memory to be read
694  *
695  * Returns:
696  *   %0: the Read WR chain was constructed successfully
697  *   %-EINVAL: there were not enough rq_pages to finish
698  *   %-ENOMEM: allocating a local resources failed
699  *   %-EIO: a DMA mapping error occurred
700  */
701 static int svc_rdma_build_read_segment(struct svc_rdma_read_info *info,
702 				       const struct svc_rdma_segment *segment)
703 {
704 	struct svc_rdma_recv_ctxt *head = info->ri_readctxt;
705 	struct svc_rdma_chunk_ctxt *cc = &info->ri_cc;
706 	struct svc_rqst *rqstp = info->ri_rqst;
707 	unsigned int sge_no, seg_len, len;
708 	struct svc_rdma_rw_ctxt *ctxt;
709 	struct scatterlist *sg;
710 	int ret;
711 
712 	len = segment->rs_length;
713 	sge_no = PAGE_ALIGN(info->ri_pageoff + len) >> PAGE_SHIFT;
714 	ctxt = svc_rdma_get_rw_ctxt(cc->cc_rdma, sge_no);
715 	if (!ctxt)
716 		return -ENOMEM;
717 	ctxt->rw_nents = sge_no;
718 
719 	sg = ctxt->rw_sg_table.sgl;
720 	for (sge_no = 0; sge_no < ctxt->rw_nents; sge_no++) {
721 		seg_len = min_t(unsigned int, len,
722 				PAGE_SIZE - info->ri_pageoff);
723 
724 		if (!info->ri_pageoff)
725 			head->rc_page_count++;
726 
727 		sg_set_page(sg, rqstp->rq_pages[info->ri_pageno],
728 			    seg_len, info->ri_pageoff);
729 		sg = sg_next(sg);
730 
731 		info->ri_pageoff += seg_len;
732 		if (info->ri_pageoff == PAGE_SIZE) {
733 			info->ri_pageno++;
734 			info->ri_pageoff = 0;
735 		}
736 		len -= seg_len;
737 
738 		/* Safety check */
739 		if (len &&
740 		    &rqstp->rq_pages[info->ri_pageno + 1] > rqstp->rq_page_end)
741 			goto out_overrun;
742 	}
743 
744 	ret = svc_rdma_rw_ctx_init(cc->cc_rdma, ctxt, segment->rs_offset,
745 				   segment->rs_handle, DMA_FROM_DEVICE);
746 	if (ret < 0)
747 		return -EIO;
748 	percpu_counter_inc(&svcrdma_stat_read);
749 
750 	list_add(&ctxt->rw_list, &cc->cc_rwctxts);
751 	cc->cc_sqecount += ret;
752 	return 0;
753 
754 out_overrun:
755 	trace_svcrdma_page_overrun_err(cc->cc_rdma, rqstp, info->ri_pageno);
756 	return -EINVAL;
757 }
758 
759 /**
760  * svc_rdma_build_read_chunk - Build RDMA Read WQEs to pull one RDMA chunk
761  * @info: context for ongoing I/O
762  * @chunk: Read chunk to pull
763  *
764  * Return values:
765  *   %0: the Read WR chain was constructed successfully
766  *   %-EINVAL: there were not enough resources to finish
767  *   %-ENOMEM: allocating a local resources failed
768  *   %-EIO: a DMA mapping error occurred
769  */
770 static int svc_rdma_build_read_chunk(struct svc_rdma_read_info *info,
771 				     const struct svc_rdma_chunk *chunk)
772 {
773 	const struct svc_rdma_segment *segment;
774 	int ret;
775 
776 	ret = -EINVAL;
777 	pcl_for_each_segment(segment, chunk) {
778 		ret = svc_rdma_build_read_segment(info, segment);
779 		if (ret < 0)
780 			break;
781 		info->ri_totalbytes += segment->rs_length;
782 	}
783 	return ret;
784 }
785 
786 /**
787  * svc_rdma_copy_inline_range - Copy part of the inline content into pages
788  * @info: context for RDMA Reads
789  * @offset: offset into the Receive buffer of region to copy
790  * @remaining: length of region to copy
791  *
792  * Take a page at a time from rqstp->rq_pages and copy the inline
793  * content from the Receive buffer into that page. Update
794  * info->ri_pageno and info->ri_pageoff so that the next RDMA Read
795  * result will land contiguously with the copied content.
796  *
797  * Return values:
798  *   %0: Inline content was successfully copied
799  *   %-EINVAL: offset or length was incorrect
800  */
801 static int svc_rdma_copy_inline_range(struct svc_rdma_read_info *info,
802 				      unsigned int offset,
803 				      unsigned int remaining)
804 {
805 	struct svc_rdma_recv_ctxt *head = info->ri_readctxt;
806 	unsigned char *dst, *src = head->rc_recv_buf;
807 	struct svc_rqst *rqstp = info->ri_rqst;
808 	unsigned int page_no, numpages;
809 
810 	numpages = PAGE_ALIGN(info->ri_pageoff + remaining) >> PAGE_SHIFT;
811 	for (page_no = 0; page_no < numpages; page_no++) {
812 		unsigned int page_len;
813 
814 		page_len = min_t(unsigned int, remaining,
815 				 PAGE_SIZE - info->ri_pageoff);
816 
817 		if (!info->ri_pageoff)
818 			head->rc_page_count++;
819 
820 		dst = page_address(rqstp->rq_pages[info->ri_pageno]);
821 		memcpy(dst + info->ri_pageno, src + offset, page_len);
822 
823 		info->ri_totalbytes += page_len;
824 		info->ri_pageoff += page_len;
825 		if (info->ri_pageoff == PAGE_SIZE) {
826 			info->ri_pageno++;
827 			info->ri_pageoff = 0;
828 		}
829 		remaining -= page_len;
830 		offset += page_len;
831 	}
832 
833 	return -EINVAL;
834 }
835 
836 /**
837  * svc_rdma_read_multiple_chunks - Construct RDMA Reads to pull data item Read chunks
838  * @info: context for RDMA Reads
839  *
840  * The chunk data lands in rqstp->rq_arg as a series of contiguous pages,
841  * like an incoming TCP call.
842  *
843  * Return values:
844  *   %0: RDMA Read WQEs were successfully built
845  *   %-EINVAL: client provided too many chunks or segments,
846  *   %-ENOMEM: rdma_rw context pool was exhausted,
847  *   %-ENOTCONN: posting failed (connection is lost),
848  *   %-EIO: rdma_rw initialization failed (DMA mapping, etc).
849  */
850 static noinline int svc_rdma_read_multiple_chunks(struct svc_rdma_read_info *info)
851 {
852 	struct svc_rdma_recv_ctxt *head = info->ri_readctxt;
853 	const struct svc_rdma_pcl *pcl = &head->rc_read_pcl;
854 	struct xdr_buf *buf = &info->ri_rqst->rq_arg;
855 	struct svc_rdma_chunk *chunk, *next;
856 	unsigned int start, length;
857 	int ret;
858 
859 	start = 0;
860 	chunk = pcl_first_chunk(pcl);
861 	length = chunk->ch_position;
862 	ret = svc_rdma_copy_inline_range(info, start, length);
863 	if (ret < 0)
864 		return ret;
865 
866 	pcl_for_each_chunk(chunk, pcl) {
867 		ret = svc_rdma_build_read_chunk(info, chunk);
868 		if (ret < 0)
869 			return ret;
870 
871 		next = pcl_next_chunk(pcl, chunk);
872 		if (!next)
873 			break;
874 
875 		start += length;
876 		length = next->ch_position - info->ri_totalbytes;
877 		ret = svc_rdma_copy_inline_range(info, start, length);
878 		if (ret < 0)
879 			return ret;
880 	}
881 
882 	start += length;
883 	length = head->rc_byte_len - start;
884 	ret = svc_rdma_copy_inline_range(info, start, length);
885 	if (ret < 0)
886 		return ret;
887 
888 	buf->len += info->ri_totalbytes;
889 	buf->buflen += info->ri_totalbytes;
890 
891 	buf->head[0].iov_base = page_address(info->ri_rqst->rq_pages[0]);
892 	buf->head[0].iov_len = min_t(size_t, PAGE_SIZE, info->ri_totalbytes);
893 	buf->pages = &info->ri_rqst->rq_pages[1];
894 	buf->page_len = info->ri_totalbytes - buf->head[0].iov_len;
895 	return 0;
896 }
897 
898 /**
899  * svc_rdma_read_data_item - Construct RDMA Reads to pull data item Read chunks
900  * @info: context for RDMA Reads
901  *
902  * The chunk data lands in the page list of rqstp->rq_arg.pages.
903  *
904  * Currently NFSD does not look at the rqstp->rq_arg.tail[0] kvec.
905  * Therefore, XDR round-up of the Read chunk and trailing
906  * inline content must both be added at the end of the pagelist.
907  *
908  * Return values:
909  *   %0: RDMA Read WQEs were successfully built
910  *   %-EINVAL: client provided too many chunks or segments,
911  *   %-ENOMEM: rdma_rw context pool was exhausted,
912  *   %-ENOTCONN: posting failed (connection is lost),
913  *   %-EIO: rdma_rw initialization failed (DMA mapping, etc).
914  */
915 static int svc_rdma_read_data_item(struct svc_rdma_read_info *info)
916 {
917 	struct svc_rdma_recv_ctxt *head = info->ri_readctxt;
918 	struct xdr_buf *buf = &info->ri_rqst->rq_arg;
919 	struct svc_rdma_chunk *chunk;
920 	unsigned int length;
921 	int ret;
922 
923 	chunk = pcl_first_chunk(&head->rc_read_pcl);
924 	ret = svc_rdma_build_read_chunk(info, chunk);
925 	if (ret < 0)
926 		goto out;
927 
928 	/* Split the Receive buffer between the head and tail
929 	 * buffers at Read chunk's position. XDR roundup of the
930 	 * chunk is not included in either the pagelist or in
931 	 * the tail.
932 	 */
933 	buf->tail[0].iov_base = buf->head[0].iov_base + chunk->ch_position;
934 	buf->tail[0].iov_len = buf->head[0].iov_len - chunk->ch_position;
935 	buf->head[0].iov_len = chunk->ch_position;
936 
937 	/* Read chunk may need XDR roundup (see RFC 8166, s. 3.4.5.2).
938 	 *
939 	 * If the client already rounded up the chunk length, the
940 	 * length does not change. Otherwise, the length of the page
941 	 * list is increased to include XDR round-up.
942 	 *
943 	 * Currently these chunks always start at page offset 0,
944 	 * thus the rounded-up length never crosses a page boundary.
945 	 */
946 	buf->pages = &info->ri_rqst->rq_pages[0];
947 	length = xdr_align_size(chunk->ch_length);
948 	buf->page_len = length;
949 	buf->len += length;
950 	buf->buflen += length;
951 
952 out:
953 	return ret;
954 }
955 
956 /**
957  * svc_rdma_read_chunk_range - Build RDMA Read WQEs for portion of a chunk
958  * @info: context for RDMA Reads
959  * @chunk: parsed Call chunk to pull
960  * @offset: offset of region to pull
961  * @length: length of region to pull
962  *
963  * Return values:
964  *   %0: RDMA Read WQEs were successfully built
965  *   %-EINVAL: there were not enough resources to finish
966  *   %-ENOMEM: rdma_rw context pool was exhausted,
967  *   %-ENOTCONN: posting failed (connection is lost),
968  *   %-EIO: rdma_rw initialization failed (DMA mapping, etc).
969  */
970 static int svc_rdma_read_chunk_range(struct svc_rdma_read_info *info,
971 				     const struct svc_rdma_chunk *chunk,
972 				     unsigned int offset, unsigned int length)
973 {
974 	const struct svc_rdma_segment *segment;
975 	int ret;
976 
977 	ret = -EINVAL;
978 	pcl_for_each_segment(segment, chunk) {
979 		struct svc_rdma_segment dummy;
980 
981 		if (offset > segment->rs_length) {
982 			offset -= segment->rs_length;
983 			continue;
984 		}
985 
986 		dummy.rs_handle = segment->rs_handle;
987 		dummy.rs_length = min_t(u32, length, segment->rs_length) - offset;
988 		dummy.rs_offset = segment->rs_offset + offset;
989 
990 		ret = svc_rdma_build_read_segment(info, &dummy);
991 		if (ret < 0)
992 			break;
993 
994 		info->ri_totalbytes += dummy.rs_length;
995 		length -= dummy.rs_length;
996 		offset = 0;
997 	}
998 	return ret;
999 }
1000 
1001 /**
1002  * svc_rdma_read_call_chunk - Build RDMA Read WQEs to pull a Long Message
1003  * @info: context for RDMA Reads
1004  *
1005  * Return values:
1006  *   %0: RDMA Read WQEs were successfully built
1007  *   %-EINVAL: there were not enough resources to finish
1008  *   %-ENOMEM: rdma_rw context pool was exhausted,
1009  *   %-ENOTCONN: posting failed (connection is lost),
1010  *   %-EIO: rdma_rw initialization failed (DMA mapping, etc).
1011  */
1012 static int svc_rdma_read_call_chunk(struct svc_rdma_read_info *info)
1013 {
1014 	struct svc_rdma_recv_ctxt *head = info->ri_readctxt;
1015 	const struct svc_rdma_chunk *call_chunk =
1016 			pcl_first_chunk(&head->rc_call_pcl);
1017 	const struct svc_rdma_pcl *pcl = &head->rc_read_pcl;
1018 	struct svc_rdma_chunk *chunk, *next;
1019 	unsigned int start, length;
1020 	int ret;
1021 
1022 	if (pcl_is_empty(pcl))
1023 		return svc_rdma_build_read_chunk(info, call_chunk);
1024 
1025 	start = 0;
1026 	chunk = pcl_first_chunk(pcl);
1027 	length = chunk->ch_position;
1028 	ret = svc_rdma_read_chunk_range(info, call_chunk, start, length);
1029 	if (ret < 0)
1030 		return ret;
1031 
1032 	pcl_for_each_chunk(chunk, pcl) {
1033 		ret = svc_rdma_build_read_chunk(info, chunk);
1034 		if (ret < 0)
1035 			return ret;
1036 
1037 		next = pcl_next_chunk(pcl, chunk);
1038 		if (!next)
1039 			break;
1040 
1041 		start += length;
1042 		length = next->ch_position - info->ri_totalbytes;
1043 		ret = svc_rdma_read_chunk_range(info, call_chunk,
1044 						start, length);
1045 		if (ret < 0)
1046 			return ret;
1047 	}
1048 
1049 	start += length;
1050 	length = call_chunk->ch_length - start;
1051 	return svc_rdma_read_chunk_range(info, call_chunk, start, length);
1052 }
1053 
1054 /**
1055  * svc_rdma_read_special - Build RDMA Read WQEs to pull a Long Message
1056  * @info: context for RDMA Reads
1057  *
1058  * The start of the data lands in the first page just after the
1059  * Transport header, and the rest lands in rqstp->rq_arg.pages.
1060  *
1061  * Assumptions:
1062  *	- A PZRC is never sent in an RDMA_MSG message, though it's
1063  *	  allowed by spec.
1064  *
1065  * Return values:
1066  *   %0: RDMA Read WQEs were successfully built
1067  *   %-EINVAL: client provided too many chunks or segments,
1068  *   %-ENOMEM: rdma_rw context pool was exhausted,
1069  *   %-ENOTCONN: posting failed (connection is lost),
1070  *   %-EIO: rdma_rw initialization failed (DMA mapping, etc).
1071  */
1072 static noinline int svc_rdma_read_special(struct svc_rdma_read_info *info)
1073 {
1074 	struct xdr_buf *buf = &info->ri_rqst->rq_arg;
1075 	int ret;
1076 
1077 	ret = svc_rdma_read_call_chunk(info);
1078 	if (ret < 0)
1079 		goto out;
1080 
1081 	buf->len += info->ri_totalbytes;
1082 	buf->buflen += info->ri_totalbytes;
1083 
1084 	buf->head[0].iov_base = page_address(info->ri_rqst->rq_pages[0]);
1085 	buf->head[0].iov_len = min_t(size_t, PAGE_SIZE, info->ri_totalbytes);
1086 	buf->pages = &info->ri_rqst->rq_pages[1];
1087 	buf->page_len = info->ri_totalbytes - buf->head[0].iov_len;
1088 
1089 out:
1090 	return ret;
1091 }
1092 
1093 /**
1094  * svc_rdma_process_read_list - Pull list of Read chunks from the client
1095  * @rdma: controlling RDMA transport
1096  * @rqstp: set of pages to use as Read sink buffers
1097  * @head: pages under I/O collect here
1098  *
1099  * The RPC/RDMA protocol assumes that the upper layer's XDR decoders
1100  * pull each Read chunk as they decode an incoming RPC message.
1101  *
1102  * On Linux, however, the server needs to have a fully-constructed RPC
1103  * message in rqstp->rq_arg when there is a positive return code from
1104  * ->xpo_recvfrom. So the Read list is safety-checked immediately when
1105  * it is received, then here the whole Read list is pulled all at once.
1106  * The ingress RPC message is fully reconstructed once all associated
1107  * RDMA Reads have completed.
1108  *
1109  * Return values:
1110  *   %1: all needed RDMA Reads were posted successfully,
1111  *   %-EINVAL: client provided too many chunks or segments,
1112  *   %-ENOMEM: rdma_rw context pool was exhausted,
1113  *   %-ENOTCONN: posting failed (connection is lost),
1114  *   %-EIO: rdma_rw initialization failed (DMA mapping, etc).
1115  */
1116 int svc_rdma_process_read_list(struct svcxprt_rdma *rdma,
1117 			       struct svc_rqst *rqstp,
1118 			       struct svc_rdma_recv_ctxt *head)
1119 {
1120 	struct svc_rdma_read_info *info;
1121 	struct svc_rdma_chunk_ctxt *cc;
1122 	int ret;
1123 
1124 	info = svc_rdma_read_info_alloc(rdma);
1125 	if (!info)
1126 		return -ENOMEM;
1127 	cc = &info->ri_cc;
1128 	info->ri_rqst = rqstp;
1129 	info->ri_readctxt = head;
1130 	info->ri_pageno = 0;
1131 	info->ri_pageoff = 0;
1132 	info->ri_totalbytes = 0;
1133 
1134 	if (pcl_is_empty(&head->rc_call_pcl)) {
1135 		if (head->rc_read_pcl.cl_count == 1)
1136 			ret = svc_rdma_read_data_item(info);
1137 		else
1138 			ret = svc_rdma_read_multiple_chunks(info);
1139 	} else
1140 		ret = svc_rdma_read_special(info);
1141 	if (ret < 0)
1142 		goto out_err;
1143 
1144 	trace_svcrdma_post_read_chunk(&cc->cc_cid, cc->cc_sqecount);
1145 	init_completion(&cc->cc_done);
1146 	ret = svc_rdma_post_chunk_ctxt(cc);
1147 	if (ret < 0)
1148 		goto out_err;
1149 
1150 	ret = 1;
1151 	wait_for_completion(&cc->cc_done);
1152 	if (cc->cc_status != IB_WC_SUCCESS)
1153 		ret = -EIO;
1154 
1155 	/* rq_respages starts after the last arg page */
1156 	rqstp->rq_respages = &rqstp->rq_pages[head->rc_page_count];
1157 	rqstp->rq_next_page = rqstp->rq_respages + 1;
1158 
1159 	/* Ensure svc_rdma_recv_ctxt_put() does not try to release pages */
1160 	head->rc_page_count = 0;
1161 
1162 out_err:
1163 	svc_rdma_read_info_free(info);
1164 	return ret;
1165 }
1166