* [PATCH v1 1/8] event/slb: introduce software load balancer PMD
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
@ 2026-11-07 1:30 ` Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 2/8] event/slb: add support for ordered queues Pravin Pathak
` (7 subsequent siblings)
8 siblings, 0 replies; 14+ messages in thread
From: Pravin Pathak @ 2026-11-07 1:30 UTC (permalink / raw)
To: dev
Cc: jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
From: Pravin Pathak <pravin.pathak.dev@gmail.com>
Add basic skeleton and initial features for the Software Load Balancer
(SLB) eventdev PMD.
This initial commit provides:
- Virtual device driver registration (event_slb) and device probing.
- Device configuration and capability reporting for parallel queues,
burst mode, multi-queue ports, and runtime port linking.
- Parallel queue support (multi-producer/multi-consumer rings) and
single-link queue support (multi-producer/single-consumer rings).
- Event port configuration, linking/unlinking, and multi-priority
queue scheduling.
- Central in-flight event credit management with batched loans
(SLB_MIN_CREDIT_LOAN = 512).
- High-performance event enqueue and dequeue burst processing.
- Meson build system integration and MAINTAINERS entry.
Signed-off-by: Pravin Pathak <pravin.pathak.dev@gmail.com>
---
MAINTAINERS | 4 +
drivers/event/meson.build | 1 +
drivers/event/slb/meson.build | 15 +
drivers/event/slb/slb_evdev.c | 585 ++++++++++++++++++++++++++++++++++
drivers/event/slb/slb_evdev.h | 176 ++++++++++
drivers/event/slb/slb_event.c | 279 ++++++++++++++++
6 files changed, 1060 insertions(+)
create mode 100644 drivers/event/slb/meson.build
create mode 100644 drivers/event/slb/slb_evdev.c
create mode 100644 drivers/event/slb/slb_evdev.h
create mode 100644 drivers/event/slb/slb_event.c
diff --git a/MAINTAINERS b/MAINTAINERS
index 93ad53027d..89ea53695a 100644
--- a/MAINTAINERS
+++ b/MAINTAINERS
@@ -1533,6 +1533,10 @@ M: Mattias Rönnblom <mattias.ronnblom@ericsson.com>
F: drivers/event/dsw/
F: doc/guides/eventdevs/dsw.rst
+Software Load Balancer (SLB) Eventdev PMD
+M: Pravin Pathak <pravin.pathak.dev@gmail.com>
+F: drivers/event/slb/
+
Software OPDL Eventdev PMD
M: Liang Ma <liangma@liangbit.com>
M: Peter Mccarthy <peter.mccarthy@intel.com>
diff --git a/drivers/event/meson.build b/drivers/event/meson.build
index ddaa0300ac..ece490dfb5 100644
--- a/drivers/event/meson.build
+++ b/drivers/event/meson.build
@@ -15,5 +15,6 @@ drivers = [
'opdl',
'skeleton',
'sw',
+ 'slb',
]
std_deps = ['eventdev', 'kvargs']
diff --git a/drivers/event/slb/meson.build b/drivers/event/slb/meson.build
new file mode 100644
index 0000000000..9a1dff524f
--- /dev/null
+++ b/drivers/event/slb/meson.build
@@ -0,0 +1,15 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2026 Pravin Pathak <pravin.pathak.dev@gmail.com>
+
+if is_windows
+ build = false
+ reason = 'not supported on Windows'
+ subdir_done()
+endif
+
+sources = files(
+ 'slb_evdev.c',
+ 'slb_event.c',
+)
+
+deps += ['bus_vdev', 'ring']
diff --git a/drivers/event/slb/slb_evdev.c b/drivers/event/slb/slb_evdev.c
new file mode 100644
index 0000000000..53af113293
--- /dev/null
+++ b/drivers/event/slb/slb_evdev.c
@@ -0,0 +1,585 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2026 Pravin Pathak <pravin.pathak.dev@gmail.com>
+ */
+
+#include <stdbool.h>
+#include <inttypes.h>
+#include <rte_bitops.h>
+#include <rte_cycles.h>
+#include <rte_malloc.h>
+#include <rte_ring_elem.h>
+#include <rte_errno.h>
+#include <rte_stdatomic.h>
+#include <eventdev_pmd.h>
+#include <eventdev_pmd_vdev.h>
+
+#include "slb_evdev.h"
+
+#define EVENTDEV_NAME_SLB_PMD event_slb
+
+/* Setup Eventdev port. Reset all mapped queues */
+static int
+slb_port_setup(struct rte_eventdev *dev, uint8_t port_id,
+ const struct rte_event_port_conf *conf)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ struct slb_port *port;
+ bool implicit_release;
+ uint16_t deq_depth;
+ uint16_t enq_depth;
+
+ port = &slb->ports[port_id];
+
+ implicit_release =
+ !(conf->event_port_cfg & RTE_EVENT_PORT_CFG_DISABLE_IMPL_REL);
+
+ deq_depth = conf->dequeue_depth;
+ if (deq_depth == 0 || deq_depth > SLB_MAX_PORT_DEQUEUE_DEPTH)
+ deq_depth = SLB_MAX_PORT_DEQUEUE_DEPTH;
+
+ enq_depth = conf->enqueue_depth;
+ if (enq_depth == 0 || enq_depth > SLB_MAX_PORT_ENQUEUE_DEPTH)
+ enq_depth = SLB_MAX_PORT_ENQUEUE_DEPTH;
+
+ *port = (struct slb_port) {
+ .id = port_id,
+ .slb = slb,
+ .dequeue_depth = deq_depth,
+ .enqueue_depth = enq_depth,
+ .new_event_threshold = conf->new_event_threshold,
+ .implicit_release = implicit_release,
+ .is_single_linked = conf->event_port_cfg &
+ RTE_EVENT_PORT_CFG_SINGLE_LINK,
+ .inflight_credits = 0,
+ .credits_drained = false,
+ .total_dequeues = 0,
+ .empty_dequeues = 0,
+ .single_linked_q = NULL,
+ .single_linked_ring = NULL,
+ };
+
+ port->remap_input_q = false;
+ rte_atomic_store_explicit(&port->active_queues, 0,
+ rte_memory_order_relaxed);
+
+ dev->data->ports[port_id] = port;
+
+ /* Reset all queues mapped to this port */
+ for (int p = 0; p < SLB_MAX_PRIORITIES; p++) {
+ port->next_q[p] = 0;
+ port->num_q[p] = 0;
+ memset(port->q_entries[p], 0, sizeof(port->q_entries[p]));
+ }
+
+ return 0;
+}
+
+/* Default Eventdev port configuration for SLB */
+static void
+slb_port_def_conf(struct rte_eventdev *dev __rte_unused,
+ uint8_t port_id __rte_unused,
+ struct rte_event_port_conf *port_conf)
+{
+ *port_conf = (struct rte_event_port_conf) {
+ .new_event_threshold = 1024,
+ .dequeue_depth = SLB_MAX_PORT_DEQUEUE_DEPTH / 4,
+ .enqueue_depth = SLB_MAX_PORT_ENQUEUE_DEPTH / 4
+ };
+}
+
+static void
+slb_port_release(void *p __rte_unused)
+{
+}
+
+/* Called to free queue resources on device close or reconfiguration */
+static void
+slb_queue_release(struct rte_eventdev *dev, uint8_t queue_id)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ struct slb_queue *queue = &slb->queues[queue_id];
+
+ if (queue->q_ring) {
+ rte_event_ring_free(queue->q_ring);
+ queue->q_ring = NULL;
+ }
+}
+
+/* Setup Eventdev queue. Eventdev queues are implemented as DPDK rings */
+static int
+slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
+ const struct rte_event_queue_conf *conf)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ struct slb_queue *queue = &slb->queues[queue_id];
+ struct rte_event_ring *ring;
+ char ring_name[RTE_RING_NAMESIZE];
+ uint32_t q_size = (slb->max_inflight > 0) ?
+ (uint32_t)slb->max_inflight : SLB_IN_RING_SIZE;
+
+ /* Release existing queue resources if reconfiguring */
+ slb_queue_release(dev, queue_id);
+
+ snprintf(ring_name, sizeof(ring_name), "slb%d_q%u", dev->data->dev_id,
+ queue_id);
+
+ if (conf->event_queue_cfg & RTE_EVENT_QUEUE_CFG_SINGLE_LINK) {
+ /* MP -> SC ring for single-link queues */
+ queue->is_single_link = true;
+ ring = rte_event_ring_create(ring_name, q_size,
+ dev->data->socket_id, RING_F_SC_DEQ | RING_F_EXACT_SZ);
+ } else if (conf->schedule_type == RTE_SCHED_TYPE_PARALLEL) {
+ /* MP -> MC ring for parallel queues */
+ queue->is_single_link = false;
+ ring = rte_event_ring_create(ring_name, q_size,
+ dev->data->socket_id, RING_F_EXACT_SZ);
+ } else {
+ SLB_LOG_DP_LINE(ERR, "Unsupported schedule_type %u",
+ conf->schedule_type);
+ return -ENOTSUP;
+ }
+
+ if (ring == NULL)
+ return -ENOMEM;
+
+ queue->q_ring = ring;
+ queue->schedule_type = conf->schedule_type;
+ queue->serving_ports_mask = 0;
+ queue->num_serving_ports = 0;
+
+ return 0;
+}
+
+/* Settings for default Eventdev queue configuration */
+static void
+slb_queue_def_conf(struct rte_eventdev *dev __rte_unused,
+ uint8_t queue_id __rte_unused,
+ struct rte_event_queue_conf *queue_conf)
+{
+ *queue_conf = (struct rte_event_queue_conf) {
+ .nb_atomic_flows = 4096,
+ .schedule_type = RTE_SCHED_TYPE_PARALLEL,
+ .priority = RTE_EVENT_DEV_PRIORITY_NORMAL
+ };
+}
+
+static void
+slb_port_unlink_queue_from_prio(struct slb_port *port, int prio, int slot)
+{
+ for (int k = slot; k < port->num_q[prio] - 1; k++)
+ port->q_entries[prio][k] = port->q_entries[prio][k + 1];
+
+ port->num_q[prio]--;
+ memset(&port->q_entries[prio][port->num_q[prio]], 0,
+ sizeof(struct slb_port_q_entry));
+ if (port->next_q[prio] >= port->num_q[prio])
+ port->next_q[prio] = 0;
+}
+
+static int
+slb_port_find_queue_slot(struct slb_port *port, int prio, uint8_t qid)
+{
+ for (int s = 0; s < port->num_q[prio]; s++) {
+ if (port->q_entries[prio][s].queue_id == qid)
+ return s;
+ }
+ return -1;
+}
+
+static int
+slb_port_link_one(struct slb_port *port, struct slb_queue *queue,
+ uint8_t qid, uint8_t app_prio, bool has_prio)
+{
+ int prio;
+ bool found = false;
+
+ if (port->is_single_linked) {
+ if (port->single_linked_q != NULL &&
+ port->single_linked_q != queue) {
+ rte_errno = EINVAL;
+ return -EINVAL;
+ }
+ port->single_linked_q = queue;
+ port->single_linked_ring = queue->q_ring;
+ } else {
+ if (has_prio)
+ prio = (app_prio < RTE_EVENT_DEV_PRIORITY_NORMAL) ? 0 : 1;
+ else
+ prio = SLB_MAX_PRIORITIES - 1;
+
+ for (int pr = 0; pr < SLB_MAX_PRIORITIES; pr++) {
+ int s = slb_port_find_queue_slot(port, pr, qid);
+
+ if (s >= 0) {
+ if (pr == prio)
+ found = true;
+ else
+ slb_port_unlink_queue_from_prio(port, pr, s);
+ break;
+ }
+ }
+ if (!found) {
+ uint16_t slot = port->num_q[prio];
+
+ if (slot >= SLB_PORT_MAX_Q_MAP_PER_PRIO) {
+ rte_errno = ENOSPC;
+ return -ENOSPC;
+ }
+ port->q_entries[prio][slot].dq_ring = queue->q_ring;
+ port->q_entries[prio][slot].queue = queue;
+ port->q_entries[prio][slot].queue_id = qid;
+ port->q_entries[prio][slot].schedule_type =
+ queue->schedule_type;
+ port->num_q[prio]++;
+ }
+
+ rte_atomic_fetch_or_explicit(&port->active_queues,
+ (1ULL << qid),
+ rte_memory_order_release);
+ }
+
+ queue->serving_ports_mask |= (1ULL << port->id);
+ queue->num_serving_ports = rte_popcount64(queue->serving_ports_mask);
+ return 0;
+}
+
+static int
+slb_port_unlink_one(struct slb_port *port, struct slb_queue *queue, uint8_t qid)
+{
+ bool found = false;
+
+ if (port->is_single_linked) {
+ if (port->single_linked_q != queue) {
+ rte_errno = EINVAL;
+ return -EINVAL;
+ }
+ port->single_linked_q = NULL;
+ port->single_linked_ring = NULL;
+ } else {
+ for (int prio = 0; prio < SLB_MAX_PRIORITIES; prio++) {
+ int s = slb_port_find_queue_slot(port, prio, qid);
+
+ if (s >= 0) {
+ slb_port_unlink_queue_from_prio(port, prio, s);
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ rte_errno = EINVAL;
+ return -EINVAL;
+ }
+ rte_atomic_fetch_and_explicit(&port->active_queues,
+ ~(1ULL << qid),
+ rte_memory_order_release);
+ port->remap_input_q = true;
+ }
+
+ queue->serving_ports_mask &= ~(1ULL << port->id);
+ queue->num_serving_ports = rte_popcount64(queue->serving_ports_mask);
+ return 0;
+}
+
+static int
+slb_port_link(struct rte_eventdev *dev, void *p, const uint8_t queues[],
+ const uint8_t priorities[], uint16_t num)
+{
+ struct slb_port *port = p;
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ uint16_t i;
+
+ for (i = 0; i < num; i++) {
+ uint8_t qid = queues[i];
+ struct slb_queue *queue = &slb->queues[qid];
+ uint8_t prio = (priorities != NULL) ? priorities[i] : 0;
+ int ret = slb_port_link_one(port, queue, qid, prio,
+ priorities != NULL);
+
+ if (ret < 0)
+ return i;
+ }
+ return num;
+}
+
+static int
+slb_port_unlink(struct rte_eventdev *dev, void *p, uint8_t queues[],
+ uint16_t num)
+{
+ struct slb_port *port = p;
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ uint16_t i;
+
+ for (i = 0; i < num; i++) {
+ uint8_t qid = queues[i];
+ struct slb_queue *queue = &slb->queues[qid];
+ int ret = slb_port_unlink_one(port, queue, qid);
+
+ if (ret < 0)
+ return i;
+ }
+ return num;
+}
+
+static void
+slb_info_get(struct rte_eventdev *dev __rte_unused,
+ struct rte_event_dev_info *info)
+{
+ *info = (struct rte_event_dev_info) {
+ .driver_name = SLB_PMD_NAME,
+ .max_event_queues = SLB_MAX_QUEUES,
+ .max_event_queue_flows = SLB_MAX_FLOWS,
+ .max_event_queue_priority_levels = SLB_MAX_PRIORITIES,
+ .max_event_priority_levels = SLB_MAX_PRIORITIES,
+ .max_event_ports = SLB_MAX_PORTS,
+ .max_event_port_dequeue_depth = SLB_MAX_PORT_DEQUEUE_DEPTH,
+ .max_event_port_enqueue_depth = SLB_MAX_PORT_ENQUEUE_DEPTH,
+ .max_num_events = SLB_MAX_EVENTS,
+ .max_profiles_per_port = 1,
+ .event_dev_cap = RTE_EVENT_DEV_CAP_BURST_MODE |
+ RTE_EVENT_DEV_CAP_PARALLEL |
+ RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED |
+ RTE_EVENT_DEV_CAP_IMPLICIT_RELEASE_DISABLE |
+ RTE_EVENT_DEV_CAP_NONSEQ_MODE |
+ RTE_EVENT_DEV_CAP_MULTIPLE_QUEUE_PORT |
+ RTE_EVENT_DEV_CAP_RUNTIME_PORT_LINK |
+ RTE_EVENT_DEV_CAP_CARRY_FLOW_ID |
+ RTE_EVENT_DEV_CAP_MAINTENANCE_FREE |
+ RTE_EVENT_DEV_CAP_INDEPENDENT_ENQ
+ };
+}
+
+static int
+slb_configure(const struct rte_eventdev *dev)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ const struct rte_event_dev_config *conf = &dev->data->dev_conf;
+ int32_t min_max_in_flight;
+ uint32_t q_size;
+
+ if (conf->nb_events_limit <= 0 ||
+ conf->nb_events_limit > SLB_MAX_EVENTS) {
+ SLB_LOG_DP_LINE(INFO,
+ "nb_events_limit (%d) invalid, max is %d",
+ conf->nb_events_limit, SLB_MAX_EVENTS);
+ return -EINVAL;
+ }
+
+ slb->num_ports = conf->nb_event_ports;
+ slb->num_queues = conf->nb_event_queues;
+
+ min_max_in_flight = conf->nb_event_ports * SLB_PORT_MAX_CREDITS;
+ if (min_max_in_flight > SLB_MAX_EVENTS)
+ min_max_in_flight = SLB_MAX_EVENTS;
+
+ q_size = rte_align32pow2((uint32_t)RTE_MAX(conf->nb_events_limit,
+ min_max_in_flight));
+ if (q_size > SLB_MAX_EVENTS)
+ q_size = SLB_MAX_EVENTS;
+
+ slb->max_inflight = (int32_t)q_size;
+ return 0;
+}
+
+static int
+slb_start(struct rte_eventdev *dev)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+
+ for (int i = 0; i < slb->num_queues; i++) {
+ if (slb->queues[i].serving_ports_mask == 0)
+ SLB_LOG_DP_LINE(INFO,
+ "Queue %d has no linked ports on start", i);
+ }
+
+ rte_atomic_store_explicit(&slb->credits_on_loan, 0,
+ rte_memory_order_relaxed);
+
+ for (int i = 0; i < slb->num_ports; i++) {
+ struct slb_port *port = &slb->ports[i];
+ port->inflight_credits = 0;
+ port->credits_drained = false;
+ for (int p = 0; p < SLB_MAX_PRIORITIES; p++)
+ port->next_q[p] = 0;
+ }
+
+ return 0;
+}
+
+static void
+slb_drain_ring(uint8_t dev_id, struct rte_event_ring *ring,
+ eventdev_stop_flush_t flush, void *flush_arg)
+{
+ struct rte_event ev;
+
+ if (ring == NULL)
+ return;
+
+ while (rte_event_ring_dequeue_burst(ring, &ev, 1, NULL)) {
+ if (flush != NULL)
+ flush(dev_id, ev, flush_arg);
+ }
+ rte_ring_reset(&ring->r);
+}
+
+static void
+slb_stop(struct rte_eventdev *dev)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ uint8_t dev_id = dev->data->dev_id;
+ eventdev_stop_flush_t flush = dev->dev_ops->dev_stop_flush;
+ void *flush_arg = dev->data->dev_stop_flush_arg;
+
+ for (uint8_t qid = 0; qid < slb->num_queues; qid++) {
+ struct slb_queue *q = &slb->queues[qid];
+ if (q->q_ring != NULL)
+ slb_drain_ring(dev_id, q->q_ring, flush, flush_arg);
+ }
+
+ for (int i = 0; i < slb->num_ports; i++) {
+ struct slb_port *port = &slb->ports[i];
+ port->inflight_credits = 0;
+ port->credits_drained = false;
+ for (int p = 0; p < SLB_MAX_PRIORITIES; p++)
+ port->next_q[p] = 0;
+ }
+
+ rte_atomic_store_explicit(&slb->credits_on_loan, 0,
+ rte_memory_order_relaxed);
+}
+
+static int
+slb_close(struct rte_eventdev *dev)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+
+ for (int i = 0; i < slb->num_queues; i++)
+ slb_queue_release(dev, i);
+
+ return 0;
+}
+
+void
+slb_dump(struct rte_eventdev *dev, FILE *f)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ int32_t pool_on_loan, pool_avail;
+
+ if (dev == NULL || f == NULL)
+ return;
+
+ pool_on_loan = rte_atomic_load_explicit(&slb->credits_on_loan,
+ rte_memory_order_relaxed);
+ pool_avail = slb->max_inflight - pool_on_loan;
+
+ fprintf(f, "\n[SLB EVENTDEV DEBUG STATS]\n");
+ fprintf(f, " Device ID: %u, Driver: %s\n", dev->data->dev_id,
+ SLB_PMD_NAME);
+ fprintf(f, " Credit Pool: %d on_loan / %d max_inflight (%d avail)\n",
+ pool_on_loan, slb->max_inflight, pool_avail);
+
+ fprintf(f, "\n PORT STATISTICS & QUEUE LINKAGES:\n");
+ for (int i = 0; i < slb->num_ports; i++) {
+ struct slb_port *p = &slb->ports[i];
+ uint64_t aq = rte_atomic_load_explicit(&p->active_queues,
+ rte_memory_order_acquire);
+ fprintf(f, " Port %2u (credits=%3d, active=0x%08" PRIx64 "): ",
+ i, p->inflight_credits, aq);
+ int linked_count = 0;
+ if (p->is_single_linked) {
+ if (p->single_linked_q) {
+ int qid = p->single_linked_q - slb->queues;
+ fprintf(f, "Q%d (single-link)", qid);
+ linked_count++;
+ }
+ } else {
+ for (int pr = 0; pr < SLB_MAX_PRIORITIES; pr++) {
+ for (int s = 0; s < p->num_q[pr]; s++) {
+ int qid = p->q_entries[pr][s].queue_id;
+ if (linked_count > 0)
+ fprintf(f, ", ");
+ fprintf(f, "Q%d(p%d)", qid, pr);
+ linked_count++;
+ }
+ }
+ }
+ if (linked_count == 0)
+ fprintf(f, "(no queues linked)");
+ fprintf(f, "\n");
+ }
+
+ fprintf(f, "\n ALL QUEUES STATUS:\n");
+ for (int i = 0; i < slb->num_queues; i++) {
+ struct slb_queue *q = &slb->queues[i];
+ uint32_t count = rte_event_ring_count(q->q_ring);
+ uint32_t ring_size = rte_event_ring_get_capacity(q->q_ring);
+ fprintf(f, " Queue %2u: %6u/%6u events, serving_ports=0x%016" PRIx64 ", sched=%u, single=%d\n",
+ i, count, ring_size, q->serving_ports_mask,
+ q->schedule_type, q->is_single_link);
+ }
+ fprintf(f, "[END DEBUG]\n\n");
+ fflush(f);
+}
+
+static struct eventdev_ops slb_evdev_ops = {
+ .port_setup = slb_port_setup,
+ .port_def_conf = slb_port_def_conf,
+ .port_release = slb_port_release,
+ .queue_setup = slb_queue_setup,
+ .queue_def_conf = slb_queue_def_conf,
+ .queue_release = slb_queue_release,
+ .port_link = slb_port_link,
+ .port_unlink = slb_port_unlink,
+ .dev_infos_get = slb_info_get,
+ .dev_configure = slb_configure,
+ .dev_start = slb_start,
+ .dev_stop = slb_stop,
+ .dev_close = slb_close,
+ .dump = slb_dump,
+};
+
+static int
+slb_probe(struct rte_vdev_device *vdev)
+{
+ const char *name = rte_vdev_device_name(vdev);
+ struct rte_eventdev *dev;
+ struct slb_evdev *slb;
+
+ dev = rte_event_pmd_vdev_init(name, sizeof(struct slb_evdev),
+ rte_socket_id(), vdev);
+ if (dev == NULL)
+ return -EFAULT;
+
+ dev->dev_ops = &slb_evdev_ops;
+ dev->enqueue_burst = slb_event_enqueue_burst;
+ dev->enqueue_new_burst = slb_event_enqueue_new_burst;
+ dev->enqueue_forward_burst = slb_event_enqueue_forward_burst;
+ dev->dequeue_burst = slb_event_dequeue_burst;
+
+ if (rte_eal_process_type() != RTE_PROC_PRIMARY)
+ return 0;
+
+ slb = dev->data->dev_private;
+ slb->data = dev->data;
+
+ event_dev_probing_finish(dev);
+ return 0;
+}
+
+static int
+slb_remove(struct rte_vdev_device *vdev)
+{
+ const char *name = rte_vdev_device_name(vdev);
+
+ if (name == NULL)
+ return -EINVAL;
+
+ return rte_event_pmd_vdev_uninit(name);
+}
+
+static struct rte_vdev_driver evdev_slb_pmd_drv = {
+ .probe = slb_probe,
+ .remove = slb_remove
+};
+
+RTE_PMD_REGISTER_VDEV(EVENTDEV_NAME_SLB_PMD, evdev_slb_pmd_drv);
+RTE_LOG_REGISTER_DEFAULT(event_slb_logtype, NOTICE);
diff --git a/drivers/event/slb/slb_evdev.h b/drivers/event/slb/slb_evdev.h
new file mode 100644
index 0000000000..d7be6746ef
--- /dev/null
+++ b/drivers/event/slb/slb_evdev.h
@@ -0,0 +1,176 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2026 Pravin Pathak <pravin.pathak.dev@gmail.com>
+ */
+
+#ifndef _SLB_EVDEV_H_
+#define _SLB_EVDEV_H_
+
+#include <rte_bitops.h>
+#include <rte_bitset.h>
+#include <rte_event_ring.h>
+#include <rte_eventdev.h>
+#include <rte_stdatomic.h>
+#include <eventdev_pmd.h>
+
+#define SLB_PMD_NAME RTE_STR(event_slb)
+
+#define SLB_MAX_PORTS (64)
+#define SLB_MAX_PORT_DEQUEUE_DEPTH (128)
+#define SLB_MAX_PORT_ENQUEUE_DEPTH (128)
+
+#define SLB_MAX_QUEUES (64)
+
+#define SLB_MAX_EVENTS (16384)
+
+#define SLB_MAX_FLOWS_BITS (16)
+#define SLB_MAX_FLOWS (1 << (SLB_MAX_FLOWS_BITS))
+#define SLB_MAX_FLOWS_MASK (SLB_MAX_FLOWS - 1)
+
+/* Avoid making small 'loans' from the central in-flight event credit pool */
+#define SLB_MIN_CREDIT_LOAN (512)
+#define SLB_PORT_MAX_CREDITS (2 * SLB_MIN_CREDIT_LOAN)
+#define SLB_PORT_MIN_CREDITS (SLB_MIN_CREDIT_LOAN)
+
+#define SLB_IN_RING_SIZE (SLB_MAX_EVENTS)
+
+/* Maximum number of queues that can be linked to a port per priority */
+#define SLB_PORT_MAX_Q_MAP_PER_PRIO (SLB_MAX_QUEUES)
+/* Number of priority levels supported by SLB. Configure 1..4. */
+#ifndef SLB_MAX_PRIORITIES
+#define SLB_MAX_PRIORITIES 2
+#endif
+
+#if (SLB_MAX_PRIORITIES < 1) || (SLB_MAX_PRIORITIES > 4)
+#error "SLB_MAX_PRIORITIES must be between 1 and 4"
+#endif
+
+/* Total max mapping slots (per-prio * priorities) */
+#define SLB_PORT_MAX_Q_MAP (SLB_PORT_MAX_Q_MAP_PER_PRIO * SLB_MAX_PRIORITIES)
+
+struct slb_queue;
+
+/* Pre-resolved queue dequeue entry for core-local port access */
+struct slb_port_q_entry {
+ struct rte_event_ring *dq_ring;
+ struct slb_queue *queue;
+ uint8_t queue_id;
+ uint8_t schedule_type;
+ uint8_t pad[6];
+};
+
+struct __rte_cache_aligned slb_port {
+ /* --- Cacheline 0: Hot Rx & Tx scheduling control (64 bytes) --- */
+ uint16_t id;
+ int32_t inflight_credits;
+ int32_t new_event_threshold;
+ RTE_ATOMIC(uint64_t) active_queues;
+ uint16_t num_q[SLB_MAX_PRIORITIES];
+ uint16_t next_q[SLB_MAX_PRIORITIES];
+ bool is_single_linked;
+ bool implicit_release;
+ bool credits_drained;
+ bool remap_input_q;
+ uint8_t pad_port[4];
+ struct rte_event_ring *single_linked_ring;
+ struct slb_evdev *slb;
+
+ /* --- Cachelines 1..6: Pre-resolved queue mappings for dequeue --- */
+ alignas(RTE_CACHE_LINE_SIZE)
+ struct slb_port_q_entry q_entries[SLB_MAX_PRIORITIES][SLB_PORT_MAX_Q_MAP_PER_PRIO];
+
+ /* --- Cold Section: Statistics counters and config --- */
+ alignas(RTE_CACHE_LINE_SIZE)
+ uint16_t dequeue_depth;
+ uint16_t enqueue_depth;
+ struct slb_queue *single_linked_q;
+ uint64_t enqueue_calls;
+ uint64_t new_enqueued;
+ uint64_t forward_enqueued;
+ uint64_t release_enqueued;
+ uint64_t dequeue_calls;
+ uint64_t dequeued;
+ uint64_t empty_dequeues;
+ uint64_t total_dequeues;
+ uint64_t queue_enqueued[SLB_MAX_QUEUES];
+ uint64_t queue_dequeued[SLB_MAX_QUEUES];
+};
+
+struct slb_queue {
+ uint8_t schedule_type;
+ bool is_single_link;
+ uint16_t num_serving_ports;
+ uint64_t serving_ports_mask;
+ alignas(RTE_CACHE_LINE_SIZE) struct rte_event_ring *q_ring;
+};
+
+/* Limited by the size of 'serving_ports_mask' and 'active_queues' bitmasks */
+static_assert(SLB_MAX_PORTS <= 64, "Max compile-time port count exceeded");
+static_assert(SLB_MAX_QUEUES <= 64, "Max compile-time queue count exceeded");
+
+struct slb_evdev {
+ struct rte_eventdev_data *data;
+
+ struct slb_port ports[SLB_MAX_PORTS];
+ uint16_t num_ports;
+ struct slb_queue queues[SLB_MAX_QUEUES];
+ uint8_t num_queues;
+ int32_t max_inflight;
+
+ alignas(RTE_CACHE_LINE_SIZE) RTE_ATOMIC(int32_t) credits_on_loan;
+};
+
+uint16_t slb_event_enqueue_burst(void *port,
+ const struct rte_event events[],
+ uint16_t events_len);
+uint16_t slb_event_enqueue_new_burst(void *port,
+ const struct rte_event events[],
+ uint16_t events_len);
+uint16_t slb_event_enqueue_forward_burst(void *port,
+ const struct rte_event events[],
+ uint16_t events_len);
+
+uint16_t slb_event_dequeue_burst(void *port, struct rte_event *events,
+ uint16_t num, uint64_t wait);
+
+void slb_port_return_credits(struct slb_evdev *slb, struct slb_port *port,
+ int32_t credits);
+
+void slb_dump(struct rte_eventdev *dev, FILE *f);
+
+static inline struct slb_evdev *
+slb_pmd_priv(const struct rte_eventdev *eventdev)
+{
+ return eventdev->data->dev_private;
+}
+
+extern int event_slb_logtype;
+#define RTE_LOGTYPE_EVENT_SLB event_slb_logtype
+#define SLB_LOG_DP_LINE(level, fmt, ...) \
+ RTE_LOG_DP_LINE(level, EVENT_SLB, "%s() line %u: " fmt, \
+ __func__, __LINE__, ## __VA_ARGS__)
+
+#define SLB_LOG_DP_PORT_LINE(level, port_id, fmt, ...) \
+ SLB_LOG_DP_LINE(level, "<Port %d> " fmt, port_id, ## __VA_ARGS__)
+
+/* Obtain serving port masks for the specified queue and set corresponding
+ * active queue bits for all served ports.
+ */
+static inline void
+slb_port_wakeup_queues(struct slb_evdev *slb, struct slb_queue *queue)
+{
+ uint8_t queue_id = queue - slb->queues;
+ uint64_t port_mask = queue->serving_ports_mask;
+ struct slb_port *port;
+ uint8_t port_id;
+
+ while (port_mask) {
+ port_id = rte_ctz64(port_mask);
+ port = &slb->ports[port_id];
+ rte_atomic_fetch_or_explicit(&port->active_queues,
+ (1ULL << queue_id),
+ rte_memory_order_release);
+ port_mask &= ~(1ULL << port_id);
+ }
+}
+
+#endif /* _SLB_EVDEV_H_ */
diff --git a/drivers/event/slb/slb_event.c b/drivers/event/slb/slb_event.c
new file mode 100644
index 0000000000..f55262154e
--- /dev/null
+++ b/drivers/event/slb/slb_event.c
@@ -0,0 +1,279 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2026 Pravin Pathak <pravin.pathak.dev@gmail.com>
+ */
+
+#include <stdbool.h>
+#include <rte_cycles.h>
+#include <rte_ring_elem.h>
+#include <rte_stdatomic.h>
+
+#include "slb_evdev.h"
+
+/* Acquire credits for new events */
+static bool
+slb_port_acquire_credits(struct slb_evdev *slb, struct slb_port *port,
+ int32_t credits)
+{
+ int32_t inflight_credits = port->inflight_credits;
+ int32_t remaining_credits = inflight_credits - credits;
+ int32_t total_inflight;
+ int32_t available;
+ int32_t required_credits;
+ int32_t new_total_inflight;
+
+ /* If port has enough credits, update and return true */
+ if (likely(remaining_credits >= 0)) {
+ port->inflight_credits = remaining_credits;
+ return true;
+ }
+
+ total_inflight = rte_atomic_load_explicit(&slb->credits_on_loan,
+ rte_memory_order_relaxed);
+ available = slb->max_inflight - total_inflight;
+ required_credits = RTE_MAX(abs(remaining_credits), SLB_PORT_MIN_CREDITS);
+
+ if (available < required_credits)
+ return false;
+
+ new_total_inflight =
+ rte_atomic_fetch_add_explicit(&slb->credits_on_loan,
+ required_credits,
+ rte_memory_order_relaxed) +
+ required_credits;
+
+ if (unlikely(new_total_inflight > slb->max_inflight)) {
+ rte_atomic_fetch_sub_explicit(&slb->credits_on_loan,
+ required_credits,
+ rte_memory_order_relaxed);
+ return false;
+ }
+
+ port->inflight_credits += required_credits;
+ port->inflight_credits -= credits;
+
+ return true;
+}
+
+/* Return accumulated credits (Terminated events) */
+void
+slb_port_return_credits(struct slb_evdev *slb, struct slb_port *port,
+ int32_t credits)
+{
+ port->inflight_credits += credits;
+
+ if (unlikely(port->inflight_credits >= SLB_PORT_MAX_CREDITS)) {
+ int32_t return_credits = port->inflight_credits -
+ SLB_PORT_MIN_CREDITS;
+ port->inflight_credits = SLB_PORT_MIN_CREDITS;
+
+ rte_atomic_fetch_sub_explicit(&slb->credits_on_loan,
+ return_credits,
+ rte_memory_order_release);
+ }
+}
+
+static void
+slb_port_enqueue_stats(struct slb_port *port, uint16_t num_new,
+ uint16_t num_forward, uint16_t num_release)
+{
+ port->new_enqueued += num_new;
+ port->forward_enqueued += num_forward;
+ port->release_enqueued += num_release;
+}
+
+static void
+slb_port_queue_enqueue_burst_stats(struct slb_port *source_port,
+ uint8_t queue_id, uint16_t num)
+{
+ source_port->queue_enqueued[queue_id] += num;
+}
+
+static inline uint16_t
+slb_event_enqueue_burst_generic(struct slb_port *source_port,
+ const struct rte_event events[],
+ uint16_t events_len, bool op_types_known,
+ uint16_t num_new, uint16_t num_forward,
+ uint16_t num_release)
+{
+ struct slb_evdev *slb = source_port->slb;
+ uint16_t enqueued = 0;
+ uint16_t idx = 0;
+
+ if (unlikely(events_len == 0))
+ return 0;
+
+ if (!op_types_known) {
+ for (uint16_t i = 0; i < events_len; i++) {
+ switch (events[i].op) {
+ case RTE_EVENT_OP_NEW:
+ num_new++;
+ break;
+ case RTE_EVENT_OP_FORWARD:
+ num_forward++;
+ break;
+ case RTE_EVENT_OP_RELEASE:
+ num_release++;
+ break;
+ }
+ }
+ }
+
+ /* Check credit threshold for NEW events */
+ if (num_new > 0) {
+ if (unlikely(rte_atomic_load_explicit(&slb->credits_on_loan,
+ rte_memory_order_relaxed) >
+ source_port->new_event_threshold))
+ return 0;
+
+ if (unlikely(!slb_port_acquire_credits(slb, source_port,
+ num_new)))
+ return 0;
+ }
+
+ if (num_release > 0)
+ slb_port_return_credits(slb, source_port, num_release);
+
+ slb_port_enqueue_stats(source_port, num_new, num_forward, num_release);
+
+ /* Enqueue contiguous chunks targeting the same queue */
+ while (idx < events_len) {
+ if (events[idx].op == RTE_EVENT_OP_RELEASE) {
+ idx++;
+ enqueued++;
+ continue;
+ }
+
+ uint8_t qid = events[idx].queue_id;
+ uint16_t chunk_len = 1;
+
+ while (idx + chunk_len < events_len &&
+ events[idx + chunk_len].queue_id == qid &&
+ events[idx + chunk_len].op != RTE_EVENT_OP_RELEASE) {
+ chunk_len++;
+ }
+
+ struct slb_queue *queue = &slb->queues[qid];
+ uint16_t actual_enq = rte_event_ring_enqueue_burst(queue->q_ring,
+ &events[idx],
+ chunk_len,
+ NULL);
+ slb_port_queue_enqueue_burst_stats(source_port, qid, actual_enq);
+ enqueued += actual_enq;
+
+ if (actual_enq < chunk_len) {
+ /* If queue ring is full, return unused credits for dropped NEW events */
+ uint16_t dropped_new = 0;
+ for (uint16_t d = actual_enq; d < chunk_len; d++) {
+ if (events[idx + d].op == RTE_EVENT_OP_NEW)
+ dropped_new++;
+ }
+ if (dropped_new > 0)
+ slb_port_return_credits(slb, source_port, dropped_new);
+ break;
+ }
+
+ idx += chunk_len;
+ }
+
+ return enqueued;
+}
+
+uint16_t
+slb_event_enqueue_burst(void *port, const struct rte_event events[],
+ uint16_t events_len)
+{
+ struct slb_port *source_port = port;
+ return slb_event_enqueue_burst_generic(source_port, events, events_len,
+ false, 0, 0, 0);
+}
+
+uint16_t
+slb_event_enqueue_new_burst(void *port, const struct rte_event events[],
+ uint16_t events_len)
+{
+ struct slb_port *source_port = port;
+ return slb_event_enqueue_burst_generic(source_port, events, events_len,
+ true, events_len, 0, 0);
+}
+
+uint16_t
+slb_event_enqueue_forward_burst(void *port, const struct rte_event events[],
+ uint16_t events_len)
+{
+ struct slb_port *source_port = port;
+ return slb_event_enqueue_burst_generic(source_port, events, events_len,
+ true, 0, events_len, 0);
+}
+
+static inline uint16_t
+slb_port_dequeue(struct slb_port *port, struct rte_event events[],
+ uint16_t num)
+{
+ uint16_t total_deq = 0;
+
+ /* Single-link direct fast path */
+ if (port->is_single_linked) {
+ if (unlikely(!port->single_linked_ring))
+ return 0;
+
+ total_deq = rte_event_ring_dequeue_burst(port->single_linked_ring,
+ events, num, NULL);
+ if (total_deq > 0) {
+ port->total_dequeues += total_deq;
+ if (port->implicit_release)
+ slb_port_return_credits(port->slb, port,
+ total_deq);
+ }
+ return total_deq;
+ }
+
+ /* Parallel queue dequeue across priority levels */
+ for (int p = 0; p < SLB_MAX_PRIORITIES; p++) {
+ uint16_t num_q = port->num_q[p];
+ if (num_q == 0)
+ continue;
+
+ uint16_t q_idx = port->next_q[p];
+ for (uint16_t i = 0; i < num_q; i++) {
+ struct slb_port_q_entry *entry = &port->q_entries[p][q_idx];
+ uint16_t req = num - total_deq;
+ uint16_t deq = rte_event_ring_dequeue_burst(entry->dq_ring,
+ &events[total_deq],
+ req, NULL);
+ if (deq > 0) {
+ port->queue_dequeued[entry->queue_id] += deq;
+ total_deq += deq;
+ }
+
+ q_idx++;
+ if (q_idx >= num_q)
+ q_idx = 0;
+
+ if (total_deq == num)
+ break;
+ }
+ port->next_q[p] = q_idx;
+
+ if (total_deq > 0)
+ break;
+ }
+
+ port->total_dequeues += total_deq;
+ return total_deq;
+}
+
+uint16_t
+slb_event_dequeue_burst(void *p, struct rte_event events[], uint16_t num,
+ uint64_t wait __rte_unused)
+{
+ struct slb_port *port = p;
+
+ port->dequeue_calls++;
+ uint16_t dequeued = slb_port_dequeue(port, events, num);
+ port->dequeued += dequeued;
+
+ if (dequeued == 0)
+ port->empty_dequeues++;
+
+ return dequeued;
+}
--
2.53.0
^ permalink raw reply related [flat|nested] 14+ messages in thread* [PATCH v1 2/8] event/slb: add support for ordered queues
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 1/8] event/slb: introduce software load balancer PMD Pravin Pathak
@ 2026-11-07 1:30 ` Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 3/8] event/slb: add support for atomic queues Pravin Pathak
` (6 subsequent siblings)
8 siblings, 0 replies; 14+ messages in thread
From: Pravin Pathak @ 2026-11-07 1:30 UTC (permalink / raw)
To: dev
Cc: jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
From: Pravin Pathak <pravin.pathak.dev@gmail.com>
Add support for RTE_SCHED_TYPE_ORDERED queues to provide strict packet
reordering guarantees across worker ports.
A reorder buffer is allocated for each ordered queue to hold forwarded
events until preceding sequence slots have completed. On dequeue from an
ordered queue, sequence numbers are atomically reserved and recorded in
the port's history tracker.
When worker ports enqueue forwarded or released events, the corresponding
reorder buffer entries are marked ready or released. Contiguous ready
events are drained in order to destination queues via slb_reorder_drain.
Signed-off-by: Pravin Pathak <pravin.pathak.dev@gmail.com>
---
drivers/event/slb/slb_evdev.c | 45 ++-
drivers/event/slb/slb_evdev.h | 39 +++
drivers/event/slb/slb_event.c | 529 +++++++++++++++++++++++++++++-----
3 files changed, 539 insertions(+), 74 deletions(-)
diff --git a/drivers/event/slb/slb_evdev.c b/drivers/event/slb/slb_evdev.c
index 53af113293..13be490636 100644
--- a/drivers/event/slb/slb_evdev.c
+++ b/drivers/event/slb/slb_evdev.c
@@ -17,6 +17,25 @@
#define EVENTDEV_NAME_SLB_PMD event_slb
+/* Reset history tracking table for a port.
+ * Token 0 is reserved for parallel and single-link events without history.
+ * Tokens 1 to 255 (255 total) are available for ordered and atomic queue events.
+ */
+static void
+slb_port_reset_history(struct slb_port *port)
+{
+ port->history[0].active = false;
+ port->history[0].next_free = SLB_PORT_HISTORY_FREE_END;
+ for (int i = 1; i < SLB_PORT_HISTORY_SIZE; i++) {
+ port->history[i].active = false;
+ port->history[i].next_free = i + 1;
+ }
+ port->history[SLB_PORT_HISTORY_SIZE - 1].next_free =
+ SLB_PORT_HISTORY_FREE_END;
+ port->history_head = 1;
+ port->history_avail = SLB_PORT_HISTORY_MAX_TOKENS;
+}
+
/* Setup Eventdev port. Reset all mapped queues */
static int
slb_port_setup(struct rte_eventdev *dev, uint8_t port_id,
@@ -54,6 +73,7 @@ slb_port_setup(struct rte_eventdev *dev, uint8_t port_id,
.credits_drained = false,
.total_dequeues = 0,
.empty_dequeues = 0,
+ .pending_releases = 0,
.single_linked_q = NULL,
.single_linked_ring = NULL,
};
@@ -71,6 +91,8 @@ slb_port_setup(struct rte_eventdev *dev, uint8_t port_id,
memset(port->q_entries[p], 0, sizeof(port->q_entries[p]));
}
+ slb_port_reset_history(port);
+
return 0;
}
@@ -99,6 +121,10 @@ slb_queue_release(struct rte_eventdev *dev, uint8_t queue_id)
struct slb_evdev *slb = slb_pmd_priv(dev);
struct slb_queue *queue = &slb->queues[queue_id];
+ if (queue->reorder_buffer) {
+ rte_free(queue->reorder_buffer);
+ queue->reorder_buffer = NULL;
+ }
if (queue->q_ring) {
rte_event_ring_free(queue->q_ring);
queue->q_ring = NULL;
@@ -128,7 +154,8 @@ slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
queue->is_single_link = true;
ring = rte_event_ring_create(ring_name, q_size,
dev->data->socket_id, RING_F_SC_DEQ | RING_F_EXACT_SZ);
- } else if (conf->schedule_type == RTE_SCHED_TYPE_PARALLEL) {
+ } else if (conf->schedule_type == RTE_SCHED_TYPE_PARALLEL ||
+ conf->schedule_type == RTE_SCHED_TYPE_ORDERED) {
/* MP -> MC ring for parallel queues */
queue->is_single_link = false;
ring = rte_event_ring_create(ring_name, q_size,
@@ -147,6 +174,21 @@ slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
queue->serving_ports_mask = 0;
queue->num_serving_ports = 0;
+ /* Ordered queue reorder buffer allocation */
+ if (conf->schedule_type == RTE_SCHED_TYPE_ORDERED) {
+ queue->reorder_buffer = rte_zmalloc_socket("slb_reorder_buf",
+ sizeof(struct slb_reorder_entry) * ring->r.size,
+ RTE_CACHE_LINE_SIZE, dev->data->socket_id);
+ if (queue->reorder_buffer == NULL) {
+ rte_event_ring_free(ring);
+ return -ENOMEM;
+ }
+ queue->reorder_drain_idx = 0;
+ rte_spinlock_init(&queue->reorder_lock);
+ } else {
+ queue->reorder_buffer = NULL;
+ }
+
return 0;
}
@@ -337,6 +379,7 @@ slb_info_get(struct rte_eventdev *dev __rte_unused,
.max_profiles_per_port = 1,
.event_dev_cap = RTE_EVENT_DEV_CAP_BURST_MODE |
RTE_EVENT_DEV_CAP_PARALLEL |
+ RTE_EVENT_DEV_CAP_ORDERED |
RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED |
RTE_EVENT_DEV_CAP_IMPLICIT_RELEASE_DISABLE |
RTE_EVENT_DEV_CAP_NONSEQ_MODE |
diff --git a/drivers/event/slb/slb_evdev.h b/drivers/event/slb/slb_evdev.h
index d7be6746ef..b73127a739 100644
--- a/drivers/event/slb/slb_evdev.h
+++ b/drivers/event/slb/slb_evdev.h
@@ -9,6 +9,7 @@
#include <rte_bitset.h>
#include <rte_event_ring.h>
#include <rte_eventdev.h>
+#include <rte_spinlock.h>
#include <rte_stdatomic.h>
#include <eventdev_pmd.h>
@@ -58,9 +59,37 @@ struct slb_port_q_entry {
uint8_t pad[6];
};
+#define SLB_PORT_HISTORY_SIZE 256
+#define SLB_PORT_HISTORY_MAX_TOKENS (SLB_PORT_HISTORY_SIZE - 1)
+#define SLB_PORT_HISTORY_FREE_END 0xFFFF
+
+struct slb_history_entry {
+ uint32_t flow_id;
+ uint16_t seq_slot;
+ uint16_t next_free;
+ uint8_t orig_queue_id;
+ uint8_t enq_port_id;
+ uint8_t sched_type;
+ bool active;
+};
+
+enum slb_reorder_state {
+ SLB_REORDER_EMPTY = 0,
+ SLB_REORDER_READY = 1,
+ SLB_REORDER_RELEASED = 2
+};
+
+struct slb_reorder_entry {
+ struct rte_event event;
+ RTE_ATOMIC(uint8_t) state;
+};
+
struct __rte_cache_aligned slb_port {
/* --- Cacheline 0: Hot Rx & Tx scheduling control (64 bytes) --- */
uint16_t id;
+ uint16_t history_head;
+ uint16_t history_avail;
+ uint16_t pending_releases;
int32_t inflight_credits;
int32_t new_event_threshold;
RTE_ATOMIC(uint64_t) active_queues;
@@ -78,6 +107,10 @@ struct __rte_cache_aligned slb_port {
alignas(RTE_CACHE_LINE_SIZE)
struct slb_port_q_entry q_entries[SLB_MAX_PRIORITIES][SLB_PORT_MAX_Q_MAP_PER_PRIO];
+ /* --- Cachelines 7+: Per-port history buffer --- */
+ alignas(RTE_CACHE_LINE_SIZE)
+ struct slb_history_entry history[SLB_PORT_HISTORY_SIZE];
+
/* --- Cold Section: Statistics counters and config --- */
alignas(RTE_CACHE_LINE_SIZE)
uint16_t dequeue_depth;
@@ -101,6 +134,10 @@ struct slb_queue {
uint16_t num_serving_ports;
uint64_t serving_ports_mask;
alignas(RTE_CACHE_LINE_SIZE) struct rte_event_ring *q_ring;
+ /* For ordered queues (dynamically allocated) */
+ struct slb_reorder_entry *reorder_buffer;
+ uint32_t reorder_drain_idx;
+ rte_spinlock_t reorder_lock;
};
/* Limited by the size of 'serving_ports_mask' and 'active_queues' bitmasks */
@@ -135,6 +172,8 @@ uint16_t slb_event_dequeue_burst(void *port, struct rte_event *events,
void slb_port_return_credits(struct slb_evdev *slb, struct slb_port *port,
int32_t credits);
+void slb_reorder_drain(struct slb_evdev *slb, struct slb_queue *queue);
+
void slb_dump(struct rte_eventdev *dev, FILE *f);
static inline struct slb_evdev *
diff --git a/drivers/event/slb/slb_event.c b/drivers/event/slb/slb_event.c
index f55262154e..b6c7a8a514 100644
--- a/drivers/event/slb/slb_event.c
+++ b/drivers/event/slb/slb_event.c
@@ -3,6 +3,7 @@
*/
#include <stdbool.h>
+#include <rte_bitops.h>
#include <rte_cycles.h>
#include <rte_ring_elem.h>
#include <rte_stdatomic.h>
@@ -81,6 +82,12 @@ slb_port_enqueue_stats(struct slb_port *port, uint16_t num_new,
port->release_enqueued += num_release;
}
+static void
+slb_port_queue_enqueue_stats(struct slb_port *source_port, uint8_t queue_id)
+{
+ source_port->queue_enqueued[queue_id]++;
+}
+
static void
slb_port_queue_enqueue_burst_stats(struct slb_port *source_port,
uint8_t queue_id, uint16_t num)
@@ -88,6 +95,67 @@ slb_port_queue_enqueue_burst_stats(struct slb_port *source_port,
source_port->queue_enqueued[queue_id] += num;
}
+#define REORDER_BURST_SZ 64
+
+/* Drain in-order contiguous ready events from ordered queue's reorder buffer */
+void
+slb_reorder_drain(struct slb_evdev *slb, struct slb_queue *queue)
+{
+ if (queue == NULL || queue->reorder_buffer == NULL)
+ return;
+
+ /* Only one thread can drain the reorder buffer at a time */
+ if (!rte_spinlock_trylock(&queue->reorder_lock))
+ return;
+
+ uint32_t drain_idx = queue->reorder_drain_idx;
+ uint32_t mask = queue->q_ring->r.mask;
+ struct slb_reorder_entry *buf = queue->reorder_buffer;
+ uint8_t state;
+
+ /* Drain from current drain index till hitting an empty slot.
+ * Released slots need no processing except marking them empty.
+ */
+ while ((state = rte_atomic_load_explicit(&buf[drain_idx].state,
+ rte_memory_order_acquire)) !=
+ SLB_REORDER_EMPTY) {
+ if (state == SLB_REORDER_RELEASED) {
+ rte_atomic_store_explicit(&buf[drain_idx].state,
+ SLB_REORDER_EMPTY,
+ rte_memory_order_release);
+ drain_idx = (drain_idx + 1) & mask;
+ continue;
+ }
+
+ /* Gather contiguous READY events destined for same target_qid */
+ struct rte_event burst[REORDER_BURST_SZ];
+ uint16_t burst_size = 0;
+ uint8_t target_qid = buf[drain_idx].event.queue_id;
+
+ while (burst_size < REORDER_BURST_SZ &&
+ rte_atomic_load_explicit(&buf[drain_idx].state,
+ rte_memory_order_acquire) ==
+ SLB_REORDER_READY &&
+ buf[drain_idx].event.queue_id == target_qid) {
+ burst[burst_size++] = buf[drain_idx].event;
+ rte_atomic_store_explicit(&buf[drain_idx].state,
+ SLB_REORDER_EMPTY,
+ rte_memory_order_release);
+ drain_idx = (drain_idx + 1) & mask;
+ }
+
+ if (burst_size == 0)
+ break;
+
+ /* Target queue is sized to max_inflight and can never be full */
+ rte_event_ring_enqueue_burst(slb->queues[target_qid].q_ring,
+ burst, burst_size, NULL);
+ }
+
+ queue->reorder_drain_idx = drain_idx;
+ rte_spinlock_unlock(&queue->reorder_lock);
+}
+
static inline uint16_t
slb_event_enqueue_burst_generic(struct slb_port *source_port,
const struct rte_event events[],
@@ -96,8 +164,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
uint16_t num_release)
{
struct slb_evdev *slb = source_port->slb;
- uint16_t enqueued = 0;
- uint16_t idx = 0;
+ uint16_t cur_pending;
if (unlikely(events_len == 0))
return 0;
@@ -133,49 +200,233 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
if (num_release > 0)
slb_port_return_credits(slb, source_port, num_release);
+ cur_pending = source_port->pending_releases;
+ RTE_VERIFY(num_forward + num_release <= cur_pending);
+
slb_port_enqueue_stats(source_port, num_new, num_forward, num_release);
- /* Enqueue contiguous chunks targeting the same queue */
- while (idx < events_len) {
- if (events[idx].op == RTE_EVENT_OP_RELEASE) {
- idx++;
- enqueued++;
- continue;
+ /* Unified zero-copy fast path: when port has no in-flight ordered events
+ * and no release ops.
+ */
+ if (likely(source_port->history_avail == SLB_PORT_HISTORY_MAX_TOKENS &&
+ num_release == 0)) {
+ uint8_t q0 = events[0].queue_id;
+ uint16_t burst_size = 1;
+
+ while (burst_size < events_len &&
+ events[burst_size].queue_id == q0)
+ burst_size++;
+
+ if (likely(burst_size == events_len)) {
+ rte_event_ring_enqueue_burst(slb->queues[q0].q_ring,
+ events, events_len, NULL);
+ slb_port_queue_enqueue_burst_stats(source_port, q0,
+ events_len);
+ } else {
+ uint8_t dest_queues[SLB_MAX_PORT_ENQUEUE_DEPTH];
+ uint16_t queue_counts[SLB_MAX_QUEUES];
+ uint16_t insert_pos[SLB_MAX_QUEUES];
+ uint16_t queue_offsets[SLB_MAX_QUEUES];
+ struct rte_event sorted_events[SLB_MAX_PORT_ENQUEUE_DEPTH];
+ uint64_t batch_queues_mask = (1ULL << q0);
+ uint64_t qmask;
+ uint16_t current_offset = 0;
+
+ queue_counts[q0] = burst_size;
+ for (uint16_t k = 0; k < burst_size; k++)
+ dest_queues[k] = q0;
+
+ for (uint16_t k = burst_size; k < events_len; k++) {
+ uint8_t qid = events[k].queue_id;
+ uint64_t q_bit = (1ULL << qid);
+
+ dest_queues[k] = qid;
+ if (!(batch_queues_mask & q_bit)) {
+ batch_queues_mask |= q_bit;
+ queue_counts[qid] = 1;
+ } else {
+ queue_counts[qid]++;
+ }
+ }
+
+ qmask = batch_queues_mask;
+ while (qmask != 0) {
+ uint8_t q = rte_ctz64(qmask);
+
+ qmask &= (qmask - 1);
+ queue_offsets[q] = current_offset;
+ insert_pos[q] = current_offset;
+ current_offset += queue_counts[q];
+ }
+
+ for (uint16_t k = 0; k < events_len; k++) {
+ uint8_t q = dest_queues[k];
+
+ sorted_events[insert_pos[q]++] = events[k];
+ }
+
+ qmask = batch_queues_mask;
+ while (qmask != 0) {
+ uint8_t q = rte_ctz64(qmask);
+ uint16_t count = queue_counts[q];
+ uint16_t offset = queue_offsets[q];
+
+ qmask &= (qmask - 1);
+ rte_event_ring_enqueue_burst(slb->queues[q].q_ring,
+ &sorted_events[offset],
+ count, NULL);
+ slb_port_queue_enqueue_burst_stats(source_port, q,
+ count);
+ }
}
- uint8_t qid = events[idx].queue_id;
- uint16_t chunk_len = 1;
+ if (num_forward > 0)
+ source_port->pending_releases -= num_forward;
+
+ SLB_LOG_DP_PORT_LINE(DEBUG, source_port->id,
+ "%d events accepted.", events_len);
+ return events_len;
+ }
- while (idx + chunk_len < events_len &&
- events[idx + chunk_len].queue_id == qid &&
- events[idx + chunk_len].op != RTE_EVENT_OP_RELEASE) {
- chunk_len++;
+ /* General path for bursts with in-flight ordered events or releases */
+ uint64_t ordered_drain_mask = 0;
+ uint16_t freed_count = 0;
+ uint16_t head = source_port->history_head;
+ uint16_t n_enq = 0;
+ uint8_t dest_queues[SLB_MAX_PORT_ENQUEUE_DEPTH];
+ const struct rte_event *enq_events[SLB_MAX_PORT_ENQUEUE_DEPTH];
+ uint16_t queue_counts[SLB_MAX_QUEUES];
+ uint64_t batch_queues_mask = 0;
+
+ for (uint16_t idx = 0; idx < events_len; idx++) {
+ const struct rte_event *ev = &events[idx];
+ uint8_t h_idx = (ev->op != RTE_EVENT_OP_NEW) ? ev->impl_opaque : 0;
+ struct slb_history_entry *hist =
+ (h_idx != 0 && source_port->history[h_idx].active) ?
+ &source_port->history[h_idx] : NULL;
+
+ if (hist != NULL && hist->sched_type == RTE_SCHED_TYPE_ORDERED) {
+ struct slb_queue *orig_q = &slb->queues[hist->orig_queue_id];
+
+ if (orig_q->reorder_buffer) {
+ uint16_t slot = hist->seq_slot;
+
+ if (ev->op == RTE_EVENT_OP_RELEASE) {
+ rte_atomic_store_explicit(
+ &orig_q->reorder_buffer[slot].state,
+ SLB_REORDER_RELEASED,
+ rte_memory_order_release);
+ } else {
+ orig_q->reorder_buffer[slot].event = *ev;
+ rte_atomic_store_explicit(
+ &orig_q->reorder_buffer[slot].state,
+ SLB_REORDER_READY,
+ rte_memory_order_release);
+ }
+ ordered_drain_mask |= (1ULL << hist->orig_queue_id);
+ }
+ hist->active = false;
+ hist->next_free = head;
+ head = h_idx;
+ freed_count++;
+ slb_port_queue_enqueue_stats(source_port, ev->queue_id);
+ } else if (ev->op == RTE_EVENT_OP_RELEASE) {
+ if (hist != NULL) {
+ hist->active = false;
+ hist->next_free = head;
+ head = h_idx;
+ freed_count++;
+ }
+ slb_port_queue_enqueue_stats(source_port, ev->queue_id);
+ } else {
+ if (hist != NULL) {
+ hist->active = false;
+ hist->next_free = head;
+ head = h_idx;
+ freed_count++;
+ }
+
+ uint8_t qid = ev->queue_id;
+ uint64_t q_bit = (1ULL << qid);
+
+ dest_queues[n_enq] = qid;
+ enq_events[n_enq] = ev;
+ n_enq++;
+
+ if (!(batch_queues_mask & q_bit)) {
+ batch_queues_mask |= q_bit;
+ queue_counts[qid] = 1;
+ } else {
+ queue_counts[qid]++;
+ }
}
+ }
- struct slb_queue *queue = &slb->queues[qid];
- uint16_t actual_enq = rte_event_ring_enqueue_burst(queue->q_ring,
- &events[idx],
- chunk_len,
- NULL);
- slb_port_queue_enqueue_burst_stats(source_port, qid, actual_enq);
- enqueued += actual_enq;
-
- if (actual_enq < chunk_len) {
- /* If queue ring is full, return unused credits for dropped NEW events */
- uint16_t dropped_new = 0;
- for (uint16_t d = actual_enq; d < chunk_len; d++) {
- if (events[idx + d].op == RTE_EVENT_OP_NEW)
- dropped_new++;
+ if (n_enq > 0) {
+ if (likely((batch_queues_mask & (batch_queues_mask - 1)) == 0 &&
+ n_enq == events_len)) {
+ uint8_t q = rte_ctz64(batch_queues_mask);
+
+ rte_event_ring_enqueue_burst(slb->queues[q].q_ring,
+ events, events_len, NULL);
+ slb_port_queue_enqueue_burst_stats(source_port, q,
+ events_len);
+ } else {
+ uint16_t queue_offsets[SLB_MAX_QUEUES];
+ uint16_t insert_pos[SLB_MAX_QUEUES];
+ struct rte_event sorted_events[SLB_MAX_PORT_ENQUEUE_DEPTH];
+ uint16_t current_offset = 0;
+ uint64_t qmask = batch_queues_mask;
+
+ while (qmask != 0) {
+ uint8_t q = rte_ctz64(qmask);
+
+ qmask &= (qmask - 1);
+ queue_offsets[q] = current_offset;
+ insert_pos[q] = current_offset;
+ current_offset += queue_counts[q];
+ }
+
+ for (uint16_t k = 0; k < n_enq; k++) {
+ uint8_t q = dest_queues[k];
+
+ sorted_events[insert_pos[q]++] = *enq_events[k];
+ }
+
+ qmask = batch_queues_mask;
+ while (qmask != 0) {
+ uint8_t q = rte_ctz64(qmask);
+ uint16_t count = queue_counts[q];
+ uint16_t offset = queue_offsets[q];
+
+ qmask &= (qmask - 1);
+ rte_event_ring_enqueue_burst(slb->queues[q].q_ring,
+ &sorted_events[offset],
+ count, NULL);
+ slb_port_queue_enqueue_burst_stats(source_port, q,
+ count);
}
- if (dropped_new > 0)
- slb_port_return_credits(slb, source_port, dropped_new);
- break;
}
+ }
+
+ if (freed_count > 0) {
+ source_port->history_head = head;
+ source_port->history_avail += freed_count;
+ }
+
+ while (ordered_drain_mask != 0) {
+ uint8_t qid = rte_ctz64(ordered_drain_mask);
- idx += chunk_len;
+ ordered_drain_mask &= ~(1ULL << qid);
+ slb_reorder_drain(slb, &slb->queues[qid]);
}
- return enqueued;
+ if (num_forward + num_release > 0)
+ source_port->pending_releases -= (num_forward + num_release);
+
+ SLB_LOG_DP_PORT_LINE(DEBUG, source_port->id,
+ "%d events accepted.", events_len);
+ return events_len;
}
uint16_t
@@ -183,6 +434,7 @@ slb_event_enqueue_burst(void *port, const struct rte_event events[],
uint16_t events_len)
{
struct slb_port *source_port = port;
+
return slb_event_enqueue_burst_generic(source_port, events, events_len,
false, 0, 0, 0);
}
@@ -192,6 +444,7 @@ slb_event_enqueue_new_burst(void *port, const struct rte_event events[],
uint16_t events_len)
{
struct slb_port *source_port = port;
+
return slb_event_enqueue_burst_generic(source_port, events, events_len,
true, events_len, 0, 0);
}
@@ -201,65 +454,184 @@ slb_event_enqueue_forward_burst(void *port, const struct rte_event events[],
uint16_t events_len)
{
struct slb_port *source_port = port;
+
return slb_event_enqueue_burst_generic(source_port, events, events_len,
true, 0, events_len, 0);
}
+static inline uint16_t
+slb_ordered_ring_dequeue_burst(struct rte_event_ring *r,
+ struct rte_event *events,
+ uint16_t num, uint32_t *start_slot)
+{
+ uint32_t cons_head, cons_next;
+ uint32_t entries;
+ uint16_t n;
+
+ n = __rte_ring_move_cons_head(&r->r, 0, num,
+ RTE_RING_QUEUE_VARIABLE,
+ &cons_head, &cons_next, &entries);
+ if (n == 0)
+ return 0;
+
+ *start_slot = cons_head & r->r.mask;
+
+ __rte_ring_dequeue_elems(&r->r, cons_head, events,
+ sizeof(struct rte_event), n);
+
+ __rte_ring_update_tail(&r->r.cons, cons_head, cons_next, 0);
+
+ return n;
+}
+
static inline uint16_t
slb_port_dequeue(struct slb_port *port, struct rte_event events[],
uint16_t num)
{
- uint16_t total_deq = 0;
+ uint16_t dequeued = 0;
- /* Single-link direct fast path */
if (port->is_single_linked) {
- if (unlikely(!port->single_linked_ring))
+ if (unlikely(port->single_linked_ring == NULL))
return 0;
-
- total_deq = rte_event_ring_dequeue_burst(port->single_linked_ring,
+ dequeued = rte_event_ring_dequeue_burst(port->single_linked_ring,
events, num, NULL);
- if (total_deq > 0) {
- port->total_dequeues += total_deq;
- if (port->implicit_release)
- slb_port_return_credits(port->slb, port,
- total_deq);
- }
- return total_deq;
+ for (uint16_t j = 0; j < dequeued; j++)
+ events[j].impl_opaque = 0;
+ return dequeued;
}
- /* Parallel queue dequeue across priority levels */
- for (int p = 0; p < SLB_MAX_PRIORITIES; p++) {
- uint16_t num_q = port->num_q[p];
- if (num_q == 0)
+ uint64_t active_queues = rte_atomic_load_explicit(
+ &port->active_queues, rte_memory_order_acquire);
+ if (!active_queues)
+ return 0;
+
+ for (int pr = 0; pr < SLB_MAX_PRIORITIES; pr++) {
+ int slots = port->num_q[pr];
+
+ if (slots == 0)
continue;
- uint16_t q_idx = port->next_q[p];
- for (uint16_t i = 0; i < num_q; i++) {
- struct slb_port_q_entry *entry = &port->q_entries[p][q_idx];
- uint16_t req = num - total_deq;
- uint16_t deq = rte_event_ring_dequeue_burst(entry->dq_ring,
- &events[total_deq],
- req, NULL);
- if (deq > 0) {
- port->queue_dequeued[entry->queue_id] += deq;
- total_deq += deq;
+ int start = port->next_q[pr];
+
+ if (unlikely(start >= slots))
+ start = 0;
+
+ for (int s = 0; s < slots; s++) {
+ int idx = start + s;
+
+ if (idx >= slots)
+ idx -= slots;
+
+ struct slb_port_q_entry *qent = &port->q_entries[pr][idx];
+ uint8_t queue_id = qent->queue_id;
+
+ if (!((active_queues >> queue_id) & 1))
+ continue;
+
+ struct rte_event_ring *ring_to_dq = qent->dq_ring;
+
+ if (unlikely(ring_to_dq == NULL))
+ continue;
+
+ uint8_t sched_type = qent->schedule_type;
+ bool is_ordered = (sched_type == RTE_SCHED_TYPE_ORDERED);
+
+ if (is_ordered) {
+ if (unlikely(port->history_avail == 0))
+ continue;
+
+ uint16_t dequeue_num = RTE_MIN(num,
+ port->history_avail);
+ uint32_t start_slot = 0;
+
+ dequeued = slb_ordered_ring_dequeue_burst(
+ ring_to_dq, events, dequeue_num,
+ &start_slot);
+
+ if (dequeued > 0) {
+ uint16_t head = port->history_head;
+ uint32_t ring_mask = ring_to_dq->r.mask;
+
+ for (uint16_t j = 0; j < dequeued; j++) {
+ uint16_t h_idx = head;
+ struct slb_history_entry *h =
+ &port->history[h_idx];
+
+ head = h->next_free;
+ h->orig_queue_id = queue_id;
+ h->sched_type = sched_type;
+ h->seq_slot = (start_slot + j) &
+ ring_mask;
+ h->active = true;
+ events[j].impl_opaque =
+ (uint8_t)h_idx;
+ }
+ port->history_head = head;
+ port->history_avail -= dequeued;
+ }
+ } else {
+ dequeued = rte_event_ring_dequeue_burst(
+ ring_to_dq, events, num, NULL);
+ for (uint16_t j = 0; j < dequeued; j++)
+ events[j].impl_opaque = 0;
}
- q_idx++;
- if (q_idx >= num_q)
- q_idx = 0;
+ if (dequeued > 0) {
+ idx++;
+ port->next_q[pr] = (idx >= slots) ? 0 : idx;
+ return dequeued;
+ }
+ }
+ }
+ return 0;
+}
- if (total_deq == num)
- break;
+static inline void
+slb_port_process_implicit_release(struct slb_port *port, struct slb_evdev *slb,
+ uint16_t pending)
+{
+ uint16_t head;
+ uint64_t ordered_drain_mask = 0;
+
+ if (port->is_single_linked ||
+ port->history_avail >= SLB_PORT_HISTORY_MAX_TOKENS)
+ goto done;
+
+ head = port->history_head;
+ for (int i = 1; i < SLB_PORT_HISTORY_SIZE; i++) {
+ struct slb_history_entry *hist = &port->history[i];
+
+ if (!hist->active)
+ continue;
+
+ if (hist->sched_type == RTE_SCHED_TYPE_ORDERED) {
+ struct slb_queue *oq = &slb->queues[hist->orig_queue_id];
+
+ if (oq->reorder_buffer) {
+ rte_atomic_store_explicit(
+ &oq->reorder_buffer[hist->seq_slot].state,
+ SLB_REORDER_RELEASED,
+ rte_memory_order_release);
+ ordered_drain_mask |= (1ULL << hist->orig_queue_id);
+ }
}
- port->next_q[p] = q_idx;
+ hist->active = false;
+ hist->next_free = head;
+ head = (uint16_t)i;
+ port->history_avail++;
+ }
+ port->history_head = head;
- if (total_deq > 0)
- break;
+ while (ordered_drain_mask != 0) {
+ uint8_t qid = rte_ctz64(ordered_drain_mask);
+
+ ordered_drain_mask &= ~(1ULL << qid);
+ slb_reorder_drain(slb, &slb->queues[qid]);
}
- port->total_dequeues += total_deq;
- return total_deq;
+done:
+ slb_port_return_credits(slb, port, pending);
+ port->pending_releases = 0;
}
uint16_t
@@ -267,13 +639,24 @@ slb_event_dequeue_burst(void *p, struct rte_event events[], uint16_t num,
uint64_t wait __rte_unused)
{
struct slb_port *port = p;
+ struct slb_evdev *slb = port->slb;
+ uint16_t pending = port->pending_releases;
+ uint16_t n;
+
+ if (port->implicit_release && pending > 0)
+ slb_port_process_implicit_release(port, slb, pending);
+
+ if (unlikely(num > port->dequeue_depth))
+ num = port->dequeue_depth;
port->dequeue_calls++;
- uint16_t dequeued = slb_port_dequeue(port, events, num);
- port->dequeued += dequeued;
+ n = slb_port_dequeue(port, events, num);
+ if (n > 0)
+ port->pending_releases += n;
+ port->dequeued += n;
- if (dequeued == 0)
+ if (n == 0)
port->empty_dequeues++;
- return dequeued;
+ return n;
}
--
2.53.0
^ permalink raw reply related [flat|nested] 14+ messages in thread* [PATCH v1 3/8] event/slb: add support for atomic queues
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 1/8] event/slb: introduce software load balancer PMD Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 2/8] event/slb: add support for ordered queues Pravin Pathak
@ 2026-11-07 1:30 ` Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 4/8] event/slb: add dynamic flow migration " Pravin Pathak
` (5 subsequent siblings)
8 siblings, 0 replies; 14+ messages in thread
From: Pravin Pathak @ 2026-11-07 1:30 UTC (permalink / raw)
To: dev
Cc: jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
From: Pravin Pathak <pravin.pathak.dev@gmail.com>
Add support for RTE_SCHED_TYPE_ATOMIC queues to provide per-flow event
serialization across worker ports.
Flow affinity is maintained using a 2048-entry flow map that pins each
flow ID to a serving worker port upon first observation. Events for each
atomic queue are held in a main ring and scheduled into dedicated per-port
rings via slb_atomic_refill using counting sort grouping.
A hold buffer retains events when destination port rings are temporarily
full to preserve flow FIFO ordering. Worker ports dequeue atomic events
from their private rings and record active flows in port history to track
returns on release or forward operations.
Signed-off-by: Pravin Pathak <pravin.pathak.dev@gmail.com>
---
drivers/event/slb/slb_evdev.c | 90 +++++++++-
drivers/event/slb/slb_evdev.h | 37 ++++
drivers/event/slb/slb_event.c | 312 +++++++++++++++++++++++++++++++---
3 files changed, 411 insertions(+), 28 deletions(-)
diff --git a/drivers/event/slb/slb_evdev.c b/drivers/event/slb/slb_evdev.c
index 13be490636..be0b3630f6 100644
--- a/drivers/event/slb/slb_evdev.c
+++ b/drivers/event/slb/slb_evdev.c
@@ -121,6 +121,10 @@ slb_queue_release(struct rte_eventdev *dev, uint8_t queue_id)
struct slb_evdev *slb = slb_pmd_priv(dev);
struct slb_queue *queue = &slb->queues[queue_id];
+ if (queue->atomic_state) {
+ rte_free(queue->atomic_state);
+ queue->atomic_state = NULL;
+ }
if (queue->reorder_buffer) {
rte_free(queue->reorder_buffer);
queue->reorder_buffer = NULL;
@@ -129,6 +133,12 @@ slb_queue_release(struct rte_eventdev *dev, uint8_t queue_id)
rte_event_ring_free(queue->q_ring);
queue->q_ring = NULL;
}
+ for (int i = 0; i < SLB_MAX_PORTS; i++) {
+ if (queue->per_port_rings[i]) {
+ rte_event_ring_free(queue->per_port_rings[i]);
+ queue->per_port_rings[i] = NULL;
+ }
+ }
}
/* Setup Eventdev queue. Eventdev queues are implemented as DPDK rings */
@@ -154,9 +164,14 @@ slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
queue->is_single_link = true;
ring = rte_event_ring_create(ring_name, q_size,
dev->data->socket_id, RING_F_SC_DEQ | RING_F_EXACT_SZ);
+ } else if (conf->schedule_type == RTE_SCHED_TYPE_ATOMIC) {
+ /* MP -> SC ring for atomic queues main ring */
+ queue->is_single_link = false;
+ ring = rte_event_ring_create(ring_name, q_size,
+ dev->data->socket_id, RING_F_SC_DEQ | RING_F_EXACT_SZ);
} else if (conf->schedule_type == RTE_SCHED_TYPE_PARALLEL ||
conf->schedule_type == RTE_SCHED_TYPE_ORDERED) {
- /* MP -> MC ring for parallel queues */
+ /* MP -> MC ring for parallel or ordered queues */
queue->is_single_link = false;
ring = rte_event_ring_create(ring_name, q_size,
dev->data->socket_id, RING_F_EXACT_SZ);
@@ -174,6 +189,31 @@ slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
queue->serving_ports_mask = 0;
queue->num_serving_ports = 0;
+ /* Atomic queue allocation */
+ if (conf->schedule_type == RTE_SCHED_TYPE_ATOMIC) {
+ for (int i = 0; i < SLB_MAX_PORTS; i++)
+ queue->per_port_rings[i] = NULL;
+
+ queue->atomic_state = rte_zmalloc_socket("slb_atomic_state",
+ sizeof(struct slb_atomic_state),
+ RTE_CACHE_LINE_SIZE, dev->data->socket_id);
+ if (queue->atomic_state == NULL) {
+ rte_event_ring_free(ring);
+ return -ENOMEM;
+ }
+ for (int i = 0; i < SLB_ATOMIC_FLOWS; i++)
+ rte_atomic_store_explicit(
+ &queue->atomic_state->flow_map.flows[i].port_id,
+ -1, rte_memory_order_relaxed);
+ queue->atomic_state->next_port_rr = 0;
+ rte_spinlock_init(&queue->atomic_state->refill_lock);
+ queue->atomic_state->hold_count = 0;
+ memset(queue->atomic_state->port_stats, 0,
+ sizeof(queue->atomic_state->port_stats));
+ } else {
+ queue->atomic_state = NULL;
+ }
+
/* Ordered queue reorder buffer allocation */
if (conf->schedule_type == RTE_SCHED_TYPE_ORDERED) {
queue->reorder_buffer = rte_zmalloc_socket("slb_reorder_buf",
@@ -244,6 +284,26 @@ slb_port_link_one(struct slb_port *port, struct slb_queue *queue,
port->single_linked_q = queue;
port->single_linked_ring = queue->q_ring;
} else {
+ if (queue->schedule_type == RTE_SCHED_TYPE_ATOMIC &&
+ queue->per_port_rings[port->id] == NULL) {
+ char ring_name[RTE_RING_NAMESIZE];
+ struct rte_event_ring *ring;
+
+ snprintf(ring_name, sizeof(ring_name),
+ "slb%d_q%u_p%u", port->slb->data->dev_id,
+ qid, port->id);
+ ring = rte_event_ring_create(
+ ring_name,
+ SLB_MAX_PORT_DEQUEUE_DEPTH * 4,
+ port->slb->data->socket_id,
+ RING_F_SP_ENQ | RING_F_SC_DEQ);
+ if (ring == NULL) {
+ rte_errno = ENOMEM;
+ return -ENOMEM;
+ }
+ queue->per_port_rings[port->id] = ring;
+ }
+
if (has_prio)
prio = (app_prio < RTE_EVENT_DEV_PRIORITY_NORMAL) ? 0 : 1;
else
@@ -262,12 +322,15 @@ slb_port_link_one(struct slb_port *port, struct slb_queue *queue,
}
if (!found) {
uint16_t slot = port->num_q[prio];
+ struct rte_event_ring *dq_ring =
+ (queue->schedule_type == RTE_SCHED_TYPE_ATOMIC) ?
+ queue->per_port_rings[port->id] : queue->q_ring;
if (slot >= SLB_PORT_MAX_Q_MAP_PER_PRIO) {
rte_errno = ENOSPC;
return -ENOSPC;
}
- port->q_entries[prio][slot].dq_ring = queue->q_ring;
+ port->q_entries[prio][slot].dq_ring = dq_ring;
port->q_entries[prio][slot].queue = queue;
port->q_entries[prio][slot].queue_id = qid;
port->q_entries[prio][slot].schedule_type =
@@ -311,6 +374,13 @@ slb_port_unlink_one(struct slb_port *port, struct slb_queue *queue, uint8_t qid)
rte_errno = EINVAL;
return -EINVAL;
}
+
+ if (queue->schedule_type == RTE_SCHED_TYPE_ATOMIC &&
+ queue->per_port_rings[port->id] != NULL) {
+ rte_event_ring_free(queue->per_port_rings[port->id]);
+ queue->per_port_rings[port->id] = NULL;
+ }
+
rte_atomic_fetch_and_explicit(&port->active_queues,
~(1ULL << qid),
rte_memory_order_release);
@@ -379,6 +449,7 @@ slb_info_get(struct rte_eventdev *dev __rte_unused,
.max_profiles_per_port = 1,
.event_dev_cap = RTE_EVENT_DEV_CAP_BURST_MODE |
RTE_EVENT_DEV_CAP_PARALLEL |
+ RTE_EVENT_DEV_CAP_ATOMIC |
RTE_EVENT_DEV_CAP_ORDERED |
RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED |
RTE_EVENT_DEV_CAP_IMPLICIT_RELEASE_DISABLE |
@@ -474,6 +545,21 @@ slb_stop(struct rte_eventdev *dev)
for (uint8_t qid = 0; qid < slb->num_queues; qid++) {
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++)
+ rte_atomic_store_explicit(
+ &q->atomic_state->flow_map.flows[f].port_id,
+ -1, rte_memory_order_relaxed);
+
+ for (uint16_t pid = 0; pid < slb->num_ports; pid++) {
+ if (q->per_port_rings[pid] != NULL)
+ slb_drain_ring(dev_id,
+ q->per_port_rings[pid],
+ flush, flush_arg);
+ }
+ }
+
if (q->q_ring != NULL)
slb_drain_ring(dev_id, q->q_ring, flush, flush_arg);
}
diff --git a/drivers/event/slb/slb_evdev.h b/drivers/event/slb/slb_evdev.h
index b73127a739..7a70d5c94c 100644
--- a/drivers/event/slb/slb_evdev.h
+++ b/drivers/event/slb/slb_evdev.h
@@ -84,6 +84,39 @@ struct slb_reorder_entry {
RTE_ATOMIC(uint8_t) state;
};
+#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)
+
+/* Per-flow entry for atomic scheduling affinity */
+struct slb_flow {
+ RTE_ATOMIC(int32_t) port_id; /* Port this flow is pinned to */
+};
+
+/* Flow-to-port affinity mapping table */
+struct slb_flow_map {
+ struct slb_flow flows[SLB_ATOMIC_FLOWS];
+};
+
+/* Per-port counters for an atomic queue */
+struct __rte_cache_aligned slb_atomic_port_stats {
+ 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 */
+};
+
+/* 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;
+ alignas(RTE_CACHE_LINE_SIZE)
+ struct slb_atomic_port_stats port_stats[SLB_MAX_PORTS];
+ uint8_t hold_ports[SLB_HOLD_BUFFER_SIZE];
+ struct rte_event hold_buffer[SLB_HOLD_BUFFER_SIZE];
+};
+
struct __rte_cache_aligned slb_port {
/* --- Cacheline 0: Hot Rx & Tx scheduling control (64 bytes) --- */
uint16_t id;
@@ -134,6 +167,10 @@ struct slb_queue {
uint16_t num_serving_ports;
uint64_t serving_ports_mask;
alignas(RTE_CACHE_LINE_SIZE) struct rte_event_ring *q_ring;
+ /* Per-port rings for atomic queues */
+ struct rte_event_ring *per_port_rings[SLB_MAX_PORTS];
+ /* For atomic queues (dynamically allocated) */
+ struct slb_atomic_state *atomic_state;
/* For ordered queues (dynamically allocated) */
struct slb_reorder_entry *reorder_buffer;
uint32_t reorder_drain_idx;
diff --git a/drivers/event/slb/slb_event.c b/drivers/event/slb/slb_event.c
index b6c7a8a514..22f40d72d4 100644
--- a/drivers/event/slb/slb_event.c
+++ b/drivers/event/slb/slb_event.c
@@ -156,6 +156,216 @@ slb_reorder_drain(struct slb_evdev *slb, struct slb_queue *queue)
rte_spinlock_unlock(&queue->reorder_lock);
}
+/* Hash flow ID to atomic flow slot index */
+static inline uint32_t
+slb_flow_hash(uint32_t flow_id)
+{
+ return (flow_id * 2654435761U) & SLB_ATOMIC_FLOWS_MASK;
+}
+
+/* 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)
+{
+ struct rte_event master_events[SLB_MAX_PORT_DEQUEUE_DEPTH];
+ uint16_t n_deq;
+ uint16_t i;
+ uint64_t full_ports_mask = 0;
+ struct slb_atomic_state *astate = queue->atomic_state;
+ uint64_t serving_mask;
+ uint16_t num_serving_ports;
+
+ if (unlikely(astate == NULL))
+ return;
+
+ if (!rte_spinlock_trylock(&astate->refill_lock))
+ return;
+
+ /* 1. Process Hold Buffer first */
+ if (astate->hold_count > 0) {
+ uint16_t retained = 0;
+ uint16_t count = astate->hold_count;
+
+ i = 0;
+ while (i < count) {
+ uint8_t dest_port_id = astate->hold_ports[i];
+ uint16_t burst_len = 1;
+
+ while ((i + burst_len) < count &&
+ astate->hold_ports[i + burst_len] == dest_port_id)
+ burst_len++;
+
+ if (full_ports_mask & (1ULL << dest_port_id)) {
+ if (retained != i) {
+ memmove(&astate->hold_buffer[retained],
+ &astate->hold_buffer[i],
+ burst_len * sizeof(struct rte_event));
+ memmove(&astate->hold_ports[retained],
+ &astate->hold_ports[i],
+ burst_len);
+ }
+ retained += burst_len;
+ i += burst_len;
+ continue;
+ }
+
+ uint16_t enqueued = rte_event_ring_enqueue_burst(
+ queue->per_port_rings[dest_port_id],
+ &astate->hold_buffer[i], burst_len, NULL);
+ astate->port_stats[dest_port_id].q_enqueued += enqueued;
+
+ if (enqueued < burst_len) {
+ uint16_t rem = burst_len - enqueued;
+ uint16_t start_idx = i + enqueued;
+
+ full_ports_mask |= (1ULL << dest_port_id);
+ if (retained != start_idx) {
+ memmove(&astate->hold_buffer[retained],
+ &astate->hold_buffer[start_idx],
+ rem * sizeof(struct rte_event));
+ memmove(&astate->hold_ports[retained],
+ &astate->hold_ports[start_idx],
+ rem);
+ }
+ retained += rem;
+ }
+
+ i += burst_len;
+ }
+
+ astate->hold_count = retained;
+ if (astate->hold_count > 0) {
+ rte_spinlock_unlock(&astate->refill_lock);
+ return;
+ }
+ }
+
+ /* 2. Dequeue events from main atomic queue in batches */
+ serving_mask = queue->serving_ports_mask;
+ num_serving_ports = queue->num_serving_ports;
+
+ if (num_serving_ports == 0) {
+ rte_spinlock_unlock(&astate->refill_lock);
+ return;
+ }
+
+ for (int iter = 0; iter < 2; iter++) {
+ n_deq = rte_event_ring_dequeue_burst(
+ queue->q_ring, master_events, RTE_DIM(master_events), NULL);
+
+ if (n_deq == 0)
+ break;
+
+ uint8_t dest_ports[SLB_MAX_PORT_DEQUEUE_DEPTH];
+ uint16_t port_counts[SLB_MAX_PORTS];
+ uint16_t insert_pos[SLB_MAX_PORTS];
+ uint16_t port_offsets[SLB_MAX_PORTS];
+ struct rte_event sorted_events[SLB_MAX_PORT_DEQUEUE_DEPTH];
+ uint64_t batch_ports_mask = 0;
+
+ for (i = 0; i < n_deq; i++) {
+ struct rte_event *ev = &master_events[i];
+ uint32_t flow_idx = slb_flow_hash(ev->flow_id);
+ int32_t dest_port_id =
+ rte_atomic_load_explicit(
+ &astate->flow_map.flows[flow_idx].port_id,
+ 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;
+ rte_atomic_store_explicit(
+ &astate->flow_map.flows[flow_idx].port_id,
+ dest_port_id, rte_memory_order_release);
+ astate->port_stats[dest_port_id].flow_count++;
+ }
+
+ dest_ports[i] = (uint8_t)dest_port_id;
+ uint64_t p_bit = (1ULL << dest_port_id);
+
+ if (!(batch_ports_mask & p_bit)) {
+ batch_ports_mask |= p_bit;
+ port_counts[dest_port_id] = 1;
+ } else {
+ port_counts[dest_port_id]++;
+ }
+ }
+
+ uint16_t current_offset = 0;
+ uint64_t pmask = batch_ports_mask;
+
+ while (pmask != 0) {
+ uint8_t p = rte_ctz64(pmask);
+
+ pmask &= (pmask - 1);
+ port_offsets[p] = current_offset;
+ insert_pos[p] = current_offset;
+ current_offset += port_counts[p];
+ }
+
+ for (i = 0; i < n_deq; i++) {
+ uint8_t p = dest_ports[i];
+
+ sorted_events[insert_pos[p]++] = master_events[i];
+ }
+
+ pmask = batch_ports_mask;
+ while (pmask != 0) {
+ uint8_t p = rte_ctz64(pmask);
+ uint16_t p_count = port_counts[p];
+ uint16_t p_start = port_offsets[p];
+
+ pmask &= (pmask - 1);
+
+ if (full_ports_mask & (1ULL << p)) {
+ uint16_t space = SLB_HOLD_BUFFER_SIZE - astate->hold_count;
+ uint16_t to_copy = RTE_MIN(p_count, space);
+
+ if (to_copy > 0) {
+ rte_memcpy(&astate->hold_buffer[astate->hold_count],
+ &sorted_events[p_start],
+ to_copy * sizeof(struct rte_event));
+ memset(&astate->hold_ports[astate->hold_count], p, to_copy);
+ astate->hold_count += to_copy;
+ }
+ continue;
+ }
+
+ uint16_t enqueued = rte_event_ring_enqueue_burst(
+ queue->per_port_rings[p], &sorted_events[p_start], p_count, NULL);
+ astate->port_stats[p].q_enqueued += enqueued;
+
+ if (enqueued < p_count) {
+ uint16_t rem = p_count - enqueued;
+ uint16_t space = SLB_HOLD_BUFFER_SIZE - astate->hold_count;
+ uint16_t to_copy = RTE_MIN(rem, space);
+
+ full_ports_mask |= (1ULL << p);
+ if (to_copy > 0) {
+ rte_memcpy(&astate->hold_buffer[astate->hold_count],
+ &sorted_events[p_start + enqueued],
+ to_copy * sizeof(struct rte_event));
+ memset(&astate->hold_ports[astate->hold_count], p, to_copy);
+ astate->hold_count += to_copy;
+ }
+ }
+ }
+ }
+
+ rte_spinlock_unlock(&astate->refill_lock);
+}
+
static inline uint16_t
slb_event_enqueue_burst_generic(struct slb_port *source_port,
const struct rte_event events[],
@@ -205,7 +415,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
slb_port_enqueue_stats(source_port, num_new, num_forward, num_release);
- /* Unified zero-copy fast path: when port has no in-flight ordered events
+ /* Unified zero-copy fast path: when port has no in-flight ordered/atomic events
* and no release ops.
*/
if (likely(source_port->history_avail == SLB_PORT_HISTORY_MAX_TOKENS &&
@@ -288,7 +498,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
return events_len;
}
- /* General path for bursts with in-flight ordered events or releases */
+ /* General path for bursts with in-flight ordered/atomic events or releases */
uint64_t ordered_drain_mask = 0;
uint16_t freed_count = 0;
uint16_t head = source_port->history_head;
@@ -332,6 +542,14 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
slb_port_queue_enqueue_stats(source_port, ev->queue_id);
} else if (ev->op == RTE_EVENT_OP_RELEASE) {
if (hist != NULL) {
+ if (hist->sched_type == RTE_SCHED_TYPE_ATOMIC) {
+ struct slb_queue *orig_q =
+ &slb->queues[hist->orig_queue_id];
+ if (orig_q->atomic_state) {
+ uint8_t ep = hist->enq_port_id;
+ orig_q->atomic_state->port_stats[ep].q_returned++;
+ }
+ }
hist->active = false;
hist->next_free = head;
head = h_idx;
@@ -340,6 +558,14 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
slb_port_queue_enqueue_stats(source_port, ev->queue_id);
} else {
if (hist != NULL) {
+ if (hist->sched_type == RTE_SCHED_TYPE_ATOMIC) {
+ struct slb_queue *orig_q =
+ &slb->queues[hist->orig_queue_id];
+ if (orig_q->atomic_state) {
+ uint8_t ep = hist->enq_port_id;
+ orig_q->atomic_state->port_stats[ep].q_returned++;
+ }
+ }
hist->active = false;
hist->next_free = head;
head = h_idx;
@@ -484,6 +710,35 @@ slb_ordered_ring_dequeue_burst(struct rte_event_ring *r,
return n;
}
+static inline void
+slb_port_record_dequeued_history(struct slb_port *port, struct rte_event events[],
+ uint16_t dequeued, uint8_t queue_id,
+ uint8_t sched_type, uint32_t start_slot,
+ uint32_t ring_mask)
+{
+ uint16_t head = port->history_head;
+ bool is_ordered = (sched_type == RTE_SCHED_TYPE_ORDERED);
+
+ for (uint16_t j = 0; j < dequeued; j++) {
+ uint16_t h_idx = head;
+ struct slb_history_entry *h = &port->history[h_idx];
+
+ head = h->next_free;
+ h->orig_queue_id = queue_id;
+ h->sched_type = sched_type;
+ if (is_ordered) {
+ h->seq_slot = (start_slot + j) & ring_mask;
+ } else {
+ h->enq_port_id = (uint8_t)port->id;
+ h->flow_id = events[j].flow_id;
+ }
+ h->active = true;
+ events[j].impl_opaque = (uint8_t)h_idx;
+ }
+ port->history_head = head;
+ port->history_avail -= dequeued;
+}
+
static inline uint16_t
slb_port_dequeue(struct slb_port *port, struct rte_event events[],
uint16_t num)
@@ -535,8 +790,9 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
uint8_t sched_type = qent->schedule_type;
bool is_ordered = (sched_type == RTE_SCHED_TYPE_ORDERED);
+ bool is_atomic = (sched_type == RTE_SCHED_TYPE_ATOMIC);
- if (is_ordered) {
+ if (is_ordered || is_atomic) {
if (unlikely(port->history_avail == 0))
continue;
@@ -544,31 +800,27 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
port->history_avail);
uint32_t start_slot = 0;
- dequeued = slb_ordered_ring_dequeue_burst(
- ring_to_dq, events, dequeue_num,
- &start_slot);
-
- if (dequeued > 0) {
- uint16_t head = port->history_head;
- uint32_t ring_mask = ring_to_dq->r.mask;
-
- for (uint16_t j = 0; j < dequeued; j++) {
- uint16_t h_idx = head;
- struct slb_history_entry *h =
- &port->history[h_idx];
-
- head = h->next_free;
- h->orig_queue_id = queue_id;
- h->sched_type = sched_type;
- h->seq_slot = (start_slot + j) &
- ring_mask;
- h->active = true;
- events[j].impl_opaque =
- (uint8_t)h_idx;
+ if (is_ordered) {
+ dequeued = slb_ordered_ring_dequeue_burst(
+ ring_to_dq, events, dequeue_num,
+ &start_slot);
+ } else {
+ dequeued = rte_event_ring_dequeue_burst(
+ ring_to_dq, events, dequeue_num,
+ NULL);
+ if (dequeued == 0) {
+ slb_atomic_refill(port, qent->queue);
+ dequeued = rte_event_ring_dequeue_burst(
+ ring_to_dq, events, dequeue_num,
+ NULL);
}
- port->history_head = head;
- port->history_avail -= dequeued;
}
+
+ if (dequeued > 0)
+ slb_port_record_dequeued_history(
+ port, events, dequeued, queue_id,
+ sched_type, start_slot,
+ ring_to_dq->r.mask);
} else {
dequeued = rte_event_ring_dequeue_burst(
ring_to_dq, events, num, NULL);
@@ -614,6 +866,14 @@ slb_port_process_implicit_release(struct slb_port *port, struct slb_evdev *slb,
rte_memory_order_release);
ordered_drain_mask |= (1ULL << hist->orig_queue_id);
}
+ } else if (hist->sched_type == RTE_SCHED_TYPE_ATOMIC) {
+ struct slb_queue *oq = &slb->queues[hist->orig_queue_id];
+
+ if (oq->atomic_state) {
+ uint8_t ep = hist->enq_port_id;
+
+ oq->atomic_state->port_stats[ep].q_returned++;
+ }
}
hist->active = false;
hist->next_free = head;
--
2.53.0
^ permalink raw reply related [flat|nested] 14+ messages in thread* [PATCH v1 4/8] event/slb: add dynamic flow migration for atomic queues
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
` (2 preceding siblings ...)
2026-11-07 1:30 ` [PATCH v1 3/8] event/slb: add support for atomic queues Pravin Pathak
@ 2026-11-07 1:30 ` Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 5/8] event/slb: add extended statistics and diagnostics Pravin Pathak
` (4 subsequent siblings)
8 siblings, 0 replies; 14+ messages in thread
From: Pravin Pathak @ 2026-11-07 1:30 UTC (permalink / raw)
To: dev
Cc: jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
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
^ permalink raw reply related [flat|nested] 14+ messages in thread* [PATCH v1 5/8] event/slb: add extended statistics and diagnostics
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
` (3 preceding siblings ...)
2026-11-07 1:30 ` [PATCH v1 4/8] event/slb: add dynamic flow migration " Pravin Pathak
@ 2026-11-07 1:30 ` Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 6/8] event/slb: add self-test and documentation Pravin Pathak
` (3 subsequent siblings)
8 siblings, 0 replies; 14+ messages in thread
From: Pravin Pathak @ 2026-11-07 1:30 UTC (permalink / raw)
To: dev
Cc: jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
From: Pravin Pathak <pravin.pathak.dev@gmail.com>
Implement the eventdev extended statistics (xstats) interface and
enhance PMD diagnostics in the dump callback.
Device and per-port statistics are exposed through xstats callbacks,
including enqueue/dequeue counts, credit loan counts, and migration
tallies. The slb_dump callback is extended to display detailed per-port
and per-queue internal state, atomic migration status, and queue depth.
Signed-off-by: Pravin Pathak <pravin.pathak.dev@gmail.com>
---
drivers/event/slb/meson.build | 5 +
drivers/event/slb/slb_evdev.c | 94 ++++++++-
drivers/event/slb/slb_evdev.h | 11 ++
drivers/event/slb/slb_xstats.c | 347 +++++++++++++++++++++++++++++++++
4 files changed, 450 insertions(+), 7 deletions(-)
create mode 100644 drivers/event/slb/slb_xstats.c
diff --git a/drivers/event/slb/meson.build b/drivers/event/slb/meson.build
index 9a1dff524f..d698a58f16 100644
--- a/drivers/event/slb/meson.build
+++ b/drivers/event/slb/meson.build
@@ -10,6 +10,11 @@ endif
sources = files(
'slb_evdev.c',
'slb_event.c',
+ 'slb_xstats.c',
)
deps += ['bus_vdev', 'ring']
+
+if cc.has_argument('-Wno-format-nonliteral')
+ cflags += '-Wno-format-nonliteral'
+endif
diff --git a/drivers/event/slb/slb_evdev.c b/drivers/event/slb/slb_evdev.c
index 1916b7faa4..04a2aba165 100644
--- a/drivers/event/slb/slb_evdev.c
+++ b/drivers/event/slb/slb_evdev.c
@@ -606,6 +606,8 @@ slb_dump(struct rte_eventdev *dev, FILE *f)
{
struct slb_evdev *slb = slb_pmd_priv(dev);
int32_t pool_on_loan, pool_avail;
+ uint64_t total_mig_in = 0, total_mig_out = 0;
+ int i;
if (dev == NULL || f == NULL)
return;
@@ -617,20 +619,41 @@ slb_dump(struct rte_eventdev *dev, FILE *f)
fprintf(f, "\n[SLB EVENTDEV DEBUG STATS]\n");
fprintf(f, " Device ID: %u, Driver: %s\n", dev->data->dev_id,
SLB_PMD_NAME);
- fprintf(f, " Credit Pool: %d on_loan / %d max_inflight (%d avail)\n",
+ fprintf(f, " Credit Pool: %d on_loan / %d max_inflight (%d available)\n",
pool_on_loan, slb->max_inflight, pool_avail);
+ fprintf(f, "\n PORT FLOW MIGRATION COUNTERS:\n");
+ for (i = 0; i < slb->num_ports; i++) {
+ struct slb_port *p = &slb->ports[i];
+
+ total_mig_in += p->migrations_in;
+ total_mig_out += p->migrations_out;
+ fprintf(f,
+ " Port %2u: Migrated-In: %6" PRIu64
+ " | Migrated-Out: %6" PRIu64 " | Net: %+5" PRId64 "\n",
+ i, p->migrations_in, p->migrations_out,
+ (int64_t)p->migrations_in - (int64_t)p->migrations_out);
+ }
+ fprintf(f, " Total Flow Migrations: IN=%" PRIu64 ", OUT=%" PRIu64 "\n",
+ total_mig_in, total_mig_out);
+
fprintf(f, "\n PORT STATISTICS & QUEUE LINKAGES:\n");
- for (int i = 0; i < slb->num_ports; i++) {
+ for (i = 0; i < slb->num_ports; i++) {
struct slb_port *p = &slb->ports[i];
uint64_t aq = rte_atomic_load_explicit(&p->active_queues,
rte_memory_order_acquire);
- fprintf(f, " Port %2u (credits=%3d, active=0x%08" PRIx64 "): ",
- i, p->inflight_credits, aq);
int linked_count = 0;
+
+ fprintf(f,
+ " Port %2u (credits=%3d, active=0x%08" PRIx64
+ ", loaded=%d, mig_in=%" PRIu64 ", mig_out=%" PRIu64 "): ",
+ i, p->inflight_credits, aq, p->is_loaded,
+ p->migrations_in, p->migrations_out);
+
if (p->is_single_linked) {
if (p->single_linked_q) {
int qid = p->single_linked_q - slb->queues;
+
fprintf(f, "Q%d (single-link)", qid);
linked_count++;
}
@@ -638,6 +661,7 @@ slb_dump(struct rte_eventdev *dev, FILE *f)
for (int pr = 0; pr < SLB_MAX_PRIORITIES; pr++) {
for (int s = 0; s < p->num_q[pr]; s++) {
int qid = p->q_entries[pr][s].queue_id;
+
if (linked_count > 0)
fprintf(f, ", ");
fprintf(f, "Q%d(p%d)", qid, pr);
@@ -648,16 +672,69 @@ slb_dump(struct rte_eventdev *dev, FILE *f)
if (linked_count == 0)
fprintf(f, "(no queues linked)");
fprintf(f, "\n");
+ fprintf(f,
+ " stats: enq_calls=%" PRIu64
+ ", new_enq=%" PRIu64 ", fwd_enq=%" PRIu64
+ ", rel_enq=%" PRIu64 ", deq_calls=%" PRIu64
+ ", deq=%" PRIu64 "\n",
+ p->enqueue_calls, p->new_enqueued,
+ p->forward_enqueued, p->release_enqueued,
+ p->dequeue_calls, p->dequeued);
}
fprintf(f, "\n ALL QUEUES STATUS:\n");
- for (int i = 0; i < slb->num_queues; i++) {
+ for (i = 0; i < slb->num_queues; i++) {
struct slb_queue *q = &slb->queues[i];
uint32_t count = rte_event_ring_count(q->q_ring);
uint32_t ring_size = rte_event_ring_get_capacity(q->q_ring);
- fprintf(f, " Queue %2u: %6u/%6u events, serving_ports=0x%016" PRIx64 ", sched=%u, single=%d\n",
+
+ fprintf(f,
+ " Queue %2u: %6u/%6u events, serving_ports=0x%016" PRIx64
+ " (count=%u), sched_type=%u, single_link=%d\n",
i, count, ring_size, q->serving_ports_mask,
- q->schedule_type, q->is_single_link);
+ q->num_serving_ports, q->schedule_type,
+ q->is_single_link);
+ if (q->schedule_type == RTE_SCHED_TYPE_ATOMIC) {
+ if (q->atomic_state) {
+ fprintf(f,
+ " Atomic State: total_migrations=%" PRIu64
+ ", hold_count=%u\n",
+ q->atomic_state->migrations_count,
+ q->atomic_state->hold_count);
+ if (q->atomic_state->migration.in_progress) {
+ fprintf(f,
+ " MIGRATION IN PROGRESS: from port %u, flow_idx=%u\n",
+ q->atomic_state->migration.port_id,
+ q->atomic_state->migration.flow_idx);
+ }
+ for (int j = 0; j < slb->num_ports; j++) {
+ if (q->per_port_rings[j]) {
+ struct slb_atomic_port_stats *ps =
+ &q->atomic_state->port_stats[j];
+
+ fprintf(f,
+ " Per-port ring %d: %u events"
+ " (load_counter=%u, flows=%u,"
+ " enq=%" PRIu64 ", ret=%" PRIu64 ")\n",
+ j,
+ rte_event_ring_count(q->per_port_rings[j]),
+ ps->load_counter,
+ ps->flow_count,
+ ps->q_enqueued,
+ ps->q_returned);
+ }
+ }
+ }
+ } else if (q->schedule_type == RTE_SCHED_TYPE_ORDERED) {
+ fprintf(f, " Ordered reorder_drain_idx=%u\n",
+ q->reorder_drain_idx);
+ }
+ if (count > 0 && q->serving_ports_mask == 0)
+ fprintf(f,
+ " ^^^ PROBLEM: Queue has %u events but serving_ports_mask=0!\n",
+ count);
+ if (count == 0 && q->serving_ports_mask == 0)
+ fprintf(f, " (unused queue)\n");
}
fprintf(f, "[END DEBUG]\n\n");
fflush(f);
@@ -678,6 +755,9 @@ static struct eventdev_ops slb_evdev_ops = {
.dev_stop = slb_stop,
.dev_close = slb_close,
.dump = slb_dump,
+ .xstats_get = slb_xstats_get,
+ .xstats_get_names = slb_xstats_get_names,
+ .xstats_get_by_name = slb_xstats_get_by_name,
};
static int
diff --git a/drivers/event/slb/slb_evdev.h b/drivers/event/slb/slb_evdev.h
index b1a6de8606..17f3154efa 100644
--- a/drivers/event/slb/slb_evdev.h
+++ b/drivers/event/slb/slb_evdev.h
@@ -231,6 +231,17 @@ void slb_reorder_drain(struct slb_evdev *slb, struct slb_queue *queue);
void slb_dump(struct rte_eventdev *dev, FILE *f);
+int slb_xstats_get_names(const struct rte_eventdev *dev,
+ enum rte_event_dev_xstats_mode mode,
+ uint8_t queue_port_id,
+ struct rte_event_dev_xstats_name *xstats_names,
+ uint64_t *ids, unsigned int capacity);
+int slb_xstats_get(const struct rte_eventdev *dev,
+ enum rte_event_dev_xstats_mode mode, uint8_t queue_port_id,
+ const uint64_t ids[], uint64_t values[], unsigned int n);
+uint64_t slb_xstats_get_by_name(const struct rte_eventdev *dev,
+ const char *name, uint64_t *id);
+
static inline struct slb_evdev *
slb_pmd_priv(const struct rte_eventdev *eventdev)
{
diff --git a/drivers/event/slb/slb_xstats.c b/drivers/event/slb/slb_xstats.c
new file mode 100644
index 0000000000..fdb7dd6cc5
--- /dev/null
+++ b/drivers/event/slb/slb_xstats.c
@@ -0,0 +1,347 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2026 Pravin Pathak <pravin.pathak.dev@gmail.com>
+ */
+
+#include "slb_evdev.h"
+
+#include <limits.h>
+#include <stdbool.h>
+#include <string.h>
+
+#include <rte_debug.h>
+#include <rte_string_fns.h>
+
+/* The high bits in the xstats id is used to store an additional
+ * parameter (beyond the queue or port id already in the xstats
+ * interface).
+ */
+#define SLB_XSTATS_ID_PARAM_BITS (8)
+#define SLB_XSTATS_ID_STAT_BITS \
+ (sizeof(uint64_t) * CHAR_BIT - SLB_XSTATS_ID_PARAM_BITS)
+#define SLB_XSTATS_ID_STAT_MASK ((UINT64_C(1) << SLB_XSTATS_ID_STAT_BITS) - 1)
+
+#define SLB_XSTATS_ID_GET_PARAM(id) \
+ ((id) >> SLB_XSTATS_ID_STAT_BITS)
+
+#define SLB_XSTATS_ID_GET_STAT(id) \
+ ((id) & SLB_XSTATS_ID_STAT_MASK)
+
+#define SLB_XSTATS_ID_CREATE(id, param_value) \
+ ((((uint64_t)(param_value)) << SLB_XSTATS_ID_STAT_BITS) | (id))
+
+typedef uint64_t (*slb_xstats_dev_get_value_fn)(struct slb_evdev *slb);
+
+struct slb_xstat_dev {
+ const char *name;
+ slb_xstats_dev_get_value_fn get_value_fn;
+};
+
+typedef uint64_t (*slb_xstats_port_get_value_fn)(struct slb_evdev *slb,
+ uint8_t port_id,
+ uint8_t queue_id);
+
+struct slb_xstats_port {
+ const char *name_fmt;
+ slb_xstats_port_get_value_fn get_value_fn;
+ bool per_queue;
+};
+
+static uint64_t
+slb_xstats_dev_credits_on_loan(struct slb_evdev *slb)
+{
+ return rte_atomic_load_explicit(&slb->credits_on_loan,
+ rte_memory_order_relaxed);
+}
+
+static struct slb_xstat_dev slb_dev_xstats[] = {
+ { "dev_credits_on_loan", slb_xstats_dev_credits_on_loan }
+};
+
+#define SLB_GEN_PORT_ACCESS_FN(_variable) \
+ static uint64_t \
+ slb_xstats_port_get_ ## _variable(struct slb_evdev *slb, \
+ uint8_t port_id, \
+ uint8_t queue_id __rte_unused) \
+ { \
+ return slb->ports[port_id]._variable; \
+ }
+
+SLB_GEN_PORT_ACCESS_FN(new_enqueued)
+SLB_GEN_PORT_ACCESS_FN(forward_enqueued)
+SLB_GEN_PORT_ACCESS_FN(release_enqueued)
+
+static uint64_t
+slb_xstats_port_get_queue_enqueued(struct slb_evdev *slb, uint8_t port_id,
+ uint8_t queue_id)
+{
+ return slb->ports[port_id].queue_enqueued[queue_id];
+}
+
+SLB_GEN_PORT_ACCESS_FN(dequeued)
+
+static uint64_t
+slb_xstats_port_get_queue_dequeued(struct slb_evdev *slb, uint8_t port_id,
+ uint8_t queue_id)
+{
+ return slb->ports[port_id].queue_dequeued[queue_id];
+}
+
+SLB_GEN_PORT_ACCESS_FN(inflight_credits)
+SLB_GEN_PORT_ACCESS_FN(pending_releases)
+SLB_GEN_PORT_ACCESS_FN(migrations_in)
+SLB_GEN_PORT_ACCESS_FN(migrations_out)
+
+static struct slb_xstats_port slb_port_xstats[] = {
+ { "port_%u_new_enqueued", slb_xstats_port_get_new_enqueued,
+ false },
+ { "port_%u_forward_enqueued", slb_xstats_port_get_forward_enqueued,
+ false },
+ { "port_%u_release_enqueued", slb_xstats_port_get_release_enqueued,
+ false },
+ { "port_%u_queue_%u_enqueued", slb_xstats_port_get_queue_enqueued,
+ true },
+ { "port_%u_dequeued", slb_xstats_port_get_dequeued,
+ false },
+ { "port_%u_queue_%u_dequeued", slb_xstats_port_get_queue_dequeued,
+ true },
+ { "port_%u_inflight_credits", slb_xstats_port_get_inflight_credits,
+ false },
+ { "port_%u_pending_releases", slb_xstats_port_get_pending_releases,
+ false },
+ { "port_%u_migrations_in", slb_xstats_port_get_migrations_in,
+ false },
+ { "port_%u_migrations_out", slb_xstats_port_get_migrations_out,
+ false },
+};
+
+typedef void (*slb_xstats_foreach_fn)(const char *xstats_name,
+ enum rte_event_dev_xstats_mode mode,
+ uint8_t queue_port_id,
+ uint64_t xstats_id,
+ void *data);
+
+static void
+slb_xstats_dev_foreach(slb_xstats_foreach_fn fn, void *fn_data)
+{
+ unsigned int i;
+
+ for (i = 0; i < RTE_DIM(slb_dev_xstats); i++)
+ fn(slb_dev_xstats[i].name, RTE_EVENT_DEV_XSTATS_DEVICE, 0,
+ i, fn_data);
+}
+
+static void
+slb_xstats_port_foreach(struct slb_evdev *slb, uint8_t port_id,
+ slb_xstats_foreach_fn fn, void *fn_data)
+{
+ uint8_t queue_id;
+ unsigned int stat_idx;
+
+ for (stat_idx = 0, queue_id = 0;
+ stat_idx < RTE_DIM(slb_port_xstats);) {
+ struct slb_xstats_port *xstat = &slb_port_xstats[stat_idx];
+ char xstats_name[RTE_EVENT_DEV_XSTATS_NAME_SIZE];
+ uint64_t xstats_id;
+
+ if (xstat->per_queue) {
+ xstats_id = SLB_XSTATS_ID_CREATE(stat_idx, queue_id);
+ snprintf(xstats_name, sizeof(xstats_name),
+ slb_port_xstats[stat_idx].name_fmt, port_id,
+ queue_id);
+ queue_id++;
+ } else {
+ xstats_id = stat_idx;
+ snprintf(xstats_name, sizeof(xstats_name),
+ slb_port_xstats[stat_idx].name_fmt, port_id);
+ }
+
+ fn(xstats_name, RTE_EVENT_DEV_XSTATS_PORT, port_id,
+ xstats_id, fn_data);
+
+ if (!(xstat->per_queue && queue_id < slb->num_queues)) {
+ stat_idx++;
+ queue_id = 0;
+ }
+ }
+}
+
+struct store_ctx {
+ struct rte_event_dev_xstats_name *names;
+ uint64_t *ids;
+ unsigned int count;
+ unsigned int capacity;
+};
+
+static void
+slb_xstats_store_stat(const char *xstats_name,
+ enum rte_event_dev_xstats_mode mode,
+ uint8_t queue_port_id, uint64_t xstats_id,
+ void *data)
+{
+ struct store_ctx *ctx = data;
+
+ RTE_SET_USED(mode);
+ RTE_SET_USED(queue_port_id);
+
+ if (ctx->count < ctx->capacity) {
+ if (ctx->names != NULL)
+ strlcpy(ctx->names[ctx->count].name, xstats_name,
+ RTE_EVENT_DEV_XSTATS_NAME_SIZE);
+ if (ctx->ids != NULL)
+ ctx->ids[ctx->count] = xstats_id;
+ }
+
+ ctx->count++;
+}
+
+int
+slb_xstats_get_names(const struct rte_eventdev *dev,
+ enum rte_event_dev_xstats_mode mode,
+ uint8_t queue_port_id,
+ struct rte_event_dev_xstats_name *xstats_names,
+ uint64_t *ids, unsigned int capacity)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ struct store_ctx ctx = {
+ .names = xstats_names,
+ .ids = ids,
+ .capacity = capacity
+ };
+
+ switch (mode) {
+ case RTE_EVENT_DEV_XSTATS_DEVICE:
+ slb_xstats_dev_foreach(slb_xstats_store_stat, &ctx);
+ return ctx.count;
+ case RTE_EVENT_DEV_XSTATS_PORT:
+ if (queue_port_id >= slb->num_ports)
+ return -EINVAL;
+ slb_xstats_port_foreach(slb, queue_port_id,
+ slb_xstats_store_stat, &ctx);
+ return ctx.count;
+ case RTE_EVENT_DEV_XSTATS_QUEUE:
+ return 0;
+ default:
+ return -EINVAL;
+ }
+}
+
+static int
+slb_xstats_dev_get(const struct rte_eventdev *dev,
+ const uint64_t ids[], uint64_t values[], unsigned int n)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ unsigned int i;
+
+ for (i = 0; i < n; i++) {
+ uint64_t id = ids[i];
+ struct slb_xstat_dev *xstat;
+
+ if (id >= RTE_DIM(slb_dev_xstats))
+ return -EINVAL;
+
+ xstat = &slb_dev_xstats[id];
+ values[i] = xstat->get_value_fn(slb);
+ }
+ return n;
+}
+
+static int
+slb_xstats_port_get(const struct rte_eventdev *dev, uint8_t port_id,
+ const uint64_t ids[], uint64_t values[], unsigned int n)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ unsigned int i;
+
+ for (i = 0; i < n; i++) {
+ uint64_t id = ids[i];
+ unsigned int stat_idx = SLB_XSTATS_ID_GET_STAT(id);
+ struct slb_xstats_port *xstat;
+ uint8_t queue_id = 0;
+
+ if (stat_idx >= RTE_DIM(slb_port_xstats))
+ return -EINVAL;
+
+ xstat = &slb_port_xstats[stat_idx];
+
+ if (xstat->per_queue) {
+ queue_id = SLB_XSTATS_ID_GET_PARAM(id);
+ if (queue_id >= slb->num_queues)
+ return -EINVAL;
+ }
+
+ values[i] = xstat->get_value_fn(slb, port_id, queue_id);
+ }
+ return n;
+}
+
+int
+slb_xstats_get(const struct rte_eventdev *dev,
+ enum rte_event_dev_xstats_mode mode, uint8_t queue_port_id,
+ const uint64_t ids[], uint64_t values[], unsigned int n)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+
+ switch (mode) {
+ case RTE_EVENT_DEV_XSTATS_DEVICE:
+ return slb_xstats_dev_get(dev, ids, values, n);
+ case RTE_EVENT_DEV_XSTATS_PORT:
+ if (queue_port_id >= slb->num_ports)
+ return -EINVAL;
+ return slb_xstats_port_get(dev, queue_port_id, ids, values, n);
+ case RTE_EVENT_DEV_XSTATS_QUEUE:
+ return 0;
+ default:
+ return -EINVAL;
+ }
+}
+
+struct find_ctx {
+ const struct rte_eventdev *dev;
+ const char *name;
+ uint64_t *id;
+ uint64_t value;
+};
+
+static void
+slb_xstats_find_stat(const char *xstats_name,
+ enum rte_event_dev_xstats_mode mode,
+ uint8_t queue_port_id, uint64_t xstats_id,
+ void *data)
+{
+ struct find_ctx *ctx = data;
+
+ if (strcmp(ctx->name, xstats_name) == 0) {
+ if (ctx->id != NULL)
+ *ctx->id = xstats_id;
+ slb_xstats_get(ctx->dev, mode, queue_port_id, &xstats_id,
+ &ctx->value, 1);
+ }
+}
+
+uint64_t
+slb_xstats_get_by_name(const struct rte_eventdev *dev, const char *name,
+ uint64_t *id)
+{
+ struct slb_evdev *slb = slb_pmd_priv(dev);
+ struct find_ctx ctx = {
+ .dev = dev,
+ .name = name,
+ .id = id,
+ .value = -EINVAL
+ };
+ uint16_t port_id;
+
+ if (id != NULL)
+ *id = (uint64_t)-1;
+
+ if (name == NULL)
+ return -EINVAL;
+
+ slb_xstats_dev_foreach(slb_xstats_find_stat, &ctx);
+
+ for (port_id = 0; port_id < slb->num_ports; port_id++)
+ slb_xstats_port_foreach(slb, port_id, slb_xstats_find_stat,
+ &ctx);
+
+ return ctx.value;
+}
--
2.53.0
^ permalink raw reply related [flat|nested] 14+ messages in thread* [PATCH v1 6/8] event/slb: add self-test and documentation
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
` (4 preceding siblings ...)
2026-11-07 1:30 ` [PATCH v1 5/8] event/slb: add extended statistics and diagnostics Pravin Pathak
@ 2026-11-07 1:30 ` Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 7/8] event/slb: update resets and credit defaults Pravin Pathak
` (2 subsequent siblings)
8 siblings, 0 replies; 14+ messages in thread
From: Pravin Pathak @ 2026-11-07 1:30 UTC (permalink / raw)
To: dev
Cc: jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
From: Pravin Pathak <pravin.pathak.dev@gmail.com>
Add driver self-test coverage and documentation for the Software
Load Balancer PMD.
The self-test verifies parallel queues, atomic queue flow ordering,
ordered queues with forward sequence restoration, single-link queues,
and device stop/restart event draining. Register the self-test in
app/test as a fast test.
Add PMD user guide, feature matrix table, release notes entry for
DPDK 26.11, and update MAINTAINERS.
Signed-off-by: Pravin Pathak <pravin.pathak.dev@gmail.com>
---
MAINTAINERS | 2 +
app/test/test_eventdev.c | 7 +
doc/guides/eventdevs/features/slb.ini | 26 ++
doc/guides/eventdevs/index.rst | 1 +
doc/guides/eventdevs/slb.rst | 68 ++++
drivers/event/slb/meson.build | 1 +
drivers/event/slb/slb_evdev.c | 1 +
drivers/event/slb/slb_evdev.h | 4 +
drivers/event/slb/slb_selftest.c | 522 ++++++++++++++++++++++++++
9 files changed, 632 insertions(+)
create mode 100644 doc/guides/eventdevs/features/slb.ini
create mode 100644 doc/guides/eventdevs/slb.rst
create mode 100644 drivers/event/slb/slb_selftest.c
diff --git a/MAINTAINERS b/MAINTAINERS
index 89ea53695a..bc46e3e8b8 100644
--- a/MAINTAINERS
+++ b/MAINTAINERS
@@ -1536,6 +1536,8 @@ F: doc/guides/eventdevs/dsw.rst
Software Load Balancer (SLB) Eventdev PMD
M: Pravin Pathak <pravin.pathak.dev@gmail.com>
F: drivers/event/slb/
+F: doc/guides/eventdevs/slb.rst
+F: doc/guides/eventdevs/features/slb.ini
Software OPDL Eventdev PMD
M: Liang Ma <liangma@liangbit.com>
diff --git a/app/test/test_eventdev.c b/app/test/test_eventdev.c
index 1241da38cb..58ec440674 100644
--- a/app/test/test_eventdev.c
+++ b/app/test/test_eventdev.c
@@ -1527,12 +1527,19 @@ test_eventdev_selftest_cn20k(void)
return test_eventdev_selftest_impl("event_cn20k", "");
}
+static int
+test_eventdev_selftest_slb(void)
+{
+ return test_eventdev_selftest_impl("event_slb", "");
+}
+
#endif /* !RTE_EXEC_ENV_WINDOWS */
REGISTER_FAST_TEST(eventdev_common_autotest, NOHUGE_OK, ASAN_OK, test_eventdev_common);
#ifndef RTE_EXEC_ENV_WINDOWS
REGISTER_FAST_TEST(eventdev_selftest_sw, NOHUGE_OK, ASAN_OK, test_eventdev_selftest_sw);
+REGISTER_FAST_TEST(eventdev_selftest_slb, NOHUGE_OK, ASAN_OK, test_eventdev_selftest_slb);
REGISTER_DRIVER_TEST(eventdev_selftest_octeontx, test_eventdev_selftest_octeontx);
REGISTER_DRIVER_TEST(eventdev_selftest_dpaa2, test_eventdev_selftest_dpaa2);
REGISTER_DRIVER_TEST(eventdev_selftest_dlb2, test_eventdev_selftest_dlb2);
diff --git a/doc/guides/eventdevs/features/slb.ini b/doc/guides/eventdevs/features/slb.ini
new file mode 100644
index 0000000000..0364840bc9
--- /dev/null
+++ b/doc/guides/eventdevs/features/slb.ini
@@ -0,0 +1,26 @@
+;
+; Supported features of the 'software load balancer' (event_slb) eventdev driver.
+;
+; Refer to default.ini for the full list of available PMD features.
+;
+[Scheduling Features]
+atomic_scheduling = Y
+ordered_scheduling = Y
+parallel_scheduling = Y
+distributed_sched = Y
+burst_mode = Y
+implicit_release_disable = Y
+nonseq_mode = Y
+runtime_port_link = Y
+multiple_queue_port = Y
+carry_flow_id = Y
+maintenance_free = Y
+independent_enq = Y
+
+[Eth Rx adapter Features]
+
+[Eth Tx adapter Features]
+
+[Crypto adapter Features]
+
+[Timer adapter Features]
diff --git a/doc/guides/eventdevs/index.rst b/doc/guides/eventdevs/index.rst
index 69a5ebf812..fdaf846503 100644
--- a/doc/guides/eventdevs/index.rst
+++ b/doc/guides/eventdevs/index.rst
@@ -20,3 +20,4 @@ application through the eventdev API.
sw
octeontx
opdl
+ slb
diff --git a/doc/guides/eventdevs/slb.rst b/doc/guides/eventdevs/slb.rst
new file mode 100644
index 0000000000..d319b94920
--- /dev/null
+++ b/doc/guides/eventdevs/slb.rst
@@ -0,0 +1,68 @@
+.. SPDX-License-Identifier: BSD-3-Clause
+ Copyright(c) 2026 Pravin Pathak <pravin.pathak.dev@gmail.com>
+
+Software Load Balancer (SLB) Eventdev Driver
+============================================
+
+The Software Load Balancer (SLB) event device is a poll-mode eventdev driver
+which distributes events among multiple worker ports and cores without requiring
+dedicated scheduling service cores or hardware coprocessors.
+
+Features
+--------
+
+Queues
+~~~~~~
+
+* **Parallel**: Lock-free multi-producer, multi-consumer ring-based queue allowing
+ concurrent dequeues across all serving worker ports.
+* **Atomic**: Flow-pinned scheduling guaranteeing strict per-flow packet ordering.
+ Includes dynamic load tracking and automatic flow migration to rebalance overloaded
+ ports while preserving in-flight sequence integrity.
+* **Ordered**: Ticket-based sequence tracking and reorder buffer providing end-to-end
+ packet order restoration across parallel processing workers.
+* **Single-Link**: Dedicated direct queue mapping from a producer port to a single
+ consumer port using optimized single-consumer rings.
+
+Ports & Scheduling Capabilities
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+* **Distributed Scheduling**: No dedicated service core required; scheduling work is
+ performed inline during enqueue and dequeue operations.
+* **Prefix Sum Burst Enqueue**: Interleaved multi-queue bursts are grouped and scattered
+ via prefix-sum offsets, dispatching single batched bursts to queue rings to eliminate
+ atomic Compare-And-Swap contention on multi-producer rings.
+* **Dynamic Flow Migration**: Atomic queues monitor port dequeue loads and migrate
+ eligible victim flows from overloaded ports to underloaded ports with cooling cycles.
+* **Credit Pool Management**: Centralized event credit pool with per-port credit caching
+ to enforce device-wide in-flight event limits (``max_num_events``).
+* **Runtime Port Linking**: Supports dynamic linking and unlinking of ports and queues
+ both when stopped and at runtime.
+* **Maintenance Free**: Operates without requiring periodic maintenance calls
+ (``RTE_EVENT_DEV_CAP_MAINTENANCE_FREE``).
+
+Configuration and Options
+-------------------------
+
+The SLB event device is a virtual device (vdev) and can be initialized from the
+application code or from the EAL command line:
+
+* From application code:
+
+ .. code-block:: c
+
+ rte_vdev_init("event_slb0", NULL);
+
+* From EAL command-line options:
+
+ .. code-block:: console
+
+ ./your_dpdk_application --vdev="event_slb0"
+
+Limitations
+-----------
+
+* **Flow Hash Table Size**: Atomic queues use an internal flow mapping table
+ dimensioned to 2048 concurrent active flows per queue (``SLB_ATOMIC_FLOWS``).
+* **Maximum Ports and Queues**: Configured for up to 64 ports (``SLB_MAX_PORTS``)
+ and 64 queues (``SLB_MAX_QUEUES``).
diff --git a/drivers/event/slb/meson.build b/drivers/event/slb/meson.build
index d698a58f16..6caf48720f 100644
--- a/drivers/event/slb/meson.build
+++ b/drivers/event/slb/meson.build
@@ -11,6 +11,7 @@ sources = files(
'slb_evdev.c',
'slb_event.c',
'slb_xstats.c',
+ 'slb_selftest.c',
)
deps += ['bus_vdev', 'ring']
diff --git a/drivers/event/slb/slb_evdev.c b/drivers/event/slb/slb_evdev.c
index 04a2aba165..c86b934b03 100644
--- a/drivers/event/slb/slb_evdev.c
+++ b/drivers/event/slb/slb_evdev.c
@@ -758,6 +758,7 @@ static struct eventdev_ops slb_evdev_ops = {
.xstats_get = slb_xstats_get,
.xstats_get_names = slb_xstats_get_names,
.xstats_get_by_name = slb_xstats_get_by_name,
+ .dev_selftest = test_slb_eventdev,
};
static int
diff --git a/drivers/event/slb/slb_evdev.h b/drivers/event/slb/slb_evdev.h
index 17f3154efa..f727a1b83c 100644
--- a/drivers/event/slb/slb_evdev.h
+++ b/drivers/event/slb/slb_evdev.h
@@ -242,6 +242,8 @@ int slb_xstats_get(const struct rte_eventdev *dev,
uint64_t slb_xstats_get_by_name(const struct rte_eventdev *dev,
const char *name, uint64_t *id);
+int test_slb_eventdev(void);
+
static inline struct slb_evdev *
slb_pmd_priv(const struct rte_eventdev *eventdev)
{
@@ -250,6 +252,8 @@ slb_pmd_priv(const struct rte_eventdev *eventdev)
extern int event_slb_logtype;
#define RTE_LOGTYPE_EVENT_SLB event_slb_logtype
+#define SLB_LOG_LINE(level, ...) \
+ RTE_LOG_LINE(level, EVENT_SLB, __VA_ARGS__)
#define SLB_LOG_DP_LINE(level, fmt, ...) \
RTE_LOG_DP_LINE(level, EVENT_SLB, "%s() line %u: " fmt, \
__func__, __LINE__, ## __VA_ARGS__)
diff --git a/drivers/event/slb/slb_selftest.c b/drivers/event/slb/slb_selftest.c
new file mode 100644
index 0000000000..4a072b811b
--- /dev/null
+++ b/drivers/event/slb/slb_selftest.c
@@ -0,0 +1,522 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2026 Pravin Pathak <pravin.pathak.dev@gmail.com>
+ */
+
+#include <stdio.h>
+#include <string.h>
+#include <stdint.h>
+#include <stdbool.h>
+#include <inttypes.h>
+
+#include <rte_bus_vdev.h>
+#include <rte_common.h>
+#include <rte_cycles.h>
+#include <rte_eventdev.h>
+
+#include "slb_evdev.h"
+
+#define NUM_PORTS 4
+#define NUM_QUEUES 4
+#define PORT_PROD 0
+#define PORT_W1 1
+#define PORT_W2 2
+#define PORT_CONS 3
+
+#define Q_PARALLEL 0
+#define Q_ATOMIC 1
+#define Q_ORDERED 2
+#define Q_SINGLE 3
+
+#define NUM_FLOWS 4
+#define EVENTS_PER_FLOW 16
+#define TOTAL_ATOMIC_EVENTS (NUM_FLOWS * EVENTS_PER_FLOW)
+
+static int g_dev_id = -1;
+
+static int
+init_selftest_device(void)
+{
+ struct rte_event_dev_config dev_conf = {
+ .dequeue_timeout_ns = 0,
+ .nb_event_queues = NUM_QUEUES,
+ .nb_event_ports = NUM_PORTS,
+ .nb_events_limit = 4096,
+ .nb_event_queue_flows = 1024,
+ .nb_event_port_dequeue_depth = 64,
+ .nb_event_port_enqueue_depth = 64,
+ };
+ struct rte_event_queue_conf q_conf;
+ struct rte_event_port_conf p_conf = {
+ .dequeue_depth = 64,
+ .enqueue_depth = 64,
+ .new_event_threshold = 2048,
+ .event_port_cfg = 0
+ };
+ struct rte_event_dev_info info;
+ uint8_t ldb_queues[] = { Q_PARALLEL, Q_ATOMIC, Q_ORDERED };
+ uint8_t single_q = Q_SINGLE;
+ int ret;
+ int p;
+
+ g_dev_id = rte_event_dev_get_dev_id(SLB_PMD_NAME);
+ if (g_dev_id < 0) {
+ ret = rte_vdev_init("event_slb_selftest", NULL);
+ if (ret < 0) {
+ SLB_LOG_LINE(ERR, "Failed to init vdev event_slb_selftest");
+ return -1;
+ }
+ g_dev_id = rte_event_dev_get_dev_id("event_slb_selftest");
+ if (g_dev_id < 0) {
+ SLB_LOG_LINE(ERR, "Cannot find initialized vdev");
+ return -1;
+ }
+ }
+
+ ret = rte_event_dev_info_get(g_dev_id, &info);
+ if (ret < 0) {
+ SLB_LOG_LINE(ERR, "Failed to get dev info");
+ return -1;
+ }
+
+ ret = rte_event_dev_configure(g_dev_id, &dev_conf);
+ if (ret < 0) {
+ SLB_LOG_LINE(ERR, "Failed to configure device, ret=%d", ret);
+ return -1;
+ }
+
+ /* Setup Queues */
+ memset(&q_conf, 0, sizeof(q_conf));
+ q_conf.nb_atomic_flows = 1024;
+ q_conf.nb_atomic_order_sequences = 1024;
+ q_conf.priority = RTE_EVENT_DEV_PRIORITY_NORMAL;
+
+ /* Q0: Parallel */
+ q_conf.schedule_type = RTE_SCHED_TYPE_PARALLEL;
+ q_conf.event_queue_cfg = 0;
+ ret = rte_event_queue_setup(g_dev_id, Q_PARALLEL, &q_conf);
+ if (ret < 0)
+ return -1;
+
+ /* Q1: Atomic */
+ q_conf.schedule_type = RTE_SCHED_TYPE_ATOMIC;
+ q_conf.event_queue_cfg = 0;
+ ret = rte_event_queue_setup(g_dev_id, Q_ATOMIC, &q_conf);
+ if (ret < 0)
+ return -1;
+
+ /* Q2: Ordered */
+ q_conf.schedule_type = RTE_SCHED_TYPE_ORDERED;
+ q_conf.event_queue_cfg = 0;
+ ret = rte_event_queue_setup(g_dev_id, Q_ORDERED, &q_conf);
+ if (ret < 0)
+ return -1;
+
+ /* Q3: Single-link */
+ q_conf.schedule_type = RTE_SCHED_TYPE_PARALLEL;
+ q_conf.event_queue_cfg = RTE_EVENT_QUEUE_CFG_SINGLE_LINK;
+ ret = rte_event_queue_setup(g_dev_id, Q_SINGLE, &q_conf);
+ if (ret < 0)
+ return -1;
+
+ /* Setup Ports */
+ for (p = 0; p < NUM_PORTS; p++) {
+ if (p == PORT_CONS)
+ p_conf.event_port_cfg = RTE_EVENT_PORT_CFG_SINGLE_LINK;
+ else
+ p_conf.event_port_cfg = 0;
+
+ ret = rte_event_port_setup(g_dev_id, p, &p_conf);
+ if (ret < 0)
+ return -1;
+ }
+
+ /* Link Queues */
+ ret = rte_event_port_link(g_dev_id, PORT_W1, ldb_queues, NULL, 3);
+ if (ret != 3)
+ return -1;
+
+ ret = rte_event_port_link(g_dev_id, PORT_W2, ldb_queues, NULL, 3);
+ if (ret != 3)
+ return -1;
+
+ ret = rte_event_port_link(g_dev_id, PORT_CONS, &single_q, NULL, 1);
+ if (ret != 1)
+ return -1;
+
+ ret = rte_event_dev_start(g_dev_id);
+ if (ret < 0)
+ return -1;
+
+ return 0;
+}
+
+static void
+cleanup_selftest_device(void)
+{
+ if (g_dev_id >= 0) {
+ rte_event_dev_stop(g_dev_id);
+ rte_event_dev_close(g_dev_id);
+ g_dev_id = -1;
+ }
+}
+
+static int
+test_parallel_queue(void)
+{
+ struct rte_event rx_ev[32];
+ struct rte_event ev[32];
+ uint16_t enqd, deqd = 0;
+ uint16_t n = 0;
+ int tries = 0;
+ int i;
+
+ memset(ev, 0, sizeof(ev));
+
+ for (i = 0; i < 32; i++) {
+ ev[i].op = RTE_EVENT_OP_NEW;
+ ev[i].sched_type = RTE_SCHED_TYPE_PARALLEL;
+ ev[i].queue_id = Q_PARALLEL;
+ ev[i].u64 = 1000 + i;
+ }
+
+ enqd = rte_event_enqueue_burst(g_dev_id, PORT_PROD, ev, 32);
+ if (enqd != 32) {
+ SLB_LOG_LINE(ERR, "Parallel enqueue failed, expected 32, got %u", enqd);
+ return -1;
+ }
+
+ while (deqd < 32 && tries++ < 1000) {
+ n = rte_event_dequeue_burst(g_dev_id, PORT_W1, rx_ev + deqd,
+ 32 - deqd, 0);
+ deqd += n;
+ if (deqd < 32) {
+ n = rte_event_dequeue_burst(g_dev_id, PORT_W2,
+ rx_ev + deqd, 32 - deqd, 0);
+ deqd += n;
+ }
+ }
+
+ if (deqd != 32) {
+ SLB_LOG_LINE(ERR, "Parallel dequeue failed, expected 32, got %u", deqd);
+ return -1;
+ }
+
+ /* Release events */
+ for (i = 0; i < 32; i++)
+ rx_ev[i].op = RTE_EVENT_OP_RELEASE;
+ rte_event_enqueue_burst(g_dev_id, PORT_W1, rx_ev, 32);
+
+ return 0;
+}
+
+static int
+test_atomic_queue(void)
+{
+ struct rte_event rx_ev[64];
+ struct rte_event ev[64];
+ int expected_seq[NUM_FLOWS];
+ uint16_t enqd = 0;
+ uint16_t deqd = 0;
+ int tries = 0;
+ int idx = 0;
+ int e, f;
+ uint16_t i, n;
+
+ memset(ev, 0, sizeof(ev));
+
+ for (e = 0; e < EVENTS_PER_FLOW; e++) {
+ for (f = 0; f < NUM_FLOWS; f++) {
+ ev[idx].op = RTE_EVENT_OP_NEW;
+ ev[idx].sched_type = RTE_SCHED_TYPE_ATOMIC;
+ ev[idx].queue_id = Q_ATOMIC;
+ ev[idx].flow_id = f;
+ ev[idx].u64 = e; /* sequence number */
+ idx++;
+ }
+ }
+
+ while (enqd < TOTAL_ATOMIC_EVENTS) {
+ n = rte_event_enqueue_burst(g_dev_id, PORT_PROD, ev + enqd,
+ TOTAL_ATOMIC_EVENTS - enqd);
+ if (n == 0)
+ break;
+ enqd += n;
+ }
+ if (enqd != TOTAL_ATOMIC_EVENTS) {
+ SLB_LOG_LINE(ERR, "Atomic enqueue failed, expected %d, got %u",
+ TOTAL_ATOMIC_EVENTS, enqd);
+ return -1;
+ }
+
+ for (f = 0; f < NUM_FLOWS; f++)
+ expected_seq[f] = 0;
+
+ while (deqd < TOTAL_ATOMIC_EVENTS && tries++ < 2000) {
+ n = rte_event_dequeue_burst(g_dev_id, PORT_W1, rx_ev, 16, 0);
+ for (i = 0; i < n; i++) {
+ int fid = rx_ev[i].flow_id;
+
+ if (rx_ev[i].u64 != (uint64_t)expected_seq[fid]) {
+ SLB_LOG_LINE(ERR,
+ "Flow %d out of order! Expected %d got %" PRIu64,
+ fid, expected_seq[fid], rx_ev[i].u64);
+ return -1;
+ }
+ expected_seq[fid]++;
+ rx_ev[i].op = RTE_EVENT_OP_RELEASE;
+ }
+ if (n > 0)
+ rte_event_enqueue_burst(g_dev_id, PORT_W1, rx_ev, n);
+ deqd += n;
+
+ n = rte_event_dequeue_burst(g_dev_id, PORT_W2, rx_ev, 16, 0);
+ for (i = 0; i < n; i++) {
+ int fid = rx_ev[i].flow_id;
+
+ if (rx_ev[i].u64 != (uint64_t)expected_seq[fid]) {
+ SLB_LOG_LINE(ERR,
+ "Flow %d out of order! Expected %d got %" PRIu64,
+ fid, expected_seq[fid], rx_ev[i].u64);
+ return -1;
+ }
+ expected_seq[fid]++;
+ rx_ev[i].op = RTE_EVENT_OP_RELEASE;
+ }
+ if (n > 0)
+ rte_event_enqueue_burst(g_dev_id, PORT_W2, rx_ev, n);
+ deqd += n;
+ }
+
+ if (deqd != TOTAL_ATOMIC_EVENTS) {
+ SLB_LOG_LINE(ERR, "Atomic dequeue failed, expected %d got %u",
+ TOTAL_ATOMIC_EVENTS, deqd);
+ return -1;
+ }
+
+ return 0;
+}
+
+static int
+test_ordered_forward(void)
+{
+ const int num_events = 32;
+ struct rte_event rx[16];
+ struct rte_event ev[32];
+ uint16_t enqd;
+ uint16_t fwd = 0;
+ uint16_t cons_rx = 0;
+ int tries = 0;
+ int i;
+ uint16_t n;
+
+ memset(ev, 0, sizeof(ev));
+
+ for (i = 0; i < num_events; i++) {
+ ev[i].op = RTE_EVENT_OP_NEW;
+ ev[i].sched_type = RTE_SCHED_TYPE_ORDERED;
+ ev[i].queue_id = Q_ORDERED;
+ ev[i].u64 = i;
+ }
+
+ enqd = rte_event_enqueue_burst(g_dev_id, PORT_PROD, ev, num_events);
+ if (enqd != num_events)
+ return -1;
+
+ /* Workers dequeue and forward to Q_SINGLE */
+ while (fwd < num_events && tries++ < 2000) {
+ n = rte_event_dequeue_burst(g_dev_id, PORT_W1, rx, 16, 0);
+ for (i = 0; i < n; i++) {
+ rx[i].op = RTE_EVENT_OP_FORWARD;
+ rx[i].queue_id = Q_SINGLE;
+ rx[i].sched_type = RTE_SCHED_TYPE_PARALLEL;
+ }
+ if (n > 0)
+ rte_event_enqueue_burst(g_dev_id, PORT_W1, rx, n);
+ fwd += n;
+
+ n = rte_event_dequeue_burst(g_dev_id, PORT_W2, rx, 16, 0);
+ for (i = 0; i < n; i++) {
+ rx[i].op = RTE_EVENT_OP_FORWARD;
+ rx[i].queue_id = Q_SINGLE;
+ rx[i].sched_type = RTE_SCHED_TYPE_PARALLEL;
+ }
+ if (n > 0)
+ rte_event_enqueue_burst(g_dev_id, PORT_W2, rx, n);
+ fwd += n;
+ }
+
+ if (fwd != num_events) {
+ SLB_LOG_LINE(ERR, "Ordered workers failed to forward all events (%u/%d)",
+ fwd, num_events);
+ return -1;
+ }
+
+ /* Consumer on PORT_CONS must receive all events in exact sequence 0..num_events-1 */
+ tries = 0;
+ while (cons_rx < num_events && tries++ < 2000) {
+ n = rte_event_dequeue_burst(g_dev_id, PORT_CONS, rx, 16, 0);
+ for (i = 0; i < n; i++) {
+ if (rx[i].u64 != cons_rx) {
+ SLB_LOG_LINE(ERR,
+ "Ordered seq violation! Expected %u got %" PRIu64,
+ cons_rx, rx[i].u64);
+ return -1;
+ }
+ cons_rx++;
+ rx[i].op = RTE_EVENT_OP_RELEASE;
+ }
+ if (n > 0)
+ rte_event_enqueue_burst(g_dev_id, PORT_CONS, rx, n);
+ }
+
+ if (cons_rx != num_events) {
+ SLB_LOG_LINE(ERR, "Consumer received %u/%d ordered events",
+ cons_rx, num_events);
+ return -1;
+ }
+
+ return 0;
+}
+
+static int
+test_single_link(void)
+{
+ struct rte_event rx[16];
+ struct rte_event ev[16];
+ uint16_t enqd;
+ uint16_t deqd = 0;
+ int tries = 0;
+ int i;
+ uint16_t n;
+
+ memset(ev, 0, sizeof(ev));
+
+ for (i = 0; i < 16; i++) {
+ ev[i].op = RTE_EVENT_OP_NEW;
+ ev[i].sched_type = RTE_SCHED_TYPE_PARALLEL;
+ ev[i].queue_id = Q_SINGLE;
+ ev[i].u64 = 5000 + i;
+ }
+
+ enqd = rte_event_enqueue_burst(g_dev_id, PORT_PROD, ev, 16);
+ if (enqd != 16)
+ return -1;
+
+ while (deqd < 16 && tries++ < 1000) {
+ n = rte_event_dequeue_burst(g_dev_id, PORT_CONS,
+ rx + deqd, 16 - deqd, 0);
+ deqd += n;
+ }
+
+ if (deqd != 16) {
+ SLB_LOG_LINE(ERR, "Single-link dequeue failed (%u/16)", deqd);
+ return -1;
+ }
+
+ for (i = 0; i < 16; i++) {
+ if (rx[i].u64 != (uint64_t)(5000 + i))
+ return -1;
+ rx[i].op = RTE_EVENT_OP_RELEASE;
+ }
+ rte_event_enqueue_burst(g_dev_id, PORT_CONS, rx, 16);
+
+ return 0;
+}
+
+static int
+test_stop_flush(void)
+{
+ struct rte_event rx[16];
+ struct rte_event ev[16];
+ uint16_t enqd;
+ uint16_t deqd;
+ int ret;
+ int i;
+
+ memset(ev, 0, sizeof(ev));
+
+ for (i = 0; i < 16; i++) {
+ ev[i].op = RTE_EVENT_OP_NEW;
+ ev[i].sched_type = RTE_SCHED_TYPE_PARALLEL;
+ ev[i].queue_id = Q_PARALLEL;
+ ev[i].u64 = 9000 + i;
+ }
+
+ enqd = rte_event_enqueue_burst(g_dev_id, PORT_PROD, ev, 16);
+ if (enqd != 16)
+ return -1;
+
+ /* Stop device flushes all queued events */
+ rte_event_dev_stop(g_dev_id);
+
+ /* Restart device */
+ ret = rte_event_dev_start(g_dev_id);
+ if (ret < 0)
+ return -1;
+
+ /* Verify no leftover events */
+ deqd = rte_event_dequeue_burst(g_dev_id, PORT_W1, rx, 16, 0);
+ if (deqd != 0) {
+ SLB_LOG_LINE(ERR, "Stop flush failed, received %u events after restart",
+ deqd);
+ return -1;
+ }
+
+ return 0;
+}
+
+int
+test_slb_eventdev(void)
+{
+ SLB_LOG_LINE(INFO, "=== Running SLB Eventdev Selftest ===");
+
+ if (init_selftest_device() < 0) {
+ SLB_LOG_LINE(ERR, "Initialization failed");
+ cleanup_selftest_device();
+ return -1;
+ }
+
+ SLB_LOG_LINE(INFO, " [1/5] Testing Parallel Queue...");
+ if (test_parallel_queue() < 0) {
+ SLB_LOG_LINE(ERR, " FAILED: Parallel Queue test failed");
+ cleanup_selftest_device();
+ return -1;
+ }
+ SLB_LOG_LINE(INFO, " [1/5] Parallel Queue: PASS");
+
+ SLB_LOG_LINE(INFO, " [2/5] Testing Atomic Queue...");
+ if (test_atomic_queue() < 0) {
+ SLB_LOG_LINE(ERR, " FAILED: Atomic Queue test failed");
+ cleanup_selftest_device();
+ return -1;
+ }
+ SLB_LOG_LINE(INFO, " [2/5] Atomic Queue: PASS");
+
+ SLB_LOG_LINE(INFO, " [3/5] Testing Ordered Queue...");
+ if (test_ordered_forward() < 0) {
+ SLB_LOG_LINE(ERR, " FAILED: Ordered Queue test failed");
+ cleanup_selftest_device();
+ return -1;
+ }
+ SLB_LOG_LINE(INFO, " [3/5] Ordered Queue: PASS");
+
+ SLB_LOG_LINE(INFO, " [4/5] Testing Single-Link Queue...");
+ if (test_single_link() < 0) {
+ SLB_LOG_LINE(ERR, " FAILED: Single-Link Queue test failed");
+ cleanup_selftest_device();
+ return -1;
+ }
+ SLB_LOG_LINE(INFO, " [4/5] Single-Link Queue: PASS");
+
+ SLB_LOG_LINE(INFO, " [5/5] Testing Device Stop & Flush...");
+ if (test_stop_flush() < 0) {
+ SLB_LOG_LINE(ERR, " FAILED: Stop Flush test failed");
+ cleanup_selftest_device();
+ return -1;
+ }
+ SLB_LOG_LINE(INFO, " [5/5] Device Stop & Flush: PASS");
+
+ cleanup_selftest_device();
+ SLB_LOG_LINE(INFO, "=== SLB Eventdev Selftest PASSED ===");
+ return 0;
+}
--
2.53.0
^ permalink raw reply related [flat|nested] 14+ messages in thread* [PATCH v1 7/8] event/slb: update resets and credit defaults
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
` (5 preceding siblings ...)
2026-11-07 1:30 ` [PATCH v1 6/8] event/slb: add self-test and documentation Pravin Pathak
@ 2026-11-07 1:30 ` Pravin Pathak
2026-11-07 1:30 ` [PATCH v1 8/8] event/slb: improve port linking, config checks and atomic refill Pravin Pathak
2026-11-07 7:29 ` [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Jerin Jacob
8 siblings, 0 replies; 14+ messages in thread
From: Pravin Pathak @ 2026-11-07 1:30 UTC (permalink / raw)
To: dev
Cc: jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
From: Pravin Pathak <pravin.pathak.dev@gmail.com>
- Force RTE_SCHED_TYPE_ATOMIC on events dequeued from single-link queues
and skip atomic/ordered state allocations for single-link queues.
- Remove RTE_EVENT_DEV_CAP_RUNTIME_PORT_LINK advertisement from driver
info, feature matrix, and release notes.
- Reset atomic queue flow-to-port mappings, hold buffers, migration
state, per-port stats, ordered reorder buffers, and port history when
the device is stopped in slb_stop().
- Default port new_event_threshold in slb_port_def_conf() to 1/4 of
SLB_IN_RING_SIZE (clamped to nb_events_limit when configured) and
lower SLB_MIN_CREDIT_LOAN from 512 to 256 so more ports can borrow
credits concurrently.
Signed-off-by: Pravin Pathak <pravin.pathak.dev@gmail.com>
---
doc/guides/eventdevs/features/slb.ini | 1 -
doc/guides/eventdevs/slb.rst | 31 ++++------
drivers/event/slb/slb_evdev.c | 81 +++++++++++++++++++++------
drivers/event/slb/slb_evdev.h | 5 +-
drivers/event/slb/slb_event.c | 23 ++++++--
drivers/event/slb/slb_selftest.c | 12 ++++
6 files changed, 110 insertions(+), 43 deletions(-)
diff --git a/doc/guides/eventdevs/features/slb.ini b/doc/guides/eventdevs/features/slb.ini
index 0364840bc9..33036301a8 100644
--- a/doc/guides/eventdevs/features/slb.ini
+++ b/doc/guides/eventdevs/features/slb.ini
@@ -11,7 +11,6 @@ distributed_sched = Y
burst_mode = Y
implicit_release_disable = Y
nonseq_mode = Y
-runtime_port_link = Y
multiple_queue_port = Y
carry_flow_id = Y
maintenance_free = Y
diff --git a/doc/guides/eventdevs/slb.rst b/doc/guides/eventdevs/slb.rst
index d319b94920..b109a1e2ca 100644
--- a/doc/guides/eventdevs/slb.rst
+++ b/doc/guides/eventdevs/slb.rst
@@ -4,9 +4,10 @@
Software Load Balancer (SLB) Eventdev Driver
============================================
-The Software Load Balancer (SLB) event device is a poll-mode eventdev driver
-which distributes events among multiple worker ports and cores without requiring
-dedicated scheduling service cores or hardware coprocessors.
+The Software Load Balancer (SLB) is a software event device and a poll-mode Eventdev driver.
+It uses a distributed scheduler, allowing it to scale well. SLB does not require dedicated
+cores for scheduling. It supports Atomic, Parallel, and Ordered queues with dynamic atomic
+flow load balancing. SLB also requires no internal core communication and is maintenance-free.
Features
--------
@@ -19,25 +20,17 @@ Queues
* **Atomic**: Flow-pinned scheduling guaranteeing strict per-flow packet ordering.
Includes dynamic load tracking and automatic flow migration to rebalance overloaded
ports while preserving in-flight sequence integrity.
-* **Ordered**: Ticket-based sequence tracking and reorder buffer providing end-to-end
- packet order restoration across parallel processing workers.
-* **Single-Link**: Dedicated direct queue mapping from a producer port to a single
- consumer port using optimized single-consumer rings.
+* **Ordered**: Sequence number-based tracking and reordering.
+* **Single-Link**: Implemented using optimized multi-producer single-consumer rings.
Ports & Scheduling Capabilities
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-* **Distributed Scheduling**: No dedicated service core required; scheduling work is
- performed inline during enqueue and dequeue operations.
-* **Prefix Sum Burst Enqueue**: Interleaved multi-queue bursts are grouped and scattered
- via prefix-sum offsets, dispatching single batched bursts to queue rings to eliminate
- atomic Compare-And-Swap contention on multi-producer rings.
+* **Distributed Scheduling**: No dedicated service core required.
* **Dynamic Flow Migration**: Atomic queues monitor port dequeue loads and migrate
- eligible victim flows from overloaded ports to underloaded ports with cooling cycles.
+ Eligible flows from overloaded ports to underloaded ports with cooling cycles.
* **Credit Pool Management**: Centralized event credit pool with per-port credit caching
to enforce device-wide in-flight event limits (``max_num_events``).
-* **Runtime Port Linking**: Supports dynamic linking and unlinking of ports and queues
- both when stopped and at runtime.
* **Maintenance Free**: Operates without requiring periodic maintenance calls
(``RTE_EVENT_DEV_CAP_MAINTENANCE_FREE``).
@@ -62,7 +55,7 @@ application code or from the EAL command line:
Limitations
-----------
-* **Flow Hash Table Size**: Atomic queues use an internal flow mapping table
- dimensioned to 2048 concurrent active flows per queue (``SLB_ATOMIC_FLOWS``).
-* **Maximum Ports and Queues**: Configured for up to 64 ports (``SLB_MAX_PORTS``)
- and 64 queues (``SLB_MAX_QUEUES``).
+* **Maximum Ports and Queues**: Supports up to 64 ports and 64 queues.
+* **Priorities and linking**: Supports two levels of service priority to link up to
+ eight queues per port at each priority. SLB currently does not support event priority.
+* **Queues Types**: Does not support all-type queues.
diff --git a/drivers/event/slb/slb_evdev.c b/drivers/event/slb/slb_evdev.c
index c86b934b03..16bfefaa68 100644
--- a/drivers/event/slb/slb_evdev.c
+++ b/drivers/event/slb/slb_evdev.c
@@ -26,6 +26,7 @@ slb_port_reset_history(struct slb_port *port)
{
port->history[0].active = false;
port->history[0].next_free = SLB_PORT_HISTORY_FREE_END;
+ /* 1 -> 2 -> 3 ..... 255 -> END */
for (int i = 1; i < SLB_PORT_HISTORY_SIZE; i++) {
port->history[i].active = false;
port->history[i].next_free = i + 1;
@@ -36,7 +37,7 @@ slb_port_reset_history(struct slb_port *port)
port->history_avail = SLB_PORT_HISTORY_MAX_TOKENS;
}
-/* Setup Eventdev port. Reset all mapped queues */
+/* Setup Eventdev port. Reset all mapped queues and history tracking */
static int
slb_port_setup(struct rte_eventdev *dev, uint8_t port_id,
const struct rte_event_port_conf *conf)
@@ -103,12 +104,18 @@ slb_port_setup(struct rte_eventdev *dev, uint8_t port_id,
/* Default Eventdev port configuration for SLB */
static void
-slb_port_def_conf(struct rte_eventdev *dev __rte_unused,
+slb_port_def_conf(struct rte_eventdev *dev,
uint8_t port_id __rte_unused,
struct rte_event_port_conf *port_conf)
{
+ int32_t threshold = SLB_IN_RING_SIZE / 4;
+
+ if (dev->data->dev_conf.nb_events_limit > 0 &&
+ threshold > dev->data->dev_conf.nb_events_limit)
+ threshold = dev->data->dev_conf.nb_events_limit;
+
*port_conf = (struct rte_event_port_conf) {
- .new_event_threshold = 1024,
+ .new_event_threshold = threshold,
.dequeue_depth = SLB_MAX_PORT_DEQUEUE_DEPTH / 4,
.enqueue_depth = SLB_MAX_PORT_ENQUEUE_DEPTH / 4
};
@@ -146,7 +153,9 @@ slb_queue_release(struct rte_eventdev *dev, uint8_t queue_id)
}
}
-/* Setup Eventdev queue. Eventdev queues are implemented as DPDK rings */
+/* Setup Eventdev queue. Eventdev queues are implemented as DPDK rings.
+ * SLB Supports single-link, atomic, parallel, and ordered queues.
+ */
static int
slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
const struct rte_event_queue_conf *conf)
@@ -190,12 +199,15 @@ slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
return -ENOMEM;
queue->q_ring = ring;
- queue->schedule_type = conf->schedule_type;
+ /* Single-link queue uses atomic schedule type */
+ queue->schedule_type = queue->is_single_link ?
+ RTE_SCHED_TYPE_ATOMIC : conf->schedule_type;
queue->serving_ports_mask = 0;
queue->num_serving_ports = 0;
- /* Atomic queue allocation */
- if (conf->schedule_type == RTE_SCHED_TYPE_ATOMIC) {
+ /* Atomic queue state allocation */
+ if (!queue->is_single_link &&
+ conf->schedule_type == RTE_SCHED_TYPE_ATOMIC) {
for (int i = 0; i < SLB_MAX_PORTS; i++)
queue->per_port_rings[i] = NULL;
@@ -224,7 +236,8 @@ slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
}
/* Ordered queue reorder buffer allocation */
- if (conf->schedule_type == RTE_SCHED_TYPE_ORDERED) {
+ if (!queue->is_single_link &&
+ conf->schedule_type == RTE_SCHED_TYPE_ORDERED) {
queue->reorder_buffer = rte_zmalloc_socket("slb_reorder_buf",
sizeof(struct slb_reorder_entry) * ring->r.size,
RTE_CACHE_LINE_SIZE, dev->data->socket_id);
@@ -293,7 +306,8 @@ slb_port_link_one(struct slb_port *port, struct slb_queue *queue,
port->single_linked_q = queue;
port->single_linked_ring = queue->q_ring;
} else {
- if (queue->schedule_type == RTE_SCHED_TYPE_ATOMIC &&
+ if (!queue->is_single_link &&
+ queue->schedule_type == RTE_SCHED_TYPE_ATOMIC &&
queue->per_port_rings[port->id] == NULL) {
char ring_name[RTE_RING_NAMESIZE];
struct rte_event_ring *ring;
@@ -332,7 +346,8 @@ slb_port_link_one(struct slb_port *port, struct slb_queue *queue,
if (!found) {
uint16_t slot = port->num_q[prio];
struct rte_event_ring *dq_ring =
- (queue->schedule_type == RTE_SCHED_TYPE_ATOMIC) ?
+ (!queue->is_single_link &&
+ queue->schedule_type == RTE_SCHED_TYPE_ATOMIC) ?
queue->per_port_rings[port->id] : queue->q_ring;
if (slot >= SLB_PORT_MAX_Q_MAP_PER_PRIO) {
@@ -344,6 +359,8 @@ slb_port_link_one(struct slb_port *port, struct slb_queue *queue,
port->q_entries[prio][slot].queue_id = qid;
port->q_entries[prio][slot].schedule_type =
queue->schedule_type;
+ port->q_entries[prio][slot].is_single_link =
+ queue->is_single_link;
port->num_q[prio]++;
}
@@ -464,7 +481,6 @@ slb_info_get(struct rte_eventdev *dev __rte_unused,
RTE_EVENT_DEV_CAP_IMPLICIT_RELEASE_DISABLE |
RTE_EVENT_DEV_CAP_NONSEQ_MODE |
RTE_EVENT_DEV_CAP_MULTIPLE_QUEUE_PORT |
- RTE_EVENT_DEV_CAP_RUNTIME_PORT_LINK |
RTE_EVENT_DEV_CAP_CARRY_FLOW_ID |
RTE_EVENT_DEV_CAP_MAINTENANCE_FREE |
RTE_EVENT_DEV_CAP_INDEPENDENT_ENQ
@@ -555,22 +571,52 @@ slb_stop(struct rte_eventdev *dev)
for (uint8_t qid = 0; qid < slb->num_queues; qid++) {
struct slb_queue *q = &slb->queues[qid];
- if (q->schedule_type == RTE_SCHED_TYPE_ATOMIC && q->atomic_state != NULL) {
+ if (q->schedule_type == RTE_SCHED_TYPE_ATOMIC &&
+ q->atomic_state != NULL) {
+ struct slb_atomic_state *astate = q->atomic_state;
+
+ for (uint16_t h = 0; h < astate->hold_count; h++) {
+ if (flush != NULL)
+ flush(dev_id, astate->hold_buffer[h],
+ flush_arg);
+ }
+ astate->hold_count = 0;
+ astate->next_port_rr = 0;
+ astate->migration.in_progress = false;
+
for (int f = 0; f < SLB_ATOMIC_FLOWS; f++) {
rte_atomic_store_explicit(
- &q->atomic_state->flow_map.flows[f].port_id,
+ &astate->flow_map.flows[f].port_id,
-1, rte_memory_order_relaxed);
- q->atomic_state->flow_last_migrated[f] = 0;
+ astate->flow_last_migrated[f] = 0;
}
- q->atomic_state->migration.in_progress = false;
+
+ memset(astate->port_stats, 0,
+ sizeof(astate->port_stats));
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],
flush, flush_arg);
}
+ } else if (q->schedule_type == RTE_SCHED_TYPE_ORDERED &&
+ q->reorder_buffer != NULL && q->q_ring != NULL) {
+ for (uint32_t s = 0; s < q->q_ring->r.size; s++) {
+ uint8_t st = rte_atomic_load_explicit(
+ &q->reorder_buffer[s].state,
+ rte_memory_order_relaxed);
+
+ if (st == SLB_REORDER_READY && flush != NULL)
+ flush(dev_id,
+ q->reorder_buffer[s].event,
+ flush_arg);
+ rte_atomic_store_explicit(
+ &q->reorder_buffer[s].state,
+ SLB_REORDER_EMPTY,
+ rte_memory_order_relaxed);
+ }
+ q->reorder_drain_idx = 0;
}
if (q->q_ring != NULL)
@@ -581,7 +627,10 @@ slb_stop(struct rte_eventdev *dev)
struct slb_port *port = &slb->ports[i];
port->inflight_credits = 0;
port->credits_drained = false;
+ port->pending_releases = 0;
port->is_loaded = false;
+ port->last_migrated_time = 0;
+ slb_port_reset_history(port);
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 f727a1b83c..bf262903cd 100644
--- a/drivers/event/slb/slb_evdev.h
+++ b/drivers/event/slb/slb_evdev.h
@@ -28,7 +28,7 @@
#define SLB_MAX_FLOWS_MASK (SLB_MAX_FLOWS - 1)
/* Avoid making small 'loans' from the central in-flight event credit pool */
-#define SLB_MIN_CREDIT_LOAN (512)
+#define SLB_MIN_CREDIT_LOAN (256)
#define SLB_PORT_MAX_CREDITS (2 * SLB_MIN_CREDIT_LOAN)
#define SLB_PORT_MIN_CREDITS (SLB_MIN_CREDIT_LOAN)
@@ -56,7 +56,8 @@ struct slb_port_q_entry {
struct slb_queue *queue;
uint8_t queue_id;
uint8_t schedule_type;
- uint8_t pad[6];
+ bool is_single_link;
+ uint8_t pad[5];
};
#define SLB_PORT_HISTORY_SIZE 256
diff --git a/drivers/event/slb/slb_event.c b/drivers/event/slb/slb_event.c
index 3a0f7f6fde..c44deba905 100644
--- a/drivers/event/slb/slb_event.c
+++ b/drivers/event/slb/slb_event.c
@@ -976,8 +976,10 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
return 0;
dequeued = rte_event_ring_dequeue_burst(port->single_linked_ring,
events, num, NULL);
- for (uint16_t j = 0; j < dequeued; j++)
+ for (uint16_t j = 0; j < dequeued; j++) {
+ events[j].sched_type = RTE_SCHED_TYPE_ATOMIC;
events[j].impl_opaque = 0;
+ }
return dequeued;
}
@@ -1015,8 +1017,11 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
continue;
uint8_t sched_type = qent->schedule_type;
- bool is_ordered = (sched_type == RTE_SCHED_TYPE_ORDERED);
- bool is_atomic = (sched_type == RTE_SCHED_TYPE_ATOMIC);
+ bool is_single_link = qent->is_single_link;
+ bool is_ordered = !is_single_link &&
+ (sched_type == RTE_SCHED_TYPE_ORDERED);
+ bool is_atomic = !is_single_link &&
+ (sched_type == RTE_SCHED_TYPE_ATOMIC);
if (is_ordered || is_atomic) {
if (unlikely(port->history_avail == 0))
@@ -1050,8 +1055,16 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
} else {
dequeued = rte_event_ring_dequeue_burst(
ring_to_dq, events, num, NULL);
- for (uint16_t j = 0; j < dequeued; j++)
- events[j].impl_opaque = 0;
+ if (is_single_link) {
+ for (uint16_t j = 0; j < dequeued; j++) {
+ events[j].sched_type =
+ RTE_SCHED_TYPE_ATOMIC;
+ events[j].impl_opaque = 0;
+ }
+ } else {
+ for (uint16_t j = 0; j < dequeued; j++)
+ events[j].impl_opaque = 0;
+ }
}
if (dequeued > 0) {
diff --git a/drivers/event/slb/slb_selftest.c b/drivers/event/slb/slb_selftest.c
index 4a072b811b..776b6d6ab4 100644
--- a/drivers/event/slb/slb_selftest.c
+++ b/drivers/event/slb/slb_selftest.c
@@ -362,6 +362,12 @@ test_ordered_forward(void)
cons_rx, rx[i].u64);
return -1;
}
+ if (rx[i].sched_type != RTE_SCHED_TYPE_ATOMIC) {
+ SLB_LOG_LINE(ERR,
+ "Single-link dequeue sched_type expected %u, got %u",
+ RTE_SCHED_TYPE_ATOMIC, rx[i].sched_type);
+ return -1;
+ }
cons_rx++;
rx[i].op = RTE_EVENT_OP_RELEASE;
}
@@ -416,6 +422,12 @@ test_single_link(void)
for (i = 0; i < 16; i++) {
if (rx[i].u64 != (uint64_t)(5000 + i))
return -1;
+ if (rx[i].sched_type != RTE_SCHED_TYPE_ATOMIC) {
+ SLB_LOG_LINE(ERR,
+ "Single-link dequeue sched_type expected %u, got %u",
+ RTE_SCHED_TYPE_ATOMIC, rx[i].sched_type);
+ return -1;
+ }
rx[i].op = RTE_EVENT_OP_RELEASE;
}
rte_event_enqueue_burst(g_dev_id, PORT_CONS, rx, 16);
--
2.53.0
^ permalink raw reply related [flat|nested] 14+ messages in thread* [PATCH v1 8/8] event/slb: improve port linking, config checks and atomic refill
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
` (6 preceding siblings ...)
2026-11-07 1:30 ` [PATCH v1 7/8] event/slb: update resets and credit defaults Pravin Pathak
@ 2026-11-07 1:30 ` Pravin Pathak
2026-11-07 7:29 ` [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Jerin Jacob
8 siblings, 0 replies; 14+ messages in thread
From: Pravin Pathak @ 2026-11-07 1:30 UTC (permalink / raw)
To: dev
Cc: jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
From: Pravin Pathak <pravin.pathak.dev@gmail.com>
- Validate duplicate/conflicting queue-port links and slot capacity at
the start of slb_port_link_one(), and fix priority levels to 2.
- Validate nb_event_ports and nb_event_queues limits in slb_configure(),
avoid rounding max_inflight to a power of 2, and precompute
migration_cooldown_cycles once during configuration.
- Verify all queues are configured and linked to at least one port in
slb_start().
- Use relaxed stores when clearing drained reorder buffer slots and
avoid duplicate acquire load on the first burst element in
slb_reorder_drain().
- Ensure hold buffer capacity before dequeuing in slb_atomic_refill()
and continue refilling uncongested ports while hold buffer space
remains.
- Enforce 1-to-1 single-link port and single-link queue pairing in
slb_port_link_one(), disallowing linking a single-link queue to a
regular port or a regular queue to a single-linked port.
- Preserve event sched_type on single-link dequeue and only clear
impl_opaque to 0.
Signed-off-by: Pravin Pathak <pravin.pathak.dev@gmail.com>
---
drivers/event/slb/slb_evdev.c | 200 +++++++++++++--------
drivers/event/slb/slb_evdev.h | 41 +----
drivers/event/slb/slb_event.c | 290 ++++++++++++++++++-------------
drivers/event/slb/slb_selftest.c | 12 --
4 files changed, 306 insertions(+), 237 deletions(-)
diff --git a/drivers/event/slb/slb_evdev.c b/drivers/event/slb/slb_evdev.c
index 16bfefaa68..346b4b38ed 100644
--- a/drivers/event/slb/slb_evdev.c
+++ b/drivers/event/slb/slb_evdev.c
@@ -84,10 +84,6 @@ slb_port_setup(struct rte_eventdev *dev, uint8_t port_id,
.last_migrated_time = 0,
};
- port->remap_input_q = false;
- rte_atomic_store_explicit(&port->active_queues, 0,
- rte_memory_order_relaxed);
-
dev->data->ports[port_id] = port;
/* Reset all queues mapped to this port */
@@ -199,9 +195,7 @@ slb_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
return -ENOMEM;
queue->q_ring = ring;
- /* Single-link queue uses atomic schedule type */
- queue->schedule_type = queue->is_single_link ?
- RTE_SCHED_TYPE_ATOMIC : conf->schedule_type;
+ queue->schedule_type = conf->schedule_type;
queue->serving_ports_mask = 0;
queue->num_serving_ports = 0;
@@ -270,16 +264,20 @@ slb_queue_def_conf(struct rte_eventdev *dev __rte_unused,
static void
slb_port_unlink_queue_from_prio(struct slb_port *port, int prio, int slot)
{
+ /* Shift the queue entries to overwriting the unlinked queue slot */
for (int k = slot; k < port->num_q[prio] - 1; k++)
port->q_entries[prio][k] = port->q_entries[prio][k + 1];
+ /* Decrement mapped queue count and Clear the last slot after shifting */
port->num_q[prio]--;
memset(&port->q_entries[prio][port->num_q[prio]], 0,
sizeof(struct slb_port_q_entry));
+ /* Adjust the next queue index if it points beyond the new count */
if (port->next_q[prio] >= port->num_q[prio])
port->next_q[prio] = 0;
}
+/* Find the slot of a queue in a specific priority level, returning -1 if not found */
static int
slb_port_find_queue_slot(struct slb_port *port, int prio, uint8_t qid)
{
@@ -290,24 +288,67 @@ slb_port_find_queue_slot(struct slb_port *port, int prio, uint8_t qid)
return -1;
}
+/* Link a queue to a port at a specific priority level */
static int
slb_port_link_one(struct slb_port *port, struct slb_queue *queue,
uint8_t qid, uint8_t app_prio, bool has_prio)
{
+ struct rte_event_ring *dq_ring;
+ uint16_t slot;
int prio;
- bool found = false;
if (port->is_single_linked) {
- if (port->single_linked_q != NULL &&
- port->single_linked_q != queue) {
+ /* Only a single-link queue can be linked to a single-linked port */
+ if (!queue->is_single_link) {
+ rte_errno = EINVAL;
+ return -EINVAL;
+ }
+ /* Only one queue can be linked to a single-linked port */
+ if (port->single_linked_q != NULL) {
+ if (port->single_linked_q == queue)
+ return 0;
+ rte_errno = EINVAL;
+ return -EINVAL;
+ }
+ /* A single-link queue can only be linked to one port */
+ if (queue->serving_ports_mask != 0) {
rte_errno = EINVAL;
return -EINVAL;
}
port->single_linked_q = queue;
port->single_linked_ring = queue->q_ring;
} else {
- if (!queue->is_single_link &&
- queue->schedule_type == RTE_SCHED_TYPE_ATOMIC &&
+ /* A single-link queue cannot be linked to a regular port */
+ if (queue->is_single_link) {
+ rte_errno = EINVAL;
+ return -EINVAL;
+ }
+
+ /* Convert application priority to internal priority level (0 or 1) */
+ if (has_prio)
+ prio = (app_prio < RTE_EVENT_DEV_PRIORITY_NORMAL) ? 0 : 1;
+ else
+ prio = SLB_MAX_PRIORITIES - 1;
+
+ /* Check if queue is already mapped to this port at requested priority */
+ for (int pr = 0; pr < SLB_MAX_PRIORITIES; pr++) {
+ if (slb_port_find_queue_slot(port, pr, qid) >= 0) {
+ if (pr == prio)
+ return 0;
+ rte_errno = EINVAL;
+ return -EINVAL;
+ }
+ }
+
+ /* Ensure a free mapping slot is available at the requested priority */
+ slot = port->num_q[prio];
+ if (slot >= SLB_PORT_MAX_Q_MAP_PER_PRIO) {
+ rte_errno = ENOSPC;
+ return -ENOSPC;
+ }
+
+ /* Allocate per-port ring when linking an atomic queue */
+ if (queue->schedule_type == RTE_SCHED_TYPE_ATOMIC &&
queue->per_port_rings[port->id] == NULL) {
char ring_name[RTE_RING_NAMESIZE];
struct rte_event_ring *ring;
@@ -327,53 +368,23 @@ slb_port_link_one(struct slb_port *port, struct slb_queue *queue,
queue->per_port_rings[port->id] = ring;
}
- if (has_prio)
- prio = (app_prio < RTE_EVENT_DEV_PRIORITY_NORMAL) ? 0 : 1;
- else
- prio = SLB_MAX_PRIORITIES - 1;
-
- for (int pr = 0; pr < SLB_MAX_PRIORITIES; pr++) {
- int s = slb_port_find_queue_slot(port, pr, qid);
-
- if (s >= 0) {
- if (pr == prio)
- found = true;
- else
- slb_port_unlink_queue_from_prio(port, pr, s);
- break;
- }
- }
- if (!found) {
- uint16_t slot = port->num_q[prio];
- struct rte_event_ring *dq_ring =
- (!queue->is_single_link &&
- queue->schedule_type == RTE_SCHED_TYPE_ATOMIC) ?
- queue->per_port_rings[port->id] : queue->q_ring;
-
- if (slot >= SLB_PORT_MAX_Q_MAP_PER_PRIO) {
- rte_errno = ENOSPC;
- return -ENOSPC;
- }
- port->q_entries[prio][slot].dq_ring = dq_ring;
- port->q_entries[prio][slot].queue = queue;
- port->q_entries[prio][slot].queue_id = qid;
- port->q_entries[prio][slot].schedule_type =
- queue->schedule_type;
- port->q_entries[prio][slot].is_single_link =
- queue->is_single_link;
- port->num_q[prio]++;
- }
+ dq_ring = (queue->schedule_type == RTE_SCHED_TYPE_ATOMIC) ?
+ queue->per_port_rings[port->id] : queue->q_ring;
- rte_atomic_fetch_or_explicit(&port->active_queues,
- (1ULL << qid),
- rte_memory_order_release);
+ port->q_entries[prio][slot].dq_ring = dq_ring;
+ port->q_entries[prio][slot].queue = queue;
+ port->q_entries[prio][slot].queue_id = qid;
+ port->q_entries[prio][slot].schedule_type = queue->schedule_type;
+ port->num_q[prio]++;
}
+ /* Add newly linked port to the queue's serving ports mask */
queue->serving_ports_mask |= (1ULL << port->id);
queue->num_serving_ports = rte_popcount64(queue->serving_ports_mask);
return 0;
}
+/* Unlink a queue from a port */
static int
slb_port_unlink_one(struct slb_port *port, struct slb_queue *queue, uint8_t qid)
{
@@ -386,7 +397,7 @@ slb_port_unlink_one(struct slb_port *port, struct slb_queue *queue, uint8_t qid)
}
port->single_linked_q = NULL;
port->single_linked_ring = NULL;
- } else {
+ } else { /* Unlink queue by finding its slot */
for (int prio = 0; prio < SLB_MAX_PRIORITIES; prio++) {
int s = slb_port_find_queue_slot(port, prio, qid);
@@ -396,21 +407,17 @@ slb_port_unlink_one(struct slb_port *port, struct slb_queue *queue, uint8_t qid)
break;
}
}
- if (!found) {
+ if (!found) { /* No link found */
rte_errno = EINVAL;
return -EINVAL;
}
+ /* For atomic queues, free the per-port ring if it exists */
if (queue->schedule_type == RTE_SCHED_TYPE_ATOMIC &&
queue->per_port_rings[port->id] != NULL) {
rte_event_ring_free(queue->per_port_rings[port->id]);
queue->per_port_rings[port->id] = NULL;
}
-
- rte_atomic_fetch_and_explicit(&port->active_queues,
- ~(1ULL << qid),
- rte_memory_order_release);
- port->remap_input_q = true;
}
queue->serving_ports_mask &= ~(1ULL << port->id);
@@ -418,6 +425,7 @@ slb_port_unlink_one(struct slb_port *port, struct slb_queue *queue, uint8_t qid)
return 0;
}
+/* Link all queues to a port at specific priorities */
static int
slb_port_link(struct rte_eventdev *dev, void *p, const uint8_t queues[],
const uint8_t priorities[], uint16_t num)
@@ -439,6 +447,7 @@ slb_port_link(struct rte_eventdev *dev, void *p, const uint8_t queues[],
return num;
}
+/* Unlink all queues from a port */
static int
slb_port_unlink(struct rte_eventdev *dev, void *p, uint8_t queues[],
uint16_t num)
@@ -458,6 +467,7 @@ slb_port_unlink(struct rte_eventdev *dev, void *p, uint8_t queues[],
return num;
}
+/* Get information about the event device */
static void
slb_info_get(struct rte_eventdev *dev __rte_unused,
struct rte_event_dev_info *info)
@@ -467,7 +477,7 @@ slb_info_get(struct rte_eventdev *dev __rte_unused,
.max_event_queues = SLB_MAX_QUEUES,
.max_event_queue_flows = SLB_MAX_FLOWS,
.max_event_queue_priority_levels = SLB_MAX_PRIORITIES,
- .max_event_priority_levels = SLB_MAX_PRIORITIES,
+ .max_event_priority_levels = 1,
.max_event_ports = SLB_MAX_PORTS,
.max_event_port_dequeue_depth = SLB_MAX_PORT_DEQUEUE_DEPTH,
.max_event_port_enqueue_depth = SLB_MAX_PORT_ENQUEUE_DEPTH,
@@ -487,13 +497,13 @@ slb_info_get(struct rte_eventdev *dev __rte_unused,
};
}
+/* Configure the event device */
static int
slb_configure(const struct rte_eventdev *dev)
{
struct slb_evdev *slb = slb_pmd_priv(dev);
const struct rte_event_dev_config *conf = &dev->data->dev_conf;
int32_t min_max_in_flight;
- uint32_t q_size;
if (conf->nb_events_limit <= 0 ||
conf->nb_events_limit > SLB_MAX_EVENTS) {
@@ -503,36 +513,60 @@ slb_configure(const struct rte_eventdev *dev)
return -EINVAL;
}
+ /* Validate max event ports and queues that SLB can support */
+ if (conf->nb_event_ports == 0 ||
+ conf->nb_event_ports > SLB_MAX_PORTS) {
+ SLB_LOG_DP_LINE(INFO,
+ "nb_event_ports (%u) invalid, max is %d",
+ conf->nb_event_ports, SLB_MAX_PORTS);
+ return -EINVAL;
+ }
+
+ if (conf->nb_event_queues == 0 ||
+ conf->nb_event_queues > SLB_MAX_QUEUES) {
+ SLB_LOG_DP_LINE(INFO,
+ "nb_event_queues (%u) invalid, max is %d",
+ conf->nb_event_queues, SLB_MAX_QUEUES);
+ return -EINVAL;
+ }
+
slb->num_ports = conf->nb_event_ports;
slb->num_queues = conf->nb_event_queues;
+ /* Compute minimum credits required to avoid deadlock */
min_max_in_flight = conf->nb_event_ports * SLB_PORT_MAX_CREDITS;
if (min_max_in_flight > SLB_MAX_EVENTS)
min_max_in_flight = SLB_MAX_EVENTS;
- q_size = rte_align32pow2((uint32_t)RTE_MAX(conf->nb_events_limit,
- min_max_in_flight));
- if (q_size > SLB_MAX_EVENTS)
- q_size = SLB_MAX_EVENTS;
-
- slb->max_inflight = (int32_t)q_size;
+ /* Set credit pool size to max of nb_events_limit and minimum required to avoid deadlock */
+ slb->max_inflight = RTE_MAX(conf->nb_events_limit, min_max_in_flight);
+ slb->migration_cooldown_cycles =
+ (rte_get_timer_hz() * SLB_MIGRATION_COOLDOWN_MS) / 1000;
return 0;
}
+/* Start the event device */
static int
slb_start(struct rte_eventdev *dev)
{
struct slb_evdev *slb = slb_pmd_priv(dev);
+ /* Make sure all queues are set up and have at least one linked port */
for (int i = 0; i < slb->num_queues; i++) {
- if (slb->queues[i].serving_ports_mask == 0)
- SLB_LOG_DP_LINE(INFO,
- "Queue %d has no linked ports on start", i);
+ if (slb->queues[i].q_ring == NULL ||
+ slb->queues[i].serving_ports_mask == 0) {
+ SLB_LOG_DP_LINE(ERR,
+ "Queue %d is not configured or has no linked ports",
+ i);
+ return -ENOLINK;
+ }
}
+ /* No events in flight */
rte_atomic_store_explicit(&slb->credits_on_loan, 0,
rte_memory_order_relaxed);
+ /* Reset port states */
for (int i = 0; i < slb->num_ports; i++) {
struct slb_port *port = &slb->ports[i];
port->inflight_credits = 0;
@@ -544,6 +578,7 @@ slb_start(struct rte_eventdev *dev)
return 0;
}
+/* Empty the event queue */
static void
slb_drain_ring(uint8_t dev_id, struct rte_event_ring *ring,
eventdev_stop_flush_t flush, void *flush_arg)
@@ -553,6 +588,7 @@ slb_drain_ring(uint8_t dev_id, struct rte_event_ring *ring,
if (ring == NULL)
return;
+ /* Dequeue all events from the ring and call user callback if provided */
while (rte_event_ring_dequeue_burst(ring, &ev, 1, NULL)) {
if (flush != NULL)
flush(dev_id, ev, flush_arg);
@@ -560,6 +596,7 @@ slb_drain_ring(uint8_t dev_id, struct rte_event_ring *ring,
rte_ring_reset(&ring->r);
}
+/* Stop the event device */
static void
slb_stop(struct rte_eventdev *dev)
{
@@ -575,6 +612,7 @@ slb_stop(struct rte_eventdev *dev)
q->atomic_state != NULL) {
struct slb_atomic_state *astate = q->atomic_state;
+ /* Call user callback for each held event */
for (uint16_t h = 0; h < astate->hold_count; h++) {
if (flush != NULL)
flush(dev_id, astate->hold_buffer[h],
@@ -584,6 +622,7 @@ slb_stop(struct rte_eventdev *dev)
astate->next_port_rr = 0;
astate->migration.in_progress = false;
+ /* Reset flow to port mapping */
for (int f = 0; f < SLB_ATOMIC_FLOWS; f++) {
rte_atomic_store_explicit(
&astate->flow_map.flows[f].port_id,
@@ -594,6 +633,7 @@ slb_stop(struct rte_eventdev *dev)
memset(astate->port_stats, 0,
sizeof(astate->port_stats));
+ /* Drain all events from per-port rings */
for (uint16_t pid = 0; pid < slb->num_ports; pid++) {
if (q->per_port_rings[pid] != NULL)
slb_drain_ring(dev_id,
@@ -606,11 +646,12 @@ slb_stop(struct rte_eventdev *dev)
uint8_t st = rte_atomic_load_explicit(
&q->reorder_buffer[s].state,
rte_memory_order_relaxed);
-
+ /* Flush all ready events from the reorder buffer. */
if (st == SLB_REORDER_READY && flush != NULL)
flush(dev_id,
q->reorder_buffer[s].event,
flush_arg);
+ /* Mark freed slot as empty */
rte_atomic_store_explicit(
&q->reorder_buffer[s].state,
SLB_REORDER_EMPTY,
@@ -623,6 +664,7 @@ slb_stop(struct rte_eventdev *dev)
slb_drain_ring(dev_id, q->q_ring, flush, flush_arg);
}
+ /* Reset each port's state, history lists and queue mappings */
for (int i = 0; i < slb->num_ports; i++) {
struct slb_port *port = &slb->ports[i];
port->inflight_credits = 0;
@@ -635,21 +677,25 @@ slb_stop(struct rte_eventdev *dev)
port->next_q[p] = 0;
}
+ /* Reset the credit pool to all available */
rte_atomic_store_explicit(&slb->credits_on_loan, 0,
rte_memory_order_relaxed);
}
+/* Close the event device */
static int
slb_close(struct rte_eventdev *dev)
{
struct slb_evdev *slb = slb_pmd_priv(dev);
+ /* Release all resources acquired by the queues */
for (int i = 0; i < slb->num_queues; i++)
slb_queue_release(dev, i);
return 0;
}
+/* Dump the event device current state and statistics */
void
slb_dump(struct rte_eventdev *dev, FILE *f)
{
@@ -689,14 +735,12 @@ slb_dump(struct rte_eventdev *dev, FILE *f)
fprintf(f, "\n PORT STATISTICS & QUEUE LINKAGES:\n");
for (i = 0; i < slb->num_ports; i++) {
struct slb_port *p = &slb->ports[i];
- uint64_t aq = rte_atomic_load_explicit(&p->active_queues,
- rte_memory_order_acquire);
int linked_count = 0;
fprintf(f,
- " Port %2u (credits=%3d, active=0x%08" PRIx64
- ", loaded=%d, mig_in=%" PRIu64 ", mig_out=%" PRIu64 "): ",
- i, p->inflight_credits, aq, p->is_loaded,
+ " Port %2u (credits=%3d, loaded=%d"
+ ", mig_in=%" PRIu64 ", mig_out=%" PRIu64 "): ",
+ i, p->inflight_credits, p->is_loaded,
p->migrations_in, p->migrations_out);
if (p->is_single_linked) {
@@ -743,7 +787,8 @@ slb_dump(struct rte_eventdev *dev, FILE *f)
i, count, ring_size, q->serving_ports_mask,
q->num_serving_ports, q->schedule_type,
q->is_single_link);
- if (q->schedule_type == RTE_SCHED_TYPE_ATOMIC) {
+ if (!q->is_single_link &&
+ q->schedule_type == RTE_SCHED_TYPE_ATOMIC) {
if (q->atomic_state) {
fprintf(f,
" Atomic State: total_migrations=%" PRIu64
@@ -774,7 +819,8 @@ slb_dump(struct rte_eventdev *dev, FILE *f)
}
}
}
- } else if (q->schedule_type == RTE_SCHED_TYPE_ORDERED) {
+ } else if (!q->is_single_link &&
+ q->schedule_type == RTE_SCHED_TYPE_ORDERED) {
fprintf(f, " Ordered reorder_drain_idx=%u\n",
q->reorder_drain_idx);
}
diff --git a/drivers/event/slb/slb_evdev.h b/drivers/event/slb/slb_evdev.h
index bf262903cd..c01290e539 100644
--- a/drivers/event/slb/slb_evdev.h
+++ b/drivers/event/slb/slb_evdev.h
@@ -16,8 +16,8 @@
#define SLB_PMD_NAME RTE_STR(event_slb)
#define SLB_MAX_PORTS (64)
-#define SLB_MAX_PORT_DEQUEUE_DEPTH (128)
-#define SLB_MAX_PORT_ENQUEUE_DEPTH (128)
+#define SLB_MAX_PORT_DEQUEUE_DEPTH (128u)
+#define SLB_MAX_PORT_ENQUEUE_DEPTH (128u)
#define SLB_MAX_QUEUES (64)
@@ -36,13 +36,13 @@
/* Maximum number of queues that can be linked to a port per priority */
#define SLB_PORT_MAX_Q_MAP_PER_PRIO (SLB_MAX_QUEUES)
-/* Number of priority levels supported by SLB. Configure 1..4. */
+/* Number of priority levels supported by SLB. Currently fixed to 2 */
#ifndef SLB_MAX_PRIORITIES
#define SLB_MAX_PRIORITIES 2
#endif
-#if (SLB_MAX_PRIORITIES < 1) || (SLB_MAX_PRIORITIES > 4)
-#error "SLB_MAX_PRIORITIES must be between 1 and 4"
+#if (SLB_MAX_PRIORITIES != 2)
+#error "SLB_MAX_PRIORITIES must be set to 2 for now"
#endif
/* Total max mapping slots (per-prio * priorities) */
@@ -56,8 +56,7 @@ struct slb_port_q_entry {
struct slb_queue *queue;
uint8_t queue_id;
uint8_t schedule_type;
- bool is_single_link;
- uint8_t pad[5];
+ uint8_t pad[6];
};
#define SLB_PORT_HISTORY_SIZE 256
@@ -139,15 +138,13 @@ struct __rte_cache_aligned slb_port {
uint16_t pending_releases;
int32_t inflight_credits;
int32_t new_event_threshold;
- RTE_ATOMIC(uint64_t) active_queues;
uint16_t num_q[SLB_MAX_PRIORITIES];
uint16_t next_q[SLB_MAX_PRIORITIES];
bool is_single_linked;
bool implicit_release;
bool credits_drained;
- bool remap_input_q;
bool is_loaded; /* Port marked loaded for atomic flow migration */
- uint8_t pad_port[3];
+ uint8_t pad_port[4];
struct rte_event_ring *single_linked_ring;
struct slb_evdev *slb;
@@ -196,7 +193,7 @@ struct slb_queue {
rte_spinlock_t reorder_lock;
};
-/* Limited by the size of 'serving_ports_mask' and 'active_queues' bitmasks */
+/* Limited by the size of 64-bit port and queue bitmasks */
static_assert(SLB_MAX_PORTS <= 64, "Max compile-time port count exceeded");
static_assert(SLB_MAX_QUEUES <= 64, "Max compile-time queue count exceeded");
@@ -208,6 +205,7 @@ struct slb_evdev {
struct slb_queue queues[SLB_MAX_QUEUES];
uint8_t num_queues;
int32_t max_inflight;
+ uint64_t migration_cooldown_cycles;
alignas(RTE_CACHE_LINE_SIZE) RTE_ATOMIC(int32_t) credits_on_loan;
};
@@ -262,25 +260,4 @@ extern int event_slb_logtype;
#define SLB_LOG_DP_PORT_LINE(level, port_id, fmt, ...) \
SLB_LOG_DP_LINE(level, "<Port %d> " fmt, port_id, ## __VA_ARGS__)
-/* Obtain serving port masks for the specified queue and set corresponding
- * active queue bits for all served ports.
- */
-static inline void
-slb_port_wakeup_queues(struct slb_evdev *slb, struct slb_queue *queue)
-{
- uint8_t queue_id = queue - slb->queues;
- uint64_t port_mask = queue->serving_ports_mask;
- struct slb_port *port;
- uint8_t port_id;
-
- while (port_mask) {
- port_id = rte_ctz64(port_mask);
- port = &slb->ports[port_id];
- rte_atomic_fetch_or_explicit(&port->active_queues,
- (1ULL << queue_id),
- rte_memory_order_release);
- port_mask &= ~(1ULL << port_id);
- }
-}
-
#endif /* _SLB_EVDEV_H_ */
diff --git a/drivers/event/slb/slb_event.c b/drivers/event/slb/slb_event.c
index c44deba905..3d0436e4a0 100644
--- a/drivers/event/slb/slb_event.c
+++ b/drivers/event/slb/slb_event.c
@@ -11,7 +11,7 @@
#include "slb_evdev.h"
-/* Acquire credits for new events */
+/* Port Acquiring credits for new events. Serve from local cache if possible. */
static bool
slb_port_acquire_credits(struct slb_evdev *slb, struct slb_port *port,
int32_t credits)
@@ -34,34 +34,38 @@ slb_port_acquire_credits(struct slb_evdev *slb, struct slb_port *port,
available = slb->max_inflight - total_inflight;
required_credits = RTE_MAX(abs(remaining_credits), SLB_PORT_MIN_CREDITS);
+ /* Pool does not have enough credits */
if (available < required_credits)
return false;
+ /* Take credits from the pool */
new_total_inflight =
rte_atomic_fetch_add_explicit(&slb->credits_on_loan,
required_credits,
rte_memory_order_relaxed) +
required_credits;
-
+ /* Check if the new total exceeds the maximum allowed */
if (unlikely(new_total_inflight > slb->max_inflight)) {
rte_atomic_fetch_sub_explicit(&slb->credits_on_loan,
required_credits,
rte_memory_order_relaxed);
return false;
}
-
+ /* Update the port's credit counts*/
port->inflight_credits += required_credits;
port->inflight_credits -= credits;
return true;
}
-/* Return accumulated credits (Terminated events) */
+/* Port returning credits (Terminated events). Keep local cache.
+ * If local cache exceeds the threshold, return excess credits to the pool
+ */
void
slb_port_return_credits(struct slb_evdev *slb, struct slb_port *port,
int32_t credits)
{
- port->inflight_credits += credits;
+ port->inflight_credits += credits; /* Local credit cache */
if (unlikely(port->inflight_credits >= SLB_PORT_MAX_CREDITS)) {
int32_t return_credits = port->inflight_credits -
@@ -74,6 +78,7 @@ slb_port_return_credits(struct slb_evdev *slb, struct slb_port *port,
}
}
+/* Port enqueue statistics */
static void
slb_port_enqueue_stats(struct slb_port *port, uint16_t num_new,
uint16_t num_forward, uint16_t num_release)
@@ -98,7 +103,7 @@ slb_port_queue_enqueue_burst_stats(struct slb_port *source_port,
#define REORDER_BURST_SZ 64
-/* Drain in-order contiguous ready events from ordered queue's reorder buffer */
+/* Drain contiguous ready events from ordered queue's reorder buffer */
void
slb_reorder_drain(struct slb_evdev *slb, struct slb_queue *queue)
{
@@ -114,8 +119,8 @@ slb_reorder_drain(struct slb_evdev *slb, struct slb_queue *queue)
struct slb_reorder_entry *buf = queue->reorder_buffer;
uint8_t state;
- /* Drain from current drain index till hitting an empty slot.
- * Released slots need no processing except marking them empty.
+ /* Drain events from the current drain index till an empty slot.
+ * Released event slots need no processing except marking them as empty.
*/
while ((state = rte_atomic_load_explicit(&buf[drain_idx].state,
rte_memory_order_acquire)) !=
@@ -123,30 +128,27 @@ slb_reorder_drain(struct slb_evdev *slb, struct slb_queue *queue)
if (state == SLB_REORDER_RELEASED) {
rte_atomic_store_explicit(&buf[drain_idx].state,
SLB_REORDER_EMPTY,
- rte_memory_order_release);
+ rte_memory_order_relaxed);
drain_idx = (drain_idx + 1) & mask;
continue;
}
- /* Gather contiguous READY events destined for same target_qid */
+ /* Gather contiguous READY events destined to same target_qid */
struct rte_event burst[REORDER_BURST_SZ];
uint16_t burst_size = 0;
uint8_t target_qid = buf[drain_idx].event.queue_id;
- while (burst_size < REORDER_BURST_SZ &&
- rte_atomic_load_explicit(&buf[drain_idx].state,
- rte_memory_order_acquire) ==
- SLB_REORDER_READY &&
- buf[drain_idx].event.queue_id == target_qid) {
+ do {
burst[burst_size++] = buf[drain_idx].event;
rte_atomic_store_explicit(&buf[drain_idx].state,
SLB_REORDER_EMPTY,
- rte_memory_order_release);
+ rte_memory_order_relaxed);
drain_idx = (drain_idx + 1) & mask;
- }
-
- if (burst_size == 0)
- break;
+ } while (burst_size < REORDER_BURST_SZ &&
+ rte_atomic_load_explicit(&buf[drain_idx].state,
+ rte_memory_order_acquire) ==
+ SLB_REORDER_READY &&
+ buf[drain_idx].event.queue_id == target_qid);
/* Target queue is sized to max_inflight and can never be full */
rte_event_ring_enqueue_burst(slb->queues[target_qid].q_ring,
@@ -157,13 +159,14 @@ slb_reorder_drain(struct slb_evdev *slb, struct slb_queue *queue)
rte_spinlock_unlock(&queue->reorder_lock);
}
-/* Hash flow ID to atomic flow slot index */
+/* Hash 20-bit flow ID into atomic flow slot index (SLB_ATOMIC_FLOWS) */
static inline uint32_t
slb_flow_hash(uint32_t flow_id)
{
return (flow_id * 2654435761U) & SLB_ATOMIC_FLOWS_MASK;
}
+/* Find the lightest flow on a given port to migrate */
static int32_t
slb_find_lightest_flow_on_port(struct slb_queue *queue, uint8_t port_id,
uint64_t now, uint64_t cooldown_cycles)
@@ -187,12 +190,14 @@ slb_find_lightest_flow_on_port(struct slb_queue *queue, uint8_t port_id,
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;
+ count = prod - cons; /* Number of atomic events in the per port ring */
memset(counts, 0, sizeof(counts));
if (count > 0) {
+ /* Skip over the ring header and point to events */
struct rte_event *ring_events = (struct rte_event *)(r + 1);
+ /* Count events for each flow in per port queue */
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);
@@ -201,9 +206,11 @@ slb_find_lightest_flow_on_port(struct slb_queue *queue, uint8_t port_id,
}
}
+ /* Find the flow with the minimum number of events */
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) {
+ /* Skip flows that have been migrated recently */
if (astate->flow_last_migrated[f] != 0 &&
(now - astate->flow_last_migrated[f]) < cooldown_cycles)
continue;
@@ -218,16 +225,19 @@ slb_find_lightest_flow_on_port(struct slb_queue *queue, uint8_t port_id,
return best_flow_idx;
}
+/* Check if port is loaded and migration is necessary. Start the flow
+ * migration processes if required.
+ */
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 cooldown_cycles = slb->migration_cooldown_cycles;
uint64_t now;
- uint64_t hz;
- uint64_t cooldown_cycles;
int32_t flow_to_migrate;
+ /* Check if the port has crossed load threshold */
pstats->load_counter++;
if (pstats->load_counter < SLB_MIGRATION_LOAD_THRESHOLD)
return;
@@ -235,20 +245,21 @@ slb_check_and_trigger_migration(struct slb_evdev *slb, struct slb_queue *queue,
if (astate->migration.in_progress)
return;
+ /* Already marked for migration */
if (slb->ports[full_port_id].is_loaded)
return;
- if (pstats->flow_count <= 1)
+ if (pstats->flow_count <= 1) /* Not enough flows to migrate */
return;
now = rte_get_timer_cycles();
- hz = rte_get_timer_hz();
- cooldown_cycles = (hz * SLB_MIGRATION_COOLDOWN_MS) / 1000;
+ /* Port is still in cooldown period */
if (slb->ports[full_port_id].last_migrated_time != 0 &&
(now - slb->ports[full_port_id].last_migrated_time) < cooldown_cycles)
return;
+ /* Find flow to migrate*/
flow_to_migrate = slb_find_lightest_flow_on_port(
queue, full_port_id, now, cooldown_cycles);
if (flow_to_migrate < 0)
@@ -261,6 +272,7 @@ slb_check_and_trigger_migration(struct slb_evdev *slb, struct slb_queue *queue,
astate->migration.flow_idx = (uint32_t)flow_to_migrate;
}
+/* Find the target port (one with minimum flows) for flow migration */
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,
@@ -291,15 +303,18 @@ slb_find_migration_target_port(struct slb_evdev *slb, struct slb_queue *queue,
return target_port;
}
+/* Complete the flow migration process from original port to 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;
+ /* Switch the flow mapping to the target port */
rte_atomic_store_explicit(&astate->flow_map.flows[f_idx].port_id,
target_port, rte_memory_order_release);
+ /* Switch all events in hold buffer to the target port */
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) {
@@ -307,38 +322,45 @@ slb_complete_flow_migration(struct slb_evdev *slb, struct slb_atomic_state *asta
}
}
+ /* Update migration timestamps */
astate->flow_last_migrated[f_idx] = now;
slb->ports[mig_port].last_migrated_time = now;
slb->ports[target_port].last_migrated_time = now;
+ /* Update flow counts */
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++;
+ /* Update migration counters */
slb->ports[mig_port].flow_migrations++;
slb->ports[mig_port].migrations_out++;
slb->ports[target_port].migrations_in++;
+ /* Mark migration as complete */
astate->port_stats[mig_port].load_counter = 0;
slb->ports[mig_port].is_loaded = false;
astate->migration.in_progress = false;
}
+/* Assign a port to unmapped flow */
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;
+ int port_offset = rr_counter % num_serving_ports; /* Pick next port */
uint64_t temp_mask = serving_mask;
uint8_t assigned_port = 0;
+ /* Convert offset to port ID */
for (int j = 0; j <= port_offset; j++) {
assigned_port = rte_ctz64(temp_mask);
temp_mask &= (temp_mask - 1);
}
+ /* If next port in RR selection is loaded then find an unloaded port */
if (slb->ports[assigned_port].is_loaded) {
uint64_t cand_mask = serving_mask;
@@ -355,14 +377,14 @@ slb_assign_unmapped_flow_port(struct slb_evdev *slb, struct slb_atomic_state *as
return assigned_port;
}
-/* If port queue is empty for an atomic queue, pull events from main atomic queue to fill
- * up port queues.
+/* If per port atomic queue is empty, pull events from the main atomic queue to fill
+ * up all port queues. Handle migration and hold buffer processing.
*/
static void
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];
+ struct rte_event rcv_events[SLB_MAX_PORT_DEQUEUE_DEPTH];
uint16_t n_deq;
uint16_t i;
uint64_t full_ports_mask = 0;
@@ -373,28 +395,27 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
if (unlikely(astate == NULL))
return;
+ /* Only one thread can refill at a time */
if (!rte_spinlock_trylock(&astate->refill_lock))
return;
- /* 0. Detect migration in progress before hold buffer processing */
+ /* 0. Handle 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;
+ /* Check if it is safe to migrate. i.e. no pending events */
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);
+ slb->migration_cooldown_cycles);
if (target_port != -1)
slb_complete_flow_migration(
slb, astate, mig_port, target_port, now);
- else {
+ else { /* If no suitable target port found, cancel migration */
astate->port_stats[mig_port].load_counter = 0;
slb->ports[mig_port].is_loaded = false;
astate->migration.in_progress = false;
@@ -402,10 +423,11 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
}
}
+ /* Add the migration port to the full ports mask to avoid scheduling events to it */
if (astate->migration.in_progress)
full_ports_mask |= (1ULL << astate->migration.port_id);
- /* 1. Process Hold Buffer first */
+ /* 1. Process Hold Buffer events */
if (astate->hold_count > 0) {
uint16_t retained = 0;
uint16_t count = astate->hold_count;
@@ -415,10 +437,12 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
uint8_t dest_port_id = astate->hold_ports[i];
uint16_t burst_len = 1;
+ /* Collect consecutive events for the same destination port */
while ((i + burst_len) < count &&
astate->hold_ports[i + burst_len] == dest_port_id)
burst_len++;
+ /* If destination port is full, retain the event in hold buffer*/
if (full_ports_mask & (1ULL << dest_port_id)) {
if (retained != i) {
memmove(&astate->hold_buffer[retained],
@@ -433,15 +457,18 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
continue;
}
+ /* Enqueue events to the destination port */
uint16_t enqueued = rte_event_ring_enqueue_burst(
queue->per_port_rings[dest_port_id],
&astate->hold_buffer[i], burst_len, NULL);
astate->port_stats[dest_port_id].q_enqueued += enqueued;
+ /* If enqueue was successful then decrement the load counter */
if (enqueued > 0 &&
astate->port_stats[dest_port_id].load_counter > 0)
astate->port_stats[dest_port_id].load_counter--;
+ /* If enqueue was not successful, retain the event in hold buffer */
if (enqueued < burst_len) {
uint16_t rem = burst_len - enqueued;
uint16_t start_idx = i + enqueued;
@@ -456,6 +483,7 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
rem);
}
retained += rem;
+ /* If no events were enqueued, trigger migration */
if (enqueued == 0)
slb_check_and_trigger_migration(
slb, queue, astate, dest_port_id);
@@ -465,7 +493,11 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
}
astate->hold_count = retained;
- if (astate->hold_count > 0) {
+ /* Return without pulling events from main atomic queue
+ * if hold buffer does not have enough free space.
+ */
+ if (SLB_HOLD_BUFFER_SIZE - astate->hold_count <
+ 3 * RTE_DIM(rcv_events)) {
rte_spinlock_unlock(&astate->refill_lock);
return;
}
@@ -475,14 +507,13 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
serving_mask = queue->serving_ports_mask;
num_serving_ports = queue->num_serving_ports;
- if (num_serving_ports == 0) {
- rte_spinlock_unlock(&astate->refill_lock);
- return;
- }
-
for (int iter = 0; iter < 2; iter++) {
+ if (SLB_HOLD_BUFFER_SIZE - astate->hold_count <
+ RTE_DIM(rcv_events))
+ break;
+
n_deq = rte_event_ring_dequeue_burst(
- queue->q_ring, master_events, RTE_DIM(master_events), NULL);
+ queue->q_ring, rcv_events, RTE_DIM(rcv_events), NULL);
if (n_deq == 0)
break;
@@ -494,14 +525,19 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
struct rte_event sorted_events[SLB_MAX_PORT_DEQUEUE_DEPTH];
uint64_t batch_ports_mask = 0;
+ /* Dequeue events from main atomic queue, sort events,
+ * enqueue events to destination port queues. If destination port queue
+ * is full, retain events in hold buffer
+ */
for (i = 0; i < n_deq; i++) {
- struct rte_event *ev = &master_events[i];
+ struct rte_event *ev = &rcv_events[i];
uint32_t flow_idx = slb_flow_hash(ev->flow_id);
int32_t dest_port_id =
rte_atomic_load_explicit(
&astate->flow_map.flows[flow_idx].port_id,
rte_memory_order_acquire);
+ /* If flow is unmapped, map it to a port */
if (unlikely(dest_port_id == -1)) {
dest_port_id = slb_assign_unmapped_flow_port(
slb, astate, serving_mask, num_serving_ports);
@@ -514,6 +550,7 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
dest_ports[i] = (uint8_t)dest_port_id;
uint64_t p_bit = (1ULL << dest_port_id);
+ /* Count events for each destination port */
if (!(batch_ports_mask & p_bit)) {
batch_ports_mask |= p_bit;
port_counts[dest_port_id] = 1;
@@ -525,6 +562,7 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
uint16_t current_offset = 0;
uint64_t pmask = batch_ports_mask;
+ /* Calculate offsets for each destination port */
while (pmask != 0) {
uint8_t p = rte_ctz64(pmask);
@@ -534,10 +572,11 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
current_offset += port_counts[p];
}
+ /* Sort events by destination port */
for (i = 0; i < n_deq; i++) {
uint8_t p = dest_ports[i];
- sorted_events[insert_pos[p]++] = master_events[i];
+ sorted_events[insert_pos[p]++] = rcv_events[i];
}
pmask = batch_ports_mask;
@@ -548,40 +587,35 @@ slb_atomic_refill(struct slb_port *port, struct slb_queue *queue)
pmask &= (pmask - 1);
+ /* If destination port is full, retain events in hold buffer */
if (full_ports_mask & (1ULL << p)) {
- uint16_t space = SLB_HOLD_BUFFER_SIZE - astate->hold_count;
- uint16_t to_copy = RTE_MIN(p_count, space);
-
- if (to_copy > 0) {
- rte_memcpy(&astate->hold_buffer[astate->hold_count],
- &sorted_events[p_start],
- to_copy * sizeof(struct rte_event));
- memset(&astate->hold_ports[astate->hold_count], p, to_copy);
- astate->hold_count += to_copy;
- }
+ rte_memcpy(&astate->hold_buffer[astate->hold_count],
+ &sorted_events[p_start],
+ p_count * sizeof(struct rte_event));
+ memset(&astate->hold_ports[astate->hold_count], p, p_count);
+ astate->hold_count += p_count;
continue;
}
+ /* Enqueue events to the destination port */
uint16_t enqueued = rte_event_ring_enqueue_burst(
queue->per_port_rings[p], &sorted_events[p_start], p_count, NULL);
astate->port_stats[p].q_enqueued += enqueued;
+ /* If all events were enqueued successfully, decrement load counter */
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);
-
+ /* Mark the destination port as full and retain remaining events */
full_ports_mask |= (1ULL << p);
- if (to_copy > 0) {
- rte_memcpy(&astate->hold_buffer[astate->hold_count],
- &sorted_events[p_start + enqueued],
- to_copy * sizeof(struct rte_event));
- memset(&astate->hold_ports[astate->hold_count], p, to_copy);
- astate->hold_count += to_copy;
- }
+ rte_memcpy(&astate->hold_buffer[astate->hold_count],
+ &sorted_events[p_start + enqueued],
+ rem * sizeof(struct rte_event));
+ memset(&astate->hold_ports[astate->hold_count], p, rem);
+ astate->hold_count += rem;
+ /* If not a single event was enqueued, check for migration */
if (enqueued == 0)
slb_check_and_trigger_migration(
slb, queue, astate, p);
@@ -605,6 +639,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
if (unlikely(events_len == 0))
return 0;
+ /* If operation types are not known, count them */
if (!op_types_known) {
for (uint16_t i = 0; i < events_len; i++) {
switch (events[i].op) {
@@ -621,7 +656,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
}
}
- /* Check credit threshold for NEW events */
+ /* Check credit threshold if there are NEW events */
if (num_new > 0) {
if (unlikely(rte_atomic_load_explicit(&slb->credits_on_loan,
rte_memory_order_relaxed) >
@@ -633,32 +668,38 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
return 0;
}
+ /* Return credits for RELEASE events */
if (num_release > 0)
slb_port_return_credits(slb, source_port, num_release);
+ /* Check for release underflow and balance pending releases */
cur_pending = source_port->pending_releases;
RTE_VERIFY(num_forward + num_release <= cur_pending);
+ source_port->pending_releases = cur_pending - (num_forward + num_release);
slb_port_enqueue_stats(source_port, num_new, num_forward, num_release);
- /* Unified zero-copy fast path: when port has no in-flight ordered/atomic events
- * and no release ops.
+ /* Unified zero-copy fast path: when burst does not have past ordered/atomic
+ * events being forwarded and there are no RELEASE events. This means no
+ * history list entries are present.
*/
if (likely(source_port->history_avail == SLB_PORT_HISTORY_MAX_TOKENS &&
num_release == 0)) {
uint8_t q0 = events[0].queue_id;
uint16_t burst_size = 1;
+ /* Collect events going to the same queue in a burst */
while (burst_size < events_len &&
events[burst_size].queue_id == q0)
burst_size++;
+ /* Enqueue the entire burst if going to the same queue */
if (likely(burst_size == events_len)) {
rte_event_ring_enqueue_burst(slb->queues[q0].q_ring,
events, events_len, NULL);
slb_port_queue_enqueue_burst_stats(source_port, q0,
events_len);
- } else {
+ } else { /* Sort events by destination queue */
uint8_t dest_queues[SLB_MAX_PORT_ENQUEUE_DEPTH];
uint16_t queue_counts[SLB_MAX_QUEUES];
uint16_t insert_pos[SLB_MAX_QUEUES];
@@ -701,6 +742,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
sorted_events[insert_pos[q]++] = events[k];
}
+ /* Enqueue events to their respective queues */
qmask = batch_queues_mask;
while (qmask != 0) {
uint8_t q = rte_ctz64(qmask);
@@ -716,15 +758,12 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
}
}
- if (num_forward > 0)
- source_port->pending_releases -= num_forward;
-
SLB_LOG_DP_PORT_LINE(DEBUG, source_port->id,
"%d events accepted.", events_len);
return events_len;
}
- /* General path for bursts with in-flight ordered/atomic events or releases */
+ /* General path for bursts with any mix of events */
uint64_t ordered_drain_mask = 0;
uint16_t freed_count = 0;
uint16_t head = source_port->history_head;
@@ -733,14 +772,19 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
const struct rte_event *enq_events[SLB_MAX_PORT_ENQUEUE_DEPTH];
uint16_t queue_counts[SLB_MAX_QUEUES];
uint64_t batch_queues_mask = 0;
+ bool has_history = (source_port->history_avail <
+ SLB_PORT_HISTORY_MAX_TOKENS);
for (uint16_t idx = 0; idx < events_len; idx++) {
const struct rte_event *ev = &events[idx];
- uint8_t h_idx = (ev->op != RTE_EVENT_OP_NEW) ? ev->impl_opaque : 0;
+ /* History list index 0 is used by NEW, parallel, and single-link events */
+ uint8_t h_idx = (has_history && ev->op != RTE_EVENT_OP_NEW) ?
+ ev->impl_opaque : 0;
struct slb_history_entry *hist =
(h_idx != 0 && source_port->history[h_idx].active) ?
&source_port->history[h_idx] : NULL;
+ /* Ordered event is being forwarded/released. */
if (hist != NULL && hist->sched_type == RTE_SCHED_TYPE_ORDERED) {
struct slb_queue *orig_q = &slb->queues[hist->orig_queue_id];
@@ -748,26 +792,32 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
uint16_t slot = hist->seq_slot;
if (ev->op == RTE_EVENT_OP_RELEASE) {
+ /* Release the reorder buffer entry */
rte_atomic_store_explicit(
&orig_q->reorder_buffer[slot].state,
SLB_REORDER_RELEASED,
rte_memory_order_release);
} else {
+ /* Put event in reorder buffer */
orig_q->reorder_buffer[slot].event = *ev;
rte_atomic_store_explicit(
&orig_q->reorder_buffer[slot].state,
SLB_REORDER_READY,
rte_memory_order_release);
}
+ /* Mark the ordered queue for draining */
ordered_drain_mask |= (1ULL << hist->orig_queue_id);
}
+ /* Free the history entry */
hist->active = false;
hist->next_free = head;
head = h_idx;
freed_count++;
slb_port_queue_enqueue_stats(source_port, ev->queue_id);
} else if (ev->op == RTE_EVENT_OP_RELEASE) {
+ /* Atomic or parallel event being released */
if (hist != NULL) {
+ /* Count atomic events being returned */
if (hist->sched_type == RTE_SCHED_TYPE_ATOMIC) {
struct slb_queue *orig_q =
&slb->queues[hist->orig_queue_id];
@@ -776,6 +826,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
orig_q->atomic_state->port_stats[ep].q_returned++;
}
}
+ /* Free the history entry */
hist->active = false;
hist->next_free = head;
head = h_idx;
@@ -783,7 +834,9 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
}
slb_port_queue_enqueue_stats(source_port, ev->queue_id);
} else {
+ /* Atomic/Parallel event being forwarded*/
if (hist != NULL) {
+ /* Count atomic events being returned */
if (hist->sched_type == RTE_SCHED_TYPE_ATOMIC) {
struct slb_queue *orig_q =
&slb->queues[hist->orig_queue_id];
@@ -792,6 +845,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
orig_q->atomic_state->port_stats[ep].q_returned++;
}
}
+ /* Free the history entry */
hist->active = false;
hist->next_free = head;
head = h_idx;
@@ -805,6 +859,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
enq_events[n_enq] = ev;
n_enq++;
+ /* Count the number of events going to each queue */
if (!(batch_queues_mask & q_bit)) {
batch_queues_mask |= q_bit;
queue_counts[qid] = 1;
@@ -814,9 +869,10 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
}
}
- if (n_enq > 0) {
+ if (n_enq > 0) { /* There are events to enqueue */
if (likely((batch_queues_mask & (batch_queues_mask - 1)) == 0 &&
n_enq == events_len)) {
+ /* Entire batch is going to a single queue */
uint8_t q = rte_ctz64(batch_queues_mask);
rte_event_ring_enqueue_burst(slb->queues[q].q_ring,
@@ -830,6 +886,7 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
uint16_t current_offset = 0;
uint64_t qmask = batch_queues_mask;
+ /* Calculate the offset for each queue */
while (qmask != 0) {
uint8_t q = rte_ctz64(qmask);
@@ -839,12 +896,14 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
current_offset += queue_counts[q];
}
+ /* Sort events by queue */
for (uint16_t k = 0; k < n_enq; k++) {
uint8_t q = dest_queues[k];
sorted_events[insert_pos[q]++] = *enq_events[k];
}
+ /* Enqueue events for each queue */
qmask = batch_queues_mask;
while (qmask != 0) {
uint8_t q = rte_ctz64(qmask);
@@ -861,11 +920,13 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
}
}
+ /* Update the history list head once after all frees*/
if (freed_count > 0) {
source_port->history_head = head;
source_port->history_avail += freed_count;
}
+ /* If events are added to reorder buffers, call the drain function */
while (ordered_drain_mask != 0) {
uint8_t qid = rte_ctz64(ordered_drain_mask);
@@ -873,14 +934,12 @@ slb_event_enqueue_burst_generic(struct slb_port *source_port,
slb_reorder_drain(slb, &slb->queues[qid]);
}
- if (num_forward + num_release > 0)
- source_port->pending_releases -= (num_forward + num_release);
-
SLB_LOG_DP_PORT_LINE(DEBUG, source_port->id,
"%d events accepted.", events_len);
return events_len;
}
+/* Enqueue burst of mix event type. (NEW, FORWARDED, RELEASED) */
uint16_t
slb_event_enqueue_burst(void *port, const struct rte_event events[],
uint16_t events_len)
@@ -891,6 +950,7 @@ slb_event_enqueue_burst(void *port, const struct rte_event events[],
false, 0, 0, 0);
}
+/* Enqueue burst of all new events */
uint16_t
slb_event_enqueue_new_burst(void *port, const struct rte_event events[],
uint16_t events_len)
@@ -901,6 +961,7 @@ slb_event_enqueue_new_burst(void *port, const struct rte_event events[],
true, events_len, 0, 0);
}
+/* Enqueue burst of all forwarded events */
uint16_t
slb_event_enqueue_forward_burst(void *port, const struct rte_event events[],
uint16_t events_len)
@@ -911,6 +972,9 @@ slb_event_enqueue_forward_burst(void *port, const struct rte_event events[],
true, 0, events_len, 0);
}
+/* Dequeue events from an ordered queue. Event's slot in the ring is
+ * used as its sequence number
+ */
static inline uint16_t
slb_ordered_ring_dequeue_burst(struct rte_event_ring *r,
struct rte_event *events,
@@ -936,6 +1000,7 @@ slb_ordered_ring_dequeue_burst(struct rte_event_ring *r,
return n;
}
+/* Create history entries for events dequeued from a queue_id of type sched_type */
static inline void
slb_port_record_dequeued_history(struct slb_port *port, struct rte_event events[],
uint16_t dequeued, uint8_t queue_id,
@@ -965,35 +1030,31 @@ slb_port_record_dequeued_history(struct slb_port *port, struct rte_event events[
port->history_avail -= dequeued;
}
+/* Dequeue events from the port */
static inline uint16_t
slb_port_dequeue(struct slb_port *port, struct rte_event events[],
uint16_t num)
{
uint16_t dequeued = 0;
+ /* If port is single-linked, it is simple dequeue from the linked queue */
if (port->is_single_linked) {
if (unlikely(port->single_linked_ring == NULL))
return 0;
dequeued = rte_event_ring_dequeue_burst(port->single_linked_ring,
events, num, NULL);
- for (uint16_t j = 0; j < dequeued; j++) {
- events[j].sched_type = RTE_SCHED_TYPE_ATOMIC;
+ for (uint16_t j = 0; j < dequeued; j++)
events[j].impl_opaque = 0;
- }
return dequeued;
}
- uint64_t active_queues = rte_atomic_load_explicit(
- &port->active_queues, rte_memory_order_acquire);
- if (!active_queues)
- return 0;
-
for (int pr = 0; pr < SLB_MAX_PRIORITIES; pr++) {
int slots = port->num_q[pr];
if (slots == 0)
- continue;
+ continue; /* No queues mapped at this priority */
+ /* Start from where we left off */
int start = port->next_q[pr];
if (unlikely(start >= slots))
@@ -1007,25 +1068,18 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
struct slb_port_q_entry *qent = &port->q_entries[pr][idx];
uint8_t queue_id = qent->queue_id;
-
- if (!((active_queues >> queue_id) & 1))
- continue;
-
struct rte_event_ring *ring_to_dq = qent->dq_ring;
if (unlikely(ring_to_dq == NULL))
continue;
uint8_t sched_type = qent->schedule_type;
- bool is_single_link = qent->is_single_link;
- bool is_ordered = !is_single_link &&
- (sched_type == RTE_SCHED_TYPE_ORDERED);
- bool is_atomic = !is_single_link &&
- (sched_type == RTE_SCHED_TYPE_ATOMIC);
+ bool is_ordered = (sched_type == RTE_SCHED_TYPE_ORDERED);
+ bool is_atomic = (sched_type == RTE_SCHED_TYPE_ATOMIC);
if (is_ordered || is_atomic) {
if (unlikely(port->history_avail == 0))
- continue;
+ continue; /* No available history entries */
uint16_t dequeue_num = RTE_MIN(num,
port->history_avail);
@@ -1035,11 +1089,12 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
dequeued = slb_ordered_ring_dequeue_burst(
ring_to_dq, events, dequeue_num,
&start_slot);
- } else {
+ } else { /* Atomic queue */
dequeued = rte_event_ring_dequeue_burst(
ring_to_dq, events, dequeue_num,
NULL);
if (dequeued == 0) {
+ /* Per port queue is empty, call for refill */
slb_atomic_refill(port, qent->queue);
dequeued = rte_event_ring_dequeue_burst(
ring_to_dq, events, dequeue_num,
@@ -1047,26 +1102,22 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
}
}
+ /* Create history entries for dequeued events */
if (dequeued > 0)
slb_port_record_dequeued_history(
port, events, dequeued, queue_id,
sched_type, start_slot,
ring_to_dq->r.mask);
- } else {
+ } else { /* Parallel queue */
dequeued = rte_event_ring_dequeue_burst(
ring_to_dq, events, num, NULL);
- if (is_single_link) {
- for (uint16_t j = 0; j < dequeued; j++) {
- events[j].sched_type =
- RTE_SCHED_TYPE_ATOMIC;
- events[j].impl_opaque = 0;
- }
- } else {
- for (uint16_t j = 0; j < dequeued; j++)
- events[j].impl_opaque = 0;
- }
+ for (uint16_t j = 0; j < dequeued; j++)
+ events[j].impl_opaque = 0;
}
+ /* If dequeued non-zero events successfully, move to the next queue
+ * for the next dequeue call and return
+ */
if (dequeued > 0) {
idx++;
port->next_q[pr] = (idx >= slots) ? 0 : idx;
@@ -1077,6 +1128,7 @@ slb_port_dequeue(struct slb_port *port, struct rte_event events[],
return 0;
}
+/* Process pending releases for the port */
static inline void
slb_port_process_implicit_release(struct slb_port *port, struct slb_evdev *slb,
uint16_t pending)
@@ -1084,10 +1136,14 @@ slb_port_process_implicit_release(struct slb_port *port, struct slb_evdev *slb,
uint16_t head;
uint64_t ordered_drain_mask = 0;
+ slb_port_return_credits(slb, port, pending);
+ port->pending_releases = 0;
+
if (port->is_single_linked ||
port->history_avail >= SLB_PORT_HISTORY_MAX_TOKENS)
- goto done;
+ return;
+ /* Process all history entries */
head = port->history_head;
for (int i = 1; i < SLB_PORT_HISTORY_SIZE; i++) {
struct slb_history_entry *hist = &port->history[i];
@@ -1095,6 +1151,7 @@ slb_port_process_implicit_release(struct slb_port *port, struct slb_evdev *slb,
if (!hist->active)
continue;
+ /* Ordered releases will free the reorder buffer slot*/
if (hist->sched_type == RTE_SCHED_TYPE_ORDERED) {
struct slb_queue *oq = &slb->queues[hist->orig_queue_id];
@@ -1106,6 +1163,7 @@ slb_port_process_implicit_release(struct slb_port *port, struct slb_evdev *slb,
ordered_drain_mask |= (1ULL << hist->orig_queue_id);
}
} else if (hist->sched_type == RTE_SCHED_TYPE_ATOMIC) {
+ /* Atomic releases are counted as returned events */
struct slb_queue *oq = &slb->queues[hist->orig_queue_id];
if (oq->atomic_state) {
@@ -1114,6 +1172,7 @@ slb_port_process_implicit_release(struct slb_port *port, struct slb_evdev *slb,
oq->atomic_state->port_stats[ep].q_returned++;
}
}
+ /* Free the history entries */
hist->active = false;
hist->next_free = head;
head = (uint16_t)i;
@@ -1121,18 +1180,16 @@ slb_port_process_implicit_release(struct slb_port *port, struct slb_evdev *slb,
}
port->history_head = head;
+ /* If there are ordered releases, call the reorder drain function */
while (ordered_drain_mask != 0) {
uint8_t qid = rte_ctz64(ordered_drain_mask);
ordered_drain_mask &= ~(1ULL << qid);
slb_reorder_drain(slb, &slb->queues[qid]);
}
-
-done:
- slb_port_return_credits(slb, port, pending);
- port->pending_releases = 0;
}
+/* Eventdev dequeue burst function */
uint16_t
slb_event_dequeue_burst(void *p, struct rte_event events[], uint16_t num,
uint64_t wait __rte_unused)
@@ -1142,6 +1199,7 @@ slb_event_dequeue_burst(void *p, struct rte_event events[], uint16_t num,
uint16_t pending = port->pending_releases;
uint16_t n;
+ /* Handle pending releases */
if (port->implicit_release && pending > 0)
slb_port_process_implicit_release(port, slb, pending);
diff --git a/drivers/event/slb/slb_selftest.c b/drivers/event/slb/slb_selftest.c
index 776b6d6ab4..4a072b811b 100644
--- a/drivers/event/slb/slb_selftest.c
+++ b/drivers/event/slb/slb_selftest.c
@@ -362,12 +362,6 @@ test_ordered_forward(void)
cons_rx, rx[i].u64);
return -1;
}
- if (rx[i].sched_type != RTE_SCHED_TYPE_ATOMIC) {
- SLB_LOG_LINE(ERR,
- "Single-link dequeue sched_type expected %u, got %u",
- RTE_SCHED_TYPE_ATOMIC, rx[i].sched_type);
- return -1;
- }
cons_rx++;
rx[i].op = RTE_EVENT_OP_RELEASE;
}
@@ -422,12 +416,6 @@ test_single_link(void)
for (i = 0; i < 16; i++) {
if (rx[i].u64 != (uint64_t)(5000 + i))
return -1;
- if (rx[i].sched_type != RTE_SCHED_TYPE_ATOMIC) {
- SLB_LOG_LINE(ERR,
- "Single-link dequeue sched_type expected %u, got %u",
- RTE_SCHED_TYPE_ATOMIC, rx[i].sched_type);
- return -1;
- }
rx[i].op = RTE_EVENT_OP_RELEASE;
}
rte_event_enqueue_burst(g_dev_id, PORT_CONS, rx, 16);
--
2.53.0
^ permalink raw reply related [flat|nested] 14+ messages in thread* Re: [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD
2026-11-07 1:30 [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Pravin Pathak
` (7 preceding siblings ...)
2026-11-07 1:30 ` [PATCH v1 8/8] event/slb: improve port linking, config checks and atomic refill Pravin Pathak
@ 2026-11-07 7:29 ` Jerin Jacob
2026-11-08 17:10 ` Stephen Hemminger
2026-11-09 2:30 ` Pravin Pathak
8 siblings, 2 replies; 14+ messages in thread
From: Jerin Jacob @ 2026-11-07 7:29 UTC (permalink / raw)
To: Pravin Pathak, mattias.ronnblom
Cc: dev, jerinj, mike.ximing.chen, bruce.richardson, thomas,
david.marchand, nipun.gupta, chenbox, tirthendu.sarkar,
Pravin Pathak
On Sat, Nov 7, 2026 at 7:00 AM Pravin Pathak <pravinpathakdev@gmail.com> wrote:
>
> From: Pravin Pathak <pravin.pathak.dev@gmail.com>
>
> This series introduces the Software Load Balancer (SLB) Eventdev PMD
> (`event_slb`) supporting parallel, atomic and ordered queues. It is
> a distributed software scheduler that scales well.
>
> Motivation
> ----------
> Existing software eventdev PMDs face trade-offs between centralized
> bottlenecks and feature coverage:
> - `event_sw` relies on a dedicated centralized scheduler consuming
> an extra CPU core. The single scheduling core also limits maximum
> throughput and it does not scale.
> - `event_dsw` distributes scheduling across worker cores without a
> central scheduler core, but does not support ordered queues and
> incurs per-port control ring messaging overhead during atomic flow
> migrations. It also does not support service priority.
>
> Design Overview
> ---------------
> `event_slb` is a distributed (`RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED`),
> maintenance-free (`RTE_EVENT_DEV_CAP_MAINTENANCE_FREE`) software eventdev
> PMD that supports Atomic, Ordered, Parallel, and Single-Link queues:
@mattias.ronnblom@ericcson.com - drivers/event/dsw/
@liangma@liangbit.com - drivers/event/opdl/
NO MAINTAINER for drivers/event/sw/
Hi Pravin,
Creating new flavours. of evendev SW drivers is a maintenance burden,
If this is close to DSW then please introduce mode in DSW driver and
have function pointer switches for fastpath,
Also claim should be backed by performance test values. Use
testeventdev test cases and share the numbers to prove one way or
another.
In summary, IMO, it better to take NEW SW PMD when
a)It operates completely different than DSW and SW and it is NOT
possible to update existing SW PMDs
b)Able to show better performance than existing drivers on the specific cases.
>
> 1. Queue-Centric Ring Architecture:
> - Parallel and Single-Link queues map directly to `rte_event_ring`
> instances (MP/MC for parallel, MP/SC for single-link), allowing
> zero-copy burst enqueue and dequeue without history tracking.
> - Ordered queues use the ring slot index at dequeue time as an implicit
> sequence number into a per-queue reorder buffer, draining contiguous
> completed events in order without global sequence counters. This
> makes ordered queue implementaion highly efficient performing close
> to parallel queues.
> - Atomic queues use a main MP/SC queue ring alongside per-port SP/SC
> rings and a 2048-entry flow-to-port affinity table. All cores
> served by the atomic queue are used for scheduling.
>
> 2. Lock-Free / Zero-Pause Atomic Flow Migration:
> - When an atomic queue refills per-port rings and detects persistent
> congestion on a destination port, SLB selects the lightest flow on
> that port (outside the migration cooldown window) and waits for
> queues in-flight events to drain naturally before switching the
> flow's port affinity and schedule hold-buffer entries.
> - Hold buffer is used for buffering events towards migrating ports
> while allowing serving other ports.
> - Implementaion avoids cross-port pause/unpause control rings while
> preserving strict per-flow ordering.
>
> 3. Credit-Based Backpressure & Priority Scheduling:
> - A global atomic credit pool with per-port local credit caches bounds
> total in-flight events (`nb_events_limit`) and enforces
> `new_event_threshold` with minimal atomic contention.
> - Ports support two strict service priority levels with
> round-robin scheduling within each priority level.
>
> 4. Observability & Testing:
> - Device and per-port/per-queue extended statistics (`xstats`),
> detailed `rte_event_dev_dump()` diagnostics, and a unit self-test
> suite (`eventdev_selftest_slb`) covering parallel, atomic, ordered,
> single-link, priority, backpressure, stop-flush, and multi-core
> atomic migration scenarios.
>
> 5. Future Plan:
> - Supporting Event priority
> - Optimizing atomic flow migration
>
> Pravin Pathak (8):
> event/slb: introduce software load balancer PMD
> event/slb: add support for ordered queues
> event/slb: add support for atomic queues
> event/slb: add dynamic flow migration for atomic queues
> event/slb: add extended statistics and diagnostics
> event/slb: add self-test and documentation
> event/slb: update resets and credit defaults
> event/slb: improve port linking, config checks and atomic refill
>
> MAINTAINERS | 6 +
> app/test/test_eventdev.c | 7 +
> doc/guides/eventdevs/features/slb.ini | 25 +
> doc/guides/eventdevs/index.rst | 1 +
> doc/guides/eventdevs/slb.rst | 61 ++
> drivers/event/meson.build | 1 +
> drivers/event/slb/meson.build | 21 +
> drivers/event/slb/slb_evdev.c | 904 ++++++++++++++++++
> drivers/event/slb/slb_evdev.h | 263 ++++++
> drivers/event/slb/slb_event.c | 1219 +++++++++++++++++++++++++
> drivers/event/slb/slb_selftest.c | 522 +++++++++++
> drivers/event/slb/slb_xstats.c | 347 +++++++
> 12 files changed, 3377 insertions(+)
> create mode 100644 doc/guides/eventdevs/features/slb.ini
> create mode 100644 doc/guides/eventdevs/slb.rst
> create mode 100644 drivers/event/slb/meson.build
> create mode 100644 drivers/event/slb/slb_evdev.c
> create mode 100644 drivers/event/slb/slb_evdev.h
> create mode 100644 drivers/event/slb/slb_event.c
> create mode 100644 drivers/event/slb/slb_selftest.c
> create mode 100644 drivers/event/slb/slb_xstats.c
>
> --
> 2.53.0
>
^ permalink raw reply [flat|nested] 14+ messages in thread* Re: [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD
2026-11-07 7:29 ` [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Jerin Jacob
@ 2026-11-08 17:10 ` Stephen Hemminger
2026-11-09 2:30 ` Pravin Pathak
1 sibling, 0 replies; 14+ messages in thread
From: Stephen Hemminger @ 2026-11-08 17:10 UTC (permalink / raw)
To: Jerin Jacob
Cc: Pravin Pathak, mattias.ronnblom, dev, jerinj, mike.ximing.chen,
bruce.richardson, thomas, david.marchand, nipun.gupta, chenbox,
tirthendu.sarkar, Pravin Pathak
On Sat, 7 Nov 2026 12:59:18 +0530
Jerin Jacob <jerinjacobk@gmail.com> wrote:
> > Design Overview
> > ---------------
> > `event_slb` is a distributed (`RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED`),
> > maintenance-free (`RTE_EVENT_DEV_CAP_MAINTENANCE_FREE`) software eventdev
> > PMD that supports Atomic, Ordered, Parallel, and Single-Link queues:
>
> @mattias.ronnblom@ericcson.com - drivers/event/dsw/
> @liangma@liangbit.com - drivers/event/opdl/
> NO MAINTAINER for drivers/event/sw/
>
> Hi Pravin,
> Creating new flavours. of evendev SW drivers is a maintenance burden,
> If this is close to DSW then please introduce mode in DSW driver and
> have function pointer switches for fastpath,
> Also claim should be backed by performance test values. Use
> testeventdev test cases and share the numbers to prove one way or
> another.
I ran AI review with Fable and the current AGENTS guidelines.
Short answer: it agrees with Jerin
Review: [PATCH v1 0/8] event/slb: introducing software load balancer
eventdev PMD
Applied cleanly on 26.11-rc2 (a13c105). Builds per-commit with
-Dwerror=true; bisect is clean. Line numbers below are in the
post-apply tree.
Cover / series-level
--------------------
Jerin's question (new PMD vs. a DSW mode) is the right one, and the
code supports it. The credit pool (slb_port_acquire_credits,
slb_port_return_credits, credits_on_loan, the max_inflight formula
in slb_configure, the RTE_VERIFY on pending_releases), the log
macros, and all of slb_xstats.c are DSW code with the prefix
changed; the comment at slb_evdev.h:30 is verbatim dsw_evdev.h:62.
What is new is the queue-side scheduling: atomic queues use a main
MP/SC ring plus per-port SP/SC rings filled by whichever port finds
its ring empty, under a per-queue spinlock; ordered queues use the
ring slot index as the sequence number into a per-queue reorder
buffer, drained under a second per-queue spinlock. Both are
serialised per queue, so "distributed" means "whichever core takes
the lock", and the atomic refill does an O(SLB_ATOMIC_FLOWS)+O(ring)
scan on every migration trigger. The scaling claim needs
testeventdev numbers against sw and dsw, as Jerin asked.
Patches 7 and 8 are fix-ups to patches 1-6 (patch 8 also deletes
code added by patch 6). Squash them; the bullet-list commit messages
describe the diff, not the change.
The cover letter says the self-test covers priority, backpressure,
and multi-core atomic migration. slb_selftest.c is single-threaded,
4 ports, no priorities, no backpressure, and never triggers a
migration. None of the concurrency paths below are exercised.
No release notes. Patch 6 says it adds a 26.11 entry and patch 7
says it edits it; neither touches doc/guides/rel_notes/. A new
driver after rc2 is 27.03 material anyway.
Patch 1: introduce software load balancer PMD
---------------------------------------------
Error:
- slb_evdev.c:781, slb_dump(): NULL dereference.
uint32_t count = rte_event_ring_count(q->q_ring);
The loop runs i < slb->num_queues, which is set by configure, but
q_ring is NULL for any queue not yet set up. rte_event_dev_dump()
after configure and before queue_setup segfaults. Guard with
if (q->q_ring == NULL) continue; (same loop at :782).
- slb_evdev.c:197, slb_queue_setup(): double free. q_ring is
assigned before the atomic/ordered allocations:
queue->q_ring = ring;
and the failure paths at :212 and :239 do
rte_event_ring_free(ring);
return -ENOMEM;
leaving queue->q_ring dangling. The next slb_queue_release()
(close, or re-setup) frees it again. Set queue->q_ring = NULL on
those paths, or assign q_ring only after all allocations succeed.
Warning:
- slb_evdev.c:167, slb_queue_setup() on an already-linked queue
frees q_ring and any per_port_rings via slb_queue_release(), but
the linked ports' q_entries[][].dq_ring (set at :374) and
single_linked_ring (:319) still point at the freed rings. The
API allows queue_setup on a stopped device without unlinking
first, so the next dequeue is a use-after-free. Either return
-EBUSY when serving_ports_mask != 0, or re-resolve the entries.
- slb_event.c:661, new_event_threshold is compared against
credits_on_loan, which includes every port's idle cache:
if (unlikely(rte_atomic_load_explicit(&slb->credits_on_loan,
rte_memory_order_relaxed) >
source_port->new_event_threshold))
return 0;
This is the DSW check, but patch 7 raises SLB_MIN_CREDIT_LOAN to
256 (DSW: 64), so a port that only dequeues and releases holds
256..511 credits forever (slb_port_return_credits keeps
SLB_PORT_MIN_CREDITS). With K terminating ports the floor on
credits_on_loan is 256*K regardless of events in flight. The
default threshold from slb_port_def_conf is min(4096,
nb_events_limit): nb_events_limit=1024 with 4 consumer ports, or
the default with 16 workers, rejects every NEW enqueue once the
caches fill, and nothing ever lowers them. Either go back to the
DSW quantum, or let the port use its own cache before applying
the global check (acquire first, check the pool only when
borrowing).
Info:
- slb_configure() logs its -EINVAL causes with SLB_LOG_DP_LINE(INFO,
...). Control path; use SLB_LOG_LINE(ERR, ...).
- drivers/event/meson.build:18: 'slb' added after 'sw'; list is
alphabetical.
Patch 2: add support for ordered queues
---------------------------------------
Error:
- slb_event.c:993 / :1021, reorder slot reuse. The sequence number
is the ring slot:
*start_slot = cons_head & r->r.mask;
h->seq_slot = (start_slot + j) & ring_mask;
but the ring slot is freed at dequeue (__rte_ring_update_tail,
:998) while the reorder slot is freed only when slb_reorder_drain
passes it. The two are only in step if the producer can never
advance r.size positions while a slot is outstanding. Credits do
not guarantee that: RELEASE returns the credit immediately
(slb_event.c:673, and :1139 for implicit release) even though the
slot stays RELEASED until drained. So: worker A dequeues slot S
and is slow; drain stops at S; other workers dequeue and release
(or implicit-release) r.size more events, all returning credits;
the producer wraps to slot S; worker B dequeues the new event with
seq_slot S. B's forward writes reorder_buffer[S] and drain
delivers it past A's event; when A then forwards, its event lands
in a slot drain has already passed and is delivered r.size events
late, and if A releases while B's event is still READY the state
is overwritten and B's event is lost with its credit. For
max_inflight=4096 r.size is 8192, i.e. one stalled worker plus a
few ms of traffic. Fix: hold the credit for an ordered event
until its slot is drained (return it in slb_reorder_drain when a
RELEASED slot is reclaimed, and skip ordered releases in the
:673/:1139 returns). Then occupied slots + ring entries <=
max_inflight < r.size and the ring cannot wrap onto a live slot.
- slb_event.c:114, lost wakeup in slb_reorder_drain():
if (!rte_spinlock_trylock(&queue->reorder_lock))
return;
The holder exits when it reads EMPTY at drain_idx (:125). A second
port that marks that very slot READY and then fails the trylock
returns too. The slot stays READY until some later forward or
release on the same ordered queue happens to call drain; nothing
in the dequeue path does. At the end of a burst the last events
are stranded in the reorder buffer. Re-check after unlock:
rte_spinlock_unlock(&queue->reorder_lock);
rte_atomic_thread_fence(rte_memory_order_seq_cst);
if (rte_atomic_load_explicit(&buf[drain_idx].state,
rte_memory_order_acquire) !=
SLB_REORDER_EMPTY)
goto retry; /* back to the trylock */
Warning:
- slb_event.c:987-998, slb_ordered_ring_dequeue_burst() calls
__rte_ring_move_cons_head(), __rte_ring_dequeue_elems() and
__rte_ring_update_tail() directly. No other driver uses the ring
private API; it is not a stable contract. If the slot index is
the design, add an exported helper to lib/ring (or
rte_event_ring) that returns the consumer head, in its own patch.
Same for slb_event.c:198, which assumes the data array follows
struct rte_ring at (r + 1).
Patch 3: add support for atomic queues
--------------------------------------
Warning:
- slb_event.c:826, :845, :1172 write q_returned from the consuming
port with a plain increment:
orig_q->atomic_state->port_stats[ep].q_returned++;
and slb_event.c:406 (patch 4) reads it from whichever core holds
the refill lock, with a plain load, to decide the flow mapping
can be switched. Single writer, but cross-core and load-bearing
(the enq==ret test is what makes migration order-preserving).
Make q_returned RTE_ATOMIC(uint64_t), store with
rte_memory_order_release, load with rte_memory_order_acquire so
the application's last access to the flow on the old port is
ordered before the new port can see events for it.
Info:
- slb_event.c:166:
return (flow_id * 2654435761U) & SLB_ATOMIC_FLOWS_MASK;
The multiplier is odd, so the low 11 bits of the product are a
permutation of the low 11 bits of flow_id; flow_id and
flow_id+2048 always collide. Fibonacci hashing takes the high
bits: (flow_id * 2654435761U) >> (32 - 11).
- slb_history_entry.flow_id is written at slb_event.c:1024 and
never read.
Patch 4: add dynamic flow migration for atomic queues
-----------------------------------------------------
Warning:
- struct slb_port fields is_loaded (slb_evdev.h:146),
last_migrated_time, flow_migrations, migrations_in,
migrations_out (:164-167) are per port but written under the
refill_lock of whichever atomic queue is migrating
(slb_event.c:269, :327-328, :337-339, :343, :420) and read by
every other atomic queue's refill (:249, :290, :364). Two atomic
queues sharing ports race on them: lost increments, and both
queues can pick the same port as migration source. Either make
them per (queue, port) inside slb_atomic_state (port_stats is
already there), or make them atomic. is_loaded also sits in the
port's hot cacheline next to inflight_credits and history_head,
so a remote refill write bounces that line under the owning
core.
Info:
- The hold buffer is per queue, not per port. Once one linked port
stops draining, SLB_HOLD_BUFFER_SIZE - 3*128 held events later
slb_atomic_refill returns at :501 without touching the main ring
and every other port on that queue starves. Worth stating in the
Limitations section of slb.rst.
Patch 5: add extended statistics and diagnostics
------------------------------------------------
Warning:
- slb_xstats.c:105 exposes port_%u_queue_%u_dequeued from
slb_port.queue_dequeued[], which nothing increments. Either count
it in slb_port_dequeue() or drop the stat.
Info:
- Leftovers: slb_port.total_dequeues, enqueue_calls (printed by
slb_dump, never incremented), credits_drained (only ever cleared),
flow_migrations (incremented, never read); SLB_MAX_FLOWS_BITS and
SLB_MAX_FLOWS_MASK unused.
Patch 6: add self-test and documentation
----------------------------------------
Warning:
- doc/guides/eventdevs/features/slb.ini:14 has
runtime_port_link = Y but slb_info_get() does not set
RTE_EVENT_DEV_CAP_RUNTIME_PORT_LINK. Patch 7 says it removed this
from the feature matrix; it removed it from the .rst only.
- slb.rst (as of patch 7) says "link up to eight queues per port at
each priority"; SLB_PORT_MAX_Q_MAP_PER_PRIO is SLB_MAX_QUEUES
(64).
- Self-test does not use TEST_ASSERT / unit_test_suite_runner and
never runs more than one lcore. At minimum add a multi-core
ordered test with one deliberately slow worker and an atomic test
that forces a migration (SLB_MIGRATION_LOAD_THRESHOLD is 16 full
per-port rings); both of the Errors above would show up.
^ permalink raw reply [flat|nested] 14+ messages in thread* Re: [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD
2026-11-07 7:29 ` [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD Jerin Jacob
2026-11-08 17:10 ` Stephen Hemminger
@ 2026-11-09 2:30 ` Pravin Pathak
2026-11-09 12:55 ` Jerin Jacob
1 sibling, 1 reply; 14+ messages in thread
From: Pravin Pathak @ 2026-11-09 2:30 UTC (permalink / raw)
To: Jerin Jacob
Cc: mattias.ronnblom, dev, jerinj, mike.ximing.chen, bruce.richardson,
thomas, david.marchand, nipun.gupta, chenbox, tirthendu.sarkar
On Sat, Nov 7, 2026 at 2:29 AM Jerin Jacob <jerinjacobk@gmail.com> wrote:
>
> On Sat, Nov 7, 2026 at 7:00 AM Pravin Pathak <pravinpathakdev@gmail.com> wrote:
> >
> > From: Pravin Pathak <pravin.pathak.dev@gmail.com>
> >
> > This series introduces the Software Load Balancer (SLB) Eventdev PMD
> > (`event_slb`) supporting parallel, atomic and ordered queues. It is
> > a distributed software scheduler that scales well.
> >
> > Motivation
> > ----------
> > Existing software eventdev PMDs face trade-offs between centralized
> > bottlenecks and feature coverage:
> > - `event_sw` relies on a dedicated centralized scheduler consuming
> > an extra CPU core. The single scheduling core also limits maximum
> > throughput and it does not scale.
> > - `event_dsw` distributes scheduling across worker cores without a
> > central scheduler core, but does not support ordered queues and
> > incurs per-port control ring messaging overhead during atomic flow
> > migrations. It also does not support service priority.
> >
> > Design Overview
> > ---------------
> > `event_slb` is a distributed (`RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED`),
> > maintenance-free (`RTE_EVENT_DEV_CAP_MAINTENANCE_FREE`) software eventdev
> > PMD that supports Atomic, Ordered, Parallel, and Single-Link queues:
>
> @mattias.ronnblom@ericcson.com - drivers/event/dsw/
> @liangma@liangbit.com - drivers/event/opdl/
> NO MAINTAINER for drivers/event/sw/
>
> Hi Pravin,
> Creating new flavours. of evendev SW drivers is a maintenance burden,
> If this is close to DSW then please introduce mode in DSW driver and
> have function pointer switches for fastpath,
> Also claim should be backed by performance test values. Use
> testeventdev test cases and share the numbers to prove one way or
> another.
>
> In summary, IMO, it better to take NEW SW PMD when
> a)It operates completely different than DSW and SW and it is NOT
> possible to update existing SW PMDs
> b)Able to show better performance than existing drivers on the specific cases.
>
This driver is architecturally completely different from DSW. And it addresses
the gap of scalable software eventdev that can support ordered queues and
service priority. The skeleton is borrowed from DSW instead of writing
a new one.
But the underlying architecture is completely different. SLB uses
queues instead of
directly enquing to destination ports. DSW is an enqueue time scheduler
whereas SLB is a dequeue time scheduler. Very basic mechanism is different.
SLB does atomic flow migration is based on the port queue associated with the
atomic queue being full. It does not compute CPU load but checks if core is
able to serve the queue at mapped priority and if not, flows are migrated.
SLB is maintenance free unlike DSW. Flow migration happens without any
core-to-core communication required. SLB architecture allows extension to
support event priorities and configurable number of service priorities.
I will make patches with AI review fixes. AI comparison is matching
the structure and not underlying architecture. So it shows similarity.
>
> >
> > 1. Queue-Centric Ring Architecture:
> > - Parallel and Single-Link queues map directly to `rte_event_ring`
> > instances (MP/MC for parallel, MP/SC for single-link), allowing
> > zero-copy burst enqueue and dequeue without history tracking.
> > - Ordered queues use the ring slot index at dequeue time as an implicit
> > sequence number into a per-queue reorder buffer, draining contiguous
> > completed events in order without global sequence counters. This
> > makes ordered queue implementaion highly efficient performing close
> > to parallel queues.
> > - Atomic queues use a main MP/SC queue ring alongside per-port SP/SC
> > rings and a 2048-entry flow-to-port affinity table. All cores
> > served by the atomic queue are used for scheduling.
> >
> > 2. Lock-Free / Zero-Pause Atomic Flow Migration:
> > - When an atomic queue refills per-port rings and detects persistent
> > congestion on a destination port, SLB selects the lightest flow on
> > that port (outside the migration cooldown window) and waits for
> > queues in-flight events to drain naturally before switching the
> > flow's port affinity and schedule hold-buffer entries.
> > - Hold buffer is used for buffering events towards migrating ports
> > while allowing serving other ports.
> > - Implementaion avoids cross-port pause/unpause control rings while
> > preserving strict per-flow ordering.
> >
> > 3. Credit-Based Backpressure & Priority Scheduling:
> > - A global atomic credit pool with per-port local credit caches bounds
> > total in-flight events (`nb_events_limit`) and enforces
> > `new_event_threshold` with minimal atomic contention.
> > - Ports support two strict service priority levels with
> > round-robin scheduling within each priority level.
> >
> > 4. Observability & Testing:
> > - Device and per-port/per-queue extended statistics (`xstats`),
> > detailed `rte_event_dev_dump()` diagnostics, and a unit self-test
> > suite (`eventdev_selftest_slb`) covering parallel, atomic, ordered,
> > single-link, priority, backpressure, stop-flush, and multi-core
> > atomic migration scenarios.
> >
> > 5. Future Plan:
> > - Supporting Event priority
> > - Optimizing atomic flow migration
> >
> > Pravin Pathak (8):
> > event/slb: introduce software load balancer PMD
> > event/slb: add support for ordered queues
> > event/slb: add support for atomic queues
> > event/slb: add dynamic flow migration for atomic queues
> > event/slb: add extended statistics and diagnostics
> > event/slb: add self-test and documentation
> > event/slb: update resets and credit defaults
> > event/slb: improve port linking, config checks and atomic refill
> >
> > MAINTAINERS | 6 +
> > app/test/test_eventdev.c | 7 +
> > doc/guides/eventdevs/features/slb.ini | 25 +
> > doc/guides/eventdevs/index.rst | 1 +
> > doc/guides/eventdevs/slb.rst | 61 ++
> > drivers/event/meson.build | 1 +
> > drivers/event/slb/meson.build | 21 +
> > drivers/event/slb/slb_evdev.c | 904 ++++++++++++++++++
> > drivers/event/slb/slb_evdev.h | 263 ++++++
> > drivers/event/slb/slb_event.c | 1219 +++++++++++++++++++++++++
> > drivers/event/slb/slb_selftest.c | 522 +++++++++++
> > drivers/event/slb/slb_xstats.c | 347 +++++++
> > 12 files changed, 3377 insertions(+)
> > create mode 100644 doc/guides/eventdevs/features/slb.ini
> > create mode 100644 doc/guides/eventdevs/slb.rst
> > create mode 100644 drivers/event/slb/meson.build
> > create mode 100644 drivers/event/slb/slb_evdev.c
> > create mode 100644 drivers/event/slb/slb_evdev.h
> > create mode 100644 drivers/event/slb/slb_event.c
> > create mode 100644 drivers/event/slb/slb_selftest.c
> > create mode 100644 drivers/event/slb/slb_xstats.c
> >
> > --
> > 2.53.0
> >
^ permalink raw reply [flat|nested] 14+ messages in thread
* Re: [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD
2026-11-09 2:30 ` Pravin Pathak
@ 2026-11-09 12:55 ` Jerin Jacob
2026-11-11 14:22 ` Sarkar, Tirthendu
0 siblings, 1 reply; 14+ messages in thread
From: Jerin Jacob @ 2026-11-09 12:55 UTC (permalink / raw)
To: Pravin Pathak
Cc: mattias.ronnblom, dev, jerinj, mike.ximing.chen, bruce.richardson,
thomas, david.marchand, nipun.gupta, chenbox, tirthendu.sarkar
On Mon, Nov 9, 2026 at 8:00 AM Pravin Pathak <pravinpathakdev@gmail.com> wrote:
>
> On Sat, Nov 7, 2026 at 2:29 AM Jerin Jacob <jerinjacobk@gmail.com> wrote:
> >
> > On Sat, Nov 7, 2026 at 7:00 AM Pravin Pathak <pravinpathakdev@gmail.com> wrote:
> > >
> > > From: Pravin Pathak <pravin.pathak.dev@gmail.com>
> > >
> > > This series introduces the Software Load Balancer (SLB) Eventdev PMD
> > > (`event_slb`) supporting parallel, atomic and ordered queues. It is
> > > a distributed software scheduler that scales well.
> > >
> > > Motivation
> > > ----------
> > > Existing software eventdev PMDs face trade-offs between centralized
> > > bottlenecks and feature coverage:
> > > - `event_sw` relies on a dedicated centralized scheduler consuming
> > > an extra CPU core. The single scheduling core also limits maximum
> > > throughput and it does not scale.
> > > - `event_dsw` distributes scheduling across worker cores without a
> > > central scheduler core, but does not support ordered queues and
> > > incurs per-port control ring messaging overhead during atomic flow
> > > migrations. It also does not support service priority.
> > >
> > > Design Overview
> > > ---------------
> > > `event_slb` is a distributed (`RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED`),
> > > maintenance-free (`RTE_EVENT_DEV_CAP_MAINTENANCE_FREE`) software eventdev
> > > PMD that supports Atomic, Ordered, Parallel, and Single-Link queues:
> >
> > @mattias.ronnblom@ericcson.com - drivers/event/dsw/
> > @liangma@liangbit.com - drivers/event/opdl/
> > NO MAINTAINER for drivers/event/sw/
> >
> > Hi Pravin,
> > Creating new flavours. of evendev SW drivers is a maintenance burden,
> > If this is close to DSW then please introduce mode in DSW driver and
> > have function pointer switches for fastpath,
> > Also claim should be backed by performance test values. Use
> > testeventdev test cases and share the numbers to prove one way or
> > another.
> >
> > In summary, IMO, it better to take NEW SW PMD when
> > a)It operates completely different than DSW and SW and it is NOT
> > possible to update existing SW PMDs
> > b)Able to show better performance than existing drivers on the specific cases.
> >
> This driver is architecturally completely different from DSW. And it addresses
> the gap of scalable software eventdev that can support ordered queues and
> service priority. The skeleton is borrowed from DSW instead of writing
> a new one.
> But the underlying architecture is completely different. SLB uses
> queues instead of
> directly enquing to destination ports. DSW is an enqueue time scheduler
> whereas SLB is a dequeue time scheduler. Very basic mechanism is different.
> SLB does atomic flow migration is based on the port queue associated with the
> atomic queue being full. It does not compute CPU load but checks if core is
> able to serve the queue at mapped priority and if not, flows are migrated.
> SLB is maintenance free unlike DSW. Flow migration happens without any
> core-to-core communication required. SLB architecture allows extension to
> support event priorities and configurable number of service priorities.
> I will make patches with AI review fixes. AI comparison is matching
> the structure and not underlying architecture. So it shows similarity.
Lets get the performance numbers for SLB vs DSW based on same test
case so that we can decide.
Performance and scalability will be the aspect to decide the usage of
YET ANOTHER SW driver.
> >
> > >
> > > 1. Queue-Centric Ring Architecture:
> > > - Parallel and Single-Link queues map directly to `rte_event_ring`
> > > instances (MP/MC for parallel, MP/SC for single-link), allowing
> > > zero-copy burst enqueue and dequeue without history tracking.
> > > - Ordered queues use the ring slot index at dequeue time as an implicit
> > > sequence number into a per-queue reorder buffer, draining contiguous
> > > completed events in order without global sequence counters. This
> > > makes ordered queue implementaion highly efficient performing close
> > > to parallel queues.
> > > - Atomic queues use a main MP/SC queue ring alongside per-port SP/SC
> > > rings and a 2048-entry flow-to-port affinity table. All cores
> > > served by the atomic queue are used for scheduling.
> > >
> > > 2. Lock-Free / Zero-Pause Atomic Flow Migration:
> > > - When an atomic queue refills per-port rings and detects persistent
> > > congestion on a destination port, SLB selects the lightest flow on
> > > that port (outside the migration cooldown window) and waits for
> > > queues in-flight events to drain naturally before switching the
> > > flow's port affinity and schedule hold-buffer entries.
> > > - Hold buffer is used for buffering events towards migrating ports
> > > while allowing serving other ports.
> > > - Implementaion avoids cross-port pause/unpause control rings while
> > > preserving strict per-flow ordering.
> > >
> > > 3. Credit-Based Backpressure & Priority Scheduling:
> > > - A global atomic credit pool with per-port local credit caches bounds
> > > total in-flight events (`nb_events_limit`) and enforces
> > > `new_event_threshold` with minimal atomic contention.
> > > - Ports support two strict service priority levels with
> > > round-robin scheduling within each priority level.
> > >
> > > 4. Observability & Testing:
> > > - Device and per-port/per-queue extended statistics (`xstats`),
> > > detailed `rte_event_dev_dump()` diagnostics, and a unit self-test
> > > suite (`eventdev_selftest_slb`) covering parallel, atomic, ordered,
> > > single-link, priority, backpressure, stop-flush, and multi-core
> > > atomic migration scenarios.
> > >
> > > 5. Future Plan:
> > > - Supporting Event priority
> > > - Optimizing atomic flow migration
> > >
> > > Pravin Pathak (8):
> > > event/slb: introduce software load balancer PMD
> > > event/slb: add support for ordered queues
> > > event/slb: add support for atomic queues
> > > event/slb: add dynamic flow migration for atomic queues
> > > event/slb: add extended statistics and diagnostics
> > > event/slb: add self-test and documentation
> > > event/slb: update resets and credit defaults
> > > event/slb: improve port linking, config checks and atomic refill
> > >
> > > MAINTAINERS | 6 +
> > > app/test/test_eventdev.c | 7 +
> > > doc/guides/eventdevs/features/slb.ini | 25 +
> > > doc/guides/eventdevs/index.rst | 1 +
> > > doc/guides/eventdevs/slb.rst | 61 ++
> > > drivers/event/meson.build | 1 +
> > > drivers/event/slb/meson.build | 21 +
> > > drivers/event/slb/slb_evdev.c | 904 ++++++++++++++++++
> > > drivers/event/slb/slb_evdev.h | 263 ++++++
> > > drivers/event/slb/slb_event.c | 1219 +++++++++++++++++++++++++
> > > drivers/event/slb/slb_selftest.c | 522 +++++++++++
> > > drivers/event/slb/slb_xstats.c | 347 +++++++
> > > 12 files changed, 3377 insertions(+)
> > > create mode 100644 doc/guides/eventdevs/features/slb.ini
> > > create mode 100644 doc/guides/eventdevs/slb.rst
> > > create mode 100644 drivers/event/slb/meson.build
> > > create mode 100644 drivers/event/slb/slb_evdev.c
> > > create mode 100644 drivers/event/slb/slb_evdev.h
> > > create mode 100644 drivers/event/slb/slb_event.c
> > > create mode 100644 drivers/event/slb/slb_selftest.c
> > > create mode 100644 drivers/event/slb/slb_xstats.c
> > >
> > > --
> > > 2.53.0
> > >
^ permalink raw reply [flat|nested] 14+ messages in thread
* RE: [PATCH v1 0/8] event/slb: introducing software load balancer eventdev PMD
2026-11-09 12:55 ` Jerin Jacob
@ 2026-11-11 14:22 ` Sarkar, Tirthendu
0 siblings, 0 replies; 14+ messages in thread
From: Sarkar, Tirthendu @ 2026-11-11 14:22 UTC (permalink / raw)
To: Jerin Jacob, Pravin Pathak
Cc: mattias.ronnblom@ericcson.com, dev@dpdk.org, jerinj@marvell.com,
mike.ximing.chen@intel.com, Richardson, Bruce,
thomas@monjalon.net, david.marchand@redhat.com,
nipun.gupta@amd.com, chenbox@nvidia.com
On Mon, Nov 9, 2026 at 6:25 PM Jerin Jacob <jerinjacobk@gmail.com> wrote:
> Lets get the performance numbers for SLB vs DSW based on same test
> case so that we can decide.
> Performance and scalability will be the aspect to decide the usage of
> YET ANOTHER SW driver.
Hi Pravin,
A few points on top of Stephen's review, plus some early numbers since
Jerin asked for SLB vs DSW data.
+1 to Stephen's two Errors in patch 2 (reorder slot reuse when RELEASE
returns credits before the slot drains, and the lost wakeup in
slb_reorder_drain()) and to the credit-threshold issue in patch 1.
Those look like the ones that can lose or strand events.
Additional comments:
1. slb_event.c, slb_event_enqueue_burst_generic(): the return value of
rte_event_ring_enqueue_burst() is ignored on every path and the
function always returns events_len. Correctness then depends
entirely on credits guaranteeing ring space. If that ever fails
(e.g. the credit-threshold or reorder issues above, or application
misuse), events are silently dropped and their credits leak. At
minimum, check the return value and report a short enqueue.
Because the burst is sorted by destination queue before enqueue,
a clean prefix count isn't possible after a partial failure, so
this probably needs a space check up front or an explicit design
statement that rings can never fill.
2. RTE_VERIFY(num_forward + num_release <= cur_pending) in the
enqueue fast path aborts the whole process on application misuse
(forwarding more events than were dequeued on that port). Please
fail the enqueue and set rte_errno instead.
3. slb_configure(): max_inflight = RTE_MAX(nb_events_limit,
nb_ports * SLB_PORT_MAX_CREDITS), so the device can hold more
events than the application configured via nb_events_limit. If
that is intended (to avoid credit-cache deadlock), please document
it in slb.rst.
4. slb_queue_setup() ignores queue_conf->priority; priority comes only
from the link priority, yet dev_info reports
max_event_queue_priority_levels = 2. Either honour queue priority or
report 1 and document that priority is per link.
Performance
-----------
Preliminary data from dpdk-test-eventdev perf_queue, series applied on
v26.11-rc2, release build, Xeon 6505P (2x12C), physical cores only,
turbo on, default device settings, worker_deq_depth 16, 1024 flows.
Mpps, 1 synthetic producer, workers on socket 0:
stlist driver 1w 2w 4w 8w
p slb 79.7 71.5 44.5 34.0
p dsw 50.1 46.3 45.1 41.9
a slb 54.3 65.3 57.0 43.9
a dsw 50.3 47.3 46.7 40.5
a,a,a slb 20.8 16.9 22.9 26.7
a,a,a dsw 20.8 18.4 20.1 18.2
o slb 61.7 29.5 36.6 27.4
o sw 7.7 10.0 11.7 13.7
o,a slb 36.0 19.5 28.3 28.9
o,a sw 3.8 5.7 7.5 7.2
(event_dsw does not support ordered queues, so ordered rows are
against event_sw.)
Observations:
- SLB is ahead at low worker counts but loses throughput faster as
workers are added on single-stage parallel/atomic queues, which is
consistent with all workers contending on one MP/MC ring per queue.
- Ordered queues are well ahead of event_sw (the only other SW PMD with
ordered support), but SLB ordered throughput halves going from 1 to
2 workers (61.7 -> 29.5 Mpps, reproduced), so the "close to
parallel" claim only holds for a single worker (parallel: 79.7 ->
71.5). This looks like the serialized reorder drain.
- With workers spanning both sockets (16/20 workers) all drivers drop
sharply; SLB falls to 5-10 Mpps vs DSW 3-14 depending on stlist.
- Note that with multiple producers perf_queue creates a separate
queue set per producer, which splits SLB's per-queue contention; SLB
scales very well in that mode (4 producers, 8 workers, atomic:
~235 Mpps vs ~158 for DSW at worker_deq_depth 16), but that does not
reflect a pipeline where all traffic shares one atomic queue.
- order_queue passes with up to 16 workers. In that test SLB slows down
as workers are added: with 8 workers it sustained ~1.0 M events/s vs
~2.5 M events/s for event_sw (steady-state rate from the progress
counter, 20M events); with 1-2 workers SLB was faster than event_sw
(rough end-to-end timings, ~3.2 vs ~1.6-2.1 M events/s).
These are untuned runs. In the 1-producer table the SLB/DSW atomic and
a,a,a rows, DSW parallel and SLB ordered were run twice and agreed
within ~5%; the other rows are single runs. Treat them as indicative.
It would help if v2 included your own perf_queue/pipeline numbers with
the exact command lines, ideally including a shared-queue case and
workers on more than one socket.
Thanks,
Tirthendu
^ permalink raw reply [flat|nested] 14+ messages in thread