2 * Definitions for the 'struct ptr_ring' datastructure.
5 * Michael S. Tsirkin <mst@redhat.com>
7 * Copyright (C) 2016 Red Hat, Inc.
9 * This program is free software; you can redistribute it and/or modify it
10 * under the terms of the GNU General Public License as published by the
11 * Free Software Foundation; either version 2 of the License, or (at your
12 * option) any later version.
14 * This is a limited-size FIFO maintaining pointers in FIFO order, with
15 * one CPU producing entries and another consuming entries from a FIFO.
17 * This implementation tries to minimize cache-contention when there is a
18 * single producer and a single consumer CPU.
21 #ifndef _LINUX_PTR_RING_H
22 #define _LINUX_PTR_RING_H 1
25 #include <linux/spinlock.h>
26 #include <linux/cache.h>
27 #include <linux/types.h>
28 #include <linux/compiler.h>
29 #include <linux/cache.h>
30 #include <linux/slab.h>
31 #include <asm/errno.h>
35 int producer ____cacheline_aligned_in_smp
;
36 spinlock_t producer_lock
;
37 int consumer_head ____cacheline_aligned_in_smp
; /* next valid entry */
38 int consumer_tail
; /* next entry to invalidate */
39 spinlock_t consumer_lock
;
40 /* Shared consumer/producer data */
41 /* Read-only by both the producer and the consumer */
42 int size ____cacheline_aligned_in_smp
; /* max entries in queue */
43 int batch
; /* number of entries to consume in a batch */
47 /* Note: callers invoking this in a loop must use a compiler barrier,
48 * for example cpu_relax(). If ring is ever resized, callers must hold
49 * producer_lock - see e.g. ptr_ring_full. Otherwise, if callers don't hold
50 * producer_lock, the next call to __ptr_ring_produce may fail.
52 static inline bool __ptr_ring_full(struct ptr_ring
*r
)
54 return r
->queue
[r
->producer
];
57 static inline bool ptr_ring_full(struct ptr_ring
*r
)
61 spin_lock(&r
->producer_lock
);
62 ret
= __ptr_ring_full(r
);
63 spin_unlock(&r
->producer_lock
);
68 static inline bool ptr_ring_full_irq(struct ptr_ring
*r
)
72 spin_lock_irq(&r
->producer_lock
);
73 ret
= __ptr_ring_full(r
);
74 spin_unlock_irq(&r
->producer_lock
);
79 static inline bool ptr_ring_full_any(struct ptr_ring
*r
)
84 spin_lock_irqsave(&r
->producer_lock
, flags
);
85 ret
= __ptr_ring_full(r
);
86 spin_unlock_irqrestore(&r
->producer_lock
, flags
);
91 static inline bool ptr_ring_full_bh(struct ptr_ring
*r
)
95 spin_lock_bh(&r
->producer_lock
);
96 ret
= __ptr_ring_full(r
);
97 spin_unlock_bh(&r
->producer_lock
);
102 /* Note: callers invoking this in a loop must use a compiler barrier,
103 * for example cpu_relax(). Callers must hold producer_lock.
104 * Callers are responsible for making sure pointer that is being queued
105 * points to a valid data.
107 static inline int __ptr_ring_produce(struct ptr_ring
*r
, void *ptr
)
109 if (unlikely(!r
->size
) || r
->queue
[r
->producer
])
112 /* Make sure the pointer we are storing points to a valid data. */
113 /* Pairs with smp_read_barrier_depends in __ptr_ring_consume. */
116 r
->queue
[r
->producer
++] = ptr
;
117 if (unlikely(r
->producer
>= r
->size
))
123 * Note: resize (below) nests producer lock within consumer lock, so if you
124 * consume in interrupt or BH context, you must disable interrupts/BH when
127 static inline int ptr_ring_produce(struct ptr_ring
*r
, void *ptr
)
131 spin_lock(&r
->producer_lock
);
132 ret
= __ptr_ring_produce(r
, ptr
);
133 spin_unlock(&r
->producer_lock
);
138 static inline int ptr_ring_produce_irq(struct ptr_ring
*r
, void *ptr
)
142 spin_lock_irq(&r
->producer_lock
);
143 ret
= __ptr_ring_produce(r
, ptr
);
144 spin_unlock_irq(&r
->producer_lock
);
149 static inline int ptr_ring_produce_any(struct ptr_ring
*r
, void *ptr
)
154 spin_lock_irqsave(&r
->producer_lock
, flags
);
155 ret
= __ptr_ring_produce(r
, ptr
);
156 spin_unlock_irqrestore(&r
->producer_lock
, flags
);
161 static inline int ptr_ring_produce_bh(struct ptr_ring
*r
, void *ptr
)
165 spin_lock_bh(&r
->producer_lock
);
166 ret
= __ptr_ring_produce(r
, ptr
);
167 spin_unlock_bh(&r
->producer_lock
);
172 /* Note: callers invoking this in a loop must use a compiler barrier,
173 * for example cpu_relax(). Callers must take consumer_lock
174 * if they dereference the pointer - see e.g. PTR_RING_PEEK_CALL.
175 * If ring is never resized, and if the pointer is merely
176 * tested, there's no need to take the lock - see e.g. __ptr_ring_empty.
177 * However, if called outside the lock, and if some other CPU
178 * consumes ring entries at the same time, the value returned
179 * is not guaranteed to be correct.
180 * In this case - to avoid incorrectly detecting the ring
181 * as empty - the CPU consuming the ring entries is responsible
182 * for either consuming all ring entries until the ring is empty,
183 * or synchronizing with some other CPU and causing it to
184 * execute __ptr_ring_peek and/or consume the ring enteries
185 * after the synchronization point.
187 static inline void *__ptr_ring_peek(struct ptr_ring
*r
)
190 return r
->queue
[r
->consumer_head
];
194 /* See __ptr_ring_peek above for locking rules. */
195 static inline bool __ptr_ring_empty(struct ptr_ring
*r
)
197 return !__ptr_ring_peek(r
);
200 static inline bool ptr_ring_empty(struct ptr_ring
*r
)
204 spin_lock(&r
->consumer_lock
);
205 ret
= __ptr_ring_empty(r
);
206 spin_unlock(&r
->consumer_lock
);
211 static inline bool ptr_ring_empty_irq(struct ptr_ring
*r
)
215 spin_lock_irq(&r
->consumer_lock
);
216 ret
= __ptr_ring_empty(r
);
217 spin_unlock_irq(&r
->consumer_lock
);
222 static inline bool ptr_ring_empty_any(struct ptr_ring
*r
)
227 spin_lock_irqsave(&r
->consumer_lock
, flags
);
228 ret
= __ptr_ring_empty(r
);
229 spin_unlock_irqrestore(&r
->consumer_lock
, flags
);
234 static inline bool ptr_ring_empty_bh(struct ptr_ring
*r
)
238 spin_lock_bh(&r
->consumer_lock
);
239 ret
= __ptr_ring_empty(r
);
240 spin_unlock_bh(&r
->consumer_lock
);
245 /* Must only be called after __ptr_ring_peek returned !NULL */
246 static inline void __ptr_ring_discard_one(struct ptr_ring
*r
)
248 /* Fundamentally, what we want to do is update consumer
249 * index and zero out the entry so producer can reuse it.
250 * Doing it naively at each consume would be as simple as:
251 * r->queue[r->consumer++] = NULL;
252 * if (unlikely(r->consumer >= r->size))
254 * but that is suboptimal when the ring is full as producer is writing
255 * out new entries in the same cache line. Defer these updates until a
256 * batch of entries has been consumed.
258 int head
= r
->consumer_head
++;
260 /* Once we have processed enough entries invalidate them in
261 * the ring all at once so producer can reuse their space in the ring.
262 * We also do this when we reach end of the ring - not mandatory
263 * but helps keep the implementation simple.
265 if (unlikely(r
->consumer_head
- r
->consumer_tail
>= r
->batch
||
266 r
->consumer_head
>= r
->size
)) {
267 /* Zero out entries in the reverse order: this way we touch the
268 * cache line that producer might currently be reading the last;
269 * producer won't make progress and touch other cache lines
270 * besides the first one until we write out all entries.
272 while (likely(head
>= r
->consumer_tail
))
273 r
->queue
[head
--] = NULL
;
274 r
->consumer_tail
= r
->consumer_head
;
276 if (unlikely(r
->consumer_head
>= r
->size
)) {
277 r
->consumer_head
= 0;
278 r
->consumer_tail
= 0;
282 static inline void *__ptr_ring_consume(struct ptr_ring
*r
)
286 ptr
= __ptr_ring_peek(r
);
288 __ptr_ring_discard_one(r
);
290 /* Make sure anyone accessing data through the pointer is up to date. */
291 /* Pairs with smp_wmb in __ptr_ring_produce. */
292 smp_read_barrier_depends();
296 static inline int __ptr_ring_consume_batched(struct ptr_ring
*r
,
302 for (i
= 0; i
< n
; i
++) {
303 ptr
= __ptr_ring_consume(r
);
313 * Note: resize (below) nests producer lock within consumer lock, so if you
314 * call this in interrupt or BH context, you must disable interrupts/BH when
317 static inline void *ptr_ring_consume(struct ptr_ring
*r
)
321 spin_lock(&r
->consumer_lock
);
322 ptr
= __ptr_ring_consume(r
);
323 spin_unlock(&r
->consumer_lock
);
328 static inline void *ptr_ring_consume_irq(struct ptr_ring
*r
)
332 spin_lock_irq(&r
->consumer_lock
);
333 ptr
= __ptr_ring_consume(r
);
334 spin_unlock_irq(&r
->consumer_lock
);
339 static inline void *ptr_ring_consume_any(struct ptr_ring
*r
)
344 spin_lock_irqsave(&r
->consumer_lock
, flags
);
345 ptr
= __ptr_ring_consume(r
);
346 spin_unlock_irqrestore(&r
->consumer_lock
, flags
);
351 static inline void *ptr_ring_consume_bh(struct ptr_ring
*r
)
355 spin_lock_bh(&r
->consumer_lock
);
356 ptr
= __ptr_ring_consume(r
);
357 spin_unlock_bh(&r
->consumer_lock
);
362 static inline int ptr_ring_consume_batched(struct ptr_ring
*r
,
367 spin_lock(&r
->consumer_lock
);
368 ret
= __ptr_ring_consume_batched(r
, array
, n
);
369 spin_unlock(&r
->consumer_lock
);
374 static inline int ptr_ring_consume_batched_irq(struct ptr_ring
*r
,
379 spin_lock_irq(&r
->consumer_lock
);
380 ret
= __ptr_ring_consume_batched(r
, array
, n
);
381 spin_unlock_irq(&r
->consumer_lock
);
386 static inline int ptr_ring_consume_batched_any(struct ptr_ring
*r
,
392 spin_lock_irqsave(&r
->consumer_lock
, flags
);
393 ret
= __ptr_ring_consume_batched(r
, array
, n
);
394 spin_unlock_irqrestore(&r
->consumer_lock
, flags
);
399 static inline int ptr_ring_consume_batched_bh(struct ptr_ring
*r
,
404 spin_lock_bh(&r
->consumer_lock
);
405 ret
= __ptr_ring_consume_batched(r
, array
, n
);
406 spin_unlock_bh(&r
->consumer_lock
);
411 /* Cast to structure type and call a function without discarding from FIFO.
412 * Function must return a value.
413 * Callers must take consumer_lock.
415 #define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))
417 #define PTR_RING_PEEK_CALL(r, f) ({ \
418 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
420 spin_lock(&(r)->consumer_lock); \
421 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
422 spin_unlock(&(r)->consumer_lock); \
423 __PTR_RING_PEEK_CALL_v; \
426 #define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
427 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
429 spin_lock_irq(&(r)->consumer_lock); \
430 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
431 spin_unlock_irq(&(r)->consumer_lock); \
432 __PTR_RING_PEEK_CALL_v; \
435 #define PTR_RING_PEEK_CALL_BH(r, f) ({ \
436 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
438 spin_lock_bh(&(r)->consumer_lock); \
439 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
440 spin_unlock_bh(&(r)->consumer_lock); \
441 __PTR_RING_PEEK_CALL_v; \
444 #define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
445 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
446 unsigned long __PTR_RING_PEEK_CALL_f;\
448 spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
449 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
450 spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
451 __PTR_RING_PEEK_CALL_v; \
454 /* Not all gfp_t flags (besides GFP_KERNEL) are allowed. See
455 * documentation for vmalloc for which of them are legal.
457 static inline void **__ptr_ring_init_queue_alloc(unsigned int size
, gfp_t gfp
)
459 if (size
> KMALLOC_MAX_SIZE
/ sizeof(void *))
461 return kvmalloc_array(size
, sizeof(void *), gfp
| __GFP_ZERO
);
464 static inline void __ptr_ring_set_size(struct ptr_ring
*r
, int size
)
467 r
->batch
= SMP_CACHE_BYTES
* 2 / sizeof(*(r
->queue
));
468 /* We need to set batch at least to 1 to make logic
469 * in __ptr_ring_discard_one work correctly.
470 * Batching too much (because ring is small) would cause a lot of
471 * burstiness. Needs tuning, for now disable batching.
473 if (r
->batch
> r
->size
/ 2 || !r
->batch
)
477 static inline int ptr_ring_init(struct ptr_ring
*r
, int size
, gfp_t gfp
)
479 r
->queue
= __ptr_ring_init_queue_alloc(size
, gfp
);
483 __ptr_ring_set_size(r
, size
);
484 r
->producer
= r
->consumer_head
= r
->consumer_tail
= 0;
485 spin_lock_init(&r
->producer_lock
);
486 spin_lock_init(&r
->consumer_lock
);
492 * Return entries into ring. Destroy entries that don't fit.
494 * Note: this is expected to be a rare slow path operation.
496 * Note: producer lock is nested within consumer lock, so if you
497 * resize you must make sure all uses nest correctly.
498 * In particular if you consume ring in interrupt or BH context, you must
499 * disable interrupts/BH when doing so.
501 static inline void ptr_ring_unconsume(struct ptr_ring
*r
, void **batch
, int n
,
502 void (*destroy
)(void *))
507 spin_lock_irqsave(&r
->consumer_lock
, flags
);
508 spin_lock(&r
->producer_lock
);
514 * Clean out buffered entries (for simplicity). This way following code
515 * can test entries for NULL and if not assume they are valid.
517 head
= r
->consumer_head
- 1;
518 while (likely(head
>= r
->consumer_tail
))
519 r
->queue
[head
--] = NULL
;
520 r
->consumer_tail
= r
->consumer_head
;
523 * Go over entries in batch, start moving head back and copy entries.
524 * Stop when we run into previously unconsumed entries.
527 head
= r
->consumer_head
- 1;
530 if (r
->queue
[head
]) {
531 /* This batch entry will have to be destroyed. */
534 r
->queue
[head
] = batch
[--n
];
535 r
->consumer_tail
= r
->consumer_head
= head
;
539 /* Destroy all entries left in the batch. */
542 spin_unlock(&r
->producer_lock
);
543 spin_unlock_irqrestore(&r
->consumer_lock
, flags
);
546 static inline void **__ptr_ring_swap_queue(struct ptr_ring
*r
, void **queue
,
548 void (*destroy
)(void *))
554 while ((ptr
= __ptr_ring_consume(r
)))
556 queue
[producer
++] = ptr
;
560 if (producer
>= size
)
562 __ptr_ring_set_size(r
, size
);
563 r
->producer
= producer
;
564 r
->consumer_head
= 0;
565 r
->consumer_tail
= 0;
573 * Note: producer lock is nested within consumer lock, so if you
574 * resize you must make sure all uses nest correctly.
575 * In particular if you consume ring in interrupt or BH context, you must
576 * disable interrupts/BH when doing so.
578 static inline int ptr_ring_resize(struct ptr_ring
*r
, int size
, gfp_t gfp
,
579 void (*destroy
)(void *))
582 void **queue
= __ptr_ring_init_queue_alloc(size
, gfp
);
588 spin_lock_irqsave(&(r
)->consumer_lock
, flags
);
589 spin_lock(&(r
)->producer_lock
);
591 old
= __ptr_ring_swap_queue(r
, queue
, size
, gfp
, destroy
);
593 spin_unlock(&(r
)->producer_lock
);
594 spin_unlock_irqrestore(&(r
)->consumer_lock
, flags
);
602 * Note: producer lock is nested within consumer lock, so if you
603 * resize you must make sure all uses nest correctly.
604 * In particular if you consume ring in interrupt or BH context, you must
605 * disable interrupts/BH when doing so.
607 static inline int ptr_ring_resize_multiple(struct ptr_ring
**rings
,
610 gfp_t gfp
, void (*destroy
)(void *))
616 queues
= kmalloc_array(nrings
, sizeof(*queues
), gfp
);
620 for (i
= 0; i
< nrings
; ++i
) {
621 queues
[i
] = __ptr_ring_init_queue_alloc(size
, gfp
);
626 for (i
= 0; i
< nrings
; ++i
) {
627 spin_lock_irqsave(&(rings
[i
])->consumer_lock
, flags
);
628 spin_lock(&(rings
[i
])->producer_lock
);
629 queues
[i
] = __ptr_ring_swap_queue(rings
[i
], queues
[i
],
631 spin_unlock(&(rings
[i
])->producer_lock
);
632 spin_unlock_irqrestore(&(rings
[i
])->consumer_lock
, flags
);
635 for (i
= 0; i
< nrings
; ++i
)
652 static inline void ptr_ring_cleanup(struct ptr_ring
*r
, void (*destroy
)(void *))
657 while ((ptr
= ptr_ring_consume(r
)))
662 #endif /* _LINUX_PTR_RING_H */