1 /* SPDX-License-Identifier: GPL-2.0 */
2 /* XDP user-space ring structure
3 * Copyright(c) 2018 Intel Corporation.
6 #ifndef _LINUX_XSK_QUEUE_H
7 #define _LINUX_XSK_QUEUE_H
9 #include <linux/types.h>
10 #include <linux/if_xdp.h>
11 #include <net/xdp_sock.h>
12 #include <net/xsk_buff_pool.h>
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.
21 u32 pad1 ____cacheline_aligned_in_smp;
22 u32 consumer ____cacheline_aligned_in_smp;
23 u32 pad2 ____cacheline_aligned_in_smp;
25 u32 pad3 ____cacheline_aligned_in_smp;
28 /* Used for the RX and TX queues for packets */
29 struct xdp_rxtx_ring {
31 struct xdp_desc desc[] ____cacheline_aligned_in_smp;
34 /* Used for the fill and completion queues for buffers */
35 struct xdp_umem_ring {
37 u64 desc[] ____cacheline_aligned_in_smp;
45 struct xdp_ring *ring;
47 u64 queue_empty_descs;
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.
59 * if (LOAD ->consumer) { (A) LOAD.acq ->producer (C)
60 * STORE $data LOAD $data
61 * STORE.rel ->producer (B) STORE.rel ->consumer (D)
64 * (A) pairs with (D), and (B) pairs with (C).
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.
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.
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.
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.
92 /* The operations on the rings are the following:
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
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.
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.
109 * The function names below reflect these operations.
112 /* Functions that read and validate content from consumer rings. */
114 static inline bool xskq_cons_read_addr_unchecked(struct xsk_queue *q, u64 *addr)
116 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
118 if (q->cached_cons != q->cached_prod) {
119 u32 idx = q->cached_cons & q->ring_mask;
121 *addr = ring->desc[idx];
128 static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
129 struct xdp_desc *desc)
131 u64 chunk, chunk_end;
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)
140 if (chunk >= pool->addrs_cnt)
148 static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool,
149 struct xdp_desc *desc)
153 base_addr = xp_unaligned_extract_addr(desc->addr);
154 addr = xp_unaligned_add_offset_to_addr(desc->addr);
156 if (desc->len > pool->chunk_size)
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))
169 static inline bool xp_validate_desc(struct xsk_buff_pool *pool,
170 struct xdp_desc *desc)
172 return pool->unaligned ? xp_unaligned_validate_desc(pool, desc) :
173 xp_aligned_validate_desc(pool, desc);
176 static inline bool xskq_cons_is_valid_desc(struct xsk_queue *q,
178 struct xsk_buff_pool *pool)
180 if (!xp_validate_desc(pool, d)) {
187 static inline bool xskq_cons_read_desc(struct xsk_queue *q,
188 struct xdp_desc *desc,
189 struct xsk_buff_pool *pool)
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;
195 *desc = ring->desc[idx];
196 if (xskq_cons_is_valid_desc(q, desc, pool))
205 static inline void xskq_cons_release_n(struct xsk_queue *q, u32 cnt)
207 q->cached_cons += cnt;
210 static inline u32 xskq_cons_read_desc_batch(struct xsk_queue *q, struct xsk_buff_pool *pool,
213 u32 cached_cons = q->cached_cons, nb_entries = 0;
214 struct xdp_desc *descs = pool->tx_descs;
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;
220 descs[nb_entries] = ring->desc[idx];
221 if (unlikely(!xskq_cons_is_valid_desc(q, &descs[nb_entries], pool))) {
231 /* Release valid plus any invalid entries */
232 xskq_cons_release_n(q, cached_cons - q->cached_cons);
236 /* Functions for consumers */
238 static inline void __xskq_cons_release(struct xsk_queue *q)
240 smp_store_release(&q->ring->consumer, q->cached_cons); /* D, matchees A */
243 static inline void __xskq_cons_peek(struct xsk_queue *q)
245 /* Refresh the local pointer */
246 q->cached_prod = smp_load_acquire(&q->ring->producer); /* C, matches B */
249 static inline void xskq_cons_get_entries(struct xsk_queue *q)
251 __xskq_cons_release(q);
255 static inline u32 xskq_cons_nb_entries(struct xsk_queue *q, u32 max)
257 u32 entries = q->cached_prod - q->cached_cons;
263 entries = q->cached_prod - q->cached_cons;
265 return entries >= max ? max : entries;
268 static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt)
270 return xskq_cons_nb_entries(q, cnt) >= cnt ? true : false;
273 static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue *q, u64 *addr)
275 if (q->cached_prod == q->cached_cons)
276 xskq_cons_get_entries(q);
277 return xskq_cons_read_addr_unchecked(q, addr);
280 static inline bool xskq_cons_peek_desc(struct xsk_queue *q,
281 struct xdp_desc *desc,
282 struct xsk_buff_pool *pool)
284 if (q->cached_prod == q->cached_cons)
285 xskq_cons_get_entries(q);
286 return xskq_cons_read_desc(q, desc, pool);
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.
293 static inline void xskq_cons_release(struct xsk_queue *q)
298 static inline bool xskq_cons_is_full(struct xsk_queue *q)
300 /* No barriers needed since data is not accessed */
301 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer) ==
305 static inline u32 xskq_cons_present_entries(struct xsk_queue *q)
307 /* No barriers needed since data is not accessed */
308 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer);
311 /* Functions for producers */
313 static inline u32 xskq_prod_nb_free(struct xsk_queue *q, u32 max)
315 u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
317 if (free_entries >= max)
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);
324 return free_entries >= max ? max : free_entries;
327 static inline bool xskq_prod_is_full(struct xsk_queue *q)
329 return xskq_prod_nb_free(q, 1) ? false : true;
332 static inline void xskq_prod_cancel(struct xsk_queue *q)
337 static inline int xskq_prod_reserve(struct xsk_queue *q)
339 if (xskq_prod_is_full(q))
347 static inline int xskq_prod_reserve_addr(struct xsk_queue *q, u64 addr)
349 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
351 if (xskq_prod_is_full(q))
355 ring->desc[q->cached_prod++ & q->ring_mask] = addr;
359 static inline void xskq_prod_write_addr_batch(struct xsk_queue *q, struct xdp_desc *descs,
362 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
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;
372 static inline int xskq_prod_reserve_desc(struct xsk_queue *q,
375 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
378 if (xskq_prod_is_full(q))
382 idx = q->cached_prod++ & q->ring_mask;
383 ring->desc[idx].addr = addr;
384 ring->desc[idx].len = len;
389 static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx)
391 smp_store_release(&q->ring->producer, idx); /* B, matches C */
394 static inline void xskq_prod_submit(struct xsk_queue *q)
396 __xskq_prod_submit(q, q->cached_prod);
399 static inline void xskq_prod_submit_addr(struct xsk_queue *q, u64 addr)
401 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
402 u32 idx = q->ring->producer;
404 ring->desc[idx++ & q->ring_mask] = addr;
406 __xskq_prod_submit(q, idx);
409 static inline void xskq_prod_submit_n(struct xsk_queue *q, u32 nb_entries)
411 __xskq_prod_submit(q, q->ring->producer + nb_entries);
414 static inline bool xskq_prod_is_empty(struct xsk_queue *q)
416 /* No barriers needed since data is not accessed */
417 return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
420 /* For both producers and consumers */
422 static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
424 return q ? q->invalid_descs : 0;
427 static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q)
429 return q ? q->queue_empty_descs : 0;
432 struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
433 void xskq_destroy(struct xsk_queue *q_ops);
435 #endif /* _LINUX_XSK_QUEUE_H */