[PATCH v3 18/29] event/sw: replace SMP barriers with C11 atomics

[Stephen Hemminger]

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 <stephen at networkplumber.org>
---
 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