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, acquire load to
read, and an acquire-release exchange to clear. The exchange cannot
lose an increment that lands after the test, so the scheduler can
keep its relaxed load of the counter first and take the locked
read-modify-write only when there is an unlink to acknowledge.
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 a2f760a98d..87a8996f9d 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 5e49b08030..22da5d1622 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
