#192 - Add eCPRI matching

AGENTS.md

@@ -73,9 +73,9 @@ clang-format -style=file -i -fallback-style=none <files>
 
 `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. 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 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).

docs/api-reference/configuration.md

@@ -230,28 +230,45 @@ 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.
-
-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.
+`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. 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
+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`

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
 

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_;

src/common.cpp

@@ -888,6 +888,37 @@ bool YAML::convert<daqiri::NetworkConfig>::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<uint8_t>(ecpri_node["msg_type"].as<uint16_t>() & 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<uint16_t>();
+      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<uint16_t>();
   } catch (const std::exception& e) {