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. On a weakly ordered CPU nothing
orders the scheduler's cq map reads against the counter test, nor the
application's reads against the counter reaching zero.
Make the counter atomic: release fetch-add on unlink, unconditional
acquire-release exchange to clear, acquire load to read.
Fixes: bd5ac24fea88 ("event/sw: implement unlinks in progress function")
Cc: [email protected]
Signed-off-by: Stephen Hemminger <[email protected]>
Acked-by: Bruce Richardson <[email protected]>
---
drivers/event/sw/sw_evdev.c | 8 +++++---
drivers/event/sw/sw_evdev.h | 2 +-
drivers/event/sw/sw_evdev_scheduler.c | 17 ++++++++++++++---
3 files changed, 20 insertions(+), 7 deletions(-)
diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 3ad82e94ac..fedf638c60 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_acquire);
}
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..207dee5854 100644
--- a/drivers/event/sw/sw_evdev_scheduler.c
+++ b/drivers/event/sw/sw_evdev_scheduler.c
@@ -523,9 +523,20 @@ 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;
+ struct sw_port *p = &sw->ports[i];
+
+ /* Ack the unlinks in progress as done.
+ * The exchange cannot lose an increment that
+ * lands after the test, and its acquire orders
+ * the cq map reads below after the unlinker's
+ * map update. Nothing to synchronize with when
+ * the counter reads zero, so test first and
+ * keep the locked op off the common path.
+ */
+ if
(rte_atomic_load_explicit(&p->unlinks_in_progress,
+ rte_memory_order_relaxed) != 0)
+
rte_atomic_exchange_explicit(&p->unlinks_in_progress,
+ 0, rte_memory_order_acq_rel);
if (sw->ports[i].is_directed)
in_pkts +=
sw_schedule_pull_port_dir(sw, i);
--
2.53.0
