lib/librte_ring/rte_ring.h

   1 /*-
   2  *   BSD LICENSE
   3  *
   4  *   Copyright(c) 2010-2012 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 /*
  36  * Derived from FreeBSD's bufring.h
  37  *
  38  **************************************************************************
  39  *
  40  * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
  41  * All rights reserved.
  42  *
  43  * Redistribution and use in source and binary forms, with or without
  44  * modification, are permitted provided that the following conditions are met:
  45  *
  46  * 1. Redistributions of source code must retain the above copyright notice,
  47  *    this list of conditions and the following disclaimer.
  48  *
  49  * 2. The name of Kip Macy nor the names of other
  50  *    contributors may be used to endorse or promote products derived from
  51  *    this software without specific prior written permission.
  52  *
  53  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  54  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  55  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  56  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  57  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  58  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  59  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  60  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  61  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  62  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  63  * POSSIBILITY OF SUCH DAMAGE.
  64  *
  65  ***************************************************************************/
  66
  67 #ifndef _RTE_RING_H_
  68 #define _RTE_RING_H_
  69
  70 /**
  71  * @file
  72  * RTE Ring
  73  *
  74  * The Ring Manager is a fixed-size queue, implemented as a table of
  75  * pointers. Head and tail pointers are modified atomically, allowing
  76  * concurrent access to it. It has the following features:
  77  *
  78  * - FIFO (First In First Out)
  79  * - Maximum size is fixed; the pointers are stored in a table.
  80  * - Lockless implementation.
  81  * - Multi- or single-consumer dequeue.
  82  * - Multi- or single-producer enqueue.
  83  * - Bulk dequeue.
  84  * - Bulk enqueue.
  85  *
  86  * Note: the ring implementation is not preemptable. A lcore must not
  87  * be interrupted by another task that uses the same ring.
  88  *
  89  */
  90
  91 #ifdef __cplusplus
  92 extern "C" {
  93 #endif
  94
  95 #include <stdint.h>
  96 #include <sys/queue.h>
  97 #include <errno.h>
  98 #include <rte_common.h>
  99 #include <rte_memory.h>
 100 #include <rte_lcore.h>
 101 #include <rte_atomic.h>
 102 #include <rte_branch_prediction.h>
 103
 104
 105 #ifdef RTE_LIBRTE_RING_DEBUG
 106 /**
 107  * A structure that stores the ring statistics (per-lcore).
 108  */
 109 struct rte_ring_debug_stats {
 110         uint64_t enq_success_bulk; /**< Successful enqueues number. */
 111         uint64_t enq_success_objs; /**< Objects successfully enqueued. */
 112         uint64_t enq_quota_bulk;   /**< Successful enqueues above watermark. */
 113         uint64_t enq_quota_objs;   /**< Objects enqueued above watermark. */
 114         uint64_t enq_fail_bulk;    /**< Failed enqueues number. */
 115         uint64_t enq_fail_objs;    /**< Objects that failed to be enqueued. */
 116         uint64_t deq_success_bulk; /**< Successful dequeues number. */
 117         uint64_t deq_success_objs; /**< Objects successfully dequeued. */
 118         uint64_t deq_fail_bulk;    /**< Failed dequeues number. */
 119         uint64_t deq_fail_objs;    /**< Objects that failed to be dequeued. */
 120 } __rte_cache_aligned;
 121 #endif
 122
 123 #define RTE_RING_NAMESIZE 32 /**< The maximum length of a ring name. */
 124
 125 /**
 126  * An RTE ring structure.
 127  *
 128  * The producer and the consumer have a head and a tail index. The particularity
 129  * of these index is that they are not between 0 and size(ring). These indexes
 130  * are between 0 and 2^32, and we mask their value when we access the ring[]
 131  * field. Thanks to this assumption, we can do subtractions between 2 index
 132  * values in a modulo-32bit base: that's why the overflow of the indexes is not
 133  * a problem.
 134  */
 135 struct rte_ring {
 136         TAILQ_ENTRY(rte_ring) next;      /**< Next in list. */
 137
 138         char name[RTE_RING_NAMESIZE];    /**< Name of the ring. */
 139         int flags;                       /**< Flags supplied at creation. */
 140
 141         /** Ring producer status. */
 142         struct prod {
 143                 volatile uint32_t bulk_default; /**< Default bulk count. */
 144                 uint32_t watermark;      /**< Maximum items before EDQUOT. */
 145                 uint32_t sp_enqueue;     /**< True, if single producer. */
 146                 uint32_t size;           /**< Size of ring. */
 147                 uint32_t mask;           /**< Mask (size-1) of ring. */
 148                 volatile uint32_t head;  /**< Producer head. */
 149                 volatile uint32_t tail;  /**< Producer tail. */
 150         } prod __rte_cache_aligned;
 151
 152         /** Ring consumer status. */
 153         struct cons {
 154                 volatile uint32_t bulk_default; /**< Default bulk count. */
 155                 uint32_t sc_dequeue;     /**< True, if single consumer. */
 156                 uint32_t size;           /**< Size of the ring. */
 157                 uint32_t mask;           /**< Mask (size-1) of ring. */
 158                 volatile uint32_t head;  /**< Consumer head. */
 159                 volatile uint32_t tail;  /**< Consumer tail. */
 160         } cons __rte_cache_aligned;
 161
 162
 163 #ifdef RTE_LIBRTE_RING_DEBUG
 164         struct rte_ring_debug_stats stats[RTE_MAX_LCORE];
 165 #endif
 166
 167         void * volatile ring[0] \
 168                         __rte_cache_aligned; /**< Memory space of ring starts here. */
 169 };
 170
 171 #define RING_F_SP_ENQ 0x0001 /**< The default enqueue is "single-producer". */
 172 #define RING_F_SC_DEQ 0x0002 /**< The default dequeue is "single-consumer". */
 173
 174 /**
 175  * When debug is enabled, store ring statistics.
 176  * @param r
 177  *   A pointer to the ring.
 178  * @param name
 179  *   The name of the statistics field to increment in the ring.
 180  * @param n
 181  *   The number to add to the object-oriented statistics.
 182  */
 183 #ifdef RTE_LIBRTE_RING_DEBUG
 184 #define __RING_STAT_ADD(r, name, n) do {                \
 185                 unsigned __lcore_id = rte_lcore_id();   \
 186                 r->stats[__lcore_id].name##_objs += n;  \
 187                 r->stats[__lcore_id].name##_bulk += 1;  \
 188         } while(0)
 189 #else
 190 #define __RING_STAT_ADD(r, name, n) do {} while(0)
 191 #endif
 192
 193 /**
 194  * Create a new ring named *name* in memory.
 195  *
 196  * This function uses ``memzone_reserve()`` to allocate memory. Its size is
 197  * set to *count*, which must be a power of two. Water marking is
 198  * disabled by default. The default bulk count is initialized to 1.
 199  * Note that the real usable ring size is *count-1* instead of
 200  * *count*.
 201  *
 202  * @param name
 203  *   The name of the ring.
 204  * @param count
 205  *   The size of the ring (must be a power of 2).
 206  * @param socket_id
 207  *   The *socket_id* argument is the socket identifier in case of
 208  *   NUMA. The value can be *SOCKET_ID_ANY* if there is no NUMA
 209  *   constraint for the reserved zone.
 210  * @param flags
 211  *   An OR of the following:
 212  *    - RING_F_SP_ENQ: If this flag is set, the default behavior when
 213  *      using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
 214  *      is "single-producer". Otherwise, it is "multi-producers".
 215  *    - RING_F_SC_DEQ: If this flag is set, the default behavior when
 216  *      using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
 217  *      is "single-consumer". Otherwise, it is "multi-consumers".
 218  * @return
 219  *   On success, the pointer to the new allocated ring. NULL on error with
 220  *    rte_errno set appropriately. Possible errno values include:
 221  *    - E_RTE_NO_CONFIG - function could not get pointer to rte_config structure
 222  *    - E_RTE_SECONDARY - function was called from a secondary process instance
 223  *    - E_RTE_NO_TAILQ - no tailq list could be got for the ring list
 224  *    - EINVAL - count provided is not a power of 2
 225  *    - ENOSPC - the maximum number of memzones has already been allocated
 226  *    - EEXIST - a memzone with the same name already exists
 227  *    - ENOMEM - no appropriate memory area found in which to create memzone
 228  */
 229 struct rte_ring *rte_ring_create(const char *name, unsigned count,
 230                                  int socket_id, unsigned flags);
 231
 232 /**
 233  * Set the default bulk count for enqueue/dequeue.
 234  *
 235  * The parameter *count* is the default number of bulk elements to
 236  * get/put when using ``rte_ring_*_{en,de}queue_bulk()``. It must be
 237  * greater than 0 and less than half of the ring size.
 238  *
 239  * @param r
 240  *   A pointer to the ring structure.
 241  * @param count
 242  *   A new water mark value.
 243  * @return
 244  *   - 0: Success; default_bulk_count changed.
 245  *   - -EINVAL: Invalid count value.
 246  */
 247 static inline int
 248 rte_ring_set_bulk_count(struct rte_ring *r, unsigned count)
 249 {
 250         if (unlikely(count == 0 || count >= r->prod.size))
 251                 return -EINVAL;
 252
 253         r->prod.bulk_default = r->cons.bulk_default = count;
 254         return 0;
 255 }
 256
 257 /**
 258  * Get the default bulk count for enqueue/dequeue.
 259  *
 260  * @param r
 261  *   A pointer to the ring structure.
 262  * @return
 263  *   The default bulk count for enqueue/dequeue.
 264  */
 265 static inline unsigned
 266 rte_ring_get_bulk_count(struct rte_ring *r)
 267 {
 268         return r->prod.bulk_default;
 269 }
 270
 271 /**
 272  * Change the high water mark.
 273  *
 274  * If *count* is 0, water marking is disabled. Otherwise, it is set to the
 275  * *count* value. The *count* value must be greater than 0 and less
 276  * than the ring size.
 277  *
 278  * This function can be called at any time (not necessarilly at
 279  * initialization).
 280  *
 281  * @param r
 282  *   A pointer to the ring structure.
 283  * @param count
 284  *   The new water mark value.
 285  * @return
 286  *   - 0: Success; water mark changed.
 287  *   - -EINVAL: Invalid water mark value.
 288  */
 289 int rte_ring_set_water_mark(struct rte_ring *r, unsigned count);
 290
 291 /**
 292  * Dump the status of the ring to the console.
 293  *
 294  * @param r
 295  *   A pointer to the ring structure.
 296  */
 297 void rte_ring_dump(const struct rte_ring *r);
 298
 299 /**
 300  * Enqueue several objects on the ring (multi-producers safe).
 301  *
 302  * This function uses a "compare and set" instruction to move the
 303  * producer index atomically.
 304  *
 305  * @param r
 306  *   A pointer to the ring structure.
 307  * @param obj_table
 308  *   A pointer to a table of void * pointers (objects).
 309  * @param n
 310  *   The number of objects to add in the ring from the obj_table. The
 311  *   value must be strictly positive.
 312  * @return
 313  *   - 0: Success; objects enqueue.
 314  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
 315  *     high water mark is exceeded.
 316  *   - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
 317  */
 318 static inline int
 319 rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
 320                          unsigned n)
 321 {
 322         uint32_t prod_head, prod_next;
 323         uint32_t cons_tail, free_entries;
 324         int success;
 325         unsigned i;
 326         uint32_t mask = r->prod.mask;
 327         int ret;
 328
 329         /* move prod.head atomically */
 330         do {
 331                 prod_head = r->prod.head;
 332                 cons_tail = r->cons.tail;
 333                 /* The subtraction is done between two unsigned 32bits value
 334                  * (the result is always modulo 32 bits even if we have
 335                  * prod_head > cons_tail). So 'free_entries' is always between 0
 336                  * and size(ring)-1. */
 337                 free_entries = (mask + cons_tail - prod_head);
 338
 339                 /* check that we have enough room in ring */
 340                 if (unlikely(n > free_entries)) {
 341                         __RING_STAT_ADD(r, enq_fail, n);
 342                         return -ENOBUFS;
 343                 }
 344
 345                 prod_next = prod_head + n;
 346                 success = rte_atomic32_cmpset(&r->prod.head, prod_head,
 347                                               prod_next);
 348         } while (unlikely(success == 0));
 349
 350         /* write entries in ring */
 351         for (i = 0; likely(i < n); i++)
 352                 r->ring[(prod_head + i) & mask] = obj_table[i];
 353         rte_wmb();
 354
 355         /* return -EDQUOT if we exceed the watermark */
 356         if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
 357                 ret = -EDQUOT;
 358                 __RING_STAT_ADD(r, enq_quota, n);
 359         }
 360         else {
 361                 ret = 0;
 362                 __RING_STAT_ADD(r, enq_success, n);
 363         }
 364
 365         /*
 366          * If there are other enqueues in progress that preceeded us,
 367          * we need to wait for them to complete
 368          */
 369         while (unlikely(r->prod.tail != prod_head))
 370                 rte_pause();
 371
 372         r->prod.tail = prod_next;
 373         return ret;
 374 }
 375
 376 /**
 377  * Enqueue several objects on a ring (NOT multi-producers safe).
 378  *
 379  * @param r
 380  *   A pointer to the ring structure.
 381  * @param obj_table
 382  *   A pointer to a table of void * pointers (objects).
 383  * @param n
 384  *   The number of objects to add in the ring from the obj_table. The
 385  *   value must be strictly positive.
 386  * @return
 387  *   - 0: Success; objects enqueued.
 388  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
 389  *     high water mark is exceeded.
 390  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
 391  */
 392 static inline int
 393 rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
 394                          unsigned n)
 395 {
 396         uint32_t prod_head, cons_tail;
 397         uint32_t prod_next, free_entries;
 398         unsigned i;
 399         uint32_t mask = r->prod.mask;
 400         int ret;
 401
 402         prod_head = r->prod.head;
 403         cons_tail = r->cons.tail;
 404         /* The subtraction is done between two unsigned 32bits value
 405          * (the result is always modulo 32 bits even if we have
 406          * prod_head > cons_tail). So 'free_entries' is always between 0
 407          * and size(ring)-1. */
 408         free_entries = mask + cons_tail - prod_head;
 409
 410         /* check that we have enough room in ring */
 411         if (unlikely(n > free_entries)) {
 412                 __RING_STAT_ADD(r, enq_fail, n);
 413                 return -ENOBUFS;
 414         }
 415
 416         prod_next = prod_head + n;
 417         r->prod.head = prod_next;
 418
 419         /* write entries in ring */
 420         for (i = 0; likely(i < n); i++)
 421                 r->ring[(prod_head + i) & mask] = obj_table[i];
 422         rte_wmb();
 423
 424         /* return -EDQUOT if we exceed the watermark */
 425         if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
 426                 ret = -EDQUOT;
 427                 __RING_STAT_ADD(r, enq_quota, n);
 428         }
 429         else {
 430                 ret = 0;
 431                 __RING_STAT_ADD(r, enq_success, n);
 432         }
 433
 434         r->prod.tail = prod_next;
 435         return ret;
 436 }
 437
 438 /**
 439  * Enqueue several objects on a ring.
 440  *
 441  * This function calls the multi-producer or the single-producer
 442  * version depending on the default behavior that was specified at
 443  * ring creation time (see flags).
 444  *
 445  * @param r
 446  *   A pointer to the ring structure.
 447  * @param obj_table
 448  *   A pointer to a table of void * pointers (objects).
 449  * @param n
 450  *   The number of objects to add in the ring from the obj_table.
 451  * @return
 452  *   - 0: Success; objects enqueued.
 453  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
 454  *     high water mark is exceeded.
 455  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
 456  */
 457 static inline int
 458 rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
 459                       unsigned n)
 460 {
 461         if (r->prod.sp_enqueue)
 462                 return rte_ring_sp_enqueue_bulk(r, obj_table, n);
 463         else
 464                 return rte_ring_mp_enqueue_bulk(r, obj_table, n);
 465 }
 466
 467 /**
 468  * Enqueue one object on a ring (multi-producers safe).
 469  *
 470  * This function uses a "compare and set" instruction to move the
 471  * producer index atomically.
 472  *
 473  * @param r
 474  *   A pointer to the ring structure.
 475  * @param obj
 476  *   A pointer to the object to be added.
 477  * @return
 478  *   - 0: Success; objects enqueued.
 479  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
 480  *     high water mark is exceeded.
 481  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
 482  */
 483 static inline int
 484 rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
 485 {
 486         return rte_ring_mp_enqueue_bulk(r, &obj, 1);
 487 }
 488
 489 /**
 490  * Enqueue one object on a ring (NOT multi-producers safe).
 491  *
 492  * @param r
 493  *   A pointer to the ring structure.
 494  * @param obj
 495  *   A pointer to the object to be added.
 496  * @return
 497  *   - 0: Success; objects enqueued.
 498  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
 499  *     high water mark is exceeded.
 500  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
 501  */
 502 static inline int
 503 rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
 504 {
 505         return rte_ring_sp_enqueue_bulk(r, &obj, 1);
 506 }
 507
 508 /**
 509  * Enqueue one object on a ring.
 510  *
 511  * This function calls the multi-producer or the single-producer
 512  * version, depending on the default behaviour that was specified at
 513  * ring creation time (see flags).
 514  *
 515  * @param r
 516  *   A pointer to the ring structure.
 517  * @param obj
 518  *   A pointer to the object to be added.
 519  * @return
 520  *   - 0: Success; objects enqueued.
 521  *   - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
 522  *     high water mark is exceeded.
 523  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
 524  */
 525 static inline int
 526 rte_ring_enqueue(struct rte_ring *r, void *obj)
 527 {
 528         if (r->prod.sp_enqueue)
 529                 return rte_ring_sp_enqueue(r, obj);
 530         else
 531                 return rte_ring_mp_enqueue(r, obj);
 532 }
 533
 534 /**
 535  * Dequeue several objects from a ring (multi-consumers safe).
 536  *
 537  * This function uses a "compare and set" instruction to move the
 538  * consumer index atomically.
 539  *
 540  * @param r
 541  *   A pointer to the ring structure.
 542  * @param obj_table
 543  *   A pointer to a table of void * pointers (objects) that will be filled.
 544  * @param n
 545  *   The number of objects to dequeue from the ring to the obj_table,
 546  *   must be strictly positive
 547  * @return
 548  *   - 0: Success; objects dequeued.
 549  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
 550  *     dequeued.
 551  */
 552 static inline int
 553 rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
 554 {
 555         uint32_t cons_head, prod_tail;
 556         uint32_t cons_next, entries;
 557         int success;
 558         unsigned i;
 559         uint32_t mask = r->prod.mask;
 560
 561         /* move cons.head atomically */
 562         do {
 563                 cons_head = r->cons.head;
 564                 prod_tail = r->prod.tail;
 565                 /* The subtraction is done between two unsigned 32bits value
 566                  * (the result is always modulo 32 bits even if we have
 567                  * cons_head > prod_tail). So 'entries' is always between 0
 568                  * and size(ring)-1. */
 569                 entries = (prod_tail - cons_head);
 570
 571                 /* check that we have enough entries in ring */
 572                 if (unlikely(n > entries)) {
 573                         __RING_STAT_ADD(r, deq_fail, n);
 574                         return -ENOENT;
 575                 }
 576
 577                 cons_next = cons_head + n;
 578                 success = rte_atomic32_cmpset(&r->cons.head, cons_head,
 579                                               cons_next);
 580         } while (unlikely(success == 0));
 581
 582         /* copy in table */
 583         rte_rmb();
 584         for (i = 0; likely(i < n); i++) {
 585                 obj_table[i] = r->ring[(cons_head + i) & mask];
 586         }
 587
 588         /*
 589          * If there are other dequeues in progress that preceeded us,
 590          * we need to wait for them to complete
 591          */
 592         while (unlikely(r->cons.tail != cons_head))
 593                 rte_pause();
 594
 595         __RING_STAT_ADD(r, deq_success, n);
 596         r->cons.tail = cons_next;
 597         return 0;
 598 }
 599
 600 /**
 601  * Dequeue several objects from a ring (NOT multi-consumers safe).
 602  *
 603  * @param r
 604  *   A pointer to the ring structure.
 605  * @param obj_table
 606  *   A pointer to a table of void * pointers (objects) that will be filled.
 607  * @param n
 608  *   The number of objects to dequeue from the ring to the obj_table,
 609  *   must be strictly positive.
 610  * @return
 611  *   - 0: Success; objects dequeued.
 612  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
 613  *     dequeued.
 614  */
 615 static inline int
 616 rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
 617 {
 618         uint32_t cons_head, prod_tail;
 619         uint32_t cons_next, entries;
 620         unsigned i;
 621         uint32_t mask = r->prod.mask;
 622
 623         cons_head = r->cons.head;
 624         prod_tail = r->prod.tail;
 625         /* The subtraction is done between two unsigned 32bits value
 626          * (the result is always modulo 32 bits even if we have
 627          * cons_head > prod_tail). So 'entries' is always between 0
 628          * and size(ring)-1. */
 629         entries = prod_tail - cons_head;
 630
 631         /* check that we have enough entries in ring */
 632         if (unlikely(n > entries)) {
 633                 __RING_STAT_ADD(r, deq_fail, n);
 634                 return -ENOENT;
 635         }
 636
 637         cons_next = cons_head + n;
 638         r->cons.head = cons_next;
 639
 640         /* copy in table */
 641         rte_rmb();
 642         for (i = 0; likely(i < n); i++) {
 643                 obj_table[i] = r->ring[(cons_head + i) & mask];
 644         }
 645
 646         __RING_STAT_ADD(r, deq_success, n);
 647         r->cons.tail = cons_next;
 648         return 0;
 649 }
 650
 651 /**
 652  * Dequeue several objects from a ring.
 653  *
 654  * This function calls the multi-consumers or the single-consumer
 655  * version, depending on the default behaviour that was specified at
 656  * ring creation time (see flags).
 657  *
 658  * @param r
 659  *   A pointer to the ring structure.
 660  * @param obj_table
 661  *   A pointer to a table of void * pointers (objects) that will be filled.
 662  * @param n
 663  *   The number of objects to dequeue from the ring to the obj_table.
 664  * @return
 665  *   - 0: Success; objects dequeued.
 666  *   - -ENOENT: Not enough entries in the ring to dequeue, no object is
 667  *     dequeued.
 668  */
 669 static inline int
 670 rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
 671 {
 672         if (r->cons.sc_dequeue)
 673                 return rte_ring_sc_dequeue_bulk(r, obj_table, n);
 674         else
 675                 return rte_ring_mc_dequeue_bulk(r, obj_table, n);
 676 }
 677
 678 /**
 679  * Dequeue one object from a ring (multi-consumers safe).
 680  *
 681  * This function uses a "compare and set" instruction to move the
 682  * consumer index atomically.
 683  *
 684  * @param r
 685  *   A pointer to the ring structure.
 686  * @param obj_p
 687  *   A pointer to a void * pointer (object) that will be filled.
 688  * @return
 689  *   - 0: Success; objects dequeued.
 690  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
 691  *     dequeued.
 692  */
 693 static inline int
 694 rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
 695 {
 696         return rte_ring_mc_dequeue_bulk(r, obj_p, 1);
 697 }
 698
 699 /**
 700  * Dequeue one object from a ring (NOT multi-consumers safe).
 701  *
 702  * @param r
 703  *   A pointer to the ring structure.
 704  * @param obj_p
 705  *   A pointer to a void * pointer (object) that will be filled.
 706  * @return
 707  *   - 0: Success; objects dequeued.
 708  *   - -ENOENT: Not enough entries in the ring to dequeue, no object is
 709  *     dequeued.
 710  */
 711 static inline int
 712 rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
 713 {
 714         return rte_ring_sc_dequeue_bulk(r, obj_p, 1);
 715 }
 716
 717 /**
 718  * Dequeue one object from a ring.
 719  *
 720  * This function calls the multi-consumers or the single-consumer
 721  * version depending on the default behaviour that was specified at
 722  * ring creation time (see flags).
 723  *
 724  * @param r
 725  *   A pointer to the ring structure.
 726  * @param obj_p
 727  *   A pointer to a void * pointer (object) that will be filled.
 728  * @return
 729  *   - 0: Success, objects dequeued.
 730  *   - -ENOENT: Not enough entries in the ring to dequeue, no object is
 731  *     dequeued.
 732  */
 733 static inline int
 734 rte_ring_dequeue(struct rte_ring *r, void **obj_p)
 735 {
 736         if (r->cons.sc_dequeue)
 737                 return rte_ring_sc_dequeue(r, obj_p);
 738         else
 739                 return rte_ring_mc_dequeue(r, obj_p);
 740 }
 741
 742 /**
 743  * Test if a ring is full.
 744  *
 745  * @param r
 746  *   A pointer to the ring structure.
 747  * @return
 748  *   - 1: The ring is full.
 749  *   - 0: The ring is not full.
 750  */
 751 static inline int
 752 rte_ring_full(const struct rte_ring *r)
 753 {
 754         uint32_t prod_tail = r->prod.tail;
 755         uint32_t cons_tail = r->cons.tail;
 756         return (((cons_tail - prod_tail - 1) & r->prod.mask) == 0);
 757 }
 758
 759 /**
 760  * Test if a ring is empty.
 761  *
 762  * @param r
 763  *   A pointer to the ring structure.
 764  * @return
 765  *   - 1: The ring is empty.
 766  *   - 0: The ring is not empty.
 767  */
 768 static inline int
 769 rte_ring_empty(const struct rte_ring *r)
 770 {
 771         uint32_t prod_tail = r->prod.tail;
 772         uint32_t cons_tail = r->cons.tail;
 773         return !!(cons_tail == prod_tail);
 774 }
 775
 776 /**
 777  * Return the number of entries in a ring.
 778  *
 779  * @param r
 780  *   A pointer to the ring structure.
 781  * @return
 782  *   The number of entries in the ring.
 783  */
 784 static inline unsigned
 785 rte_ring_count(const struct rte_ring *r)
 786 {
 787         uint32_t prod_tail = r->prod.tail;
 788         uint32_t cons_tail = r->cons.tail;
 789         return ((prod_tail - cons_tail) & r->prod.mask);
 790 }
 791
 792 /**
 793  * Return the number of free entries in a ring.
 794  *
 795  * @param r
 796  *   A pointer to the ring structure.
 797  * @return
 798  *   The number of free entries in the ring.
 799  */
 800 static inline unsigned
 801 rte_ring_free_count(const struct rte_ring *r)
 802 {
 803         uint32_t prod_tail = r->prod.tail;
 804         uint32_t cons_tail = r->cons.tail;
 805         return ((cons_tail - prod_tail - 1) & r->prod.mask);
 806 }
 807
 808 /**
 809  * Dump the status of all rings on the console
 810  */
 811 void rte_ring_list_dump(void);
 812
 813 /**
 814  * Search a ring from its name
 815  *
 816  * @param name
 817  *   The name of the ring.
 818  * @return
 819  *   The pointer to the ring matching the name, or NULL if not found,
 820  *   with rte_errno set appropriately. Possible rte_errno values include:
 821  *    - ENOENT - required entry not available to return.
 822  */
 823 struct rte_ring *rte_ring_lookup(const char *name);
 824
 825 #ifdef __cplusplus
 826 }
 827 #endif
 828
 829 #endif /* _RTE_RING_H_ */