diff --git a/AGENTS.md b/AGENTS.md index d9d639a..3c2fc53 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -81,7 +81,7 @@ The always-built socket engine implements Linux UDP/TCP streams directly. Applic 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. +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 in either direction — RX (post-decap) and TX egress (post-encap) — sizing each direction with its own transform wire overhead (jumbo frames silently drop 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 bbc016e..3e69b69 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -433,9 +433,10 @@ the pre-encap packet. - **`match`**: Same standard UDP/IP match keys as RX flows. Omit `match` for a catch-all TX transform. -DAQIRI validates tunnel overhead against the configured packet buffer size and -the supported jumbo-frame bound. For RX decap/pop, MTU sizing accounts for the -outer wire frame while packet buffers contain the post-decap frame. +DAQIRI validates transform overhead against the configured packet buffer size and +the supported jumbo-frame bound. For RX decap/pop and TX encap/push, MTU sizing +accounts for the outer wire frame; packet buffers hold the post-decap (RX) / +pre-encap (TX) frame. ### Accurate Send diff --git a/docs/benchmarks/raw_benchmarking.md b/docs/benchmarks/raw_benchmarking.md index fa10a2a..31807b6 100644 --- a/docs/benchmarks/raw_benchmarking.md +++ b/docs/benchmarks/raw_benchmarking.md @@ -323,6 +323,8 @@ interface and queue. ??? abstract "See an example output" + This is an illustrative excerpt; source line numbers and pointer values vary by build. + ```log [INFO] /workspace/daqiri/src/../include/daqiri/common.h:1045: Finished reading DAQIRI configuration [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1732: Attempting to use 2 ports for high-speed network @@ -356,12 +358,10 @@ interface and queue. [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1850: DPDK init (0005:03:00.0) -- RX: ENABLED TX: ENABLED [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1863: Configuring RX queue: UNUSED_P0_Q0 (0) on port 0 [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1896: Created mempool RXP_P0_Q0_MR0 : mbufs=32768 elsize=9228 ptr=0x17fca4380 - [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1908: Max packet size needed for RX: 9100 [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1955: Configuring TX queue: tx_q_0 (0) on port 0 [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1981: Created mempool TXP_P0_Q0_MR0 : mbufs=32768 elsize=8064 ptr=0x148cdc980 - [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1995: Max packet size needed with TX: 9100 - [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1996: Max packet size needed with RX only: 9100 - [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:2012: Setting port config for port 0 mtu:9082 + [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1995: Max frame needed - RX: 0 TX: 8064 + [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:2012: Setting port config for port 0 mtu:8046 [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:2026: Enabling RX scatter offload for single-segment RX queues (min buffer size: 9100) [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:2062: Initializing port 0 with 1 RX queues and 1 TX queues... [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:2079: Successfully configured ethdev @@ -380,8 +380,7 @@ interface and queue. [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1908: Max packet size needed for RX: 8064 [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1955: Configuring TX queue: UNUSED_TX_P1_Q0 (0) on port 1 [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1981: Created mempool TXP_P1_Q0_MR0 : mbufs=32768 elsize=9100 ptr=0x1470e9e00 - [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1995: Max packet size needed with TX: 9100 - [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1996: Max packet size needed with RX only: 8064 + [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:1995: Max frame needed - RX: 8064 TX: 0 [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:2012: Setting port config for port 1 mtu:8046 [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:2026: Enabling RX scatter offload for single-segment RX queues (min buffer size: 8064) [INFO] /workspace/daqiri/src/engines/dpdk/daqiri_dpdk_engine.cpp:2062: Initializing port 1 with 1 RX queues and 1 TX queues... diff --git a/include/daqiri/types.h b/include/daqiri/types.h index 288c712..14b9fbb 100644 --- a/include/daqiri/types.h +++ b/include/daqiri/types.h @@ -881,6 +881,18 @@ inline size_t flow_max_decap_wire_overhead(const std::vector& flows) return overhead; } +inline size_t flow_max_encap_wire_overhead(const std::vector& flows) { + size_t overhead = 0; + for (const auto& flow : flows) { + size_t per_flow = 0; + for (const auto& action : flow.actions_) { + per_flow += flow_action_wire_overhead(action); + } + overhead = std::max(overhead, per_flow); + } + return overhead; +} + struct CommonConfig { int version; int master_core_; diff --git a/src/engines/dpdk/daqiri_dpdk_engine.cpp b/src/engines/dpdk/daqiri_dpdk_engine.cpp index 0c4e406..07a24e2 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.cpp +++ b/src/engines/dpdk/daqiri_dpdk_engine.cpp @@ -2319,13 +2319,22 @@ void DpdkEngine::initialize() { intf.rx_.queues_.size() > 0 ? "ENABLED" : "DISABLED", intf.tx_.queues_.size() > 0 ? "ENABLED" : "DISABLED"); - // Queue setup - size_t max_pkt_size = 0; + // Queue setup. MTU sizing is direction-aware: RX and TX frames are + // accumulated separately so each direction folds in only its own transform + // (decap/pop for RX, encap/push for TX) wire overhead. size_t max_rx_pkt_size = 0; + size_t max_tx_pkt_size = 0; bool single_seg_rx_needs_scatter = false; size_t min_single_seg_rx_buf_size = std::numeric_limits::max(); const auto& rx = intf.rx_; + // Generated dummy RX queues (name prefix "UNUSED_P") are setup-only and + // must not drive MTU sizing. Prefix match mirrors is_dummy_tx_queue; the + // 8th char differs from "UNUSED_TX_P" so this excludes only dummy RX. + const auto is_dummy_rx_queue = [](const CommonQueueConfig& queue) { + return queue.name_.find("UNUSED_P") == 0; + }; + for (auto& q : rx.queues_) { DAQIRI_LOG_INFO("Configuring RX queue: {} ({}) on port {}", q.common_.name_, @@ -2365,9 +2374,10 @@ void DpdkEngine::initialize() { } } - max_pkt_size = std::max(max_pkt_size, q_packet_size); - max_rx_pkt_size = std::max(max_rx_pkt_size, q_packet_size); - DAQIRI_LOG_INFO("Max packet size needed for RX: {}", max_pkt_size); + if (!is_dummy_rx_queue(q.common_)) { + max_rx_pkt_size = std::max(max_rx_pkt_size, q_packet_size); + DAQIRI_LOG_INFO("Max packet size needed for RX: {}", max_rx_pkt_size); + } // Keep scatter available for single-segment RX so queue setup is not tied to MTU sizing. if (q.common_.mrs_.size() == 1 && q_packet_size > 0) { @@ -2471,37 +2481,42 @@ void DpdkEngine::initialize() { q_packet_size += mr.buf_size_; } - // Exclude generated dummy TX queues so RX-only ports are not inflated to the - // dummy jumbo frame size when MTU sizing uses the combined RX/TX maximum. - // Generated dummy RX queues are left unchanged to preserve TX-only behavior. + // Generated dummy RX and TX queues are setup-only and are excluded from + // direction-aware MTU sizing, so they never inflate a port to the dummy + // jumbo frame size. if (!is_dummy_tx_queue(q.common_)) { - max_pkt_size = std::max(max_pkt_size, q_packet_size); + max_tx_pkt_size = std::max(max_tx_pkt_size, q_packet_size); } uint32_t key = generate_queue_key(intf.port_id_, q.common_.id_); tx_dpdk_q_map_[key] = q_backend; } - const size_t rx_tunnel_overhead = flow_max_decap_wire_overhead(rx.flows_); - if (rx_tunnel_overhead > 0) { - DAQIRI_LOG_INFO("Adding {} bytes of RX tunnel overhead for MTU sizing", rx_tunnel_overhead); - max_rx_pkt_size += rx_tunnel_overhead; - max_pkt_size = std::max(max_pkt_size, max_rx_pkt_size); + const size_t rx_transform_overhead = flow_max_decap_wire_overhead(rx.flows_); + if (rx_transform_overhead > 0) { + DAQIRI_LOG_INFO("Adding {} bytes of RX transform overhead for MTU sizing", + rx_transform_overhead); } - - DAQIRI_LOG_INFO("Max packet size needed with TX: {}", max_pkt_size); - DAQIRI_LOG_INFO("Max packet size needed with RX only: {}", max_rx_pkt_size); + const size_t tx_transform_overhead = flow_max_encap_wire_overhead(tx.flows_); + if (tx_transform_overhead > 0) { + DAQIRI_LOG_INFO("Adding {} bytes of TX transform overhead for MTU sizing", + tx_transform_overhead); + } + const size_t rx_required = max_rx_pkt_size + rx_transform_overhead; + const size_t tx_required = max_tx_pkt_size + tx_transform_overhead; + DAQIRI_LOG_INFO("Max frame needed - RX: {} TX: {}", rx_required, tx_required); local_port_conf[intf.port_id_].txmode.offloads = 0; - // Compute MTU from the largest RX/TX queue buffer frame needed on this port - // (max_pkt_size already folds in the existing RX tunnel decap overhead). - // Clamp to jumbo bounds and enforce a valid minimum MTU. - max_pkt_size = std::max(max_pkt_size, 64UL); - const size_t requested_frame_size = - std::min(max_pkt_size, static_cast(JUMBOFRAME_SIZE)); + // Compute MTU from the largest frame needed on this port in either + // direction: each direction's queue buffer frame plus its own transform + // (RX decap/pop, TX encap/push) wire overhead. Clamp to jumbo bounds and + // enforce a valid minimum MTU. + const size_t required_frame = std::max({rx_required, tx_required, 64UL}); + const size_t clamped_frame_size = + std::min(required_frame, static_cast(JUMBOFRAME_SIZE)); const size_t crc_and_hdr = RTE_ETHER_CRC_LEN + RTE_ETHER_HDR_LEN; const size_t requested_mtu = - (requested_frame_size > crc_and_hdr) ? (requested_frame_size - crc_and_hdr) : 0; + (clamped_frame_size > crc_and_hdr) ? (clamped_frame_size - crc_and_hdr) : 0; local_port_conf[intf.port_id_].rxmode.mtu = std::max(requested_mtu, static_cast(RTE_ETHER_MIN_MTU)); local_port_conf[intf.port_id_].rxmode.max_lro_pkt_size = diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp index ac19815..68a334b 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp @@ -2322,7 +2322,7 @@ void IbverbsEngine::initialize() { } // Raise each port's netdev MTU to cover the configured frame size before any - // traffic flows (the kernel netdev MTU gates jumbo RX on this path). + // traffic flows (the kernel netdev MTU gates jumbo RX and TX egress on this path). ensure_port_mtus(); // All RX queues (and their TIRs) now exist -- install per-port flow steering. @@ -3293,10 +3293,12 @@ void IbverbsEngine::ensure_port_mtus() { constexpr unsigned kEthHdr = 14; // L2 header bytes excluded from the MTU constexpr unsigned kStdMtu = 1500; // never lower below the standard MTU for (const auto& intf : cfg_.ifs_) { - // Largest L2 frame the NIC may deliver = max over queues of the sum of the - // queue's memory-region buffer sizes (HDS spreads one frame across regions). - size_t max_frame = 0; - auto accumulate = [&](const CommonQueueConfig& c) { + // Largest L2 frame the NIC may handle in each direction = max over that + // direction's queues of the sum of the queue's memory-region buffer sizes + // (HDS spreads one frame across regions). MTU sizing is direction-aware: + // RX folds in decap/pop overhead, TX egress folds in encap/push overhead. + size_t rx_frame = 0, tx_frame = 0; + auto queue_frame = [&](const CommonQueueConfig& c) { size_t sum = 0; for (const auto& name : c.mrs_) { auto it = cfg_.mrs_.find(name); @@ -3304,24 +3306,34 @@ void IbverbsEngine::ensure_port_mtus() { sum += it->second.buf_size_; } } - max_frame = std::max(max_frame, sum); + return sum; }; for (const auto& q : intf.rx_.queues_) { - accumulate(q.common_); + rx_frame = std::max(rx_frame, queue_frame(q.common_)); } for (const auto& q : intf.tx_.queues_) { - accumulate(q.common_); + tx_frame = std::max(tx_frame, queue_frame(q.common_)); } - max_frame += flow_max_decap_wire_overhead(intf.rx_.flows_); - if (max_frame <= kStdMtu + kEthHdr) { + const size_t rx_required = rx_frame + flow_max_decap_wire_overhead(intf.rx_.flows_); + const size_t tx_required = tx_frame + flow_max_encap_wire_overhead(intf.tx_.flows_); + const size_t required_frame = std::max(rx_required, tx_required); + if (required_frame <= kStdMtu + kEthHdr) { continue; } // standard MTU already fits - const unsigned need = static_cast(max_frame) - kEthHdr; + const unsigned need = + static_cast(required_frame > kEthHdr ? required_frame - kEthHdr : 0); + // Computed-target log: unconditional w.r.t. the current netdev MTU, so the + // target is observable even when cur >= need and the raise is skipped + // (parity with DPDK's always-on MTU log). Placed after the standard-MTU + // skip so no-op ports don't log a 0 target. + DAQIRI_LOG_INFO("MTU: port {} requires frame {} B (target netdev MTU {})", intf.port_id_, + required_frame, need); const std::string netdev = port_netdev(intf.port_id_); if (netdev.empty()) { - DAQIRI_LOG_WARN("MTU: no netdev for port {}; cannot raise MTU (need {}) -- jumbo RX may drop", - intf.port_id_, need); + DAQIRI_LOG_WARN( + "MTU: no netdev for port {}; cannot raise MTU (need {}) -- jumbo frames may drop", + intf.port_id_, need); continue; } // Read the current MTU; only raise it (never shrink, to avoid disturbing @@ -3349,12 +3361,12 @@ void IbverbsEngine::ensure_port_mtus() { ifr.ifr_mtu = static_cast(need); if (ioctl(fd, SIOCSIFMTU, &ifr) != 0) { DAQIRI_LOG_WARN( - "MTU: could not raise {} MTU {}->{} ({}); jumbo RX may silently drop. Set it manually " - "(ip link set {} mtu {}) or run with privileges.", + "MTU: could not raise {} MTU {}->{} ({}); jumbo frames may silently drop. Set it " + "manually (ip link set {} mtu {}) or run with privileges.", netdev, cur, need, strerror(errno), netdev, need); } else { DAQIRI_LOG_INFO("MTU: raised {} (port {}) from {} to {} for max frame {} B", netdev, - intf.port_id_, cur, need, max_frame); + intf.port_id_, cur, need, required_frame); } close(fd); } diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.h b/src/engines/ibverbs/daqiri_ibverbs_engine.h index e3d2ec2..fcfe2b0 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.h +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.h @@ -475,7 +475,8 @@ class IbverbsEngine : public Engine { std::string port_netdev(int port) const; // Raise each port's netdev MTU to cover the largest configured frame. Unlike // DPDK, the ibverbs path uses the kernel netdev directly and does not own the - // port MTU, so jumbo frames are silently dropped on RX if the MTU is too low. + // port MTU, so jumbo frames are silently dropped on RX and TX egress + // (post-encap) if the MTU is too low. void ensure_port_mtus(); // ---- dynamic RX flow lifecycle ----