Issuing a zero objects dequeue with a single consumer has no effect. Doing so with multiple consumers, can get more than one thread to succeed the compare-and-set operation and observe starvation or even deadlock in the while loop that checks for preceding dequeues. The problematic piece of code when n = 0:
cons_next = cons_head + n; success = rte_atomic32_cmpset(&r->cons.head, cons_head, cons_next); The same is possible on the enqueue path. Signed-off-by: Lazaros Koromilas <l at nofutznetworks.com> --- lib/librte_ring/rte_ring.h | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h index 943c97c..2bf9ce3 100644 --- a/lib/librte_ring/rte_ring.h +++ b/lib/librte_ring/rte_ring.h @@ -100,6 +100,7 @@ extern "C" { #include <rte_lcore.h> #include <rte_atomic.h> #include <rte_branch_prediction.h> +#include <rte_debug.h> #define RTE_TAILQ_RING_NAME "RTE_RING" @@ -211,6 +212,19 @@ struct rte_ring { #endif /** + * @internal Assert macro. + * @param exp + * The expression to evaluate. + */ +#define RTE_RING_ASSERT(exp) do { \ + if (!(exp)) { \ + rte_panic("line%d\t" \ + "assert \"" #exp "\" failed\n", \ + __LINE__); \ + } \ + } while (0) + +/** * Calculate the memory size needed for a ring * * This function returns the number of bytes needed for a ring, given @@ -406,6 +420,7 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r); * A pointer to a table of void * pointers (objects). * @param n * The number of objects to add in the ring from the obj_table. + * Must be greater than zero. * @param behavior * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring * RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring @@ -431,6 +446,8 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table, uint32_t mask = r->prod.mask; int ret; + RTE_RING_ASSERT(n > 0); + /* move prod.head atomically */ do { /* Reset n to the initial burst count */ @@ -510,6 +527,7 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table, * A pointer to a table of void * pointers (objects). * @param n * The number of objects to add in the ring from the obj_table. + * Must be greater than zero. * @param behavior * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring * RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring @@ -533,6 +551,8 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table, uint32_t mask = r->prod.mask; int ret; + RTE_RING_ASSERT(n > 0); + prod_head = r->prod.head; cons_tail = r->cons.tail; /* The subtraction is done between two unsigned 32bits value @@ -594,6 +614,7 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table, * A pointer to a table of void * pointers (objects) that will be filled. * @param n * The number of objects to dequeue from the ring to the obj_table. + * Must be greater than zero. * @param behavior * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring * RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring @@ -618,6 +639,8 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table, unsigned i, rep = 0; uint32_t mask = r->prod.mask; + RTE_RING_ASSERT(n > 0); + /* move cons.head atomically */ do { /* Restore n as it may change every loop */ @@ -689,6 +712,7 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table, * A pointer to a table of void * pointers (objects) that will be filled. * @param n * The number of objects to dequeue from the ring to the obj_table. + * Must be greater than zero. * @param behavior * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring * RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring @@ -710,6 +734,8 @@ __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table, unsigned i; uint32_t mask = r->prod.mask; + RTE_RING_ASSERT(n > 0); + cons_head = r->cons.head; prod_tail = r->prod.tail; /* The subtraction is done between two unsigned 32bits value -- 1.9.1