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 ____cacheline_aligned_in_smp;
38 spinlock_t consumer_lock;
39 /* Shared consumer/producer data */
40 /* Read-only by both the producer and the consumer */
41 int size ____cacheline_aligned_in_smp; /* max entries in queue */
45 /* Note: callers invoking this in a loop must use a compiler barrier,
46 * for example cpu_relax().
47 * Callers don't need to take producer lock - if they don't
48 * the next call to __ptr_ring_produce may fail.
50 static inline bool __ptr_ring_full(struct ptr_ring *r)
52 return r->queue[r->producer];
55 static inline bool ptr_ring_full(struct ptr_ring *r)
58 return __ptr_ring_full(r);
61 /* Note: callers invoking this in a loop must use a compiler barrier,
62 * for example cpu_relax().
64 static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
66 if (__ptr_ring_full(r))
69 r->queue[r->producer++] = ptr;
70 if (unlikely(r->producer >= r->size))
75 static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
79 spin_lock(&r->producer_lock);
80 ret = __ptr_ring_produce(r, ptr);
81 spin_unlock(&r->producer_lock);
86 static inline int ptr_ring_produce_irq(struct ptr_ring *r, void *ptr)
90 spin_lock_irq(&r->producer_lock);
91 ret = __ptr_ring_produce(r, ptr);
92 spin_unlock_irq(&r->producer_lock);
97 static inline int ptr_ring_produce_any(struct ptr_ring *r, void *ptr)
102 spin_lock_irqsave(&r->producer_lock, flags);
103 ret = __ptr_ring_produce(r, ptr);
104 spin_unlock_irqrestore(&r->producer_lock, flags);
109 static inline int ptr_ring_produce_bh(struct ptr_ring *r, void *ptr)
113 spin_lock_bh(&r->producer_lock);
114 ret = __ptr_ring_produce(r, ptr);
115 spin_unlock_bh(&r->producer_lock);
120 /* Note: callers invoking this in a loop must use a compiler barrier,
121 * for example cpu_relax(). Callers must take consumer_lock
122 * if they dereference the pointer - see e.g. PTR_RING_PEEK_CALL.
123 * There's no need for a lock if pointer is merely tested - see e.g.
126 static inline void *__ptr_ring_peek(struct ptr_ring *r)
128 return r->queue[r->consumer];
131 static inline bool ptr_ring_empty(struct ptr_ring *r)
134 return !__ptr_ring_peek(r);
137 /* Must only be called after __ptr_ring_peek returned !NULL */
138 static inline void __ptr_ring_discard_one(struct ptr_ring *r)
140 r->queue[r->consumer++] = NULL;
141 if (unlikely(r->consumer >= r->size))
145 static inline void *__ptr_ring_consume(struct ptr_ring *r)
149 ptr = __ptr_ring_peek(r);
151 __ptr_ring_discard_one(r);
156 static inline void *ptr_ring_consume(struct ptr_ring *r)
160 spin_lock(&r->consumer_lock);
161 ptr = __ptr_ring_consume(r);
162 spin_unlock(&r->consumer_lock);
167 static inline void *ptr_ring_consume_irq(struct ptr_ring *r)
171 spin_lock_irq(&r->consumer_lock);
172 ptr = __ptr_ring_consume(r);
173 spin_unlock_irq(&r->consumer_lock);
178 static inline void *ptr_ring_consume_any(struct ptr_ring *r)
183 spin_lock_irqsave(&r->consumer_lock, flags);
184 ptr = __ptr_ring_consume(r);
185 spin_unlock_irqrestore(&r->consumer_lock, flags);
190 static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
194 spin_lock_bh(&r->consumer_lock);
195 ptr = __ptr_ring_consume(r);
196 spin_unlock_bh(&r->consumer_lock);
201 /* Cast to structure type and call a function without discarding from FIFO.
202 * Function must return a value.
203 * Callers must take consumer_lock.
205 #define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))
207 #define PTR_RING_PEEK_CALL(r, f) ({ \
208 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
210 spin_lock(&(r)->consumer_lock); \
211 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
212 spin_unlock(&(r)->consumer_lock); \
213 __PTR_RING_PEEK_CALL_v; \
216 #define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
217 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
219 spin_lock_irq(&(r)->consumer_lock); \
220 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
221 spin_unlock_irq(&(r)->consumer_lock); \
222 __PTR_RING_PEEK_CALL_v; \
225 #define PTR_RING_PEEK_CALL_BH(r, f) ({ \
226 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
228 spin_lock_bh(&(r)->consumer_lock); \
229 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
230 spin_unlock_bh(&(r)->consumer_lock); \
231 __PTR_RING_PEEK_CALL_v; \
234 #define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
235 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
236 unsigned long __PTR_RING_PEEK_CALL_f;\
238 spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
239 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
240 spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
241 __PTR_RING_PEEK_CALL_v; \
244 static inline int ptr_ring_init(struct ptr_ring *r, int size, gfp_t gfp)
246 r->queue = kzalloc(ALIGN(size * sizeof *(r->queue), SMP_CACHE_BYTES),
252 r->producer = r->consumer = 0;
253 spin_lock_init(&r->producer_lock);
254 spin_lock_init(&r->consumer_lock);
259 static inline void ptr_ring_cleanup(struct ptr_ring *r)
264 #endif /* _LINUX_PTR_RING_H */