|SUCCESS| [v3,2/2] ring: introduce peek API for soring

[dpdklab at iol.unh.edu]

Test-Label: ai-code-review
Test-Status: SUCCESS
https://dpdk.org/patch/163179

_AI Code Review_

=== Patch Review: series.patch (via Claude) ===
Review date: 2026-04-17

# Review of DPDK Patch Series

git sha: 20260417212358.212692-2-konstantin.ananyev at huawei.com
Author: Konstantin Ananyev <konstantin.ananyev at huawei.com>
Subject: [PATCH v3 1/2] ring: make soring to finalize its own stage only

This patch modifies the soring finalization logic to only be called from
the release function of a given stage, rather than from multiple
locations. This is intended to reduce contention in multi-worker
scenarios.

> 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

[ ... ]

> @@ -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;
> -	}

The removed retry logic in soring_dequeue attempts to finalize elements
and retry dequeue operations when the initial attempt fails. This removal
changes the API behavior. Does this code break the RTE_RING_QUEUE_VARIABLE
mode when elements are not yet finalized? Previously, the code would
actively try to finalize up to 2*num elements to make progress. Now,
if stage finalization is pending, does the dequeue immediately fail?

> @@ -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;

[ ... ]

> @@ -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);

Similar concern in soring_acquire - does removing the finalization retry
logic cause stage acquire operations to fail when they would have
previously succeeded after finalizing the previous stage?

> @@ -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;

[ ... ]

> @@ -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);

In soring_release, does removing the tail position check before calling
__rte_soring_stage_finalize cause unnecessary work? Previously the code
only attempted finalization when pos matched the tail. Now finalization
is attempted unconditionally on every release. Is the performance impact
of this change measured in scenarios with many concurrent release calls
where pos != tail?


git sha: 20260417212358.212692-3-konstantin.ananyev at huawei.com
Author: Konstantin Ananyev <konstantin.ananyev at huawei.com>
Subject: [PATCH v3 2/2] ring: introduce peek API for soring

This patch introduces peek-style APIs for soring rings, allowing split
enqueue/dequeue operations for certain synchronization types.

> diff --git a/app/test/test_soring_peek_stress.c b/app/test/test_soring_peek_stress.c
> new file mode 100644
> index 0000000000..6ac638f8d7
> --- /dev/null
> +++ b/app/test/test_soring_peek_stress.c

[ ... ]

> diff --git a/lib/ring/rte_soring.h b/lib/ring/rte_soring.h
> index 95c3cc4080..16f042e612 100644
> --- a/lib/ring/rte_soring.h
> +++ b/lib/ring/rte_soring.h

[ ... ]

> @@ -607,6 +607,270 @@ void
>  rte_soring_releasx(struct rte_soring *r, const void *objs,
>  	const void *meta, uint32_t stage, uint32_t n, uint32_t ftoken);
>  
> +/**
> + * SORING Peek API
> + * Same as with rte_ring, for some sync modes it is possible to split
> + * public enqueue/dequeue API into two phases:
> + * - enqueue/dequeue start
> + * - enqueue/dequeue finish
> + * That allows user to inspect objects in the soring without removing them
> + * from it (aka MT safe peek).
> + * Note that right now this new API is available only for two sync modes:
> + * 1) Single Producer/Single Consumer (RTE_RING_SYNC_ST)
> + * 2) Serialized Producer/Serialized Consumer (RTE_RING_SYNC_MT_HTS).
> + * It is a user responsibility to create/init soring with appropriate sync
> + * modes selected for enqueue/dequeue.
> + * For more information, please refer to corresping rte_ring peek API.

Typo: "corresping" should be "corresponding"

> + */
> +
> +/**
> + * @warning
> + * @b EXPERIMENTAL: this API may change without prior notice.
> + *
> + * Start to enqueue excact number of objects on the soring.

Typo: "excact" should be "exact"

> +/**
> + * @warning
> + * @b EXPERIMENTAL: this API may change without prior notice.
> + *
> + * Complete to enqueue several objects plus metadata on the soring.
> + * Note that number of objects to enqueue should not exceed previous
> + * enqueue_start return value.
> + *
> + * @param r
> + *   A pointer to the soring structure.
> + * @param objs
> + *   A pointer to an array of objects to enqueue.
> + *   Size of objects to enqueue must be the same value as 'elem_size' parameter
> + *   used while creating the soring. Otherwise the results are undefined.
> + * @param meta
> + *   A pointer to an array of metadata values for each object to enqueue.
> + *   Note that if user not using object metadata values, then this parameter
> + *   can be NULL.
> + *   Size of elements in this array must be the same value as 'meta_size'
> + *   parameter used while creating the soring. If user created the soring with
> + *   'meta_size' value equals zero, then 'meta' parameter should be NULL.
> + *   Otherwise the results are undefined.
> + * @param n
> + *   The number of objects to add in the soring from the 'objs'.
> + */
> +__rte_experimental
> +void
> +rte_soring_enqueux_finish(struct rte_soring *r, const void *objs,
> +	const void *meta, uint32_t n);

