[dpdk-dev] [PATCH v2] ring: enforce reading the tails before ring operations

[Ilya Maximets]

On 07.03.2019 9:45, gavin hu wrote:
> In weak memory models, like arm64, reading the {prod,cons}.tail may get
> reordered after reading or writing the ring slots, which corrupts the ring
> and stale data is observed.
> 
> This issue was reported by NXP on 8-A72 DPAA2 board. The problem is most
> likely caused by missing the acquire semantics when reading cons.tail (in
> SP enqueue) or prod.tail (in SC dequeue) which makes it possible to read a
> stale value from the ring slots.
> 
> For MP (and MC) case, rte_atomic32_cmpset() already provides the required
> ordering. This patch is to prevent reading and writing the ring slots get
> reordered before reading {prod,cons}.tail for SP (and SC) case.

Read barrier rte_smp_rmb() is OK to prevent reading the ring get reordered
before reading the tail. However, to prevent *writing* the ring get reordered
*before reading* the tail you need a full memory barrier, i.e. rte_smp_mb().

> 
> Signed-off-by: gavin hu <gavin.hu at arm.com>
> Reviewed-by: Ola Liljedahl <Ola.Liljedahl at arm.com>
> Tested-by: Nipun Gupta <nipun.gupta at nxp.com>
> ---
>  lib/librte_ring/rte_ring_generic.h | 16 ++++++++++------
>  1 file changed, 10 insertions(+), 6 deletions(-)
> 
> diff --git a/lib/librte_ring/rte_ring_generic.h b/lib/librte_ring/rte_ring_generic.h
> index ea7dbe5..1bd3dfd 100644
> --- a/lib/librte_ring/rte_ring_generic.h
> +++ b/lib/librte_ring/rte_ring_generic.h
> @@ -90,9 +90,11 @@ __rte_ring_move_prod_head(struct rte_ring *r, unsigned int is_sp,
>  			return 0;
>  
>  		*new_head = *old_head + n;
> -		if (is_sp)
> -			r->prod.head = *new_head, success = 1;
> -		else
> +		if (is_sp) {
> +			r->prod.head = *new_head;
> +			rte_smp_rmb();
> +			success = 1;
> +		} else
>  			success = rte_atomic32_cmpset(&r->prod.head,
>  					*old_head, *new_head);
>  	} while (unlikely(success == 0));
> @@ -158,9 +160,11 @@ __rte_ring_move_cons_head(struct rte_ring *r, unsigned int is_sc,
>  			return 0;
>  
>  		*new_head = *old_head + n;
> -		if (is_sc)
> -			r->cons.head = *new_head, success = 1;
> -		else
> +		if (is_sc) {
> +			r->cons.head = *new_head;
> +			rte_smp_rmb();
> +			success = 1;
> +		} else
>  			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
>  					*new_head);
>  	} while (unlikely(success == 0));
>