patch 'ring: establish safe partial order in RTS mode' has been queued to stable release 23.11.6

[Shani Peretz <shperetz@nvidia.com>]

Hi,

FYI, your patch has been queued to stable release 23.11.6

Note it hasn't been pushed to http://dpdk.org/browse/dpdk-stable yet.
It will be pushed if I get no objections before 12/30/25. So please
shout if anyone has objections.

Also note that after the patch there's a diff of the upstream commit vs the
patch applied to the branch. This will indicate if there was any rebasing
needed to apply to the stable branch. If there were code changes for rebasing
(ie: not only metadata diffs), please double check that the rebase was
correctly done.

Queued patches are on a temporary branch at:
https://github.com/shanipr/dpdk-stable

This queued commit can be viewed at:
https://github.com/shanipr/dpdk-stable/commit/0916a3572d6e6f5ce60a16390889aabbc48003d0

Thanks.

Shani

---
From 0916a3572d6e6f5ce60a16390889aabbc48003d0 Mon Sep 17 00:00:00 2001
From: Wathsala Vithanage <wathsala.vithanage@arm.com>
Date: Tue, 2 Dec 2025 20:39:28 +0000
Subject: [PATCH] ring: establish safe partial order in RTS mode

[ upstream commit 36b69b5f958e10eb5beb4292ade57199a722a045 ]

Enforce a safe partial order by making the CAS and the preceding head
load use release and acquire semantics. This creates a pairwise
happens-before relationship between threads of the same kind
(i.e. between consumers or between producers).

Combine the two load-acquire operations of ht->head.raw, which were
previously split across the two paths of a conditional branch, into
__rte_ring_rts_head_wait. This simplifies the branching logic and makes
the synchronization behavior easier to understand.

Add comments to explain synchronizes with edges in detail.

Signed-off-by: Wathsala Vithanage <wathsala.vithanage@arm.com>
Signed-off-by: Ola Liljedahl <ola.liljedahl@arm.com>
---
 lib/ring/rte_ring_rts_elem_pvt.h | 96 ++++++++++++++++++++++++--------
 1 file changed, 72 insertions(+), 24 deletions(-)

diff --git a/lib/ring/rte_ring_rts_elem_pvt.h b/lib/ring/rte_ring_rts_elem_pvt.h
index 122650346b..867175e15f 100644
--- a/lib/ring/rte_ring_rts_elem_pvt.h
+++ b/lib/ring/rte_ring_rts_elem_pvt.h
@@ -31,6 +31,17 @@ __rte_ring_rts_update_tail(struct rte_ring_rts_headtail *ht)
 	 * might preceded us, then don't update tail with new value.
 	 */
 
+	/*
+	 * A0 = {A0.a, A0.b}: Synchronizes with the CAS at R0.
+	 * The CAS at R0 in same typed thread establishes a happens-before
+	 * relationship with this load acquire. Ensures that this thread
+	 * observes the same or later values for h.raw/h.val.cnt
+	 * observed by the other thread when it updated ht->tail.raw.
+	 * If not, ht->tail.raw may get updated out of sync (e.g. getting
+	 * updated to the same value twice). A0.a makes sure this condition
+	 * holds when CAS succeeds and A0.b when it fails.
+	 */
+	/* A0.a */
 	ot.raw = rte_atomic_load_explicit(&ht->tail.raw, rte_memory_order_acquire);
 
 	do {
@@ -41,6 +52,11 @@ __rte_ring_rts_update_tail(struct rte_ring_rts_headtail *ht)
 		if (++nt.val.cnt == h.val.cnt)
 			nt.val.pos = h.val.pos;
 
+	/*
+	 * R0: Synchronizes with A2 of a different thread of the opposite type and A0.b
+	 * of a different thread of the same type.
+	 */
+	/* A0.b */
 	} while (rte_atomic_compare_exchange_strong_explicit(&ht->tail.raw,
 			(uint64_t *)(uintptr_t)&ot.raw, nt.raw,
 			rte_memory_order_release, rte_memory_order_acquire) == 0);
