[PATCH v9 16/25] bus/vmbus: convert from rte_atomic to stdatomic

[Stephen Hemminger <stephen@networkplumber.org>]

The tx ring producer uses two cursors:
- tbr->windex which is driver private. It lets producers reserve space
  concurrently without a lock
- vbr->windex is the host visible commit cursor and has to advance in
  reservation order so the host never sees it point past unwritten data.

Convert the reservation CAS on tbr->windex to a weak compare exchange,
and replace the smp_wmb plus CAS spin that published the commit with
rte_wait_until_equal_32 on the previous producer followed by a release
store. The release ordering on that store covers what the barrier did.

The wait on the previous producer uses acquire ordering, not relaxed.
A release store only carries the writes of the storing thread and
anything it has a happens-before edge to. Producers publish in
reservation order, so when this thread stores its own windex the host
may then consume the data of every earlier producer as well. The
acquire on the wait pairs with the earlier producer's release store,
putting that producer's data writes into this thread's happens-before
set so they are carried by this store too.

vbr->windex stays volatile uint32_t in the packed bufring struct.
The host is not a C11 thread.

The atomic qualifier is applied by cast at the single publishing store;
the (uintptr_t) launder there avoids a bogus misaligned atomic warning
from the packed attribute, since windex sits at offset 0 of a page
aligned struct.

Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/bus/vmbus/private.h       |  2 +-
 drivers/bus/vmbus/vmbus_bufring.c | 40 +++++++++++++++++--------------
 2 files changed, 23 insertions(+), 19 deletions(-)

diff --git a/drivers/bus/vmbus/private.h b/drivers/bus/vmbus/private.h
index 6efac86b77..6b7782724f 100644
--- a/drivers/bus/vmbus/private.h
+++ b/drivers/bus/vmbus/private.h
@@ -25,7 +25,7 @@ extern int vmbus_logtype_bus;
 struct vmbus_br {
 	struct vmbus_bufring *vbr;
 	uint32_t	dsize;
-	uint32_t	windex; /* next available location */
+	RTE_ATOMIC(uint32_t) windex; /* next available location */
 };
 
 #define UIO_NAME_MAX 64
diff --git a/drivers/bus/vmbus/vmbus_bufring.c b/drivers/bus/vmbus/vmbus_bufring.c
index fcb97287dc..dcd66390a3 100644
--- a/drivers/bus/vmbus/vmbus_bufring.c
+++ b/drivers/bus/vmbus/vmbus_bufring.c
@@ -15,7 +15,7 @@
 #include <rte_tailq.h>
 #include <rte_log.h>
 #include <rte_malloc.h>
-#include <rte_atomic.h>
+#include <rte_stdatomic.h>
 #include <rte_memory.h>
 #include <rte_pause.h>
 #include <rte_bus_vmbus.h>
@@ -121,17 +121,13 @@ vmbus_txbr_write(struct vmbus_br *tbr, const struct iovec iov[], int iovlen,
 		total += iov[i].iov_len;
 	total += sizeof(save_windex);
 
+	/* Get current free location */
+	old_windex = rte_atomic_load_explicit(&tbr->windex,
+					      rte_memory_order_relaxed);
+
 	/* Reserve space in ring */
 	do {
-		uint32_t avail;
-
-		/* Get current free location */
-		old_windex = tbr->windex;
-
-		/* Prevent compiler reordering this with calculation */
-		rte_compiler_barrier();
-
-		avail = vmbus_br_availwrite(tbr, old_windex);
+		uint32_t avail = vmbus_br_availwrite(tbr, old_windex);
 
 		/* If not enough space in ring, then tell caller. */
 		if (avail <= total)
@@ -139,8 +135,12 @@ vmbus_txbr_write(struct vmbus_br *tbr, const struct iovec iov[], int iovlen,
 
 		next_windex = vmbus_br_idxinc(old_windex, total, ring_size);
 
-		/* Atomic update of next write_index for other threads */
-	} while (!rte_atomic32_cmpset(&tbr->windex, old_windex, next_windex));
+		/* Atomic update of next write_index for other threads
+		 * Can use weak since easy to recompute and retry.
+		 */
+	} while (!rte_atomic_compare_exchange_weak_explicit(
+			 &tbr->windex, &old_windex, next_windex,
+			 rte_memory_order_acquire, rte_memory_order_relaxed));
 
 	/* Space from old..new is now reserved */
 	windex = old_windex;
@@ -157,12 +157,15 @@ vmbus_txbr_write(struct vmbus_br *tbr, const struct iovec iov[], int iovlen,
 	/* The region reserved should match region used */
 	RTE_ASSERT(windex == next_windex);
 
-	/* Ensure that data is available before updating host index */
-	rte_smp_wmb();
+	/* Wait for previous producer to publish their windex update */
+	rte_wait_until_equal_32(&vbr->windex, old_windex, rte_memory_order_acquire);
 
-	/* Checkin for our reservation. wait for our turn to update host */
-	while (!rte_atomic32_cmpset(&vbr->windex, old_windex, next_windex))
-		rte_pause();
+	/* Publish our windex update; prior data writes ordered via release.
+	 * windex is 4-byte aligned in practice (struct is page-aligned, windex
+	 * at offset 0); cast launders the packed-struct alignment-1 attribute.
+	 */
+	rte_atomic_store_explicit((volatile __rte_atomic uint32_t *)(uintptr_t)&vbr->windex,
+				  next_windex, rte_memory_order_release);
 
 	/* If host had read all data before this, then need to signal */
 	*need_sig |= vmbus_txbr_need_signal(vbr, old_windex);
@@ -224,7 +227,8 @@ vmbus_rxbr_read(struct vmbus_br *rbr, void *data, size_t dlen, size_t skip)
 		return -EAGAIN;
 
 	/* Record where host was when we started read (for debug) */
-	rbr->windex = rbr->vbr->windex;
+	rte_atomic_store_explicit(&rbr->windex, rbr->vbr->windex,
+				  rte_memory_order_relaxed);
 
 	/*
 	 * Copy channel packet from RX bufring.
-- 
2.53.0