The write barrier plus plain store publish patterns (cq map count, port initialized, device started) become release stores; the barrier after clearing started in sw_stop() becomes a release fence. All are control path.
The scheduler load of started is acquire, to complete the handshake with the release store in sw_start(). It runs once per service call, not per event. Scheduler reads of cq_num_mapped_cqs stay plain loads: acquire loads would land in the inner scheduling loops on weakly ordered CPUs. The transient window where a new count is visible before the map entry on such CPUs predates this change and is unchanged by it; unlink is covered by the counter handshake. Signed-off-by: Stephen Hemminger <[email protected]> --- drivers/event/sw/sw_evdev.c | 29 ++++++++++++++++++--------- drivers/event/sw/sw_evdev_scheduler.c | 7 ++++++- 2 files changed, 26 insertions(+), 10 deletions(-) diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c index 87a8996f9d..165d8e3950 100644 --- a/drivers/event/sw/sw_evdev.c +++ b/drivers/event/sw/sw_evdev.c @@ -84,8 +84,10 @@ sw_port_link(struct rte_eventdev *dev, void *port, const uint8_t queues[], } q->cq_map[q->cq_num_mapped_cqs] = p->id; - rte_smp_wmb(); - q->cq_num_mapped_cqs++; + /* Release publishes the map entry before the new count */ + rte_atomic_store_explicit( + (uint32_t __rte_atomic *)&q->cq_num_mapped_cqs, + q->cq_num_mapped_cqs + 1, rte_memory_order_release); } return i; } @@ -105,8 +107,14 @@ sw_port_unlink(struct rte_eventdev *dev, void *port, uint8_t queues[], if (q->cq_map[j] == p->id) { q->cq_map[j] = q->cq_map[q->cq_num_mapped_cqs - 1]; - rte_smp_wmb(); - q->cq_num_mapped_cqs--; + /* Release publishes the map update + * before the new count + */ + rte_atomic_store_explicit( + (uint32_t __rte_atomic *) + &q->cq_num_mapped_cqs, + q->cq_num_mapped_cqs - 1, + rte_memory_order_release); unlinked++; p->num_qids_mapped--; @@ -210,8 +218,9 @@ sw_port_setup(struct rte_eventdev *dev, uint8_t port_id, } dev->data->ports[port_id] = p; - rte_smp_wmb(); - p->initialized = 1; + /* Release publishes the port setup before initialized flag */ + rte_atomic_store_explicit((uint8_t __rte_atomic *)&p->initialized, 1, + rte_memory_order_release); return 0; } @@ -817,8 +826,9 @@ sw_start(struct rte_eventdev *dev) if (sw_xstats_init(sw) < 0) return -EINVAL; - rte_smp_wmb(); - sw->started = 1; + /* Release publishes device state before the started flag */ + rte_atomic_store_explicit((uint8_t __rte_atomic *)&sw->started, 1, + rte_memory_order_release); return 0; } @@ -847,7 +857,8 @@ sw_stop(struct rte_eventdev *dev) sw_clean_qid_iqs(dev); sw_xstats_uninit(sw); sw->started = 0; - rte_smp_wmb(); + /* Order the started store before re-enabling the service */ + rte_atomic_thread_fence(rte_memory_order_release); if (runstate == 1) rte_service_runstate_set(sw->service_id, 1); diff --git a/drivers/event/sw/sw_evdev_scheduler.c b/drivers/event/sw/sw_evdev_scheduler.c index 207dee5854..5b15c2a9ad 100644 --- a/drivers/event/sw/sw_evdev_scheduler.c +++ b/drivers/event/sw/sw_evdev_scheduler.c @@ -513,7 +513,12 @@ sw_event_schedule(struct rte_eventdev *dev) uint32_t i; sw->sched_called++; - if (unlikely(!sw->started)) + /* Acquire pairs with the release store in sw_start(), so the device + * state it published is visible before any of it is used here. Once + * per service call, not per event. + */ + if (unlikely(!rte_atomic_load_explicit((uint8_t __rte_atomic *)&sw->started, + rte_memory_order_acquire))) return -EAGAIN; do { -- 2.53.0
