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 | 11 +++++------
drivers/event/sw/sw_evdev_scheduler.c | 17 ++++++++++++++---
3 files changed, 24 insertions(+), 12 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..2c11d02911 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -154,13 +154,12 @@ struct sw_port {
/* A numeric ID for the port */
uint8_t id;
- /* An atomic counter for when the port has been unlinked, and the
- * scheduler has not yet acked this unlink - hence there may still be
- * events in the buffers going to the port. When the unlinks in
- * progress is read by the scheduler, no more events will be pushed to
- * the port - hence the scheduler core can just assign zero.
+ /* Count of unlinks the scheduler has not yet acked, so there may
+ * still be events in the buffers going to the port. Incremented by
+ * the unlinking thread, cleared by the scheduler with an exchange so
+ * an increment racing with the ack is not lost.
*/
- 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
