96232d35d9fc140e23f5439c6e642f17a07f0fc3
[dpdk.git] / lib / librte_ring / rte_ring.h
1 /*-
2  *   BSD LICENSE
3  * 
4  *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
5  *   All rights reserved.
6  * 
7  *   Redistribution and use in source and binary forms, with or without
8  *   modification, are permitted provided that the following conditions
9  *   are met:
10  * 
11  *     * Redistributions of source code must retain the above copyright
12  *       notice, this list of conditions and the following disclaimer.
13  *     * Redistributions in binary form must reproduce the above copyright
14  *       notice, this list of conditions and the following disclaimer in
15  *       the documentation and/or other materials provided with the
16  *       distribution.
17  *     * Neither the name of Intel Corporation nor the names of its
18  *       contributors may be used to endorse or promote products derived
19  *       from this software without specific prior written permission.
20  * 
21  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
24  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
25  *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
28  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
29  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
30  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
31  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32  */
33
34 /*
35  * Derived from FreeBSD's bufring.h
36  *
37  **************************************************************************
38  *
39  * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
40  * All rights reserved.
41  *
42  * Redistribution and use in source and binary forms, with or without
43  * modification, are permitted provided that the following conditions are met:
44  *
45  * 1. Redistributions of source code must retain the above copyright notice,
46  *    this list of conditions and the following disclaimer.
47  *
48  * 2. The name of Kip Macy nor the names of other
49  *    contributors may be used to endorse or promote products derived from
50  *    this software without specific prior written permission.
51  *
52  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
53  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
54  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
55  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
56  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
57  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
58  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
59  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
60  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
61  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
62  * POSSIBILITY OF SUCH DAMAGE.
63  *
64  ***************************************************************************/
65
66 #ifndef _RTE_RING_H_
67 #define _RTE_RING_H_
68
69 /**
70  * @file
71  * RTE Ring
72  *
73  * The Ring Manager is a fixed-size queue, implemented as a table of
74  * pointers. Head and tail pointers are modified atomically, allowing
75  * concurrent access to it. It has the following features:
76  *
77  * - FIFO (First In First Out)
78  * - Maximum size is fixed; the pointers are stored in a table.
79  * - Lockless implementation.
80  * - Multi- or single-consumer dequeue.
81  * - Multi- or single-producer enqueue.
82  * - Bulk dequeue.
83  * - Bulk enqueue.
84  *
85  * Note: the ring implementation is not preemptable. A lcore must not
86  * be interrupted by another task that uses the same ring.
87  *
88  */
89
90 #ifdef __cplusplus
91 extern "C" {
92 #endif
93
94 #include <stdint.h>
95 #include <sys/queue.h>
96 #include <errno.h>
97 #include <rte_common.h>
98 #include <rte_memory.h>
99 #include <rte_lcore.h>
100 #include <rte_atomic.h>
101 #include <rte_branch_prediction.h>
102
103 enum rte_ring_queue_behavior {
104         RTE_RING_QUEUE_FIXED = 0, /* Enq/Deq a fixed number of items from a ring */
105         RTE_RING_QUEUE_VARIABLE   /* Enq/Deq as many items a possible from ring */
106 };
107
108 #ifdef RTE_LIBRTE_RING_DEBUG
109 /**
110  * A structure that stores the ring statistics (per-lcore).
111  */
112 struct rte_ring_debug_stats {
113         uint64_t enq_success_bulk; /**< Successful enqueues number. */
114         uint64_t enq_success_objs; /**< Objects successfully enqueued. */
115         uint64_t enq_quota_bulk;   /**< Successful enqueues above watermark. */
116         uint64_t enq_quota_objs;   /**< Objects enqueued above watermark. */
117         uint64_t enq_fail_bulk;    /**< Failed enqueues number. */
118         uint64_t enq_fail_objs;    /**< Objects that failed to be enqueued. */
119         uint64_t deq_success_bulk; /**< Successful dequeues number. */
120         uint64_t deq_success_objs; /**< Objects successfully dequeued. */
121         uint64_t deq_fail_bulk;    /**< Failed dequeues number. */
122         uint64_t deq_fail_objs;    /**< Objects that failed to be dequeued. */
123 } __rte_cache_aligned;
124 #endif
125
126 #define RTE_RING_NAMESIZE 32 /**< The maximum length of a ring name. */
127 #define RTE_RING_MZ_PREFIX "RG_"
128
129 /**
130  * An RTE ring structure.
131  *
132  * The producer and the consumer have a head and a tail index. The particularity
133  * of these index is that they are not between 0 and size(ring). These indexes
134  * are between 0 and 2^32, and we mask their value when we access the ring[]
135  * field. Thanks to this assumption, we can do subtractions between 2 index
136  * values in a modulo-32bit base: that's why the overflow of the indexes is not
137  * a problem.
138  */
139 struct rte_ring {
140         TAILQ_ENTRY(rte_ring) next;      /**< Next in list. */
141
142         char name[RTE_RING_NAMESIZE];    /**< Name of the ring. */
143         int flags;                       /**< Flags supplied at creation. */
144
145         /** Ring producer status. */
146         struct prod {
147                 uint32_t watermark;      /**< Maximum items before EDQUOT. */
148                 uint32_t sp_enqueue;     /**< True, if single producer. */
149                 uint32_t size;           /**< Size of ring. */
150                 uint32_t mask;           /**< Mask (size-1) of ring. */
151                 volatile uint32_t head;  /**< Producer head. */
152                 volatile uint32_t tail;  /**< Producer tail. */
153         } prod __rte_cache_aligned;
154
155         /** Ring consumer status. */
156         struct cons {
157                 uint32_t sc_dequeue;     /**< True, if single consumer. */
158                 uint32_t size;           /**< Size of the ring. */
159                 uint32_t mask;           /**< Mask (size-1) of ring. */
160                 volatile uint32_t head;  /**< Consumer head. */
161                 volatile uint32_t tail;  /**< Consumer tail. */
162 #ifdef RTE_RING_SPLIT_PROD_CONS
163         } cons __rte_cache_aligned;
164 #else
165         } cons;
166 #endif
167
168 #ifdef RTE_LIBRTE_RING_DEBUG
169         struct rte_ring_debug_stats stats[RTE_MAX_LCORE];
170 #endif
171
172         void * ring[0] __rte_cache_aligned; /**< Memory space of ring starts here.
173                                                                                  * not volatile so need to be careful
174                                                                                  * about compiler re-ordering */
175 };
176
177 #define RING_F_SP_ENQ 0x0001 /**< The default enqueue is "single-producer". */
178 #define RING_F_SC_DEQ 0x0002 /**< The default dequeue is "single-consumer". */
179 #define RTE_RING_QUOT_EXCEED (1 << 31)  /**< Quota exceed for burst ops */
180 #define RTE_RING_SZ_MASK  (unsigned)(0x0fffffff) /**< Ring size mask */
181
182 /**
183  * @internal When debug is enabled, store ring statistics.
184  * @param r
185  *   A pointer to the ring.
186  * @param name
187  *   The name of the statistics field to increment in the ring.
188  * @param n
189  *   The number to add to the object-oriented statistics.
190  */
191 #ifdef RTE_LIBRTE_RING_DEBUG
192 #define __RING_STAT_ADD(r, name, n) do {                \
193                 unsigned __lcore_id = rte_lcore_id();   \
194                 r->stats[__lcore_id].name##_objs += n;  \
195                 r->stats[__lcore_id].name##_bulk += 1;  \
196         } while(0)
197 #else
198 #define __RING_STAT_ADD(r, name, n) do {} while(0)
199 #endif
200
201 /**
202  * Calculate the memory size needed for a ring
203  *
204  * This function returns the number of bytes needed for a ring, given
205  * the number of elements in it. This value is the sum of the size of
206  * the structure rte_ring and the size of the memory needed by the
207  * objects pointers. The value is aligned to a cache line size.
208  *
209  * @param count
210  *   The number of elements in the ring (must be a power of 2).
211  * @return
212  *   - The memory size needed for the ring on success.
213  *   - -EINVAL if count is not a power of 2.
214  */
215 ssize_t rte_ring_get_memsize(unsigned count);
216
217 /**
218  * Initialize a ring structure.
219  *
220  * Initialize a ring structure in memory pointed by "r". The size of the
221  * memory area must be large enough to store the ring structure and the
222  * object table. It is advised to use rte_ring_get_memsize() to get the
223  * appropriate size.
224  *
225  * The ring size is set to *count*, which must be a power of two. Water
226  * marking is disabled by default. The real usable ring size is
227  * *count-1* instead of *count* to differentiate a free ring from an
228  * empty ring.
229  *
230  * The ring is not added in RTE_TAILQ_RING global list. Indeed, the
231  * memory given by the caller may not be shareable among dpdk
232  * processes.
233  *
234  * @param r
235  *   The pointer to the ring structure followed by the objects table.
236  * @param name
237  *   The name of the ring.
238  * @param count
239  *   The number of elements in the ring (must be a power of 2).
240  * @param flags
241  *   An OR of the following:
242  *    - RING_F_SP_ENQ: If this flag is set, the default behavior when
243  *      using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
244  *      is "single-producer". Otherwise, it is "multi-producers".
245  *    - RING_F_SC_DEQ: If this flag is set, the default behavior when
246  *      using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
247  *      is "single-consumer". Otherwise, it is "multi-consumers".
248  * @return
249  *   0 on success, or a negative value on error.
250  */
251 int rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
252         unsigned flags);
253
254 /**
255  * Create a new ring named *name* in memory.
256  *
257  * This function uses ``memzone_reserve()`` to allocate memory. Then it
258  * calls rte_ring_init() to initialize an empty ring.
259  *
260  * The new ring size is set to *count*, which must be a power of
261  * two. Water marking is disabled by default. The real usable ring size
262  * is *count-1* instead of *count* to differentiate a free ring from an
263  * empty ring.
264  *
265  * The ring is added in RTE_TAILQ_RING list.
266  *
267  * @param name
268  *   The name of the ring.
269  * @param count
270  *   The size of the ring (must be a power of 2).
271  * @param socket_id
272  *   The *socket_id* argument is the socket identifier in case of
273  *   NUMA. The value can be *SOCKET_ID_ANY* if there is no NUMA
274  *   constraint for the reserved zone.
275  * @param flags
276  *   An OR of the following:
277  *    - RING_F_SP_ENQ: If this flag is set, the default behavior when
278  *      using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
279  *      is "single-producer". Otherwise, it is "multi-producers".
280  *    - RING_F_SC_DEQ: If this flag is set, the default behavior when
281  *      using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
282  *      is "single-consumer". Otherwise, it is "multi-consumers".
283  * @return
284  *   On success, the pointer to the new allocated ring. NULL on error with
285  *    rte_errno set appropriately. Possible errno values include:
286  *    - E_RTE_NO_CONFIG - function could not get pointer to rte_config structure
287  *    - E_RTE_SECONDARY - function was called from a secondary process instance
288  *    - E_RTE_NO_TAILQ - no tailq list could be got for the ring list
289  *    - EINVAL - count provided is not a power of 2
290  *    - ENOSPC - the maximum number of memzones has already been allocated
291  *    - EEXIST - a memzone with the same name already exists
292  *    - ENOMEM - no appropriate memory area found in which to create memzone
293  */
294 struct rte_ring *rte_ring_create(const char *name, unsigned count,
295                                  int socket_id, unsigned flags);
296
297 /**
298  * Change the high water mark.
299  *
300  * If *count* is 0, water marking is disabled. Otherwise, it is set to the
301  * *count* value. The *count* value must be greater than 0 and less
302  * than the ring size.
303  *
304  * This function can be called at any time (not necessarily at
305  * initialization).
306  *
307  * @param r
308  *   A pointer to the ring structure.
309  * @param count
310  *   The new water mark value.
311  * @return
312  *   - 0: Success; water mark changed.
313  *   - -EINVAL: Invalid water mark value.
314  */
315 int rte_ring_set_water_mark(struct rte_ring *r, unsigned count);
316
317 /**
318  * Dump the status of the ring to the console.
319  *
320  * @param r
321  *   A pointer to the ring structure.
322  */
323 void rte_ring_dump(const struct rte_ring *r);
324
325 /* the actual enqueue of pointers on the ring. 
326  * Placed here since identical code needed in both
327  * single and multi producer enqueue functions */
328 #define ENQUEUE_PTRS() do { \
329         const uint32_t size = r->prod.size; \
330         uint32_t idx = prod_head & mask; \
331         if (likely(idx + n < size)) { \
332                 for (i = 0; i < (n & ((~(unsigned)0x3))); i+=4, idx+=4) { \
333                         r->ring[idx] = obj_table[i]; \
334                         r->ring[idx+1] = obj_table[i+1]; \
335                         r->ring[idx+2] = obj_table[i+2]; \
336                         r->ring[idx+3] = obj_table[i+3]; \
337                 } \
338                 switch (n & 0x3) { \
339                         case 3: r->ring[idx++] = obj_table[i++]; \
340                         case 2: r->ring[idx++] = obj_table[i++]; \
341                         case 1: r->ring[idx++] = obj_table[i++]; \
342                 } \
343         } else { \
344                 for (i = 0; idx < size; i++, idx++)\
345                         r->ring[idx] = obj_table[i]; \
346                 for (idx = 0; i < n; i++, idx++) \
347                         r->ring[idx] = obj_table[i]; \
348         } \
349 } while(0)
350
351 /* the actual copy of pointers on the ring to obj_table. 
352  * Placed here since identical code needed in both
353  * single and multi consumer dequeue functions */
354 #define DEQUEUE_PTRS() do { \
355         uint32_t idx = cons_head & mask; \
356         const uint32_t size = r->cons.size; \
357         if (likely(idx + n < size)) { \
358                 for (i = 0; i < (n & (~(unsigned)0x3)); i+=4, idx+=4) {\
359                         obj_table[i] = r->ring[idx]; \
360                         obj_table[i+1] = r->ring[idx+1]; \
361                         obj_table[i+2] = r->ring[idx+2]; \
362                         obj_table[i+3] = r->ring[idx+3]; \
363                 } \
364                 switch (n & 0x3) { \
365                         case 3: obj_table[i++] = r->ring[idx++]; \
366                         case 2: obj_table[i++] = r->ring[idx++]; \
367                         case 1: obj_table[i++] = r->ring[idx++]; \
368                 } \
369         } else { \
370                 for (i = 0; idx < size; i++, idx++) \
371                         obj_table[i] = r->ring[idx]; \
372                 for (idx = 0; i < n; i++, idx++) \
373                         obj_table[i] = r->ring[idx]; \
374         } \
375 } while (0)
376
377 /**
378  * @internal Enqueue several objects on the ring (multi-producers safe).
379  *
380  * This function uses a "compare and set" instruction to move the
381  * producer index atomically.
382  *
383  * @param r
384  *   A pointer to the ring structure.
385  * @param obj_table
386  *   A pointer to a table of void * pointers (objects).
387  * @param n
388  *   The number of objects to add in the ring from the obj_table.
389  * @param behavior
390  *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
391  *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
392  * @return
393  *   Depend on the behavior value
394  *   if behavior = RTE_RING_QUEUE_FIXED
395  *   - 0: Success; objects enqueue.
396  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
397  *     high water mark is exceeded.
398  *   - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
399  *   if behavior = RTE_RING_QUEUE_VARIABLE
400  *   - n: Actual number of objects enqueued.
401  */
402 static inline int __attribute__((always_inline))
403 __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
404                          unsigned n, enum rte_ring_queue_behavior behavior)
405 {
406         uint32_t prod_head, prod_next;
407         uint32_t cons_tail, free_entries;
408         const unsigned max = n;
409         int success;
410         unsigned i;
411         uint32_t mask = r->prod.mask;
412         int ret;
413
414         /* move prod.head atomically */
415         do {
416                 /* Reset n to the initial burst count */
417                 n = max;
418
419                 prod_head = r->prod.head;
420                 cons_tail = r->cons.tail;
421                 /* The subtraction is done between two unsigned 32bits value
422                  * (the result is always modulo 32 bits even if we have
423                  * prod_head > cons_tail). So 'free_entries' is always between 0
424                  * and size(ring)-1. */
425                 free_entries = (mask + cons_tail - prod_head);
426
427                 /* check that we have enough room in ring */
428                 if (unlikely(n > free_entries)) {
429                         if (behavior == RTE_RING_QUEUE_FIXED) {
430                                 __RING_STAT_ADD(r, enq_fail, n);
431                                 return -ENOBUFS;
432                         }
433                         else {
434                                 /* No free entry available */
435                                 if (unlikely(free_entries == 0)) {
436                                         __RING_STAT_ADD(r, enq_fail, n);
437                                         return 0;
438                                 }
439
440                                 n = free_entries;
441                         }
442                 }
443
444                 prod_next = prod_head + n;
445                 success = rte_atomic32_cmpset(&r->prod.head, prod_head,
446                                               prod_next);
447         } while (unlikely(success == 0));
448
449         /* write entries in ring */
450         ENQUEUE_PTRS();
451         rte_compiler_barrier();
452
453         /* if we exceed the watermark */
454         if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
455                 ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT :
456                                 (int)(n | RTE_RING_QUOT_EXCEED);
457                 __RING_STAT_ADD(r, enq_quota, n);
458         }
459         else {
460                 ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n;
461                 __RING_STAT_ADD(r, enq_success, n);
462         }
463
464         /*
465          * If there are other enqueues in progress that preceeded us,
466          * we need to wait for them to complete
467          */
468         while (unlikely(r->prod.tail != prod_head))
469                 rte_pause();
470
471         r->prod.tail = prod_next;
472         return ret;
473 }
474
475 /**
476  * @internal Enqueue several objects on a ring (NOT multi-producers safe).
477  *
478  * @param r
479  *   A pointer to the ring structure.
480  * @param obj_table
481  *   A pointer to a table of void * pointers (objects).
482  * @param n
483  *   The number of objects to add in the ring from the obj_table.
484  * @param behavior
485  *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
486  *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
487  * @return
488  *   Depend on the behavior value
489  *   if behavior = RTE_RING_QUEUE_FIXED
490  *   - 0: Success; objects enqueue.
491  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
492  *     high water mark is exceeded.
493  *   - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
494  *   if behavior = RTE_RING_QUEUE_VARIABLE
495  *   - n: Actual number of objects enqueued.
496  */
497 static inline int __attribute__((always_inline))
498 __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
499                          unsigned n, enum rte_ring_queue_behavior behavior)
500 {
501         uint32_t prod_head, cons_tail;
502         uint32_t prod_next, free_entries;
503         unsigned i;
504         uint32_t mask = r->prod.mask;
505         int ret;
506
507         prod_head = r->prod.head;
508         cons_tail = r->cons.tail;
509         /* The subtraction is done between two unsigned 32bits value
510          * (the result is always modulo 32 bits even if we have
511          * prod_head > cons_tail). So 'free_entries' is always between 0
512          * and size(ring)-1. */
513         free_entries = mask + cons_tail - prod_head;
514
515         /* check that we have enough room in ring */
516         if (unlikely(n > free_entries)) {
517                 if (behavior == RTE_RING_QUEUE_FIXED) {
518                         __RING_STAT_ADD(r, enq_fail, n);
519                         return -ENOBUFS;
520                 }
521                 else {
522                         /* No free entry available */
523                         if (unlikely(free_entries == 0)) {
524                                 __RING_STAT_ADD(r, enq_fail, n);
525                                 return 0;
526                         }
527
528                         n = free_entries;
529                 }
530         }
531
532         prod_next = prod_head + n;
533         r->prod.head = prod_next;
534
535         /* write entries in ring */
536         ENQUEUE_PTRS();
537         rte_compiler_barrier();
538
539         /* if we exceed the watermark */
540         if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
541                 ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT :
542                         (int)(n | RTE_RING_QUOT_EXCEED);
543                 __RING_STAT_ADD(r, enq_quota, n);
544         }
545         else {
546                 ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n;
547                 __RING_STAT_ADD(r, enq_success, n);
548         }
549
550         r->prod.tail = prod_next;
551         return ret;
552 }
553
554 /**
555  * @internal Dequeue several objects from a ring (multi-consumers safe). When
556  * the request objects are more than the available objects, only dequeue the
557  * actual number of objects
558  *
559  * This function uses a "compare and set" instruction to move the
560  * consumer index atomically.
561  *
562  * @param r
563  *   A pointer to the ring structure.
564  * @param obj_table
565  *   A pointer to a table of void * pointers (objects) that will be filled.
566  * @param n
567  *   The number of objects to dequeue from the ring to the obj_table.
568  * @param behavior
569  *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
570  *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
571  * @return
572  *   Depend on the behavior value
573  *   if behavior = RTE_RING_QUEUE_FIXED
574  *   - 0: Success; objects dequeued.
575  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
576  *     dequeued.
577  *   if behavior = RTE_RING_QUEUE_VARIABLE
578  *   - n: Actual number of objects dequeued.
579  */
580
581 static inline int __attribute__((always_inline))
582 __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
583                  unsigned n, enum rte_ring_queue_behavior behavior)
584 {
585         uint32_t cons_head, prod_tail;
586         uint32_t cons_next, entries;
587         const unsigned max = n;
588         int success;
589         unsigned i;
590         uint32_t mask = r->prod.mask;
591
592         /* move cons.head atomically */
593         do {
594                 /* Restore n as it may change every loop */
595                 n = max;
596
597                 cons_head = r->cons.head;
598                 prod_tail = r->prod.tail;
599                 /* The subtraction is done between two unsigned 32bits value
600                  * (the result is always modulo 32 bits even if we have
601                  * cons_head > prod_tail). So 'entries' is always between 0
602                  * and size(ring)-1. */
603                 entries = (prod_tail - cons_head);
604
605                 /* Set the actual entries for dequeue */
606                 if (n > entries) {
607                         if (behavior == RTE_RING_QUEUE_FIXED) {
608                                 __RING_STAT_ADD(r, deq_fail, n);
609                                 return -ENOENT;
610                         }
611                         else {
612                                 if (unlikely(entries == 0)){
613                                         __RING_STAT_ADD(r, deq_fail, n);
614                                         return 0;
615                                 }
616
617                                 n = entries;
618                         }
619                 }
620
621                 cons_next = cons_head + n;
622                 success = rte_atomic32_cmpset(&r->cons.head, cons_head,
623                                               cons_next);
624         } while (unlikely(success == 0));
625
626         /* copy in table */
627         DEQUEUE_PTRS();
628         rte_compiler_barrier();
629
630         /*
631          * If there are other dequeues in progress that preceded us,
632          * we need to wait for them to complete
633          */
634         while (unlikely(r->cons.tail != cons_head))
635                 rte_pause();
636
637         __RING_STAT_ADD(r, deq_success, n);
638         r->cons.tail = cons_next;
639
640         return behavior == RTE_RING_QUEUE_FIXED ? 0 : n;
641 }
642
643 /**
644  * @internal Dequeue several objects from a ring (NOT multi-consumers safe).
645  * When the request objects are more than the available objects, only dequeue
646  * the actual number of objects
647  *
648  * @param r
649  *   A pointer to the ring structure.
650  * @param obj_table
651  *   A pointer to a table of void * pointers (objects) that will be filled.
652  * @param n
653  *   The number of objects to dequeue from the ring to the obj_table.
654  * @param behavior
655  *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
656  *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
657  * @return
658  *   Depend on the behavior value
659  *   if behavior = RTE_RING_QUEUE_FIXED
660  *   - 0: Success; objects dequeued.
661  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
662  *     dequeued.
663  *   if behavior = RTE_RING_QUEUE_VARIABLE
664  *   - n: Actual number of objects dequeued.
665  */
666 static inline int __attribute__((always_inline))
667 __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
668                  unsigned n, enum rte_ring_queue_behavior behavior)
669 {
670         uint32_t cons_head, prod_tail;
671         uint32_t cons_next, entries;
672         unsigned i;
673         uint32_t mask = r->prod.mask;
674
675         cons_head = r->cons.head;
676         prod_tail = r->prod.tail;
677         /* The subtraction is done between two unsigned 32bits value
678          * (the result is always modulo 32 bits even if we have
679          * cons_head > prod_tail). So 'entries' is always between 0
680          * and size(ring)-1. */
681         entries = prod_tail - cons_head;
682
683         if (n > entries) {
684                 if (behavior == RTE_RING_QUEUE_FIXED) {
685                         __RING_STAT_ADD(r, deq_fail, n);
686                         return -ENOENT;
687                 }
688                 else {
689                         if (unlikely(entries == 0)){
690                                 __RING_STAT_ADD(r, deq_fail, n);
691                                 return 0;
692                         }
693
694                         n = entries;
695                 }
696         }
697
698         cons_next = cons_head + n;
699         r->cons.head = cons_next;
700
701         /* copy in table */
702         DEQUEUE_PTRS();
703         rte_compiler_barrier();
704
705         __RING_STAT_ADD(r, deq_success, n);
706         r->cons.tail = cons_next;
707         return behavior == RTE_RING_QUEUE_FIXED ? 0 : n;
708 }
709
710 /**
711  * Enqueue several objects on the ring (multi-producers safe).
712  *
713  * This function uses a "compare and set" instruction to move the
714  * producer index atomically.
715  *
716  * @param r
717  *   A pointer to the ring structure.
718  * @param obj_table
719  *   A pointer to a table of void * pointers (objects).
720  * @param n
721  *   The number of objects to add in the ring from the obj_table.
722  * @return
723  *   - 0: Success; objects enqueue.
724  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
725  *     high water mark is exceeded.
726  *   - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
727  */
728 static inline int __attribute__((always_inline))
729 rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
730                          unsigned n)
731 {
732         return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
733 }
734
735 /**
736  * Enqueue several objects on a ring (NOT multi-producers safe).
737  *
738  * @param r
739  *   A pointer to the ring structure.
740  * @param obj_table
741  *   A pointer to a table of void * pointers (objects).
742  * @param n
743  *   The number of objects to add in the ring from the obj_table.
744  * @return
745  *   - 0: Success; objects enqueued.
746  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
747  *     high water mark is exceeded.
748  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
749  */
750 static inline int __attribute__((always_inline))
751 rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
752                          unsigned n)
753 {
754         return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
755 }
756
757 /**
758  * Enqueue several objects on a ring.
759  *
760  * This function calls the multi-producer or the single-producer
761  * version depending on the default behavior that was specified at
762  * ring creation time (see flags).
763  *
764  * @param r
765  *   A pointer to the ring structure.
766  * @param obj_table
767  *   A pointer to a table of void * pointers (objects).
768  * @param n
769  *   The number of objects to add in the ring from the obj_table.
770  * @return
771  *   - 0: Success; objects enqueued.
772  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
773  *     high water mark is exceeded.
774  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
775  */
776 static inline int __attribute__((always_inline))
777 rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
778                       unsigned n)
779 {
780         if (r->prod.sp_enqueue)
781                 return rte_ring_sp_enqueue_bulk(r, obj_table, n);
782         else
783                 return rte_ring_mp_enqueue_bulk(r, obj_table, n);
784 }
785
786 /**
787  * Enqueue one object on a ring (multi-producers safe).
788  *
789  * This function uses a "compare and set" instruction to move the
790  * producer index atomically.
791  *
792  * @param r
793  *   A pointer to the ring structure.
794  * @param obj
795  *   A pointer to the object to be added.
796  * @return
797  *   - 0: Success; objects enqueued.
798  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
799  *     high water mark is exceeded.
800  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
801  */
802 static inline int __attribute__((always_inline))
803 rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
804 {
805         return rte_ring_mp_enqueue_bulk(r, &obj, 1);
806 }
807
808 /**
809  * Enqueue one object on a ring (NOT multi-producers safe).
810  *
811  * @param r
812  *   A pointer to the ring structure.
813  * @param obj
814  *   A pointer to the object to be added.
815  * @return
816  *   - 0: Success; objects enqueued.
817  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
818  *     high water mark is exceeded.
819  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
820  */
821 static inline int __attribute__((always_inline))
822 rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
823 {
824         return rte_ring_sp_enqueue_bulk(r, &obj, 1);
825 }
826
827 /**
828  * Enqueue one object on a ring.
829  *
830  * This function calls the multi-producer or the single-producer
831  * version, depending on the default behaviour that was specified at
832  * ring creation time (see flags).
833  *
834  * @param r
835  *   A pointer to the ring structure.
836  * @param obj
837  *   A pointer to the object to be added.
838  * @return
839  *   - 0: Success; objects enqueued.
840  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
841  *     high water mark is exceeded.
842  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
843  */
844 static inline int __attribute__((always_inline))
845 rte_ring_enqueue(struct rte_ring *r, void *obj)
846 {
847         if (r->prod.sp_enqueue)
848                 return rte_ring_sp_enqueue(r, obj);
849         else
850                 return rte_ring_mp_enqueue(r, obj);
851 }
852
853 /**
854  * Dequeue several objects from a ring (multi-consumers safe).
855  *
856  * This function uses a "compare and set" instruction to move the
857  * consumer index atomically.
858  *
859  * @param r
860  *   A pointer to the ring structure.
861  * @param obj_table
862  *   A pointer to a table of void * pointers (objects) that will be filled.
863  * @param n
864  *   The number of objects to dequeue from the ring to the obj_table.
865  * @return
866  *   - 0: Success; objects dequeued.
867  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
868  *     dequeued.
869  */
870 static inline int __attribute__((always_inline))
871 rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
872 {
873         return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
874 }
875
876 /**
877  * Dequeue several objects from a ring (NOT multi-consumers safe).
878  *
879  * @param r
880  *   A pointer to the ring structure.
881  * @param obj_table
882  *   A pointer to a table of void * pointers (objects) that will be filled.
883  * @param n
884  *   The number of objects to dequeue from the ring to the obj_table,
885  *   must be strictly positive.
886  * @return
887  *   - 0: Success; objects dequeued.
888  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
889  *     dequeued.
890  */
891 static inline int __attribute__((always_inline))
892 rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
893 {
894         return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
895 }
896
897 /**
898  * Dequeue several objects from a ring.
899  *
900  * This function calls the multi-consumers or the single-consumer
901  * version, depending on the default behaviour that was specified at
902  * ring creation time (see flags).
903  *
904  * @param r
905  *   A pointer to the ring structure.
906  * @param obj_table
907  *   A pointer to a table of void * pointers (objects) that will be filled.
908  * @param n
909  *   The number of objects to dequeue from the ring to the obj_table.
910  * @return
911  *   - 0: Success; objects dequeued.
912  *   - -ENOENT: Not enough entries in the ring to dequeue, no object is
913  *     dequeued.
914  */
915 static inline int __attribute__((always_inline))
916 rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
917 {
918         if (r->cons.sc_dequeue)
919                 return rte_ring_sc_dequeue_bulk(r, obj_table, n);
920         else
921                 return rte_ring_mc_dequeue_bulk(r, obj_table, n);
922 }
923
924 /**
925  * Dequeue one object from a ring (multi-consumers safe).
926  *
927  * This function uses a "compare and set" instruction to move the
928  * consumer index atomically.
929  *
930  * @param r
931  *   A pointer to the ring structure.
932  * @param obj_p
933  *   A pointer to a void * pointer (object) that will be filled.
934  * @return
935  *   - 0: Success; objects dequeued.
936  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
937  *     dequeued.
938  */
939 static inline int __attribute__((always_inline))
940 rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
941 {
942         return rte_ring_mc_dequeue_bulk(r, obj_p, 1);
943 }
944
945 /**
946  * Dequeue one object from a ring (NOT multi-consumers safe).
947  *
948  * @param r
949  *   A pointer to the ring structure.
950  * @param obj_p
951  *   A pointer to a void * pointer (object) that will be filled.
952  * @return
953  *   - 0: Success; objects dequeued.
954  *   - -ENOENT: Not enough entries in the ring to dequeue, no object is
955  *     dequeued.
956  */
957 static inline int __attribute__((always_inline))
958 rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
959 {
960         return rte_ring_sc_dequeue_bulk(r, obj_p, 1);
961 }
962
963 /**
964  * Dequeue one object from a ring.
965  *
966  * This function calls the multi-consumers or the single-consumer
967  * version depending on the default behaviour that was specified at
968  * ring creation time (see flags).
969  *
970  * @param r
971  *   A pointer to the ring structure.
972  * @param obj_p
973  *   A pointer to a void * pointer (object) that will be filled.
974  * @return
975  *   - 0: Success, objects dequeued.
976  *   - -ENOENT: Not enough entries in the ring to dequeue, no object is
977  *     dequeued.
978  */
979 static inline int __attribute__((always_inline))
980 rte_ring_dequeue(struct rte_ring *r, void **obj_p)
981 {
982         if (r->cons.sc_dequeue)
983                 return rte_ring_sc_dequeue(r, obj_p);
984         else
985                 return rte_ring_mc_dequeue(r, obj_p);
986 }
987
988 /**
989  * Test if a ring is full.
990  *
991  * @param r
992  *   A pointer to the ring structure.
993  * @return
994  *   - 1: The ring is full.
995  *   - 0: The ring is not full.
996  */
997 static inline int
998 rte_ring_full(const struct rte_ring *r)
999 {
1000         uint32_t prod_tail = r->prod.tail;
1001         uint32_t cons_tail = r->cons.tail;
1002         return (((cons_tail - prod_tail - 1) & r->prod.mask) == 0);
1003 }
1004
1005 /**
1006  * Test if a ring is empty.
1007  *
1008  * @param r
1009  *   A pointer to the ring structure.
1010  * @return
1011  *   - 1: The ring is empty.
1012  *   - 0: The ring is not empty.
1013  */
1014 static inline int
1015 rte_ring_empty(const struct rte_ring *r)
1016 {
1017         uint32_t prod_tail = r->prod.tail;
1018         uint32_t cons_tail = r->cons.tail;
1019         return !!(cons_tail == prod_tail);
1020 }
1021
1022 /**
1023  * Return the number of entries in a ring.
1024  *
1025  * @param r
1026  *   A pointer to the ring structure.
1027  * @return
1028  *   The number of entries in the ring.
1029  */
1030 static inline unsigned
1031 rte_ring_count(const struct rte_ring *r)
1032 {
1033         uint32_t prod_tail = r->prod.tail;
1034         uint32_t cons_tail = r->cons.tail;
1035         return ((prod_tail - cons_tail) & r->prod.mask);
1036 }
1037
1038 /**
1039  * Return the number of free entries in a ring.
1040  *
1041  * @param r
1042  *   A pointer to the ring structure.
1043  * @return
1044  *   The number of free entries in the ring.
1045  */
1046 static inline unsigned
1047 rte_ring_free_count(const struct rte_ring *r)
1048 {
1049         uint32_t prod_tail = r->prod.tail;
1050         uint32_t cons_tail = r->cons.tail;
1051         return ((cons_tail - prod_tail - 1) & r->prod.mask);
1052 }
1053
1054 /**
1055  * Dump the status of all rings on the console
1056  */
1057 void rte_ring_list_dump(void);
1058
1059 /**
1060  * Search a ring from its name
1061  *
1062  * @param name
1063  *   The name of the ring.
1064  * @return
1065  *   The pointer to the ring matching the name, or NULL if not found,
1066  *   with rte_errno set appropriately. Possible rte_errno values include:
1067  *    - ENOENT - required entry not available to return.
1068  */
1069 struct rte_ring *rte_ring_lookup(const char *name);
1070
1071 /**
1072  * Enqueue several objects on the ring (multi-producers safe).
1073  *
1074  * This function uses a "compare and set" instruction to move the
1075  * producer index atomically.
1076  *
1077  * @param r
1078  *   A pointer to the ring structure.
1079  * @param obj_table
1080  *   A pointer to a table of void * pointers (objects).
1081  * @param n
1082  *   The number of objects to add in the ring from the obj_table.
1083  * @return
1084  *   - n: Actual number of objects enqueued.
1085  */
1086 static inline int __attribute__((always_inline))
1087 rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
1088                          unsigned n)
1089 {
1090         return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
1091 }
1092
1093 /**
1094  * Enqueue several objects on a ring (NOT multi-producers safe).
1095  *
1096  * @param r
1097  *   A pointer to the ring structure.
1098  * @param obj_table
1099  *   A pointer to a table of void * pointers (objects).
1100  * @param n
1101  *   The number of objects to add in the ring from the obj_table.
1102  * @return
1103  *   - n: Actual number of objects enqueued.
1104  */
1105 static inline int __attribute__((always_inline))
1106 rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
1107                          unsigned n)
1108 {
1109         return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
1110 }
1111
1112 /**
1113  * Enqueue several objects on a ring.
1114  *
1115  * This function calls the multi-producer or the single-producer
1116  * version depending on the default behavior that was specified at
1117  * ring creation time (see flags).
1118  *
1119  * @param r
1120  *   A pointer to the ring structure.
1121  * @param obj_table
1122  *   A pointer to a table of void * pointers (objects).
1123  * @param n
1124  *   The number of objects to add in the ring from the obj_table.
1125  * @return
1126  *   - n: Actual number of objects enqueued.
1127  */
1128 static inline int __attribute__((always_inline))
1129 rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
1130                       unsigned n)
1131 {
1132         if (r->prod.sp_enqueue)
1133                 return  rte_ring_sp_enqueue_burst(r, obj_table, n);
1134         else
1135                 return  rte_ring_mp_enqueue_burst(r, obj_table, n);
1136 }
1137
1138 /**
1139  * Dequeue several objects from a ring (multi-consumers safe). When the request
1140  * objects are more than the available objects, only dequeue the actual number
1141  * of objects
1142  *
1143  * This function uses a "compare and set" instruction to move the
1144  * consumer index atomically.
1145  *
1146  * @param r
1147  *   A pointer to the ring structure.
1148  * @param obj_table
1149  *   A pointer to a table of void * pointers (objects) that will be filled.
1150  * @param n
1151  *   The number of objects to dequeue from the ring to the obj_table.
1152  * @return
1153  *   - n: Actual number of objects dequeued, 0 if ring is empty
1154  */
1155 static inline int __attribute__((always_inline))
1156 rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
1157 {
1158         return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
1159 }
1160
1161 /**
1162  * Dequeue several objects from a ring (NOT multi-consumers safe).When the
1163  * request objects are more than the available objects, only dequeue the
1164  * actual number of objects
1165  *
1166  * @param r
1167  *   A pointer to the ring structure.
1168  * @param obj_table
1169  *   A pointer to a table of void * pointers (objects) that will be filled.
1170  * @param n
1171  *   The number of objects to dequeue from the ring to the obj_table.
1172  * @return
1173  *   - n: Actual number of objects dequeued, 0 if ring is empty
1174  */
1175 static inline int __attribute__((always_inline))
1176 rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
1177 {
1178         return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
1179 }
1180
1181 /**
1182  * Dequeue multiple objects from a ring up to a maximum number.
1183  *
1184  * This function calls the multi-consumers or the single-consumer
1185  * version, depending on the default behaviour that was specified at
1186  * ring creation time (see flags).
1187  *
1188  * @param r
1189  *   A pointer to the ring structure.
1190  * @param obj_table
1191  *   A pointer to a table of void * pointers (objects) that will be filled.
1192  * @param n
1193  *   The number of objects to dequeue from the ring to the obj_table.
1194  * @return
1195  *   - Number of objects dequeued, or a negative error code on error
1196  */
1197 static inline int __attribute__((always_inline))
1198 rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
1199 {
1200         if (r->cons.sc_dequeue)
1201                 return rte_ring_sc_dequeue_burst(r, obj_table, n);
1202         else
1203                 return rte_ring_mc_dequeue_burst(r, obj_table, n);
1204 }
1205
1206 #ifdef __cplusplus
1207 }
1208 #endif
1209
1210 #endif /* _RTE_RING_H_ */