1
2
3
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
16struct xdp_ring {
17 u32 producer ____cacheline_aligned_in_smp;
18
19
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
29struct xdp_rxtx_ring {
30 struct xdp_ring ptrs;
31 struct xdp_desc desc[] ____cacheline_aligned_in_smp;
32};
33
34
35struct xdp_umem_ring {
36 struct xdp_ring ptrs;
37 u64 desc[] ____cacheline_aligned_in_smp;
38};
39
40struct 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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114static 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
128static 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
148static 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 xp_desc_crosses_non_contig_pg(pool, addr, desc->len))
161 return false;
162
163 if (desc->options)
164 return false;
165 return true;
166}
167
168static inline bool xp_validate_desc(struct xsk_buff_pool *pool,
169 struct xdp_desc *desc)
170{
171 return pool->unaligned ? xp_unaligned_validate_desc(pool, desc) :
172 xp_aligned_validate_desc(pool, desc);
173}
174
175static inline bool xskq_cons_is_valid_desc(struct xsk_queue *q,
176 struct xdp_desc *d,
177 struct xsk_buff_pool *pool)
178{
179 if (!xp_validate_desc(pool, d)) {
180 q->invalid_descs++;
181 return false;
182 }
183 return true;
184}
185
186static inline bool xskq_cons_read_desc(struct xsk_queue *q,
187 struct xdp_desc *desc,
188 struct xsk_buff_pool *pool)
189{
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;
193
194 *desc = ring->desc[idx];
195 if (xskq_cons_is_valid_desc(q, desc, pool))
196 return true;
197
198 q->cached_cons++;
199 }
200
201 return false;
202}
203
204static inline u32 xskq_cons_read_desc_batch(struct xsk_queue *q,
205 struct xdp_desc *descs,
206 struct xsk_buff_pool *pool, u32 max)
207{
208 u32 cached_cons = q->cached_cons, nb_entries = 0;
209
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;
213
214 descs[nb_entries] = ring->desc[idx];
215 if (unlikely(!xskq_cons_is_valid_desc(q, &descs[nb_entries], pool))) {
216
217 cached_cons++;
218 continue;
219 }
220
221 nb_entries++;
222 cached_cons++;
223 }
224
225 return nb_entries;
226}
227
228
229
230static inline void __xskq_cons_release(struct xsk_queue *q)
231{
232 smp_store_release(&q->ring->consumer, q->cached_cons);
233}
234
235static inline void __xskq_cons_peek(struct xsk_queue *q)
236{
237
238 q->cached_prod = smp_load_acquire(&q->ring->producer);
239}
240
241static inline void xskq_cons_get_entries(struct xsk_queue *q)
242{
243 __xskq_cons_release(q);
244 __xskq_cons_peek(q);
245}
246
247static inline u32 xskq_cons_nb_entries(struct xsk_queue *q, u32 max)
248{
249 u32 entries = q->cached_prod - q->cached_cons;
250
251 if (entries >= max)
252 return max;
253
254 __xskq_cons_peek(q);
255 entries = q->cached_prod - q->cached_cons;
256
257 return entries >= max ? max : entries;
258}
259
260static inline bool xskq_cons_has_entries(struct xsk_queue *q, u32 cnt)
261{
262 return xskq_cons_nb_entries(q, cnt) >= cnt ? true : false;
263}
264
265static inline bool xskq_cons_peek_addr_unchecked(struct xsk_queue *q, u64 *addr)
266{
267 if (q->cached_prod == q->cached_cons)
268 xskq_cons_get_entries(q);
269 return xskq_cons_read_addr_unchecked(q, addr);
270}
271
272static inline bool xskq_cons_peek_desc(struct xsk_queue *q,
273 struct xdp_desc *desc,
274 struct xsk_buff_pool *pool)
275{
276 if (q->cached_prod == q->cached_cons)
277 xskq_cons_get_entries(q);
278 return xskq_cons_read_desc(q, desc, pool);
279}
280
281static inline u32 xskq_cons_peek_desc_batch(struct xsk_queue *q, struct xdp_desc *descs,
282 struct xsk_buff_pool *pool, u32 max)
283{
284 u32 entries = xskq_cons_nb_entries(q, max);
285
286 return xskq_cons_read_desc_batch(q, descs, pool, entries);
287}
288
289
290
291
292
293static inline void xskq_cons_release(struct xsk_queue *q)
294{
295 q->cached_cons++;
296}
297
298static inline void xskq_cons_release_n(struct xsk_queue *q, u32 cnt)
299{
300 q->cached_cons += cnt;
301}
302
303static inline bool xskq_cons_is_full(struct xsk_queue *q)
304{
305
306 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer) ==
307 q->nentries;
308}
309
310static inline u32 xskq_cons_present_entries(struct xsk_queue *q)
311{
312
313 return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer);
314}
315
316
317
318static inline u32 xskq_prod_nb_free(struct xsk_queue *q, u32 max)
319{
320 u32 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
321
322 if (free_entries >= max)
323 return max;
324
325
326 q->cached_cons = READ_ONCE(q->ring->consumer);
327 free_entries = q->nentries - (q->cached_prod - q->cached_cons);
328
329 return free_entries >= max ? max : free_entries;
330}
331
332static inline bool xskq_prod_is_full(struct xsk_queue *q)
333{
334 return xskq_prod_nb_free(q, 1) ? false : true;
335}
336
337static inline void xskq_prod_cancel(struct xsk_queue *q)
338{
339 q->cached_prod--;
340}
341
342static inline int xskq_prod_reserve(struct xsk_queue *q)
343{
344 if (xskq_prod_is_full(q))
345 return -ENOSPC;
346
347
348 q->cached_prod++;
349 return 0;
350}
351
352static inline int xskq_prod_reserve_addr(struct xsk_queue *q, u64 addr)
353{
354 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
355
356 if (xskq_prod_is_full(q))
357 return -ENOSPC;
358
359
360 ring->desc[q->cached_prod++ & q->ring_mask] = addr;
361 return 0;
362}
363
364static inline u32 xskq_prod_reserve_addr_batch(struct xsk_queue *q, struct xdp_desc *descs,
365 u32 max)
366{
367 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
368 u32 nb_entries, i, cached_prod;
369
370 nb_entries = xskq_prod_nb_free(q, max);
371
372
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;
377
378 return nb_entries;
379}
380
381static inline int xskq_prod_reserve_desc(struct xsk_queue *q,
382 u64 addr, u32 len)
383{
384 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
385 u32 idx;
386
387 if (xskq_prod_is_full(q))
388 return -ENOSPC;
389
390
391 idx = q->cached_prod++ & q->ring_mask;
392 ring->desc[idx].addr = addr;
393 ring->desc[idx].len = len;
394
395 return 0;
396}
397
398static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx)
399{
400 smp_store_release(&q->ring->producer, idx);
401}
402
403static inline void xskq_prod_submit(struct xsk_queue *q)
404{
405 __xskq_prod_submit(q, q->cached_prod);
406}
407
408static inline void xskq_prod_submit_addr(struct xsk_queue *q, u64 addr)
409{
410 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
411 u32 idx = q->ring->producer;
412
413 ring->desc[idx++ & q->ring_mask] = addr;
414
415 __xskq_prod_submit(q, idx);
416}
417
418static inline void xskq_prod_submit_n(struct xsk_queue *q, u32 nb_entries)
419{
420 __xskq_prod_submit(q, q->ring->producer + nb_entries);
421}
422
423static inline bool xskq_prod_is_empty(struct xsk_queue *q)
424{
425
426 return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
427}
428
429
430
431static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
432{
433 return q ? q->invalid_descs : 0;
434}
435
436static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q)
437{
438 return q ? q->queue_empty_descs : 0;
439}
440
441struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
442void xskq_destroy(struct xsk_queue *q_ops);
443
444#endif
445