[RFC 15/32] event/sw: fix unlinks in progress counter races

[Stephen Hemminger <stephen@networkplumber.org>]

The counter is written by both the application thread (increment on
unlink) and the scheduler (clear on ack) as a plain uint8_t. An
increment is lost if it lands between the scheduler's test and clear,
making rte_event_port_unlinks_in_progress() report completion before
the scheduler has seen the unlink. Nothing orders the scheduler's
later cq map reads against the counter test on a weakly ordered CPU
either.

Make the counter atomic: release fetch-add on unlink, acquire
exchange to clear. The exchange cannot lose a concurrent increment,
and the acquire guarantees the scheduler only acks unlinks whose cq
map update it can observe, replacing the full barrier in unlink.

Fixes: bd5ac24fea88 ("event/sw: implement unlinks in progress function")
Cc: stable@dpdk.org

Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/event/sw/sw_evdev.c           |  8 +++++---
 drivers/event/sw/sw_evdev.h           |  2 +-
 drivers/event/sw/sw_evdev_scheduler.c | 13 ++++++++++---
 3 files changed, 16 insertions(+), 7 deletions(-)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 3ad82e94ac..bb6f50e03b 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -119,8 +119,9 @@ sw_port_unlink(struct rte_eventdev *dev, void *port, uint8_t queues[],
 		}
 	}
 
-	p->unlinks_in_progress += unlinked;
-	rte_smp_mb();
+	/* Pairs with the acquire exchange in the scheduler */
+	rte_atomic_fetch_add_explicit(&p->unlinks_in_progress, unlinked,
+				      rte_memory_order_release);
 
 	return unlinked;
 }
@@ -130,7 +131,8 @@ sw_port_unlinks_in_progress(struct rte_eventdev *dev, void *port)
 {
 	RTE_SET_USED(dev);
 	struct sw_port *p = port;
-	return p->unlinks_in_progress;
+	return rte_atomic_load_explicit(&p->unlinks_in_progress,
+					rte_memory_order_relaxed);
 }
 
 static int
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
index c159be21be..8b9118bf91 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -160,7 +160,7 @@ struct sw_port {
 	 * progress is read by the scheduler, no more events will be pushed to
 	 * the port - hence the scheduler core can just assign zero.
 	 */
-	uint8_t unlinks_in_progress;
+	RTE_ATOMIC(uint8_t) unlinks_in_progress;
 
 	int16_t is_directed; /** Takes from a single directed QID */
 	/**
diff --git a/drivers/event/sw/sw_evdev_scheduler.c b/drivers/event/sw/sw_evdev_scheduler.c
index a5fdcf301b..f4bce2cbb8 100644
--- a/drivers/event/sw/sw_evdev_scheduler.c
+++ b/drivers/event/sw/sw_evdev_scheduler.c
@@ -523,9 +523,16 @@ sw_event_schedule(struct rte_eventdev *dev)
 		do {
 			in_pkts = 0;
 			for (i = 0; i < sw->port_count; i++) {
-				/* ack the unlinks in progress as done */
-				if (sw->ports[i].unlinks_in_progress)
-					sw->ports[i].unlinks_in_progress = 0;
+				/* Ack the unlinks in progress as done. The
+				 * acquire exchange orders the cq map reads
+				 * below after the unlinker's map update.
+				 */
+				if (rte_atomic_load_explicit(
+						&sw->ports[i].unlinks_in_progress,
+						rte_memory_order_relaxed))
+					rte_atomic_exchange_explicit(
+						&sw->ports[i].unlinks_in_progress,
+						0, rte_memory_order_acquire);
 
 				if (sw->ports[i].is_directed)
 					in_pkts += sw_schedule_pull_port_dir(sw, i);
-- 
2.53.0