1 // SPDX-License-Identifier: GPL-2.0
2 /*
3  * Copyright (c) 2020 Oracle. All rights reserved.
4  */
5 
6 #include <linux/sunrpc/svc_rdma.h>
7 #include <linux/sunrpc/rpc_rdma.h>
8 
9 #include "xprt_rdma.h"
10 #include <trace/events/rpcrdma.h>
11 
12 /**
13  * pcl_free - Release all memory associated with a parsed chunk list
14  * @pcl: parsed chunk list
15  *
16  */
17 void pcl_free(struct svc_rdma_pcl *pcl)
18 {
19 	while (!list_empty(&pcl->cl_chunks)) {
20 		struct svc_rdma_chunk *chunk;
21 
22 		chunk = pcl_first_chunk(pcl);
23 		list_del(&chunk->ch_list);
24 		kfree(chunk);
25 	}
26 }
27 
28 static struct svc_rdma_chunk *pcl_alloc_chunk(u32 segcount, u32 position)
29 {
30 	struct svc_rdma_chunk *chunk;
31 
32 	chunk = kmalloc(struct_size(chunk, ch_segments, segcount), GFP_KERNEL);
33 	if (!chunk)
34 		return NULL;
35 
36 	chunk->ch_position = position;
37 	chunk->ch_length = 0;
38 	chunk->ch_payload_length = 0;
39 	chunk->ch_segcount = 0;
40 	return chunk;
41 }
42 
43 static struct svc_rdma_chunk *
44 pcl_lookup_position(struct svc_rdma_pcl *pcl, u32 position)
45 {
46 	struct svc_rdma_chunk *pos;
47 
48 	pcl_for_each_chunk(pos, pcl) {
49 		if (pos->ch_position == position)
50 			return pos;
51 	}
52 	return NULL;
53 }
54 
55 static void pcl_insert_position(struct svc_rdma_pcl *pcl,
56 				struct svc_rdma_chunk *chunk)
57 {
58 	struct svc_rdma_chunk *pos;
59 
60 	pcl_for_each_chunk(pos, pcl) {
61 		if (pos->ch_position > chunk->ch_position)
62 			break;
63 	}
64 	__list_add(&chunk->ch_list, pos->ch_list.prev, &pos->ch_list);
65 	pcl->cl_count++;
66 }
67 
68 static void pcl_set_read_segment(const struct svc_rdma_recv_ctxt *rctxt,
69 				 struct svc_rdma_chunk *chunk,
70 				 u32 handle, u32 length, u64 offset)
71 {
72 	struct svc_rdma_segment *segment;
73 
74 	segment = &chunk->ch_segments[chunk->ch_segcount];
75 	segment->rs_handle = handle;
76 	segment->rs_length = length;
77 	segment->rs_offset = offset;
78 
79 	trace_svcrdma_decode_rseg(&rctxt->rc_cid, chunk, segment);
80 
81 	chunk->ch_length += length;
82 	chunk->ch_segcount++;
83 }
84 
85 /**
86  * pcl_alloc_call - Construct a parsed chunk list for the Call body
87  * @rctxt: Ingress receive context
88  * @p: Start of an un-decoded Read list
89  *
90  * Assumptions:
91  * - The incoming Read list has already been sanity checked.
92  * - cl_count is already set to the number of segments in
93  *   the un-decoded list.
94  * - The list might not be in order by position.
95  *
96  * Return values:
97  *       %true: Parsed chunk list was successfully constructed, and
98  *              cl_count is updated to be the number of chunks (ie.
99  *              unique positions) in the Read list.
100  *      %false: Memory allocation failed.
101  */
102 bool pcl_alloc_call(struct svc_rdma_recv_ctxt *rctxt, __be32 *p)
103 {
104 	struct svc_rdma_pcl *pcl = &rctxt->rc_call_pcl;
105 	unsigned int i, segcount = pcl->cl_count;
106 
107 	pcl->cl_count = 0;
108 	for (i = 0; i < segcount; i++) {
109 		struct svc_rdma_chunk *chunk;
110 		u32 position, handle, length;
111 		u64 offset;
112 
113 		p++;	/* skip the list discriminator */
114 		p = xdr_decode_read_segment(p, &position, &handle,
115 					    &length, &offset);
116 		if (position != 0)
117 			continue;
118 
119 		if (pcl_is_empty(pcl)) {
120 			chunk = pcl_alloc_chunk(segcount, position);
121 			if (!chunk)
122 				return false;
123 			pcl_insert_position(pcl, chunk);
124 		} else {
125 			chunk = list_first_entry(&pcl->cl_chunks,
126 						 struct svc_rdma_chunk,
127 						 ch_list);
128 		}
129 
130 		pcl_set_read_segment(rctxt, chunk, handle, length, offset);
131 	}
132 
133 	return true;
134 }
135 
136 /**
137  * pcl_alloc_read - Construct a parsed chunk list for normal Read chunks
138  * @rctxt: Ingress receive context
139  * @p: Start of an un-decoded Read list
140  *
141  * Assumptions:
142  * - The incoming Read list has already been sanity checked.
143  * - cl_count is already set to the number of segments in
144  *   the un-decoded list.
145  * - The list might not be in order by position.
146  *
147  * Return values:
148  *       %true: Parsed chunk list was successfully constructed, and
149  *              cl_count is updated to be the number of chunks (ie.
150  *              unique position values) in the Read list.
151  *      %false: Memory allocation failed.
152  *
153  * TODO:
154  * - Check for chunk range overlaps
155  */
156 bool pcl_alloc_read(struct svc_rdma_recv_ctxt *rctxt, __be32 *p)
157 {
158 	struct svc_rdma_pcl *pcl = &rctxt->rc_read_pcl;
159 	unsigned int i, segcount = pcl->cl_count;
160 
161 	pcl->cl_count = 0;
162 	for (i = 0; i < segcount; i++) {
163 		struct svc_rdma_chunk *chunk;
164 		u32 position, handle, length;
165 		u64 offset;
166 
167 		p++;	/* skip the list discriminator */
168 		p = xdr_decode_read_segment(p, &position, &handle,
169 					    &length, &offset);
170 		if (position == 0)
171 			continue;
172 
173 		chunk = pcl_lookup_position(pcl, position);
174 		if (!chunk) {
175 			chunk = pcl_alloc_chunk(segcount, position);
176 			if (!chunk)
177 				return false;
178 			pcl_insert_position(pcl, chunk);
179 		}
180 
181 		pcl_set_read_segment(rctxt, chunk, handle, length, offset);
182 	}
183 
184 	return true;
185 }
186 
187 /**
188  * pcl_alloc_write - Construct a parsed chunk list from a Write list
189  * @rctxt: Ingress receive context
190  * @pcl: Parsed chunk list to populate
191  * @p: Start of an un-decoded Write list
192  *
193  * Assumptions:
194  * - The incoming Write list has already been sanity checked, and
195  * - cl_count is set to the number of chunks in the un-decoded list.
196  *
197  * Return values:
198  *       %true: Parsed chunk list was successfully constructed.
199  *      %false: Memory allocation failed.
200  */
201 bool pcl_alloc_write(struct svc_rdma_recv_ctxt *rctxt,
202 		     struct svc_rdma_pcl *pcl, __be32 *p)
203 {
204 	struct svc_rdma_segment *segment;
205 	struct svc_rdma_chunk *chunk;
206 	unsigned int i, j;
207 	u32 segcount;
208 
209 	for (i = 0; i < pcl->cl_count; i++) {
210 		p++;	/* skip the list discriminator */
211 		segcount = be32_to_cpup(p++);
212 
213 		chunk = pcl_alloc_chunk(segcount, 0);
214 		if (!chunk)
215 			return false;
216 		list_add_tail(&chunk->ch_list, &pcl->cl_chunks);
217 
218 		for (j = 0; j < segcount; j++) {
219 			segment = &chunk->ch_segments[j];
220 			p = xdr_decode_rdma_segment(p, &segment->rs_handle,
221 						    &segment->rs_length,
222 						    &segment->rs_offset);
223 			trace_svcrdma_decode_wseg(&rctxt->rc_cid, chunk, j);
224 
225 			chunk->ch_length += segment->rs_length;
226 			chunk->ch_segcount++;
227 		}
228 	}
229 	return true;
230 }
231 
232 static int pcl_process_region(const struct xdr_buf *xdr,
233 			      unsigned int offset, unsigned int length,
234 			      int (*actor)(const struct xdr_buf *, void *),
235 			      void *data)
236 {
237 	struct xdr_buf subbuf;
238 
239 	if (!length)
240 		return 0;
241 	if (xdr_buf_subsegment(xdr, &subbuf, offset, length))
242 		return -EMSGSIZE;
243 	return actor(&subbuf, data);
244 }
245 
246 /**
247  * pcl_process_nonpayloads - Process non-payload regions inside @xdr
248  * @pcl: Chunk list to process
249  * @xdr: xdr_buf to process
250  * @actor: Function to invoke on each non-payload region
251  * @data: Arguments for @actor
252  *
253  * This mechanism must ignore not only result payloads that were already
254  * sent via RDMA Write, but also XDR padding for those payloads that
255  * the upper layer has added.
256  *
257  * Assumptions:
258  *  The xdr->len and ch_position fields are aligned to 4-byte multiples.
259  *
260  * Returns:
261  *   On success, zero,
262  *   %-EMSGSIZE on XDR buffer overflow, or
263  *   The return value of @actor
264  */
265 int pcl_process_nonpayloads(const struct svc_rdma_pcl *pcl,
266 			    const struct xdr_buf *xdr,
267 			    int (*actor)(const struct xdr_buf *, void *),
268 			    void *data)
269 {
270 	struct svc_rdma_chunk *chunk, *next;
271 	unsigned int start;
272 	int ret;
273 
274 	chunk = pcl_first_chunk(pcl);
275 
276 	/* No result payloads were generated */
277 	if (!chunk || !chunk->ch_payload_length)
278 		return actor(xdr, data);
279 
280 	/* Process the region before the first result payload */
281 	ret = pcl_process_region(xdr, 0, chunk->ch_position, actor, data);
282 	if (ret < 0)
283 		return ret;
284 
285 	/* Process the regions between each middle result payload */
286 	while ((next = pcl_next_chunk(pcl, chunk))) {
287 		if (!next->ch_payload_length)
288 			break;
289 
290 		start = pcl_chunk_end_offset(chunk);
291 		ret = pcl_process_region(xdr, start, next->ch_position - start,
292 					 actor, data);
293 		if (ret < 0)
294 			return ret;
295 
296 		chunk = next;
297 	}
298 
299 	/* Process the region after the last result payload */
300 	start = pcl_chunk_end_offset(chunk);
301 	ret = pcl_process_region(xdr, start, xdr->len - start, actor, data);
302 	if (ret < 0)
303 		return ret;
304 
305 	return 0;
306 }
307