* Placed here since identical code needed in both
* single and multi producer enqueue functions */
#define ENQUEUE_PTRS() do { \
+ unsigned int i; \
const uint32_t size = r->size; \
uint32_t idx = prod_head & mask; \
if (likely(idx + n < size)) { \
* Placed here since identical code needed in both
* single and multi consumer dequeue functions */
#define DEQUEUE_PTRS() do { \
+ unsigned int i; \
uint32_t idx = cons_head & mask; \
const uint32_t size = r->size; \
if (likely(idx + n < size)) { \
*/
static inline unsigned int __attribute__((always_inline))
__rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
+ unsigned int n, enum rte_ring_queue_behavior behavior,
+ unsigned int *free_space)
{
uint32_t prod_head, prod_next;
uint32_t cons_tail, free_entries;
- const unsigned max = n;
+ const unsigned int max = n;
int success;
- unsigned int i;
uint32_t mask = r->mask;
- /* Avoid the unnecessary cmpset operation below, which is also
- * potentially harmful when n equals 0. */
- if (n == 0)
- return 0;
-
/* move prod.head atomically */
do {
/* Reset n to the initial burst count */
free_entries = (mask + cons_tail - prod_head);
/* check that we have enough room in ring */
- if (unlikely(n > free_entries)) {
- if (behavior == RTE_RING_QUEUE_FIXED)
- return 0;
- else {
- /* No free entry available */
- if (unlikely(free_entries == 0))
- return 0;
- n = free_entries;
- }
- }
+ if (unlikely(n > free_entries))
+ n = (behavior == RTE_RING_QUEUE_FIXED) ?
+ 0 : free_entries;
+
+ if (n == 0)
+ goto end;
prod_next = prod_head + n;
success = rte_atomic32_cmpset(&r->prod.head, prod_head,
rte_pause();
r->prod.tail = prod_next;
+end:
+ if (free_space != NULL)
+ *free_space = free_entries - n;
return n;
}
*/
static inline unsigned int __attribute__((always_inline))
__rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
+ unsigned int n, enum rte_ring_queue_behavior behavior,
+ unsigned int *free_space)
{
uint32_t prod_head, cons_tail;
uint32_t prod_next, free_entries;
- unsigned int i;
uint32_t mask = r->mask;
prod_head = r->prod.head;
free_entries = mask + cons_tail - prod_head;
/* check that we have enough room in ring */
- if (unlikely(n > free_entries)) {
- if (behavior == RTE_RING_QUEUE_FIXED)
- return 0;
- else {
- /* No free entry available */
- if (unlikely(free_entries == 0))
- return 0;
- n = free_entries;
- }
- }
+ if (unlikely(n > free_entries))
+ n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : free_entries;
+
+ if (n == 0)
+ goto end;
+
prod_next = prod_head + n;
r->prod.head = prod_next;
rte_smp_wmb();
r->prod.tail = prod_next;
+end:
+ if (free_space != NULL)
+ *free_space = free_entries - n;
return n;
}
static inline unsigned int __attribute__((always_inline))
__rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
+ unsigned int n, enum rte_ring_queue_behavior behavior,
+ unsigned int *available)
{
uint32_t cons_head, prod_tail;
uint32_t cons_next, entries;
const unsigned max = n;
int success;
- unsigned int i;
uint32_t mask = r->mask;
- /* Avoid the unnecessary cmpset operation below, which is also
- * potentially harmful when n equals 0. */
- if (n == 0)
- return 0;
-
/* move cons.head atomically */
do {
/* Restore n as it may change every loop */
entries = (prod_tail - cons_head);
/* Set the actual entries for dequeue */
- if (n > entries) {
- if (behavior == RTE_RING_QUEUE_FIXED)
- return 0;
- else {
- if (unlikely(entries == 0))
- return 0;
- n = entries;
- }
- }
+ if (n > entries)
+ n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : entries;
+
+ if (unlikely(n == 0))
+ goto end;
cons_next = cons_head + n;
success = rte_atomic32_cmpset(&r->cons.head, cons_head,
rte_pause();
r->cons.tail = cons_next;
-
+end:
+ if (available != NULL)
+ *available = entries - n;
return n;
}
*/
static inline unsigned int __attribute__((always_inline))
__rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
+ unsigned int n, enum rte_ring_queue_behavior behavior,
+ unsigned int *available)
{
uint32_t cons_head, prod_tail;
uint32_t cons_next, entries;
- unsigned int i;
uint32_t mask = r->mask;
cons_head = r->cons.head;
* and size(ring)-1. */
entries = prod_tail - cons_head;
- if (n > entries) {
- if (behavior == RTE_RING_QUEUE_FIXED)
- return 0;
- else {
- if (unlikely(entries == 0))
- return 0;
- n = entries;
- }
- }
+ if (n > entries)
+ n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : entries;
+
+ if (unlikely(entries == 0))
+ goto end;
cons_next = cons_head + n;
r->cons.head = cons_next;
rte_smp_rmb();
r->cons.tail = cons_next;
+end:
+ if (available != NULL)
+ *available = entries - n;
return n;
}
* A pointer to a table of void * pointers (objects).
* @param n
* The number of objects to add in the ring from the obj_table.
+ * @param free_space
+ * if non-NULL, returns the amount of space in the ring after the
+ * enqueue operation has finished.
* @return
* The number of objects enqueued, either 0 or n
*/
static inline unsigned int __attribute__((always_inline))
rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
+ unsigned int n, unsigned int *free_space)
{
- return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+ return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
+ free_space);
}
/**
* A pointer to a table of void * pointers (objects).
* @param n
* The number of objects to add in the ring from the obj_table.
+ * @param free_space
+ * if non-NULL, returns the amount of space in the ring after the
+ * enqueue operation has finished.
* @return
* The number of objects enqueued, either 0 or n
*/
static inline unsigned int __attribute__((always_inline))
rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
+ unsigned int n, unsigned int *free_space)
{
- return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+ return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
+ free_space);
}
/**
* A pointer to a table of void * pointers (objects).
* @param n
* The number of objects to add in the ring from the obj_table.
+ * @param free_space
+ * if non-NULL, returns the amount of space in the ring after the
+ * enqueue operation has finished.
* @return
* The number of objects enqueued, either 0 or n
*/
static inline unsigned int __attribute__((always_inline))
rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
+ unsigned int n, unsigned int *free_space)
{
if (r->prod.single)
- return rte_ring_sp_enqueue_bulk(r, obj_table, n);
+ return rte_ring_sp_enqueue_bulk(r, obj_table, n, free_space);
else
- return rte_ring_mp_enqueue_bulk(r, obj_table, n);
+ return rte_ring_mp_enqueue_bulk(r, obj_table, n, free_space);
}
/**
static inline int __attribute__((always_inline))
rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
{
- return rte_ring_mp_enqueue_bulk(r, &obj, 1) ? 0 : -ENOBUFS;
+ return rte_ring_mp_enqueue_bulk(r, &obj, 1, NULL) ? 0 : -ENOBUFS;
}
/**
static inline int __attribute__((always_inline))
rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
{
- return rte_ring_sp_enqueue_bulk(r, &obj, 1) ? 0 : -ENOBUFS;
+ return rte_ring_sp_enqueue_bulk(r, &obj, 1, NULL) ? 0 : -ENOBUFS;
}
/**
static inline int __attribute__((always_inline))
rte_ring_enqueue(struct rte_ring *r, void *obj)
{
- return rte_ring_enqueue_bulk(r, &obj, 1) ? 0 : -ENOBUFS;
+ return rte_ring_enqueue_bulk(r, &obj, 1, NULL) ? 0 : -ENOBUFS;
}
/**
* 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.
+ * @param available
+ * If non-NULL, returns the number of remaining ring entries after the
+ * dequeue has finished.
* @return
* The number of objects dequeued, either 0 or n
*/
static inline unsigned int __attribute__((always_inline))
-rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table,
+ unsigned int n, unsigned int *available)
{
- return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+ return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
+ available);
}
/**
* @param n
* The number of objects to dequeue from the ring to the obj_table,
* must be strictly positive.
+ * @param available
+ * If non-NULL, returns the number of remaining ring entries after the
+ * dequeue has finished.
* @return
* The number of objects dequeued, either 0 or n
*/
static inline unsigned int __attribute__((always_inline))
-rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table,
+ unsigned int n, unsigned int *available)
{
- return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+ return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
+ available);
}
/**
* 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.
+ * @param available
+ * If non-NULL, returns the number of remaining ring entries after the
+ * dequeue has finished.
* @return
* The number of objects dequeued, either 0 or n
*/
static inline unsigned int __attribute__((always_inline))
-rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned int n,
+ unsigned int *available)
{
if (r->cons.single)
- return rte_ring_sc_dequeue_bulk(r, obj_table, n);
+ return rte_ring_sc_dequeue_bulk(r, obj_table, n, available);
else
- return rte_ring_mc_dequeue_bulk(r, obj_table, n);
+ return rte_ring_mc_dequeue_bulk(r, obj_table, n, available);
}
/**
static inline int __attribute__((always_inline))
rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
{
- return rte_ring_mc_dequeue_bulk(r, obj_p, 1) ? 0 : -ENOBUFS;
+ return rte_ring_mc_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOBUFS;
}
/**
static inline int __attribute__((always_inline))
rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
{
- return rte_ring_sc_dequeue_bulk(r, obj_p, 1) ? 0 : -ENOBUFS;
+ return rte_ring_sc_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOBUFS;
}
/**
static inline int __attribute__((always_inline))
rte_ring_dequeue(struct rte_ring *r, void **obj_p)
{
- return rte_ring_dequeue_bulk(r, obj_p, 1) ? 0 : -ENOBUFS;
+ return rte_ring_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOBUFS;
}
/**
* A pointer to a table of void * pointers (objects).
* @param n
* The number of objects to add in the ring from the obj_table.
+ * @param free_space
+ * if non-NULL, returns the amount of space in the ring after the
+ * enqueue operation has finished.
* @return
* - n: Actual number of objects enqueued.
*/
static inline unsigned __attribute__((always_inline))
rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
+ unsigned int n, unsigned int *free_space)
{
- return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+ return __rte_ring_mp_do_enqueue(r, obj_table, n,
+ RTE_RING_QUEUE_VARIABLE, free_space);
}
/**
* A pointer to a table of void * pointers (objects).
* @param n
* The number of objects to add in the ring from the obj_table.
+ * @param free_space
+ * if non-NULL, returns the amount of space in the ring after the
+ * enqueue operation has finished.
* @return
* - n: Actual number of objects enqueued.
*/
static inline unsigned __attribute__((always_inline))
rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
+ unsigned int n, unsigned int *free_space)
{
- return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+ return __rte_ring_sp_do_enqueue(r, obj_table, n,
+ RTE_RING_QUEUE_VARIABLE, free_space);
}
/**
* A pointer to a table of void * pointers (objects).
* @param n
* The number of objects to add in the ring from the obj_table.
+ * @param free_space
+ * if non-NULL, returns the amount of space in the ring after the
+ * enqueue operation has finished.
* @return
* - n: Actual number of objects enqueued.
*/
static inline unsigned __attribute__((always_inline))
rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
+ unsigned int n, unsigned int *free_space)
{
if (r->prod.single)
- return rte_ring_sp_enqueue_burst(r, obj_table, n);
+ return rte_ring_sp_enqueue_burst(r, obj_table, n, free_space);
else
- return rte_ring_mp_enqueue_burst(r, obj_table, n);
+ return rte_ring_mp_enqueue_burst(r, obj_table, n, free_space);
}
/**
* 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.
+ * @param available
+ * If non-NULL, returns the number of remaining ring entries after the
+ * dequeue has finished.
* @return
* - n: Actual number of objects dequeued, 0 if ring is empty
*/
static inline unsigned __attribute__((always_inline))
-rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table,
+ unsigned int n, unsigned int *available)
{
- return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+ return __rte_ring_mc_do_dequeue(r, obj_table, n,
+ RTE_RING_QUEUE_VARIABLE, available);
}
/**
* 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.
+ * @param available
+ * If non-NULL, returns the number of remaining ring entries after the
+ * dequeue has finished.
* @return
* - n: Actual number of objects dequeued, 0 if ring is empty
*/
static inline unsigned __attribute__((always_inline))
-rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table,
+ unsigned int n, unsigned int *available)
{
- return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+ return __rte_ring_sc_do_dequeue(r, obj_table, n,
+ RTE_RING_QUEUE_VARIABLE, available);
}
/**
* 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.
+ * @param available
+ * If non-NULL, returns the number of remaining ring entries after the
+ * dequeue has finished.
* @return
* - Number of objects dequeued
*/
static inline unsigned __attribute__((always_inline))
-rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table,
+ unsigned int n, unsigned int *available)
{
if (r->cons.single)
- return rte_ring_sc_dequeue_burst(r, obj_table, n);
+ return rte_ring_sc_dequeue_burst(r, obj_table, n, available);
else
- return rte_ring_mc_dequeue_burst(r, obj_table, n);
+ return rte_ring_mc_dequeue_burst(r, obj_table, n, available);
}
#ifdef __cplusplus
git.droids-corp.org / dpdk.git / blobdiff