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 <[email protected]> --- 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
