GNU Linux-libre 4.19.211-gnu1
[releases.git] / net / xdp / xsk_queue.h
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
13 #define RX_BATCH_SIZE 16
14 #define LAZY_UPDATE_THRESHOLD 128
15
16 struct xdp_ring {
17         u32 producer ____cacheline_aligned_in_smp;
18         u32 consumer ____cacheline_aligned_in_smp;
19 };
20
21 /* Used for the RX and TX queues for packets */
22 struct xdp_rxtx_ring {
23         struct xdp_ring ptrs;
24         struct xdp_desc desc[0] ____cacheline_aligned_in_smp;
25 };
26
27 /* Used for the fill and completion queues for buffers */
28 struct xdp_umem_ring {
29         struct xdp_ring ptrs;
30         u64 desc[0] ____cacheline_aligned_in_smp;
31 };
32
33 struct xsk_queue {
34         struct xdp_umem_props umem_props;
35         u32 ring_mask;
36         u32 nentries;
37         u32 prod_head;
38         u32 prod_tail;
39         u32 cons_head;
40         u32 cons_tail;
41         struct xdp_ring *ring;
42         u64 invalid_descs;
43 };
44
45 /* Common functions operating for both RXTX and umem queues */
46
47 static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
48 {
49         return q ? q->invalid_descs : 0;
50 }
51
52 static inline u32 xskq_nb_avail(struct xsk_queue *q, u32 dcnt)
53 {
54         u32 entries = q->prod_tail - q->cons_tail;
55
56         if (entries == 0) {
57                 /* Refresh the local pointer */
58                 q->prod_tail = READ_ONCE(q->ring->producer);
59                 entries = q->prod_tail - q->cons_tail;
60         }
61
62         return (entries > dcnt) ? dcnt : entries;
63 }
64
65 static inline u32 xskq_nb_free(struct xsk_queue *q, u32 producer, u32 dcnt)
66 {
67         u32 free_entries = q->nentries - (producer - q->cons_tail);
68
69         if (free_entries >= dcnt)
70                 return free_entries;
71
72         /* Refresh the local tail pointer */
73         q->cons_tail = READ_ONCE(q->ring->consumer);
74         return q->nentries - (producer - q->cons_tail);
75 }
76
77 /* UMEM queue */
78
79 static inline bool xskq_is_valid_addr(struct xsk_queue *q, u64 addr)
80 {
81         if (addr >= q->umem_props.size) {
82                 q->invalid_descs++;
83                 return false;
84         }
85
86         return true;
87 }
88
89 static inline u64 *xskq_validate_addr(struct xsk_queue *q, u64 *addr)
90 {
91         while (q->cons_tail != q->cons_head) {
92                 struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
93                 unsigned int idx = q->cons_tail & q->ring_mask;
94
95                 *addr = READ_ONCE(ring->desc[idx]) & q->umem_props.chunk_mask;
96                 if (xskq_is_valid_addr(q, *addr))
97                         return addr;
98
99                 q->cons_tail++;
100         }
101
102         return NULL;
103 }
104
105 static inline u64 *xskq_peek_addr(struct xsk_queue *q, u64 *addr)
106 {
107         if (q->cons_tail == q->cons_head) {
108                 WRITE_ONCE(q->ring->consumer, q->cons_tail);
109                 q->cons_head = q->cons_tail + xskq_nb_avail(q, RX_BATCH_SIZE);
110
111                 /* Order consumer and data */
112                 smp_rmb();
113         }
114
115         return xskq_validate_addr(q, addr);
116 }
117
118 static inline void xskq_discard_addr(struct xsk_queue *q)
119 {
120         q->cons_tail++;
121 }
122
123 static inline int xskq_produce_addr(struct xsk_queue *q, u64 addr)
124 {
125         struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
126
127         if (xskq_nb_free(q, q->prod_tail, 1) == 0)
128                 return -ENOSPC;
129
130         ring->desc[q->prod_tail++ & q->ring_mask] = addr;
131
132         /* Order producer and data */
133         smp_wmb();
134
135         WRITE_ONCE(q->ring->producer, q->prod_tail);
136         return 0;
137 }
138
139 static inline int xskq_produce_addr_lazy(struct xsk_queue *q, u64 addr)
140 {
141         struct xdp_umem_ring *ring = (struct xdp_umem_ring *)q->ring;
142
143         if (xskq_nb_free(q, q->prod_head, LAZY_UPDATE_THRESHOLD) == 0)
144                 return -ENOSPC;
145
146         ring->desc[q->prod_head++ & q->ring_mask] = addr;
147         return 0;
148 }
149
150 static inline void xskq_produce_flush_addr_n(struct xsk_queue *q,
151                                              u32 nb_entries)
152 {
153         /* Order producer and data */
154         smp_wmb();
155
156         q->prod_tail += nb_entries;
157         WRITE_ONCE(q->ring->producer, q->prod_tail);
158 }
159
160 static inline int xskq_reserve_addr(struct xsk_queue *q)
161 {
162         if (xskq_nb_free(q, q->prod_head, 1) == 0)
163                 return -ENOSPC;
164
165         q->prod_head++;
166         return 0;
167 }
168
169 /* Rx/Tx queue */
170
171 static inline bool xskq_is_valid_desc(struct xsk_queue *q, struct xdp_desc *d)
172 {
173         if (!xskq_is_valid_addr(q, d->addr))
174                 return false;
175
176         if (((d->addr + d->len) & q->umem_props.chunk_mask) !=
177             (d->addr & q->umem_props.chunk_mask)) {
178                 q->invalid_descs++;
179                 return false;
180         }
181
182         return true;
183 }
184
185 static inline struct xdp_desc *xskq_validate_desc(struct xsk_queue *q,
186                                                   struct xdp_desc *desc)
187 {
188         while (q->cons_tail != q->cons_head) {
189                 struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
190                 unsigned int idx = q->cons_tail & q->ring_mask;
191
192                 *desc = READ_ONCE(ring->desc[idx]);
193                 if (xskq_is_valid_desc(q, desc))
194                         return desc;
195
196                 q->cons_tail++;
197         }
198
199         return NULL;
200 }
201
202 static inline struct xdp_desc *xskq_peek_desc(struct xsk_queue *q,
203                                               struct xdp_desc *desc)
204 {
205         if (q->cons_tail == q->cons_head) {
206                 WRITE_ONCE(q->ring->consumer, q->cons_tail);
207                 q->cons_head = q->cons_tail + xskq_nb_avail(q, RX_BATCH_SIZE);
208
209                 /* Order consumer and data */
210                 smp_rmb();
211         }
212
213         return xskq_validate_desc(q, desc);
214 }
215
216 static inline void xskq_discard_desc(struct xsk_queue *q)
217 {
218         q->cons_tail++;
219 }
220
221 static inline int xskq_produce_batch_desc(struct xsk_queue *q,
222                                           u64 addr, u32 len)
223 {
224         struct xdp_rxtx_ring *ring = (struct xdp_rxtx_ring *)q->ring;
225         unsigned int idx;
226
227         if (xskq_nb_free(q, q->prod_head, 1) == 0)
228                 return -ENOSPC;
229
230         idx = (q->prod_head++) & q->ring_mask;
231         ring->desc[idx].addr = addr;
232         ring->desc[idx].len = len;
233
234         return 0;
235 }
236
237 static inline void xskq_produce_flush_desc(struct xsk_queue *q)
238 {
239         /* Order producer and data */
240         smp_wmb();
241
242         q->prod_tail = q->prod_head;
243         WRITE_ONCE(q->ring->producer, q->prod_tail);
244 }
245
246 static inline bool xskq_full_desc(struct xsk_queue *q)
247 {
248         /* No barriers needed since data is not accessed */
249         return READ_ONCE(q->ring->producer) - READ_ONCE(q->ring->consumer) ==
250                 q->nentries;
251 }
252
253 static inline bool xskq_empty_desc(struct xsk_queue *q)
254 {
255         /* No barriers needed since data is not accessed */
256         return READ_ONCE(q->ring->consumer) == READ_ONCE(q->ring->producer);
257 }
258
259 void xskq_set_umem(struct xsk_queue *q, struct xdp_umem_props *umem_props);
260 struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
261 void xskq_destroy(struct xsk_queue *q_ops);
262
263 #endif /* _LINUX_XSK_QUEUE_H */