From 225adf31d255da5d44d05b4516b1d811efd06d11 Mon Sep 17 00:00:00 2001 From: Cliff Burdick Date: Fri, 17 Jul 2026 15:36:26 +0000 Subject: [PATCH 1/2] #233 - Add automatic RSS for multi-queue flows Signed-off-by: Cliff Burdick --- AGENTS.md | 8 +- CMakeLists.txt | 2 +- docs/api-reference/configuration.md | 26 +- docs/api-reference/cpp.md | 22 +- docs/api-reference/python.md | 13 +- docs/benchmarks/raw_benchmarking.md | 5 +- docs/concepts.md | 18 +- docs/getting-started.md | 6 +- docs/images/daqiri_rx_path.svg | 2 +- docs/tutorials/configuration-walkthrough.md | 4 +- examples/README.md | 4 +- examples/dynamic_rx_flow_example.cpp | 123 +++++-- include/daqiri/types.h | 28 +- python/daqiri_common_pybind.cpp | 1 + src/common.cpp | 24 +- src/engine.cpp | 67 +++- src/engines/dpdk/daqiri_dpdk_engine.cpp | 181 ++++++---- src/engines/dpdk/daqiri_dpdk_engine.h | 6 +- src/engines/ibverbs/daqiri_ibverbs_engine.cpp | 320 +++++++++++++----- src/engines/ibverbs/daqiri_ibverbs_engine.h | 27 +- src/engines/ibverbs/mlx5_prm_min.h | 44 +++ src/rss.h | 29 ++ 22 files changed, 717 insertions(+), 243 deletions(-) create mode 100644 src/rss.h diff --git a/AGENTS.md b/AGENTS.md index 129f1da..479f602 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -23,7 +23,7 @@ CMake options (full table in `docs/getting-started.md`): - `DAQIRI_ENABLE_S3` — enable AWS SDK-backed asynchronous raw packet writes to S3 (off by default). - `DAQIRI_PREFER_SYSTEM_YAML_CPP` — prefer system `yaml-cpp` over the vendored `third_party/yaml-cpp` submodule (off by default; keep off when a conda/miniforge env is on `PATH`). -Package versions use CalVer (`YYYY.MM.PATCH`) from the top-level `VERSION` file. CMake reads that value into `project(daqiri VERSION ...)`, generates `daqiri/version.h`, feeds pkg-config/CMake package metadata, and exposes the same value in Python. `DAQIRI_ABI_VERSION` is separate and currently `0`; do not tie ABI policy to the CalVer year. +Package versions use CalVer (`YYYY.MM.PATCH`) from the top-level `VERSION` file. CMake reads that value into `project(daqiri VERSION ...)`, generates `daqiri/version.h`, feeds pkg-config/CMake package metadata, and exposes the same value in Python. `DAQIRI_ABI_VERSION` is separate and currently `1`; do not tie ABI policy to the CalVer year. CUDA architectures default to `80;90` (A100, H100), with `121` (GB10) added when configuring with CUDA Toolkit 13.0 or newer. Override `CMAKE_CUDA_ARCHITECTURES` when targeting other GPUs. @@ -36,7 +36,7 @@ There is no unit test suite. Verification is done via the benchmark executables | Executable | Source | Typical config | |---|---|---| | `daqiri_bench_raw_gpudirect` | `raw_gpudirect_bench.cpp` | `daqiri_bench_raw_tx_rx.yaml`, `daqiri_bench_raw_tx_rx_4q.yaml`, `daqiri_bench_raw_tx_rx_spark.yaml`, `daqiri_bench_raw_{tx,rx}_spark_xhost.yaml`, `daqiri_bench_raw_sw_loopback.yaml`, `daqiri_bench_raw_rx_multi_q.yaml`, `daqiri_bench_raw_tx_rx_vxlan.yaml`, `daqiri_bench_raw_tx_rx_vlan.yaml`, `daqiri_bench_raw_tx_rx_gre.yaml`, `daqiri_bench_raw_tx_rx_nvgre.yaml`, `daqiri_bench_raw_tx_rx_spark_mq.yaml` (mq base; `run_spark_mq_bench.sh` derives the 4 cells via `scripts/gen_spark_mq_config.py`), `daqiri_bench_raw_tx_rx_pacing.yaml` (per-queue `pacing_mbps`; DPDK engine only) | -| `daqiri_example_dynamic_rx_flow` | `dynamic_rx_flow_example.cpp` | `daqiri_example_dynamic_rx_flow.yaml` — `flow_isolation: true` startup followed by runtime RX queue-steering and raw-engine decap/pop flow add/delete | +| `daqiri_example_dynamic_rx_flow` | `dynamic_rx_flow_example.cpp` | `daqiri_example_dynamic_rx_flow.yaml` — `flow_isolation: true` startup followed by runtime scalar queue steering, multi-queue RSS, and raw-engine decap/pop flow add/delete | | `daqiri_bench_raw_hds` | `raw_hds_bench.cpp` | `daqiri_bench_raw_tx_rx_hds.yaml` | | `daqiri_bench_raw_reorder_seq` | `raw_reorder_seq_bench.cpp` | `daqiri_bench_raw_tx_rx_reorder_seq_1024*.yaml`, `daqiri_bench_raw_rx_reorder_seq_*.yaml` | | `daqiri_bench_raw_reorder_quantize` | `raw_reorder_quantize_bench.cpp` | `daqiri_bench_raw_tx_rx_reorder_quantize_seq_batch.yaml` | @@ -79,7 +79,7 @@ clang-format -style=file -i -fallback-style=none The always-built socket engine implements Linux UDP/TCP streams directly. Applications that need kernel socket tuning call `socket_setsockopt(conn_id, level, optname, optval, optlen)` after resolving a TCP/UDP connection ID; DAQIRI passes the numeric Linux constants through without maintaining a symbolic option map. `socket_setsockopt` is not supported for `roce://` connections, which delegate to the RDMA/ibverbs path. -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 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 queue targets / `flex_item_id` values before NIC programming. Queue actions with two or more IDs use mlx5 Toeplitz IPv4/UDP RSS. 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 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. @@ -100,7 +100,7 @@ Vendored under `third_party/` as submodules (`.gitmodules`): `yaml-cpp` (config ### Current limitations - TX header fill currently supports UDP only (see README). -- Raw Ethernet RX flow legacy `action.id` or final `actions:` queue action must match an `rx.queues` ID, and flex-item flows must reference a valid `flex_item_id` on the same interface; `daqiri_init()` aborts if RX flow rules, send-to-kernel fallbacks (`flow_isolation: true`), transform flow actions, or `tx_eth_src` offload rules cannot be programmed on the NIC. +- Raw Ethernet RX flow scalar `action.id` or every entry in `action.ids` (including the final `actions:` queue action) must match an `rx.queues` ID, and flex-item flows must reference a valid `flex_item_id` on the same interface; `daqiri_init()` aborts if RX flow rules, RSS destinations, send-to-kernel fallbacks (`flow_isolation: true`), transform flow actions, or `tx_eth_src` offload rules cannot be programmed on the NIC. - Raw Ethernet tunnel/VLAN transform flows are hardware-only on the DPDK and ibverbs raw engines. TX flows may contain only push/encap transform actions and RX transform flows must use pop/decap actions ending in a queue; socket/RDMA engines reject these actions instead of adding a software fallback. - Raw Ethernet RX flow steering: a single interface cannot mix standard (UDP/IP) and flex-item flows, and flex-item flows cannot combine with tunnel/VLAN transform actions; `DpdkEngine::validate_config()` rejects mixed configs at init. diff --git a/CMakeLists.txt b/CMakeLists.txt index fc2d1dc..9c8ecb1 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -31,7 +31,7 @@ include(GNUInstallDirs) include(CMakePackageConfigHelpers) find_package(CUDAToolkit REQUIRED) -set(DAQIRI_ABI_VERSION "0" CACHE STRING "DAQIRI shared library ABI version") +set(DAQIRI_ABI_VERSION "1" CACHE STRING "DAQIRI shared library ABI version") if(NOT DAQIRI_ABI_VERSION MATCHES "^(0|[1-9][0-9]*)$") message(FATAL_ERROR "DAQIRI_ABI_VERSION must be a non-negative integer") endif() diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index c49323b..0bdb957 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -207,11 +207,16 @@ RX flows can also perform hardware VLAN pop or tunnel decapsulation before queue - **`id`**: Flow ID. Retrievable at runtime via `get_packet_flow_id()`. - type: `integer` - **`action`**: Legacy single action map. Existing configs may keep using - `action: {type: queue, id: ...}`. + `action: {type: queue, id: ...}`. A multi-queue destination uses + `action: {type: queue, ids: [0, 1]}`. - **`actions`**: Ordered action list. Use this for tunnel/VLAN transforms. RX transform flows must end with `type: queue`. - **`type: queue`**: Steer matched packets to an RX queue. - **`id`**: Queue ID under `rx.queues` on the same interface. + - **`ids`**: Non-empty list of unique queue IDs under `rx.queues` on the same + interface. One entry is direct steering. Two or more entries automatically + enable flow-affine Toeplitz RSS over source/destination IPv4 address and + source/destination UDP port. `id` and `ids` are mutually exclusive. - **`type: vlan_pop`**: Pop one VLAN tag in hardware. - **`type: tunnel_decap`**: Decapsulate a hardware tunnel before queue delivery. - **`tunnel.type`**: `vxlan`, `gre`, or `nvgre`. @@ -262,6 +267,22 @@ Each class installs its own DPDK group-0 jump rule, and these conflict when mixe 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. +Multi-queue RSS is supported for standard IPv4/UDP and flex-item flows. A flex +item selects the rule, but its sampled value is not an RSS input; distribution +still uses the packet's IPv4/UDP tuple. VLAN-pop rules hash the IPv4/UDP packet +after the VLAN header, and tunnel-decap rules hash the inner IPv4/UDP tuple. +eCPRI flows cannot use multi-queue RSS because they have no applicable UDP/IP +five tuple. Queue-list order affects hash-to-queue mapping, but does not express +weights. + +RSS is flow-affine: every packet with an unchanged five tuple stays on one +queue. Roughly even packet counts require enough distinct tuples with reasonably +balanced traffic; this is not packet striping or exact round-robin delivery. +There is no queue-action mode field in schema v1; a future stripe mode can be +added without changing the multi-ID RSS default. If +the NIC rejects an RSS action, static initialization or the dynamic flow +completion fails rather than falling back to one queue. + ### Flow Isolation `rx.flow_isolation:` When `true`, only packets matching an explicit flow rule are delivered @@ -346,7 +367,8 @@ v1 batch-size requirement: nibble first; `network` endianness swaps byte-multiple input types wider than 8 bits - **`flow_ids`**: List of RX flow IDs this reorder config applies to. - type: `list[integer]` - - notes: flow IDs cannot overlap across reorder configs on the same interface + - notes: flow IDs cannot overlap across reorder configs on the same interface; + a referenced flow must use direct steering to one queue, not multi-queue RSS - **`method`**: Exactly one method must be configured: - **`seq_batch_number`** - `sequence_number.bit_offset` diff --git a/docs/api-reference/cpp.md b/docs/api-reference/cpp.md index ad43987..db6e8cb 100644 --- a/docs/api-reference/cpp.md +++ b/docs/api-reference/cpp.md @@ -159,8 +159,11 @@ Raw Ethernet RX flows can be added and deleted after `daqiri_init()` on the `dpdk` and raw `ibverbs` engines. This supports queues-only startup configs, including `rx.flow_isolation: true` with no initial `rx.flows`. Static YAML flows still use explicit configured IDs and are not deletable through this API. -The legacy `FlowRuleConfig::action_` field remains the shorthand for a single -queue action; `FlowRuleConfig::actions_` is the ordered form used when a dynamic +The legacy `FlowRuleConfig::action_` field remains the shorthand for a queue +action; set `FlowAction::id_` for direct steering or `FlowAction::ids_` for a +queue list. A list with one entry is direct steering, while two or more entries +automatically enable flow-affine IPv4/UDP five-tuple RSS. `FlowRuleConfig::actions_` +is the ordered form used when a dynamic RX rule needs hardware VLAN pop or tunnel decapsulation before queue delivery. Dynamic TX transform flows are not part of v1; configure TX encapsulation/push rules statically under `tx.flows`. @@ -201,6 +204,21 @@ while (flow_id == 0) { } ``` +To distribute distinct IPv4/UDP flows across two queues, replace the scalar +target with a queue list: + +```cpp +flow.action_.ids_ = {0, 1}; +``` + +The queue list must be non-empty, duplicate-free, and contain only configured RX +queue IDs. Each unchanged five tuple remains on one queue; approximately even +packet totals require many distinct tuples with reasonably balanced traffic. +Tunnel decapsulation hashes the inner tuple. Multi-queue RSS is not available for +eCPRI flows or flows referenced by an RX reorder configuration. Hardware creation +failure is reported by the static initialization or dynamic operation completion; +DAQIRI does not silently select one queue. + For a dynamic VXLAN decap rule, use ordered actions and make the final action the target queue: diff --git a/docs/api-reference/python.md b/docs/api-reference/python.md index f7dcb7c..ba160ea 100644 --- a/docs/api-reference/python.md +++ b/docs/api-reference/python.md @@ -622,8 +622,15 @@ The workflow sections above show the common call order and ownership rules. | `rdma_get_port_queue(conn_id)` | Return `(Status, port, queue)`. | | `rdma_get_server_conn_id(server_addr, server_port)` | Return `(Status, conn_id)`. | -Dynamic RX flows are RX-only in v1. The `action` attribute remains the single queue-action -shorthand. Use ordered `actions` when a raw DPDK or raw ibverbs dynamic rule needs hardware +Dynamic RX flows are RX-only in v1. The `action` attribute remains the queue-action +shorthand. Set `flow.action.id = 0` for direct steering, or set +`flow.action.ids = [0, 1]` to enable automatic flow-affine IPv4/UDP five-tuple +RSS. A one-entry `ids` list is direct steering. Lists must be non-empty, +duplicate-free, and contain configured RX queue IDs. An unchanged five tuple +stays on one queue; balanced packet totals require enough distinct, reasonably +balanced tuples. Multi-queue RSS is unsupported for eCPRI and reorder-owned +flows, and hardware creation errors are returned rather than falling back to a +single queue. Use ordered `actions` when a raw DPDK or raw ibverbs dynamic rule needs hardware VLAN pop or VXLAN/GRE/NVGRE decapsulation before the final queue action. Static TX encapsulation/push rules are configured in YAML under `tx.flows`. @@ -696,7 +703,7 @@ names that mostly omit the trailing underscore from the C++ member name (e.g. | `MemoryRegionConfig` | Memory region kind, affinity, access flags, sizes, counts, and ownership. | | `VlanActionConfig` | VLAN push parameters: VLAN ID, priority, DEI, and ethertype. | | `TunnelConfig` | VXLAN, GRE, or NVGRE tunnel template fields for hardware encap/decap actions. | -| `FlowAction` | Flow action type, queue target ID, optional VLAN config, and optional tunnel config. | +| `FlowAction` | Flow action type, scalar queue target `id`, queue-list target `ids`, optional VLAN config, and optional tunnel config. | | `FlowMatch` | Flow match fields for UDP, IPv4, and flex item matching. | | `FlowConfig` | Static named flow rule combining legacy `action`, ordered `actions`, and match fields. | | `FlowRuleConfig` | Dynamic RX flow rule combining legacy `action`, ordered `actions`, and match fields. | diff --git a/docs/benchmarks/raw_benchmarking.md b/docs/benchmarks/raw_benchmarking.md index b9faabe..2f195a7 100644 --- a/docs/benchmarks/raw_benchmarking.md +++ b/docs/benchmarks/raw_benchmarking.md @@ -250,8 +250,9 @@ flow programming test. |------|------------------|----------| | Build smoke | `daqiri_bench_raw_sw_loopback.yaml --seconds 5` | Init succeeds, no NIC flows created | | 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` | +| 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, tests scalar queue rules, then varies UDP source ports and verifies one `[0, 1]` RSS rule reaches both queues with the expected flow ID and tolerance | +| Bad queue target | Copy `daqiri_bench_raw_tx_rx.yaml`; try `flows[0].action.id: 99`, `ids: []`, duplicate/unknown IDs, or both `id` and `ids` | Fails validation before NIC initialization | +| Static RSS flow | Change a standard IPv4/UDP flow action to `{type: queue, ids: [0, 1]}` and generate many distinct five tuples | Both queues receive packets; an unchanged tuple remains on one queue. Validate per-queue rates with `mlnx_perf` | | 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 | diff --git a/docs/concepts.md b/docs/concepts.md index a6fd534..d61c920 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -247,7 +247,8 @@ buffers (CPU hugepages, GPU device memory, or pinned host memory). ### Flow A **flow** is a match pattern paired with one or more actions. The -common RX action is to steer matching packets into a specific queue. For +common RX action is to steer matching packets into one queue or a list of +queues. For example, all UDP-destination-port-4096 packets can be routed into a queue backed by GPU memory. Matching and the resulting actions both run entirely in NIC hardware. @@ -274,12 +275,21 @@ application buffers as pre-encap packets and change only the wire frame. Dynamic RX flows use the same ordered action model for runtime decap/pop rules, while TX transform flows remain static startup configuration. +A queue action with two or more queue IDs enables **receive-side scaling +(RSS)**. The NIC computes a Toeplitz hash from the IPv4/UDP five tuple and uses +it to select one requested queue. This is flow-affine: every packet in an +unchanged flow stays on one queue, preserving per-flow ordering. RSS is not +packet striping, so roughly even queue packet counts require enough distinct +tuples with reasonably balanced traffic. Tunnel-decap rules hash the inner +tuple; flex-item rules still hash the IPv4/UDP tuple rather than the flex value. +eCPRI flows do not support multi-queue RSS. + ### Flow Steering **Flow steering** is the NIC-level mechanism that classifies an -incoming packet against the configured flows and writes it into the -matching queue's buffer, entirely in hardware. Multi-queue RX works by -routing each flow to a separate queue for parallel processing. +incoming packet against the configured flows and writes it into a selected +queue's buffer, entirely in hardware. Multi-queue RX can use separate scalar +rules or one RSS rule for parallel processing. For Raw Ethernet, flow steering is implemented on top of RTE Flow in the DPDK engine and mlx5 Direct Rules in the ibverbs engine. Flow rules are diff --git a/docs/getting-started.md b/docs/getting-started.md index eb78eeb..7a37de8 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -220,8 +220,10 @@ passing the numeric `level` and option constants from system headers. DAQIRI doe not maintain symbolic socket-option mappings in YAML. For Raw Ethernet (`stream_type: "raw"`), `daqiri_init()` validates that each `rx.flows` -entry's legacy `action.id` or final ordered `actions:` queue action matches an -`rx.queues` ID on the same interface, then programs flow rules into the NIC. +entry's legacy scalar `action.id` or queue-list `action.ids` (including the final +queue action in ordered `actions:`) references configured `rx.queues` IDs on the +same interface, then programs flow rules into the NIC. Two or more queue IDs +automatically enable flow-affine IPv4/UDP five-tuple RSS. Initialization fails if any RX flow rule, TX transform flow, send-to-kernel fallback (when `flow_isolation: true`), or `tx_eth_src` offload rule cannot be installed. Raw DPDK and raw ibverbs can offload VLAN push/pop and VXLAN, GRE, or NVGRE diff --git a/docs/images/daqiri_rx_path.svg b/docs/images/daqiri_rx_path.svg index 75bff4e..31b57d5 100644 --- a/docs/images/daqiri_rx_path.svg +++ b/docs/images/daqiri_rx_path.svg @@ -339,7 +339,7 @@ ① Number of memory_regions per queue picks the receive mode (1 = batched / CPU-only, 2 = header-data split). See docs/configuration.md §Memory Regions. - ② Same flow rule schema serves both columns: rx.flows[*].action.id chooses the queue; mode is determined by that queue's memory_regions list. + ② Same flow rule schema serves both columns: rx.flows[*].action.id(s) chooses one queue or an RSS queue list; receive mode is determined by each queue's memory_regions list. ③ On integrated GPUs (e.g. NVIDIA GB10 / DGX Spark) substitute kind: host_pinned for device — the NIC can't peer-DMA into iGPU VRAM. diff --git a/docs/tutorials/configuration-walkthrough.md b/docs/tutorials/configuration-walkthrough.md index 54c99ac..155a3ed 100644 --- a/docs/tutorials/configuration-walkthrough.md +++ b/docs/tutorials/configuration-walkthrough.md @@ -107,7 +107,7 @@ For a shorter selection guide, start with the [Benchmarking overview](../benchma ??? question "4. I need flow-based load balancing across multiple RX queues" - **Closed-loop TX+RX with four queues**: [`daqiri_bench_raw_tx_rx_4q.yaml`](https://github.com/nvidia/daqiri/blob/main/examples/daqiri_bench_raw_tx_rx_4q.yaml) (runs on `daqiri_bench_raw_gpudirect`). - [`daqiri_bench_raw_rx_multi_q.yaml`](https://github.com/nvidia/daqiri/blob/main/examples/daqiri_bench_raw_rx_multi_q.yaml) (runs on `daqiri_bench_raw_gpudirect`). - - **Dynamic RX flow lifecycle**: [`daqiri_example_dynamic_rx_flow.yaml`](https://github.com/nvidia/daqiri/blob/main/examples/daqiri_example_dynamic_rx_flow.yaml) (runs on `daqiri_example_dynamic_rx_flow`). Starts with `flow_isolation: true` and no configured flows, then dynamically routes one UDP flow to RX queue 0 and queue 1 in sequence. + - **Dynamic RX flow lifecycle**: [`daqiri_example_dynamic_rx_flow.yaml`](https://github.com/nvidia/daqiri/blob/main/examples/daqiri_example_dynamic_rx_flow.yaml) (runs on `daqiri_example_dynamic_rx_flow`). Starts with `flow_isolation: true` and no configured flows, dynamically routes traffic to RX queue 0 and queue 1 in sequence, then installs one RSS flow across both queues. The four-queue TX+RX config is self-contained and maps each `bench_tx`/`bench_rx` list entry to the matching DAQIRI queue. The RX-only config is for an external traffic source. The dynamic-flow example demonstrates queues-only startup and runtime flow insertion/deletion. All three demonstrate flow-rule-based routing across multiple RX queues, with explicit CPU cores for both DAQIRI queue workers and benchmark application workers. @@ -248,7 +248,7 @@ bench_tx: # (25)! 19. **`flow_isolation`** · `boolean` · *default: `false`*: When `true`, static startup flows send unmatched traffic in their flow class back to the Linux kernel via fallback rules, while queues-only dynamic configs deliver no traffic to DAQIRI queues until dynamic rules are installed. Useful for letting the interface still handle ARP, ICMP, etc. while DAQIRI takes the application packets. When `false`, every packet hitting the interface must be processed (or dropped) by your application. 20. The list of static startup flows. Flows route packets to a queue based on packet fields. Queues-only configs may omit this list and add dynamic RX flows after initialization. For Raw Ethernet, each configured rule is programmed into the NIC during `daqiri_init()`, and initialization fails if any rule or the send-to-kernel fallback (when `flow_isolation: true`) cannot be installed. Per interface, use only standard UDP/IP flows or only flex-item flows, not both. 21. **`id`** · `integer` · *required*: Tag attached to packets that match this flow. Useful when multiple flows route to a single queue and the application needs to distinguish which rule matched. -22. What to do with packets that match this flow. Existing configs can use the legacy single `action:` map with `type: queue` to send packets to the queue with the given `id`. New hardware transform flows use ordered `actions:`; RX transform flows can `vlan_pop` or `tunnel_decap` and must end with `type: queue`. That queue `id` must match an `rx.queues` entry on this interface; `daqiri_init()` rejects unknown queue IDs during config validation. +22. What to do with packets that match this flow. Existing configs can use the legacy single `action:` map with `type: queue` to send packets to the queue with the given `id`. Use `ids: [0, 1]` instead to enable flow-affine IPv4/UDP five-tuple RSS across a non-empty list of unique configured queues; do not specify both `id` and `ids`. New hardware transform flows use ordered `actions:`; RX transform flows can `vlan_pop` or `tunnel_decap` and must end with `type: queue`. An unchanged five tuple remains on one queue, so roughly even packet counts require many reasonably balanced flows rather than one high-rate flow. 23. :material-package-variant: List of rules to match packets against. **All** rules must hold for a packet to match the flow. Currently supported keys: `udp_src` / `udp_dst` (UDP source/destination port numbers, integer), `ipv4_len` (full IPv4 packet length in bytes, integer). **Adjust to match your incoming traffic.** 24. The `bench_rx` section is specific to the benchmark application. It is a list of application worker configs. List entries map to DAQIRI RX queues on the named interface, either by explicit `queue_id` or by queue-list order when `queue_id` is omitted. `cpu_core` pins the benchmark application's RX worker thread. It is separate from the DAQIRI queue `cpu_core` above, and can use the same core only when you intentionally want to share. In this base config there is one RX queue, so there is one entry. Other DAQIRI binaries (e.g. the reorder-quantize bench) may add fields here. See those configs for details. 25. :material-package-variant: The `bench_tx` section configures the TX side of the benchmark: the benchmark application's TX worker core, packet sizes, and the Ethernet/IP/UDP header fields embedded in outgoing packets. It is a list for the same reason as `bench_rx`: each entry maps to a DAQIRI TX queue on the named interface, either by explicit `queue_id` or by queue-list order when `queue_id` is omitted. The `cpu_core` field pins the application TX thread. The `eth_dst_addr` is required when `flow_isolation` is enabled on the RX interface. The `ip_src_addr` / `ip_dst_addr` are only needed when traffic is routed across subnets. For a direct cable loopback, any value works. The `payload_size`, `header_size`, and UDP ports should match your application's packet format. Hardware TX VLAN push or tunnel encapsulation is configured under `daqiri.cfg.interfaces[].tx.flows`, not in `bench_tx`. Application buffers remain the pre-encap packet. diff --git a/examples/README.md b/examples/README.md index 0d1210b..2f4243c 100644 --- a/examples/README.md +++ b/examples/README.md @@ -8,8 +8,8 @@ Standalone benchmark applications for testing performance of DAQIRI with various - `daqiri_bench_raw_reorder_quantize`: raw RX sequence reorder with payload conversion - `daqiri_example_pcap_writer`: RX pcap writer with optional GPUDirect demo TX traffic - `daqiri_example_dynamic_rx_flow`: raw TX/RX example that starts with RX flow - isolation and no configured flows, then dynamically steers one UDP flow to - queues 0 and 1 in sequence + isolation and no configured flows, dynamically steers UDP traffic to queues 0 + and 1 in sequence, then verifies five-tuple RSS across both queues and flow IDs - `daqiri_bench_rdma`: RDMA benchmark logic (former `rdma_bench.h`) - `daqiri_bench_socket`: TCP/UDP socket benchmark logic - `daqiri_example_gds_write`: one-shot capture that demonstrates synchronous and diff --git a/examples/dynamic_rx_flow_example.cpp b/examples/dynamic_rx_flow_example.cpp index ee2ba76..c6d081d 100644 --- a/examples/dynamic_rx_flow_example.cpp +++ b/examples/dynamic_rx_flow_example.cpp @@ -30,6 +30,7 @@ #include #include #include +#include #include #include "raw_bench_common.h" @@ -52,6 +53,7 @@ struct RxCounter { std::atomic packets{0}; std::atomic bytes{0}; std::atomic bursts{0}; + std::atomic unexpected_flow_ids{0}; }; Args parse_args(int argc, char** argv) { @@ -109,9 +111,8 @@ uint16_t single_udp_port(const std::string& spec, const char* field) { return ports.front(); } -void rx_worker(const daqiri::bench::RawBenchRxConfig& cfg, - RxCounter& counter, - std::atomic& stop) { +void rx_worker(const daqiri::bench::RawBenchRxConfig& cfg, RxCounter& counter, + const std::atomic& expected_flow_id, std::atomic& stop) { const int port_id = daqiri::get_port_id(cfg.interface_name); if (port_id < 0) { std::cerr << "Invalid RX interface_name: " << cfg.interface_name << "\n"; @@ -127,8 +128,16 @@ void rx_worker(const daqiri::bench::RawBenchRxConfig& cfg, continue; } - counter.packets.fetch_add(static_cast(daqiri::get_num_packets(burst)), - std::memory_order_relaxed); + const int num_packets = daqiri::get_num_packets(burst); + const daqiri::FlowId expected = expected_flow_id.load(std::memory_order_relaxed); + if (expected != 0) { + for (int i = 0; i < num_packets; ++i) { + if (daqiri::get_packet_flow_id(burst, i) != expected) { + counter.unexpected_flow_ids.fetch_add(1, std::memory_order_relaxed); + } + } + } + counter.packets.fetch_add(static_cast(num_packets), std::memory_order_relaxed); counter.bytes.fetch_add(daqiri::get_burst_tot_byte(burst), std::memory_order_relaxed); counter.bursts.fetch_add(1, std::memory_order_relaxed); daqiri::free_all_packets_and_burst_rx(burst); @@ -207,11 +216,19 @@ void tx_worker(const daqiri::bench::RawBenchTxConfig& cfg, for (int i = 0; i < num_pkts; ++i) { void* pkt = daqiri::get_segment_packet_ptr(msg, 0, i); if (initialized_tx_buffers.insert(pkt).second) { - if (cudaMemcpyAsync(pkt, - packet_template.data(), - packet_template.size(), - cudaMemcpyHostToDevice, - nullptr) != cudaSuccess || + constexpr uint32_t kFirstDynamicPort = 1024; + constexpr uint32_t kDynamicPortCount = 65536 - kFirstDynamicPort; + const uint32_t base_port = udp_src >= kFirstDynamicPort ? udp_src - kFirstDynamicPort : 0; + const auto source_port = static_cast( + kFirstDynamicPort + + ((base_port + initialized_tx_buffers.size() - 1) % kDynamicPortCount)); + std::vector packet = packet_template; + daqiri::bench::populate_udp_ipv4_headers(packet.data(), cfg.header_size, cfg.payload_size, + eth_src, eth_dst, ip_src, ip_dst, source_port, + udp_dst); + daqiri::bench::finalize_udp_ipv4_checksums(packet.data()); + if (cudaMemcpyAsync(pkt, packet.data(), packet.size(), cudaMemcpyHostToDevice, nullptr) != + cudaSuccess || cudaEventRecord(copy_done, nullptr) != cudaSuccess || cudaEventSynchronize(copy_done) != cudaSuccess) { initialized_tx_buffers.erase(pkt); @@ -280,16 +297,12 @@ daqiri::FlowId wait_for_flow_op(daqiri::FlowOpId op_id, throw std::runtime_error("timed out waiting for dynamic flow operation"); } -daqiri::FlowId add_dynamic_udp_flow(int port_id, - uint16_t queue_id, - uint16_t udp_src, - uint16_t udp_dst) { +daqiri::FlowId add_dynamic_udp_flow(int port_id, uint16_t queue_id, uint16_t udp_dst) { daqiri::FlowRuleConfig flow; flow.name_ = "dynamic_udp_" + std::to_string(udp_dst) + "_q" + std::to_string(queue_id); flow.action_.type_ = daqiri::FlowType::QUEUE; flow.action_.id_ = queue_id; flow.match_.type_ = daqiri::FlowMatchType::IPV4_UDP; - flow.match_.udp_src_ = udp_src; flow.match_.udp_dst_ = udp_dst; daqiri::FlowOpId op_id = 0; @@ -300,6 +313,23 @@ daqiri::FlowId add_dynamic_udp_flow(int port_id, return wait_for_flow_op(op_id, daqiri::FlowOpType::ADD_RX, std::chrono::seconds(5)); } +daqiri::FlowId add_dynamic_udp_rss_flow(int port_id, std::vector queue_ids, + uint16_t udp_dst) { + daqiri::FlowRuleConfig flow; + flow.name_ = "dynamic_udp_" + std::to_string(udp_dst) + "_rss"; + flow.action_.type_ = daqiri::FlowType::QUEUE; + flow.action_.ids_ = std::move(queue_ids); + flow.match_.type_ = daqiri::FlowMatchType::IPV4_UDP; + flow.match_.udp_dst_ = udp_dst; + + daqiri::FlowOpId op_id = 0; + const auto status = daqiri::add_rx_flow_async(port_id, flow, &op_id); + if (status != daqiri::Status::SUCCESS) { + throw std::runtime_error("add_rx_flow_async RSS rule was not accepted"); + } + return wait_for_flow_op(op_id, daqiri::FlowOpType::ADD_RX, std::chrono::seconds(5)); +} + void delete_dynamic_flow(daqiri::FlowId flow_id) { daqiri::FlowOpId op_id = 0; const auto status = daqiri::delete_flow_async(flow_id, &op_id); @@ -315,12 +345,12 @@ uint64_t packets(const RxCounter& counter) { void print_rx_counter(const daqiri::bench::RawBenchRxConfig& cfg, const RxCounter& counter) { - std::cout << "RX complete: interface=" << cfg.interface_name - << " queue=" << cfg.queue_id + std::cout << "RX complete: interface=" << cfg.interface_name << " queue=" << cfg.queue_id << " packets=" << counter.packets.load(std::memory_order_relaxed) << " bytes=" << counter.bytes.load(std::memory_order_relaxed) << " bursts=" << counter.bursts.load(std::memory_order_relaxed) - << "\n"; + << " unexpected_flow_ids=" + << counter.unexpected_flow_ids.load(std::memory_order_relaxed) << "\n"; } } // namespace @@ -328,6 +358,7 @@ void print_rx_counter(const daqiri::bench::RawBenchRxConfig& cfg, int main(int argc, char** argv) { bool daqiri_initialized = false; std::atomic stop{false}; + std::atomic expected_flow_id{0}; std::thread tx_thread; std::vector rx_threads; @@ -359,14 +390,16 @@ int main(int argc, char** argv) { std::vector counters(2); rx_threads.reserve(rx_configs.size()); for (size_t i = 0; i < rx_configs.size(); ++i) { - rx_threads.emplace_back(rx_worker, std::cref(rx_configs[i]), std::ref(counters[i]), std::ref(stop)); + rx_threads.emplace_back(rx_worker, std::cref(rx_configs[i]), std::ref(counters[i]), + std::cref(expected_flow_id), std::ref(stop)); } daqiri::bench::TokenBucketPacer pacer(args.target_gbps); tx_thread = std::thread(tx_worker, std::cref(tx_configs[0]), std::ref(pacer), std::ref(stop)); - std::cout << "Initial drop window: UDP " << udp_src << " -> " << udp_dst - << " has no configured flow for " << args.drop_ms << " ms\n"; + std::cout << "Initial drop window: varied UDP source ports (base " << udp_src + << ") -> destination " << udp_dst << " have no configured flow for " << args.drop_ms + << " ms\n"; std::this_thread::sleep_for(std::chrono::milliseconds(args.drop_ms)); const uint64_t dropped_window_packets = packets(counters[0]) + packets(counters[1]); if (dropped_window_packets != 0) { @@ -374,18 +407,35 @@ int main(int argc, char** argv) { } std::cout << "Adding dynamic UDP flow to RX queue 0 for " << args.active_ms << " ms\n"; - const daqiri::FlowId q0_flow = add_dynamic_udp_flow(rx_port_id, 0, udp_src, udp_dst); + const daqiri::FlowId q0_flow = add_dynamic_udp_flow(rx_port_id, 0, udp_dst); + expected_flow_id.store(q0_flow, std::memory_order_relaxed); std::this_thread::sleep_for(std::chrono::milliseconds(args.active_ms)); delete_dynamic_flow(q0_flow); + expected_flow_id.store(0, std::memory_order_relaxed); std::this_thread::sleep_for(std::chrono::milliseconds(args.drain_ms)); const uint64_t q0_after = packets(counters[0]); const uint64_t q1_after_q0 = packets(counters[1]); std::cout << "Adding dynamic UDP flow to RX queue 1 for " << args.active_ms << " ms\n"; - const daqiri::FlowId q1_flow = add_dynamic_udp_flow(rx_port_id, 1, udp_src, udp_dst); + const daqiri::FlowId q1_flow = add_dynamic_udp_flow(rx_port_id, 1, udp_dst); + expected_flow_id.store(q1_flow, std::memory_order_relaxed); std::this_thread::sleep_for(std::chrono::milliseconds(args.active_ms)); delete_dynamic_flow(q1_flow); + expected_flow_id.store(0, std::memory_order_relaxed); std::this_thread::sleep_for(std::chrono::milliseconds(args.drain_ms)); + const uint64_t q0_before_rss = packets(counters[0]); + const uint64_t q1_before_rss = packets(counters[1]); + + std::cout << "Adding dynamic UDP RSS flow to RX queues [0, 1] for " << args.active_ms + << " ms\n"; + const daqiri::FlowId rss_flow = add_dynamic_udp_rss_flow(rx_port_id, {0, 1}, udp_dst); + expected_flow_id.store(rss_flow, std::memory_order_relaxed); + std::this_thread::sleep_for(std::chrono::milliseconds(args.active_ms)); + delete_dynamic_flow(rss_flow); + expected_flow_id.store(0, std::memory_order_relaxed); + std::this_thread::sleep_for(std::chrono::milliseconds(args.drain_ms)); + const uint64_t q0_rss_packets = packets(counters[0]) - q0_before_rss; + const uint64_t q1_rss_packets = packets(counters[1]) - q1_before_rss; stop.store(true); if (tx_thread.joinable()) { @@ -400,25 +450,30 @@ int main(int argc, char** argv) { print_rx_counter(rx_configs[0], counters[0]); print_rx_counter(rx_configs[1], counters[1]); - const uint64_t q0_packets = packets(counters[0]); - const uint64_t q1_packets = packets(counters[1]); - if (q0_packets == 0 || q1_packets == 0) { - throw std::runtime_error("Expected nonzero packet counts on both RX queues"); + if (q0_rss_packets == 0 || q1_rss_packets == 0) { + throw std::runtime_error("Expected the RSS phase to deliver packets to both RX queues"); } if (q1_after_q0 != 0) { throw std::runtime_error("Queue 1 received packets while only the queue 0 flow was active"); } - if (q0_packets < q0_after) { - throw std::runtime_error("Queue 0 packet counter regressed"); + if (q0_after == 0 || q1_before_rss == q1_after_q0) { + throw std::runtime_error("Expected both scalar queue phases to receive packets"); + } + if (q0_before_rss != q0_after) { + throw std::runtime_error("Queue 0 received packets while only the queue 1 flow was active"); + } + if (counters[0].unexpected_flow_ids.load(std::memory_order_relaxed) != 0 || + counters[1].unexpected_flow_ids.load(std::memory_order_relaxed) != 0) { + throw std::runtime_error("Received packets with an unexpected dynamic flow ID"); } - const auto min_packets = static_cast(std::min(q0_packets, q1_packets)); - const auto max_packets = static_cast(std::max(q0_packets, q1_packets)); + const auto min_packets = static_cast(std::min(q0_rss_packets, q1_rss_packets)); + const auto max_packets = static_cast(std::max(q0_rss_packets, q1_rss_packets)); const double relative_delta = (max_packets - min_packets) / max_packets; - std::cout << "Queue packet relative delta=" << relative_delta - << " tolerance=" << args.tolerance << "\n"; + std::cout << "RSS queue packets: q0=" << q0_rss_packets << " q1=" << q1_rss_packets + << " relative_delta=" << relative_delta << " tolerance=" << args.tolerance << "\n"; if (relative_delta > args.tolerance) { - throw std::runtime_error("RX queue packet counts are not similar"); + throw std::runtime_error("RSS queue packet counts are not similar"); } daqiri::print_stats(); diff --git a/include/daqiri/types.h b/include/daqiri/types.h index 14b9fbb..c167254 100644 --- a/include/daqiri/types.h +++ b/include/daqiri/types.h @@ -731,6 +731,10 @@ struct FlowAction { uint16_t id_ = 0; VlanActionConfig vlan_; TunnelConfig tunnel_; + // Kept after the legacy fields so positional scalar aggregate initializers + // remain source-compatible. A non-empty list replaces id_ as the queue + // destination; two or more entries request flow-affine RSS. + std::vector ids_; }; struct FlexItemMatch { @@ -809,6 +813,13 @@ inline std::vector flow_rule_actions(const FlowRuleConfig& flow) { return {flow.action_}; } +inline std::vector flow_config_actions(const FlowConfig& flow) { + if (!flow.actions_.empty()) { + return flow.actions_; + } + return {flow.action_}; +} + inline FlowAction flow_queue_action(const std::vector& actions) { auto it = std::find_if(actions.begin(), actions.end(), [](const FlowAction& action) { return action.type_ == FlowType::QUEUE; @@ -816,12 +827,23 @@ inline FlowAction flow_queue_action(const std::vector& actions) { return it == actions.end() ? FlowAction{} : *it; } +inline std::vector flow_queue_ids(const FlowAction& action) { + if (!action.ids_.empty()) { + return action.ids_; + } + return {action.id_}; +} + +inline bool flow_queue_action_uses_rss(const FlowAction& action) { + return action.type_ == FlowType::QUEUE && action.ids_.size() > 1; +} + inline bool flow_actions_have_transform(const std::vector& actions) { return std::any_of(actions.begin(), actions.end(), flow_action_is_transform); } inline bool flow_has_transform_actions(const FlowConfig& flow) { - return flow_actions_have_transform(flow.actions_); + return flow_actions_have_transform(flow_config_actions(flow)); } inline bool flow_rule_has_transform_actions(const FlowRuleConfig& flow) { @@ -873,7 +895,7 @@ inline size_t flow_max_decap_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_) { + for (const auto& action : flow_config_actions(flow)) { per_flow += flow_decap_wire_overhead(action); } overhead = std::max(overhead, per_flow); @@ -885,7 +907,7 @@ 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_) { + for (const auto& action : flow_config_actions(flow)) { per_flow += flow_action_wire_overhead(action); } overhead = std::max(overhead, per_flow); diff --git a/python/daqiri_common_pybind.cpp b/python/daqiri_common_pybind.cpp index a25f0fb..0481095 100644 --- a/python/daqiri_common_pybind.cpp +++ b/python/daqiri_common_pybind.cpp @@ -771,6 +771,7 @@ void bind_config_types(py::module_ &m) { .def(py::init<>()) .def_readwrite("type", &FlowAction::type_) .def_readwrite("id", &FlowAction::id_) + .def_readwrite("ids", &FlowAction::ids_) .def_readwrite("vlan", &FlowAction::vlan_) .def_readwrite("tunnel", &FlowAction::tunnel_); diff --git a/src/common.cpp b/src/common.cpp index f8666c1..e557844 100644 --- a/src/common.cpp +++ b/src/common.cpp @@ -164,8 +164,28 @@ bool parse_flow_action_config(const YAML::Node& action_node, FlowAction& action) } if (action.type_ == FlowType::QUEUE) { - if (!parse_u16_field(action_node, "id", &action.id_)) { - DAQIRI_LOG_ERROR("Queue flow action requires integer 'id'"); + if (action_node["id"] && action_node["ids"]) { + DAQIRI_LOG_ERROR("Queue flow action must use either 'id' or 'ids', not both"); + return false; + } + action.ids_.clear(); + if (action_node["ids"]) { + const YAML::Node ids = action_node["ids"]; + if (!ids.IsSequence() || ids.size() == 0) { + DAQIRI_LOG_ERROR("Queue flow action 'ids' must be a non-empty sequence"); + return false; + } + action.ids_.reserve(ids.size()); + for (const auto& id_node : ids) { + uint64_t parsed = 0; + if (!parse_u64_scalar(id_node, &parsed) || parsed > std::numeric_limits::max()) { + DAQIRI_LOG_ERROR("Queue flow action 'ids' entries must be 16-bit integers"); + return false; + } + action.ids_.push_back(static_cast(parsed)); + } + } else if (!parse_u16_field(action_node, "id", &action.id_)) { + DAQIRI_LOG_ERROR("Queue flow action requires integer 'id' or non-empty 'ids'"); return false; } return true; diff --git a/src/engine.cpp b/src/engine.cpp index 73be9bc..b92cca4 100644 --- a/src/engine.cpp +++ b/src/engine.cpp @@ -541,6 +541,11 @@ bool Engine::validate_config() const { for (const auto& mr : cfg_.mrs_) { mr_names.emplace(mr.second.name_); } for (const auto& intf : cfg_.ifs_) { + std::set rx_queue_ids; + for (const auto& rxq : intf.rx_.queues_) { + rx_queue_ids.insert(rxq.common_.id_); + } + std::set rss_flow_ids; size_t max_rx_payload_frame = 0; size_t max_tx_payload_frame = 0; auto queue_frame_size = [&](const CommonQueueConfig& queue) { @@ -563,20 +568,40 @@ bool Engine::validate_config() const { } for (const auto& flow : intf.rx_.flows_) { - if (flow.actions_.empty()) { - DAQIRI_LOG_ERROR("RX flow '{}' on interface '{}' has no actions", flow.name_, intf.name_); - pass = false; - continue; - } - if (flow.actions_.size() > kMaxFlowActions) { + const auto actions = flow_config_actions(flow); + if (actions.size() > kMaxFlowActions) { DAQIRI_LOG_ERROR("RX flow '{}' on interface '{}' has {} actions; maximum supported is {}", - flow.name_, intf.name_, flow.actions_.size(), kMaxFlowActions); + flow.name_, intf.name_, actions.size(), kMaxFlowActions); pass = false; } - if (flow.actions_.back().type_ != FlowType::QUEUE) { + if (actions.back().type_ != FlowType::QUEUE) { DAQIRI_LOG_ERROR("RX flow '{}' on interface '{}' must end with a queue action", flow.name_, intf.name_); pass = false; + } else { + const FlowAction& queue_action = actions.back(); + const auto queue_ids = flow_queue_ids(queue_action); + std::set unique_ids; + for (const uint16_t queue_id : queue_ids) { + if (!unique_ids.insert(queue_id).second) { + DAQIRI_LOG_ERROR("RX flow '{}' on interface '{}' repeats queue id {}", flow.name_, + intf.name_, queue_id); + pass = false; + } + if (rx_queue_ids.find(queue_id) == rx_queue_ids.end()) { + DAQIRI_LOG_ERROR("RX flow '{}' references unknown RX queue {} on interface '{}'", + flow.name_, queue_id, intf.name_); + pass = false; + } + } + if (queue_ids.size() > 1) { + rss_flow_ids.insert(flow.id_); + if (flow.match_.type_ == FlowMatchType::ECPRI) { + DAQIRI_LOG_ERROR("RX flow '{}' on interface '{}' cannot use RSS with eCPRI matching", + flow.name_, intf.name_); + pass = false; + } + } } const bool has_transform = flow_has_transform_actions(flow); if (has_transform && flow.match_.type_ == FlowMatchType::FLEX_ITEM) { @@ -592,7 +617,7 @@ bool Engine::validate_config() const { pass = false; } size_t rx_overhead = 0; - for (const auto& action : flow.actions_) { + for (const auto& action : actions) { if (action.type_ == FlowType::VLAN_PUSH || action.type_ == FlowType::TUNNEL_ENCAP) { DAQIRI_LOG_ERROR("RX flow '{}' on interface '{}' can only use decap/pop transform " "actions before queue", @@ -609,15 +634,23 @@ bool Engine::validate_config() const { } } - for (const auto& flow : intf.tx_.flows_) { - if (flow.actions_.empty()) { - DAQIRI_LOG_ERROR("TX flow '{}' on interface '{}' has no actions", flow.name_, intf.name_); - pass = false; - continue; + for (const auto& reorder : intf.rx_.reorder_configs_) { + for (const FlowId flow_id : reorder.flow_ids_) { + if (rss_flow_ids.find(flow_id) != rss_flow_ids.end()) { + DAQIRI_LOG_ERROR( + "Reorder config '{}' on interface '{}' references RSS flow ID {}; reorder flows " + "must target exactly one RX queue", + reorder.name_, intf.name_, flow_id); + pass = false; + } } - if (flow.actions_.size() > kMaxFlowActions) { + } + + for (const auto& flow : intf.tx_.flows_) { + const auto actions = flow_config_actions(flow); + if (actions.size() > kMaxFlowActions) { DAQIRI_LOG_ERROR("TX flow '{}' on interface '{}' has {} actions; maximum supported is {}", - flow.name_, intf.name_, flow.actions_.size(), kMaxFlowActions); + flow.name_, intf.name_, actions.size(), kMaxFlowActions); pass = false; } if (flow.match_.type_ == FlowMatchType::FLEX_ITEM) { @@ -628,7 +661,7 @@ bool Engine::validate_config() const { } bool has_transform = false; size_t tx_overhead = 0; - for (const auto& action : flow.actions_) { + for (const auto& action : actions) { if (action.type_ == FlowType::QUEUE) { DAQIRI_LOG_ERROR("TX flow '{}' on interface '{}' cannot contain a queue action", flow.name_, intf.name_); diff --git a/src/engines/dpdk/daqiri_dpdk_engine.cpp b/src/engines/dpdk/daqiri_dpdk_engine.cpp index 3f6e050..e012c67 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.cpp +++ b/src/engines/dpdk/daqiri_dpdk_engine.cpp @@ -41,12 +41,43 @@ #include "src/dpdk_log.h" #include "daqiri_dpdk_engine.h" #include "src/kernels.h" +#include "src/rss.h" #include using namespace std::chrono; namespace daqiri { +namespace { + +struct DpdkRxDestination { + explicit DpdkRxDestination(const FlowAction& action, uint32_t rss_level = 0) + : queue_ids(flow_queue_ids(action)) { + queue.index = queue_ids.front(); + rss.func = RTE_ETH_HASH_FUNCTION_TOEPLITZ; + rss.level = rss_level; + rss.types = RTE_ETH_RSS_NONFRAG_IPV4_UDP; + rss.key_len = static_cast(kToeplitzRssKey.size()); + rss.queue_num = static_cast(queue_ids.size()); + rss.key = kToeplitzRssKey.data(); + rss.queue = queue_ids.data(); + } + + void append(struct rte_flow_action actions[], int* index) { + if (queue_ids.size() > 1) { + actions[(*index)++] = {.type = RTE_FLOW_ACTION_TYPE_RSS, .conf = &rss}; + } else { + actions[(*index)++] = {.type = RTE_FLOW_ACTION_TYPE_QUEUE, .conf = &queue}; + } + } + + std::vector queue_ids; + struct rte_flow_action_queue queue {}; + struct rte_flow_action_rss rss {}; +}; + +} // namespace + static bool looks_like_mac_address(const std::string& address) { static const std::regex mac_regex( "^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$", std::regex::ECMAScript); @@ -503,7 +534,8 @@ bool DpdkEngine::init_reorder_queue_state(const InterfaceConfig& intf, const RxQ DAQIRI_LOG_ERROR("Duplicate flow ID {} in interface '{}'", flow.id_, intf.name_); return false; } - flow_id_to_queue[flow.id_] = flow.action_.id_; + flow_id_to_queue[flow.id_] = + flow_queue_ids(flow_queue_action(flow_config_actions(flow))).front(); } for (const auto& reorder_cfg : intf.rx_.reorder_configs_) { @@ -2738,10 +2770,13 @@ void DpdkEngine::initialize() { for (const auto& flow : rx.flows_) { DAQIRI_LOG_INFO("Adding RX flow {}", flow.name_); struct rte_flow* created = nullptr; + const FlowAction queue_action = flow_queue_action(flow_config_actions(flow)); if (flow.match_.type_ == FlowMatchType::FLEX_ITEM) { - 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; } + created = add_flex_item_flow(intf.port_id_, flow.match_.flex_item_match_, queue_action, + 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) { @@ -3177,8 +3212,20 @@ bool DpdkEngine::validate_dynamic_rx_flow(int port, const FlowRuleConfig& flow) return false; } const FlowAction queue_action = flow_queue_action(actions); - if (!is_valid_rx_queue(port, queue_action.id_)) { - DAQIRI_LOG_ERROR("Dynamic RX flow targets invalid port/queue {}/{}", port, queue_action.id_); + const auto queue_ids = flow_queue_ids(queue_action); + std::unordered_set unique_queue_ids; + for (const uint16_t queue_id : queue_ids) { + if (!unique_queue_ids.insert(queue_id).second) { + DAQIRI_LOG_ERROR("Dynamic RX flow '{}' repeats queue id {}", flow.name_, queue_id); + return false; + } + if (!is_valid_rx_queue(port, queue_id)) { + DAQIRI_LOG_ERROR("Dynamic RX flow targets invalid port/queue {}/{}", port, queue_id); + return false; + } + } + if (queue_ids.size() > 1 && flow.match_.type_ == FlowMatchType::ECPRI) { + DAQIRI_LOG_ERROR("Dynamic RX flow '{}' cannot use RSS with eCPRI matching", flow.name_); return false; } const bool has_transform = flow_actions_have_transform(actions); @@ -3515,8 +3562,7 @@ Status DpdkEngine::create_dynamic_flow_legacy_locked(int port, struct rte_flow* rte_flow = nullptr; std::shared_ptr resource; 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); + rte_flow = add_flex_item_flow(port, cfg.match_.flex_item_match_, cfg.action_, cfg.id_, false); } else if (cfg.match_.type_ == FlowMatchType::ECPRI) { rte_flow = add_ecpri_flow(port, cfg, false); } else { @@ -3530,7 +3576,7 @@ Status DpdkEngine::create_dynamic_flow_legacy_locked(int port, DynamicFlowEntry entry; entry.flow_id = flow_id; entry.port = static_cast(port); - entry.queue = flow.action_.id_; + entry.queue = flow_queue_ids(cfg.action_).front(); entry.flow = rte_flow; entry.backend = DynamicFlowBackend::LEGACY; entry.state = DynamicFlowState::ACTIVE; @@ -3594,11 +3640,9 @@ Status DpdkEngine::enqueue_rx_flow_template_create_locked(int port, auto storage = std::make_shared(); build_ipv4_udp_flow_pattern(flow.match_, storage->pattern, &storage->ip_spec, &storage->udp_spec); - build_mark_queue_actions(flow_id, - flow.action_.id_, - storage->action, - &storage->mark, - &storage->queue); + const FlowAction queue_action = flow_queue_action(flow_rule_actions(flow)); + const uint16_t queue_id = flow_queue_ids(queue_action).front(); + build_mark_queue_actions(flow_id, queue_id, storage->action, &storage->mark, &storage->queue); uint8_t template_index = 0; if (!ipv4_udp_flow_template_index(flow.match_, &template_index)) { @@ -3625,7 +3669,7 @@ Status DpdkEngine::enqueue_rx_flow_template_create_locked(int port, DynamicFlowEntry entry; entry.flow_id = flow_id; entry.port = static_cast(port); - entry.queue = flow.action_.id_; + entry.queue = queue_id; entry.flow = rte_flow; entry.backend = DynamicFlowBackend::TEMPLATE; entry.state = DynamicFlowState::ADDING; @@ -4128,7 +4172,9 @@ Status DpdkEngine::add_rx_flow_async(int port, const FlowRuleConfig& flow, FlowO const FlowOpId new_op_id = allocate_flow_op_id(); *op_id = new_op_id; - if (is_ipv4_udp_flow_match(flow.match_) && !flow_rule_has_transform_actions(flow)) { + const FlowAction queue_action = flow_queue_action(flow_rule_actions(flow)); + if (is_ipv4_udp_flow_match(flow.match_) && !flow_rule_has_transform_actions(flow) && + !flow_queue_action_uses_rss(queue_action)) { return add_rx_flow_template_locked(port, flow, flow_id, new_op_id); } return add_rx_flow_legacy_locked(port, flow, flow_id, new_op_id); @@ -4176,7 +4222,9 @@ Status DpdkEngine::add_rx_flows_async(int port, const bool all_ipv4_udp = std::all_of(flows.begin(), flows.end(), [this](const FlowRuleConfig& flow) { - return is_ipv4_udp_flow_match(flow.match_) && !flow_rule_has_transform_actions(flow); + const FlowAction queue_action = flow_queue_action(flow_rule_actions(flow)); + return is_ipv4_udp_flow_match(flow.match_) && !flow_rule_has_transform_actions(flow) && + !flow_queue_action_uses_rss(queue_action); }); if (all_ipv4_udp) { return add_rx_flows_template_locked(port, flows, flow_ids, new_op_id); @@ -4350,18 +4398,15 @@ void DpdkEngine::destroy_owned_flows() { owned_flows_.clear(); } -struct rte_flow* DpdkEngine::add_flex_item_flow( - int port, - const FlexItemMatch& match_info, - uint16_t queue_id, - FlowId mark_id, - bool track) { +struct rte_flow* DpdkEngine::add_flex_item_flow(int port, const FlexItemMatch& match_info, + const FlowAction& queue_action, FlowId mark_id, + bool track) { /* Declaring structs being used. 8< */ 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 = queue_id}; + DpdkRxDestination destination(queue_action); struct rte_flow_action_mark mark = {.id = mark_id}; struct rte_flow_error error; struct rte_flow_item_udp udp_spec; @@ -4385,17 +4430,13 @@ struct rte_flow* DpdkEngine::add_flex_item_flow( memset(&udp_spec, 0, sizeof(struct rte_flow_item_udp)); memset(&udp_mask, 0, sizeof(struct rte_flow_item_udp)); + int action_index = 0; if (mark_id != 0) { - 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; - } else { - action[0].type = RTE_FLOW_ACTION_TYPE_QUEUE; - action[0].conf = &queue; - action[1].type = RTE_FLOW_ACTION_TYPE_END; + action[action_index].type = RTE_FLOW_ACTION_TYPE_MARK; + action[action_index++].conf = &mark; } + destination.append(action, &action_index); + action[action_index].type = RTE_FLOW_ACTION_TYPE_END; pattern[0].type = RTE_FLOW_ITEM_TYPE_ETH; pattern[1].type = RTE_FLOW_ITEM_TYPE_IPV4; @@ -4669,10 +4710,9 @@ bool fill_gre_raw_storage(DpdkEngine::DpdkFlowResource::ActionStorage& storage, } void append_normal_match(struct rte_flow_item* pattern, int* idx, const FlowMatch& match, - struct rte_flow_item_ipv4* ip_spec, - struct rte_flow_item_ipv4* ip_mask, - struct rte_flow_item_udp* udp_spec, - struct rte_flow_item_udp* udp_mask) { + struct rte_flow_item_ipv4* ip_spec, struct rte_flow_item_ipv4* ip_mask, + struct rte_flow_item_udp* udp_spec, struct rte_flow_item_udp* udp_mask, + bool force_ipv4_udp = false) { bool has_ip_match = false; bool has_udp_match = false; if (match.ipv4_len_ > 0) { @@ -4700,7 +4740,10 @@ void append_normal_match(struct rte_flow_item* pattern, int* idx, const FlowMatc udp_mask->hdr.dst_port = 0xffff; has_udp_match = true; } - if (has_udp_match) { has_ip_match = true; } + if (has_udp_match || force_ipv4_udp) { + has_udp_match = true; + has_ip_match = true; + } if (has_ip_match) { pattern[(*idx)++] = {.type = RTE_FLOW_ITEM_TYPE_IPV4, .spec = ip_spec, @@ -4715,8 +4758,8 @@ void append_normal_match(struct rte_flow_item* pattern, int* idx, const FlowMatc } } -int normal_match_pattern_item_count(const FlowMatch& match) { - const bool has_udp_match = match.udp_src_ > 0 || match.udp_dst_ > 0; +int normal_match_pattern_item_count(const FlowMatch& match, bool force_ipv4_udp = false) { + const bool has_udp_match = force_ipv4_udp || match.udp_src_ > 0 || match.udp_dst_ > 0; const bool has_ip_match = has_udp_match || match.ipv4_len_ > 0 || match.ipv4_src_ != INADDR_ANY || match.ipv4_dst_ != INADDR_ANY; @@ -4861,7 +4904,6 @@ bool append_transform_action(struct rte_flow_action* actions, int* idx, } // namespace - // Taken from flow_block.c DPDK example */ struct rte_flow* DpdkEngine::add_flow(int port, const FlowConfig& cfg, @@ -4870,7 +4912,11 @@ struct rte_flow* DpdkEngine::add_flow(int port, struct rte_flow_attr attr {}; struct rte_flow_item pattern[MAX_PATTERN_NUM] {}; struct rte_flow_action action[MAX_ACTION_NUM] {}; - struct rte_flow_action_queue queue = {.index = cfg.action_.id_}; + const auto flow_actions = flow_config_actions(cfg); + const bool hash_inner = std::any_of( + flow_actions.begin(), flow_actions.end(), + [](const FlowAction& flow_action) { return flow_action.type_ == FlowType::TUNNEL_DECAP; }); + DpdkRxDestination destination(flow_queue_action(flow_actions), hash_inner ? 2 : 0); struct rte_flow_action_mark mark = {.id = cfg.id_}; struct rte_flow_error error {}; struct rte_flow_item_udp udp_spec {}; @@ -4882,11 +4928,12 @@ struct rte_flow* DpdkEngine::add_flow(int port, auto resource = std::make_shared(); resource->port = port; - resource->action_storage.reserve(cfg.actions_.size() * 2 + 2); + resource->action_storage.reserve(flow_actions.size() * 2 + 2); - int required_pattern_items = 1 + normal_match_pattern_item_count(cfg.match_) + 1; + const bool use_rss = destination.queue_ids.size() > 1; + int required_pattern_items = 1 + normal_match_pattern_item_count(cfg.match_, use_rss) + 1; int required_action_items = 2 + 1; - for (const auto& flow_action : cfg.actions_) { + for (const auto& flow_action : flow_actions) { required_pattern_items += transform_pattern_item_count(flow_action); required_action_items += transform_action_item_count(flow_action); } @@ -4903,7 +4950,7 @@ struct rte_flow* DpdkEngine::add_flow(int port, int pi = 0; bool has_outer_transform = false; - for (const auto& flow_action : cfg.actions_) { + for (const auto& flow_action : flow_actions) { if (flow_action.type_ == FlowType::TUNNEL_DECAP) { if (!append_tunnel_pattern(pattern, &pi, *resource, flow_action.tunnel_)) { DAQIRI_LOG_CRITICAL("Failed to build tunnel pattern for RX flow '{}'", cfg.name_); @@ -4917,11 +4964,11 @@ struct rte_flow* DpdkEngine::add_flow(int port, } } pattern[pi++].type = RTE_FLOW_ITEM_TYPE_ETH; - append_normal_match(pattern, &pi, cfg.match_, &ip_spec, &ip_mask, &udp_spec, &udp_mask); + append_normal_match(pattern, &pi, cfg.match_, &ip_spec, &ip_mask, &udp_spec, &udp_mask, use_rss); pattern[pi].type = RTE_FLOW_ITEM_TYPE_END; int ai = 0; - for (const auto& flow_action : cfg.actions_) { + for (const auto& flow_action : flow_actions) { if (flow_action.type_ == FlowType::QUEUE) { continue; } if (!append_transform_action(action, &ai, *resource, flow_action)) { DAQIRI_LOG_CRITICAL("Failed to build RX action '{}' for flow '{}'", @@ -4930,7 +4977,7 @@ struct rte_flow* DpdkEngine::add_flow(int port, } } action[ai++] = {.type = RTE_FLOW_ACTION_TYPE_MARK, .conf = &mark}; - action[ai++] = {.type = RTE_FLOW_ACTION_TYPE_QUEUE, .conf = &queue}; + destination.append(action, &ai); action[ai].type = RTE_FLOW_ACTION_TYPE_END; attr.ingress = 1; @@ -4971,14 +5018,15 @@ struct rte_flow* DpdkEngine::add_tx_flow(int port, const FlowConfig& cfg) { struct rte_flow_item_udp udp_mask {}; struct rte_flow_item_ipv4 ip_spec {}; struct rte_flow_item_ipv4 ip_mask {}; + const auto flow_actions = flow_config_actions(cfg); auto resource = std::make_shared(); resource->port = port; - resource->action_storage.reserve(cfg.actions_.size() * 2 + 1); + resource->action_storage.reserve(flow_actions.size() * 2 + 1); const int required_pattern_items = 1 + normal_match_pattern_item_count(cfg.match_) + 1; int required_action_items = 1; - for (const auto& flow_action : cfg.actions_) { + for (const auto& flow_action : flow_actions) { required_action_items += transform_action_item_count(flow_action); } if (required_pattern_items > MAX_PATTERN_NUM) { @@ -4998,7 +5046,7 @@ struct rte_flow* DpdkEngine::add_tx_flow(int port, const FlowConfig& cfg) { pattern[pi].type = RTE_FLOW_ITEM_TYPE_END; int ai = 0; - for (const auto& flow_action : cfg.actions_) { + for (const auto& flow_action : flow_actions) { if (!append_transform_action(action, &ai, *resource, flow_action)) { DAQIRI_LOG_CRITICAL("Failed to build TX action '{}' for flow '{}'", flow_type_to_string(flow_action.type_), cfg.name_); @@ -5039,7 +5087,7 @@ struct rte_flow* DpdkEngine::add_ecpri_flow(int port, const FlowConfig& cfg, boo 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_}; + DpdkRxDestination destination(flow_queue_action(flow_config_actions(cfg))); struct rte_flow_action_mark mark = {.id = cfg.id_}; struct rte_flow_error error; struct rte_flow_item_eth eth_spec; @@ -5062,9 +5110,9 @@ struct rte_flow* DpdkEngine::add_ecpri_flow(int port, const FlowConfig& cfg, boo 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; + int action_index = 1; + destination.append(action, &action_index); + action[action_index].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); @@ -5482,14 +5530,15 @@ bool DpdkEngine::validate_config() const { DAQIRI_LOG_ERROR("Duplicate flow ID {} on interface '{}'", flow.id_, intf.name_); return false; } - if (rx_queue_ids.find(flow.action_.id_) == rx_queue_ids.end()) { - DAQIRI_LOG_ERROR("Flow '{}' references unknown RX queue {} on interface '{}'", - flow.name_, - flow.action_.id_, - intf.name_); - return false; + const auto destination_ids = flow_queue_ids(flow_queue_action(flow_config_actions(flow))); + for (const uint16_t destination_id : destination_ids) { + if (rx_queue_ids.find(destination_id) == rx_queue_ids.end()) { + DAQIRI_LOG_ERROR("Flow '{}' references unknown RX queue {} on interface '{}'", flow.name_, + destination_id, intf.name_); + return false; + } } - flow_to_queue.emplace(flow.id_, flow.action_.id_); + flow_to_queue.emplace(flow.id_, destination_ids.front()); if (flow.match_.type_ == FlowMatchType::FLEX_ITEM) { has_flex_item_flows = true; const uint16_t flex_item_id = flow.match_.flex_item_match_.flex_item_id_; @@ -5868,7 +5917,7 @@ int DpdkEngine::rx_core_multi_q_worker(void* arg) { uint16_t num_queues = tparams->q_params.size(); std::string pq_str = ""; - for (const auto &pq : tparams->q_params) { + for (const auto& pq : tparams->q_params) { pq_str += std::to_string(pq.port) + "/" + std::to_string(pq.queue) + " "; } @@ -6301,7 +6350,9 @@ int DpdkEngine::tx_core_worker(void* arg) { (void*)tparams->ring); while (!force_quit.load()) { - if (rte_ring_dequeue(tparams->ring, reinterpret_cast(&msg)) != 0) { continue; } + if (rte_ring_dequeue(tparams->ring, reinterpret_cast(&msg)) != 0) { + continue; + } // Scatter mode needs to chain all the buffers if (msg->hdr.hdr.num_segs > 1) { diff --git a/src/engines/dpdk/daqiri_dpdk_engine.h b/src/engines/dpdk/daqiri_dpdk_engine.h index b88cb3e..985abd6 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.h +++ b/src/engines/dpdk/daqiri_dpdk_engine.h @@ -212,10 +212,8 @@ class DpdkEngine : public Engine { struct rte_flow* add_modify_flow_set(int port, int queue, const char* buf, int len, Direction direction); - struct rte_flow* add_flex_item_flow(int port, - const FlexItemMatch& match, - uint16_t queue_id, - FlowId mark_id = 0, + struct rte_flow* add_flex_item_flow(int port, const FlexItemMatch& match, + const FlowAction& queue_action, FlowId mark_id = 0, bool track = true); // eCPRI-over-Ethernet RX flow (EtherType 0xAEFE) via RTE_FLOW_ITEM_TYPE_ECPRI. diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp index de4c6de..86aa835 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp @@ -18,6 +18,7 @@ #include "src/engines/ibverbs/daqiri_ibverbs_engine.h" #include "src/engines/ibverbs/mlx5_prm_min.h" #include "src/kernels.h" +#include "src/rss.h" #include #include @@ -1059,14 +1060,23 @@ bool IbverbsEngine::validate_dynamic_rx_flow_locked(int port, const FlowRuleConf return false; } const FlowAction queue_action = flow_queue_action(actions); - - const auto queue_it = std::find_if(intf->rx_.queues_.begin(), intf->rx_.queues_.end(), - [&](const RxQueueConfig& q) { - return q.common_.id_ == queue_action.id_; - }); - if (queue_it == intf->rx_.queues_.end()) { - DAQIRI_LOG_ERROR("Dynamic RX flow targets invalid port/queue {}/{}", port, - queue_action.id_); + const auto queue_ids = flow_queue_ids(queue_action); + std::set unique_queue_ids; + for (const uint16_t queue_id : queue_ids) { + if (!unique_queue_ids.insert(queue_id).second) { + DAQIRI_LOG_ERROR("Dynamic RX flow '{}' repeats queue id {}", flow.name_, queue_id); + return false; + } + const auto queue_it = + std::find_if(intf->rx_.queues_.begin(), intf->rx_.queues_.end(), + [&](const RxQueueConfig& q) { return q.common_.id_ == queue_id; }); + if (queue_it == intf->rx_.queues_.end()) { + DAQIRI_LOG_ERROR("Dynamic RX flow targets invalid port/queue {}/{}", port, queue_id); + return false; + } + } + if (queue_ids.size() > 1 && flow.match_.type_ == FlowMatchType::ECPRI) { + DAQIRI_LOG_ERROR("Dynamic RX flow '{}' cannot use RSS with eCPRI matching", flow.name_); return false; } const bool has_transform = flow_actions_have_transform(actions); @@ -1116,17 +1126,131 @@ bool IbverbsEngine::validate_dynamic_rx_flow_locked(int port, const FlowRuleConf return false; } -bool IbverbsEngine::create_dr_rule_locked(int port, - PortSteering& st, - uint16_t criteria, - struct mlx5dv_flow_match_parameters* mask, - struct mlx5dv_flow_match_parameters* value, - IbvRxQueue* dest, - int priority, - FlowId flow_id, - const char* desc, - DynamicFlowEntry* dynamic_entry, - const std::vector& reformats) { +Status IbverbsEngine::resolve_rx_destination(int port, PortSteering& st, + const FlowAction& queue_action, bool inner, + struct mlx5dv_dr_action** action, + uint16_t* primary_queue, + RssDestinationPtr* rss_destination) { + if (action == nullptr || primary_queue == nullptr || rss_destination == nullptr) { + return Status::NULL_PTR; + } + *action = nullptr; + *primary_queue = 0; + rss_destination->reset(); + + const auto queue_ids = flow_queue_ids(queue_action); + std::vector queues; + queues.reserve(queue_ids.size()); + for (const uint16_t queue_id : queue_ids) { + IbvRxQueue* queue = find_rx_queue(port, queue_id); + if (queue == nullptr || queue->dr_action == nullptr) { + DAQIRI_LOG_ERROR("Flow targets unknown queue id {} on port {}", queue_id, port); + return Status::INVALID_PARAMETER; + } + queues.push_back(queue); + } + *primary_queue = queue_ids.front(); + if (queues.size() == 1) { + *action = queues.front()->dr_action; + return Status::SUCCESS; + } + + std::string cache_key = inner ? "inner:" : "outer:"; + for (const uint16_t queue_id : queue_ids) { + cache_key += std::to_string(queue_id) + ","; + } + auto cached = st.rss_destinations.find(cache_key); + if (cached != st.rss_destinations.end()) { + if (auto destination = cached->second.lock()) { + *action = destination->action; + *rss_destination = std::move(destination); + return Status::SUCCESS; + } + } + + const size_t table_size = rss_table_size(queues.size()); + const size_t rqt_bytes = DEVX_ST_SZ_BYTES(create_rqt_in) + (table_size - 1) * sizeof(uint32_t); + std::vector rqt_in((rqt_bytes + sizeof(uint32_t) - 1) / sizeof(uint32_t), 0); + uint32_t rqt_out[DEVX_ST_SZ_DW(create_rqt_out)] = {0}; + void* rqtc = DEVX_ADDR_OF(create_rqt_in, rqt_in.data(), rqt_context); + DEVX_SET(create_rqt_in, rqt_in.data(), opcode, MLX5_CMD_OP_CREATE_RQT); + DEVX_SET(rqtc, rqtc, rqt_max_size, table_size); + DEVX_SET(rqtc, rqtc, rqt_actual_size, table_size); + auto* rqt_entries = reinterpret_cast(DEVX_ADDR_OF(rqtc, rqtc, rq_num)); + for (size_t i = 0; i < table_size; ++i) { + rqt_entries[i] = htobe32(queues[i % queues.size()]->rqn & 0x00ffffffu); + } + + auto destination = RssDestinationPtr(new RssDestination, [](RssDestination* value) { + if (value->action != nullptr) { + mlx5dv_dr_action_destroy(value->action); + } + if (value->tir_obj != nullptr) { + mlx5dv_devx_obj_destroy(value->tir_obj); + } + if (value->rqt_obj != nullptr) { + mlx5dv_devx_obj_destroy(value->rqt_obj); + } + delete value; + }); + destination->queue_ids = queue_ids; + destination->inner = inner; + destination->rqt_obj = mlx5dv_devx_obj_create(queues.front()->ctx, rqt_in.data(), rqt_bytes, + rqt_out, sizeof(rqt_out)); + if (destination->rqt_obj == nullptr) { + DAQIRI_LOG_ERROR("CREATE_RQT failed for RSS flow on port {}: {} (syndrome 0x{:x})", port, + strerror(errno), DEVX_GET(create_rqt_out, rqt_out, syndrome)); + return Status::GENERIC_FAILURE; + } + const uint32_t rqtn = DEVX_GET(create_rqt_out, rqt_out, rqtn); + + uint32_t tir_in[DEVX_ST_SZ_DW(create_tir_in)] = {0}; + uint32_t tir_out[DEVX_ST_SZ_DW(create_tir_out)] = {0}; + void* tirc = DEVX_ADDR_OF(create_tir_in, tir_in, ctx); + DEVX_SET(create_tir_in, tir_in, opcode, MLX5_CMD_OP_CREATE_TIR); + DEVX_SET(tirc, tirc, disp_type, MLX5_TIRC_DISP_TYPE_INDIRECT); + DEVX_SET(tirc, tirc, indirect_table, rqtn); + DEVX_SET(tirc, tirc, rx_hash_fn, MLX5_RX_HASH_FN_TOEPLITZ); + DEVX_SET(tirc, tirc, transport_domain, queues.front()->td_num); + if (inner) { + DEVX_SET(tirc, tirc, tunneled_offload_en, 1); + } + void* selector = inner ? DEVX_ADDR_OF(tirc, tirc, rx_hash_field_selector_inner) + : DEVX_ADDR_OF(tirc, tirc, rx_hash_field_selector_outer); + DEVX_SET(rx_hash_field_select, selector, l3_prot_type, MLX5_L3_PROT_TYPE_IPV4); + DEVX_SET(rx_hash_field_select, selector, l4_prot_type, MLX5_L4_PROT_TYPE_UDP); + DEVX_SET(rx_hash_field_select, selector, selected_fields, + MLX5_HASH_FIELD_SEL_SRC_IP | MLX5_HASH_FIELD_SEL_DST_IP | MLX5_HASH_FIELD_SEL_L4_SPORT | + MLX5_HASH_FIELD_SEL_L4_DPORT); + void* key = DEVX_ADDR_OF(tirc, tirc, rx_hash_toeplitz_key); + memcpy(key, kToeplitzRssKey.data(), kToeplitzRssKey.size()); + + destination->tir_obj = + mlx5dv_devx_obj_create(queues.front()->ctx, tir_in, sizeof(tir_in), tir_out, sizeof(tir_out)); + if (destination->tir_obj == nullptr) { + DAQIRI_LOG_ERROR("CREATE_TIR failed for RSS flow on port {}: {} (syndrome 0x{:x})", port, + strerror(errno), DEVX_GET(create_tir_out, tir_out, syndrome)); + return Status::GENERIC_FAILURE; + } + destination->action = mlx5dv_dr_action_create_dest_devx_tir(destination->tir_obj); + if (destination->action == nullptr) { + DAQIRI_LOG_ERROR("Failed to create RSS destination action on port {}: {}", port, + strerror(errno)); + return Status::GENERIC_FAILURE; + } + + st.rss_destinations[cache_key] = destination; + *action = destination->action; + *rss_destination = destination; + return Status::SUCCESS; +} + +bool IbverbsEngine::create_dr_rule_locked( + int port, PortSteering& st, uint16_t criteria, struct mlx5dv_flow_match_parameters* mask, + struct mlx5dv_flow_match_parameters* value, struct mlx5dv_dr_action* destination_action, + uint16_t primary_queue, const RssDestinationPtr& rss_destination, int priority, FlowId flow_id, + const char* desc, DynamicFlowEntry* dynamic_entry, + const std::vector& reformats) { struct mlx5dv_dr_matcher* matcher = mlx5dv_dr_matcher_create(st.table, priority, criteria, mask); if (matcher == nullptr) { DAQIRI_LOG_CRITICAL("dr_matcher_create failed (port {} {}): {}", port, desc, strerror(errno)); @@ -1155,7 +1279,7 @@ bool IbverbsEngine::create_dr_rule_locked(int port, } actions[num_actions++] = reformat; } - actions[num_actions++] = dest->dr_action; + actions[num_actions++] = destination_action; struct mlx5dv_dr_rule* rule = mlx5dv_dr_rule_create(matcher, value, num_actions, actions); if (rule == nullptr) { @@ -1167,10 +1291,11 @@ bool IbverbsEngine::create_dr_rule_locked(int port, if (dynamic_entry != nullptr) { dynamic_entry->port = port; - dynamic_entry->queue = static_cast(dest->queue_id); + dynamic_entry->queue = primary_queue; dynamic_entry->matcher = matcher; dynamic_entry->rule = rule; dynamic_entry->tag_action = tag_action; + dynamic_entry->rss_destination = rss_destination; dynamic_entry->state = DynamicFlowState::ACTIVE; return true; } @@ -1179,7 +1304,8 @@ bool IbverbsEngine::create_dr_rule_locked(int port, st.rules.push_back(rule); if (tag_action != nullptr) { st.tag_actions.push_back(tag_action); } - PortSteering::RuleSpec spec{matcher, dest->dr_action, tag_action, reformats, 0, {}}; + PortSteering::RuleSpec spec{ + matcher, destination_action, tag_action, reformats, rss_destination, 0, {}}; spec.value_sz = std::min(value->match_sz, sizeof(spec.value)); memcpy(spec.value, value->match_buf, spec.value_sz); st.rule_specs.push_back(spec); @@ -1201,10 +1327,19 @@ Status IbverbsEngine::install_flow_rule_locked(int port, const auto actions = flow_rule_actions(flow); const FlowAction queue_action = flow_queue_action(actions); - IbvRxQueue* dest = find_rx_queue(port, queue_action.id_); - if (dest == nullptr || dest->dr_action == nullptr) { - DAQIRI_LOG_ERROR("Flow '{}' targets unknown queue id {} on port {}", flow.name_, queue_action.id_, - port); + const bool inner = std::any_of(actions.begin(), actions.end(), [](const FlowAction& action) { + return action.type_ == FlowType::TUNNEL_DECAP; + }); + struct mlx5dv_dr_action* destination_action = nullptr; + uint16_t primary_queue = 0; + RssDestinationPtr rss_destination; + const Status destination_status = resolve_rx_destination( + port, st, queue_action, inner, &destination_action, &primary_queue, &rss_destination); + if (destination_status != Status::SUCCESS) { + return destination_status; + } + IbvRxQueue* primary = find_rx_queue(port, primary_queue); + if (primary == nullptr) { return Status::INVALID_PARAMETER; } @@ -1285,18 +1420,14 @@ Status IbverbsEngine::install_flow_rule_locked(int port, mt.flex_item_match_.val_ & mt.flex_item_match_.mask_); const bool ok = - create_dr_rule_locked(port, - st, - MLX5_DR_MATCH_CRITERIA_OUTER | MLX5_DR_MATCH_CRITERIA_MISC4, + create_dr_rule_locked(port, st, MLX5_DR_MATCH_CRITERIA_OUTER | MLX5_DR_MATCH_CRITERIA_MISC4, reinterpret_cast(&mask), reinterpret_cast(&value), - dest, - priority, - flow_id, - flow.name_.c_str(), - dynamic_entry, - flow_reformats); - if (!ok) { cleanup_dynamic_reformats(); } + destination_action, primary_queue, rss_destination, priority, flow_id, + flow.name_.c_str(), dynamic_entry, flow_reformats); + if (!ok) { + cleanup_dynamic_reformats(); + } return ok ? Status::SUCCESS : Status::GENERIC_FAILURE; } @@ -1306,14 +1437,14 @@ Status IbverbsEngine::install_flow_rule_locked(int port, 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), + primary->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) + return create_dr_rule_locked( + port, st, criteria, reinterpret_cast(&mask), + reinterpret_cast(&value), destination_action, + primary_queue, rss_destination, priority, flow_id, flow.name_.c_str(), dynamic_entry) ? Status::SUCCESS : Status::GENERIC_FAILURE; } @@ -1337,6 +1468,10 @@ Status IbverbsEngine::install_flow_rule_locked(int port, DEVX_SET(fte_match_set_lyr_2_4, value_buf, ip_protocol, MLX5_IP_PROTOCOL_UDP); }; + if (rss_destination != nullptr) { + pin_ipv4_udp(); + } + if (mt.udp_src_ > 0) { pin_ipv4_udp(); DEVX_SET(fte_match_set_lyr_2_4, mask_buf, udp_sport, 0xffff); @@ -1366,9 +1501,8 @@ Status IbverbsEngine::install_flow_rule_locked(int port, if (mt.ipv4_len_ > 0) { if (st.ipv4_len_node.obj == nullptr) { uint32_t sample_id = 0; - struct mlx5dv_devx_obj* node = - create_flex_parser_node(dest->ctx, MLX5_GRAPH_ARC_NODE_MAC, MLX5_ETHERTYPE_IPV4, 0, - &sample_id); + struct mlx5dv_devx_obj* node = create_flex_parser_node(primary->ctx, MLX5_GRAPH_ARC_NODE_MAC, + MLX5_ETHERTYPE_IPV4, 0, &sample_id); if (node == nullptr) { DAQIRI_LOG_CRITICAL("ipv4_len: parse-graph node creation failed on port {}", port); return Status::GENERIC_FAILURE; @@ -1394,19 +1528,14 @@ Status IbverbsEngine::install_flow_rule_locked(int port, return Status::INVALID_PARAMETER; } - const bool ok = - create_dr_rule_locked(port, - st, - criteria, - reinterpret_cast(&mask), - reinterpret_cast(&value), - dest, - priority, - flow_id, - flow.name_.c_str(), - dynamic_entry, - flow_reformats); - if (!ok) { cleanup_dynamic_reformats(); } + const bool ok = create_dr_rule_locked( + port, st, criteria, reinterpret_cast(&mask), + reinterpret_cast(&value), destination_action, + primary_queue, rss_destination, priority, flow_id, flow.name_.c_str(), dynamic_entry, + flow_reformats); + if (!ok) { + cleanup_dynamic_reformats(); + } return ok ? Status::SUCCESS : Status::GENERIC_FAILURE; } @@ -1681,11 +1810,12 @@ Status IbverbsEngine::install_port_flows() { // Build the per-flow rules. Each flow with a queue action steers its matched // 5-tuple to that queue's TIR. A flow that matches no supported field, and // the no-flows case, fall through to a catch-all -> the first queue. - auto add_rule = [&](uint16_t crit, struct mlx5dv_flow_match_parameters* mask, - struct mlx5dv_flow_match_parameters* val, IbvRxQueue* dest, int prio, - uint32_t tag, const char* desc, - const std::vector& reformats = - std::vector{}) -> bool { + auto add_rule = + [&](uint16_t crit, struct mlx5dv_flow_match_parameters* mask, + struct mlx5dv_flow_match_parameters* val, struct mlx5dv_dr_action* destination_action, + const RssDestinationPtr& rss_destination, int prio, uint32_t tag, const char* desc, + const std::vector& reformats = + std::vector{}) -> bool { struct mlx5dv_dr_matcher* m = mlx5dv_dr_matcher_create(st.table, prio, crit, mask); if (m == nullptr) { DAQIRI_LOG_CRITICAL("dr_matcher_create failed (port {} {}): {}", port, desc, @@ -1716,14 +1846,15 @@ Status IbverbsEngine::install_port_flows() { } actions[na++] = reformat; } - actions[na++] = dest->dr_action; + actions[na++] = destination_action; struct mlx5dv_dr_rule* r = mlx5dv_dr_rule_create(m, val, na, actions); if (r == nullptr) { DAQIRI_LOG_CRITICAL("dr_rule_create failed (port {} {}): {}", port, desc, strerror(errno)); return false; } st.rules.push_back(r); - PortSteering::RuleSpec spec{m, dest->dr_action, tag_act, reformats, 0, {}}; + PortSteering::RuleSpec spec{m, destination_action, tag_act, reformats, rss_destination, 0, + {}}; spec.value_sz = std::min(val->match_sz, sizeof(spec.value)); memcpy(spec.value, val->match_buf, spec.value_sz); st.rule_specs.push_back(spec); @@ -1733,18 +1864,24 @@ Status IbverbsEngine::install_port_flows() { int installed = 0; int prio = 0; for (const auto& fl : intf.rx_.flows_) { - if (fl.action_.type_ != FlowType::QUEUE) { + const auto actions = flow_config_actions(fl); + const FlowAction queue_action = flow_queue_action(actions); + if (queue_action.type_ != FlowType::QUEUE) { continue; } - auto it = by_id.find(fl.action_.id_); - if (it == by_id.end()) { - DAQIRI_LOG_ERROR("Flow '{}' targets unknown queue id {} on port {}", fl.name_, - fl.action_.id_, port); - continue; + const bool inner = std::any_of(actions.begin(), actions.end(), [](const FlowAction& action) { + return action.type_ == FlowType::TUNNEL_DECAP; + }); + struct mlx5dv_dr_action* destination_action = nullptr; + uint16_t primary_queue = 0; + RssDestinationPtr rss_destination; + if (resolve_rx_destination(port, st, queue_action, inner, &destination_action, &primary_queue, + &rss_destination) != Status::SUCCESS) { + return Status::GENERIC_FAILURE; } const FlowMatch& mt = fl.match_; std::vector flow_reformats; - for (const auto& action : fl.actions_) { + for (const auto& action : actions) { if (!flow_action_is_transform(action)) { continue; } st.reformat_buffers.emplace_back(); auto& buffer = st.reformat_buffers.back(); @@ -1800,13 +1937,14 @@ Status IbverbsEngine::install_port_flows() { mt.flex_item_match_.val_ & mt.flex_item_match_.mask_); if (!add_rule(MLX5_DR_MATCH_CRITERIA_OUTER | MLX5_DR_MATCH_CRITERIA_MISC4, reinterpret_cast(&fmask), - reinterpret_cast(&fval), it->second, - prio++, fl.id_, fl.name_.c_str(), flow_reformats)) { + reinterpret_cast(&fval), + destination_action, rss_destination, prio++, fl.id_, fl.name_.c_str(), + flow_reformats)) { return Status::GENERIC_FAILURE; } DAQIRI_LOG_INFO( "Flow '{}' (flex item {}) -> queue {} (port {}): udp_dst={} sample==0x{:x}/0x{:x}", - fl.name_, fid, fl.action_.id_, port, fcfg->udp_dst_port_, mt.flex_item_match_.val_, + fl.name_, fid, primary_queue, port, fcfg->udp_dst_port_, mt.flex_item_match_.val_, mt.flex_item_match_.mask_); installed++; continue; @@ -1826,13 +1964,13 @@ Status IbverbsEngine::install_port_flows() { return Status::GENERIC_FAILURE; } if (!add_rule(ecrit, reinterpret_cast(&emask), - reinterpret_cast(&eval), it->second, - prio++, fl.id_, fl.name_.c_str())) { + reinterpret_cast(&eval), + destination_action, rss_destination, 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_, + fl.name_, primary_queue, port, mt.ecpri_match_.msg_type_, mt.ecpri_match_.match_msg_type_, mt.ecpri_match_.id_, mt.ecpri_match_.match_id_); installed++; continue; @@ -1858,6 +1996,9 @@ Status IbverbsEngine::install_port_flows() { DEVX_SET(fte_match_set_lyr_2_4, mk, ip_protocol, 0xff); DEVX_SET(fte_match_set_lyr_2_4, vl, ip_protocol, MLX5_IP_PROTOCOL_UDP); }; + if (rss_destination != nullptr) { + pin_ipv4_udp(); + } if (mt.udp_src_ > 0) { pin_ipv4_udp(); DEVX_SET(fte_match_set_lyr_2_4, mk, udp_sport, 0xffff); @@ -1916,12 +2057,13 @@ Status IbverbsEngine::install_port_flows() { continue; } if (!add_rule(crit, reinterpret_cast(&mask), - reinterpret_cast(&val), it->second, - prio++, fl.id_, fl.name_.c_str(), flow_reformats)) { + reinterpret_cast(&val), + destination_action, rss_destination, prio++, fl.id_, fl.name_.c_str(), + flow_reformats)) { return Status::GENERIC_FAILURE; } DAQIRI_LOG_INFO("Flow '{}' -> queue {} (port {}): udp_src={} udp_dst={} ipv4_len={}", - fl.name_, fl.action_.id_, port, mt.udp_src_, mt.udp_dst_, mt.ipv4_len_); + fl.name_, primary_queue, port, mt.udp_src_, mt.udp_dst_, mt.ipv4_len_); installed++; } @@ -1936,8 +2078,8 @@ Status IbverbsEngine::install_port_flows() { val.match_sz = sizeof(val.buf); IbvRxQueue* dest = by_id.begin()->second; if (!add_rule(0, reinterpret_cast(&mask), - reinterpret_cast(&val), dest, - kIbverbsCatchAllPriority, 0, "catch-all")) { + reinterpret_cast(&val), dest->dr_action, + nullptr, kIbverbsCatchAllPriority, 0, "catch-all")) { return Status::GENERIC_FAILURE; } DAQIRI_LOG_INFO("DevX catch-all flow -> queue {} (port {})", dest->queue_id, port); @@ -2043,7 +2185,7 @@ Status IbverbsEngine::install_tx_flows() { struct mlx5dv_dr_action* actions[8]; int na = 0; - for (const auto& action : fl.actions_) { + for (const auto& action : flow_config_actions(fl)) { st.reformat_buffers.emplace_back(); auto& buffer = st.reformat_buffers.back(); enum mlx5dv_flow_action_packet_reformat_type reformat_type {}; @@ -2267,7 +2409,10 @@ Status IbverbsEngine::setup_rx_queue(IbvRxQueue& q, const InterfaceConfig& intf, // Per-queue flow id: if a configured flow steers to this queue, report its id // for packets received here (see get_packet_flow_id). for (const auto& fl : intf.rx_.flows_) { - if (fl.action_.type_ == FlowType::QUEUE && fl.action_.id_ == q.queue_id) { + const FlowAction queue_action = flow_queue_action(flow_config_actions(fl)); + const auto queue_ids = flow_queue_ids(queue_action); + if (queue_action.type_ == FlowType::QUEUE && + std::find(queue_ids.begin(), queue_ids.end(), q.queue_id) != queue_ids.end()) { q.flow_id = fl.id_; break; } @@ -2760,7 +2905,7 @@ Status IbverbsEngine::init_reorder(IbvRxQueue& q, const InterfaceConfig& intf, // flow id -> queue, to find which reorder configs belong to this queue. std::unordered_map flow_to_queue; for (const auto& fl : intf.rx_.flows_) { - flow_to_queue[fl.id_] = fl.action_.id_; + flow_to_queue[fl.id_] = flow_queue_ids(flow_queue_action(flow_config_actions(fl))).front(); } auto st = std::make_unique(); @@ -3667,6 +3812,11 @@ void IbverbsEngine::shutdown() { mlx5dv_dr_rule_destroy(r); } } + st.rules.clear(); + // Rules no longer reference their destination actions, so release shared + // RSS TIR/RQT resources before the queues and their direct TIRs disappear. + st.rule_specs.clear(); + st.rss_destinations.clear(); for (auto* a : st.tag_actions) { if (a) { mlx5dv_dr_action_destroy(a); diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.h b/src/engines/ibverbs/daqiri_ibverbs_engine.h index 8508f59..e8eb5d6 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.h +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.h @@ -485,6 +485,15 @@ class IbverbsEngine : public Engine { // (post-encap) if the MTU is too low. void ensure_port_mtus(); + struct RssDestination { + std::vector queue_ids; + bool inner = false; + struct mlx5dv_devx_obj* rqt_obj = nullptr; + struct mlx5dv_devx_obj* tir_obj = nullptr; + struct mlx5dv_dr_action* action = nullptr; + }; + using RssDestinationPtr = std::shared_ptr; + // ---- dynamic RX flow lifecycle ---- enum class DynamicFlowState { ACTIVE, DELETING }; struct DynamicFlowEntry { @@ -494,6 +503,7 @@ class IbverbsEngine : public Engine { struct mlx5dv_dr_matcher* matcher = nullptr; struct mlx5dv_dr_rule* rule = nullptr; struct mlx5dv_dr_action* tag_action = nullptr; + RssDestinationPtr rss_destination; std::vector reformat_actions; std::vector> reformat_buffers; DynamicFlowState state = DynamicFlowState::ACTIVE; @@ -553,10 +563,12 @@ class IbverbsEngine : public Engine { struct mlx5dv_dr_action* action; // dest-TIR struct mlx5dv_dr_action* tag = nullptr; // optional MARK tag action std::vector reformats; + RssDestinationPtr rss_destination; size_t value_sz = 0; // bytes of `value` in use uint64_t value[64]; // up to full fte_match_param (512 B) }; std::vector rule_specs; + std::unordered_map> rss_destinations; // Flex-parser (parse-graph) nodes for arbitrary-offset / ipv4_len matching, // keyed by the config flex-item id. The DevX object must outlive any matcher // that references its sample_field_id, so these are torn down after rules. @@ -587,18 +599,17 @@ class IbverbsEngine : public Engine { }; std::map port_steering_; // port_id -> steering - bool create_dr_rule_locked(int port, - PortSteering& st, - uint16_t criteria, + bool create_dr_rule_locked(int port, PortSteering& st, uint16_t criteria, struct mlx5dv_flow_match_parameters* mask, struct mlx5dv_flow_match_parameters* value, - IbvRxQueue* dest, - int priority, - FlowId flow_id, - const char* desc, - DynamicFlowEntry* dynamic_entry, + struct mlx5dv_dr_action* destination_action, uint16_t primary_queue, + const RssDestinationPtr& rss_destination, int priority, FlowId flow_id, + const char* desc, DynamicFlowEntry* dynamic_entry, const std::vector& reformats = std::vector{}); + Status resolve_rx_destination(int port, PortSteering& st, const FlowAction& queue_action, + bool inner, struct mlx5dv_dr_action** action, + uint16_t* primary_queue, RssDestinationPtr* rss_destination); Status install_flow_rule_locked(int port, PortSteering& st, const InterfaceConfig& intf, diff --git a/src/engines/ibverbs/mlx5_prm_min.h b/src/engines/ibverbs/mlx5_prm_min.h index 883d4f2..3371639 100644 --- a/src/engines/ibverbs/mlx5_prm_min.h +++ b/src/engines/ibverbs/mlx5_prm_min.h @@ -38,6 +38,7 @@ enum { MLX5_CMD_OP_CREATE_TIR = 0x900, MLX5_CMD_OP_CREATE_RQ = 0x908, MLX5_CMD_OP_MODIFY_RQ = 0x909, + MLX5_CMD_OP_CREATE_RQT = 0x916, MLX5_CMD_OP_CREATE_GENERAL_OBJECT = 0xa00, MLX5_CMD_OP_QUERY_GENERAL_OBJECT = 0xa02, }; @@ -105,9 +106,19 @@ enum { }; enum { MLX5_TIRC_DISP_TYPE_DIRECT = 0x0, + MLX5_TIRC_DISP_TYPE_INDIRECT = 0x1, }; enum { MLX5_RX_HASH_FN_NONE = 0x0, + MLX5_RX_HASH_FN_TOEPLITZ = 0x2, +}; +enum { + MLX5_L3_PROT_TYPE_IPV4 = 0x0, + MLX5_L4_PROT_TYPE_UDP = 0x1, + MLX5_HASH_FIELD_SEL_SRC_IP = 1 << 0, + MLX5_HASH_FIELD_SEL_DST_IP = 1 << 1, + MLX5_HASH_FIELD_SEL_L4_SPORT = 1 << 2, + MLX5_HASH_FIELD_SEL_L4_DPORT = 1 << 3, }; enum { MLX5_MODIFY_RQ_RST2RDY = 0x1, @@ -233,6 +244,39 @@ struct mlx5_ifc_modify_rq_out_bits { uint8_t reserved_at_40[0x40]; }; +struct mlx5_ifc_rqtc_bits { + uint8_t reserved_at_0[0xa0]; + uint8_t reserved_at_a0[0x5]; + uint8_t list_q_type[0x3]; + uint8_t reserved_at_a8[0x8]; + uint8_t rqt_max_size[0x10]; + uint8_t rq_vhca_id_format[0x1]; + uint8_t reserved_at_c1[0xf]; + uint8_t rqt_actual_size[0x10]; + uint8_t reserved_at_e0[0x6a0]; + // Flexible array of 32-bit entries (8 reserved bits + 24-bit RQN). Command + // buffers allocate additional entries after this sentinel. + uint8_t rq_num[1][0x20]; +}; + +struct mlx5_ifc_create_rqt_in_bits { + uint8_t opcode[0x10]; + uint8_t uid[0x10]; + uint8_t reserved_at_20[0x10]; + uint8_t op_mod[0x10]; + uint8_t reserved_at_40[0xc0]; + struct mlx5_ifc_rqtc_bits rqt_context; +}; + +struct mlx5_ifc_create_rqt_out_bits { + uint8_t status[0x8]; + uint8_t reserved_at_8[0x18]; + uint8_t syndrome[0x20]; + uint8_t reserved_at_40[0x8]; + uint8_t rqtn[0x18]; + uint8_t reserved_at_60[0x20]; +}; + struct mlx5_ifc_rx_hash_field_select_bits { uint8_t l3_prot_type[0x1]; uint8_t l4_prot_type[0x1]; diff --git a/src/rss.h b/src/rss.h new file mode 100644 index 0000000..05d8284 --- /dev/null +++ b/src/rss.h @@ -0,0 +1,29 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include +#include + +namespace daqiri { + +// The conventional 40-byte Microsoft RSS key. Keeping the key fixed makes a +// given five-tuple map consistently across restarts and across both raw engines. +inline constexpr std::array kToeplitzRssKey = { + 0x6d, 0x5a, 0x56, 0xda, 0x25, 0x5b, 0x0e, 0xc2, 0x41, 0x67, 0x25, 0x3d, 0x43, 0xa3, + 0x8f, 0xb0, 0xd0, 0xca, 0x2b, 0xcb, 0xae, 0x7b, 0x30, 0xb4, 0x77, 0xcb, 0x2d, 0xa3, + 0x80, 0x30, 0xf2, 0x0c, 0x6a, 0x42, 0xb7, 0x3b, 0xbe, 0xac, 0x01, 0xfa}; + +inline size_t rss_table_size(size_t queue_count) { + size_t size = 1; + while (size < queue_count) { + size <<= 1; + } + return size; +} + +} // namespace daqiri From 593390b492c747d3bd59c3fb945618408a8767e2 Mon Sep 17 00:00:00 2001 From: Cliff Burdick Date: Fri, 17 Jul 2026 21:45:42 +0000 Subject: [PATCH 2/2] #233 - Document multi-queue RSS in README Signed-off-by: Cliff Burdick --- README.md | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index a1a21ff..5684fbf 100644 --- a/README.md +++ b/README.md @@ -38,10 +38,14 @@ DAQIRI provides direct NIC hardware access in userspace, bypassing the Linux ker S3-compatible object store through the AWS SDK for C++. - **Flow Steering** — Configure the NIC's hardware flow engine to route packets by UDP source/destination port or flex-item payload fields. Raw RX flows can be configured - statically in YAML or added/deleted dynamically after `daqiri_init()`. Per RX - interface, use standard UDP/IP flows or flex-item flows, not both. Raw DPDK and - raw ibverbs flows can also use hardware-only VLAN push/pop and VXLAN, GRE, or - NVGRE encap/decap actions; socket/RDMA streams reject those tunnel actions. + statically in YAML or added/deleted dynamically after `daqiri_init()`. A scalar + queue target steers directly to one queue; `ids: [0, 1, ...]` automatically enables + flow-affine IPv4/UDP Toeplitz RSS across the listed queues on the raw DPDK and + ibverbs engines. One unchanged five-tuple remains on one queue, so balanced packet + counts require varied tuples. Per RX interface, use standard UDP/IP flows or + flex-item flows, not both. Raw DPDK and raw ibverbs flows can also use hardware-only + VLAN push/pop and VXLAN, GRE, or NVGRE encap/decap actions; socket/RDMA streams reject + those tunnel actions. - **RDMA** — RDMA verbs (READ, WRITE, SEND) over RoCE on Ethernet NICs or InfiniBand. - **Linux socket control** — TCP/UDP socket streams expose connection IDs and `socket_setsockopt()` for native Linux `setsockopt` tuning without YAML option