96 #include <sys/queue.h> 105 #define RTE_TAILQ_RING_NAME "RTE_RING" 107 enum rte_ring_queue_behavior {
108 RTE_RING_QUEUE_FIXED = 0,
109 RTE_RING_QUEUE_VARIABLE
112 #ifdef RTE_LIBRTE_RING_DEBUG 116 struct rte_ring_debug_stats {
117 uint64_t enq_success_bulk;
118 uint64_t enq_success_objs;
119 uint64_t enq_quota_bulk;
120 uint64_t enq_quota_objs;
121 uint64_t enq_fail_bulk;
122 uint64_t enq_fail_objs;
123 uint64_t deq_success_bulk;
124 uint64_t deq_success_objs;
125 uint64_t deq_fail_bulk;
126 uint64_t deq_fail_objs;
130 #define RTE_RING_MZ_PREFIX "RG_" 132 #define RTE_RING_NAMESIZE (RTE_MEMZONE_NAMESIZE - \ 133 sizeof(RTE_RING_MZ_PREFIX) + 1) 135 #ifndef RTE_RING_PAUSE_REP_COUNT 136 #define RTE_RING_PAUSE_REP_COUNT 0 169 volatile uint32_t head;
170 volatile uint32_t tail;
178 volatile uint32_t head;
179 volatile uint32_t tail;
180 #ifdef RTE_RING_SPLIT_PROD_CONS 186 #ifdef RTE_LIBRTE_RING_DEBUG 187 struct rte_ring_debug_stats stats[RTE_MAX_LCORE];
195 #define RING_F_SP_ENQ 0x0001 196 #define RING_F_SC_DEQ 0x0002 197 #define RTE_RING_QUOT_EXCEED (1 << 31) 198 #define RTE_RING_SZ_MASK (unsigned)(0x0fffffff) 209 #ifdef RTE_LIBRTE_RING_DEBUG 210 #define __RING_STAT_ADD(r, name, n) do { \ 211 unsigned __lcore_id = rte_lcore_id(); \ 212 if (__lcore_id < RTE_MAX_LCORE) { \ 213 r->stats[__lcore_id].name##_objs += n; \ 214 r->stats[__lcore_id].name##_bulk += 1; \ 218 #define __RING_STAT_ADD(r, name, n) do {} while(0) 314 int socket_id,
unsigned flags);
356 #define ENQUEUE_PTRS() do { \ 357 const uint32_t size = r->prod.size; \ 358 uint32_t idx = prod_head & mask; \ 359 if (likely(idx + n < size)) { \ 360 for (i = 0; i < (n & ((~(unsigned)0x3))); i+=4, idx+=4) { \ 361 r->ring[idx] = obj_table[i]; \ 362 r->ring[idx+1] = obj_table[i+1]; \ 363 r->ring[idx+2] = obj_table[i+2]; \ 364 r->ring[idx+3] = obj_table[i+3]; \ 367 case 3: r->ring[idx++] = obj_table[i++]; \ 368 case 2: r->ring[idx++] = obj_table[i++]; \ 369 case 1: r->ring[idx++] = obj_table[i++]; \ 372 for (i = 0; idx < size; i++, idx++)\ 373 r->ring[idx] = obj_table[i]; \ 374 for (idx = 0; i < n; i++, idx++) \ 375 r->ring[idx] = obj_table[i]; \ 382 #define DEQUEUE_PTRS() do { \ 383 uint32_t idx = cons_head & mask; \ 384 const uint32_t size = r->cons.size; \ 385 if (likely(idx + n < size)) { \ 386 for (i = 0; i < (n & (~(unsigned)0x3)); i+=4, idx+=4) {\ 387 obj_table[i] = r->ring[idx]; \ 388 obj_table[i+1] = r->ring[idx+1]; \ 389 obj_table[i+2] = r->ring[idx+2]; \ 390 obj_table[i+3] = r->ring[idx+3]; \ 393 case 3: obj_table[i++] = r->ring[idx++]; \ 394 case 2: obj_table[i++] = r->ring[idx++]; \ 395 case 1: obj_table[i++] = r->ring[idx++]; \ 398 for (i = 0; idx < size; i++, idx++) \ 399 obj_table[i] = r->ring[idx]; \ 400 for (idx = 0; i < n; i++, idx++) \ 401 obj_table[i] = r->ring[idx]; \ 430 static inline int __attribute__((always_inline))
431 __rte_ring_mp_do_enqueue(struct
rte_ring *r,
void * const *obj_table,
432 unsigned n, enum rte_ring_queue_behavior behavior)
434 uint32_t prod_head, prod_next;
435 uint32_t cons_tail, free_entries;
436 const unsigned max = n;
439 uint32_t mask = r->prod.mask;
452 prod_head = r->prod.head;
453 cons_tail = r->cons.tail;
458 free_entries = (mask + cons_tail - prod_head);
462 if (behavior == RTE_RING_QUEUE_FIXED) {
463 __RING_STAT_ADD(r, enq_fail, n);
469 __RING_STAT_ADD(r, enq_fail, n);
477 prod_next = prod_head + n;
487 if (
unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
488 ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT :
490 __RING_STAT_ADD(r, enq_quota, n);
493 ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n;
494 __RING_STAT_ADD(r, enq_success, n);
501 while (
unlikely(r->prod.tail != prod_head)) {
513 r->prod.tail = prod_next;
539 static inline int __attribute__((always_inline))
540 __rte_ring_sp_do_enqueue(struct
rte_ring *r,
void * const *obj_table,
541 unsigned n, enum rte_ring_queue_behavior behavior)
543 uint32_t prod_head, cons_tail;
544 uint32_t prod_next, free_entries;
546 uint32_t mask = r->prod.mask;
549 prod_head = r->prod.head;
550 cons_tail = r->cons.tail;
555 free_entries = mask + cons_tail - prod_head;
559 if (behavior == RTE_RING_QUEUE_FIXED) {
560 __RING_STAT_ADD(r, enq_fail, n);
566 __RING_STAT_ADD(r, enq_fail, n);
574 prod_next = prod_head + n;
575 r->prod.head = prod_next;
582 if (
unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
583 ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT :
585 __RING_STAT_ADD(r, enq_quota, n);
588 ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n;
589 __RING_STAT_ADD(r, enq_success, n);
592 r->prod.tail = prod_next;
623 static inline int __attribute__((always_inline))
624 __rte_ring_mc_do_dequeue(struct
rte_ring *r,
void **obj_table,
625 unsigned n, enum rte_ring_queue_behavior behavior)
627 uint32_t cons_head, prod_tail;
628 uint32_t cons_next, entries;
629 const unsigned max = n;
632 uint32_t mask = r->prod.mask;
644 cons_head = r->cons.head;
645 prod_tail = r->prod.tail;
650 entries = (prod_tail - cons_head);
654 if (behavior == RTE_RING_QUEUE_FIXED) {
655 __RING_STAT_ADD(r, deq_fail, n);
660 __RING_STAT_ADD(r, deq_fail, n);
668 cons_next = cons_head + n;
681 while (
unlikely(r->cons.tail != cons_head)) {
693 __RING_STAT_ADD(r, deq_success, n);
694 r->cons.tail = cons_next;
696 return behavior == RTE_RING_QUEUE_FIXED ? 0 : n;
722 static inline int __attribute__((always_inline))
723 __rte_ring_sc_do_dequeue(struct
rte_ring *r,
void **obj_table,
724 unsigned n, enum rte_ring_queue_behavior behavior)
726 uint32_t cons_head, prod_tail;
727 uint32_t cons_next, entries;
729 uint32_t mask = r->prod.mask;
731 cons_head = r->cons.head;
732 prod_tail = r->prod.tail;
737 entries = prod_tail - cons_head;
740 if (behavior == RTE_RING_QUEUE_FIXED) {
741 __RING_STAT_ADD(r, deq_fail, n);
746 __RING_STAT_ADD(r, deq_fail, n);
754 cons_next = cons_head + n;
755 r->cons.head = cons_next;
761 __RING_STAT_ADD(r, deq_success, n);
762 r->cons.tail = cons_next;
763 return behavior == RTE_RING_QUEUE_FIXED ? 0 : n;
784 static inline int __attribute__((always_inline))
788 return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
806 static inline int __attribute__((always_inline))
810 return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
832 static inline int __attribute__((always_inline))
858 static inline int __attribute__((always_inline))
877 static inline int __attribute__((always_inline))
900 static inline int __attribute__((always_inline))
926 static inline int __attribute__((always_inline))
929 return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
947 static inline int __attribute__((always_inline))
950 return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
971 static inline int __attribute__((always_inline))
995 static inline int __attribute__((always_inline))
1013 static inline int __attribute__((always_inline))
1035 static inline int __attribute__((always_inline))
1057 uint32_t cons_tail = r->cons.
tail;
1058 return ((cons_tail - prod_tail - 1) & r->
prod.
mask) == 0;
1074 uint32_t cons_tail = r->cons.
tail;
1075 return !!(cons_tail == prod_tail);
1086 static inline unsigned 1090 uint32_t cons_tail = r->cons.
tail;
1091 return (prod_tail - cons_tail) & r->
prod.
mask;
1102 static inline unsigned 1106 uint32_t cons_tail = r->cons.
tail;
1107 return (cons_tail - prod_tail - 1) & r->
prod.
mask;
1145 static inline unsigned __attribute__((always_inline))
1149 return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
1164 static inline unsigned __attribute__((always_inline))
1168 return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
1187 static inline unsigned __attribute__((always_inline))
1214 static inline unsigned __attribute__((always_inline))
1217 return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
1234 static inline unsigned __attribute__((always_inline))
1237 return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
1256 static inline unsigned __attribute__((always_inline))
static void rte_smp_rmb(void)
static int rte_atomic32_cmpset(volatile uint32_t *dst, uint32_t exp, uint32_t src)
static int rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
static int rte_ring_dequeue(struct rte_ring *r, void **obj_p)
static int rte_ring_mp_enqueue_bulk(struct rte_ring *r, void *const *obj_table, unsigned n)
int rte_ring_set_water_mark(struct rte_ring *r, unsigned count)
static int rte_ring_empty(const struct rte_ring *r)
static int rte_ring_sp_enqueue_bulk(struct rte_ring *r, void *const *obj_table, unsigned n)
void rte_ring_list_dump(FILE *f)
#define RTE_RING_QUOT_EXCEED
static int rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
static int rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
#define RTE_RING_PAUSE_REP_COUNT
static int rte_ring_enqueue_bulk(struct rte_ring *r, void *const *obj_table, unsigned n)
void rte_ring_free(struct rte_ring *r)
static unsigned rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
static unsigned rte_ring_sp_enqueue_burst(struct rte_ring *r, void *const *obj_table, unsigned n)
static void rte_smp_wmb(void)
char name[RTE_MEMZONE_NAMESIZE]
static int rte_ring_enqueue(struct rte_ring *r, void *obj)
static unsigned rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
static unsigned rte_ring_enqueue_burst(struct rte_ring *r, void *const *obj_table, unsigned n)
static int rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
void rte_ring_dump(FILE *f, const struct rte_ring *r)
static unsigned rte_ring_count(const struct rte_ring *r)
struct rte_ring * rte_ring_create(const char *name, unsigned count, int socket_id, unsigned flags)
static int rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
#define __rte_cache_aligned
static unsigned rte_ring_mp_enqueue_burst(struct rte_ring *r, void *const *obj_table, unsigned n)
struct rte_ring * rte_ring_lookup(const char *name)
static unsigned rte_ring_free_count(const struct rte_ring *r)
static int rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
static unsigned rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
int rte_ring_init(struct rte_ring *r, const char *name, unsigned count, unsigned flags)
static int rte_ring_full(const struct rte_ring *r)
static int rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
ssize_t rte_ring_get_memsize(unsigned count)
#define RTE_MEMZONE_NAMESIZE