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