1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17#ifndef _LINUX_PTR_RING_H
18#define _LINUX_PTR_RING_H 1
19
20#ifdef __KERNEL__
21#include <linux/spinlock.h>
22#include <linux/cache.h>
23#include <linux/types.h>
24#include <linux/compiler.h>
25#include <linux/slab.h>
26#include <linux/mm.h>
27#include <asm/errno.h>
28#endif
29
30struct ptr_ring {
31 int producer ____cacheline_aligned_in_smp;
32 spinlock_t producer_lock;
33 int consumer_head ____cacheline_aligned_in_smp;
34 int consumer_tail;
35 spinlock_t consumer_lock;
36
37
38 int size ____cacheline_aligned_in_smp;
39 int batch;
40 void **queue;
41};
42
43
44
45
46
47
48
49static inline bool __ptr_ring_full(struct ptr_ring *r)
50{
51 return r->queue[r->producer];
52}
53
54static inline bool ptr_ring_full(struct ptr_ring *r)
55{
56 bool ret;
57
58 spin_lock(&r->producer_lock);
59 ret = __ptr_ring_full(r);
60 spin_unlock(&r->producer_lock);
61
62 return ret;
63}
64
65static inline bool ptr_ring_full_irq(struct ptr_ring *r)
66{
67 bool ret;
68
69 spin_lock_irq(&r->producer_lock);
70 ret = __ptr_ring_full(r);
71 spin_unlock_irq(&r->producer_lock);
72
73 return ret;
74}
75
76static inline bool ptr_ring_full_any(struct ptr_ring *r)
77{
78 unsigned long flags;
79 bool ret;
80
81 spin_lock_irqsave(&r->producer_lock, flags);
82 ret = __ptr_ring_full(r);
83 spin_unlock_irqrestore(&r->producer_lock, flags);
84
85 return ret;
86}
87
88static inline bool ptr_ring_full_bh(struct ptr_ring *r)
89{
90 bool ret;
91
92 spin_lock_bh(&r->producer_lock);
93 ret = __ptr_ring_full(r);
94 spin_unlock_bh(&r->producer_lock);
95
96 return ret;
97}
98
99
100
101
102
103
104static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
105{
106 if (unlikely(!r->size) || r->queue[r->producer])
107 return -ENOSPC;
108
109
110
111 smp_wmb();
112
113 WRITE_ONCE(r->queue[r->producer++], ptr);
114 if (unlikely(r->producer >= r->size))
115 r->producer = 0;
116 return 0;
117}
118
119
120
121
122
123
124static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
125{
126 int ret;
127
128 spin_lock(&r->producer_lock);
129 ret = __ptr_ring_produce(r, ptr);
130 spin_unlock(&r->producer_lock);
131
132 return ret;
133}
134
135static inline int ptr_ring_produce_irq(struct ptr_ring *r, void *ptr)
136{
137 int ret;
138
139 spin_lock_irq(&r->producer_lock);
140 ret = __ptr_ring_produce(r, ptr);
141 spin_unlock_irq(&r->producer_lock);
142
143 return ret;
144}
145
146static inline int ptr_ring_produce_any(struct ptr_ring *r, void *ptr)
147{
148 unsigned long flags;
149 int ret;
150
151 spin_lock_irqsave(&r->producer_lock, flags);
152 ret = __ptr_ring_produce(r, ptr);
153 spin_unlock_irqrestore(&r->producer_lock, flags);
154
155 return ret;
156}
157
158static inline int ptr_ring_produce_bh(struct ptr_ring *r, void *ptr)
159{
160 int ret;
161
162 spin_lock_bh(&r->producer_lock);
163 ret = __ptr_ring_produce(r, ptr);
164 spin_unlock_bh(&r->producer_lock);
165
166 return ret;
167}
168
169static inline void *__ptr_ring_peek(struct ptr_ring *r)
170{
171 if (likely(r->size))
172 return READ_ONCE(r->queue[r->consumer_head]);
173 return NULL;
174}
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194static inline bool __ptr_ring_empty(struct ptr_ring *r)
195{
196 if (likely(r->size))
197 return !r->queue[READ_ONCE(r->consumer_head)];
198 return true;
199}
200
201static inline bool ptr_ring_empty(struct ptr_ring *r)
202{
203 bool ret;
204
205 spin_lock(&r->consumer_lock);
206 ret = __ptr_ring_empty(r);
207 spin_unlock(&r->consumer_lock);
208
209 return ret;
210}
211
212static inline bool ptr_ring_empty_irq(struct ptr_ring *r)
213{
214 bool ret;
215
216 spin_lock_irq(&r->consumer_lock);
217 ret = __ptr_ring_empty(r);
218 spin_unlock_irq(&r->consumer_lock);
219
220 return ret;
221}
222
223static inline bool ptr_ring_empty_any(struct ptr_ring *r)
224{
225 unsigned long flags;
226 bool ret;
227
228 spin_lock_irqsave(&r->consumer_lock, flags);
229 ret = __ptr_ring_empty(r);
230 spin_unlock_irqrestore(&r->consumer_lock, flags);
231
232 return ret;
233}
234
235static inline bool ptr_ring_empty_bh(struct ptr_ring *r)
236{
237 bool ret;
238
239 spin_lock_bh(&r->consumer_lock);
240 ret = __ptr_ring_empty(r);
241 spin_unlock_bh(&r->consumer_lock);
242
243 return ret;
244}
245
246
247static inline void __ptr_ring_discard_one(struct ptr_ring *r)
248{
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264 int consumer_head = r->consumer_head;
265 int head = consumer_head++;
266
267
268
269
270
271
272 if (unlikely(consumer_head - r->consumer_tail >= r->batch ||
273 consumer_head >= r->size)) {
274
275
276
277
278
279 while (likely(head >= r->consumer_tail))
280 r->queue[head--] = NULL;
281 r->consumer_tail = consumer_head;
282 }
283 if (unlikely(consumer_head >= r->size)) {
284 consumer_head = 0;
285 r->consumer_tail = 0;
286 }
287
288 WRITE_ONCE(r->consumer_head, consumer_head);
289}
290
291static inline void *__ptr_ring_consume(struct ptr_ring *r)
292{
293 void *ptr;
294
295
296
297
298
299 ptr = __ptr_ring_peek(r);
300 if (ptr)
301 __ptr_ring_discard_one(r);
302
303 return ptr;
304}
305
306static inline int __ptr_ring_consume_batched(struct ptr_ring *r,
307 void **array, int n)
308{
309 void *ptr;
310 int i;
311
312 for (i = 0; i < n; i++) {
313 ptr = __ptr_ring_consume(r);
314 if (!ptr)
315 break;
316 array[i] = ptr;
317 }
318
319 return i;
320}
321
322
323
324
325
326
327static inline void *ptr_ring_consume(struct ptr_ring *r)
328{
329 void *ptr;
330
331 spin_lock(&r->consumer_lock);
332 ptr = __ptr_ring_consume(r);
333 spin_unlock(&r->consumer_lock);
334
335 return ptr;
336}
337
338static inline void *ptr_ring_consume_irq(struct ptr_ring *r)
339{
340 void *ptr;
341
342 spin_lock_irq(&r->consumer_lock);
343 ptr = __ptr_ring_consume(r);
344 spin_unlock_irq(&r->consumer_lock);
345
346 return ptr;
347}
348
349static inline void *ptr_ring_consume_any(struct ptr_ring *r)
350{
351 unsigned long flags;
352 void *ptr;
353
354 spin_lock_irqsave(&r->consumer_lock, flags);
355 ptr = __ptr_ring_consume(r);
356 spin_unlock_irqrestore(&r->consumer_lock, flags);
357
358 return ptr;
359}
360
361static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
362{
363 void *ptr;
364
365 spin_lock_bh(&r->consumer_lock);
366 ptr = __ptr_ring_consume(r);
367 spin_unlock_bh(&r->consumer_lock);
368
369 return ptr;
370}
371
372static inline int ptr_ring_consume_batched(struct ptr_ring *r,
373 void **array, int n)
374{
375 int ret;
376
377 spin_lock(&r->consumer_lock);
378 ret = __ptr_ring_consume_batched(r, array, n);
379 spin_unlock(&r->consumer_lock);
380
381 return ret;
382}
383
384static inline int ptr_ring_consume_batched_irq(struct ptr_ring *r,
385 void **array, int n)
386{
387 int ret;
388
389 spin_lock_irq(&r->consumer_lock);
390 ret = __ptr_ring_consume_batched(r, array, n);
391 spin_unlock_irq(&r->consumer_lock);
392
393 return ret;
394}
395
396static inline int ptr_ring_consume_batched_any(struct ptr_ring *r,
397 void **array, int n)
398{
399 unsigned long flags;
400 int ret;
401
402 spin_lock_irqsave(&r->consumer_lock, flags);
403 ret = __ptr_ring_consume_batched(r, array, n);
404 spin_unlock_irqrestore(&r->consumer_lock, flags);
405
406 return ret;
407}
408
409static inline int ptr_ring_consume_batched_bh(struct ptr_ring *r,
410 void **array, int n)
411{
412 int ret;
413
414 spin_lock_bh(&r->consumer_lock);
415 ret = __ptr_ring_consume_batched(r, array, n);
416 spin_unlock_bh(&r->consumer_lock);
417
418 return ret;
419}
420
421
422
423
424
425#define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))
426
427#define PTR_RING_PEEK_CALL(r, f) ({ \
428 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
429 \
430 spin_lock(&(r)->consumer_lock); \
431 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
432 spin_unlock(&(r)->consumer_lock); \
433 __PTR_RING_PEEK_CALL_v; \
434})
435
436#define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
437 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
438 \
439 spin_lock_irq(&(r)->consumer_lock); \
440 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
441 spin_unlock_irq(&(r)->consumer_lock); \
442 __PTR_RING_PEEK_CALL_v; \
443})
444
445#define PTR_RING_PEEK_CALL_BH(r, f) ({ \
446 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
447 \
448 spin_lock_bh(&(r)->consumer_lock); \
449 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
450 spin_unlock_bh(&(r)->consumer_lock); \
451 __PTR_RING_PEEK_CALL_v; \
452})
453
454#define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
455 typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
456 unsigned long __PTR_RING_PEEK_CALL_f;\
457 \
458 spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
459 __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
460 spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
461 __PTR_RING_PEEK_CALL_v; \
462})
463
464
465
466
467static inline void **__ptr_ring_init_queue_alloc(unsigned int size, gfp_t gfp)
468{
469 if (size > KMALLOC_MAX_SIZE / sizeof(void *))
470 return NULL;
471 return kvmalloc_array(size, sizeof(void *), gfp | __GFP_ZERO);
472}
473
474static inline void __ptr_ring_set_size(struct ptr_ring *r, int size)
475{
476 r->size = size;
477 r->batch = SMP_CACHE_BYTES * 2 / sizeof(*(r->queue));
478
479
480
481
482
483 if (r->batch > r->size / 2 || !r->batch)
484 r->batch = 1;
485}
486
487static inline int ptr_ring_init(struct ptr_ring *r, int size, gfp_t gfp)
488{
489 r->queue = __ptr_ring_init_queue_alloc(size, gfp);
490 if (!r->queue)
491 return -ENOMEM;
492
493 __ptr_ring_set_size(r, size);
494 r->producer = r->consumer_head = r->consumer_tail = 0;
495 spin_lock_init(&r->producer_lock);
496 spin_lock_init(&r->consumer_lock);
497
498 return 0;
499}
500
501
502
503
504
505
506
507
508
509
510
511static inline void ptr_ring_unconsume(struct ptr_ring *r, void **batch, int n,
512 void (*destroy)(void *))
513{
514 unsigned long flags;
515 int head;
516
517 spin_lock_irqsave(&r->consumer_lock, flags);
518 spin_lock(&r->producer_lock);
519
520 if (!r->size)
521 goto done;
522
523
524
525
526
527 head = r->consumer_head - 1;
528 while (likely(head >= r->consumer_tail))
529 r->queue[head--] = NULL;
530 r->consumer_tail = r->consumer_head;
531
532
533
534
535
536 while (n) {
537 head = r->consumer_head - 1;
538 if (head < 0)
539 head = r->size - 1;
540 if (r->queue[head]) {
541
542 goto done;
543 }
544 r->queue[head] = batch[--n];
545 r->consumer_tail = head;
546
547 WRITE_ONCE(r->consumer_head, head);
548 }
549
550done:
551
552 while (n)
553 destroy(batch[--n]);
554 spin_unlock(&r->producer_lock);
555 spin_unlock_irqrestore(&r->consumer_lock, flags);
556}
557
558static inline void **__ptr_ring_swap_queue(struct ptr_ring *r, void **queue,
559 int size, gfp_t gfp,
560 void (*destroy)(void *))
561{
562 int producer = 0;
563 void **old;
564 void *ptr;
565
566 while ((ptr = __ptr_ring_consume(r)))
567 if (producer < size)
568 queue[producer++] = ptr;
569 else if (destroy)
570 destroy(ptr);
571
572 if (producer >= size)
573 producer = 0;
574 __ptr_ring_set_size(r, size);
575 r->producer = producer;
576 r->consumer_head = 0;
577 r->consumer_tail = 0;
578 old = r->queue;
579 r->queue = queue;
580
581 return old;
582}
583
584
585
586
587
588
589
590static inline int ptr_ring_resize(struct ptr_ring *r, int size, gfp_t gfp,
591 void (*destroy)(void *))
592{
593 unsigned long flags;
594 void **queue = __ptr_ring_init_queue_alloc(size, gfp);
595 void **old;
596
597 if (!queue)
598 return -ENOMEM;
599
600 spin_lock_irqsave(&(r)->consumer_lock, flags);
601 spin_lock(&(r)->producer_lock);
602
603 old = __ptr_ring_swap_queue(r, queue, size, gfp, destroy);
604
605 spin_unlock(&(r)->producer_lock);
606 spin_unlock_irqrestore(&(r)->consumer_lock, flags);
607
608 kvfree(old);
609
610 return 0;
611}
612
613
614
615
616
617
618
619static inline int ptr_ring_resize_multiple(struct ptr_ring **rings,
620 unsigned int nrings,
621 int size,
622 gfp_t gfp, void (*destroy)(void *))
623{
624 unsigned long flags;
625 void ***queues;
626 int i;
627
628 queues = kmalloc_array(nrings, sizeof(*queues), gfp);
629 if (!queues)
630 goto noqueues;
631
632 for (i = 0; i < nrings; ++i) {
633 queues[i] = __ptr_ring_init_queue_alloc(size, gfp);
634 if (!queues[i])
635 goto nomem;
636 }
637
638 for (i = 0; i < nrings; ++i) {
639 spin_lock_irqsave(&(rings[i])->consumer_lock, flags);
640 spin_lock(&(rings[i])->producer_lock);
641 queues[i] = __ptr_ring_swap_queue(rings[i], queues[i],
642 size, gfp, destroy);
643 spin_unlock(&(rings[i])->producer_lock);
644 spin_unlock_irqrestore(&(rings[i])->consumer_lock, flags);
645 }
646
647 for (i = 0; i < nrings; ++i)
648 kvfree(queues[i]);
649
650 kfree(queues);
651
652 return 0;
653
654nomem:
655 while (--i >= 0)
656 kvfree(queues[i]);
657
658 kfree(queues);
659
660noqueues:
661 return -ENOMEM;
662}
663
664static inline void ptr_ring_cleanup(struct ptr_ring *r, void (*destroy)(void *))
665{
666 void *ptr;
667
668 if (destroy)
669 while ((ptr = ptr_ring_consume(r)))
670 destroy(ptr);
671 kvfree(r->queue);
672}
673
674#endif
675