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: [email protected]
Signed-off-by: Stephen Hemminger <[email protected]>
---
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
