lib/librte_ring/rte_ring_elem.h

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