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(). If ring is ever resized, callers must hold
47 * producer_lock - see e.g. ptr_ring_full. Otherwise, if callers don't hold
48 * producer_lock, 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)
59 spin_lock(&r->producer_lock);
60 ret = __ptr_ring_full(r);
61 spin_unlock(&r->producer_lock);
66 static inline bool ptr_ring_full_irq(struct ptr_ring *r)
70 spin_lock_irq(&r->producer_lock);
71 ret = __ptr_ring_full(r);
72 spin_unlock_irq(&r->producer_lock);
77 static inline bool ptr_ring_full_any(struct ptr_ring *r)
82 spin_lock_irqsave(&r->producer_lock, flags);
83 ret = __ptr_ring_full(r);
84 spin_unlock_irqrestore(&r->producer_lock, flags);
89 static inline bool ptr_ring_full_bh(struct ptr_ring *r)
93 spin_lock_bh(&r->producer_lock);
94 ret = __ptr_ring_full(r);
95 spin_unlock_bh(&r->producer_lock);
100 /* Note: callers invoking this in a loop must use a compiler barrier,
101 * for example cpu_relax(). Callers must hold producer_lock.
102 * Callers are responsible for making sure pointer that is being queued
103 * points to a valid data.
105 static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
107 if (unlikely(!r->size) || r->queue[r->producer])
110 /* Make sure the pointer we are storing points to a valid data. */
111 /* Pairs with smp_read_barrier_depends in __ptr_ring_consume. */
114 r->queue[r->producer++] = ptr;
115 if (unlikely(r->producer >= r->size))
121 * Note: resize (below) nests producer lock within consumer lock, so if you
122 * consume in interrupt or BH context, you must disable interrupts/BH when
125 static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
129 spin_lock(&r->producer_lock);
130 ret = __ptr_ring_produce(r, ptr);
131 spin_unlock(&r->producer_lock);
136 static inline int ptr_ring_produce_irq(struct ptr_ring *r, void *ptr)
140 spin_lock_irq(&r->producer_lock);
141 ret = __ptr_ring_produce(r, ptr);
142 spin_unlock_irq(&r->producer_lock);
147 static inline int ptr_ring_produce_any(struct ptr_ring *r, void *ptr)
152 spin_lock_irqsave(&r->producer_lock, flags);
153 ret = __ptr_ring_produce(r, ptr);
154 spin_unlock_irqrestore(&r->producer_lock, flags);
159 static inline int ptr_ring_produce_bh(struct ptr_ring *r, void *ptr)
163 spin_lock_bh(&r->producer_lock);
164 ret = __ptr_ring_produce(r, ptr);
165 spin_unlock_bh(&r->producer_lock);
170 /* Note: callers invoking this in a loop must use a compiler barrier,
171 * for example cpu_relax(). Callers must take consumer_lock
172 * if they dereference the pointer - see e.g. PTR_RING_PEEK_CALL.
173 * If ring is never resized, and if the pointer is merely
174 * tested, there's no need to take the lock - see e.g. __ptr_ring_empty.
176 static inline void *__ptr_ring_peek(struct ptr_ring *r)
179 return r->queue[r->consumer];
183 /* Note: callers invoking this in a loop must use a compiler barrier,
184 * for example cpu_relax(). Callers must take consumer_lock
185 * if the ring is ever resized - see e.g. ptr_ring_empty.
187 static inline bool __ptr_ring_empty(struct ptr_ring *r)
189 return !__ptr_ring_peek(r);
192 static inline bool ptr_ring_empty(struct ptr_ring *r)
196 spin_lock(&r->consumer_lock);
197 ret = __ptr_ring_empty(r);
198 spin_unlock(&r->consumer_lock);
203 static inline bool ptr_ring_empty_irq(struct ptr_ring *r)
207 spin_lock_irq(&r->consumer_lock);
208 ret = __ptr_ring_empty(r);
209 spin_unlock_irq(&r->consumer_lock);
214 static inline bool ptr_ring_empty_any(struct ptr_ring *r)
219 spin_lock_irqsave(&r->consumer_lock, flags);
220 ret = __ptr_ring_empty(r);
221 spin_unlock_irqrestore(&r->consumer_lock, flags);
226 static inline bool ptr_ring_empty_bh(struct ptr_ring *r)
230 spin_lock_bh(&r->consumer_lock);
231 ret = __ptr_ring_empty(r);
232 spin_unlock_bh(&r->consumer_lock);
237 /* Must only be called after __ptr_ring_peek returned !NULL */
238 static inline void __ptr_ring_discard_one(struct ptr_ring *r)
240 r->queue[r->consumer++] = NULL;
241 if (unlikely(r->consumer >= r->size))
245 static inline void *__ptr_ring_consume(struct ptr_ring *r)
249 ptr = __ptr_ring_peek(r);
251 __ptr_ring_discard_one(r);
253 /* Make sure anyone accessing data through the pointer is up to date. */
254 /* Pairs with smp_wmb in __ptr_ring_produce. */
255 smp_read_barrier_depends();
260 * Note: resize (below) nests producer lock within consumer lock, so if you
261 * call this in interrupt or BH context, you must disable interrupts/BH when
264 static inline void *ptr_ring_consume(struct ptr_ring *r)
268 spin_lock(&r->consumer_lock);
269 ptr = __ptr_ring_consume(r);
270 spin_unlock(&r->consumer_lock);
275 static inline void *ptr_ring_consume_irq(struct ptr_ring *r)
279 spin_lock_irq(&r->consumer_lock);
280 ptr = __ptr_ring_consume(r);
281 spin_unlock_irq(&r->consumer_lock);
286 static inline void *ptr_ring_consume_any(struct ptr_ring *r)
291 spin_lock_irqsave(&r->consumer_lock, flags);
292 ptr = __ptr_ring_consume(r);
293 spin_unlock_irqrestore(&r->consumer_lock, flags);
298 static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
302 spin_lock_bh(&r->consumer_lock);
303 ptr = __ptr_ring_consume(r);
304 spin_unlock_bh(&r->consumer_lock);
309 /* Cast to structure type and call a function without discarding from FIFO.
310 * Function must return a value.
311 * Callers must take consumer_lock.
313 #define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))
315 #define PTR_RING_PEEK_CALL(r, f) ({ \
316 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
318 spin_lock(&(r)->consumer_lock); \
319 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
320 spin_unlock(&(r)->consumer_lock); \
321 __PTR_RING_PEEK_CALL_v; \
324 #define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
325 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
327 spin_lock_irq(&(r)->consumer_lock); \
328 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
329 spin_unlock_irq(&(r)->consumer_lock); \
330 __PTR_RING_PEEK_CALL_v; \
333 #define PTR_RING_PEEK_CALL_BH(r, f) ({ \
334 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
336 spin_lock_bh(&(r)->consumer_lock); \
337 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
338 spin_unlock_bh(&(r)->consumer_lock); \
339 __PTR_RING_PEEK_CALL_v; \
342 #define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
343 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
344 unsigned long __PTR_RING_PEEK_CALL_f;\
346 spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
347 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
348 spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
349 __PTR_RING_PEEK_CALL_v; \
352 static inline void **__ptr_ring_init_queue_alloc(unsigned int size, gfp_t gfp)
354 if (size > KMALLOC_MAX_SIZE / sizeof(void *))
356 return kcalloc(size, sizeof(void *), gfp);
359 static inline int ptr_ring_init(struct ptr_ring *r, int size, gfp_t gfp)
361 r->queue = __ptr_ring_init_queue_alloc(size, gfp);
366 r->producer = r->consumer = 0;
367 spin_lock_init(&r->producer_lock);
368 spin_lock_init(&r->consumer_lock);
373 static inline void **__ptr_ring_swap_queue(struct ptr_ring *r, void **queue,
375 void (*destroy)(void *))
381 while ((ptr = __ptr_ring_consume(r)))
383 queue[producer++] = ptr;
387 if (producer >= size)
390 r->producer = producer;
399 * Note: producer lock is nested within consumer lock, so if you
400 * resize you must make sure all uses nest correctly.
401 * In particular if you consume ring in interrupt or BH context, you must
402 * disable interrupts/BH when doing so.
404 static inline int ptr_ring_resize(struct ptr_ring *r, int size, gfp_t gfp,
405 void (*destroy)(void *))
408 void **queue = __ptr_ring_init_queue_alloc(size, gfp);
414 spin_lock_irqsave(&(r)->consumer_lock, flags);
415 spin_lock(&(r)->producer_lock);
417 old = __ptr_ring_swap_queue(r, queue, size, gfp, destroy);
419 spin_unlock(&(r)->producer_lock);
420 spin_unlock_irqrestore(&(r)->consumer_lock, flags);
428 * Note: producer lock is nested within consumer lock, so if you
429 * resize you must make sure all uses nest correctly.
430 * In particular if you consume ring in interrupt or BH context, you must
431 * disable interrupts/BH when doing so.
433 static inline int ptr_ring_resize_multiple(struct ptr_ring **rings,
436 gfp_t gfp, void (*destroy)(void *))
442 queues = kmalloc_array(nrings, sizeof(*queues), gfp);
446 for (i = 0; i < nrings; ++i) {
447 queues[i] = __ptr_ring_init_queue_alloc(size, gfp);
452 for (i = 0; i < nrings; ++i) {
453 spin_lock_irqsave(&(rings[i])->consumer_lock, flags);
454 spin_lock(&(rings[i])->producer_lock);
455 queues[i] = __ptr_ring_swap_queue(rings[i], queues[i],
457 spin_unlock(&(rings[i])->producer_lock);
458 spin_unlock_irqrestore(&(rings[i])->consumer_lock, flags);
461 for (i = 0; i < nrings; ++i)
478 static inline void ptr_ring_cleanup(struct ptr_ring *r, void (*destroy)(void *))
483 while ((ptr = ptr_ring_consume(r)))
488 #endif /* _LINUX_PTR_RING_H */