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