From ce06c7713d4d984843bcbc938b76355f25e94066 Mon Sep 17 00:00:00 2001 From: Cliff Burdick Date: Tue, 16 Jun 2026 21:30:41 +0000 Subject: [PATCH 1/3] Add eCPRI-over-Ethernet (EtherType 0xAEFE) as a new RX flow-match class in both the DPDK and ibverbs engines, alongside the existing standard UDP/IP and flex-item classes. Config: new FlowMatchType::ECPRI and EcpriMatch (optional message type + pc_id/rtc_id); selected by a match.ecpri map in YAML. Matching the identifier requires a msg_type (hardware constraint). DPDK: add_ecpri_flow() builds an ETH(0xAEFE)+RTE_FLOW_ITEM_TYPE_ECPRI pattern in a dedicated flow group (2) with its own group-0 jump rule and send-to-kernel fallback; validate_config() now enforces three-way mutual exclusivity of flow classes. ibverbs: a per-port mlx5 flex-parser node anchored at L2 on the eCPRI EtherType samples the common-header and message-body dwords, matched via misc_parameters_4; wired into static and dynamic flow installation and teardown. Signed-off-by: Cliff Burdick --- AGENTS.md | 4 +- docs/api-reference/configuration.md | 28 ++- docs/benchmarks/raw_benchmarking.md | 3 +- include/daqiri/types.h | 15 ++ src/common.cpp | 31 +++ src/engines/dpdk/daqiri_dpdk_engine.cpp | 129 +++++++++++- src/engines/dpdk/daqiri_dpdk_engine.h | 5 + src/engines/ibverbs/daqiri_ibverbs_engine.cpp | 186 +++++++++++++++++- src/engines/ibverbs/daqiri_ibverbs_engine.h | 25 +++ src/engines/ibverbs/mlx5_prm_min.h | 1 + 10 files changed, 410 insertions(+), 17 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 322c09bb..32bdb88f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -73,9 +73,9 @@ clang-format -style=file -i -fallback-style=none `EngineType` (`include/daqiri/types.h`) is resolved from `(stream_type, engine)`: `raw` defaults to `EngineType::DPDK`; `raw` + `engine: "ibverbs"` selects `EngineType::IBVERBS` (the MPRQ engine); `socket` + a `roce://` endpoint (or `engine: "ibverbs"`) selects `EngineType::RDMA`. The stream-aware `config_engine_from_string(str, stream_type)` overload encodes the `ibverbs`→`{IBVERBS for raw, RDMA for socket}` split. `EngineFactory` (in `engine.h`) is a singleton that instantiates the active engine. `daqiri_init(...)` resolves which engine to use from the `NetworkConfig` and then delegates everything through the `Engine` vtable. There is only ever **one** active `Engine` per process. -The default `dpdk` raw engine (`src/engines/dpdk/`, `DpdkEngine`) programs RX steering, send-to-kernel fallbacks (`flow_isolation: true`), and `tx_eth_src` TX offloads via DPDK RTE Flow during `daqiri_init()`. Standard UDP/IP and flex-item RX flows use separate flow groups; `validate_config()` rejects mixed flow classes per interface, duplicate `flex_item_id` values, and unknown `action.id` / `flex_item_id` values before NIC programming. Flex-item parser handles are created per `(port, flex_item_id)` (scoped per interface). All programmed `rte_flow` rules and flex-item handles are tracked and destroyed in order on shutdown, init failure, and engine teardown (programmed flows → flex items → group-0 ETH jump rules). See `docs/benchmarks/raw_benchmarking.md` (Flow programming smoke test) for manual verification steps. +The default `dpdk` raw engine (`src/engines/dpdk/`, `DpdkEngine`) programs RX steering, send-to-kernel fallbacks (`flow_isolation: true`), and `tx_eth_src` TX offloads via DPDK RTE Flow during `daqiri_init()`. Standard UDP/IP (group 3), flex-item (group 1) and eCPRI-over-Ethernet (group 2, EtherType `0xAEFE`, via `RTE_FLOW_ITEM_TYPE_ECPRI` matching message type and pc_id/rtc_id) RX flows use separate flow groups; `validate_config()` rejects mixing these flow classes per interface, duplicate `flex_item_id` values, and unknown `action.id` / `flex_item_id` values before NIC programming. Flex-item parser handles are created per `(port, flex_item_id)` (scoped per interface). All programmed `rte_flow` rules and flex-item handles are tracked and destroyed in order on shutdown, init failure, and engine teardown (programmed flows → flex items → group-0 ETH jump rules). See `docs/benchmarks/raw_benchmarking.md` (Flow programming smoke test) for manual verification steps. -The `ibverbs` raw engine (`src/engines/ibverbs/`, `IbverbsEngine`) drives a Mellanox/mlx5 Multi-Packet (striding) Receive Queue via **DevX** (`mlx5dv_devx_obj_create` against vendored PRM structs in `mlx5_prm_min.h`): a DevX CQ + striding RQ + TIR + `mlx5dv_dr` flow steering, with manual WQE/doorbell management and worker-driven cyclic refill. RX packets DMA strided into one pre-posted MR (host or GPU via `ibv_reg_dmabuf_mr`); a queue with >1 memory region instead uses a non-striding DevX *regular* RQ with multi-segment scatter WQEs for **physical** header-data split (header → CPU MR, payload → GPU MR). TX builds mlx5 send WQEs directly on a raw-packet QP's SQ (via `mlx5dv_init_obj`, bypassing `ibv_post_send`) from a slab of registered slots tracked by cyclic index counters, with NIC checksum offload and a `tx_eth_src` offload. It uses the libdpdk-free `daqiri::Ring`/`daqiri::ObjectPool` for the worker→app burst handoff (like the rdma engine — neither links DPDK) and drives the NIC through libibverbs/mlx5dv directly. Feature set: RX (MPRQ), TX, GPUDirect, physical/logical HDS, multi-queue 5-tuple flow steering with per-packet flow IDs, flex-item arbitrary-offset and IPv4-total-length flow matching (mlx5 flex parser / `misc_parameters_4`), per-packet RX hardware timestamps, accurate TX send scheduling (wait-on-time WAIT WQE), and GPU reorder/quantize. Because it uses the kernel netdev directly, `ensure_port_mtus` raises the netdev MTU at init to cover the configured frame size (jumbo frames silently drop on RX otherwise). Queues sharing a `cpu_core` are serviced round-robin by one poller thread. +The `ibverbs` raw engine (`src/engines/ibverbs/`, `IbverbsEngine`) drives a Mellanox/mlx5 Multi-Packet (striding) Receive Queue via **DevX** (`mlx5dv_devx_obj_create` against vendored PRM structs in `mlx5_prm_min.h`): a DevX CQ + striding RQ + TIR + `mlx5dv_dr` flow steering, with manual WQE/doorbell management and worker-driven cyclic refill. RX packets DMA strided into one pre-posted MR (host or GPU via `ibv_reg_dmabuf_mr`); a queue with >1 memory region instead uses a non-striding DevX *regular* RQ with multi-segment scatter WQEs for **physical** header-data split (header → CPU MR, payload → GPU MR). TX builds mlx5 send WQEs directly on a raw-packet QP's SQ (via `mlx5dv_init_obj`, bypassing `ibv_post_send`) from a slab of registered slots tracked by cyclic index counters, with NIC checksum offload and a `tx_eth_src` offload. It uses the libdpdk-free `daqiri::Ring`/`daqiri::ObjectPool` for the worker→app burst handoff (like the rdma engine — neither links DPDK) and drives the NIC through libibverbs/mlx5dv directly. Feature set: RX (MPRQ), TX, GPUDirect, physical/logical HDS, multi-queue 5-tuple flow steering with per-packet flow IDs, flex-item arbitrary-offset, IPv4-total-length and eCPRI-over-Ethernet (EtherType `0xAEFE`, message type + pc_id/rtc_id) flow matching (mlx5 flex parser / `misc_parameters_4`), per-packet RX hardware timestamps, accurate TX send scheduling (wait-on-time WAIT WQE), and GPU reorder/quantize. Because it uses the kernel netdev directly, `ensure_port_mtus` raises the netdev MTU at init to cover the configured frame size (jumbo frames silently drop on RX otherwise). Queues sharing a `cpu_core` are serviced round-robin by one poller thread. ### Zero-copy / BurstParams All packet data flows through `BurstParams`, a batch of packets. Only pointers are passed between NIC, DAQIRI internals, and the application — the caller reads directly from the buffers the NIC DMA'd into. **The caller must explicitly free bursts**; a missed free drains the pool and produces `NO_FREE_BURST_BUFFERS` / `NO_FREE_PACKET_BUFFERS` errors and NIC drops. See `docs/concepts.md` (Zero-Copy Ownership) and `docs/api-reference/cpp.md` (free-function call patterns). diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index 66debe41..1d5bc8d0 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -230,28 +230,42 @@ RX flows can also perform hardware VLAN pop or tunnel decapsulation before queue - type: `integer` - **`mask`**: 32-bit mask applied before matching (with flex items). - type: `integer` + - **`ecpri`**: eCPRI-over-Ethernet match (EtherType `0xAEFE`). Presence of this map selects + the eCPRI flow class; the EtherType is matched implicitly. Cannot be combined with UDP/IP + or flex-item matching. A flow with an empty `ecpri: {}` map matches all eCPRI frames. + - **`msg_type`**: eCPRI common-header message type (e.g. `0` = IQ data, `2` = real-time + control). Optional. + - type: `integer` + - **`pc_id`** / **`rtc_id`**: eCPRI message identifier (the 16-bit physical-channel ID for + message types 0/1, or real-time-control ID for type 2). `pc_id` and `rtc_id` are aliases + for the same field. Optional, but matching it requires a `msg_type` (the NIC needs a known + message type to locate the identifier in the eCPRI header). + - type: `integer` For Raw Ethernet (`stream_type: "raw"`), each flow rule is programmed into the NIC during `daqiri_init()`. If any rule cannot be installed, or the send-to-kernel fallback cannot be created when `flow_isolation: true`, initialization fails with a critical log and -`daqiri_init()` returns an error status. +`daqiri_init()` returns an error status. eCPRI matching is supported by both the `dpdk` +(via the mlx5 eCPRI flow item) and `ibverbs` (via an mlx5 flex-parser node anchored on the +eCPRI EtherType) engines. -A single RX interface must use either standard UDP/IP flows or flex-item flows, not both. -Both classes install conflicting DPDK group-0 jump rules, so only one is reachable when mixed. -`daqiri_init` rejects such configs with a clear error. +A single RX interface must use exactly one flow class — standard UDP/IP, flex-item, or eCPRI. +Each class installs its own DPDK group-0 jump rule, and these conflict when mixed, so only one +class is reachable per interface. `daqiri_init` rejects mixed configs with a clear error. Flex-item flows cannot be combined with VLAN/tunnel transform actions in v1. ### Flow Isolation `rx.flow_isolation:` — When `true`, only packets matching an explicit flow rule are delivered to the application. Static startup flows install send-to-kernel fallback rules per flow class -(standard or flex-item), so unmatched traffic in those classes is steered back to the Linux +(standard, flex-item, or eCPRI), so unmatched traffic in those classes is steered back to the Linux kernel. Queues-only configs can set `flow_isolation: true` and then install dynamic RX flows after `daqiri_init()`; the first dynamic RX flow installs the send-to-kernel fallback for that flow class (so unmatched control traffic such as ARP keeps reaching the kernel), and until a dynamic rule is added, application traffic is not delivered to DAQIRI RX queues. When `false`, -unmatched packets go to a default queue. Mixing standard and flex-item flow classes on one -interface is not supported, including across dynamic flow additions or within one dynamic batch. +unmatched packets go to a default queue. Mixing standard, flex-item, and eCPRI flow classes on +one interface is not supported, including across dynamic flow additions or within one dynamic +batch. - type: `boolean` - default: `false` diff --git a/docs/benchmarks/raw_benchmarking.md b/docs/benchmarks/raw_benchmarking.md index fa18426b..fa10a2a0 100644 --- a/docs/benchmarks/raw_benchmarking.md +++ b/docs/benchmarks/raw_benchmarking.md @@ -252,7 +252,8 @@ flow programming test. | Good NIC config | `daqiri_bench_raw_tx_rx.yaml` (filled placeholders, cabled NIC) | Init succeeds; RX and `tx_eth_src` flows programmed | | Dynamic RX flow config | `daqiri_example_dynamic_rx_flow.yaml` with `daqiri_example_dynamic_rx_flow` | Starts with `flow_isolation: true` and no `rx.flows`, drops unmatched traffic, then adds/deletes runtime UDP steering rules | | Bad queue ID | Copy `daqiri_bench_raw_tx_rx.yaml`, set `flows[0].action.id: 99` | Fails in `validate_config()` before EAL/NIC init with `references unknown RX queue` | -| Mixed flows (optional) | On one interface, add both a standard UDP/IP flow and a flex-item flow (see `rx.flex_items` in the [configuration reference](../api-reference/configuration.md)) | Fails in `validate_config()` with `mixes standard (UDP/IP) and flex-item` | +| Mixed flows (optional) | On one interface, add two of the three flow classes — standard UDP/IP, flex-item (see `rx.flex_items`), or eCPRI (`match.ecpri`) — in the [configuration reference](../api-reference/configuration.md) | Fails in `validate_config()` with `mixes standard (UDP/IP), flex-item and/or eCPRI` | +| eCPRI flow (optional) | On a cabled NIC, add a flow with `match: { ecpri: { msg_type: 0, pc_id: 1 } }` (see the [configuration reference](../api-reference/configuration.md)) | Init succeeds; eCPRI-over-Ethernet (EtherType 0xAEFE) frames matching the message type and pc_id steer to the flow's queue | ## Cap the transmit rate with packet pacing diff --git a/include/daqiri/types.h b/include/daqiri/types.h index 89feabf4..288c712f 100644 --- a/include/daqiri/types.h +++ b/include/daqiri/types.h @@ -739,9 +739,23 @@ struct FlexItemMatch { uint32_t mask_ = 0; }; +// eCPRI-over-Ethernet (EtherType 0xAEFE) RX flow match. The eCPRI EtherType is +// matched implicitly; the common-header message type and the message-body +// identifier (pc_id for message types 0/1, rtc_id for type 2) are matched only +// when their match_*_ flag is set. Matching the identifier requires a message +// type (the hardware needs a known message type to locate the body field), so +// match_id_ implies match_msg_type_. +struct EcpriMatch { + bool match_msg_type_ = false; + uint8_t msg_type_ = 0; // eCPRI common-header message type (RTE_ECPRI_MSG_TYPE_*) + bool match_id_ = false; + uint16_t id_ = 0; // pc_id (msg type 0/1) or rtc_id (msg type 2), at eCPRI offset 4 +}; + enum class FlowMatchType { IPV4_UDP, FLEX_ITEM, + ECPRI, }; struct FlowMatch { @@ -752,6 +766,7 @@ struct FlowMatch { in_addr_t ipv4_src_ = INADDR_ANY; in_addr_t ipv4_dst_ = INADDR_ANY; FlexItemMatch flex_item_match_; + EcpriMatch ecpri_match_; }; struct FlowConfig { std::string name_; diff --git a/src/common.cpp b/src/common.cpp index 5b2a3d18..a7cdaf56 100644 --- a/src/common.cpp +++ b/src/common.cpp @@ -888,6 +888,37 @@ bool YAML::convert::parse_flow_config( const YAML::Node match = flow_item["match"]; + // eCPRI-over-Ethernet match: selected by the presence of a `match.ecpri` map. + // Matches the eCPRI EtherType (0xAEFE) plus an optional common-header message + // type and message identifier (pc_id/rtc_id). Detected before the UDP/IP and + // flex-item paths because it is a distinct, mutually exclusive match class. + const YAML::Node ecpri_node = match["ecpri"]; + if (ecpri_node && ecpri_node.IsMap()) { + flow.match_.type_ = daqiri::FlowMatchType::ECPRI; + if (ecpri_node["msg_type"]) { + flow.match_.ecpri_match_.msg_type_ = + static_cast(ecpri_node["msg_type"].as() & 0xff); + flow.match_.ecpri_match_.match_msg_type_ = true; + } + // pc_id (msg type 0/1) and rtc_id (msg type 2) name the same 16-bit field. + const YAML::Node id_node = ecpri_node["pc_id"] ? ecpri_node["pc_id"] : ecpri_node["rtc_id"]; + if (id_node) { + flow.match_.ecpri_match_.id_ = id_node.as(); + flow.match_.ecpri_match_.match_id_ = true; + } + if (flow.match_.ecpri_match_.match_id_ && !flow.match_.ecpri_match_.match_msg_type_) { + DAQIRI_LOG_ERROR( + "eCPRI flow '{}' matches pc_id/rtc_id but no msg_type; matching the eCPRI message " + "identifier requires a msg_type", + flow.name_); + return false; + } + DAQIRI_LOG_INFO("Using eCPRI match: msg_type={} (matched={}), id={} (matched={})", + flow.match_.ecpri_match_.msg_type_, flow.match_.ecpri_match_.match_msg_type_, + flow.match_.ecpri_match_.id_, flow.match_.ecpri_match_.match_id_); + return true; + } + try { flow.match_.udp_src_ = match["udp_src"].as(); } catch (const std::exception& e) { diff --git a/src/engines/dpdk/daqiri_dpdk_engine.cpp b/src/engines/dpdk/daqiri_dpdk_engine.cpp index ed7ad450..b7bde54d 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.cpp +++ b/src/engines/dpdk/daqiri_dpdk_engine.cpp @@ -2682,11 +2682,12 @@ void DpdkEngine::initialize() { if (rx.hardware_timestamps_ && !calibrate_rx_timestamp_clock(intf.port_id_)) { return; } - // Standard (group 3) and flex-item (group 1) flows use separate DPDK flow - // groups with conflicting group-0 jump rules; validate_config() rejects mixed - // configs per interface. + // Standard (group 3), flex-item (group 1) and eCPRI (group 2) flows use + // separate DPDK flow groups with conflicting group-0 jump rules; + // validate_config() rejects mixing these classes per interface. bool has_standard_flows = false; bool has_flex_item_flows = false; + bool has_ecpri_flows = false; for (const auto& flow : rx.flows_) { DAQIRI_LOG_INFO("Adding RX flow {}", flow.name_); struct rte_flow* created = nullptr; @@ -2694,6 +2695,11 @@ void DpdkEngine::initialize() { created = add_flex_item_flow( intf.port_id_, flow.match_.flex_item_match_, flow.action_.id_, flow.id_); if (created != nullptr) { has_flex_item_flows = true; } + } else if (flow.match_.type_ == FlowMatchType::ECPRI) { + created = add_ecpri_flow(intf.port_id_, flow); + if (created != nullptr) { + has_ecpri_flows = true; + } } else { created = add_flow(intf.port_id_, flow); if (created != nullptr) { has_standard_flows = true; } @@ -2704,6 +2710,9 @@ void DpdkEngine::initialize() { if (intf.rx_.flow_isolation_) { if (has_standard_flows && !add_send_to_kernel_fallback(intf.port_id_, 3)) { return; } if (has_flex_item_flows && !add_send_to_kernel_fallback(intf.port_id_, 1)) { return; } + if (has_ecpri_flows && !add_send_to_kernel_fallback(intf.port_id_, 2)) { + return; + } } for (const auto& flow : tx.flows_) { @@ -3148,8 +3157,15 @@ bool DpdkEngine::validate_dynamic_rx_flow(int port, const FlowRuleConfig& flow) } return true; } + if (flow.match_.type_ == FlowMatchType::ECPRI) { + if (flow.match_.ecpri_match_.match_id_ && !flow.match_.ecpri_match_.match_msg_type_) { + DAQIRI_LOG_ERROR("Dynamic eCPRI RX flow matches pc_id/rtc_id but no msg_type"); + return false; + } + return true; + } - DAQIRI_LOG_ERROR("Dynamic RX flow must define an IPv4/UDP or flex-item match"); + DAQIRI_LOG_ERROR("Dynamic RX flow must define an IPv4/UDP, flex-item or eCPRI match"); return false; } @@ -3438,6 +3454,8 @@ Status DpdkEngine::create_dynamic_flow_legacy_locked(int port, if (cfg.match_.type_ == FlowMatchType::FLEX_ITEM) { rte_flow = add_flex_item_flow( port, cfg.match_.flex_item_match_, cfg.action_.id_, cfg.id_, false); + } else if (cfg.match_.type_ == FlowMatchType::ECPRI) { + rte_flow = add_ecpri_flow(port, cfg, false); } else { rte_flow = add_flow(port, cfg, false, &resource); } @@ -4953,6 +4971,92 @@ struct rte_flow* DpdkEngine::add_tx_flow(int port, const FlowConfig& cfg) { return flow; } +struct rte_flow* DpdkEngine::add_ecpri_flow(int port, const FlowConfig& cfg, bool track) { + struct rte_flow_attr attr; + struct rte_flow_item pattern[MAX_PATTERN_NUM]; + struct rte_flow_action action[MAX_ACTION_NUM]; + struct rte_flow* flow = NULL; + struct rte_flow_action_queue queue = {.index = cfg.action_.id_}; + struct rte_flow_action_mark mark = {.id = cfg.id_}; + struct rte_flow_error error; + struct rte_flow_item_eth eth_spec; + struct rte_flow_item_eth eth_mask; + struct rte_flow_item_ecpri ecpri_spec; + struct rte_flow_item_ecpri ecpri_mask; + int res; + + if (!ensure_eth_jump_rule(port, 2)) { + return nullptr; + } + + memset(pattern, 0, sizeof(pattern)); + memset(action, 0, sizeof(action)); + memset(&attr, 0, sizeof(struct rte_flow_attr)); + memset(ð_spec, 0, sizeof(struct rte_flow_item_eth)); + memset(ð_mask, 0, sizeof(struct rte_flow_item_eth)); + memset(&ecpri_spec, 0, sizeof(struct rte_flow_item_ecpri)); + memset(&ecpri_mask, 0, sizeof(struct rte_flow_item_ecpri)); + + action[0].type = RTE_FLOW_ACTION_TYPE_MARK; + action[0].conf = &mark; + action[1].type = RTE_FLOW_ACTION_TYPE_QUEUE; + action[1].conf = &queue; + action[2].type = RTE_FLOW_ACTION_TYPE_END; + + // Pin the eCPRI EtherType so the rule only fires on eCPRI-over-Ethernet frames. + eth_spec.hdr.ether_type = rte_cpu_to_be_16(RTE_ETHER_TYPE_ECPRI); + eth_mask.hdr.ether_type = 0xffff; + pattern[0].type = RTE_FLOW_ITEM_TYPE_ETH; + pattern[0].spec = ð_spec; + pattern[0].mask = ð_mask; + + // The mlx5 eCPRI parser matches the common-header dword (message type at the + // second byte, network-order mask 0x00ff0000) and the message-body dword + // (hdr.dummy[0], shared by type0.pc_id / type2.rtc_id; the identifier is the + // top 16 bits, mask 0xffff0000). Both are filled in network byte order. The + // PMD only matches the body when the common-header mask is non-zero, so + // pc_id/rtc_id matching requires a msg_type (enforced at config parse). + const EcpriMatch& em = cfg.match_.ecpri_match_; + if (em.match_msg_type_) { + ecpri_spec.hdr.common.u32 = rte_cpu_to_be_32(static_cast(em.msg_type_) << 16); + ecpri_mask.hdr.common.u32 = rte_cpu_to_be_32(0x00ff0000u); + } + if (em.match_id_) { + ecpri_spec.hdr.type0.pc_id = rte_cpu_to_be_16(em.id_); + ecpri_mask.hdr.type0.pc_id = rte_cpu_to_be_16(0xffff); + } + + uint8_t npat = 1; + if (em.match_msg_type_ || em.match_id_) { + pattern[npat].type = RTE_FLOW_ITEM_TYPE_ECPRI; + pattern[npat].spec = &ecpri_spec; + pattern[npat].mask = &ecpri_mask; + npat++; + } + pattern[npat].type = RTE_FLOW_ITEM_TYPE_END; + + attr.ingress = 1; + attr.priority = 1; // below the drop-all rule (priority 0) + attr.group = 2; + + res = rte_flow_validate(port, &attr, pattern, action, &error); + if (res != 0) { + DAQIRI_LOG_CRITICAL("Failed to validate eCPRI RX flow '{}' on port {}: {} ({})", cfg.name_, + port, error.message ? error.message : "unknown error", + rte_strerror(rte_errno)); + return nullptr; + } + + flow = rte_flow_create(port, &attr, pattern, action, &error); + if (flow == nullptr) { + DAQIRI_LOG_CRITICAL("Failed to create eCPRI RX flow '{}' on port {}: {} ({})", cfg.name_, port, + error.message ? error.message : "unknown error", rte_strerror(rte_errno)); + } else if (track) { + track_flow(static_cast(port), flow); + } + return flow; +} + bool DpdkEngine::add_send_to_kernel_fallback(int port, uint32_t group) { const uint64_t key = eth_jump_key(port, group); if (send_to_kernel_fallback_installed_.count(key) != 0) { return true; } @@ -5309,6 +5413,7 @@ bool DpdkEngine::validate_config() const { std::unordered_map flow_to_queue; bool has_standard_flows = false; bool has_flex_item_flows = false; + bool has_ecpri_flows = false; for (const auto& flow : intf.rx_.flows_) { if (flow_to_queue.find(flow.id_) != flow_to_queue.end()) { DAQIRI_LOG_ERROR("Duplicate flow ID {} on interface '{}'", flow.id_, intf.name_); @@ -5338,12 +5443,24 @@ bool DpdkEngine::validate_config() const { intf.name_); return false; } + } else if (flow.match_.type_ == FlowMatchType::ECPRI) { + has_ecpri_flows = true; + if (flow.match_.ecpri_match_.match_id_ && !flow.match_.ecpri_match_.match_msg_type_) { + DAQIRI_LOG_ERROR("eCPRI flow '{}' matches pc_id/rtc_id but no msg_type on interface '{}'", + flow.name_, intf.name_); + return false; + } } else { has_standard_flows = true; } - if (has_standard_flows && has_flex_item_flows) { + // Standard (group 3), flex-item (group 1) and eCPRI (group 2) flows install + // conflicting group-0 ETH jump rules, so only one class is allowed per + // interface. + if (static_cast(has_standard_flows) + static_cast(has_flex_item_flows) + + static_cast(has_ecpri_flows) > + 1) { DAQIRI_LOG_ERROR( - "Interface '{}' mixes standard (UDP/IP) and flex-item RX flows, which is not " + "Interface '{}' mixes standard (UDP/IP), flex-item and/or eCPRI RX flows, which is not " "supported. Use only one flow class per interface.", intf.name_); return false; diff --git a/src/engines/dpdk/daqiri_dpdk_engine.h b/src/engines/dpdk/daqiri_dpdk_engine.h index 3dabec26..639ed342 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.h +++ b/src/engines/dpdk/daqiri_dpdk_engine.h @@ -218,6 +218,11 @@ class DpdkEngine : public Engine { FlowId mark_id = 0, bool track = true); + // eCPRI-over-Ethernet RX flow (EtherType 0xAEFE) via RTE_FLOW_ITEM_TYPE_ECPRI. + // Programmed in a dedicated flow group (2), like the standard (3) and + // flex-item (1) classes, with its own group-0 ETH jump rule. + struct rte_flow* add_ecpri_flow(int port, const FlowConfig& cfg, bool track = true); + enum class DynamicFlowBackend { LEGACY, TEMPLATE, diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp index 95627f9c..ac19815e 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp @@ -1046,13 +1046,21 @@ bool IbverbsEngine::validate_dynamic_rx_flow_locked(int port, const FlowRuleConf return true; } + if (match.type_ == FlowMatchType::ECPRI) { + if (match.ecpri_match_.match_id_ && !match.ecpri_match_.match_msg_type_) { + DAQIRI_LOG_ERROR("Dynamic eCPRI RX flow matches pc_id/rtc_id but no msg_type"); + return false; + } + return true; + } + if (match.type_ == FlowMatchType::IPV4_UDP && (match.udp_src_ > 0 || match.udp_dst_ > 0 || match.ipv4_src_ != INADDR_ANY || match.ipv4_dst_ != INADDR_ANY || match.ipv4_len_ > 0)) { return true; } - DAQIRI_LOG_ERROR("Dynamic RX flow must define an IPv4/UDP or flex-item match"); + DAQIRI_LOG_ERROR("Dynamic RX flow must define an IPv4/UDP, flex-item or eCPRI match"); return false; } @@ -1240,6 +1248,24 @@ Status IbverbsEngine::install_flow_rule_locked(int port, return ok ? Status::SUCCESS : Status::GENERIC_FAILURE; } + if (mt.type_ == FlowMatchType::ECPRI) { + DrMatchParam mask{}, value{}; + mask.match_sz = sizeof(mask.buf); + value.match_sz = sizeof(value.buf); + uint16_t criteria = 0; + if (build_ecpri_match_locked( + dest->ctx, st, mt.ecpri_match_, reinterpret_cast(mask.buf), + reinterpret_cast(value.buf), &criteria) != Status::SUCCESS) { + return Status::GENERIC_FAILURE; + } + return create_dr_rule_locked(port, st, criteria, + reinterpret_cast(&mask), + reinterpret_cast(&value), + dest, priority, flow_id, flow.name_.c_str(), dynamic_entry) + ? Status::SUCCESS + : Status::GENERIC_FAILURE; + } + DrMatchParam mask{}, value{}; mask.match_sz = sizeof(mask.buf); value.match_sz = sizeof(value.buf); @@ -1420,6 +1446,135 @@ struct mlx5dv_devx_obj* IbverbsEngine::create_flex_parser_node(struct ibv_contex return obj; } +struct mlx5dv_devx_obj* IbverbsEngine::create_ecpri_parser_node(struct ibv_context* ctx, + uint32_t* out_type_sample_id, + uint32_t* out_id_sample_id) { + uint32_t in[DEVX_ST_SZ_DW(create_flex_parser_in)] = {0}; + uint32_t out[DEVX_ST_SZ_DW(create_flex_parser_out)] = {0}; + void* hdr = DEVX_ADDR_OF(create_flex_parser_in, in, hdr); + void* flex = DEVX_ADDR_OF(create_flex_parser_in, in, flex); + DEVX_SET(general_obj_in_cmd_hdr, hdr, opcode, MLX5_CMD_OP_CREATE_GENERAL_OBJECT); + DEVX_SET(general_obj_in_cmd_hdr, hdr, obj_type, MLX5_GENERAL_OBJ_TYPE_FLEX_PARSE_GRAPH); + + // Fixed-length custom header covering both 4-byte samples (offsets 0 and 4). + DEVX_SET(parse_graph_flex, flex, header_length_mode, MLX5_GRAPH_NODE_LEN_FIXED); + DEVX_SET(parse_graph_flex, flex, header_length_base_value, 8u); + + // Sample 0: eCPRI common header (message type lives in the second byte), at + // offset 0. Sample 1: eCPRI message body (pc_id/rtc_id), at offset 4. + void* s0 = DEVX_ADDR_OF(parse_graph_flex, flex, sample_table[0]); + DEVX_SET(parse_graph_flow_match_sample, s0, flow_match_sample_en, 1); + DEVX_SET(parse_graph_flow_match_sample, s0, flow_match_sample_offset_mode, + MLX5_GRAPH_SAMPLE_OFFSET_FIXED); + DEVX_SET(parse_graph_flow_match_sample, s0, flow_match_sample_field_base_offset, 0); + void* s1 = DEVX_ADDR_OF(parse_graph_flex, flex, sample_table[1]); + DEVX_SET(parse_graph_flow_match_sample, s1, flow_match_sample_en, 1); + DEVX_SET(parse_graph_flow_match_sample, s1, flow_match_sample_offset_mode, + MLX5_GRAPH_SAMPLE_OFFSET_FIXED); + DEVX_SET(parse_graph_flow_match_sample, s1, flow_match_sample_field_base_offset, 4); + + // Anchor the node at L2, entered on the eCPRI EtherType. + void* in_arc = DEVX_ADDR_OF(parse_graph_flex, flex, input_arc); + DEVX_SET(parse_graph_arc, in_arc, arc_parse_graph_node, MLX5_GRAPH_ARC_NODE_MAC); + DEVX_SET(parse_graph_arc, in_arc, compare_condition_value, MLX5_ETHERTYPE_ECPRI); + + struct mlx5dv_devx_obj* obj = mlx5dv_devx_obj_create(ctx, in, sizeof(in), out, sizeof(out)); + if (obj == nullptr) { + DAQIRI_LOG_CRITICAL( + "CREATE_GENERAL_OBJECT(FLEX_PARSE_GRAPH/eCPRI) failed: {} (syndrome 0x{:x})", + strerror(errno), DEVX_GET(general_obj_out_cmd_hdr, out, syndrome)); + return nullptr; + } + const uint32_t node_id = DEVX_GET(general_obj_out_cmd_hdr, out, obj_id); + + // Query the device-assigned sample field ids back (not reliably echoed by CREATE). + uint32_t qin[DEVX_ST_SZ_DW(general_obj_in_cmd_hdr)] = {0}; + uint32_t qout[DEVX_ST_SZ_DW(create_flex_parser_out)] = {0}; + DEVX_SET(general_obj_in_cmd_hdr, qin, opcode, MLX5_CMD_OP_QUERY_GENERAL_OBJECT); + DEVX_SET(general_obj_in_cmd_hdr, qin, obj_type, MLX5_GENERAL_OBJ_TYPE_FLEX_PARSE_GRAPH); + DEVX_SET(general_obj_in_cmd_hdr, qin, obj_id, node_id); + if (mlx5dv_devx_obj_query(obj, qin, sizeof(qin), qout, sizeof(qout)) != 0) { + DAQIRI_LOG_CRITICAL("QUERY_GENERAL_OBJECT(FLEX_PARSE_GRAPH/eCPRI) failed: {} (syndrome 0x{:x})", + strerror(errno), DEVX_GET(general_obj_out_cmd_hdr, qout, syndrome)); + mlx5dv_devx_obj_destroy(obj); + return nullptr; + } + void* q_flex = DEVX_ADDR_OF(create_flex_parser_out, qout, flex); + void* q_s0 = DEVX_ADDR_OF(parse_graph_flex, q_flex, sample_table[0]); + void* q_s1 = DEVX_ADDR_OF(parse_graph_flex, q_flex, sample_table[1]); + *out_type_sample_id = DEVX_GET(parse_graph_flow_match_sample, q_s0, flow_match_sample_field_id); + *out_id_sample_id = DEVX_GET(parse_graph_flow_match_sample, q_s1, flow_match_sample_field_id); + DAQIRI_LOG_INFO("eCPRI flex parser node created: id {} -> type_sample_id {} id_sample_id {}", + node_id, *out_type_sample_id, *out_id_sample_id); + return obj; +} + +Status IbverbsEngine::build_ecpri_match_locked(struct ibv_context* ctx, PortSteering& st, + const EcpriMatch& em, uint8_t* mask_buf, + uint8_t* value_buf, uint16_t* criteria) { + // Always pin the eCPRI EtherType so the rule only matches eCPRI-over-Ethernet + // frames (the parse-graph sample register is meaningful only for them). + DEVX_SET(fte_match_set_lyr_2_4, mask_buf, ethertype, 0xffff); + DEVX_SET(fte_match_set_lyr_2_4, value_buf, ethertype, MLX5_ETHERTYPE_ECPRI); + *criteria |= MLX5_DR_MATCH_CRITERIA_OUTER; + + if (!em.match_msg_type_ && !em.match_id_) { + return Status::SUCCESS; // EtherType-only match -- no flex sampling needed. + } + + if (st.ecpri_node.obj == nullptr) { + uint32_t type_sid = 0; + uint32_t id_sid = 0; + struct mlx5dv_devx_obj* node = create_ecpri_parser_node(ctx, &type_sid, &id_sid); + if (node == nullptr) { + DAQIRI_LOG_CRITICAL("eCPRI: parse-graph node creation failed"); + return Status::GENERIC_FAILURE; + } + st.ecpri_node = PortSteering::EcpriNode{node, type_sid, id_sid}; + } + + void* m4_mask = DEVX_ADDR_OF(fte_match_param, mask_buf, misc_parameters_4); + void* m4_value = DEVX_ADDR_OF(fte_match_param, value_buf, misc_parameters_4); + // prog_sample_field_id_N selects which parse-graph sample register; set + // identically in mask and value (a selector, not a matched field). The sample + // register holds the 4 sampled bytes in big-endian wire order, so DEVX_SET's + // host->BE conversion of the (value, mask) pair lines up with it -- the same + // arithmetic the ipv4_len path relies on. + int slot = 0; + auto set_sample = [&](uint32_t sid, uint32_t v, uint32_t msk) { + switch (slot) { + case 0: + DEVX_SET(fte_match_set_misc4, m4_mask, prog_sample_field_id_0, sid); + DEVX_SET(fte_match_set_misc4, m4_value, prog_sample_field_id_0, sid); + DEVX_SET(fte_match_set_misc4, m4_mask, prog_sample_field_value_0, msk); + DEVX_SET(fte_match_set_misc4, m4_value, prog_sample_field_value_0, v & msk); + break; + case 1: + DEVX_SET(fte_match_set_misc4, m4_mask, prog_sample_field_id_1, sid); + DEVX_SET(fte_match_set_misc4, m4_value, prog_sample_field_id_1, sid); + DEVX_SET(fte_match_set_misc4, m4_mask, prog_sample_field_value_1, msk); + DEVX_SET(fte_match_set_misc4, m4_value, prog_sample_field_value_1, v & msk); + break; + default: + break; + } + slot++; + }; + + // Message type: second byte of the common-header dword (sample at offset 0) -> + // big-endian register bits [23:16], mask 0x00ff0000. + if (em.match_msg_type_) { + set_sample(st.ecpri_node.type_sample_id, static_cast(em.msg_type_) << 16, + 0x00ff0000u); + } + // Identifier: top 16 bits of the message-body dword (sample at offset 4). + if (em.match_id_) { + set_sample(st.ecpri_node.id_sample_id, static_cast(em.id_) << 16, 0xffff0000u); + } + *criteria |= MLX5_DR_MATCH_CRITERIA_MISC4; + return Status::SUCCESS; +} + Status IbverbsEngine::install_port_flows() { for (const auto& intf : cfg_.ifs_) { if (intf.rx_.queues_.empty()) { @@ -1605,6 +1760,32 @@ Status IbverbsEngine::install_port_flows() { continue; } + // eCPRI-over-Ethernet match: pin the eCPRI EtherType (outer) and, when a + // message type / identifier is requested, the parse-graph sample registers + // (misc_parameters_4). One shared eCPRI node per port, created lazily. + if (mt.type_ == FlowMatchType::ECPRI) { + DrMatchParam emask{}, eval{}; + emask.match_sz = sizeof(emask.buf); + eval.match_sz = sizeof(eval.buf); + uint16_t ecrit = 0; + if (build_ecpri_match_locked( + ctx, st, mt.ecpri_match_, reinterpret_cast(emask.buf), + reinterpret_cast(eval.buf), &ecrit) != Status::SUCCESS) { + return Status::GENERIC_FAILURE; + } + if (!add_rule(ecrit, reinterpret_cast(&emask), + reinterpret_cast(&eval), it->second, + prio++, fl.id_, fl.name_.c_str())) { + return Status::GENERIC_FAILURE; + } + DAQIRI_LOG_INFO( + "Flow '{}' (eCPRI) -> queue {} (port {}): msg_type={}(matched={}) id={}(matched={})", + fl.name_, fl.action_.id_, port, mt.ecpri_match_.msg_type_, + mt.ecpri_match_.match_msg_type_, mt.ecpri_match_.id_, mt.ecpri_match_.match_id_); + installed++; + continue; + } + // Full-size param so an ipv4_len match can reach misc_parameters_4; outer // fields live at offset 0 either way. criteria accumulates as fields are set. DrMatchParam mask{}, val{}; @@ -3454,6 +3635,9 @@ void IbverbsEngine::shutdown() { if (st.ipv4_len_node.obj) { mlx5dv_devx_obj_destroy(st.ipv4_len_node.obj); } + if (st.ecpri_node.obj) { + mlx5dv_devx_obj_destroy(st.ecpri_node.obj); + } } port_steering_.clear(); diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.h b/src/engines/ibverbs/daqiri_ibverbs_engine.h index 92f2e243..e3d2ec2c 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.h +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.h @@ -417,6 +417,13 @@ class IbverbsEngine : public Engine { struct mlx5dv_devx_obj* create_flex_parser_node(struct ibv_context* ctx, uint8_t arc_node, uint16_t compare_value, uint16_t offset, uint32_t* out_sample_id); + // Create the per-port eCPRI parse-graph node: anchored at MAC on the eCPRI + // EtherType (0xAEFE) with two 4-byte samples -- offset 0 (common header, the + // message type) and offset 4 (message body, the pc_id/rtc_id). Returns the + // DevX object and the two device-assigned sample field ids. + struct mlx5dv_devx_obj* create_ecpri_parser_node(struct ibv_context* ctx, + uint32_t* out_type_sample_id, + uint32_t* out_id_sample_id); void devx_build_wqe(IbvRxQueue& q, uint32_t wqe_idx); // write one RQ WQE void devx_ring_rq_doorbell(IbvRxQueue& q); void devx_advance_producer(IbvRxQueue& q); // worker-only: refill freed regions @@ -552,6 +559,17 @@ class IbverbsEngine : public Engine { // lazily when the first ipv4_len flow is installed (anchored at L2 on the // IPv4 ethertype). Torn down with the flex_nodes after rules. FlexNode ipv4_len_node; + // Shared parse-graph node for eCPRI-over-Ethernet matching (anchored at L2 + // on the eCPRI EtherType), created lazily when the first eCPRI flow that + // matches a message type or identifier is installed. Carries two sample + // registers: type_sample_id (common header, offset 0) and id_sample_id + // (message body, offset 4). Torn down with the flex_nodes after rules. + struct EcpriNode { + struct mlx5dv_devx_obj* obj = nullptr; + uint32_t type_sample_id = 0; + uint32_t id_sample_id = 0; + }; + EcpriNode ecpri_node; bool dropped = false; // Continue directly after static rules; sparse high priorities can fail on // some mlx5 DR stacks even when the same matcher/action works at init. @@ -578,6 +596,13 @@ class IbverbsEngine : public Engine { FlowId flow_id, int priority, DynamicFlowEntry* dynamic_entry); + // Fill an eCPRI flow's match mask/value (always pinning the eCPRI EtherType in + // outer_headers, and the message type / identifier in misc_parameters_4 when + // requested), lazily creating the port's shared eCPRI parse-graph node. + // Accumulates the match criteria bits. Shared by the static and dynamic flow + // installers. + Status build_ecpri_match_locked(struct ibv_context* ctx, PortSteering& st, const EcpriMatch& em, + uint8_t* mask_buf, uint8_t* value_buf, uint16_t* criteria); std::mutex flow_lock_; FlowId next_dynamic_flow_id_ = 1; diff --git a/src/engines/ibverbs/mlx5_prm_min.h b/src/engines/ibverbs/mlx5_prm_min.h index cc05b46f..12f88f2c 100644 --- a/src/engines/ibverbs/mlx5_prm_min.h +++ b/src/engines/ibverbs/mlx5_prm_min.h @@ -439,6 +439,7 @@ enum { MLX5_DR_MATCH_CRITERIA_OUTER = 1 << 0, // match_criteria_enable: outer_headers MLX5_DR_MATCH_CRITERIA_MISC4 = 1 << 5, // match_criteria_enable: misc_parameters_4 MLX5_ETHERTYPE_IPV4 = 0x0800, + MLX5_ETHERTYPE_ECPRI = 0xAEFE, // eCPRI over Ethernet MLX5_IP_PROTOCOL_UDP = 0x11, }; From a336d48486fb985d139a6a26fb0e8e3e0ee6544e Mon Sep 17 00:00:00 2001 From: Cliff Burdick Date: Tue, 16 Jun 2026 23:14:40 +0000 Subject: [PATCH 2/3] #192 - Use firmware steering for DPDK eCPRI flows The mlx5 PMD honors the native eCPRI flow item (RTE_FLOW_ITEM_TYPE_ECPRI) only under firmware steering (dv_flow_en=1); under HW steering (dv_flow_en=2, the engine default) the rule installs but never matches on ConnectX-class NICs. initialize() now auto-switches any interface carrying eCPRI RX flows to dv_flow_en=1 and logs a warning. Other interfaces keep dv_flow_en=2. Trade-off: the async/template dynamic-RX-flow path (which needs HW steering) is unavailable on an eCPRI interface. Validated on ConnectX-7: DAQIRI TX (raw eCPRI frames) -> DAQIRI RX steers each (msg_type, pc_id) variant to the correct queue with the correct per-packet flow id, on both the dpdk (dv_flow_en=1) and ibverbs engines. Signed-off-by: Cliff Burdick --- AGENTS.md | 2 +- docs/api-reference/configuration.md | 5 ++++- src/engines/dpdk/daqiri_dpdk_engine.cpp | 29 ++++++++++++++++++++++++- 3 files changed, 33 insertions(+), 3 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 32bdb88f..fe54c717 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -73,7 +73,7 @@ clang-format -style=file -i -fallback-style=none `EngineType` (`include/daqiri/types.h`) is resolved from `(stream_type, engine)`: `raw` defaults to `EngineType::DPDK`; `raw` + `engine: "ibverbs"` selects `EngineType::IBVERBS` (the MPRQ engine); `socket` + a `roce://` endpoint (or `engine: "ibverbs"`) selects `EngineType::RDMA`. The stream-aware `config_engine_from_string(str, stream_type)` overload encodes the `ibverbs`→`{IBVERBS for raw, RDMA for socket}` split. `EngineFactory` (in `engine.h`) is a singleton that instantiates the active engine. `daqiri_init(...)` resolves which engine to use from the `NetworkConfig` and then delegates everything through the `Engine` vtable. There is only ever **one** active `Engine` per process. -The default `dpdk` raw engine (`src/engines/dpdk/`, `DpdkEngine`) programs RX steering, send-to-kernel fallbacks (`flow_isolation: true`), and `tx_eth_src` TX offloads via DPDK RTE Flow during `daqiri_init()`. Standard UDP/IP (group 3), flex-item (group 1) and eCPRI-over-Ethernet (group 2, EtherType `0xAEFE`, via `RTE_FLOW_ITEM_TYPE_ECPRI` matching message type and pc_id/rtc_id) RX flows use separate flow groups; `validate_config()` rejects mixing these flow classes per interface, duplicate `flex_item_id` values, and unknown `action.id` / `flex_item_id` values before NIC programming. Flex-item parser handles are created per `(port, flex_item_id)` (scoped per interface). All programmed `rte_flow` rules and flex-item handles are tracked and destroyed in order on shutdown, init failure, and engine teardown (programmed flows → flex items → group-0 ETH jump rules). See `docs/benchmarks/raw_benchmarking.md` (Flow programming smoke test) for manual verification steps. +The default `dpdk` raw engine (`src/engines/dpdk/`, `DpdkEngine`) programs RX steering, send-to-kernel fallbacks (`flow_isolation: true`), and `tx_eth_src` TX offloads via DPDK RTE Flow during `daqiri_init()`. Standard UDP/IP (group 3), flex-item (group 1) and eCPRI-over-Ethernet (group 2, EtherType `0xAEFE`, via `RTE_FLOW_ITEM_TYPE_ECPRI` matching message type and pc_id/rtc_id) RX flows use separate flow groups; `validate_config()` rejects mixing these flow classes per interface, duplicate `flex_item_id` values, and unknown `action.id` / `flex_item_id` values before NIC programming. The mlx5 PMD honors the native eCPRI flow item only under **firmware steering** (`dv_flow_en=1`); under HW steering (`dv_flow_en=2`, the default) the rule installs but silently never matches on ConnectX-class NICs. So `initialize()` auto-switches any interface carrying eCPRI RX flows to `dv_flow_en=1` (with a `WARN`), which means the async/template dynamic-RX-flow path is unavailable on that port. Flex-item parser handles are created per `(port, flex_item_id)` (scoped per interface). All programmed `rte_flow` rules and flex-item handles are tracked and destroyed in order on shutdown, init failure, and engine teardown (programmed flows → flex items → group-0 ETH jump rules). See `docs/benchmarks/raw_benchmarking.md` (Flow programming smoke test) for manual verification steps. The `ibverbs` raw engine (`src/engines/ibverbs/`, `IbverbsEngine`) drives a Mellanox/mlx5 Multi-Packet (striding) Receive Queue via **DevX** (`mlx5dv_devx_obj_create` against vendored PRM structs in `mlx5_prm_min.h`): a DevX CQ + striding RQ + TIR + `mlx5dv_dr` flow steering, with manual WQE/doorbell management and worker-driven cyclic refill. RX packets DMA strided into one pre-posted MR (host or GPU via `ibv_reg_dmabuf_mr`); a queue with >1 memory region instead uses a non-striding DevX *regular* RQ with multi-segment scatter WQEs for **physical** header-data split (header → CPU MR, payload → GPU MR). TX builds mlx5 send WQEs directly on a raw-packet QP's SQ (via `mlx5dv_init_obj`, bypassing `ibv_post_send`) from a slab of registered slots tracked by cyclic index counters, with NIC checksum offload and a `tx_eth_src` offload. It uses the libdpdk-free `daqiri::Ring`/`daqiri::ObjectPool` for the worker→app burst handoff (like the rdma engine — neither links DPDK) and drives the NIC through libibverbs/mlx5dv directly. Feature set: RX (MPRQ), TX, GPUDirect, physical/logical HDS, multi-queue 5-tuple flow steering with per-packet flow IDs, flex-item arbitrary-offset, IPv4-total-length and eCPRI-over-Ethernet (EtherType `0xAEFE`, message type + pc_id/rtc_id) flow matching (mlx5 flex parser / `misc_parameters_4`), per-packet RX hardware timestamps, accurate TX send scheduling (wait-on-time WAIT WQE), and GPU reorder/quantize. Because it uses the kernel netdev directly, `ensure_port_mtus` raises the netdev MTU at init to cover the configured frame size (jumbo frames silently drop on RX otherwise). Queues sharing a `cpu_core` are serviced round-robin by one poller thread. diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index 1d5bc8d0..edeee177 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -247,7 +247,10 @@ For Raw Ethernet (`stream_type: "raw"`), each flow rule is programmed into the N created when `flow_isolation: true`, initialization fails with a critical log and `daqiri_init()` returns an error status. eCPRI matching is supported by both the `dpdk` (via the mlx5 eCPRI flow item) and `ibverbs` (via an mlx5 flex-parser node anchored on the -eCPRI EtherType) engines. +eCPRI EtherType) engines. On the `dpdk` engine the mlx5 eCPRI flow item is only honored under +firmware steering, so any interface with eCPRI flows is automatically switched to +`dv_flow_en=1` (logged as a warning); a side effect is that the async/template dynamic-RX-flow +API is unavailable on that interface. The `ibverbs` engine has no such restriction. A single RX interface must use exactly one flow class — standard UDP/IP, flex-item, or eCPRI. Each class installs its own DPDK group-0 jump rule, and these conflict when mixed, so only one diff --git a/src/engines/dpdk/daqiri_dpdk_engine.cpp b/src/engines/dpdk/daqiri_dpdk_engine.cpp index b7bde54d..5ec27218 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.cpp +++ b/src/engines/dpdk/daqiri_dpdk_engine.cpp @@ -2196,10 +2196,37 @@ void DpdkEngine::initialize() { } } + // The mlx5 native eCPRI flow item (RTE_FLOW_ITEM_TYPE_ECPRI) is honored only + // under software/firmware steering (dv_flow_en=1); HW steering (dv_flow_en=2) + // silently fails to match it on ConnectX-class NICs. Switch any interface with + // eCPRI RX flows to firmware steering. Trade-off: the async/template + // dynamic-RX-flow path requires HW steering, so it is unavailable on that port. + auto interface_has_ecpri = [&](const std::string& dev_addr) -> bool { + for (const auto& intf : cfg_.ifs_) { + if (intf.address_ != dev_addr) { + continue; + } + for (const auto& fl : intf.rx_.flows_) { + if (fl.match_.type_ == FlowMatchType::ECPRI) { + return true; + } + } + } + return false; + }; + if (loopback_ != LoopbackType::LOOPBACK_TYPE_SW) { for (const auto& name : ifs) { strncpy(_argv[arg++], "-a", max_arg_size - 1); - std::string devargs = name + std::string(",txq_inline_max=0,dv_flow_en=2"); + const int dv_flow_en = interface_has_ecpri(name) ? 1 : 2; + if (dv_flow_en == 1) { + DAQIRI_LOG_WARN( + "Interface '{}' has eCPRI RX flows: switching mlx5 to firmware steering " + "(dv_flow_en=1). HW steering (dv_flow_en=2) does not honor the eCPRI flow item on " + "this NIC; async/template dynamic RX flows are unavailable on this port.", + name); + } + std::string devargs = name + ",txq_inline_max=0,dv_flow_en=" + std::to_string(dv_flow_en); if (needs_send_sched) { devargs += ",tx_pp=500"; // 500 ns scheduling granularity } From befef5088808a2f2b91735e37eb4a28278b52e88 Mon Sep 17 00:00:00 2001 From: Cliff Burdick Date: Tue, 16 Jun 2026 23:25:56 +0000 Subject: [PATCH 3/3] #192 - Reject dynamic eCPRI RX flows on HW-steering ports validate_dynamic_rx_flow() accepted eCPRI flows without checking the port's steering mode. initialize() switches a port to firmware steering (dv_flow_en=1) only when it has a static eCPRI flow; on a HW-steering (dv_flow_en=2) port add_ecpri_flow() returns success but the mlx5 PMD silently never matches the eCPRI item, so packets bypass the queue with no error. Require at least one static eCPRI flow on the interface (the only signal that initialize() picked firmware steering) before accepting a dynamic eCPRI flow. The ibverbs engine has no such dependency and is unaffected. Signed-off-by: Cliff Burdick --- src/engines/dpdk/daqiri_dpdk_engine.cpp | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/engines/dpdk/daqiri_dpdk_engine.cpp b/src/engines/dpdk/daqiri_dpdk_engine.cpp index 5ec27218..0c4e406e 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.cpp +++ b/src/engines/dpdk/daqiri_dpdk_engine.cpp @@ -3189,6 +3189,22 @@ bool DpdkEngine::validate_dynamic_rx_flow(int port, const FlowRuleConfig& flow) DAQIRI_LOG_ERROR("Dynamic eCPRI RX flow matches pc_id/rtc_id but no msg_type"); return false; } + // The eCPRI flow item is honored only under firmware steering (dv_flow_en=1). + // initialize() switches a port to firmware steering only when it has at least + // one static eCPRI flow; otherwise the port stays on HW steering (dv_flow_en=2) + // where add_ecpri_flow() returns success but the rule silently never matches. + // A static eCPRI flow on the interface is the only signal that the switch + // happened, so require one before accepting a dynamic eCPRI flow. + const auto& intf_flows = cfg_.ifs_[port].rx_.flows_; + if (std::none_of(intf_flows.begin(), intf_flows.end(), + [](const FlowConfig& f) { return f.match_.type_ == FlowMatchType::ECPRI; })) { + DAQIRI_LOG_ERROR( + "Dynamic eCPRI RX flow on port {} requires at least one static eCPRI flow in the " + "initial config; without it the port keeps HW steering (dv_flow_en=2) and the eCPRI " + "flow item silently never matches.", + port); + return false; + } return true; }