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
Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.