[PATCH v3 1/2] ring: make soring to finalize its own stage only

[Konstantin Ananyev <konstantin.ananyev@huawei.com>]

While finalize() is MT-safe and can be called from multiple places:
from 'acquire()' for next stage or even from consumer's 'dequeue(),
doing so usually creates extra un-necessary contention and might
slow-down ring operations, especially for the cases when we have
multiple threads doing acquire/release for some stage.
Things become worse when we have single producer/consumer and multiple
workers doing acquire/release for the same ring.
So instead of calling finalize() from different places, just make
release() for given stage to always try to perform it.

Accorging to the soring_stress_autotest, for multiple workers (8+)
it reduces number of cycles spent by 1.5x-2.0x factor.
For l3fwd-like workload it improves things by ~20%.
For small number of workers didn't observe any serious change.
As another benefit - it allows to simplify the code.

Signed-off-by: Konstantin Ananyev <konstantin.ananyev@huawei.com>
---
 lib/ring/soring.c | 52 ++++++++---------------------------------------
 1 file changed, 9 insertions(+), 43 deletions(-)

diff --git a/lib/ring/soring.c b/lib/ring/soring.c
index 3b90521bdb..fc7fc55a21 100644
--- a/lib/ring/soring.c
+++ b/lib/ring/soring.c
@@ -50,10 +50,8 @@
  * from current stage tail up to its head, check state[] and move stage tail
  * through elements that already are in SORING_ST_FINISH state.
  * Along with that, corresponding state[] values are reset to zero.
- * Note that 'finalize()' for given stage can be done from multiple places:
- * 'release()' for that stage or from 'acquire()' for next stage
- * even from consumer's 'dequeue()' - in case given stage is the last one.
- * So 'finalize()' has to be MT-safe and inside it we have to
+ * Note that 'finalize()' for given stage can be called from multiple threads
+ * in parallel, so 'finalize()' has to be MT-safe and inside it we have to
  * guarantee that only one thread will update state[] and stage's tail values.
  */
 
@@ -284,7 +282,7 @@ soring_dequeue(struct rte_soring *r, void *objs, void *meta,
 	uint32_t *available)
 {
 	enum rte_ring_sync_type st;
-	uint32_t entries, cons_head, cons_next, n, ns, reqn;
+	uint32_t entries, cons_head, cons_next, n, ns;
 
 	RTE_ASSERT(r != NULL && r->nb_stage > 0);
 	RTE_ASSERT(meta == NULL || r->meta != NULL);
@@ -293,22 +291,8 @@ soring_dequeue(struct rte_soring *r, void *objs, void *meta,
 	st = r->cons.ht.sync_type;
 
 	/* try to grab exactly @num elems first */
-	n = __rte_soring_move_cons_head(r, ns, num, RTE_RING_QUEUE_FIXED, st,
+	n = __rte_soring_move_cons_head(r, ns, num, behavior, st,
 			&cons_head, &cons_next, &entries);
-	if (n == 0) {
-		/* try to finalize some elems from previous stage */
-		n = __rte_soring_stage_finalize(&r->stage[ns].sht, ns,
-			r->state, r->mask, 2 * num);
-		entries += n;
-
-		/* repeat attempt to grab elems */
-		reqn = (behavior == RTE_RING_QUEUE_FIXED) ? num : 0;
-		if (entries >= reqn)
-			n = __rte_soring_move_cons_head(r, ns, num, behavior,
-				st, &cons_head, &cons_next, &entries);
-		else
-			n = 0;
-	}
 
 	/* we have some elems to consume */
 	if (n != 0) {
@@ -382,7 +366,7 @@ soring_acquire(struct rte_soring *r, void *objs, void *meta,
 	uint32_t stage, uint32_t num, enum rte_ring_queue_behavior behavior,
 	uint32_t *ftoken, uint32_t *available)
 {
-	uint32_t avail, head, idx, n, next, reqn;
+	uint32_t avail, head, idx, n, next;
 	struct soring_stage *pstg;
 	struct soring_stage_headtail *cons;
 
@@ -399,22 +383,7 @@ soring_acquire(struct rte_soring *r, void *objs, void *meta,
 
 		/* try to grab exactly @num elems */
 		n = __rte_soring_stage_move_head(cons, &pstg->ht, 0, num,
-			RTE_RING_QUEUE_FIXED, &head, &next, &avail);
-		if (n == 0) {
-			/* try to finalize some elems from previous stage */
-			n = __rte_soring_stage_finalize(&pstg->sht, stage - 1,
-				r->state, r->mask, 2 * num);
-			avail += n;
-
-			/* repeat attempt to grab elems */
-			reqn = (behavior == RTE_RING_QUEUE_FIXED) ? num : 0;
-			if (avail >= reqn)
-				n = __rte_soring_stage_move_head(cons,
-					&pstg->ht, 0, num, behavior, &head,
-					&next, &avail);
-			else
-				n = 0;
-		}
+			behavior, &head, &next, &avail);
 	}
 
 	if (n != 0) {
@@ -442,7 +411,7 @@ static __rte_always_inline void
 soring_release(struct rte_soring *r, const void *objs,
 	const void *meta, uint32_t stage, uint32_t n, uint32_t ftoken)
 {
-	uint32_t idx, pos, tail;
+	uint32_t idx, pos;
 	struct soring_stage *stg;
 	union soring_state st;
 
@@ -479,11 +448,8 @@ soring_release(struct rte_soring *r, const void *objs,
 	rte_atomic_store_explicit(&r->state[idx].raw, st.raw,
 			rte_memory_order_relaxed);
 
-	/* try to do finalize(), if appropriate */
-	tail = rte_atomic_load_explicit(&stg->sht.tail.pos,
-			rte_memory_order_relaxed);
-	if (tail == pos)
-		__rte_soring_stage_finalize(&stg->sht, stage, r->state, r->mask,
+	/* try to do finalize(), if possible */
+	__rte_soring_stage_finalize(&stg->sht, stage, r->state, r->mask,
 				r->capacity);
 }
 
-- 
2.51.0