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