Add a sched_yield() syscall if the thread spins for too long,
waiting other thread to finish its operations on the ring.
That gives pre-empted thread a chance to proceed and finish
with ring enqueue/dequeue operation.
The purpose is to reduce contention on the ring.
By ring_perf_test, it doesn't shows additional perf penalty.
Signed-off-by: Cunming Liang <cunming.liang@intel.com>
Acked-by: Olivier Matz <olivier.matz@6wind.com>
Acked-by: Konstantin Ananyev <konstantin.ananyev@intel.com>
CONFIG_RTE_LIBRTE_RING=y
CONFIG_RTE_LIBRTE_RING_DEBUG=n
CONFIG_RTE_RING_SPLIT_PROD_CONS=n
CONFIG_RTE_LIBRTE_RING=y
CONFIG_RTE_LIBRTE_RING_DEBUG=n
CONFIG_RTE_RING_SPLIT_PROD_CONS=n
+CONFIG_RTE_RING_PAUSE_REP_COUNT=0
#
# Compile librte_mempool
#
# Compile librte_mempool
CONFIG_RTE_LIBRTE_RING=y
CONFIG_RTE_LIBRTE_RING_DEBUG=n
CONFIG_RTE_RING_SPLIT_PROD_CONS=n
CONFIG_RTE_LIBRTE_RING=y
CONFIG_RTE_LIBRTE_RING_DEBUG=n
CONFIG_RTE_RING_SPLIT_PROD_CONS=n
+CONFIG_RTE_RING_PAUSE_REP_COUNT=0
#
# Compile librte_mempool
#
# Compile librte_mempool
#define RTE_RING_NAMESIZE 32 /**< The maximum length of a ring name. */
#define RTE_RING_MZ_PREFIX "RG_"
#define RTE_RING_NAMESIZE 32 /**< The maximum length of a ring name. */
#define RTE_RING_MZ_PREFIX "RG_"
+#ifndef RTE_RING_PAUSE_REP_COUNT
+#define RTE_RING_PAUSE_REP_COUNT 0 /**< Yield after pause num of times, no yield
+ * if RTE_RING_PAUSE_REP not defined. */
+#endif
+
/**
* An RTE ring structure.
*
/**
* An RTE ring structure.
*
uint32_t cons_tail, free_entries;
const unsigned max = n;
int success;
uint32_t cons_tail, free_entries;
const unsigned max = n;
int success;
uint32_t mask = r->prod.mask;
int ret;
uint32_t mask = r->prod.mask;
int ret;
* If there are other enqueues in progress that preceded us,
* we need to wait for them to complete
*/
* If there are other enqueues in progress that preceded us,
* we need to wait for them to complete
*/
- while (unlikely(r->prod.tail != prod_head))
+ while (unlikely(r->prod.tail != prod_head)) {
+ /* Set RTE_RING_PAUSE_REP_COUNT to avoid spin too long waiting
+ * for other thread finish. It gives pre-empted thread a chance
+ * to proceed and finish with ring dequeue operation. */
+ if (RTE_RING_PAUSE_REP_COUNT &&
+ ++rep == RTE_RING_PAUSE_REP_COUNT) {
+ rep = 0;
+ sched_yield();
+ }
+ }
r->prod.tail = prod_next;
return ret;
}
r->prod.tail = prod_next;
return ret;
}
uint32_t cons_next, entries;
const unsigned max = n;
int success;
uint32_t cons_next, entries;
const unsigned max = n;
int success;
uint32_t mask = r->prod.mask;
/* move cons.head atomically */
uint32_t mask = r->prod.mask;
/* move cons.head atomically */
* If there are other dequeues in progress that preceded us,
* we need to wait for them to complete
*/
* If there are other dequeues in progress that preceded us,
* we need to wait for them to complete
*/
- while (unlikely(r->cons.tail != cons_head))
+ while (unlikely(r->cons.tail != cons_head)) {
+ /* Set RTE_RING_PAUSE_REP_COUNT to avoid spin too long waiting
+ * for other thread finish. It gives pre-empted thread a chance
+ * to proceed and finish with ring dequeue operation. */
+ if (RTE_RING_PAUSE_REP_COUNT &&
+ ++rep == RTE_RING_PAUSE_REP_COUNT) {
+ rep = 0;
+ sched_yield();
+ }
+ }
__RING_STAT_ADD(r, deq_success, n);
r->cons.tail = cons_next;
__RING_STAT_ADD(r, deq_success, n);
r->cons.tail = cons_next;