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;
48 size_t ring_vmalloc_size;
51 /* The structure of the shared state of the rings are a simple
52 * circular buffer, as outlined in
53 * Documentation/core-api/circular-buffers.rst. For the Rx and
54 * completion ring, the kernel is the producer and user space is the
55 * consumer. For the Tx and fill rings, the kernel is the consumer and
56 * user space is the producer.
60 * if (LOAD ->consumer) { (A) LOAD.acq ->producer (C)
61 * STORE $data LOAD $data
62 * STORE.rel ->producer (B) STORE.rel ->consumer (D)
65 * (A) pairs with (D), and (B) pairs with (C).
67 * Starting with (B), it protects the data from being written after
68 * the producer pointer. If this barrier was missing, the consumer
69 * could observe the producer pointer being set and thus load the data
70 * before the producer has written the new data. The consumer would in
71 * this case load the old data.
73 * (C) protects the consumer from speculatively loading the data before
74 * the producer pointer actually has been read. If we do not have this
75 * barrier, some architectures could load old data as speculative loads
76 * are not discarded as the CPU does not know there is a dependency
77 * between ->producer and data.
79 * (A) is a control dependency that separates the load of ->consumer
80 * from the stores of $data. In case ->consumer indicates there is no
81 * room in the buffer to store $data we do not. The dependency will
82 * order both of the stores after the loads. So no barrier is needed.
84 * (D) protects the load of the data to be observed to happen after the
85 * store of the consumer pointer. If we did not have this memory
86 * barrier, the producer could observe the consumer pointer being set
87 * and overwrite the data with a new value before the consumer got the
88 * chance to read the old value. The consumer would thus miss reading
89 * the old entry and very likely read the new entry twice, once right
90 * now and again after circling through the ring.
93 /* The operations on the rings are the following:
97 * RESERVE entries PEEK in the ring for entries
98 * WRITE data into the ring READ data from the ring
99 * SUBMIT entries RELEASE entries
101 * The producer reserves one or more entries in the ring. It can then
102 * fill in these entries and finally submit them so that they can be
103 * seen and read by the consumer.
105 * The consumer peeks into the ring to see if the producer has written
106 * any new entries. If so, the consumer can then read these entries
107 * and when it is done reading them release them back to the producer
108 * so that the producer can use these slots to fill in new entries.
110 * The function names below reflect these operations.
113 /* Functions that read and validate content from consumer rings. */
115 static inline void __xskq_cons_read_addr_unchecked(struct xsk_queue *q, u32 cached_cons, u64 *addr)
117 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
118 u32 idx = cached_cons & q->ring_mask;
120 *addr = ring->desc[idx];
123 static inline bool xskq_cons_read_addr_unchecked(struct xsk_queue *q, u64 *addr)
125 if (q->cached_cons != q->cached_prod) {
126 __xskq_cons_read_addr_unchecked(q, q->cached_cons, addr);
133 static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
134 struct xdp_desc *desc)
136 u64 offset = desc->addr & (pool->chunk_size - 1);
138 if (offset + desc->len > pool->chunk_size)
141 if (desc->addr >= pool->addrs_cnt)
149 static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool,
150 struct xdp_desc *desc)
152 u64 addr = xp_unaligned_add_offset_to_addr(desc->addr);
154 if (desc->len > pool->chunk_size)
157 if (addr >= pool->addrs_cnt || addr + desc->len > pool->addrs_cnt ||
158 xp_desc_crosses_non_contig_pg(pool, addr, desc->len))
166 static inline bool xp_validate_desc(struct xsk_buff_pool *pool,
167 struct xdp_desc *desc)
169 return pool->unaligned ? xp_unaligned_validate_desc(pool, desc) :
170 xp_aligned_validate_desc(pool, desc);
173 static inline bool xskq_cons_is_valid_desc(struct xsk_queue *q,
175 struct xsk_buff_pool *pool)
177 if (!xp_validate_desc(pool, d)) {
184 static inline bool xskq_cons_read_desc(struct xsk_queue *q,
185 struct xdp_desc *desc,
186 struct xsk_buff_pool *pool)
188 while (q->cached_cons != q->cached_prod) {
189 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
190 u32 idx = q->cached_cons & q->ring_mask;
192 *desc = ring->desc[idx];
193 if (xskq_cons_is_valid_desc(q, desc, pool))
202 static inline void xskq_cons_release_n(struct xsk_queue *q, u32 cnt)
204 q->cached_cons += cnt;
207 static inline u32 xskq_cons_read_desc_batch(struct xsk_queue *q, struct xsk_buff_pool *pool,
210 u32 cached_cons = q->cached_cons, nb_entries = 0;
211 struct xdp_desc *descs = pool->tx_descs;
213 while (cached_cons != q->cached_prod && nb_entries < max) {
214 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
215 u32 idx = cached_cons & q->ring_mask;
217 descs[nb_entries] = ring->desc[idx];
218 if (unlikely(!xskq_cons_is_valid_desc(q, &descs[nb_entries], pool))) {
228 /* Release valid plus any invalid entries */
229 xskq_cons_release_n(q, cached_cons - q->cached_cons);
233 /* Functions for consumers */
235 static inline void __xskq_cons_release(struct xsk_queue *q)
237 smp_store_release(&q->ring->consumer, q->cached_cons); /* D, matchees A */
240 static inline void __xskq_cons_peek(struct xsk_queue *q)
242 /* Refresh the local pointer */
243 q->cached_prod = smp_load_acquire(&q->ring->producer); /* C, matches B */
246 static inline void xskq_cons_get_entries(struct xsk_queue *q)
248 __xskq_cons_release(q);
252 static inline u32 xskq_cons_nb_entries(struct xsk_queue *q, u32 max)
254 u32 entries = q->cached_prod - q->cached_cons;
260 entries = q->cached_prod - q->cached_cons;
262 return entries >= max ? max : entries;
265 static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt)
267 return xskq_cons_nb_entries(q, cnt) >= cnt;
270 static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue *q, u64 *addr)
272 if (q->cached_prod == q->cached_cons)
273 xskq_cons_get_entries(q);
274 return xskq_cons_read_addr_unchecked(q, addr);
277 static inline bool xskq_cons_peek_desc(struct xsk_queue *q,
278 struct xdp_desc *desc,
279 struct xsk_buff_pool *pool)
281 if (q->cached_prod == q->cached_cons)
282 xskq_cons_get_entries(q);
283 return xskq_cons_read_desc(q, desc, pool);
286 /* To improve performance in the xskq_cons_release functions, only update local state here.
287 * Reflect this to global state when we get new entries from the ring in
288 * xskq_cons_get_entries() and whenever Rx or Tx processing are completed in the NAPI loop.
290 static inline void xskq_cons_release(struct xsk_queue *q)
295 static inline u32 xskq_cons_present_entries(struct xsk_queue *q)
297 /* No barriers needed since data is not accessed */
298 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer);
301 /* Functions for producers */
303 static inline u32 xskq_prod_nb_free(struct xsk_queue *q, u32 max)
305 u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
307 if (free_entries >= max)
310 /* Refresh the local tail pointer */
311 q->cached_cons = READ_ONCE(q->ring->consumer);
312 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
314 return free_entries >= max ? max : free_entries;
317 static inline bool xskq_prod_is_full(struct xsk_queue *q)
319 return xskq_prod_nb_free(q, 1) ? false : true;
322 static inline void xskq_prod_cancel(struct xsk_queue *q)
327 static inline int xskq_prod_reserve(struct xsk_queue *q)
329 if (xskq_prod_is_full(q))
337 static inline int xskq_prod_reserve_addr(struct xsk_queue *q, u64 addr)
339 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
341 if (xskq_prod_is_full(q))
345 ring->desc[q->cached_prod++ & q->ring_mask] = addr;
349 static inline void xskq_prod_write_addr_batch(struct xsk_queue *q, struct xdp_desc *descs,
352 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
356 cached_prod = q->cached_prod;
357 for (i = 0; i < nb_entries; i++)
358 ring->desc[cached_prod++ & q->ring_mask] = descs[i].addr;
359 q->cached_prod = cached_prod;
362 static inline int xskq_prod_reserve_desc(struct xsk_queue *q,
365 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
368 if (xskq_prod_is_full(q))
372 idx = q->cached_prod++ & q->ring_mask;
373 ring->desc[idx].addr = addr;
374 ring->desc[idx].len = len;
379 static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx)
381 smp_store_release(&q->ring->producer, idx); /* B, matches C */
384 static inline void xskq_prod_submit(struct xsk_queue *q)
386 __xskq_prod_submit(q, q->cached_prod);
389 static inline void xskq_prod_submit_addr(struct xsk_queue *q, u64 addr)
391 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
392 u32 idx = q->ring->producer;
394 ring->desc[idx++ & q->ring_mask] = addr;
396 __xskq_prod_submit(q, idx);
399 static inline void xskq_prod_submit_n(struct xsk_queue *q, u32 nb_entries)
401 __xskq_prod_submit(q, q->ring->producer + nb_entries);
404 static inline bool xskq_prod_is_empty(struct xsk_queue *q)
406 /* No barriers needed since data is not accessed */
407 return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
410 /* For both producers and consumers */
412 static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
414 return q ? q->invalid_descs : 0;
417 static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q)
419 return q ? q->queue_empty_descs : 0;
422 struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
423 void xskq_destroy(struct xsk_queue *q_ops);
425 #endif /* _LINUX_XSK_QUEUE_H */