1 /* SPDX-License-Identifier: GPL-2.0 */
2 /* XDP user-space ring structure
3 * Copyright(c) 2018 Intel Corporation.
4 */
5
6 #ifndef _LINUX_XSK_QUEUE_H
7 #define _LINUX_XSK_QUEUE_H
8
9 #include <linux/types.h>
10 #include <linux/if_xdp.h>
11 #include <net/xdp_sock.h>
12 #include <net/xsk_buff_pool.h>
13
14 #include "xsk.h"
15
16 struct xdp_ring {
17 u32 producer ____cacheline_aligned_in_smp;
18 /* Hinder the adjacent cache prefetcher to prefetch the consumer
19 * pointer if the producer pointer is touched and vice versa.
20 */
21 u32 pad1 ____cacheline_aligned_in_smp;
22 u32 consumer ____cacheline_aligned_in_smp;
23 u32 pad2 ____cacheline_aligned_in_smp;
24 u32 flags;
25 u32 pad3 ____cacheline_aligned_in_smp;
26 };
27
28 /* Used for the RX and TX queues for packets */
29 struct xdp_rxtx_ring {
30 struct xdp_ring ptrs;
31 struct xdp_desc desc[] ____cacheline_aligned_in_smp;
32 };
33
34 /* Used for the fill and completion queues for buffers */
35 struct xdp_umem_ring {
36 struct xdp_ring ptrs;
37 u64 desc[] ____cacheline_aligned_in_smp;
38 };
39
40 struct xsk_queue {
41 u32 ring_mask;
42 u32 nentries;
43 u32 cached_prod;
44 u32 cached_cons;
45 struct xdp_ring *ring;
46 u64 invalid_descs;
47 u64 queue_empty_descs;
48 size_t ring_vmalloc_size;
49 };
50
51 struct parsed_desc {
52 u32 mb;
53 u32 valid;
54 };
55
56 /* The structure of the shared state of the rings are a simple
57 * circular buffer, as outlined in
58 * Documentation/core-api/circular-buffers.rst. For the Rx and
59 * completion ring, the kernel is the producer and user space is the
60 * consumer. For the Tx and fill rings, the kernel is the consumer and
61 * user space is the producer.
62 *
63 * producer consumer
64 *
65 * if (LOAD ->consumer) { (A) LOAD.acq ->producer (C)
66 * STORE $data LOAD $data
67 * STORE.rel ->producer (B) STORE.rel ->consumer (D)
68 * }
69 *
70 * (A) pairs with (D), and (B) pairs with (C).
71 *
72 * Starting with (B), it protects the data from being written after
73 * the producer pointer. If this barrier was missing, the consumer
74 * could observe the producer pointer being set and thus load the data
75 * before the producer has written the new data. The consumer would in
76 * this case load the old data.
77 *
78 * (C) protects the consumer from speculatively loading the data before
79 * the producer pointer actually has been read. If we do not have this
80 * barrier, some architectures could load old data as speculative loads
81 * are not discarded as the CPU does not know there is a dependency
82 * between ->producer and data.
83 *
84 * (A) is a control dependency that separates the load of ->consumer
85 * from the stores of $data. In case ->consumer indicates there is no
86 * room in the buffer to store $data we do not. The dependency will
87 * order both of the stores after the loads. So no barrier is needed.
88 *
89 * (D) protects the load of the data to be observed to happen after the
90 * store of the consumer pointer. If we did not have this memory
91 * barrier, the producer could observe the consumer pointer being set
92 * and overwrite the data with a new value before the consumer got the
93 * chance to read the old value. The consumer would thus miss reading
94 * the old entry and very likely read the new entry twice, once right
95 * now and again after circling through the ring.
96 */
97
98 /* The operations on the rings are the following:
99 *
100 * producer consumer
101 *
102 * RESERVE entries PEEK in the ring for entries
103 * WRITE data into the ring READ data from the ring
104 * SUBMIT entries RELEASE entries
105 *
106 * The producer reserves one or more entries in the ring. It can then
107 * fill in these entries and finally submit them so that they can be
108 * seen and read by the consumer.
109 *
110 * The consumer peeks into the ring to see if the producer has written
111 * any new entries. If so, the consumer can then read these entries
112 * and when it is done reading them release them back to the producer
113 * so that the producer can use these slots to fill in new entries.
114 *
115 * The function names below reflect these operations.
116 */
117
118 /* Functions that read and validate content from consumer rings. */
119
120 static inline void __xskq_cons_read_addr_unchecked(struct xsk_queue *q, u32 cached_cons, u64 *addr)
121 {
122 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
123 u32 idx = cached_cons & q->ring_mask;
124
125 *addr = ring->desc[idx];
126 }
127
128 static inline bool xskq_cons_read_addr_unchecked(struct xsk_queue *q, u64 *addr)
129 {
130 if (q->cached_cons != q->cached_prod) {
131 __xskq_cons_read_addr_unchecked(q, q->cached_cons, addr);
132 return true;
133 }
134
135 return false;
136 }
137
138 static inline bool xp_unused_options_set(u32 options)
139 {
140 return options & ~(XDP_PKT_CONTD | XDP_TX_METADATA);
141 }
142
143 static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
144 struct xdp_desc *desc)
145 {
146 u64 addr = desc->addr - pool->tx_metadata_len;
147 u64 len = desc->len + pool->tx_metadata_len;
148 u64 offset = addr & (pool->chunk_size - 1);
149
150 if (!desc->len)
151 return false;
152
153 if (offset + len > pool->chunk_size)
154 return false;
155
156 if (addr >= pool->addrs_cnt)
157 return false;
158
159 if (xp_unused_options_set(desc->options))
160 return false;
161 return true;
162 }
163
164 static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool,
165 struct xdp_desc *desc)
166 {
167 u64 addr = xp_unaligned_add_offset_to_addr(desc->addr) - pool->tx_metadata_len;
168 u64 len = desc->len + pool->tx_metadata_len;
169
170 if (!desc->len)
171 return false;
172
173 if (len > pool->chunk_size)
174 return false;
175
176 if (addr >= pool->addrs_cnt || addr + len > pool->addrs_cnt ||
177 xp_desc_crosses_non_contig_pg(pool, addr, len))
178 return false;
179
180 if (xp_unused_options_set(desc->options))
181 return false;
182 return true;
183 }
184
185 static inline bool xp_validate_desc(struct xsk_buff_pool *pool,
186 struct xdp_desc *desc)
187 {
188 return pool->unaligned ? xp_unaligned_validate_desc(pool, desc) :
189 xp_aligned_validate_desc(pool, desc);
190 }
191
192 static inline bool xskq_has_descs(struct xsk_queue *q)
193 {
194 return q->cached_cons != q->cached_prod;
195 }
196
197 static inline bool xskq_cons_is_valid_desc(struct xsk_queue *q,
198 struct xdp_desc *d,
199 struct xsk_buff_pool *pool)
200 {
201 if (!xp_validate_desc(pool, d)) {
202 q->invalid_descs++;
203 return false;
204 }
205 return true;
206 }
207
208 static inline bool xskq_cons_read_desc(struct xsk_queue *q,
209 struct xdp_desc *desc,
210 struct xsk_buff_pool *pool)
211 {
212 if (q->cached_cons != q->cached_prod) {
213 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
214 u32 idx = q->cached_cons & q->ring_mask;
215
216 *desc = ring->desc[idx];
217 return xskq_cons_is_valid_desc(q, desc, pool);
218 }
219
220 q->queue_empty_descs++;
221 return false;
222 }
223
224 static inline void xskq_cons_release_n(struct xsk_queue *q, u32 cnt)
225 {
226 q->cached_cons += cnt;
227 }
228
229 static inline void parse_desc(struct xsk_queue *q, struct xsk_buff_pool *pool,
230 struct xdp_desc *desc, struct parsed_desc *parsed)
231 {
232 parsed->valid = xskq_cons_is_valid_desc(q, desc, pool);
233 parsed->mb = xp_mb_desc(desc);
234 }
235
236 static inline
237 u32 xskq_cons_read_desc_batch(struct xsk_queue *q, struct xsk_buff_pool *pool,
238 u32 max)
239 {
240 u32 cached_cons = q->cached_cons, nb_entries = 0;
241 struct xdp_desc *descs = pool->tx_descs;
242 u32 total_descs = 0, nr_frags = 0;
243
244 /* track first entry, if stumble upon *any* invalid descriptor, rewind
245 * current packet that consists of frags and stop the processing
246 */
247 while (cached_cons != q->cached_prod && nb_entries < max) {
248 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
249 u32 idx = cached_cons & q->ring_mask;
250 struct parsed_desc parsed;
251
252 descs[nb_entries] = ring->desc[idx];
253 cached_cons++;
254 parse_desc(q, pool, &descs[nb_entries], &parsed);
255 if (unlikely(!parsed.valid))
256 break;
257
258 if (likely(!parsed.mb)) {
259 total_descs += (nr_frags + 1);
260 nr_frags = 0;
261 } else {
262 nr_frags++;
263 if (nr_frags == pool->netdev->xdp_zc_max_segs) {
264 nr_frags = 0;
265 break;
266 }
267 }
268 nb_entries++;
269 }
270
271 cached_cons -= nr_frags;
272 /* Release valid plus any invalid entries */
273 xskq_cons_release_n(q, cached_cons - q->cached_cons);
274 return total_descs;
275 }
276
277 /* Functions for consumers */
278
279 static inline void __xskq_cons_release(struct xsk_queue *q)
280 {
281 smp_store_release(&q->ring->consumer, q->cached_cons); /* D, matchees A */
282 }
283
284 static inline void __xskq_cons_peek(struct xsk_queue *q)
285 {
286 /* Refresh the local pointer */
287 q->cached_prod = smp_load_acquire(&q->ring->producer); /* C, matches B */
288 }
289
290 static inline void xskq_cons_get_entries(struct xsk_queue *q)
291 {
292 __xskq_cons_release(q);
293 __xskq_cons_peek(q);
294 }
295
296 static inline u32 xskq_cons_nb_entries(struct xsk_queue *q, u32 max)
297 {
298 u32 entries = q->cached_prod - q->cached_cons;
299
300 if (entries >= max)
301 return max;
302
303 __xskq_cons_peek(q);
304 entries = q->cached_prod - q->cached_cons;
305
306 return entries >= max ? max : entries;
307 }
308
309 static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt)
310 {
311 return xskq_cons_nb_entries(q, cnt) >= cnt;
312 }
313
314 static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue *q, u64 *addr)
315 {
316 if (q->cached_prod == q->cached_cons)
317 xskq_cons_get_entries(q);
318 return xskq_cons_read_addr_unchecked(q, addr);
319 }
320
321 static inline bool xskq_cons_peek_desc(struct xsk_queue *q,
322 struct xdp_desc *desc,
323 struct xsk_buff_pool *pool)
324 {
325 if (q->cached_prod == q->cached_cons)
326 xskq_cons_get_entries(q);
327 return xskq_cons_read_desc(q, desc, pool);
328 }
329
330 /* To improve performance in the xskq_cons_release functions, only update local state here.
331 * Reflect this to global state when we get new entries from the ring in
332 * xskq_cons_get_entries() and whenever Rx or Tx processing are completed in the NAPI loop.
333 */
334 static inline void xskq_cons_release(struct xsk_queue *q)
335 {
336 q->cached_cons++;
337 }
338
339 static inline void xskq_cons_cancel_n(struct xsk_queue *q, u32 cnt)
340 {
341 q->cached_cons -= cnt;
342 }
343
344 static inline u32 xskq_cons_present_entries(struct xsk_queue *q)
345 {
346 /* No barriers needed since data is not accessed */
347 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer);
348 }
349
350 /* Functions for producers */
351
352 static inline u32 xskq_prod_nb_free(struct xsk_queue *q, u32 max)
353 {
354 u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
355
356 if (free_entries >= max)
357 return max;
358
359 /* Refresh the local tail pointer */
360 q->cached_cons = READ_ONCE(q->ring->consumer);
361 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
362
363 return free_entries >= max ? max : free_entries;
364 }
365
366 static inline bool xskq_prod_is_full(struct xsk_queue *q)
367 {
368 return xskq_prod_nb_free(q, 1) ? false : true;
369 }
370
371 static inline void xskq_prod_cancel_n(struct xsk_queue *q, u32 cnt)
372 {
373 q->cached_prod -= cnt;
374 }
375
376 static inline int xskq_prod_reserve(struct xsk_queue *q)
377 {
378 if (xskq_prod_is_full(q))
379 return -ENOSPC;
380
381 /* A, matches D */
382 q->cached_prod++;
383 return 0;
384 }
385
386 static inline int xskq_prod_reserve_addr(struct xsk_queue *q, u64 addr)
387 {
388 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
389
390 if (xskq_prod_is_full(q))
391 return -ENOSPC;
392
393 /* A, matches D */
394 ring->desc[q->cached_prod++ & q->ring_mask] = addr;
395 return 0;
396 }
397
398 static inline void xskq_prod_write_addr_batch(struct xsk_queue *q, struct xdp_desc *descs,
399 u32 nb_entries)
400 {
401 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
402 u32 i, cached_prod;
403
404 /* A, matches D */
405 cached_prod = q->cached_prod;
406 for (i = 0; i < nb_entries; i++)
407 ring->desc[cached_prod++ & q->ring_mask] = descs[i].addr;
408 q->cached_prod = cached_prod;
409 }
410
411 static inline int xskq_prod_reserve_desc(struct xsk_queue *q,
412 u64 addr, u32 len, u32 flags)
413 {
414 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
415 u32 idx;
416
417 if (xskq_prod_is_full(q))
418 return -ENOBUFS;
419
420 /* A, matches D */
421 idx = q->cached_prod++ & q->ring_mask;
422 ring->desc[idx].addr = addr;
423 ring->desc[idx].len = len;
424 ring->desc[idx].options = flags;
425
426 return 0;
427 }
428
429 static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx)
430 {
431 smp_store_release(&q->ring->producer, idx); /* B, matches C */
432 }
433
434 static inline void xskq_prod_submit(struct xsk_queue *q)
435 {
436 __xskq_prod_submit(q, q->cached_prod);
437 }
438
439 static inline void xskq_prod_submit_n(struct xsk_queue *q, u32 nb_entries)
440 {
441 __xskq_prod_submit(q, q->ring->producer + nb_entries);
442 }
443
444 static inline bool xskq_prod_is_empty(struct xsk_queue *q)
445 {
446 /* No barriers needed since data is not accessed */
447 return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
448 }
449
450 /* For both producers and consumers */
451
452 static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
453 {
454 return q ? q->invalid_descs : 0;
455 }
456
457 static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q)
458 {
459 return q ? q->queue_empty_descs : 0;
460 }
461
462 struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
463 void xskq_destroy(struct xsk_queue *q_ops);
464
465 #endif /* _LINUX_XSK_QUEUE_H */
466
Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.