1 /*-
2 * SPDX-License-Identifier: BSD-2-Clause
3 *
4 * Copyright (c) 2007-2009 Kip Macy <kmacy@freebsd.org>
5 * All rights reserved.
6 * Copyright (c) 2024 Arm Ltd
7 *
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions
10 * are met:
11 * 1. Redistributions of source code must retain the above copyright
12 * notice, this list of conditions and the following disclaimer.
13 * 2. Redistributions in binary form must reproduce the above copyright
14 * notice, this list of conditions and the following disclaimer in the
15 * documentation and/or other materials provided with the distribution.
16 *
17 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
18 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20 * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
21 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
22 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
23 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
24 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
26 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
27 * SUCH DAMAGE.
28 *
29 */
30
31 #ifndef _SYS_BUF_RING_H_
32 #define _SYS_BUF_RING_H_
33
34 #include <sys/param.h>
35 #include <sys/kassert.h>
36 #include <machine/atomic.h>
37 #include <machine/cpu.h>
38
39 #if defined(_KERNEL)
40 #include <sys/lock.h>
41 #include <sys/counter.h>
42 #define BR_LOCK_ASSERT(br) \
43 LOCK_CLASS((br)->br_lock)->lc_assert((br)->br_lock, LA_XLOCKED)
44 #else
45 #define BR_LOCK_ASSERT(br) do {} while (0)
46 typedef uint64_t counter_u64_t;
47 #define counter_u64_add(c,v) do { (c) += (v); } while (0)
48 #endif
49
50 /*
51 * We only apply the mask to the head and tail values when calculating the
52 * index into br_ring to access. This means the upper bits can be used as
53 * epoch to reduce the chance the atomic_cmpset succeedes when it should
54 * fail, e.g. when the head wraps while the CPU is in an interrupt. This
55 * is a probablistic fix as there is still a very unlikely chance the
56 * value wraps back to the expected value.
57 *
58 */
59 struct buf_ring {
60 uint32_t br_prod_head;
61 uint32_t br_prod_tail;
62 int br_prod_size;
63 int br_prod_mask;
64 counter_u64_t br_drops;
65 uint32_t br_cons_head __aligned(CACHE_LINE_SIZE);
66 uint32_t br_cons_tail;
67 int br_cons_size;
68 int br_cons_mask;
69 struct lock_object *br_lock;
70 void *br_ring[0] __aligned(CACHE_LINE_SIZE);
71 };
72
73 /*
74 * multi-producer safe lock-free ring buffer enqueue
75 *
76 */
77 static __inline int
buf_ring_enqueue(struct buf_ring * br,void * buf)78 buf_ring_enqueue(struct buf_ring *br, void *buf)
79 {
80 uint32_t prod_head, prod_next, prod_idx;
81 uint32_t cons_tail, mask;
82
83 mask = br->br_prod_mask;
84 #ifdef DEBUG_BUFRING
85 /*
86 * Note: It is possible to encounter an mbuf that was removed
87 * via drbr_peek(), and then re-added via drbr_putback() and
88 * trigger a spurious panic.
89 */
90 for (uint32_t i = atomic_load_32(&br->br_cons_head);
91 i != atomic_load_32(&br->br_prod_head); i++)
92 if (br->br_ring[i & mask] == buf)
93 panic("buf=%p already enqueue at %d prod=%d cons=%d",
94 buf, i, atomic_load_32(&br->br_prod_tail),
95 atomic_load_32(&br->br_cons_tail));
96 #endif
97 critical_enter();
98 do {
99 /*
100 * br->br_prod_head needs to be read before br->br_cons_tail.
101 * If not then we could perform the dequeue and enqueue
102 * between reading br_cons_tail and reading br_prod_head. This
103 * could give us values where br_cons_head == br_prod_tail
104 * (after masking).
105 *
106 * To work around this us a load acquire. This is just to
107 * ensure ordering within this thread.
108 */
109 prod_head = atomic_load_acq_32(&br->br_prod_head);
110 prod_next = prod_head + 1;
111 cons_tail = atomic_load_acq_32(&br->br_cons_tail);
112
113 if ((int32_t)(cons_tail + br->br_prod_size - prod_next) < 1) {
114 if (prod_head == atomic_load_32(&br->br_prod_head) &&
115 cons_tail == atomic_load_32(&br->br_cons_tail)) {
116 critical_exit();
117 counter_u64_add(br->br_drops, 1);
118 return (ENOBUFS);
119 }
120 continue;
121 }
122 } while (!atomic_cmpset_32(&br->br_prod_head, prod_head, prod_next));
123 prod_idx = prod_head & mask;
124 #ifdef DEBUG_BUFRING
125 if (br->br_ring[prod_idx] != NULL)
126 panic("dangling value in enqueue");
127 #endif
128 br->br_ring[prod_idx] = buf;
129
130 /*
131 * If there are other enqueues in progress
132 * that preceded us, we need to wait for them
133 * to complete
134 */
135 while (atomic_load_32(&br->br_prod_tail) != prod_head)
136 cpu_spinwait();
137 atomic_store_rel_32(&br->br_prod_tail, prod_next);
138 critical_exit();
139 return (0);
140 }
141
142 /*
143 * multi-consumer safe dequeue
144 *
145 */
146 static __inline void *
buf_ring_dequeue_mc(struct buf_ring * br)147 buf_ring_dequeue_mc(struct buf_ring *br)
148 {
149 uint32_t cons_head, cons_next, cons_idx;
150 uint32_t prod_tail, mask;
151 void *buf;
152
153 critical_enter();
154 mask = br->br_cons_mask;
155 do {
156 /*
157 * As with buf_ring_enqueue ensure we read the head before
158 * the tail. If we read them in the wrong order we may
159 * think the bug_ring is full when it is empty.
160 */
161 cons_head = atomic_load_acq_32(&br->br_cons_head);
162 cons_next = cons_head + 1;
163 prod_tail = atomic_load_acq_32(&br->br_prod_tail);
164
165 if (cons_head == prod_tail) {
166 critical_exit();
167 return (NULL);
168 }
169 } while (!atomic_cmpset_32(&br->br_cons_head, cons_head, cons_next));
170 cons_idx = cons_head & mask;
171
172 buf = br->br_ring[cons_idx];
173 #ifdef DEBUG_BUFRING
174 br->br_ring[cons_idx] = NULL;
175 #endif
176 /*
177 * If there are other dequeues in progress
178 * that preceded us, we need to wait for them
179 * to complete
180 */
181 while (atomic_load_32(&br->br_cons_tail) != cons_head)
182 cpu_spinwait();
183
184 atomic_store_rel_32(&br->br_cons_tail, cons_next);
185 critical_exit();
186
187 return (buf);
188 }
189
190 /*
191 * single-consumer dequeue
192 * use where dequeue is protected by a lock
193 * e.g. a network driver's tx queue lock
194 */
195 static __inline void *
buf_ring_dequeue_sc(struct buf_ring * br)196 buf_ring_dequeue_sc(struct buf_ring *br)
197 {
198 uint32_t cons_head, cons_next, cons_idx;
199 uint32_t prod_tail, mask;
200 void *buf;
201
202 BR_LOCK_ASSERT(br);
203
204 mask = br->br_cons_mask;
205 cons_head = atomic_load_32(&br->br_cons_head);
206 prod_tail = atomic_load_acq_32(&br->br_prod_tail);
207
208 cons_next = cons_head + 1;
209
210 if (cons_head == prod_tail)
211 return (NULL);
212
213 cons_idx = cons_head & mask;
214 atomic_store_32(&br->br_cons_head, cons_next);
215 buf = br->br_ring[cons_idx];
216
217 #ifdef DEBUG_BUFRING
218 br->br_ring[cons_idx] = NULL;
219 if (atomic_load_32(&br->br_cons_tail) != cons_head)
220 panic("inconsistent list cons_tail=%d cons_head=%d",
221 atomic_load_32(&br->br_cons_tail), cons_head);
222 #endif
223 atomic_store_rel_32(&br->br_cons_tail, cons_next);
224 return (buf);
225 }
226
227 /*
228 * single-consumer advance after a peek
229 * use where it is protected by a lock
230 * e.g. a network driver's tx queue lock
231 */
232 static __inline void
buf_ring_advance_sc(struct buf_ring * br)233 buf_ring_advance_sc(struct buf_ring *br)
234 {
235 uint32_t cons_head, cons_next, prod_tail;
236 #ifdef DEBUG_BUFRING
237 uint32_t mask;
238
239 mask = br->br_cons_mask;
240 #endif
241 cons_head = atomic_load_32(&br->br_cons_head);
242 prod_tail = atomic_load_32(&br->br_prod_tail);
243
244 cons_next = cons_head + 1;
245 if (cons_head == prod_tail)
246 return;
247 atomic_store_32(&br->br_cons_head, cons_next);
248 #ifdef DEBUG_BUFRING
249 br->br_ring[cons_head & mask] = NULL;
250 #endif
251 atomic_store_rel_32(&br->br_cons_tail, cons_next);
252 }
253
254 /*
255 * Used to return a buffer (most likely already there)
256 * to the top of the ring. The caller should *not*
257 * have used any dequeue to pull it out of the ring
258 * but instead should have used the peek() function.
259 * This is normally used where the transmit queue
260 * of a driver is full, and an mbuf must be returned.
261 * Most likely whats in the ring-buffer is what
262 * is being put back (since it was not removed), but
263 * sometimes the lower transmit function may have
264 * done a pullup or other function that will have
265 * changed it. As an optimization we always put it
266 * back (since jhb says the store is probably cheaper),
267 * if we have to do a multi-queue version we will need
268 * the compare and an atomic.
269 */
270 static __inline void
buf_ring_putback_sc(struct buf_ring * br,void * buf)271 buf_ring_putback_sc(struct buf_ring *br, void *buf)
272 {
273 uint32_t cons_idx, mask;
274
275 mask = br->br_cons_mask;
276 cons_idx = atomic_load_32(&br->br_cons_head) & mask;
277 KASSERT(cons_idx != (atomic_load_32(&br->br_prod_tail) & mask),
278 ("Buf-Ring has none in putback")) ;
279 br->br_ring[cons_idx] = buf;
280 }
281
282 /*
283 * return a pointer to the first entry in the ring
284 * without modifying it, or NULL if the ring is empty
285 * race-prone if not protected by a lock
286 */
287 static __inline void *
buf_ring_peek(struct buf_ring * br)288 buf_ring_peek(struct buf_ring *br)
289 {
290 uint32_t cons_head, prod_tail, mask;
291
292 if (br->br_lock != NULL)
293 BR_LOCK_ASSERT(br);
294
295 mask = br->br_cons_mask;
296 prod_tail = atomic_load_acq_32(&br->br_prod_tail);
297 cons_head = atomic_load_32(&br->br_cons_head);
298
299 if (cons_head == prod_tail)
300 return (NULL);
301
302 return (br->br_ring[cons_head & mask]);
303 }
304
305 static __inline void *
buf_ring_peek_clear_sc(struct buf_ring * br)306 buf_ring_peek_clear_sc(struct buf_ring *br)
307 {
308 uint32_t cons_head, prod_tail, mask;
309 void *buf;
310
311 BR_LOCK_ASSERT(br);
312
313 mask = br->br_cons_mask;
314 prod_tail = atomic_load_acq_32(&br->br_prod_tail);
315 cons_head = atomic_load_32(&br->br_cons_head);
316
317 if (cons_head == prod_tail)
318 return (NULL);
319
320 buf = br->br_ring[cons_head & mask];
321 #ifdef DEBUG_BUFRING
322 /*
323 * Single consumer, i.e. cons_head will not move while we are
324 * running, so atomic_swap_ptr() is not necessary here.
325 */
326 br->br_ring[cons_head & mask] = NULL;
327 #endif
328 return (buf);
329 }
330
331 static __inline int
buf_ring_full(struct buf_ring * br)332 buf_ring_full(struct buf_ring *br)
333 {
334 return (atomic_load_32(&br->br_prod_head) ==
335 atomic_load_32(&br->br_cons_tail) + br->br_cons_size - 1);
336 }
337
338 static __inline int
buf_ring_empty(struct buf_ring * br)339 buf_ring_empty(struct buf_ring *br)
340 {
341 return (atomic_load_32(&br->br_cons_head) ==
342 atomic_load_32(&br->br_prod_tail));
343 }
344
345 static __inline int
buf_ring_count(struct buf_ring * br)346 buf_ring_count(struct buf_ring *br)
347 {
348 uint32_t cons_tail, prod_tail;
349
350 cons_tail = atomic_load_32(&br->br_cons_tail);
351 prod_tail = atomic_load_32(&br->br_prod_tail);
352 return ((br->br_prod_size + prod_tail - cons_tail) & br->br_prod_mask);
353 }
354
355 #ifdef _KERNEL
356 struct buf_ring *_buf_ring_alloc(int count, struct malloc_type *type,
357 int flags, struct lock_object *);
358 #define buf_ring_alloc(c, mt, f, lk) _Generic((lk), \
359 struct mtx *: _buf_ring_alloc((c), (mt), (f), lk2lo(lk)), \
360 struct mtx_padalign *: _buf_ring_alloc((c), (mt), (f), lk2lo(lk)),\
361 struct rwlock *: _buf_ring_alloc((c), (mt), (f), lk2lo(lk)), \
362 struct rmlock *: _buf_ring_alloc((c), (mt), (f), lk2lo(lk)))
363 void buf_ring_free(struct buf_ring *br, struct malloc_type *type);
364
365 static inline uint64_t
buf_ring_drops(struct buf_ring * br)366 buf_ring_drops(struct buf_ring *br)
367 {
368 return (counter_u64_fetch(br->br_drops));
369 }
370
371 #else /* !_KERNEL */
372 #include <stdlib.h>
373
374 static inline struct buf_ring *
buf_ring_alloc(int count)375 buf_ring_alloc(int count)
376 {
377 struct buf_ring *br;
378
379 KASSERT(powerof2(count), ("buf ring must be size power of 2"));
380
381 br = calloc(1, sizeof(struct buf_ring) + count * sizeof(void *));
382 if (br == NULL)
383 return (NULL);
384 br->br_prod_size = br->br_cons_size = count;
385 br->br_prod_mask = br->br_cons_mask = count - 1;
386 br->br_prod_head = br->br_cons_head = 0;
387 br->br_prod_tail = br->br_cons_tail = 0;
388 return (br);
389 }
390
391 static inline void
buf_ring_free(struct buf_ring * br)392 buf_ring_free(struct buf_ring *br)
393 {
394 free(br);
395 }
396
397 static inline uint64_t
buf_ring_drops(struct buf_ring * br)398 buf_ring_drops(struct buf_ring *br)
399 {
400 return (br->br_drops);
401 }
402
403 #endif /* !_KERNEL */
404 #endif /* _SYS_BUF_RING_H_ */
405