@@ -50,18 +66,22 @@ __rte_ring_rts_update_tail(struct rte_ring_rts_headtail *ht)
  * @internal This function waits till head/tail distance wouldn't
  * exceed pre-defined max value.
  */
-static __rte_always_inline void
+static __rte_always_inline union __rte_ring_rts_poscnt
 __rte_ring_rts_head_wait(const struct rte_ring_rts_headtail *ht,
-	union __rte_ring_rts_poscnt *h)
+			 int memorder)
 {
-	uint32_t max;
+	union __rte_ring_rts_poscnt h;
+	uint32_t max = ht->htd_max;
 
-	max = ht->htd_max;
 
-	while (h->val.pos - ht->tail.val.pos > max) {
+	h.raw = rte_atomic_load_explicit(&ht->head.raw, memorder);
+
+	while (h.val.pos - ht->tail.val.pos > max) {
 		rte_pause();
-		h->raw = rte_atomic_load_explicit(&ht->head.raw, rte_memory_order_acquire);
+		h.raw = rte_atomic_load_explicit(&ht->head.raw, memorder);
 	}
+
+	return h;
 }
 
 /**
@@ -72,13 +92,11 @@ __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
 	enum rte_ring_queue_behavior behavior, uint32_t *old_head,
 	uint32_t *free_entries)
 {
-	uint32_t n;
+	uint32_t n, cons_tail;
 	union __rte_ring_rts_poscnt nh, oh;
 
 	const uint32_t capacity = r->capacity;
 
-	oh.raw = rte_atomic_load_explicit(&r->rts_prod.head.raw, rte_memory_order_acquire);
-
 	do {
 		/* Reset n to the initial burst count */
 		n = num;
@@ -88,7 +106,20 @@ __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
 		 * make sure that we read prod head *before*
 		 * reading cons tail.
 		 */
-		__rte_ring_rts_head_wait(&r->rts_prod, &oh);
+		/*
+		 * A1 Synchronizes with the CAS at R1.
+		 * Establishes a happens-before relationship with a thread of the same
+		 * type that released the ht.raw, ensuring this thread observes all of
+		 * its memory effects needed to maintain a safe partial order.
+		 */
+		oh = __rte_ring_rts_head_wait(&r->rts_prod, rte_memory_order_acquire);
+
+		/*
+		 * A2: Establish a synchronizes-with edge using a store-release at R0.
+		 * This ensures that all memory effects from the preceding opposing
+		 * thread are observed.
+		 */
+		cons_tail = rte_atomic_load_explicit(&r->cons.tail, rte_memory_order_acquire);
 
 		/*
 		 *  The subtraction is done between two unsigned 32bits value
@@ -96,7 +127,7 @@ __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
 		 * *old_head > cons_tail). So 'free_entries' is always between 0
 		 * and capacity (which is < size).
 		 */
-		*free_entries = capacity + r->cons.tail - oh.val.pos;
+		*free_entries = capacity + cons_tail - oh.val.pos;
 
 		/* check that we have enough room in ring */
 		if (unlikely(n > *free_entries))
@@ -110,13 +141,16 @@ __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
 		nh.val.cnt = oh.val.cnt + 1;
 
 	/*
-	 * this CAS(ACQUIRE, ACQUIRE) serves as a hoist barrier to prevent:
-	 *  - OOO reads of cons tail value
-	 *  - OOO copy of elems to the ring
+	 * R1: Establishes a synchronizes-with edge with the load-acquire
+	 * of ht.raw at A1. Ensures that the store-release to the tail by
+	 * this thread, if it was of the opposite type, becomes
+	 * visible to another thread of the current type. That thread will
+	 * then observe the updates in the same order, keeping a safe
+	 * partial order.
 	 */
 	} while (rte_atomic_compare_exchange_strong_explicit(&r->rts_prod.head.raw,
 			(uint64_t *)(uintptr_t)&oh.raw, nh.raw,
-			rte_memory_order_acquire, rte_memory_order_acquire) == 0);
+			rte_memory_order_release, rte_memory_order_relaxed) == 0);
 
 	*old_head = oh.val.pos;
 	return n;
