diff --git a/AGENTS.md b/AGENTS.md index 322c09bb..fe54c717 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. 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). diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index 66debe41..edeee177 100644 --- a/docs/api-reference/configuration.md +++ b/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` 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..0c4e406e 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 } @@ -2682,11 +2709,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 +2722,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 +2737,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 +3184,31 @@ 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; + } + // 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; + } - 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 +3497,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 +5014,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 +5456,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 +5486,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, };