[PATCH v1 4/8] event/slb: add dynamic flow migration for atomic queues

[Pravin Pathak <pravinpathakdev@gmail.com>]

From: Pravin Pathak <pravin.pathak.dev@gmail.com>

Add dynamic flow migration to automatically rebalance flows across
worker ports when an atomic queue's destination port becomes congested.

When consecutive enqueue attempts to a worker's private atomic ring fail
and exceed SLB_MIGRATION_LOAD_THRESHOLD, migration is initiated. The
PMD inspects the overloaded ring and selects the lightest active flow
not currently in migration cooldown.

To preserve strict per-flow event ordering, migration completes only
after all previously scheduled events for the source port have been
returned (q_enqueued == q_returned). The flow affinity is then remapped
to the least-loaded worker port, and any matching events retained in the
hold buffer are reassigned.

Signed-off-by: Pravin Pathak <pravin.pathak.dev@gmail.com>
---
 drivers/event/slb/slb_evdev.c |  16 ++-
 drivers/event/slb/slb_evdev.h |  20 ++-
 drivers/event/slb/slb_event.c | 252 ++++++++++++++++++++++++++++++++--
 3 files changed, 273 insertions(+), 15 deletions(-)

diff --git a/drivers/event/slb/slb_evdev.c b/drivers/event/slb/slb_evdev.c
index be0b3630f6..1916b7faa4 100644
--- a/drivers/event/slb/slb_evdev.c
+++ b/drivers/event/slb/slb_evdev.c
@@ -74,8 +74,13 @@ slb_port_setup(struct rte_eventdev *dev, uint8_t port_id,
 		.total_dequeues = 0,
 		.empty_dequeues = 0,
 		.pending_releases = 0,
+		.is_loaded = false,
 		.single_linked_q = NULL,
 		.single_linked_ring = NULL,
+		.flow_migrations = 0,
+		.migrations_in = 0,
+		.migrations_out = 0,
+		.last_migrated_time = 0,
 	};
 
 	port->remap_input_q = false;
@@ -208,8 +213,12 @@ slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
 		queue->atomic_state->next_port_rr = 0;
 		rte_spinlock_init(&queue->atomic_state->refill_lock);
 		queue->atomic_state->hold_count = 0;
+		queue->atomic_state->migration.in_progress = false;
+		queue->atomic_state->migrations_count = 0;
 		memset(queue->atomic_state->port_stats, 0,
 		       sizeof(queue->atomic_state->port_stats));
+		memset(queue->atomic_state->flow_last_migrated, 0,
+		       sizeof(queue->atomic_state->flow_last_migrated));
 	} else {
 		queue->atomic_state = NULL;
 	}