Typo in function name: "enqueux_finish" should be "enqueue_finish" to
match the pattern. This inconsistency appears in both the header and
implementation.

> diff --git a/lib/ring/soring.c b/lib/ring/soring.c
> index fc7fc55a21..b7d0ad6da8 100644
> --- a/lib/ring/soring.c
> +++ b/lib/ring/soring.c

[ ... ]

> @@ -296,6 +294,70 @@ soring_enqueue(struct rte_soring *r, const void *objs,
>  	return n;
>  }
>  
> +static inline uint32_t
> +soring_enqueue_start(struct rte_soring *r, uint32_t num,
> +	enum rte_ring_queue_behavior behavior, uint32_t *free_space)
> +{
> +	enum rte_ring_sync_type st;
> +	uint32_t free, head, n, next;
> +
> +	RTE_ASSERT(r != NULL && r->nb_stage > 0);
> +
> +	st = r->prod.ht.sync_type;
> +
> +	switch (st) {
> +	case RTE_RING_SYNC_ST:
> +		n = __rte_ring_headtail_move_head(&r->prod.ht, &r->cons.ht,
> +			r->capacity, RTE_RING_SYNC_ST, num, behavior,
> +			&head, &next, &free);
> +		break;
> +	case RTE_RING_SYNC_MT_HTS:
> +		n = __rte_ring_hts_move_head(&r->prod.hts, &r->cons.ht,
> +			r->capacity, num, behavior, &head, &free);
> +		break;
> +	default:
> +		/* unsupported mode, shouldn't be here */
> +		RTE_ASSERT(0);
> +		free = 0;
> +		n = 0;
> +	}

In soring_enqueue_start default case, does setting free=0 and n=0 after
RTE_ASSERT(0) make sense? If assertions are disabled, this could lead
to silent failures. Should this return an error or log a warning?

> +static inline void
> +soring_enqueue_finish(struct rte_soring *r, const void *objs, const void *meta,
> +	uint32_t num)
> +{
> +	enum rte_ring_sync_type st;
> +	uint32_t n, tail;
> +
> +	RTE_ASSERT(r != NULL && r->nb_stage > 0);
> +	RTE_ASSERT(meta == NULL || r->meta != NULL);
> +
> +	st = r->prod.ht.sync_type;
> +
> +	switch (st) {
> +	case RTE_RING_SYNC_ST:
> +		n = __rte_ring_st_get_tail(&r->prod.ht, &tail, num);
> +		if (n != 0)
> +			__enqueue_elems(r, objs, meta, tail, n);
> +		__rte_ring_st_set_head_tail(&r->prod.ht, tail, n, 1);
> +		break;
> +	case RTE_RING_SYNC_MT_HTS:
> +		n = __rte_ring_hts_get_tail(&r->prod.hts, &tail, num);
> +		if (n != 0)
> +			__enqueue_elems(r, objs, meta, tail, n);
> +		__rte_ring_hts_set_head_tail(&r->prod.hts, tail, n, 1);
> +		break;
> +	default:
> +		/* unsupported mode, shouldn't be here */
> +		RTE_ASSERT(0);
> +	}
> +}

In soring_enqueue_finish, does the code have undefined behavior in the
default case when assertions are disabled? The function continues without
initializing n or tail, potentially leading to inconsistent ring state.

> +static inline uint32_t
> +soring_dequeue_start(struct rte_soring *r, void *objs, void *meta,
> +	uint32_t num, enum rte_ring_queue_behavior behavior,
> +	uint32_t *available)
> +{
> +	enum rte_ring_sync_type st;
> +	uint32_t avail, head, next, n, ns;
> +
> +	RTE_ASSERT(r != NULL && r->nb_stage > 0);
> +	RTE_ASSERT(meta == NULL || r->meta != NULL);
> +
> +	ns = r->nb_stage - 1;
> +	st = r->cons.ht.sync_type;
> +
> +	switch (st) {
> +	case RTE_RING_SYNC_ST:
> +		n = __rte_ring_headtail_move_head(&r->cons.ht, &r->stage[ns].ht,
> +			0, RTE_RING_SYNC_ST, num, behavior, &head, &next,
> +			&avail);
> +		break;
> +	case RTE_RING_SYNC_MT_HTS:
> +		n = __rte_ring_hts_move_head(&r->cons.hts, &r->stage[ns].ht,
> +			0,