ptr_ring: try vmalloc() when kmalloc() fails
[sfrench/cifs-2.6.git] / include / linux / ptr_ring.h
1 /*
2  *      Definitions for the 'struct ptr_ring' datastructure.
3  *
4  *      Author:
5  *              Michael S. Tsirkin <mst@redhat.com>
6  *
7  *      Copyright (C) 2016 Red Hat, Inc.
8  *
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.
13  *
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.
16  *
17  *      This implementation tries to minimize cache-contention when there is a
18  *      single producer and a single consumer CPU.
19  */
20
21 #ifndef _LINUX_PTR_RING_H
22 #define _LINUX_PTR_RING_H 1
23
24 #ifdef __KERNEL__
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>
32 #endif
33
34 struct ptr_ring {
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 */
44         void **queue;
45 };
46
47 /* Note: callers invoking this in a loop must use a compiler barrier,
48  * for example cpu_relax().
49  *
50  * NB: this is unlike __ptr_ring_empty in that callers must hold producer_lock:
51  * see e.g. ptr_ring_full.
52  */
53 static inline bool __ptr_ring_full(struct ptr_ring *r)
54 {
55         return r->queue[r->producer];
56 }
57
58 static inline bool ptr_ring_full(struct ptr_ring *r)
59 {
60         bool ret;
61
62         spin_lock(&r->producer_lock);
63         ret = __ptr_ring_full(r);
64         spin_unlock(&r->producer_lock);
65
66         return ret;
67 }
68
69 static inline bool ptr_ring_full_irq(struct ptr_ring *r)
70 {
71         bool ret;
72
73         spin_lock_irq(&r->producer_lock);
74         ret = __ptr_ring_full(r);
75         spin_unlock_irq(&r->producer_lock);
76
77         return ret;
78 }
79
80 static inline bool ptr_ring_full_any(struct ptr_ring *r)
81 {
82         unsigned long flags;
83         bool ret;
84
85         spin_lock_irqsave(&r->producer_lock, flags);
86         ret = __ptr_ring_full(r);
87         spin_unlock_irqrestore(&r->producer_lock, flags);
88
89         return ret;
90 }
91
92 static inline bool ptr_ring_full_bh(struct ptr_ring *r)
93 {
94         bool ret;
95
96         spin_lock_bh(&r->producer_lock);
97         ret = __ptr_ring_full(r);
98         spin_unlock_bh(&r->producer_lock);
99
100         return ret;
101 }
102
103 /* Note: callers invoking this in a loop must use a compiler barrier,
104  * for example cpu_relax(). Callers must hold producer_lock.
105  * Callers are responsible for making sure pointer that is being queued
106  * points to a valid data.
107  */
108 static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
109 {
110         if (unlikely(!r->size) || r->queue[r->producer])
111                 return -ENOSPC;
112
113         /* Make sure the pointer we are storing points to a valid data. */
114         /* Pairs with smp_read_barrier_depends in __ptr_ring_consume. */
115         smp_wmb();
116
117         WRITE_ONCE(r->queue[r->producer++], ptr);
118         if (unlikely(r->producer >= r->size))
119                 r->producer = 0;
120         return 0;
121 }
122
123 /*
124  * Note: resize (below) nests producer lock within consumer lock, so if you
125  * consume in interrupt or BH context, you must disable interrupts/BH when
126  * calling this.
127  */
128 static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
129 {
130         int ret;
131
132         spin_lock(&r->producer_lock);
133         ret = __ptr_ring_produce(r, ptr);
134         spin_unlock(&r->producer_lock);
135
136         return ret;
137 }
138
139 static inline int ptr_ring_produce_irq(struct ptr_ring *r, void *ptr)
140 {
141         int ret;
142
143         spin_lock_irq(&r->producer_lock);
144         ret = __ptr_ring_produce(r, ptr);
145         spin_unlock_irq(&r->producer_lock);
146
147         return ret;
148 }
149
150 static inline int ptr_ring_produce_any(struct ptr_ring *r, void *ptr)
151 {
152         unsigned long flags;
153         int ret;
154
155         spin_lock_irqsave(&r->producer_lock, flags);
156         ret = __ptr_ring_produce(r, ptr);
157         spin_unlock_irqrestore(&r->producer_lock, flags);
158
159         return ret;
160 }
161
162 static inline int ptr_ring_produce_bh(struct ptr_ring *r, void *ptr)
163 {
164         int ret;
165
166         spin_lock_bh(&r->producer_lock);
167         ret = __ptr_ring_produce(r, ptr);
168         spin_unlock_bh(&r->producer_lock);
169
170         return ret;
171 }
172
173 static inline void *__ptr_ring_peek(struct ptr_ring *r)
174 {
175         if (likely(r->size))
176                 return READ_ONCE(r->queue[r->consumer_head]);
177         return NULL;
178 }
179
180 /*
181  * Test ring empty status without taking any locks.
182  *
183  * NB: This is only safe to call if ring is never resized.
184  *
185  * However, if some other CPU consumes ring entries at the same time, the value
186  * returned is not guaranteed to be correct.
187  *
188  * In this case - to avoid incorrectly detecting the ring
189  * as empty - the CPU consuming the ring entries is responsible
190  * for either consuming all ring entries until the ring is empty,
191  * or synchronizing with some other CPU and causing it to
192  * re-test __ptr_ring_empty and/or consume the ring enteries
193  * after the synchronization point.
194  *
195  * Note: callers invoking this in a loop must use a compiler barrier,
196  * for example cpu_relax().
197  */
198 static inline bool __ptr_ring_empty(struct ptr_ring *r)
199 {
200         if (likely(r->size))
201                 return !r->queue[READ_ONCE(r->consumer_head)];
202         return true;
203 }
204
205 static inline bool ptr_ring_empty(struct ptr_ring *r)
206 {
207         bool ret;
208
209         spin_lock(&r->consumer_lock);
210         ret = __ptr_ring_empty(r);
211         spin_unlock(&r->consumer_lock);
212
213         return ret;
214 }
215
216 static inline bool ptr_ring_empty_irq(struct ptr_ring *r)
217 {
218         bool ret;
219
220         spin_lock_irq(&r->consumer_lock);
221         ret = __ptr_ring_empty(r);
222         spin_unlock_irq(&r->consumer_lock);
223
224         return ret;
225 }
226
227 static inline bool ptr_ring_empty_any(struct ptr_ring *r)
228 {
229         unsigned long flags;
230         bool ret;
231
232         spin_lock_irqsave(&r->consumer_lock, flags);
233         ret = __ptr_ring_empty(r);
234         spin_unlock_irqrestore(&r->consumer_lock, flags);
235
236         return ret;
237 }
238
239 static inline bool ptr_ring_empty_bh(struct ptr_ring *r)
240 {
241         bool ret;
242
243         spin_lock_bh(&r->consumer_lock);
244         ret = __ptr_ring_empty(r);
245         spin_unlock_bh(&r->consumer_lock);
246
247         return ret;
248 }
249
250 /* Must only be called after __ptr_ring_peek returned !NULL */
251 static inline void __ptr_ring_discard_one(struct ptr_ring *r)
252 {
253         /* Fundamentally, what we want to do is update consumer
254          * index and zero out the entry so producer can reuse it.
255          * Doing it naively at each consume would be as simple as:
256          *       consumer = r->consumer;
257          *       r->queue[consumer++] = NULL;
258          *       if (unlikely(consumer >= r->size))
259          *               consumer = 0;
260          *       r->consumer = consumer;
261          * but that is suboptimal when the ring is full as producer is writing
262          * out new entries in the same cache line.  Defer these updates until a
263          * batch of entries has been consumed.
264          */
265         /* Note: we must keep consumer_head valid at all times for __ptr_ring_empty
266          * to work correctly.
267          */
268         int consumer_head = r->consumer_head;
269         int head = consumer_head++;
270
271         /* Once we have processed enough entries invalidate them in
272          * the ring all at once so producer can reuse their space in the ring.
273          * We also do this when we reach end of the ring - not mandatory
274          * but helps keep the implementation simple.
275          */
276         if (unlikely(consumer_head - r->consumer_tail >= r->batch ||
277                      consumer_head >= r->size)) {
278                 /* Zero out entries in the reverse order: this way we touch the
279                  * cache line that producer might currently be reading the last;
280                  * producer won't make progress and touch other cache lines
281                  * besides the first one until we write out all entries.
282                  */
283                 while (likely(head >= r->consumer_tail))
284                         r->queue[head--] = NULL;
285                 r->consumer_tail = consumer_head;
286         }
287         if (unlikely(consumer_head >= r->size)) {
288                 consumer_head = 0;
289                 r->consumer_tail = 0;
290         }
291         /* matching READ_ONCE in __ptr_ring_empty for lockless tests */
292         WRITE_ONCE(r->consumer_head, consumer_head);
293 }
294
295 static inline void *__ptr_ring_consume(struct ptr_ring *r)
296 {
297         void *ptr;
298
299         ptr = __ptr_ring_peek(r);
300         if (ptr)
301                 __ptr_ring_discard_one(r);
302
303         /* Make sure anyone accessing data through the pointer is up to date. */
304         /* Pairs with smp_wmb in __ptr_ring_produce. */
305         smp_read_barrier_depends();
306         return ptr;
307 }
308
309 static inline int __ptr_ring_consume_batched(struct ptr_ring *r,
310                                              void **array, int n)
311 {
312         void *ptr;
313         int i;
314
315         for (i = 0; i < n; i++) {
316                 ptr = __ptr_ring_consume(r);
317                 if (!ptr)
318                         break;
319                 array[i] = ptr;
320         }
321
322         return i;
323 }
324
325 /*
326  * Note: resize (below) nests producer lock within consumer lock, so if you
327  * call this in interrupt or BH context, you must disable interrupts/BH when
328  * producing.
329  */
330 static inline void *ptr_ring_consume(struct ptr_ring *r)
331 {
332         void *ptr;
333
334         spin_lock(&r->consumer_lock);
335         ptr = __ptr_ring_consume(r);
336         spin_unlock(&r->consumer_lock);
337
338         return ptr;
339 }
340
341 static inline void *ptr_ring_consume_irq(struct ptr_ring *r)
342 {
343         void *ptr;
344
345         spin_lock_irq(&r->consumer_lock);
346         ptr = __ptr_ring_consume(r);
347         spin_unlock_irq(&r->consumer_lock);
348
349         return ptr;
350 }
351
352 static inline void *ptr_ring_consume_any(struct ptr_ring *r)
353 {
354         unsigned long flags;
355         void *ptr;
356
357         spin_lock_irqsave(&r->consumer_lock, flags);
358         ptr = __ptr_ring_consume(r);
359         spin_unlock_irqrestore(&r->consumer_lock, flags);
360
361         return ptr;
362 }
363
364 static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
365 {
366         void *ptr;
367
368         spin_lock_bh(&r->consumer_lock);
369         ptr = __ptr_ring_consume(r);
370         spin_unlock_bh(&r->consumer_lock);
371
372         return ptr;
373 }
374
375 static inline int ptr_ring_consume_batched(struct ptr_ring *r,
376                                            void **array, int n)
377 {
378         int ret;
379
380         spin_lock(&r->consumer_lock);
381         ret = __ptr_ring_consume_batched(r, array, n);
382         spin_unlock(&r->consumer_lock);
383
384         return ret;
385 }
386
387 static inline int ptr_ring_consume_batched_irq(struct ptr_ring *r,
388                                                void **array, int n)
389 {
390         int ret;
391
392         spin_lock_irq(&r->consumer_lock);
393         ret = __ptr_ring_consume_batched(r, array, n);
394         spin_unlock_irq(&r->consumer_lock);
395
396         return ret;
397 }
398
399 static inline int ptr_ring_consume_batched_any(struct ptr_ring *r,
400                                                void **array, int n)
401 {
402         unsigned long flags;
403         int ret;
404
405         spin_lock_irqsave(&r->consumer_lock, flags);
406         ret = __ptr_ring_consume_batched(r, array, n);
407         spin_unlock_irqrestore(&r->consumer_lock, flags);
408
409         return ret;
410 }
411
412 static inline int ptr_ring_consume_batched_bh(struct ptr_ring *r,
413                                               void **array, int n)
414 {
415         int ret;
416
417         spin_lock_bh(&r->consumer_lock);
418         ret = __ptr_ring_consume_batched(r, array, n);
419         spin_unlock_bh(&r->consumer_lock);
420
421         return ret;
422 }
423
424 /* Cast to structure type and call a function without discarding from FIFO.
425  * Function must return a value.
426  * Callers must take consumer_lock.
427  */
428 #define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))
429
430 #define PTR_RING_PEEK_CALL(r, f) ({ \
431         typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
432         \
433         spin_lock(&(r)->consumer_lock); \
434         __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
435         spin_unlock(&(r)->consumer_lock); \
436         __PTR_RING_PEEK_CALL_v; \
437 })
438
439 #define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
440         typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
441         \
442         spin_lock_irq(&(r)->consumer_lock); \
443         __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
444         spin_unlock_irq(&(r)->consumer_lock); \
445         __PTR_RING_PEEK_CALL_v; \
446 })
447
448 #define PTR_RING_PEEK_CALL_BH(r, f) ({ \
449         typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
450         \
451         spin_lock_bh(&(r)->consumer_lock); \
452         __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
453         spin_unlock_bh(&(r)->consumer_lock); \
454         __PTR_RING_PEEK_CALL_v; \
455 })
456
457 #define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
458         typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
459         unsigned long __PTR_RING_PEEK_CALL_f;\
460         \
461         spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
462         __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
463         spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
464         __PTR_RING_PEEK_CALL_v; \
465 })
466
467 /* Not all gfp_t flags (besides GFP_KERNEL) are allowed. See
468  * documentation for vmalloc for which of them are legal.
469  */
470 static inline void **__ptr_ring_init_queue_alloc(unsigned int size, gfp_t gfp)
471 {
472         if (size * sizeof(void *) > KMALLOC_MAX_SIZE)
473                 return NULL;
474         return kvmalloc_array(size, sizeof(void *), gfp | __GFP_ZERO);
475 }
476
477 static inline void __ptr_ring_set_size(struct ptr_ring *r, int size)
478 {
479         r->size = size;
480         r->batch = SMP_CACHE_BYTES * 2 / sizeof(*(r->queue));
481         /* We need to set batch at least to 1 to make logic
482          * in __ptr_ring_discard_one work correctly.
483          * Batching too much (because ring is small) would cause a lot of
484          * burstiness. Needs tuning, for now disable batching.
485          */
486         if (r->batch > r->size / 2 || !r->batch)
487                 r->batch = 1;
488 }
489
490 static inline int ptr_ring_init(struct ptr_ring *r, int size, gfp_t gfp)
491 {
492         r->queue = __ptr_ring_init_queue_alloc(size, gfp);
493         if (!r->queue)
494                 return -ENOMEM;
495
496         __ptr_ring_set_size(r, size);
497         r->producer = r->consumer_head = r->consumer_tail = 0;
498         spin_lock_init(&r->producer_lock);
499         spin_lock_init(&r->consumer_lock);
500
501         return 0;
502 }
503
504 /*
505  * Return entries into ring. Destroy entries that don't fit.
506  *
507  * Note: this is expected to be a rare slow path operation.
508  *
509  * Note: producer lock is nested within consumer lock, so if you
510  * resize you must make sure all uses nest correctly.
511  * In particular if you consume ring in interrupt or BH context, you must
512  * disable interrupts/BH when doing so.
513  */
514 static inline void ptr_ring_unconsume(struct ptr_ring *r, void **batch, int n,
515                                       void (*destroy)(void *))
516 {
517         unsigned long flags;
518         int head;
519
520         spin_lock_irqsave(&r->consumer_lock, flags);
521         spin_lock(&r->producer_lock);
522
523         if (!r->size)
524                 goto done;
525
526         /*
527          * Clean out buffered entries (for simplicity). This way following code
528          * can test entries for NULL and if not assume they are valid.
529          */
530         head = r->consumer_head - 1;
531         while (likely(head >= r->consumer_tail))
532                 r->queue[head--] = NULL;
533         r->consumer_tail = r->consumer_head;
534
535         /*
536          * Go over entries in batch, start moving head back and copy entries.
537          * Stop when we run into previously unconsumed entries.
538          */
539         while (n) {
540                 head = r->consumer_head - 1;
541                 if (head < 0)
542                         head = r->size - 1;
543                 if (r->queue[head]) {
544                         /* This batch entry will have to be destroyed. */
545                         goto done;
546                 }
547                 r->queue[head] = batch[--n];
548                 r->consumer_tail = head;
549                 /* matching READ_ONCE in __ptr_ring_empty for lockless tests */
550                 WRITE_ONCE(r->consumer_head, head);
551         }
552
553 done:
554         /* Destroy all entries left in the batch. */
555         while (n)
556                 destroy(batch[--n]);
557         spin_unlock(&r->producer_lock);
558         spin_unlock_irqrestore(&r->consumer_lock, flags);
559 }
560
561 static inline void **__ptr_ring_swap_queue(struct ptr_ring *r, void **queue,
562                                            int size, gfp_t gfp,
563                                            void (*destroy)(void *))
564 {
565         int producer = 0;
566         void **old;
567         void *ptr;
568
569         while ((ptr = __ptr_ring_consume(r)))
570                 if (producer < size)
571                         queue[producer++] = ptr;
572                 else if (destroy)
573                         destroy(ptr);
574
575         __ptr_ring_set_size(r, size);
576         r->producer = producer;
577         r->consumer_head = 0;
578         r->consumer_tail = 0;
579         old = r->queue;
580         r->queue = queue;
581
582         return old;
583 }
584
585 /*
586  * Note: producer lock is nested within consumer lock, so if you
587  * resize you must make sure all uses nest correctly.
588  * In particular if you consume ring in interrupt or BH context, you must
589  * disable interrupts/BH when doing so.
590  */
591 static inline int ptr_ring_resize(struct ptr_ring *r, int size, gfp_t gfp,
592                                   void (*destroy)(void *))
593 {
594         unsigned long flags;
595         void **queue = __ptr_ring_init_queue_alloc(size, gfp);
596         void **old;
597
598         if (!queue)
599                 return -ENOMEM;
600
601         spin_lock_irqsave(&(r)->consumer_lock, flags);
602         spin_lock(&(r)->producer_lock);
603
604         old = __ptr_ring_swap_queue(r, queue, size, gfp, destroy);
605
606         spin_unlock(&(r)->producer_lock);
607         spin_unlock_irqrestore(&(r)->consumer_lock, flags);
608
609         kvfree(old);
610
611         return 0;
612 }
613
614 /*
615  * Note: producer lock is nested within consumer lock, so if you
616  * resize you must make sure all uses nest correctly.
617  * In particular if you consume ring in interrupt or BH context, you must
618  * disable interrupts/BH when doing so.
619  */
620 static inline int ptr_ring_resize_multiple(struct ptr_ring **rings,
621                                            unsigned int nrings,
622                                            int size,
623                                            gfp_t gfp, void (*destroy)(void *))
624 {
625         unsigned long flags;
626         void ***queues;
627         int i;
628
629         queues = kmalloc_array(nrings, sizeof(*queues), gfp);
630         if (!queues)
631                 goto noqueues;
632
633         for (i = 0; i < nrings; ++i) {
634                 queues[i] = __ptr_ring_init_queue_alloc(size, gfp);
635                 if (!queues[i])
636                         goto nomem;
637         }
638
639         for (i = 0; i < nrings; ++i) {
640                 spin_lock_irqsave(&(rings[i])->consumer_lock, flags);
641                 spin_lock(&(rings[i])->producer_lock);
642                 queues[i] = __ptr_ring_swap_queue(rings[i], queues[i],
643                                                   size, gfp, destroy);
644                 spin_unlock(&(rings[i])->producer_lock);
645                 spin_unlock_irqrestore(&(rings[i])->consumer_lock, flags);
646         }
647
648         for (i = 0; i < nrings; ++i)
649                 kvfree(queues[i]);
650
651         kfree(queues);
652
653         return 0;
654
655 nomem:
656         while (--i >= 0)
657                 kvfree(queues[i]);
658
659         kfree(queues);
660
661 noqueues:
662         return -ENOMEM;
663 }
664
665 static inline void ptr_ring_cleanup(struct ptr_ring *r, void (*destroy)(void *))
666 {
667         void *ptr;
668
669         if (destroy)
670                 while ((ptr = ptr_ring_consume(r)))
671                         destroy(ptr);
672         kvfree(r->queue);
673 }
674
675 #endif /* _LINUX_PTR_RING_H  */