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 };
49
50 /* The structure of the shared state of the rings are a simple
51 * circular buffer, as outlined in
52 * Documentation/core-api/circular-buffers.rst. For the Rx and
53 * completion ring, the kernel is the producer and user space is the
54 * consumer. For the Tx and fill rings, the kernel is the consumer and
55 * user space is the producer.
56 *
57 * producer consumer
58 *
59 * if (LOAD ->consumer) { (A) LOAD.acq ->producer (C)
60 * STORE $data LOAD $data
61 * STORE.rel ->producer (B) STORE.rel ->consumer (D)
62 * }
63 *
64 * (A) pairs with (D), and (B) pairs with (C).
65 *
66 * Starting with (B), it protects the data from being written after
67 * the producer pointer. If this barrier was missing, the consumer
68 * could observe the producer pointer being set and thus load the data
69 * before the producer has written the new data. The consumer would in
70 * this case load the old data.
71 *
72 * (C) protects the consumer from speculatively loading the data before
73 * the producer pointer actually has been read. If we do not have this
74 * barrier, some architectures could load old data as speculative loads
75 * are not discarded as the CPU does not know there is a dependency
76 * between ->producer and data.
77 *
78 * (A) is a control dependency that separates the load of ->consumer
79 * from the stores of $data. In case ->consumer indicates there is no
80 * room in the buffer to store $data we do not. The dependency will
81 * order both of the stores after the loads. So no barrier is needed.
82 *
83 * (D) protects the load of the data to be observed to happen after the
84 * store of the consumer pointer. If we did not have this memory
85 * barrier, the producer could observe the consumer pointer being set
86 * and overwrite the data with a new value before the consumer got the
87 * chance to read the old value. The consumer would thus miss reading
88 * the old entry and very likely read the new entry twice, once right
89 * now and again after circling through the ring.
90 */
91
92 /* The operations on the rings are the following:
93 *
94 * producer consumer
95 *
96 * RESERVE entries PEEK in the ring for entries
97 * WRITE data into the ring READ data from the ring
98 * SUBMIT entries RELEASE entries
99 *
100 * The producer reserves one or more entries in the ring. It can then
101 * fill in these entries and finally submit them so that they can be
102 * seen and read by the consumer.
103 *
104 * The consumer peeks into the ring to see if the producer has written
105 * any new entries. If so, the consumer can then read these entries
106 * and when it is done reading them release them back to the producer
107 * so that the producer can use these slots to fill in new entries.
108 *
109 * The function names below reflect these operations.
110 */
111
112 /* Functions that read and validate content from consumer rings. */
113
xskq_cons_read_addr_unchecked(struct xsk_queue * q,u64 * addr)114 static inline bool xskq_cons_read_addr_unchecked(struct xsk_queue *q, u64 *addr)
115 {
116 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
117
118 if (q->cached_cons != q->cached_prod) {
119 u32 idx = q->cached_cons & q->ring_mask;
120
121 *addr = ring->desc[idx];
122 return true;
123 }
124
125 return false;
126 }
127
xp_aligned_validate_desc(struct xsk_buff_pool * pool,struct xdp_desc * desc)128 static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
129 struct xdp_desc *desc)
130 {
131 u64 chunk, chunk_end;
132
133 chunk = xp_aligned_extract_addr(pool, desc->addr);
134 if (likely(desc->len)) {
135 chunk_end = xp_aligned_extract_addr(pool, desc->addr + desc->len - 1);
136 if (chunk != chunk_end)
137 return false;
138 }
139
140 if (chunk >= pool->addrs_cnt)
141 return false;
142
143 if (desc->options)
144 return false;
145 return true;
146 }
147
xp_unaligned_validate_desc(struct xsk_buff_pool * pool,struct xdp_desc * desc)148 static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool,
149 struct xdp_desc *desc)
150 {
151 u64 addr, base_addr;
152
153 base_addr = xp_unaligned_extract_addr(desc->addr);
154 addr = xp_unaligned_add_offset_to_addr(desc->addr);
155
156 if (desc->len > pool->chunk_size)
157 return false;
158
159 if (base_addr >= pool->addrs_cnt || addr >= pool->addrs_cnt ||
160 addr + desc->len > pool->addrs_cnt ||
161 xp_desc_crosses_non_contig_pg(pool, addr, desc->len))
162 return false;
163
164 if (desc->options)
165 return false;
166 return true;
167 }
168
xp_validate_desc(struct xsk_buff_pool * pool,struct xdp_desc * desc)169 static inline bool xp_validate_desc(struct xsk_buff_pool *pool,
170 struct xdp_desc *desc)
171 {
172 return pool->unaligned ? xp_unaligned_validate_desc(pool, desc) :
173 xp_aligned_validate_desc(pool, desc);
174 }
175
xskq_cons_is_valid_desc(struct xsk_queue * q,struct xdp_desc * d,struct xsk_buff_pool * pool)176 static inline bool xskq_cons_is_valid_desc(struct xsk_queue *q,
177 struct xdp_desc *d,
178 struct xsk_buff_pool *pool)
179 {
180 if (!xp_validate_desc(pool, d)) {
181 q->invalid_descs++;
182 return false;
183 }
184 return true;
185 }
186
xskq_cons_read_desc(struct xsk_queue * q,struct xdp_desc * desc,struct xsk_buff_pool * pool)187 static inline bool xskq_cons_read_desc(struct xsk_queue *q,
188 struct xdp_desc *desc,
189 struct xsk_buff_pool *pool)
190 {
191 while (q->cached_cons != q->cached_prod) {
192 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
193 u32 idx = q->cached_cons & q->ring_mask;
194
195 *desc = ring->desc[idx];
196 if (xskq_cons_is_valid_desc(q, desc, pool))
197 return true;
198
199 q->cached_cons++;
200 }
201
202 return false;
203 }
204
xskq_cons_release_n(struct xsk_queue * q,u32 cnt)205 static inline void xskq_cons_release_n(struct xsk_queue *q, u32 cnt)
206 {
207 q->cached_cons += cnt;
208 }
209
xskq_cons_read_desc_batch(struct xsk_queue * q,struct xsk_buff_pool * pool,u32 max)210 static inline u32 xskq_cons_read_desc_batch(struct xsk_queue *q, struct xsk_buff_pool *pool,
211 u32 max)
212 {
213 u32 cached_cons = q->cached_cons, nb_entries = 0;
214 struct xdp_desc *descs = pool->tx_descs;
215
216 while (cached_cons != q->cached_prod && nb_entries < max) {
217 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
218 u32 idx = cached_cons & q->ring_mask;
219
220 descs[nb_entries] = ring->desc[idx];
221 if (unlikely(!xskq_cons_is_valid_desc(q, &descs[nb_entries], pool))) {
222 /* Skip the entry */
223 cached_cons++;
224 continue;
225 }
226
227 nb_entries++;
228 cached_cons++;
229 }
230
231 /* Release valid plus any invalid entries */
232 xskq_cons_release_n(q, cached_cons - q->cached_cons);
233 return nb_entries;
234 }
235
236 /* Functions for consumers */
237
__xskq_cons_release(struct xsk_queue * q)238 static inline void __xskq_cons_release(struct xsk_queue *q)
239 {
240 smp_store_release(&q->ring->consumer, q->cached_cons); /* D, matchees A */
241 }
242
__xskq_cons_peek(struct xsk_queue * q)243 static inline void __xskq_cons_peek(struct xsk_queue *q)
244 {
245 /* Refresh the local pointer */
246 q->cached_prod = smp_load_acquire(&q->ring->producer); /* C, matches B */
247 }
248
xskq_cons_get_entries(struct xsk_queue * q)249 static inline void xskq_cons_get_entries(struct xsk_queue *q)
250 {
251 __xskq_cons_release(q);
252 __xskq_cons_peek(q);
253 }
254
xskq_cons_nb_entries(struct xsk_queue * q,u32 max)255 static inline u32 xskq_cons_nb_entries(struct xsk_queue *q, u32 max)
256 {
257 u32 entries = q->cached_prod - q->cached_cons;
258
259 if (entries >= max)
260 return max;
261
262 __xskq_cons_peek(q);
263 entries = q->cached_prod - q->cached_cons;
264
265 return entries >= max ? max : entries;
266 }
267
xskq_cons_has_entries(struct xsk_queue * q,u32 cnt)268 static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt)
269 {
270 return xskq_cons_nb_entries(q, cnt) >= cnt ? true : false;
271 }
272
xskq_cons_peek_addr_unchecked(struct xsk_queue * q,u64 * addr)273 static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue *q, u64 *addr)
274 {
275 if (q->cached_prod == q->cached_cons)
276 xskq_cons_get_entries(q);
277 return xskq_cons_read_addr_unchecked(q, addr);
278 }
279
xskq_cons_peek_desc(struct xsk_queue * q,struct xdp_desc * desc,struct xsk_buff_pool * pool)280 static inline bool xskq_cons_peek_desc(struct xsk_queue *q,
281 struct xdp_desc *desc,
282 struct xsk_buff_pool *pool)
283 {
284 if (q->cached_prod == q->cached_cons)
285 xskq_cons_get_entries(q);
286 return xskq_cons_read_desc(q, desc, pool);
287 }
288
289 /* To improve performance in the xskq_cons_release functions, only update local state here.
290 * Reflect this to global state when we get new entries from the ring in
291 * xskq_cons_get_entries() and whenever Rx or Tx processing are completed in the NAPI loop.
292 */
xskq_cons_release(struct xsk_queue * q)293 static inline void xskq_cons_release(struct xsk_queue *q)
294 {
295 q->cached_cons++;
296 }
297
xskq_cons_is_full(struct xsk_queue * q)298 static inline bool xskq_cons_is_full(struct xsk_queue *q)
299 {
300 /* No barriers needed since data is not accessed */
301 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer) ==
302 q->nentries;
303 }
304
xskq_cons_present_entries(struct xsk_queue * q)305 static inline u32 xskq_cons_present_entries(struct xsk_queue *q)
306 {
307 /* No barriers needed since data is not accessed */
308 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer);
309 }
310
311 /* Functions for producers */
312
xskq_prod_nb_free(struct xsk_queue * q,u32 max)313 static inline u32 xskq_prod_nb_free(struct xsk_queue *q, u32 max)
314 {
315 u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
316
317 if (free_entries >= max)
318 return max;
319
320 /* Refresh the local tail pointer */
321 q->cached_cons = READ_ONCE(q->ring->consumer);
322 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
323
324 return free_entries >= max ? max : free_entries;
325 }
326
xskq_prod_is_full(struct xsk_queue * q)327 static inline bool xskq_prod_is_full(struct xsk_queue *q)
328 {
329 return xskq_prod_nb_free(q, 1) ? false : true;
330 }
331
xskq_prod_cancel(struct xsk_queue * q)332 static inline void xskq_prod_cancel(struct xsk_queue *q)
333 {
334 q->cached_prod--;
335 }
336
xskq_prod_reserve(struct xsk_queue * q)337 static inline int xskq_prod_reserve(struct xsk_queue *q)
338 {
339 if (xskq_prod_is_full(q))
340 return -ENOSPC;
341
342 /* A, matches D */
343 q->cached_prod++;
344 return 0;
345 }
346
xskq_prod_reserve_addr(struct xsk_queue * q,u64 addr)347 static inline int xskq_prod_reserve_addr(struct xsk_queue *q, u64 addr)
348 {
349 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
350
351 if (xskq_prod_is_full(q))
352 return -ENOSPC;
353
354 /* A, matches D */
355 ring->desc[q->cached_prod++ & q->ring_mask] = addr;
356 return 0;
357 }
358
xskq_prod_write_addr_batch(struct xsk_queue * q,struct xdp_desc * descs,u32 nb_entries)359 static inline void xskq_prod_write_addr_batch(struct xsk_queue *q, struct xdp_desc *descs,
360 u32 nb_entries)
361 {
362 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
363 u32 i, cached_prod;
364
365 /* A, matches D */
366 cached_prod = q->cached_prod;
367 for (i = 0; i < nb_entries; i++)
368 ring->desc[cached_prod++ & q->ring_mask] = descs[i].addr;
369 q->cached_prod = cached_prod;
370 }
371
xskq_prod_reserve_desc(struct xsk_queue * q,u64 addr,u32 len)372 static inline int xskq_prod_reserve_desc(struct xsk_queue *q,
373 u64 addr, u32 len)
374 {
375 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
376 u32 idx;
377
378 if (xskq_prod_is_full(q))
379 return -ENOSPC;
380
381 /* A, matches D */
382 idx = q->cached_prod++ & q->ring_mask;
383 ring->desc[idx].addr = addr;
384 ring->desc[idx].len = len;
385
386 return 0;
387 }
388
__xskq_prod_submit(struct xsk_queue * q,u32 idx)389 static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx)
390 {
391 smp_store_release(&q->ring->producer, idx); /* B, matches C */
392 }
393
xskq_prod_submit(struct xsk_queue * q)394 static inline void xskq_prod_submit(struct xsk_queue *q)
395 {
396 __xskq_prod_submit(q, q->cached_prod);
397 }
398
xskq_prod_submit_addr(struct xsk_queue * q,u64 addr)399 static inline void xskq_prod_submit_addr(struct xsk_queue *q, u64 addr)
400 {
401 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
402 u32 idx = q->ring->producer;
403
404 ring->desc[idx++ & q->ring_mask] = addr;
405
406 __xskq_prod_submit(q, idx);
407 }
408
xskq_prod_submit_n(struct xsk_queue * q,u32 nb_entries)409 static inline void xskq_prod_submit_n(struct xsk_queue *q, u32 nb_entries)
410 {
411 __xskq_prod_submit(q, q->ring->producer + nb_entries);
412 }
413
xskq_prod_is_empty(struct xsk_queue * q)414 static inline bool xskq_prod_is_empty(struct xsk_queue *q)
415 {
416 /* No barriers needed since data is not accessed */
417 return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
418 }
419
420 /* For both producers and consumers */
421
xskq_nb_invalid_descs(struct xsk_queue * q)422 static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
423 {
424 return q ? q->invalid_descs : 0;
425 }
426
xskq_nb_queue_empty_descs(struct xsk_queue * q)427 static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q)
428 {
429 return q ? q->queue_empty_descs : 0;
430 }
431
432 struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
433 void xskq_destroy(struct xsk_queue *q_ops);
434
435 #endif /* _LINUX_XSK_QUEUE_H */
436