@@ -547,12 +556,16 @@ slb_stop(struct rte_eventdev *dev)
 		struct slb_queue *q = &slb->queues[qid];
 
 		if (q->schedule_type == RTE_SCHED_TYPE_ATOMIC && q->atomic_state != NULL) {
-			for (int f = 0; f < SLB_ATOMIC_FLOWS; f++)
+			for (int f = 0; f < SLB_ATOMIC_FLOWS; f++) {
 				rte_atomic_store_explicit(
 					&q->atomic_state->flow_map.flows[f].port_id,
 					-1, rte_memory_order_relaxed);
+				q->atomic_state->flow_last_migrated[f] = 0;
+			}
+			q->atomic_state->migration.in_progress = false;
 
 			for (uint16_t pid = 0; pid < slb->num_ports; pid++) {
+				q->atomic_state->port_stats[pid].load_counter = 0;
 				if (q->per_port_rings[pid] != NULL)
 					slb_drain_ring(dev_id,
 						       q->per_port_rings[pid],
@@ -568,6 +581,7 @@ slb_stop(struct rte_eventdev *dev)
 		struct slb_port *port = &slb->ports[i];
 		port->inflight_credits = 0;
 		port->credits_drained = false;
+		port->is_loaded = false;
 		for (int p = 0; p < SLB_MAX_PRIORITIES; p++)
 			port->next_q[p] = 0;
 	}
diff --git a/drivers/event/slb/slb_evdev.h b/drivers/event/slb/slb_evdev.h
index 7a70d5c94c..b1a6de8606 100644
--- a/drivers/event/slb/slb_evdev.h
+++ b/drivers/event/slb/slb_evdev.h
@@ -87,6 +87,8 @@ struct slb_reorder_entry {
 #define SLB_ATOMIC_FLOWS 2048
 #define SLB_ATOMIC_FLOWS_MASK (SLB_ATOMIC_FLOWS - 1)
 #define SLB_HOLD_BUFFER_SIZE (SLB_MAX_PORT_DEQUEUE_DEPTH * 64)
+#define SLB_MIGRATION_LOAD_THRESHOLD 16
+#define SLB_MIGRATION_COOLDOWN_MS 500
 
 /* Per-flow entry for atomic scheduling affinity */
 struct slb_flow {
@@ -100,17 +102,28 @@ struct slb_flow_map {
 
 /* Per-port counters for an atomic queue */
 struct __rte_cache_aligned slb_atomic_port_stats {
+	uint32_t load_counter; /* Incremented when full, decremented when space available */
 	uint32_t flow_count;   /* Number of active flows mapped to this port */
 	uint64_t q_enqueued;   /* Events scheduled into this port's atomic port-Q */
 	uint64_t q_returned;   /* Events returned (FORWARD/RELEASE) by this port */
 };
 
+/* Active migration state for an atomic queue */
+struct slb_migration_state {
+	bool in_progress;
+	uint8_t port_id;       /* Source port being drained */
+	uint32_t flow_idx;     /* Flow hash index being migrated */
+};
+
 /* Dynamically allocated state for atomic queues */
 struct slb_atomic_state {
 	struct slb_flow_map flow_map;
 	uint32_t next_port_rr;
 	rte_spinlock_t refill_lock;
 	uint16_t hold_count;
+	struct slb_migration_state migration;
+	uint64_t migrations_count;
+	uint64_t flow_last_migrated[SLB_ATOMIC_FLOWS];
 	alignas(RTE_CACHE_LINE_SIZE)
 	struct slb_atomic_port_stats port_stats[SLB_MAX_PORTS];
 	uint8_t hold_ports[SLB_HOLD_BUFFER_SIZE];
@@ -132,7 +145,8 @@ struct __rte_cache_aligned slb_port {
 	bool implicit_release;
 	bool credits_drained;
 	bool remap_input_q;
-	uint8_t pad_port[4];
+	bool is_loaded; /* Port marked loaded for atomic flow migration */
+	uint8_t pad_port[3];
 	struct rte_event_ring *single_linked_ring;
 	struct slb_evdev *slb;
 
@@ -149,6 +163,10 @@ struct __rte_cache_aligned slb_port {
 	uint16_t dequeue_depth;
 	uint16_t enqueue_depth;
 	struct slb_queue *single_linked_q;
+	uint64_t flow_migrations;
+	uint64_t migrations_in;
+	uint64_t migrations_out;
+	uint64_t last_migrated_time;
 	uint64_t enqueue_calls;
 	uint64_t new_enqueued;
 	uint64_t forward_enqueued;
diff --git a/drivers/event/slb/slb_event.c b/drivers/event/slb/slb_event.c
index 22f40d72d4..3a0f7f6fde 100644
--- a/drivers/event/slb/slb_event.c
+++ b/drivers/event/slb/slb_event.c
@@ -3,6 +3,7 @@
  */
 
 #include <stdbool.h>
+#include <limits.h>
 #include <rte_bitops.h>
 #include <rte_cycles.h>
 #include <rte_ring_elem.h>
@@ -163,12 +164,204 @@ slb_flow_hash(uint32_t flow_id)
 	return (flow_id * 2654435761U) & SLB_ATOMIC_FLOWS_MASK;
 }
 
+static int32_t
+slb_find_lightest_flow_on_port(struct slb_queue *queue, uint8_t port_id,
+			       uint64_t now, uint64_t cooldown_cycles)
+{
+	struct rte_event_ring *ering = queue->per_port_rings[port_id];
+	struct slb_atomic_state *astate = queue->atomic_state;
+	struct rte_ring *r;
+	uint32_t cons;
+	uint32_t prod;
+	uint32_t count;
+	uint16_t counts[SLB_ATOMIC_FLOWS];
+	int32_t best_flow_idx = -1;
+	uint32_t min_events = UINT32_MAX;
+
+	if (unlikely(ering == NULL || astate == NULL))
+		return -1;
+
+	if (astate->port_stats[port_id].flow_count <= 1)
+		return -1;
+
+	r = &ering->r;
+	cons = rte_atomic_load_explicit(&r->cons.tail, rte_memory_order_relaxed);
+	prod = rte_atomic_load_explicit(&r->prod.tail, rte_memory_order_relaxed);
+	count = prod - cons;
+
+	memset(counts, 0, sizeof(counts));
+	if (count > 0) {
+		struct rte_event *ring_events = (struct rte_event *)(r + 1);
+
+		for (uint32_t i = 0; i < count; i++) {
+			uint32_t slot = (cons + i) & r->mask;
+			uint32_t f_idx = slb_flow_hash(ring_events[slot].flow_id);
+
+			counts[f_idx]++;
+		}
+	}
+
+	for (uint32_t f = 0; f < SLB_ATOMIC_FLOWS; f++) {
+		if (rte_atomic_load_explicit(&astate->flow_map.flows[f].port_id,
+					     rte_memory_order_relaxed) == (int32_t)port_id) {
+			if (astate->flow_last_migrated[f] != 0 &&
+			    (now - astate->flow_last_migrated[f]) < cooldown_cycles)
+				continue;
+
+			if (counts[f] < min_events) {
+				min_events = counts[f];
+				best_flow_idx = (int32_t)f;
+			}
+		}
+	}
+
+	return best_flow_idx;
+}
+
+static void
+slb_check_and_trigger_migration(struct slb_evdev *slb, struct slb_queue *queue,
+				struct slb_atomic_state *astate, uint8_t full_port_id)
+{
+	struct slb_atomic_port_stats *pstats = &astate->port_stats[full_port_id];
+	uint64_t now;
+	uint64_t hz;
+	uint64_t cooldown_cycles;
+	int32_t flow_to_migrate;
+
+	pstats->load_counter++;
+	if (pstats->load_counter < SLB_MIGRATION_LOAD_THRESHOLD)
+		return;
+
+	if (astate->migration.in_progress)
+		return;
+
+	if (slb->ports[full_port_id].is_loaded)
+		return;
+
+	if (pstats->flow_count <= 1)
+		return;
+
+	now = rte_get_timer_cycles();
+	hz = rte_get_timer_hz();
+	cooldown_cycles = (hz * SLB_MIGRATION_COOLDOWN_MS) / 1000;
+
+	if (slb->ports[full_port_id].last_migrated_time != 0 &&
+	    (now - slb->ports[full_port_id].last_migrated_time) < cooldown_cycles)
+		return;
+
+	flow_to_migrate = slb_find_lightest_flow_on_port(
+		queue, full_port_id, now, cooldown_cycles);
+	if (flow_to_migrate < 0)
+		return;
+
+	pstats->load_counter = 0;
+	slb->ports[full_port_id].is_loaded = true;
+	astate->migration.in_progress = true;
+	astate->migration.port_id = full_port_id;
+	astate->migration.flow_idx = (uint32_t)flow_to_migrate;
+}
+
+static inline int
+slb_find_migration_target_port(struct slb_evdev *slb, struct slb_queue *queue,
+			       struct slb_atomic_state *astate, uint8_t mig_port,
+			       uint64_t now, uint64_t cooldown_cycles)
+{
+	uint64_t temp_mask = queue->serving_ports_mask;
+	int target_port = -1;
+	int min_flows = INT_MAX;
+
+	while (temp_mask != 0) {
+		uint8_t cand_port = rte_ctz64(temp_mask);
+		struct slb_port *cp = &slb->ports[cand_port];
+
+		temp_mask &= ~(1ULL << cand_port);
+		if (cand_port == mig_port || cp->is_loaded)
+			continue;
+
+		if (cp->last_migrated_time == 0 ||
+		    (now - cp->last_migrated_time) >= cooldown_cycles) {
+			if ((int)astate->port_stats[cand_port].flow_count < min_flows) {
+				min_flows = (int)astate->port_stats[cand_port].flow_count;
+				target_port = cand_port;
+			}
+		} else if (target_port == -1) {
+			target_port = cand_port;
+		}
+	}
+	return target_port;
+}
+
+static inline void
+slb_complete_flow_migration(struct slb_evdev *slb, struct slb_atomic_state *astate,
+			    uint8_t mig_port, int target_port, uint64_t now)
+{
+	uint32_t f_idx = astate->migration.flow_idx;
+
+	rte_atomic_store_explicit(&astate->flow_map.flows[f_idx].port_id,
+				  target_port, rte_memory_order_release);
+
+	for (uint16_t h = 0; h < astate->hold_count; h++) {
+		if (astate->hold_ports[h] == mig_port &&
+		    slb_flow_hash(astate->hold_buffer[h].flow_id) == f_idx) {
+			astate->hold_ports[h] = (uint8_t)target_port;
+		}
+	}
+
+	astate->flow_last_migrated[f_idx] = now;
+	slb->ports[mig_port].last_migrated_time = now;
+	slb->ports[target_port].last_migrated_time = now;
+
+	if (astate->port_stats[mig_port].flow_count > 0)
+		astate->port_stats[mig_port].flow_count--;
+	astate->port_stats[target_port].flow_count++;
+	astate->migrations_count++;
+
+	slb->ports[mig_port].flow_migrations++;
+	slb->ports[mig_port].migrations_out++;
+	slb->ports[target_port].migrations_in++;
+
+	astate->port_stats[mig_port].load_counter = 0;
+	slb->ports[mig_port].is_loaded = false;
+	astate->migration.in_progress = false;
+}
+
+static inline uint8_t
+slb_assign_unmapped_flow_port(struct slb_evdev *slb, struct slb_atomic_state *astate,
+			      uint64_t serving_mask, uint16_t num_serving_ports)
+{
+	uint32_t rr_counter = astate->next_port_rr++;
+	int port_offset = rr_counter % num_serving_ports;
+	uint64_t temp_mask = serving_mask;
+	uint8_t assigned_port = 0;
+
+	for (int j = 0; j <= port_offset; j++) {
+		assigned_port = rte_ctz64(temp_mask);
+		temp_mask &= (temp_mask - 1);
+	}
+
+	if (slb->ports[assigned_port].is_loaded) {
+		uint64_t cand_mask = serving_mask;
+
+		while (cand_mask != 0) {
+			uint8_t cp = rte_ctz64(cand_mask);
+
+			cand_mask &= (cand_mask - 1);
+			if (!slb->ports[cp].is_loaded) {
+				assigned_port = cp;
+				break;
+			}
+		}
+	}
+	return assigned_port;
+}
+
 /* If port queue is empty for an atomic queue, pull events from main atomic queue to fill
  * up port queues.
  */
 static void
-slb_atomic_refill(struct slb_port *port __rte_unused, struct slb_queue *queue)
+slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
 {
+	struct slb_evdev *slb = port->slb;
 	struct rte_event master_events[SLB_MAX_PORT_DEQUEUE_DEPTH];
 	uint16_t n_deq;
 	uint16_t i;
@@ -183,6 +376,35 @@ slb_atomic_refill(struct slb_port *port __rte_unused, struct slb_queue *queue)
 	if (!rte_spinlock_trylock(&astate->refill_lock))
 		return;
 
+	/* 0. Detect migration in progress before hold buffer processing */
+	if (astate->migration.in_progress) {
+		uint8_t mig_port = astate->migration.port_id;
+		uint64_t enq_cnt = astate->port_stats[mig_port].q_enqueued;
+		uint64_t ret_cnt = astate->port_stats[mig_port].q_returned;
+
+		if (enq_cnt == ret_cnt) {
+			uint64_t now = rte_get_timer_cycles();
+			uint64_t hz = rte_get_timer_hz();
+			uint64_t cooldown_cycles =
+				(hz * SLB_MIGRATION_COOLDOWN_MS) / 1000;
+			int target_port = slb_find_migration_target_port(
+				slb, queue, astate, mig_port, now,
+				cooldown_cycles);
+
+			if (target_port != -1)
+				slb_complete_flow_migration(
+					slb, astate, mig_port, target_port, now);
+			else {
+				astate->port_stats[mig_port].load_counter = 0;
+				slb->ports[mig_port].is_loaded = false;
+				astate->migration.in_progress = false;
+			}
+		}
+	}
+
+	if (astate->migration.in_progress)
+		full_ports_mask |= (1ULL << astate->migration.port_id);
+
 	/* 1. Process Hold Buffer first */
 	if (astate->hold_count > 0) {
 		uint16_t retained = 0;
@@ -216,6 +438,10 @@ slb_atomic_refill(struct slb_port *port __rte_unused, struct slb_queue *queue)
 				&astate->hold_buffer[i], burst_len, NULL);
 			astate->port_stats[dest_port_id].q_enqueued += enqueued;
 
+			if (enqueued > 0 &&
+			    astate->port_stats[dest_port_id].load_counter > 0)
+				astate->port_stats[dest_port_id].load_counter--;
+
 			if (enqueued < burst_len) {
 				uint16_t rem = burst_len - enqueued;
 				uint16_t start_idx = i + enqueued;
@@ -230,6 +456,9 @@ slb_atomic_refill(struct slb_port *port __rte_unused, struct slb_queue *queue)
 						rem);
 				}
 				retained += rem;
+				if (enqueued == 0)
+					slb_check_and_trigger_migration(
+						slb, queue, astate, dest_port_id);
 			}
 
 			i += burst_len;
@@ -274,17 +503,8 @@ slb_atomic_refill(struct slb_port *port __rte_unused, struct slb_queue *queue)
 					rte_memory_order_acquire);
 
 			if (unlikely(dest_port_id == -1)) {
-				uint32_t rr_counter = astate->next_port_rr++;
-				int port_offset = rr_counter % num_serving_ports;
-				uint64_t temp_mask = serving_mask;
-				uint8_t assigned_port = 0;
-
-				for (int j = 0; j <= port_offset; j++) {
-					assigned_port = rte_ctz64(temp_mask);
-					temp_mask &= (temp_mask - 1);
-				}
-
-				dest_port_id = assigned_port;
+				dest_port_id = slb_assign_unmapped_flow_port(
+					slb, astate, serving_mask, num_serving_ports);
 				rte_atomic_store_explicit(
 					&astate->flow_map.flows[flow_idx].port_id,
 					dest_port_id, rte_memory_order_release);
@@ -346,7 +566,10 @@ slb_atomic_refill(struct slb_port *port __rte_unused, struct slb_queue *queue)
 				queue->per_port_rings[p], &sorted_events[p_start], p_count, NULL);
 			astate->port_stats[p].q_enqueued += enqueued;
 
-			if (enqueued < p_count) {
+			if (enqueued == p_count) {
+				if (unlikely(astate->port_stats[p].load_counter > 0))
+					astate->port_stats[p].load_counter--;
+			} else {
 				uint16_t rem = p_count - enqueued;
 				uint16_t space = SLB_HOLD_BUFFER_SIZE - astate->hold_count;
 				uint16_t to_copy = RTE_MIN(rem, space);
@@ -359,6 +582,9 @@ slb_atomic_refill(struct slb_port *port __rte_unused, struct slb_queue *queue)
 					memset(&astate->hold_ports[astate->hold_count], p, to_copy);
 					astate->hold_count += to_copy;
 				}
+				if (enqueued == 0)
+					slb_check_and_trigger_migration(
+						slb, queue, astate, p);
 			}
 		}
 	}
-- 
2.53.0