lib/librte_ring/rte_ring.h

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  *
   3  * Copyright (c) 2010-2020 Intel Corporation
   4  * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
   5  * All rights reserved.
   6  * Derived from FreeBSD's bufring.h
   7  * Used as BSD-3 Licensed with permission from Kip Macy.
   8  */
   9
  10 #ifndef _RTE_RING_H_
  11 #define _RTE_RING_H_
  12
  13 /**
  14  * @file
  15  * RTE Ring
  16  *
  17  * The Ring Manager is a fixed-size queue, implemented as a table of
  18  * pointers. Head and tail pointers are modified atomically, allowing
  19  * concurrent access to it. It has the following features:
  20  *
  21  * - FIFO (First In First Out)
  22  * - Maximum size is fixed; the pointers are stored in a table.
  23  * - Lockless implementation.
  24  * - Multi- or single-consumer dequeue.
  25  * - Multi- or single-producer enqueue.
  26  * - Bulk dequeue.
  27  * - Bulk enqueue.
  28  * - Ability to select different sync modes for producer/consumer.
  29  * - Dequeue start/finish (depending on consumer sync modes).
  30  * - Enqueue start/finish (depending on producer sync mode).
  31  *
  32  * Note: the ring implementation is not preemptible. Refer to Programmer's
  33  * guide/Environment Abstraction Layer/Multiple pthread/Known Issues/rte_ring
  34  * for more information.
  35  *
  36  */
  37
  38 #ifdef __cplusplus
  39 extern "C" {
  40 #endif
  41
  42 #include <rte_ring_core.h>
  43 #include <rte_ring_elem.h>
  44
  45 /**
  46  * Calculate the memory size needed for a ring
  47  *
  48  * This function returns the number of bytes needed for a ring, given
  49  * the number of elements in it. This value is the sum of the size of
  50  * the structure rte_ring and the size of the memory needed by the
  51  * objects pointers. The value is aligned to a cache line size.
  52  *
  53  * @param count
  54  *   The number of elements in the ring (must be a power of 2).
  55  * @return
  56  *   - The memory size needed for the ring on success.
  57  *   - -EINVAL if count is not a power of 2.
  58  */
  59 ssize_t rte_ring_get_memsize(unsigned int count);
  60
  61 /**
  62  * Initialize a ring structure.
  63  *
  64  * Initialize a ring structure in memory pointed by "r". The size of the
  65  * memory area must be large enough to store the ring structure and the
  66  * object table. It is advised to use rte_ring_get_memsize() to get the
  67  * appropriate size.
  68  *
  69  * The ring size is set to *count*, which must be a power of two. Water
  70  * marking is disabled by default. The real usable ring size is
  71  * *count-1* instead of *count* to differentiate a free ring from an
  72  * empty ring.
  73  *
  74  * The ring is not added in RTE_TAILQ_RING global list. Indeed, the
  75  * memory given by the caller may not be shareable among dpdk
  76  * processes.
  77  *
  78  * @param r
  79  *   The pointer to the ring structure followed by the objects table.
  80  * @param name
  81  *   The name of the ring.
  82  * @param count
  83  *   The number of elements in the ring (must be a power of 2).
  84  * @param flags
  85  *   An OR of the following:
  86  *   - One of mutually exclusive flags that define producer behavior:
  87  *      - RING_F_SP_ENQ: If this flag is set, the default behavior when
  88  *        using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
  89  *        is "single-producer".
  90  *      - RING_F_MP_RTS_ENQ: If this flag is set, the default behavior when
  91  *        using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
  92  *        is "multi-producer RTS mode".
  93  *      - RING_F_MP_HTS_ENQ: If this flag is set, the default behavior when
  94  *        using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
  95  *        is "multi-producer HTS mode".
  96  *     If none of these flags is set, then default "multi-producer"
  97  *     behavior is selected.
  98  *   - One of mutually exclusive flags that define consumer behavior:
  99  *      - RING_F_SC_DEQ: If this flag is set, the default behavior when
 100  *        using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
 101  *        is "single-consumer". Otherwise, it is "multi-consumers".
 102  *      - RING_F_MC_RTS_DEQ: If this flag is set, the default behavior when
 103  *        using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
 104  *        is "multi-consumer RTS mode".
 105  *      - RING_F_MC_HTS_DEQ: If this flag is set, the default behavior when
 106  *        using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
 107  *        is "multi-consumer HTS mode".
 108  *     If none of these flags is set, then default "multi-consumer"
 109  *     behavior is selected.
 110  * @return
 111  *   0 on success, or a negative value on error.
 112  */
 113 int rte_ring_init(struct rte_ring *r, const char *name, unsigned int count,
 114         unsigned int flags);
 115
 116 /**
 117  * Create a new ring named *name* in memory.
 118  *
 119  * This function uses ``memzone_reserve()`` to allocate memory. Then it
 120  * calls rte_ring_init() to initialize an empty ring.
 121  *
 122  * The new ring size is set to *count*, which must be a power of
 123  * two. Water marking is disabled by default. The real usable ring size
 124  * is *count-1* instead of *count* to differentiate a free ring from an
 125  * empty ring.
 126  *
 127  * The ring is added in RTE_TAILQ_RING list.
 128  *
 129  * @param name
 130  *   The name of the ring.
 131  * @param count
 132  *   The size of the ring (must be a power of 2).
 133  * @param socket_id
 134  *   The *socket_id* argument is the socket identifier in case of
 135  *   NUMA. The value can be *SOCKET_ID_ANY* if there is no NUMA
 136  *   constraint for the reserved zone.
 137  * @param flags
 138  *   An OR of the following:
 139  *   - One of mutually exclusive flags that define producer behavior:
 140  *      - RING_F_SP_ENQ: If this flag is set, the default behavior when
 141  *        using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
 142  *        is "single-producer".
 143  *      - RING_F_MP_RTS_ENQ: If this flag is set, the default behavior when
 144  *        using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
 145  *        is "multi-producer RTS mode".
 146  *      - RING_F_MP_HTS_ENQ: If this flag is set, the default behavior when
 147  *        using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
 148  *        is "multi-producer HTS mode".
 149  *     If none of these flags is set, then default "multi-producer"
 150  *     behavior is selected.
 151  *   - One of mutually exclusive flags that define consumer behavior:
 152  *      - RING_F_SC_DEQ: If this flag is set, the default behavior when
 153  *        using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
 154  *        is "single-consumer". Otherwise, it is "multi-consumers".
 155  *      - RING_F_MC_RTS_DEQ: If this flag is set, the default behavior when
 156  *        using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
 157  *        is "multi-consumer RTS mode".
 158  *      - RING_F_MC_HTS_DEQ: If this flag is set, the default behavior when
 159  *        using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
 160  *        is "multi-consumer HTS mode".
 161  *     If none of these flags is set, then default "multi-consumer"
 162  *     behavior is selected.
 163  * @return
 164  *   On success, the pointer to the new allocated ring. NULL on error with
 165  *    rte_errno set appropriately. Possible errno values include:
 166  *    - E_RTE_NO_CONFIG - function could not get pointer to rte_config structure
 167  *    - E_RTE_SECONDARY - function was called from a secondary process instance
 168  *    - EINVAL - count provided is not a power of 2
 169  *    - ENOSPC - the maximum number of memzones has already been allocated
 170  *    - EEXIST - a memzone with the same name already exists
 171  *    - ENOMEM - no appropriate memory area found in which to create memzone
 172  */
 173 struct rte_ring *rte_ring_create(const char *name, unsigned int count,
 174                                  int socket_id, unsigned int flags);
 175
 176 /**
 177  * De-allocate all memory used by the ring.
 178  *
 179  * @param r
 180  *   Ring to free
 181  */
 182 void rte_ring_free(struct rte_ring *r);
 183
 184 /**
 185  * Dump the status of the ring to a file.
 186  *
 187  * @param f
 188  *   A pointer to a file for output
 189  * @param r
 190  *   A pointer to the ring structure.
 191  */
 192 void rte_ring_dump(FILE *f, const struct rte_ring *r);
 193
 194 /* the actual enqueue of pointers on the ring.
 195  * Placed here since identical code needed in both
 196  * single and multi producer enqueue functions */
 197 #define ENQUEUE_PTRS(r, ring_start, prod_head, obj_table, n, obj_type) do { \
 198         unsigned int i; \
 199         const uint32_t size = (r)->size; \
 200         uint32_t idx = prod_head & (r)->mask; \
 201         obj_type *ring = (obj_type *)ring_start; \
 202         if (likely(idx + n < size)) { \
 203                 for (i = 0; i < (n & ~0x3); i += 4, idx += 4) { \
 204                         ring[idx] = obj_table[i]; \
 205                         ring[idx + 1] = obj_table[i + 1]; \
 206                         ring[idx + 2] = obj_table[i + 2]; \
 207                         ring[idx + 3] = obj_table[i + 3]; \
 208                 } \
 209                 switch (n & 0x3) { \
 210                 case 3: \
 211                         ring[idx++] = obj_table[i++]; /* fallthrough */ \
 212                 case 2: \
 213                         ring[idx++] = obj_table[i++]; /* fallthrough */ \
 214                 case 1: \
 215                         ring[idx++] = obj_table[i++]; \
 216                 } \
 217         } else { \
 218                 for (i = 0; idx < size; i++, idx++)\
 219                         ring[idx] = obj_table[i]; \
 220                 for (idx = 0; i < n; i++, idx++) \
 221                         ring[idx] = obj_table[i]; \
 222         } \
 223 } while (0)
 224
 225 /* the actual copy of pointers on the ring to obj_table.
 226  * Placed here since identical code needed in both
 227  * single and multi consumer dequeue functions */
 228 #define DEQUEUE_PTRS(r, ring_start, cons_head, obj_table, n, obj_type) do { \
 229         unsigned int i; \
 230         uint32_t idx = cons_head & (r)->mask; \
 231         const uint32_t size = (r)->size; \
 232         obj_type *ring = (obj_type *)ring_start; \
 233         if (likely(idx + n < size)) { \
 234                 for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {\
 235                         obj_table[i] = ring[idx]; \
 236                         obj_table[i + 1] = ring[idx + 1]; \
 237                         obj_table[i + 2] = ring[idx + 2]; \
 238                         obj_table[i + 3] = ring[idx + 3]; \
 239                 } \
 240                 switch (n & 0x3) { \
 241                 case 3: \
 242                         obj_table[i++] = ring[idx++]; /* fallthrough */ \
 243                 case 2: \
 244                         obj_table[i++] = ring[idx++]; /* fallthrough */ \
 245                 case 1: \
 246                         obj_table[i++] = ring[idx++]; \
 247                 } \
 248         } else { \
 249                 for (i = 0; idx < size; i++, idx++) \
 250                         obj_table[i] = ring[idx]; \
 251                 for (idx = 0; i < n; i++, idx++) \
 252                         obj_table[i] = ring[idx]; \
 253         } \
 254 } while (0)
 255
 256 /* Between load and load. there might be cpu reorder in weak model
 257  * (powerpc/arm).
 258  * There are 2 choices for the users
 259  * 1.use rmb() memory barrier
 260  * 2.use one-direction load_acquire/store_release barrier,defined by
 261  * CONFIG_RTE_USE_C11_MEM_MODEL=y
 262  * It depends on performance test results.
 263  * By default, move common functions to rte_ring_generic.h
 264  */
 265 #ifdef RTE_USE_C11_MEM_MODEL
 266 #include "rte_ring_c11_mem.h"
 267 #else
 268 #include "rte_ring_generic.h"
 269 #endif
 270
 271 /**
 272  * @internal Enqueue several objects on the ring
 273  *
 274   * @param r
 275  *   A pointer to the ring structure.
 276  * @param obj_table
 277  *   A pointer to a table of void * pointers (objects).
 278  * @param n
 279  *   The number of objects to add in the ring from the obj_table.
 280  * @param behavior
 281  *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
 282  *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
 283  * @param is_sp
 284  *   Indicates whether to use single producer or multi-producer head update
 285  * @param free_space
 286  *   returns the amount of space after the enqueue operation has finished
 287  * @return
 288  *   Actual number of objects enqueued.
 289  *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
 290  */
 291 static __rte_always_inline unsigned int
 292 __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 293                  unsigned int n, enum rte_ring_queue_behavior behavior,
 294                  unsigned int is_sp, unsigned int *free_space)
 295 {
 296         uint32_t prod_head, prod_next;
 297         uint32_t free_entries;
 298
 299         n = __rte_ring_move_prod_head(r, is_sp, n, behavior,
 300                         &prod_head, &prod_next, &free_entries);
 301         if (n == 0)
 302                 goto end;
 303
 304         ENQUEUE_PTRS(r, &r[1], prod_head, obj_table, n, void *);
 305
 306         update_tail(&r->prod, prod_head, prod_next, is_sp, 1);
 307 end:
 308         if (free_space != NULL)
 309                 *free_space = free_entries - n;
 310         return n;
 311 }
 312
 313 /**
 314  * @internal Dequeue several objects from the ring
 315  *
 316  * @param r
 317  *   A pointer to the ring structure.
 318  * @param obj_table
 319  *   A pointer to a table of void * pointers (objects).
 320  * @param n
 321  *   The number of objects to pull from the ring.
 322  * @param behavior
 323  *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
 324  *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
 325  * @param is_sc
 326  *   Indicates whether to use single consumer or multi-consumer head update
 327  * @param available
 328  *   returns the number of remaining ring entries after the dequeue has finished
 329  * @return
 330  *   - Actual number of objects dequeued.
 331  *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
 332  */
 333 static __rte_always_inline unsigned int
 334 __rte_ring_do_dequeue(struct rte_ring *r, void **obj_table,
 335                  unsigned int n, enum rte_ring_queue_behavior behavior,
 336                  unsigned int is_sc, unsigned int *available)
 337 {
 338         uint32_t cons_head, cons_next;
 339         uint32_t entries;
 340
 341         n = __rte_ring_move_cons_head(r, (int)is_sc, n, behavior,
 342                         &cons_head, &cons_next, &entries);
 343         if (n == 0)
 344                 goto end;
 345
 346         DEQUEUE_PTRS(r, &r[1], cons_head, obj_table, n, void *);
 347
 348         update_tail(&r->cons, cons_head, cons_next, is_sc, 0);
 349
 350 end:
 351         if (available != NULL)
 352                 *available = entries - n;
 353         return n;
 354 }
 355
 356 /**
 357  * Enqueue several objects on the ring (multi-producers safe).
 358  *
 359  * This function uses a "compare and set" instruction to move the
 360  * producer index atomically.
 361  *
 362  * @param r
 363  *   A pointer to the ring structure.
 364  * @param obj_table
 365  *   A pointer to a table of void * pointers (objects).
 366  * @param n
 367  *   The number of objects to add in the ring from the obj_table.
 368  * @param free_space
 369  *   if non-NULL, returns the amount of space in the ring after the
 370  *   enqueue operation has finished.
 371  * @return
 372  *   The number of objects enqueued, either 0 or n
 373  */
 374 static __rte_always_inline unsigned int
 375 rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
 376                          unsigned int n, unsigned int *free_space)
 377 {
 378         return __rte_ring_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
 379                         RTE_RING_SYNC_MT, free_space);
 380 }
 381
 382 /**
 383  * Enqueue several objects on a ring (NOT multi-producers safe).
 384  *
 385  * @param r
 386  *   A pointer to the ring structure.
 387  * @param obj_table
 388  *   A pointer to a table of void * pointers (objects).
 389  * @param n
 390  *   The number of objects to add in the ring from the obj_table.
 391  * @param free_space
 392  *   if non-NULL, returns the amount of space in the ring after the
 393  *   enqueue operation has finished.
 394  * @return
 395  *   The number of objects enqueued, either 0 or n
 396  */
 397 static __rte_always_inline unsigned int
 398 rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
 399                          unsigned int n, unsigned int *free_space)
 400 {
 401         return __rte_ring_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
 402                         RTE_RING_SYNC_ST, free_space);
 403 }
 404
 405 /**
 406  * Enqueue several objects on a ring.
 407  *
 408  * This function calls the multi-producer or the single-producer
 409  * version depending on the default behavior that was specified at
 410  * ring creation time (see flags).
 411  *
 412  * @param r
 413  *   A pointer to the ring structure.
 414  * @param obj_table
 415  *   A pointer to a table of void * pointers (objects).
 416  * @param n
 417  *   The number of objects to add in the ring from the obj_table.
 418  * @param free_space
 419  *   if non-NULL, returns the amount of space in the ring after the
 420  *   enqueue operation has finished.
 421  * @return
 422  *   The number of objects enqueued, either 0 or n
 423  */
 424 static __rte_always_inline unsigned int
 425 rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
 426                       unsigned int n, unsigned int *free_space)
 427 {
 428         switch (r->prod.sync_type) {
 429         case RTE_RING_SYNC_MT:
 430                 return rte_ring_mp_enqueue_bulk(r, obj_table, n, free_space);
 431         case RTE_RING_SYNC_ST:
 432                 return rte_ring_sp_enqueue_bulk(r, obj_table, n, free_space);
 433 #ifdef ALLOW_EXPERIMENTAL_API
 434         case RTE_RING_SYNC_MT_RTS:
 435                 return rte_ring_mp_rts_enqueue_bulk(r, obj_table, n,
 436                         free_space);
 437         case RTE_RING_SYNC_MT_HTS:
 438                 return rte_ring_mp_hts_enqueue_bulk(r, obj_table, n,
 439                         free_space);
 440 #endif
 441         }
 442
 443         /* valid ring should never reach this point */
 444         RTE_ASSERT(0);
 445         return 0;
 446 }
 447
 448 /**
 449  * Enqueue one object on a ring (multi-producers safe).
 450  *
 451  * This function uses a "compare and set" instruction to move the
 452  * producer index atomically.
 453  *
 454  * @param r
 455  *   A pointer to the ring structure.
 456  * @param obj
 457  *   A pointer to the object to be added.
 458  * @return
 459  *   - 0: Success; objects enqueued.
 460  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
 461  */
 462 static __rte_always_inline int
 463 rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
 464 {
 465         return rte_ring_mp_enqueue_bulk(r, &obj, 1, NULL) ? 0 : -ENOBUFS;
 466 }
 467
 468 /**
 469  * Enqueue one object on a ring (NOT multi-producers safe).
 470  *
 471  * @param r
 472  *   A pointer to the ring structure.
 473  * @param obj
 474  *   A pointer to the object to be added.
 475  * @return
 476  *   - 0: Success; objects enqueued.
 477  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
 478  */
 479 static __rte_always_inline int
 480 rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
 481 {
 482         return rte_ring_sp_enqueue_bulk(r, &obj, 1, NULL) ? 0 : -ENOBUFS;
 483 }
 484
 485 /**
 486  * Enqueue one object on a ring.
 487  *
 488  * This function calls the multi-producer or the single-producer
 489  * version, depending on the default behaviour that was specified at
 490  * ring creation time (see flags).
 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  *   - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
 499  */
 500 static __rte_always_inline int
 501 rte_ring_enqueue(struct rte_ring *r, void *obj)
 502 {
 503         return rte_ring_enqueue_bulk(r, &obj, 1, NULL) ? 0 : -ENOBUFS;
 504 }
 505
 506 /**
 507  * Dequeue several objects from a ring (multi-consumers safe).
 508  *
 509  * This function uses a "compare and set" instruction to move the
 510  * consumer index atomically.
 511  *
 512  * @param r
 513  *   A pointer to the ring structure.
 514  * @param obj_table
 515  *   A pointer to a table of void * pointers (objects) that will be filled.
 516  * @param n
 517  *   The number of objects to dequeue from the ring to the obj_table.
 518  * @param available
 519  *   If non-NULL, returns the number of remaining ring entries after the
 520  *   dequeue has finished.
 521  * @return
 522  *   The number of objects dequeued, either 0 or n
 523  */
 524 static __rte_always_inline unsigned int
 525 rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table,
 526                 unsigned int n, unsigned int *available)
 527 {
 528         return __rte_ring_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
 529                         RTE_RING_SYNC_MT, available);
 530 }
 531
 532 /**
 533  * Dequeue several objects from a ring (NOT multi-consumers safe).
 534  *
 535  * @param r
 536  *   A pointer to the ring structure.
 537  * @param obj_table
 538  *   A pointer to a table of void * pointers (objects) that will be filled.
 539  * @param n
 540  *   The number of objects to dequeue from the ring to the obj_table,
 541  *   must be strictly positive.
 542  * @param available
 543  *   If non-NULL, returns the number of remaining ring entries after the
 544  *   dequeue has finished.
 545  * @return
 546  *   The number of objects dequeued, either 0 or n
 547  */
 548 static __rte_always_inline unsigned int
 549 rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table,
 550                 unsigned int n, unsigned int *available)
 551 {
 552         return __rte_ring_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
 553                         RTE_RING_SYNC_ST, available);
 554 }
 555
 556 /**
 557  * Dequeue several objects from a ring.
 558  *
 559  * This function calls the multi-consumers or the single-consumer
 560  * version, depending on the default behaviour that was specified at
 561  * ring creation time (see flags).
 562  *
 563  * @param r
 564  *   A pointer to the ring structure.
 565  * @param obj_table
 566  *   A pointer to a table of void * pointers (objects) that will be filled.
 567  * @param n
 568  *   The number of objects to dequeue from the ring to the obj_table.
 569  * @param available
 570  *   If non-NULL, returns the number of remaining ring entries after the
 571  *   dequeue has finished.
 572  * @return
 573  *   The number of objects dequeued, either 0 or n
 574  */
 575 static __rte_always_inline unsigned int
 576 rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned int n,
 577                 unsigned int *available)
 578 {
 579         switch (r->cons.sync_type) {
 580         case RTE_RING_SYNC_MT:
 581                 return rte_ring_mc_dequeue_bulk(r, obj_table, n, available);
 582         case RTE_RING_SYNC_ST:
 583                 return rte_ring_sc_dequeue_bulk(r, obj_table, n, available);
 584 #ifdef ALLOW_EXPERIMENTAL_API
 585         case RTE_RING_SYNC_MT_RTS:
 586                 return rte_ring_mc_rts_dequeue_bulk(r, obj_table, n, available);
 587         case RTE_RING_SYNC_MT_HTS:
 588                 return rte_ring_mc_hts_dequeue_bulk(r, obj_table, n, available);
 589 #endif
 590         }
 591
 592         /* valid ring should never reach this point */
 593         RTE_ASSERT(0);
 594         return 0;
 595 }
 596
 597 /**
 598  * Dequeue one object from a ring (multi-consumers safe).
 599  *
 600  * This function uses a "compare and set" instruction to move the
 601  * consumer index atomically.
 602  *
 603  * @param r
 604  *   A pointer to the ring structure.
 605  * @param obj_p
 606  *   A pointer to a void * pointer (object) that will be filled.
 607  * @return
 608  *   - 0: Success; objects dequeued.
 609  *   - -ENOENT: Not enough entries in the ring to dequeue; no object is
 610  *     dequeued.
 611  */
 612 static __rte_always_inline int
 613 rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
 614 {
 615         return rte_ring_mc_dequeue_bulk(r, obj_p, 1, NULL)  ? 0 : -ENOENT;
 616 }
 617
 618 /**
 619  * Dequeue one object from a ring (NOT multi-consumers safe).
 620  *
 621  * @param r
 622  *   A pointer to the ring structure.
 623  * @param obj_p
 624  *   A pointer to a void * pointer (object) that will be filled.
 625  * @return
 626  *   - 0: Success; objects dequeued.
 627  *   - -ENOENT: Not enough entries in the ring to dequeue, no object is
 628  *     dequeued.
 629  */
 630 static __rte_always_inline int
 631 rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
 632 {
 633         return rte_ring_sc_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOENT;
 634 }
 635
 636 /**
 637  * Dequeue one object from a ring.
 638  *
 639  * This function calls the multi-consumers or the single-consumer
 640  * version depending on the default behaviour that was specified at
 641  * ring creation time (see flags).
 642  *
 643  * @param r
 644  *   A pointer to the ring structure.
 645  * @param obj_p
 646  *   A pointer to a void * pointer (object) that will be filled.
 647  * @return
 648  *   - 0: Success, objects dequeued.
 649  *   - -ENOENT: Not enough entries in the ring to dequeue, no object is
 650  *     dequeued.
 651  */
 652 static __rte_always_inline int
 653 rte_ring_dequeue(struct rte_ring *r, void **obj_p)
 654 {
 655         return rte_ring_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOENT;
 656 }
 657
 658 /**
 659  * Flush a ring.
 660  *
 661  * This function flush all the elements in a ring
 662  *
 663  * @warning
 664  * Make sure the ring is not in use while calling this function.
 665  *
 666  * @param r
 667  *   A pointer to the ring structure.
 668  */
 669 void
 670 rte_ring_reset(struct rte_ring *r);
 671
 672 /**
 673  * Return the number of entries in a ring.
 674  *
 675  * @param r
 676  *   A pointer to the ring structure.
 677  * @return
 678  *   The number of entries in the ring.
 679  */
 680 static inline unsigned int
 681 rte_ring_count(const struct rte_ring *r)
 682 {
 683         uint32_t prod_tail = r->prod.tail;
 684         uint32_t cons_tail = r->cons.tail;
 685         uint32_t count = (prod_tail - cons_tail) & r->mask;
 686         return (count > r->capacity) ? r->capacity : count;
 687 }
 688
 689 /**
 690  * Return the number of free entries in a ring.
 691  *
 692  * @param r
 693  *   A pointer to the ring structure.
 694  * @return
 695  *   The number of free entries in the ring.
 696  */
 697 static inline unsigned int
 698 rte_ring_free_count(const struct rte_ring *r)
 699 {
 700         return r->capacity - rte_ring_count(r);
 701 }
 702
 703 /**
 704  * Test if a ring is full.
 705  *
 706  * @param r
 707  *   A pointer to the ring structure.
 708  * @return
 709  *   - 1: The ring is full.
 710  *   - 0: The ring is not full.
 711  */
 712 static inline int
 713 rte_ring_full(const struct rte_ring *r)
 714 {
 715         return rte_ring_free_count(r) == 0;
 716 }
 717
 718 /**
 719  * Test if a ring is empty.
 720  *
 721  * @param r
 722  *   A pointer to the ring structure.
 723  * @return
 724  *   - 1: The ring is empty.
 725  *   - 0: The ring is not empty.
 726  */
 727 static inline int
 728 rte_ring_empty(const struct rte_ring *r)
 729 {
 730         uint32_t prod_tail = r->prod.tail;
 731         uint32_t cons_tail = r->cons.tail;
 732         return cons_tail == prod_tail;
 733 }
 734
 735 /**
 736  * Return the size of the ring.
 737  *
 738  * @param r
 739  *   A pointer to the ring structure.
 740  * @return
 741  *   The size of the data store used by the ring.
 742  *   NOTE: this is not the same as the usable space in the ring. To query that
 743  *   use ``rte_ring_get_capacity()``.
 744  */
 745 static inline unsigned int
 746 rte_ring_get_size(const struct rte_ring *r)
 747 {
 748         return r->size;
 749 }
 750
 751 /**
 752  * Return the number of elements which can be stored in the ring.
 753  *
 754  * @param r
 755  *   A pointer to the ring structure.
 756  * @return
 757  *   The usable size of the ring.
 758  */
 759 static inline unsigned int
 760 rte_ring_get_capacity(const struct rte_ring *r)
 761 {
 762         return r->capacity;
 763 }
 764
 765 /**
 766  * Return sync type used by producer in the ring.
 767  *
 768  * @param r
 769  *   A pointer to the ring structure.
 770  * @return
 771  *   Producer sync type value.
 772  */
 773 static inline enum rte_ring_sync_type
 774 rte_ring_get_prod_sync_type(const struct rte_ring *r)
 775 {
 776         return r->prod.sync_type;
 777 }
 778
 779 /**
 780  * Check is the ring for single producer.
 781  *
 782  * @param r
 783  *   A pointer to the ring structure.
 784  * @return
 785  *   true if ring is SP, zero otherwise.
 786  */
 787 static inline int
 788 rte_ring_is_prod_single(const struct rte_ring *r)
 789 {
 790         return (rte_ring_get_prod_sync_type(r) == RTE_RING_SYNC_ST);
 791 }
 792
 793 /**
 794  * Return sync type used by consumer in the ring.
 795  *
 796  * @param r
 797  *   A pointer to the ring structure.
 798  * @return
 799  *   Consumer sync type value.
 800  */
 801 static inline enum rte_ring_sync_type
 802 rte_ring_get_cons_sync_type(const struct rte_ring *r)
 803 {
 804         return r->cons.sync_type;
 805 }
 806
 807 /**
 808  * Check is the ring for single consumer.
 809  *
 810  * @param r
 811  *   A pointer to the ring structure.
 812  * @return
 813  *   true if ring is SC, zero otherwise.
 814  */
 815 static inline int
 816 rte_ring_is_cons_single(const struct rte_ring *r)
 817 {
 818         return (rte_ring_get_cons_sync_type(r) == RTE_RING_SYNC_ST);
 819 }
 820
 821 /**
 822  * Dump the status of all rings on the console
 823  *
 824  * @param f
 825  *   A pointer to a file for output
 826  */
 827 void rte_ring_list_dump(FILE *f);
 828
 829 /**
 830  * Search a ring from its name
 831  *
 832  * @param name
 833  *   The name of the ring.
 834  * @return
 835  *   The pointer to the ring matching the name, or NULL if not found,
 836  *   with rte_errno set appropriately. Possible rte_errno values include:
 837  *    - ENOENT - required entry not available to return.
 838  */
 839 struct rte_ring *rte_ring_lookup(const char *name);
 840
 841 /**
 842  * Enqueue several objects on the ring (multi-producers safe).
 843  *
 844  * This function uses a "compare and set" instruction to move the
 845  * producer index atomically.
 846  *
 847  * @param r
 848  *   A pointer to the ring structure.
 849  * @param obj_table
 850  *   A pointer to a table of void * pointers (objects).
 851  * @param n
 852  *   The number of objects to add in the ring from the obj_table.
 853  * @param free_space
 854  *   if non-NULL, returns the amount of space in the ring after the
 855  *   enqueue operation has finished.
 856  * @return
 857  *   - n: Actual number of objects enqueued.
 858  */
 859 static __rte_always_inline unsigned int
 860 rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
 861                          unsigned int n, unsigned int *free_space)
 862 {
 863         return __rte_ring_do_enqueue(r, obj_table, n,
 864                         RTE_RING_QUEUE_VARIABLE, RTE_RING_SYNC_MT, free_space);
 865 }
 866
 867 /**
 868  * Enqueue several objects on a ring (NOT multi-producers safe).
 869  *
 870  * @param r
 871  *   A pointer to the ring structure.
 872  * @param obj_table
 873  *   A pointer to a table of void * pointers (objects).
 874  * @param n
 875  *   The number of objects to add in the ring from the obj_table.
 876  * @param free_space
 877  *   if non-NULL, returns the amount of space in the ring after the
 878  *   enqueue operation has finished.
 879  * @return
 880  *   - n: Actual number of objects enqueued.
 881  */
 882 static __rte_always_inline unsigned int
 883 rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
 884                          unsigned int n, unsigned int *free_space)
 885 {
 886         return __rte_ring_do_enqueue(r, obj_table, n,
 887                         RTE_RING_QUEUE_VARIABLE, RTE_RING_SYNC_ST, free_space);
 888 }
 889
 890 /**
 891  * Enqueue several objects on a ring.
 892  *
 893  * This function calls the multi-producer or the single-producer
 894  * version depending on the default behavior that was specified at
 895  * ring creation time (see flags).
 896  *
 897  * @param r
 898  *   A pointer to the ring structure.
 899  * @param obj_table
 900  *   A pointer to a table of void * pointers (objects).
 901  * @param n
 902  *   The number of objects to add in the ring from the obj_table.
 903  * @param free_space
 904  *   if non-NULL, returns the amount of space in the ring after the
 905  *   enqueue operation has finished.
 906  * @return
 907  *   - n: Actual number of objects enqueued.
 908  */
 909 static __rte_always_inline unsigned int
 910 rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
 911                       unsigned int n, unsigned int *free_space)
 912 {
 913         switch (r->prod.sync_type) {
 914         case RTE_RING_SYNC_MT:
 915                 return rte_ring_mp_enqueue_burst(r, obj_table, n, free_space);
 916         case RTE_RING_SYNC_ST:
 917                 return rte_ring_sp_enqueue_burst(r, obj_table, n, free_space);
 918 #ifdef ALLOW_EXPERIMENTAL_API
 919         case RTE_RING_SYNC_MT_RTS:
 920                 return rte_ring_mp_rts_enqueue_burst(r, obj_table, n,
 921                         free_space);
 922         case RTE_RING_SYNC_MT_HTS:
 923                 return rte_ring_mp_hts_enqueue_burst(r, obj_table, n,
 924                         free_space);
 925 #endif
 926         }
 927
 928         /* valid ring should never reach this point */
 929         RTE_ASSERT(0);
 930         return 0;
 931 }
 932
 933 /**
 934  * Dequeue several objects from a ring (multi-consumers safe). When the request
 935  * objects are more than the available objects, only dequeue the actual number
 936  * of objects
 937  *
 938  * This function uses a "compare and set" instruction to move the
 939  * consumer index atomically.
 940  *
 941  * @param r
 942  *   A pointer to the ring structure.
 943  * @param obj_table
 944  *   A pointer to a table of void * pointers (objects) that will be filled.
 945  * @param n
 946  *   The number of objects to dequeue from the ring to the obj_table.
 947  * @param available
 948  *   If non-NULL, returns the number of remaining ring entries after the
 949  *   dequeue has finished.
 950  * @return
 951  *   - n: Actual number of objects dequeued, 0 if ring is empty
 952  */
 953 static __rte_always_inline unsigned int
 954 rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table,
 955                 unsigned int n, unsigned int *available)
 956 {
 957         return __rte_ring_do_dequeue(r, obj_table, n,
 958                         RTE_RING_QUEUE_VARIABLE, RTE_RING_SYNC_MT, available);
 959 }
 960
 961 /**
 962  * Dequeue several objects from a ring (NOT multi-consumers safe).When the
 963  * request objects are more than the available objects, only dequeue the
 964  * actual number of objects
 965  *
 966  * @param r
 967  *   A pointer to the ring structure.
 968  * @param obj_table
 969  *   A pointer to a table of void * pointers (objects) that will be filled.
 970  * @param n
 971  *   The number of objects to dequeue from the ring to the obj_table.
 972  * @param available
 973  *   If non-NULL, returns the number of remaining ring entries after the
 974  *   dequeue has finished.
 975  * @return
 976  *   - n: Actual number of objects dequeued, 0 if ring is empty
 977  */
 978 static __rte_always_inline unsigned int
 979 rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table,
 980                 unsigned int n, unsigned int *available)
 981 {
 982         return __rte_ring_do_dequeue(r, obj_table, n,
 983                         RTE_RING_QUEUE_VARIABLE, RTE_RING_SYNC_ST, available);
 984 }
 985
 986 /**
 987  * Dequeue multiple objects from a ring up to a maximum number.
 988  *
 989  * This function calls the multi-consumers or the single-consumer
 990  * version, depending on the default behaviour that was specified at
 991  * ring creation time (see flags).
 992  *
 993  * @param r
 994  *   A pointer to the ring structure.
 995  * @param obj_table
 996  *   A pointer to a table of void * pointers (objects) that will be filled.
 997  * @param n
 998  *   The number of objects to dequeue from the ring to the obj_table.
 999  * @param available
1000  *   If non-NULL, returns the number of remaining ring entries after the
1001  *   dequeue has finished.
1002  * @return
1003  *   - Number of objects dequeued
1004  */
1005 static __rte_always_inline unsigned int
1006 rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table,
1007                 unsigned int n, unsigned int *available)
1008 {
1009         switch (r->cons.sync_type) {
1010         case RTE_RING_SYNC_MT:
1011                 return rte_ring_mc_dequeue_burst(r, obj_table, n, available);
1012         case RTE_RING_SYNC_ST:
1013                 return rte_ring_sc_dequeue_burst(r, obj_table, n, available);
1014 #ifdef ALLOW_EXPERIMENTAL_API
1015         case RTE_RING_SYNC_MT_RTS:
1016                 return rte_ring_mc_rts_dequeue_burst(r, obj_table, n,
1017                         available);
1018         case RTE_RING_SYNC_MT_HTS:
1019                 return rte_ring_mc_hts_dequeue_burst(r, obj_table, n,
1020                         available);
1021 #endif
1022         }
1023
1024         /* valid ring should never reach this point */
1025         RTE_ASSERT(0);
1026         return 0;
1027 }
1028
1029 #ifdef __cplusplus
1030 }
1031 #endif
1032
1033 #endif /* _RTE_RING_H_ */