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

[Wei Hu]

> -----Original Message-----
> From: Stephen Hemminger <stephen at networkplumber.org>
> Sent: Friday, September 18, 2026 4:10 AM
> To: dev at dpdk.org
> Cc: Stephen Hemminger <stephen at networkplumber.org>; Wei Hu
> <weh at microsoft.com>
> Subject: [EXTERNAL] [PATCH v8 16/25] bus/vmbus: convert from rte_atomic
> to stdatomic
> 
> 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 at networkplumber.org>

Reviewed-by: Wei Hu <weh at microsoft.com>

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