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 xp_desc_crosses_non_contig_pg(pool
, addr
, desc
->len
))
168 static inline bool xp_validate_desc(struct xsk_buff_pool
*pool
,
169 struct xdp_desc
*desc
)
171 return pool
->unaligned
? xp_unaligned_validate_desc(pool
, desc
) :
172 xp_aligned_validate_desc(pool
, desc
);
175 static inline bool xskq_cons_is_valid_desc(struct xsk_queue
*q
,
177 struct xsk_buff_pool
*pool
)
179 if (!xp_validate_desc(pool
, d
)) {
186 static inline bool xskq_cons_read_desc(struct xsk_queue
*q
,
187 struct xdp_desc
*desc
,
188 struct xsk_buff_pool
*pool
)
190 while (q
->cached_cons
!= q
->cached_prod
) {
191 struct xdp_rxtx_ring
*ring
= (struct xdp_rxtx_ring
*)q
->ring
;
192 u32 idx
= q
->cached_cons
& q
->ring_mask
;
194 *desc
= ring
->desc
[idx
];
195 if (xskq_cons_is_valid_desc(q
, desc
, pool
))
204 static inline u32
xskq_cons_read_desc_batch(struct xsk_queue
*q
,
205 struct xdp_desc
*descs
,
206 struct xsk_buff_pool
*pool
, u32 max
)
208 u32 cached_cons
= q
->cached_cons
, nb_entries
= 0;
210 while (cached_cons
!= q
->cached_prod
&& nb_entries
< max
) {
211 struct xdp_rxtx_ring
*ring
= (struct xdp_rxtx_ring
*)q
->ring
;
212 u32 idx
= cached_cons
& q
->ring_mask
;
214 descs
[nb_entries
] = ring
->desc
[idx
];
215 if (unlikely(!xskq_cons_is_valid_desc(q
, &descs
[nb_entries
], pool
))) {
228 /* Functions for consumers */
230 static inline void __xskq_cons_release(struct xsk_queue
*q
)
232 smp_store_release(&q
->ring
->consumer
, q
->cached_cons
); /* D, matchees A */
235 static inline void __xskq_cons_peek(struct xsk_queue
*q
)
237 /* Refresh the local pointer */
238 q
->cached_prod
= smp_load_acquire(&q
->ring
->producer
); /* C, matches B */
241 static inline void xskq_cons_get_entries(struct xsk_queue
*q
)
243 __xskq_cons_release(q
);
247 static inline u32
xskq_cons_nb_entries(struct xsk_queue
*q
, u32 max
)
249 u32 entries
= q
->cached_prod
- q
->cached_cons
;
255 entries
= q
->cached_prod
- q
->cached_cons
;
257 return entries
>= max
? max
: entries
;
260 static inline bool xskq_cons_has_entries(struct xsk_queue
*q
, u32 cnt
)
262 return xskq_cons_nb_entries(q
, cnt
) >= cnt
? true : false;
265 static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue
*q
, u64
*addr
)
267 if (q
->cached_prod
== q
->cached_cons
)
268 xskq_cons_get_entries(q
);
269 return xskq_cons_read_addr_unchecked(q
, addr
);
272 static inline bool xskq_cons_peek_desc(struct xsk_queue
*q
,
273 struct xdp_desc
*desc
,
274 struct xsk_buff_pool
*pool
)
276 if (q
->cached_prod
== q
->cached_cons
)
277 xskq_cons_get_entries(q
);
278 return xskq_cons_read_desc(q
, desc
, pool
);
281 static inline u32
xskq_cons_peek_desc_batch(struct xsk_queue
*q
, struct xdp_desc
*descs
,
282 struct xsk_buff_pool
*pool
, u32 max
)
284 u32 entries
= xskq_cons_nb_entries(q
, max
);
286 return xskq_cons_read_desc_batch(q
, descs
, pool
, entries
);
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 void xskq_cons_release_n(struct xsk_queue
*q
, u32 cnt
)
300 q
->cached_cons
+= cnt
;
303 static inline bool xskq_cons_is_full(struct xsk_queue
*q
)
305 /* No barriers needed since data is not accessed */
306 return READ_ONCE(q
->ring
->producer
) - READ_ONCE(q
->ring
->consumer
) ==
310 static inline u32
xskq_cons_present_entries(struct xsk_queue
*q
)
312 /* No barriers needed since data is not accessed */
313 return READ_ONCE(q
->ring
->producer
) - READ_ONCE(q
->ring
->consumer
);
316 /* Functions for producers */
318 static inline u32
xskq_prod_nb_free(struct xsk_queue
*q
, u32 max
)
320 u32 free_entries
= q
->nentries
- (q
->cached_prod
- q
->cached_cons
);
322 if (free_entries
>= max
)
325 /* Refresh the local tail pointer */
326 q
->cached_cons
= READ_ONCE(q
->ring
->consumer
);
327 free_entries
= q
->nentries
- (q
->cached_prod
- q
->cached_cons
);
329 return free_entries
>= max
? max
: free_entries
;
332 static inline bool xskq_prod_is_full(struct xsk_queue
*q
)
334 return xskq_prod_nb_free(q
, 1) ? false : true;
337 static inline void xskq_prod_cancel(struct xsk_queue
*q
)
342 static inline int xskq_prod_reserve(struct xsk_queue
*q
)
344 if (xskq_prod_is_full(q
))
352 static inline int xskq_prod_reserve_addr(struct xsk_queue
*q
, u64 addr
)
354 struct xdp_umem_ring
*ring
= (struct xdp_umem_ring
*)q
->ring
;
356 if (xskq_prod_is_full(q
))
360 ring
->desc
[q
->cached_prod
++ & q
->ring_mask
] = addr
;
364 static inline u32
xskq_prod_reserve_addr_batch(struct xsk_queue
*q
, struct xdp_desc
*descs
,
367 struct xdp_umem_ring
*ring
= (struct xdp_umem_ring
*)q
->ring
;
368 u32 nb_entries
, i
, cached_prod
;
370 nb_entries
= xskq_prod_nb_free(q
, max
);
373 cached_prod
= q
->cached_prod
;
374 for (i
= 0; i
< nb_entries
; i
++)
375 ring
->desc
[cached_prod
++ & q
->ring_mask
] = descs
[i
].addr
;
376 q
->cached_prod
= cached_prod
;
381 static inline int xskq_prod_reserve_desc(struct xsk_queue
*q
,
384 struct xdp_rxtx_ring
*ring
= (struct xdp_rxtx_ring
*)q
->ring
;
387 if (xskq_prod_is_full(q
))
391 idx
= q
->cached_prod
++ & q
->ring_mask
;
392 ring
->desc
[idx
].addr
= addr
;
393 ring
->desc
[idx
].len
= len
;
398 static inline void __xskq_prod_submit(struct xsk_queue
*q
, u32 idx
)
400 smp_store_release(&q
->ring
->producer
, idx
); /* B, matches C */
403 static inline void xskq_prod_submit(struct xsk_queue
*q
)
405 __xskq_prod_submit(q
, q
->cached_prod
);
408 static inline void xskq_prod_submit_addr(struct xsk_queue
*q
, u64 addr
)
410 struct xdp_umem_ring
*ring
= (struct xdp_umem_ring
*)q
->ring
;
411 u32 idx
= q
->ring
->producer
;
413 ring
->desc
[idx
++ & q
->ring_mask
] = addr
;
415 __xskq_prod_submit(q
, idx
);
418 static inline void xskq_prod_submit_n(struct xsk_queue
*q
, u32 nb_entries
)
420 __xskq_prod_submit(q
, q
->ring
->producer
+ nb_entries
);
423 static inline bool xskq_prod_is_empty(struct xsk_queue
*q
)
425 /* No barriers needed since data is not accessed */
426 return READ_ONCE(q
->ring
->consumer
) == READ_ONCE(q
->ring
->producer
);
429 /* For both producers and consumers */
431 static inline u64
xskq_nb_invalid_descs(struct xsk_queue
*q
)
433 return q
? q
->invalid_descs
: 0;
436 static inline u64
xskq_nb_queue_empty_descs(struct xsk_queue
*q
)
438 return q
? q
->queue_empty_descs
: 0;
441 struct xsk_queue
*xskq_create(u32 nentries
, bool umem_queue
);
442 void xskq_destroy(struct xsk_queue
*q_ops
);
444 #endif /* _LINUX_XSK_QUEUE_H */