@@ -130,11 +164,9 @@ __rte_ring_rts_move_cons_head(struct rte_ring *r, uint32_t num,
 	enum rte_ring_queue_behavior behavior, uint32_t *old_head,
 	uint32_t *entries)
 {
-	uint32_t n;
+	uint32_t n, prod_tail;
 	union __rte_ring_rts_poscnt nh, oh;
 
-	oh.raw = rte_atomic_load_explicit(&r->rts_cons.head.raw, rte_memory_order_acquire);
-
 	/* move cons.head atomically */
 	do {
 		/* Restore n as it may change every loop */
@@ -145,14 +177,27 @@ __rte_ring_rts_move_cons_head(struct rte_ring *r, uint32_t num,
 		 * make sure that we read cons head *before*
 		 * reading prod tail.
 		 */
-		__rte_ring_rts_head_wait(&r->rts_cons, &oh);
+		/*
+		 * A3: Synchronizes with the CAS at R2.
+		 * Establishes a happens-before relationship with a thread of the same
+		 * type that released the ht.raw, ensuring this thread observes all of
+		 * its memory effects needed to maintain a safe partial order.
+		 */
+		oh = __rte_ring_rts_head_wait(&r->rts_cons, rte_memory_order_acquire);
+
+		/*
+		 * A4: Establish a synchronizes-with edge using a store-release at R0.
+		 * This ensures that all memory effects from the preceding opposing
+		 * thread are observed.
+		 */
+		prod_tail = rte_atomic_load_explicit(&r->prod.tail, rte_memory_order_acquire);
 
 		/* The subtraction is done between two unsigned 32bits value
 		 * (the result is always modulo 32 bits even if we have
 		 * cons_head > prod_tail). So 'entries' is always between 0
 		 * and size(ring)-1.
 		 */
-		*entries = r->prod.tail - oh.val.pos;
+		*entries = prod_tail - oh.val.pos;
 
 		/* Set the actual entries for dequeue */
 		if (n > *entries)
@@ -165,13 +210,16 @@ __rte_ring_rts_move_cons_head(struct rte_ring *r, uint32_t num,
 		nh.val.cnt = oh.val.cnt + 1;
 
 	/*
-	 * this CAS(ACQUIRE, ACQUIRE) serves as a hoist barrier to prevent:
-	 *  - OOO reads of prod tail value
-	 *  - OOO copy of elems from the ring
+	 * R2: Establishes a synchronizes-with edge with the load-acquire
+	 * of ht.raw at A3. Ensures that the store-release to the tail by
+	 * this thread, if it was of the opposite type, becomes
+	 * visible to another thread of the current type. That thread will
+	 * then observe the updates in the same order, keeping a safe
+	 * partial order.
 	 */
 	} while (rte_atomic_compare_exchange_strong_explicit(&r->rts_cons.head.raw,
 			(uint64_t *)(uintptr_t)&oh.raw, nh.raw,
-			rte_memory_order_acquire, rte_memory_order_acquire) == 0);
+			rte_memory_order_release, rte_memory_order_relaxed) == 0);
 
 	*old_head = oh.val.pos;
 	return n;
-- 
2.43.0

---
  Diff of the applied patch vs upstream commit (please double-check if non-empty:
---
--- -	2025-12-25 11:16:39.892313987 +0200
+++ 0071-ring-establish-safe-partial-order-in-RTS-mode.patch	2025-12-25 11:16:36.077854000 +0200
@@ -1 +1 @@
-From 36b69b5f958e10eb5beb4292ade57199a722a045 Mon Sep 17 00:00:00 2001
+From 0916a3572d6e6f5ce60a16390889aabbc48003d0 Mon Sep 17 00:00:00 2001
@@ -3 +3 @@
-Date: Tue, 11 Nov 2025 18:37:19 +0000
+Date: Tue, 2 Dec 2025 20:39:28 +0000
@@ -5,0 +6,2 @@
+[ upstream commit 36b69b5f958e10eb5beb4292ade57199a722a045 ]
+
@@ -18,3 +19,0 @@
-Fixes: e6ba4731c0f3a ("ring: introduce RTS ring mode")
-Cc: stable@dpdk.org
-
@@ -24,2 +23,2 @@
- lib/ring/rte_ring_rts_elem_pvt.h | 67 +++++++++++++++++++++++---------
- 1 file changed, 49 insertions(+), 18 deletions(-)
+ lib/ring/rte_ring_rts_elem_pvt.h | 96 ++++++++++++++++++++++++--------
+ 1 file changed, 72 insertions(+), 24 deletions(-)
@@ -28 +27 @@
-index 96825931f8..0280369b21 100644
+index 122650346b..867175e15f 100644
@@ -49,2 +48 @@
-@@ -40,7 +51,11 @@ __rte_ring_rts_update_tail(struct rte_ring_rts_headtail *ht)
- 		nt.raw = ot.raw;
+@@ -41,6 +52,11 @@ __rte_ring_rts_update_tail(struct rte_ring_rts_headtail *ht)
@@ -53 +51 @@
--
+ 
@@ -62 +60 @@
-@@ -50,18 +65,21 @@ __rte_ring_rts_update_tail(struct rte_ring_rts_headtail *ht)
+@@ -50,18 +66,22 @@ __rte_ring_rts_update_tail(struct rte_ring_rts_headtail *ht)
@@ -70 +68 @@
-+			 rte_memory_order memorder)
++			 int memorder)
@@ -77 +74,0 @@
-+	h.raw = rte_atomic_load_explicit(&ht->head.raw, memorder);
@@ -79,0 +77,2 @@
++	h.raw = rte_atomic_load_explicit(&ht->head.raw, memorder);
++
@@ -90 +89 @@
-@@ -94,12 +112,9 @@ __rte_ring_rts_move_head(struct rte_ring_rts_headtail *d,
+@@ -72,13 +92,11 @@ __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
@@ -92 +91 @@
- 	uint32_t *entries)
+ 	uint32_t *free_entries)
@@ -95 +94 @@
-+	uint32_t n, stail;
++	uint32_t n, cons_tail;
@@ -98,2 +97,3 @@
--	oh.raw = rte_atomic_load_explicit(&d->head.raw,
--			rte_memory_order_acquire);
+ 	const uint32_t capacity = r->capacity;
+ 
+-	oh.raw = rte_atomic_load_explicit(&r->rts_prod.head.raw, rte_memory_order_acquire);
@@ -104 +104 @@
-@@ -109,7 +124,20 @@ __rte_ring_rts_move_head(struct rte_ring_rts_headtail *d,
+@@ -88,7 +106,20 @@ __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
@@ -108 +108 @@
--		__rte_ring_rts_head_wait(d, &oh);
+-		__rte_ring_rts_head_wait(&r->rts_prod, &oh);
@@ -115 +115 @@
-+		oh = __rte_ring_rts_head_wait(d, rte_memory_order_acquire);
++		oh = __rte_ring_rts_head_wait(&r->rts_prod, rte_memory_order_acquire);
@@ -122 +122 @@
-+		stail = rte_atomic_load_explicit(&s->tail, rte_memory_order_acquire);
++		cons_tail = rte_atomic_load_explicit(&r->cons.tail, rte_memory_order_acquire);
@@ -126,2 +126,2 @@
-@@ -117,7 +145,7 @@ __rte_ring_rts_move_head(struct rte_ring_rts_headtail *d,
- 		 * *old_head > cons_tail). So 'entries' is always between 0
+@@ -96,7 +127,7 @@ __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
+ 		 * *old_head > cons_tail). So 'free_entries' is always between 0
@@ -130,2 +130,2 @@
--		*entries = capacity + s->tail - oh.val.pos;
-+		*entries = capacity + stail - oh.val.pos;
+-		*free_entries = capacity + r->cons.tail - oh.val.pos;
++		*free_entries = capacity + cons_tail - oh.val.pos;
@@ -134,2 +134,2 @@
- 		if (unlikely(n > *entries))
-@@ -131,14 +159,17 @@ __rte_ring_rts_move_head(struct rte_ring_rts_headtail *d,
+ 		if (unlikely(n > *free_entries))
+@@ -110,13 +141,16 @@ __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
@@ -149 +149,65 @@
- 	} while (rte_atomic_compare_exchange_strong_explicit(&d->head.raw,
+ 	} while (rte_atomic_compare_exchange_strong_explicit(&r->rts_prod.head.raw,
+ 			(uint64_t *)(uintptr_t)&oh.raw, nh.raw,
+-			rte_memory_order_acquire, rte_memory_order_acquire) == 0);
++			rte_memory_order_release, rte_memory_order_relaxed) == 0);
+ 
+ 	*old_head = oh.val.pos;
+ 	return n;
+@@ -130,11 +164,9 @@ __rte_ring_rts_move_cons_head(struct rte_ring *r, uint32_t num,
+ 	enum rte_ring_queue_behavior behavior, uint32_t *old_head,
+ 	uint32_t *entries)
+ {
+-	uint32_t n;
++	uint32_t n, prod_tail;
+ 	union __rte_ring_rts_poscnt nh, oh;
+ 
+-	oh.raw = rte_atomic_load_explicit(&r->rts_cons.head.raw, rte_memory_order_acquire);
+-
+ 	/* move cons.head atomically */
+ 	do {
+ 		/* Restore n as it may change every loop */
+@@ -145,14 +177,27 @@ __rte_ring_rts_move_cons_head(struct rte_ring *r, uint32_t num,
+ 		 * make sure that we read cons head *before*
+ 		 * reading prod tail.
+ 		 */
+-		__rte_ring_rts_head_wait(&r->rts_cons, &oh);
++		/*
++		 * A3: Synchronizes with the CAS at R2.
++		 * Establishes a happens-before relationship with a thread of the same
++		 * type that released the ht.raw, ensuring this thread observes all of
++		 * its memory effects needed to maintain a safe partial order.
++		 */
++		oh = __rte_ring_rts_head_wait(&r->rts_cons, rte_memory_order_acquire);
++
++		/*
++		 * A4: Establish a synchronizes-with edge using a store-release at R0.
++		 * This ensures that all memory effects from the preceding opposing
++		 * thread are observed.
++		 */
++		prod_tail = rte_atomic_load_explicit(&r->prod.tail, rte_memory_order_acquire);
+ 
+ 		/* The subtraction is done between two unsigned 32bits value
+ 		 * (the result is always modulo 32 bits even if we have
+ 		 * cons_head > prod_tail). So 'entries' is always between 0
+ 		 * and size(ring)-1.
+ 		 */
+-		*entries = r->prod.tail - oh.val.pos;
++		*entries = prod_tail - oh.val.pos;
+ 
+ 		/* Set the actual entries for dequeue */
+ 		if (n > *entries)
+@@ -165,13 +210,16 @@ __rte_ring_rts_move_cons_head(struct rte_ring *r, uint32_t num,
+ 		nh.val.cnt = oh.val.cnt + 1;
+ 
+ 	/*
+-	 * this CAS(ACQUIRE, ACQUIRE) serves as a hoist barrier to prevent:
+-	 *  - OOO reads of prod tail value
+-	 *  - OOO copy of elems from the ring
++	 * R2: Establishes a synchronizes-with edge with the load-acquire
++	 * of ht.raw at A3. Ensures that the store-release to the tail by
++	 * this thread, if it was of the opposite type, becomes
++	 * visible to another thread of the current type. That thread will
++	 * then observe the updates in the same order, keeping a safe
++	 * partial order.
+ 	 */
+ 	} while (rte_atomic_compare_exchange_strong_explicit(&r->rts_cons.head.raw,
@@ -151,4 +215,2 @@
--			rte_memory_order_acquire,
--			rte_memory_order_acquire) == 0);
-+			rte_memory_order_release,
-+			rte_memory_order_relaxed) == 0);
+-			rte_memory_order_acquire, rte_memory_order_acquire) == 0);
++			rte_memory_order_release, rte_memory_order_relaxed) == 0);