diff --git a/docs/api-reference/cpp.md b/docs/api-reference/cpp.md index d09c610b..6566e589 100644 --- a/docs/api-reference/cpp.md +++ b/docs/api-reference/cpp.md @@ -883,9 +883,32 @@ workflow sections above show the common call order and ownership rules. ### Ports, Traffic, Socket, and RDMA +Raw userspace TX must supply a destination MAC because transmitting a completed +Ethernet frame bypasses the kernel route and ARP processing used by sockets. +The raw ibverbs engine can perform that lookup explicitly: + +```cpp +char dst_mac[6] = {}; +daqiri::Status status = daqiri::resolve_ipv4_mac(port, "192.0.2.20", dst_mac, 3000); +if (status != daqiri::Status::SUCCESS) { + // Stop before sending data packets. +} +``` + +The result is a snapshot, not a managed DAQIRI neighbor entry. Reuse it for +packet templates to keep the data path efficient, and call +`resolve_ipv4_mac()` again if the peer, gateway, route, link, or network +namespace may have changed. DAQIRI does not refresh previously returned MAC +addresses in the background. If the port also has DAQIRI RX queues, configure +`rx.flow_isolation: true` so unmatched ARP traffic stays on the kernel path; a +TX-only interface with no `rx.queues` needs no `rx` section. See +[Destination MAC resolution](../concepts.md#destination-mac-resolution) for the +full ownership model. + | Function | Purpose | | --- | --- | | `get_mac_addr(port, mac)` | Copy a port MAC address into a six-byte buffer. | +| `resolve_ipv4_mac(port, dst, mac, timeout_ms)` | Resolve an IPv4 destination's next-hop MAC through the raw ibverbs port's Linux route and ARP table. `dst` may be a host-order `uint32_t` or a dotted-decimal string. | | `format_eth_addr(dst, addr)` | Convert a `xx:xx:xx:xx:xx:xx` MAC string into a six-byte buffer. Invalid input zeroes the buffer. | | `get_port_id(key)` | Resolve an interface name or PCIe address to a port ID. | | `get_num_rx_queues(port_id)` | Return the configured or engine-reported RX queue count. | diff --git a/docs/api-reference/python.md b/docs/api-reference/python.md index 98db538d..129ec2fe 100644 --- a/docs/api-reference/python.md +++ b/docs/api-reference/python.md @@ -690,9 +690,21 @@ The workflow sections above show the common call order and ownership rules. ### Ports, Traffic, Socket, and RDMA +Raw userspace TX bypasses the automatic route and ARP processing performed for +Linux sockets. With the raw ibverbs engine, call +`resolve_ipv4_mac(port, dst, timeout_ms)` after initialization, then reuse the +returned MAC in packet templates. This is a one-shot lookup: DAQIRI does not +cache or refresh the returned value, so the application must re-query after a +peer, gateway, route, link, or network-namespace change, or according to its +own failure/refresh policy. Configure `rx.flow_isolation: true` so unmatched +ARP traffic remains on the kernel path when the port also has DAQIRI RX queues. +A TX-only interface with no `rx.queues` needs no `rx` section. See +[Destination MAC resolution](../concepts.md#destination-mac-resolution). + | Function | Purpose | | --- | --- | | `get_mac_addr(port)` | Return `(Status, "aa:bb:cc:dd:ee:ff")`. | +| `resolve_ipv4_mac(port, dst, timeout_ms)` | Return `(Status, mac_string)` after resolving a host-order integer or dotted-decimal IPv4 destination through the raw ibverbs port's Linux route and ARP table. The binding releases the GIL while waiting. | | `format_eth_addr(addr)` | Return six MAC-address bytes from a `xx:xx:xx:xx:xx:xx` MAC string. Invalid input returns zero bytes. | | `get_port_id(key)` | Resolve an interface name or PCIe address to a port ID. | | `get_num_rx_queues(port_id)` | Return configured RX queue count for a port. | diff --git a/docs/benchmarks/raw_benchmarking.md b/docs/benchmarks/raw_benchmarking.md index 9ad8d230..2ba87656 100644 --- a/docs/benchmarks/raw_benchmarking.md +++ b/docs/benchmarks/raw_benchmarking.md @@ -5,7 +5,7 @@ hide: # Raw Ethernet Benchmarking -DAQIRI provides raw Ethernet benchmark applications that use DPDK to drive an NVIDIA NIC directly. This page walks through `daqiri_bench_raw_gpudirect`, the TX/RX loopback config, and the raw Ethernet checks needed before interpreting throughput results. +DAQIRI provides raw Ethernet benchmark applications that use the DPDK or ibverbs engine to drive an NVIDIA NIC directly. This page walks through `daqiri_bench_raw_gpudirect`, the TX/RX loopback config, and the raw Ethernet checks needed before interpreting throughput results. Make sure to [build the DAQIRI library](../getting-started.md#build-the-daqiri-library) beforehand. @@ -81,14 +81,14 @@ docker run --rm -it --privileged \ If you have two DGX Sparks cross-cabled p0↔p0 instead of a chassis QSFP loop on one machine, use the `_xhost` configs. Each host runs only its own role, so the YAML on each side configures one port instead of two. Both hosts must already be set up per the [DGX Spark profile](../tutorials/system_configuration.md#dgx-spark-profile), with one adjustment: the `daqiri-tx` (`1.1.1.1/24`) and `daqiri-rx` (`2.2.2.2/24`) nmcli profiles are *split across* the two hosts. Bring up `daqiri-tx` on the TX host's p0 and `daqiri-rx` on the RX host's p0, instead of both on one box. -**Network prerequisite (required for RDMA, recommended for raw).** Assigning `/24` addresses on each host is not enough for the kernel to reach the peer over a direct cable. RDMA-CM uses the kernel stack, so you need a host route and a static neighbor on the cabled port before ping or RoCE will work. Run [`scripts/setup_spark_xhost_net.sh`](https://github.com/nvidia/daqiri/blob/main/scripts/setup_spark_xhost_net.sh) on **both** hosts after bringing up the nmcli profile. See the [cross-host variant](../tutorials/system_configuration.md#cross-host-variant-two-sparks) in System Configuration for the full steps. +**Network prerequisite.** Assigning `/24` addresses on each host is not enough for the kernel to reach the peer over a direct cable. Install a host route on the cabled port by running [`scripts/setup_spark_xhost_net.sh`](https://github.com/nvidia/daqiri/blob/main/scripts/setup_spark_xhost_net.sh) on **both** hosts after bringing up the nmcli profile. The ibverbs raw engine uses that route and Linux ARP when `eth_dst_addr` is omitted; RDMA-CM uses the same kernel route. See the [cross-host variant](../tutorials/system_configuration.md#cross-host-variant-two-sparks) in System Configuration for the full steps. ```bash -# TX host (peer MAC from RX: cat /sys/class/net/enp1s0f0np0/address) -sudo scripts/setup_spark_xhost_net.sh --role tx --peer-mac +# TX host +sudo scripts/setup_spark_xhost_net.sh --role tx -# RX host (peer MAC from TX) -sudo scripts/setup_spark_xhost_net.sh --role rx --peer-mac +# RX host +sudo scripts/setup_spark_xhost_net.sh --role rx # Verify on each host before starting benches ping -c 3 # 2.2.2.2 on TX, 1.1.1.1 on RX @@ -101,7 +101,7 @@ ip route get # must name enp1s0f0np0, not lo # RX host sudo ./daqiri_bench_raw_gpudirect daqiri_bench_raw_rx_spark_xhost.yaml --seconds 30 -# TX host (set eth_dst_addr to the RX host p0's MAC first: cat /sys/class/net/enp1s0f0np0/address on the RX host) +# TX host (the ibverbs config resolves the RX MAC through Linux ARP) sudo ./daqiri_bench_raw_gpudirect daqiri_bench_raw_tx_spark_xhost.yaml --seconds 30 ``` @@ -255,7 +255,7 @@ interfaces: To run the benchmarking application to run a loopback on your system, you'll need to modify the `bench_tx` section which configures the application itself, to create the packet headers, pin the application TX worker, and direct the packets to the NIC. Make sure to remove the template brackets `< >`. -- `eth_dst_addr` with the MAC address (and not the PCIe address) of the NIC interface you want to use for Rx. You can get the MAC address of your `if_name` interface with `#!bash cat /sys/class/net/$if_name/address`: +- `eth_dst_addr` with the MAC address (and not the PCIe address) of the NIC interface you want to use for Rx. You can get the MAC address of your `if_name` interface with `#!bash cat /sys/class/net/$if_name/address`. Keep this field for DPDK and loopback configurations. A cross-host raw ibverbs configuration may omit it when the TX netdev has a usable IPv4 address and route; the benchmark then resolves `ip_dst_addr` through Linux routing and ARP once before its TX loop: - `cpu_core` with the CPU core for the benchmark application's TX thread. This is separate from the DAQIRI TX queue `cpu_core`; use a different isolated core when you have one, or deliberately share when the machine has a tight core budget. ```yaml hl_lines="3 5" @@ -263,7 +263,7 @@ bench_tx: - interface_name: "tx_port" # Name of the TX port from the daqiri config cpu_core: 10 # Benchmark application TX thread affinity ... - eth_dst_addr: <00:00:00:00:00:00> # Destination MAC address - required when Rx flow_isolation=true + eth_dst_addr: <00:00:00:00:00:00> # Explicit MAC for DPDK or loopback ... ``` @@ -274,13 +274,14 @@ bench_tx: - interface_name: "tx_port" # Name of the TX port from the daqiri config cpu_core: 10 # Benchmark application TX thread affinity ... - eth_dst_addr: 48:b0:2d:ee:83:ad # Destination MAC address - required when Rx flow_isolation=true + eth_dst_addr: 48:b0:2d:ee:83:ad # Explicit MAC for DPDK or loopback ... ``` ??? info "Show explanation" - - `eth_dst_addr` - the destination ethernet MAC address - will be embedded in the packet headers by the application. This is required here because the Rx interface above has `flow_isolation: true` (explained in more details below). In that configuration, only the packets listing the adequate destination MAC address will be accepted by the Rx interface. + - `eth_dst_addr` - when present, this destination Ethernet MAC is embedded in packet headers without a route or neighbor lookup. When it is absent on raw ibverbs, the benchmark resolves the next-hop MAC from `ip_dst_addr` once before sending. Raw userspace TX does not otherwise invoke Linux routing or ARP because the application supplies complete Ethernet frames. If the interface also has DAQIRI RX queues, use `rx.flow_isolation: true` so ARP remains on the kernel path. A TX-only interface with no `rx.queues` needs no `rx` section. + - The benchmark reuses the resolved MAC for the entire run; it does not monitor Linux neighbor changes. Long-running applications should define their own refresh policy and call `resolve_ipv4_mac()` again when a peer, gateway, route, link, or namespace may have changed. See [Destination MAC resolution](../concepts.md#destination-mac-resolution). - `cpu_core` - the benchmark application's own TX worker thread affinity. Set the matching `bench_rx.cpu_core` for RX workers too. These app-thread fields are distinct from the DAQIRI queue `cpu_core` values that poll the NIC. - We ignore the IP fields (`ip_src_addr`, `ip_dst_addr`) for now, as we are testing on a layer 2 network by just connecting a cable between the two interfaces on our system, therefore having mock values has no impact. - You might have noted the lack of a `eth_src_addr` field in this `bench_tx` section. This is because the source Ethernet MAC address can be inferred automatically by the DAQIRI library from the PCIe address of the Tx interface referenced above. diff --git a/docs/concepts.md b/docs/concepts.md index 14e81646..2e19b0e2 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -69,6 +69,43 @@ packets into one pre-posted buffer to avoid per-packet allocation. Set Raw Ethernet requires an NVIDIA ConnectX-6 Dx or later NIC. Packet pacing and timed transmission require ConnectX-7 or later. +#### Destination MAC resolution + +A raw-Ethernet application constructs the complete layer-2 frame in user +space. Unlike a UDP or TCP socket send, that packet does not pass through the +Linux IP stack, so Linux does not automatically select a route, identify the +next hop, run ARP, or fill the destination MAC address. Merely configuring an +IP destination in the packet header is therefore insufficient: userspace +packet processing cannot rely on the kernel routing and neighbor tables being +consulted when the packet is transmitted. + +The raw ibverbs engine provides `resolve_ipv4_mac()` as an explicit bridge to +Linux routing and ARP. After `daqiri_init()`, the application supplies a DAQIRI +port and IPv4 destination. DAQIRI asks Linux for the route constrained to that +port's kernel netdev and primary IPv4 source, resolves the gateway when the +route is off-link or the destination when it is on-link, and returns the +current six-byte next-hop MAC. An existing confirmed Linux neighbor entry +returns immediately. DAQIRI refreshes a stale entry before returning it; +otherwise Linux performs ARP while DAQIRI waits up to the caller's deadline. +This API is available only for the raw ibverbs engine. DPDK and the socket/RDMA +engines return `NOT_SUPPORTED`. + +Resolution is deliberately a one-shot operation. DAQIRI does not maintain a +second neighbor cache, start a refresh thread, or change a MAC already placed +in application packet templates. For efficiency, resolve once and reuse the +result while the route and peer are stable. The application is responsible for +calling `resolve_ipv4_mac()` again when its refresh policy requires it—for +example after a link, peer NIC, gateway, route, or network-namespace change, or +after a transmission/reachability failure. Re-querying still uses Linux's +neighbor table and is inexpensive while its entry remains usable. + +Automatic ARP also requires the NIC's control traffic to remain visible to the +kernel. If the raw ibverbs interface has DAQIRI RX queues, configure +`rx.flow_isolation: true` so DAQIRI installs only explicit application flow +rules and unmatched ARP traffic continues on the kernel path. A TX-only +interface with no `rx.queues` installs no DAQIRI RX catch-all and does not need +an `rx` section. + ### Socket *YAML:* `stream_type: "socket"`. The specific transport is chosen by endpoint diff --git a/docs/tutorials/configuration-walkthrough.md b/docs/tutorials/configuration-walkthrough.md index 90d1c959..8d730d62 100644 --- a/docs/tutorials/configuration-walkthrough.md +++ b/docs/tutorials/configuration-walkthrough.md @@ -284,7 +284,7 @@ bench_tx: # (25)! 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. +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. An explicit `eth_dst_addr` is used unchanged. With the raw ibverbs engine, it may instead be omitted so DAQIRI resolves `ip_dst_addr` (or its gateway) through the selected port's Linux route and ARP table. When that interface also has DAQIRI RX queues, configure `rx.flow_isolation: true` so ARP remains on the kernel path; a TX-only interface with no `rx.queues` needs no `rx` section. Other engines still require an explicit destination MAC. 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. ### Flow steering diff --git a/docs/tutorials/system_configuration.md b/docs/tutorials/system_configuration.md index 7877a70e..c351e967 100644 --- a/docs/tutorials/system_configuration.md +++ b/docs/tutorials/system_configuration.md @@ -1547,14 +1547,14 @@ DAQIRI requires an [**NVIDIA SmartNIC**](https://www.nvidia.com/en-us/networking - TX host: `sudo nmcli connection up daqiri-tx` only (`1.1.1.1/24` on `enp1s0f0np0`) - RX host: `sudo nmcli connection up daqiri-rx` only (`2.2.2.2/24` on `enp1s0f0np0`) - The `/24` assignment alone does not give the kernel a route to the peer. On **each** host, add a host route and static neighbor for the peer on the cabled port (read the peer's MAC on the other box with `cat /sys/class/net/enp1s0f0np0/address`): + The `/24` assignment alone does not give the kernel a route to the peer. On **each** host, add a host route on the cabled port. The ibverbs raw engine can then ask Linux to resolve the peer with ARP when `bench_tx.eth_dst_addr` is omitted: ```bash # TX host - sudo scripts/setup_spark_xhost_net.sh --role tx --peer-mac + sudo scripts/setup_spark_xhost_net.sh --role tx # RX host - sudo scripts/setup_spark_xhost_net.sh --role rx --peer-mac + sudo scripts/setup_spark_xhost_net.sh --role rx ``` Verify before running `_xhost` benchmarks (raw or RDMA): diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 92fa9a14..7f22444c 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -69,7 +69,7 @@ function(link_daqiri_bench target) target_compile_features(${target} PRIVATE cxx_std_17) if(NOT BUILD_SHARED_LIBS AND UNIX AND NOT APPLE) set(DAQIRI_EXAMPLE_STATIC_LIBS daqiri::daqiri) - foreach(engine IN ITEMS dpdk socket rdma) + foreach(engine IN ITEMS dpdk socket rdma ibverbs) if(TARGET daqiri_${engine}) list(APPEND DAQIRI_EXAMPLE_STATIC_LIBS daqiri_${engine}) endif() diff --git a/examples/daqiri_bench_raw_rx_spark_xhost.yaml b/examples/daqiri_bench_raw_rx_spark_xhost.yaml index 3bbd610e..d86f213b 100644 --- a/examples/daqiri_bench_raw_rx_spark_xhost.yaml +++ b/examples/daqiri_bench_raw_rx_spark_xhost.yaml @@ -18,6 +18,7 @@ daqiri: cfg: version: 1 stream_type: "raw" + engine: "ibverbs" master_core: 8 debug: false log_level: "info" @@ -44,7 +45,7 @@ daqiri: - "Data_RX_GPU" flows: - name: "flow_0" - id: 0 + id: 1 action: type: queue id: 0 diff --git a/examples/daqiri_bench_raw_tx_spark_xhost.yaml b/examples/daqiri_bench_raw_tx_spark_xhost.yaml index 31242ae0..94b43e4b 100644 --- a/examples/daqiri_bench_raw_tx_spark_xhost.yaml +++ b/examples/daqiri_bench_raw_tx_spark_xhost.yaml @@ -10,11 +10,10 @@ # - master_core: 8; cpu_core: 17 for the DAQIRI TX queue and 16 for the app # TX worker -- separate cores so they don't contend (isolated big-cluster # X925 16-19) -# - eth_dst_addr is the *peer* RX port's MAC -- replace with your own: -# cat /sys/class/net//address # on the RX host -# - ip_src/ip_dst: arbitrary 1.1.1.1 -> 2.2.2.2 (kernel stack bypassed by -# the DPDK PMD; only the UDP src/dst ports below are matched by the RX -# flow rule in daqiri_bench_raw_rx_spark_xhost.yaml) +# - engine: ibverbs leaves ARP on the kernel path. This TX-only interface has +# no DAQIRI RX queues, so no RX catch-all can divert ARP from Linux. +# eth_dst_addr is omitted, so DAQIRI resolves 2.2.2.2 through the kernel +# route and neighbour tables. # %YAML 1.2 --- @@ -22,6 +21,7 @@ daqiri: cfg: version: 1 stream_type: "raw" + engine: "ibverbs" master_core: 8 debug: false log_level: "info" @@ -41,7 +41,7 @@ daqiri: queues: - name: "tx_q_0" id: 0 - batch_size: 10240 + batch_size: 4096 cpu_core: 17 memory_regions: - "Data_TX_GPU" @@ -51,10 +51,9 @@ daqiri: bench_tx: interface_name: "tx_port" cpu_core: 16 - batch_size: 10240 + batch_size: 4096 payload_size: 8000 header_size: 64 - eth_dst_addr: <00:00:00:00:00:00> ip_src_addr: 1.1.1.1 ip_dst_addr: 2.2.2.2 udp_src_port: 4096 diff --git a/examples/dynamic_rx_flow_example.cpp b/examples/dynamic_rx_flow_example.cpp index c6d081d7..09f9c6b4 100644 --- a/examples/dynamic_rx_flow_example.cpp +++ b/examples/dynamic_rx_flow_example.cpp @@ -144,12 +144,12 @@ void rx_worker(const daqiri::bench::RawBenchRxConfig& cfg, RxCounter& counter, } } -void tx_worker(const daqiri::bench::RawBenchTxConfig& cfg, - daqiri::bench::TokenBucketPacer& pacer, - std::atomic& stop) { +void tx_worker(const daqiri::bench::RawBenchTxConfig& cfg, daqiri::bench::TokenBucketPacer& pacer, + std::atomic& stop, std::atomic& failed) { const int port_id = daqiri::get_port_id(cfg.interface_name); if (port_id < 0) { std::cerr << "Invalid TX interface_name: " << cfg.interface_name << "\n"; + failed.store(true); stop.store(true); return; } @@ -157,7 +157,11 @@ void tx_worker(const daqiri::bench::RawBenchTxConfig& cfg, char eth_dst[6] = {0}; char eth_src[6] = {0}; daqiri::format_eth_addr(eth_src, cfg.eth_src_addr); - daqiri::format_eth_addr(eth_dst, cfg.eth_dst_addr); + if (!daqiri::bench::resolve_tx_eth_dst(port_id, cfg.ip_dst_addr, cfg.eth_dst_addr, eth_dst)) { + failed.store(true); + stop.store(true); + return; + } const uint32_t ip_src = parse_ipv4_host_order(cfg.ip_src_addr); const uint32_t ip_dst = parse_ipv4_host_order(cfg.ip_dst_addr); @@ -358,6 +362,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 failed{false}; std::atomic expected_flow_id{0}; std::thread tx_thread; std::vector rx_threads; @@ -395,12 +400,16 @@ int main(int argc, char** argv) { } daqiri::bench::TokenBucketPacer pacer(args.target_gbps); - tx_thread = std::thread(tx_worker, std::cref(tx_configs[0]), std::ref(pacer), std::ref(stop)); + tx_thread = std::thread(tx_worker, std::cref(tx_configs[0]), std::ref(pacer), std::ref(stop), + std::ref(failed)); 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)); + if (failed.load()) { + throw std::runtime_error("TX worker failed before sending packets"); + } const uint64_t dropped_window_packets = packets(counters[0]) + packets(counters[1]); if (dropped_window_packets != 0) { throw std::runtime_error("Packets arrived during the initial flow-isolated drop window"); diff --git a/examples/gds_write_example.cpp b/examples/gds_write_example.cpp index 2b1bcc38..12404334 100644 --- a/examples/gds_write_example.cpp +++ b/examples/gds_write_example.cpp @@ -114,7 +114,9 @@ bool fill_and_send_one_burst(const daqiri::bench::RawBenchTxConfig &cfg, char eth_dst[6] = {}; char eth_src[6] = {}; daqiri::format_eth_addr(eth_src, cfg.eth_src_addr); - daqiri::format_eth_addr(eth_dst, cfg.eth_dst_addr); + if (!daqiri::bench::resolve_tx_eth_dst(port_id, cfg.ip_dst_addr, cfg.eth_dst_addr, eth_dst)) { + return false; + } uint32_t ip_src = 0; uint32_t ip_dst = 0; diff --git a/examples/pcap_writer_example.cpp b/examples/pcap_writer_example.cpp index 2c02ee9f..bf11582c 100644 --- a/examples/pcap_writer_example.cpp +++ b/examples/pcap_writer_example.cpp @@ -112,9 +112,9 @@ bool parse_ipv4(const std::string &text, std::array *out) { return true; } -std::vector -make_udp_packet_template(const daqiri::bench::RawBenchTxConfig &cfg, - const std::string ð_src_addr) { +std::vector make_udp_packet_template(const daqiri::bench::RawBenchTxConfig& cfg, + const std::string& eth_src_addr, + const char* dst_mac) { if (cfg.header_size < kUdpIpv4EthernetHeaderLen) { throw std::runtime_error("bench_tx.header_size must be at least 42 bytes"); } @@ -138,13 +138,11 @@ make_udp_packet_template(const daqiri::bench::RawBenchTxConfig &cfg, const auto src_ports = daqiri::bench::parse_udp_ports(cfg.udp_src_port); const auto dst_ports = daqiri::bench::parse_udp_ports(cfg.udp_dst_port); - char dst_mac[6] = {0}; char src_mac[6] = {0}; - daqiri::format_eth_addr(dst_mac, cfg.eth_dst_addr); daqiri::format_eth_addr(src_mac, eth_src_addr); std::vector packet(packet_len, 0); - std::memcpy(packet.data(), dst_mac, sizeof(dst_mac)); + std::memcpy(packet.data(), dst_mac, 6); std::memcpy(packet.data() + 6, src_mac, sizeof(src_mac)); store_be16(packet.data() + 12, kEtherTypeIpv4); @@ -422,9 +420,10 @@ PcapTxConfig parse_pcap_tx_config(const YAML::Node &root) { return cfg; } -void tx_worker(PcapTxConfig cfg, std::atomic *stop) { +void tx_worker(PcapTxConfig cfg, std::atomic* stop, std::atomic* failed) { if (!daqiri::bench::set_current_thread_affinity(cfg.raw.cpu_core, "bench_tx")) { + failed->store(true, std::memory_order_relaxed); stop->store(true, std::memory_order_relaxed); return; } @@ -433,11 +432,19 @@ void tx_worker(PcapTxConfig cfg, std::atomic *stop) { if (port_id < 0) { std::cerr << "Invalid TX interface_name: " << cfg.raw.interface_name << "\n"; + failed->store(true, std::memory_order_relaxed); stop->store(true, std::memory_order_relaxed); return; } - auto packet_template = make_udp_packet_template(cfg.raw, cfg.eth_src_addr); + char eth_dst[6] = {}; + if (!daqiri::bench::resolve_tx_eth_dst(port_id, cfg.raw.ip_dst_addr, cfg.raw.eth_dst_addr, + eth_dst)) { + failed->store(true, std::memory_order_relaxed); + stop->store(true, std::memory_order_relaxed); + return; + } + auto packet_template = make_udp_packet_template(cfg.raw, cfg.eth_src_addr, eth_dst); std::unordered_set initialized_buffers; uint64_t bursts = 0; uint64_t packets = 0; @@ -460,7 +467,7 @@ void tx_worker(PcapTxConfig cfg, std::atomic *stop) { } const int num_packets = daqiri::get_num_packets(msg); - bool failed = false; + bool burst_failed = false; for (int i = 0; i < num_packets; ++i) { auto *dst = daqiri::get_segment_packet_ptr(msg, 0, i); @@ -471,7 +478,8 @@ void tx_worker(PcapTxConfig cfg, std::atomic *stop) { if (err != cudaSuccess) { std::cerr << "TX cudaMemcpy failed: " << cudaGetErrorString(err) << "\n"; - failed = true; + failed->store(true, std::memory_order_relaxed); + burst_failed = true; break; } } @@ -480,12 +488,13 @@ void tx_worker(PcapTxConfig cfg, std::atomic *stop) { msg, i, {static_cast(cfg.raw.header_size + cfg.raw.payload_size)}) != daqiri::Status::SUCCESS) { - failed = true; + failed->store(true, std::memory_order_relaxed); + burst_failed = true; break; } } - if (failed) { + if (burst_failed) { daqiri::free_all_packets_and_burst_tx(msg); stop->store(true, std::memory_order_relaxed); break; @@ -493,6 +502,7 @@ void tx_worker(PcapTxConfig cfg, std::atomic *stop) { if (daqiri::send_tx_burst(msg) != daqiri::Status::SUCCESS) { std::cerr << "send_tx_burst failed\n"; + failed->store(true, std::memory_order_relaxed); stop->store(true, std::memory_order_relaxed); break; } @@ -600,9 +610,10 @@ int main(int argc, char **argv) { writer.open_file(); std::atomic stop{false}; + std::atomic failed{false}; std::thread tx_thread; if (tx_cfg) { - tx_thread = std::thread(tx_worker, *tx_cfg, &stop); + tx_thread = std::thread(tx_worker, *tx_cfg, &stop, &failed); } std::cout << "Capturing to " << args.output_path @@ -619,7 +630,7 @@ int main(int argc, char **argv) { daqiri::print_stats(); daqiri::shutdown(); daqiri_initialized = false; - return 0; + return failed.load(std::memory_order_relaxed) ? 1 : 0; } catch (const std::exception &e) { std::cerr << "pcap writer failed: " << e.what() << "\n"; if (daqiri_initialized) { diff --git a/examples/raw_bench_common.cpp b/examples/raw_bench_common.cpp index 9c35450f..91ea0baa 100644 --- a/examples/raw_bench_common.cpp +++ b/examples/raw_bench_common.cpp @@ -58,6 +58,19 @@ bool has_bench_config(const YAML::Node &root, const char *key) { return node.IsMap() && node["interface_name"]; } +int mac_hex_digit(char value) { + if (value >= '0' && value <= '9') { + return value - '0'; + } + if (value >= 'a' && value <= 'f') { + return value - 'a' + 10; + } + if (value >= 'A' && value <= 'F') { + return value - 'A' + 10; + } + return -1; +} + std::vector queue_ids_for_interface(const YAML::Node &root, const std::string &interface_name, const char *direction) { @@ -195,6 +208,40 @@ RawBenchTxConfig parse_tx_item(const YAML::Node &tx) { } // namespace +bool resolve_tx_eth_dst(int port, const std::string& ip_dst_addr, const std::string& eth_dst_addr, + char* eth_dst, uint32_t timeout_ms) { + if (eth_dst == nullptr) { + std::cerr << "Missing Ethernet destination output buffer\n"; + return false; + } + + if (!eth_dst_addr.empty()) { + if (eth_dst_addr.size() != 17) { + std::cerr << "Invalid eth_dst_addr: " << eth_dst_addr << "\n"; + return false; + } + for (size_t octet = 0; octet < 6; ++octet) { + const size_t offset = octet * 3; + if ((octet != 5 && eth_dst_addr[offset + 2] != ':') || + mac_hex_digit(eth_dst_addr[offset]) < 0 || mac_hex_digit(eth_dst_addr[offset + 1]) < 0) { + std::cerr << "Invalid eth_dst_addr: " << eth_dst_addr << "\n"; + return false; + } + eth_dst[octet] = static_cast((mac_hex_digit(eth_dst_addr[offset]) << 4) | + mac_hex_digit(eth_dst_addr[offset + 1])); + } + return true; + } + + const Status status = daqiri::resolve_ipv4_mac(port, ip_dst_addr, eth_dst, timeout_ms); + if (status != Status::SUCCESS) { + std::cerr << "Failed to resolve MAC for " << ip_dst_addr << " (DAQIRI status " + << static_cast(status) << ")\n"; + return false; + } + return true; +} + PinnedHostBuffer::PinnedHostBuffer(PinnedHostBuffer &&other) noexcept { ptr_ = other.ptr_; capacity_ = other.capacity_; diff --git a/examples/raw_bench_common.h b/examples/raw_bench_common.h index e57cf410..1cb0ea79 100644 --- a/examples/raw_bench_common.h +++ b/examples/raw_bench_common.h @@ -74,7 +74,7 @@ struct RawBenchTxConfig { std::string ip_src_addr = "1.2.3.4"; std::string ip_dst_addr = "5.6.7.8"; std::string eth_src_addr = "00:00:00:00:00:00"; - std::string eth_dst_addr = "00:00:00:00:00:00"; + std::string eth_dst_addr; }; struct RawBenchRxConfig { @@ -121,6 +121,8 @@ std::vector parse_rx_configs(const YAML::Node &root); std::vector parse_tx_configs(const YAML::Node &root); std::vector parse_udp_ports(const std::string &spec); bool set_current_thread_affinity(int cpu_core, const std::string &thread_name); +bool resolve_tx_eth_dst(int port, const std::string& ip_dst_addr, const std::string& eth_dst_addr, + char* eth_dst, uint32_t timeout_ms = 3000); void populate_udp_ipv4_headers(uint8_t *pkt_data, uint32_t header_size, uint32_t payload_size, const char *eth_src, diff --git a/examples/raw_gpudirect_bench.cpp b/examples/raw_gpudirect_bench.cpp index 6c788f08..a8df2851 100644 --- a/examples/raw_gpudirect_bench.cpp +++ b/examples/raw_gpudirect_bench.cpp @@ -37,10 +37,10 @@ namespace { -void tx_worker(const daqiri::bench::RawBenchTxConfig &cfg, - daqiri::bench::TokenBucketPacer &pacer, - std::atomic &stop) { +void tx_worker(const daqiri::bench::RawBenchTxConfig& cfg, daqiri::bench::TokenBucketPacer& pacer, + std::atomic& stop, std::atomic& failed) { if (!daqiri::bench::set_current_thread_affinity(cfg.cpu_core, "bench_tx")) { + failed.store(true); stop.store(true); return; } @@ -48,6 +48,7 @@ void tx_worker(const daqiri::bench::RawBenchTxConfig &cfg, const int port_id = daqiri::get_port_id(cfg.interface_name); if (port_id < 0) { std::cerr << "Invalid TX interface_name: " << cfg.interface_name << "\n"; + failed.store(true); stop.store(true); return; } @@ -55,7 +56,11 @@ void tx_worker(const daqiri::bench::RawBenchTxConfig &cfg, char eth_dst[6] = {0}; char eth_src[6] = {0}; daqiri::format_eth_addr(eth_src, cfg.eth_src_addr); - daqiri::format_eth_addr(eth_dst, cfg.eth_dst_addr); + if (!daqiri::bench::resolve_tx_eth_dst(port_id, cfg.ip_dst_addr, cfg.eth_dst_addr, eth_dst)) { + failed.store(true); + stop.store(true); + return; + } uint32_t ip_src = 0; uint32_t ip_dst = 0; @@ -200,6 +205,7 @@ int main(int argc, char **argv) { } std::atomic stop{false}; + std::atomic failed{false}; std::vector tx_threads; std::vector rx_threads; daqiri::bench::TokenBucketPacer tx_pacer(target_gbps); @@ -229,7 +235,7 @@ int main(int argc, char **argv) { } tx_threads.reserve(tx_configs.size()); for (const auto &cfg : tx_configs) { - tx_threads.emplace_back(tx_worker, cfg, std::ref(tx_pacer), std::ref(stop)); + tx_threads.emplace_back(tx_worker, cfg, std::ref(tx_pacer), std::ref(stop), std::ref(failed)); } daqiri::bench::wait_for_stop(run_seconds, stop); @@ -250,5 +256,5 @@ int main(int argc, char **argv) { if (prometheus_metrics) { daqiri::bench::grafana::shutdown_prometheus_metrics(); } - return 0; + return failed.load() ? 1 : 0; } diff --git a/examples/raw_hds_bench.cpp b/examples/raw_hds_bench.cpp index b9e4c6a1..39cc1119 100644 --- a/examples/raw_hds_bench.cpp +++ b/examples/raw_hds_bench.cpp @@ -35,9 +35,10 @@ namespace { -void tx_worker(const daqiri::bench::RawBenchTxConfig &cfg, - std::atomic &stop) { +void tx_worker(const daqiri::bench::RawBenchTxConfig& cfg, std::atomic& stop, + std::atomic& failed) { if (!daqiri::bench::set_current_thread_affinity(cfg.cpu_core, "bench_tx")) { + failed.store(true); stop.store(true); return; } @@ -45,12 +46,17 @@ void tx_worker(const daqiri::bench::RawBenchTxConfig &cfg, const int port_id = daqiri::get_port_id(cfg.interface_name); if (port_id < 0) { std::cerr << "Invalid TX interface_name: " << cfg.interface_name << "\n"; + failed.store(true); stop.store(true); return; } char eth_dst[6] = {0}; - daqiri::format_eth_addr(eth_dst, cfg.eth_dst_addr); + if (!daqiri::bench::resolve_tx_eth_dst(port_id, cfg.ip_dst_addr, cfg.eth_dst_addr, eth_dst)) { + failed.store(true); + stop.store(true); + return; + } uint32_t ip_src = 0; uint32_t ip_dst = 0; @@ -166,6 +172,7 @@ int main(int argc, char **argv) { } std::atomic stop{false}; + std::atomic failed{false}; std::thread tx_thread; std::thread rx_thread; @@ -194,7 +201,7 @@ int main(int argc, char **argv) { } if (has_tx) { tx_thread = - std::thread(tx_worker, daqiri::bench::parse_tx(root), std::ref(stop)); + std::thread(tx_worker, daqiri::bench::parse_tx(root), std::ref(stop), std::ref(failed)); } daqiri::bench::wait_for_stop(run_seconds, stop); @@ -208,5 +215,5 @@ int main(int argc, char **argv) { daqiri::print_stats(); daqiri::shutdown(); - return 0; + return failed.load() ? 1 : 0; } diff --git a/examples/raw_latency_bench.cpp b/examples/raw_latency_bench.cpp index c691994d..0105cbb3 100644 --- a/examples/raw_latency_bench.cpp +++ b/examples/raw_latency_bench.cpp @@ -190,7 +190,7 @@ BenchConfig parse_bench_config(const YAML::Node& root) { cfg.rx_queue_id = node["rx_queue_id"].as(common_queue); cfg.cpu_core = node["cpu_core"].as(cfg.cpu_core); cfg.ptp_device = node["ptp_device"].as(""); - cfg.eth_dst_addr = node["eth_dst_addr"].as(); + cfg.eth_dst_addr = node["eth_dst_addr"].as(""); cfg.ip_src_addr = node["ip_src_addr"].as(cfg.ip_src_addr); cfg.ip_dst_addr = node["ip_dst_addr"].as(cfg.ip_dst_addr); cfg.udp_src_port = node["udp_src_port"].as(cfg.udp_src_port); @@ -318,24 +318,51 @@ void initialize_cuda() { std::array parse_mac(const std::string& value) { std::array mac{}; - unsigned int bytes[ETH_ALEN]{}; - if (std::sscanf(value.c_str(), "%x:%x:%x:%x:%x:%x", &bytes[0], &bytes[1], &bytes[2], &bytes[3], - &bytes[4], &bytes[5]) != ETH_ALEN) { + const auto hex_digit = [](char digit) { + if (digit >= '0' && digit <= '9') { + return digit - '0'; + } + if (digit >= 'a' && digit <= 'f') { + return digit - 'a' + 10; + } + if (digit >= 'A' && digit <= 'F') { + return digit - 'A' + 10; + } + return -1; + }; + if (value.size() != 17) { throw std::invalid_argument("invalid eth_dst_addr: " + value); } for (size_t i = 0; i < mac.size(); ++i) { - if (bytes[i] > 0xff) { + const size_t offset = i * 3; + const int high = hex_digit(value[offset]); + const int low = hex_digit(value[offset + 1]); + if (high < 0 || low < 0 || (i != mac.size() - 1 && value[offset + 2] != ':')) { throw std::invalid_argument("invalid eth_dst_addr: " + value); } - mac[i] = static_cast(bytes[i]); + mac[i] = static_cast((high << 4) | low); + } + return mac; +} + +std::array resolve_mac(int port, const BenchConfig& cfg) { + if (!cfg.eth_dst_addr.empty()) { + return parse_mac(cfg.eth_dst_addr); + } + std::array mac{}; + const daqiri::Status status = + daqiri::resolve_ipv4_mac(port, cfg.ip_dst_addr, reinterpret_cast(mac.data()), 3000); + if (status != daqiri::Status::SUCCESS) { + throw std::runtime_error("failed to resolve bench_latency destination MAC (DAQIRI status " + + std::to_string(static_cast(status)) + ")"); } return mac; } -std::vector make_packet_template(uint32_t packet_size, const BenchConfig& cfg) { +std::vector make_packet_template(uint32_t packet_size, const BenchConfig& cfg, + const std::array& dst) { std::vector data(packet_size, 0); auto* pkt = reinterpret_cast(data.data()); - const auto dst = parse_mac(cfg.eth_dst_addr); std::memcpy(pkt->eth.h_dest, dst.data(), dst.size()); pkt->eth.h_proto = htons(ETH_P_IP); pkt->ip.version = 4; @@ -536,6 +563,7 @@ int main(int argc, char** argv) { daqiri::shutdown(); return 1; } + const auto dst_mac = resolve_mac(tx_port_id, cfg); std::vector all_samples; all_samples.reserve(static_cast(options.samples) * 8); @@ -556,7 +584,7 @@ int main(int argc, char** argv) { std::cout << '\n'; for (uint32_t packet_size = kMinPacketSize; packet_size <= kMaxPacketSize; packet_size *= 2) { - const auto packet_template = make_packet_template(packet_size, cfg); + const auto packet_template = make_packet_template(packet_size, cfg, dst_mac); drain_rx(rx_port_id, cfg.rx_queue_id); const std::optional clock_before = cfg.ptp_device.empty() diff --git a/examples/raw_reorder_quantize_bench.cpp b/examples/raw_reorder_quantize_bench.cpp index f65b2f11..35a85ca4 100644 --- a/examples/raw_reorder_quantize_bench.cpp +++ b/examples/raw_reorder_quantize_bench.cpp @@ -92,7 +92,7 @@ struct BenchConfig { uint32_t payload_size = 256; uint32_t header_size = 64; std::string eth_src_addr = "00:00:00:00:00:00"; - std::string eth_dst_addr = "00:00:00:00:00:00"; + std::string eth_dst_addr; std::string ip_src_addr = "1.2.3.4"; std::string ip_dst_addr = "5.6.7.8"; uint16_t udp_src_port = 4096; @@ -629,7 +629,11 @@ void tx_worker(const BenchConfig &cfg, const ReorderPlanConfig &plan, char eth_dst[6] = {0}; char eth_src[6] = {0}; daqiri::format_eth_addr(eth_src, cfg.eth_src_addr); - daqiri::format_eth_addr(eth_dst, cfg.eth_dst_addr); + if (!daqiri::bench::resolve_tx_eth_dst(port_id, cfg.ip_dst_addr, cfg.eth_dst_addr, eth_dst)) { + stats.failures.fetch_add(1); + stop.store(true); + return; + } uint32_t ip_src = 0; uint32_t ip_dst = 0; diff --git a/examples/raw_reorder_seq_bench.cpp b/examples/raw_reorder_seq_bench.cpp index f0c00b57..30b8b3ae 100644 --- a/examples/raw_reorder_seq_bench.cpp +++ b/examples/raw_reorder_seq_bench.cpp @@ -104,9 +104,10 @@ parse_gpu_reorder_plans(const YAML::Node &root) { return plans; } -void tx_worker(const SequenceTxConfig &cfg, std::atomic &stop) { +void tx_worker(const SequenceTxConfig& cfg, std::atomic& stop, std::atomic& failed) { if (!daqiri::bench::set_current_thread_affinity(cfg.packet.cpu_core, "bench_tx")) { + failed.store(true); stop.store(true); return; } @@ -115,18 +116,25 @@ void tx_worker(const SequenceTxConfig &cfg, std::atomic &stop) { if (port_id < 0) { std::cerr << "Invalid TX interface_name: " << cfg.packet.interface_name << "\n"; + failed.store(true); stop.store(true); return; } if (cfg.sequence_number_offset + sizeof(uint32_t) > cfg.packet.payload_size) { std::cerr << "sequence_number_offset out of payload range\n"; + failed.store(true); stop.store(true); return; } char eth_dst[6] = {0}; - daqiri::format_eth_addr(eth_dst, cfg.packet.eth_dst_addr); + if (!daqiri::bench::resolve_tx_eth_dst(port_id, cfg.packet.ip_dst_addr, cfg.packet.eth_dst_addr, + eth_dst)) { + failed.store(true); + stop.store(true); + return; + } uint32_t ip_src = 0; uint32_t ip_dst = 0; @@ -397,6 +405,7 @@ int main(int argc, char **argv) { } std::atomic stop{false}; + std::atomic failed{false}; std::thread tx_thread; std::thread rx_thread; @@ -405,7 +414,7 @@ int main(int argc, char **argv) { std::ref(stop)); } if (has_tx) { - tx_thread = std::thread(tx_worker, parse_sequence_tx(root), std::ref(stop)); + tx_thread = std::thread(tx_worker, parse_sequence_tx(root), std::ref(stop), std::ref(failed)); } daqiri::bench::wait_for_stop(run_seconds, stop); @@ -422,5 +431,5 @@ int main(int argc, char **argv) { cudaStreamDestroy(reorder_stream); } daqiri::shutdown(); - return 0; + return failed.load() ? 1 : 0; } diff --git a/include/daqiri/common.h b/include/daqiri/common.h index 3ba63f66..cbc18488 100644 --- a/include/daqiri/common.h +++ b/include/daqiri/common.h @@ -931,6 +931,38 @@ void set_connection_id(BurstParams *burst, uintptr_t conn_id); */ Status get_mac_addr(int port, char *mac); +/** + * @brief Resolve an IPv4 destination's next-hop MAC through Linux + * + * This operation is supported by the raw ibverbs engine. The selected port's + * kernel netdev supplies routing and ARP. Other engines return NOT_SUPPORTED. + * + * @param port Port number of interface + * @param dst_host IPv4 destination in host byte order + * @param mac Output buffer of at least six bytes + * @param timeout_ms Maximum time to wait for ARP resolution + * @return SUCCESS on resolution; NULL_PTR for a missing output buffer; + * INVALID_PARAMETER for an invalid port, destination, or timeout; + * NOT_READY when the neighbor fails or remains unresolved at the deadline; + * CONNECT_FAILURE for netdev, route, permission, or netlink failures; and + * NOT_SUPPORTED for engines other than raw ibverbs + */ +Status resolve_ipv4_mac(int port, uint32_t dst_host, char* mac, uint32_t timeout_ms); + +/** + * @brief Resolve an IPv4 destination's next-hop MAC from dotted-decimal text + * + * This overload has the same behavior as the host-order overload and returns + * INVALID_PARAMETER when @p dst_addr is not a valid IPv4 address. + * + * @param port Port number of interface + * @param dst_addr IPv4 destination in dotted-decimal notation + * @param mac Output buffer of at least six bytes + * @param timeout_ms Maximum time to wait for ARP resolution + * @return See the host-order resolve_ipv4_mac overload + */ +Status resolve_ipv4_mac(int port, const std::string& dst_addr, char* mac, uint32_t timeout_ms); + /** * @brief Drop all traffic on a port * diff --git a/python/daqiri_common_pybind.cpp b/python/daqiri_common_pybind.cpp index 71f35874..56c845c7 100644 --- a/python/daqiri_common_pybind.cpp +++ b/python/daqiri_common_pybind.cpp @@ -1325,6 +1325,46 @@ PYBIND11_MODULE(_daqiri, m) { return py::make_tuple(status, std::string(formatted)); }, "port"_a); + m.def( + "resolve_ipv4_mac", + [](int port, uint32_t dst_host, uint32_t timeout_ms) { + char mac[6] = {}; + Status status = Status::SUCCESS; + { + py::gil_scoped_release release; + status = resolve_ipv4_mac(port, dst_host, mac, timeout_ms); + } + if (status != Status::SUCCESS) { + return py::make_tuple(status, std::string()); + } + char formatted[18] = {}; + std::snprintf(formatted, sizeof(formatted), "%02x:%02x:%02x:%02x:%02x:%02x", + static_cast(mac[0]), static_cast(mac[1]), + static_cast(mac[2]), static_cast(mac[3]), + static_cast(mac[4]), static_cast(mac[5])); + return py::make_tuple(status, std::string(formatted)); + }, + "port"_a, "dst"_a, "timeout_ms"_a); + m.def( + "resolve_ipv4_mac", + [](int port, const std::string& dst_addr, uint32_t timeout_ms) { + char mac[6] = {}; + Status status = Status::SUCCESS; + { + py::gil_scoped_release release; + status = resolve_ipv4_mac(port, dst_addr, mac, timeout_ms); + } + if (status != Status::SUCCESS) { + return py::make_tuple(status, std::string()); + } + char formatted[18] = {}; + std::snprintf(formatted, sizeof(formatted), "%02x:%02x:%02x:%02x:%02x:%02x", + static_cast(mac[0]), static_cast(mac[1]), + static_cast(mac[2]), static_cast(mac[3]), + static_cast(mac[4]), static_cast(mac[5])); + return py::make_tuple(status, std::string(formatted)); + }, + "port"_a, "dst"_a, "timeout_ms"_a); m.def( "format_eth_addr", [](const std::string &addr) { diff --git a/scripts/setup_spark_xhost_net.sh b/scripts/setup_spark_xhost_net.sh index adafe62c..e77d6423 100755 --- a/scripts/setup_spark_xhost_net.sh +++ b/scripts/setup_spark_xhost_net.sh @@ -1,10 +1,11 @@ #!/bin/bash # Host network config for two-DGX-Spark cross-cable benches (p0 <-> p0). -# Adds the host route and static neighbor the kernel needs to reach the peer +# Adds the host route the kernel needs to reach the peer # when daqiri-tx / daqiri-rx nmcli profiles are split across two boxes. # # Matches examples/*_spark_xhost.yaml (1.1.1.1 on TX, 2.2.2.2 on RX). -# Re-running is safe: replaces route and neighbor entries. +# Re-running is safe: replaces the route. An optional peer MAC preserves the +# old static-neighbor setup, but DAQIRI's ibverbs raw engine can resolve it. # # Conflicts with scripts/setup_spark_rdma_loopback.sh on the same host: that # script reassigns 1.1.1.1 / 2.2.2.2 across two local ports for inter-port @@ -12,8 +13,8 @@ # re-up the right nmcli profile (daqiri-tx or daqiri-rx), then re-run this. # # Usage: -# sudo scripts/setup_spark_xhost_net.sh --role tx --peer-mac -# sudo scripts/setup_spark_xhost_net.sh --role rx --peer-mac +# sudo scripts/setup_spark_xhost_net.sh --role tx +# sudo scripts/setup_spark_xhost_net.sh --role rx # # Env overrides: SPARK_XHOST_IFACE, SPARK_TX_IP, SPARK_RX_IP, # SPARK_TX_INT_IP, SPARK_RX_INT_IP (169.254 cable-side addresses for SSH/mgmt) @@ -31,13 +32,12 @@ peer_mac="" usage() { cat < [--iface IFACE] [--peer-ip IP] +Usage: $0 --role tx|rx [--iface IFACE] [--peer-ip IP] [--peer-mac MAC] tx local ${tx_ip} on IFACE; peer defaults to ${rx_ip} rx local ${rx_ip} on IFACE; peer defaults to ${tx_ip} -Read the peer MAC on the other host: - cat /sys/class/net/${iface}/address +--peer-mac is optional and installs a permanent neighbor for compatibility. Verify after running: ping -c1 @@ -86,8 +86,8 @@ while [[ $# -gt 0 ]]; do esac done -if [[ -z "$role" || -z "$peer_mac" ]]; then - echo "error: --role and --peer-mac are required" >&2 +if [[ -z "$role" ]]; then + echo "error: --role is required" >&2 usage exit 1 fi @@ -120,9 +120,13 @@ fi # Bench peer: pin egress source to the local bench address, not 169.254/16. ip route replace "${peer_ip}/32" dev "${iface}" src "${local_ip}" -ip neigh replace "${peer_ip}" lladdr "${peer_mac}" dev "${iface}" nud permanent +if [[ -n "${peer_mac}" ]]; then + ip neigh replace "${peer_ip}" lladdr "${peer_mac}" dev "${iface}" nud permanent +else + ip neigh del "${peer_ip}" dev "${iface}" 2>/dev/null || true +fi -# Cable internal address (169.254.x on p0): route + neighbor for SSH between hosts. +# Cable internal address (169.254.x on p0): route for SSH between hosts. # Only applied when the configured local_int_ip is actually assigned to iface; # some hosts put 169.254/16 on the *other* cable instead, in which case skip. if [[ "$role" == "tx" ]]; then @@ -132,10 +136,15 @@ else peer_int_ip="${peer_int_ip:-$tx_int_ip}" local_int_ip="$rx_int_ip" fi +if [[ -z "${peer_mac}" && -n "${peer_int_ip}" ]]; then + ip neigh del "${peer_int_ip}" dev "${iface}" 2>/dev/null || true +fi if [[ -n "$peer_int_ip" && "$peer_int_ip" != "$peer_ip" ]] && \ ip -4 -o addr show dev "${iface}" | grep -qw "${local_int_ip}"; then ip route replace "${peer_int_ip}/32" dev "${iface}" src "${local_int_ip}" - ip neigh replace "${peer_int_ip}" lladdr "${peer_mac}" dev "${iface}" nud permanent + if [[ -n "${peer_mac}" ]]; then + ip neigh replace "${peer_int_ip}" lladdr "${peer_mac}" dev "${iface}" nud permanent + fi int_applied=1 else int_applied=0 @@ -143,9 +152,13 @@ fi echo "Cross-host network config applied (${role} role)." echo " ${iface} (${local_mac}) bench ${local_ip}/24" -echo " peer bench ${peer_ip} -> ${peer_mac}" +if [[ -n "${peer_mac}" ]]; then + echo " peer bench ${peer_ip} -> ${peer_mac} (static neighbor)" +else + echo " peer bench ${peer_ip} (dynamic ARP)" +fi if [[ "${int_applied}" == "1" ]]; then - echo " peer internal ${peer_int_ip} -> ${peer_mac} (via ${iface}, src ${local_int_ip})" + echo " peer internal ${peer_int_ip} (via ${iface}, src ${local_int_ip})" else echo " peer internal route skipped (${local_int_ip} not on ${iface})" fi diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index bd7b007f..8b182782 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -145,6 +145,7 @@ add_library(daqiri_common kernels.cu engine.cpp logging.cpp + net_neighbor.cpp net_pause.cpp ) if(DAQIRI_ENABLE_OTEL_METRICS) diff --git a/src/common.cpp b/src/common.cpp index 6f63e897..d2cc9d45 100644 --- a/src/common.cpp +++ b/src/common.cpp @@ -496,6 +496,22 @@ Status get_mac_addr(int port, char* mac) { return g_daqiri_engine->get_mac_addr(port, mac); } +Status resolve_ipv4_mac(int port, uint32_t dst_host, char* mac, uint32_t timeout_ms) { + ASSERT_DAQIRI_ENGINE_INITIALIZED(); + return g_daqiri_engine->resolve_ipv4_mac(port, dst_host, mac, timeout_ms); +} + +Status resolve_ipv4_mac(int port, const std::string& dst_addr, char* mac, uint32_t timeout_ms) { + if (mac == nullptr) { + return Status::NULL_PTR; + } + in_addr dst{}; + if (inet_pton(AF_INET, dst_addr.c_str(), &dst) != 1) { + return Status::INVALID_PARAMETER; + } + return resolve_ipv4_mac(port, ntohl(dst.s_addr), mac, timeout_ms); +} + Status drop_all_traffic(int port) { ASSERT_DAQIRI_ENGINE_INITIALIZED(); return g_daqiri_engine->drop_all_traffic(port); diff --git a/src/engine.cpp b/src/engine.cpp index 3ce28b72..1076cbb6 100644 --- a/src/engine.cpp +++ b/src/engine.cpp @@ -1579,6 +1579,18 @@ Status Engine::wait_for_tx_idle(uint32_t timeout_ms) { return Status::NOT_SUPPORTED; } +Status Engine::resolve_ipv4_mac(int port, uint32_t dst_host, char* mac, uint32_t timeout_ms) { + if (mac == nullptr) { + return Status::NULL_PTR; + } + if (port < 0 || port >= static_cast(cfg_.ifs_.size()) || dst_host == INADDR_ANY || + dst_host == INADDR_BROADCAST || IN_MULTICAST(dst_host) || IN_BADCLASS(dst_host) || + timeout_ms == 0) { + return Status::INVALID_PARAMETER; + } + return Status::NOT_SUPPORTED; +} + Status Engine::get_rx_burst(BurstParams** burst, int port_id) { // Check if the port_id is valid if (port_id < 0 || port_id >= static_cast(cfg_.ifs_.size())) { diff --git a/src/engine.h b/src/engine.h index a64f4196..1a523918 100644 --- a/src/engine.h +++ b/src/engine.h @@ -187,6 +187,7 @@ class Engine { virtual Status send_tx_burst(BurstParams* burst) = 0; virtual Status wait_for_tx_idle(uint32_t timeout_ms); virtual Status get_mac_addr(int port, char* mac) = 0; + virtual Status resolve_ipv4_mac(int port, uint32_t dst_host, char* mac, uint32_t timeout_ms); virtual Status drop_all_traffic(int port); virtual Status allow_all_traffic(int port); virtual Status add_rx_flow_async(int port, const FlowRuleConfig& flow, FlowOpId* op_id); diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp index 50c92118..3d778b66 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/net_neighbor.h" #include "src/net_pause.h" #include "src/rss.h" @@ -8218,6 +8219,33 @@ Status IbverbsEngine::set_eth_header(BurstParams* burst, int idx, char* dst_addr return Status::SUCCESS; } +Status IbverbsEngine::resolve_ipv4_mac(int port, uint32_t dst_host, char* mac, + uint32_t timeout_ms) { + if (mac == nullptr) { + return Status::NULL_PTR; + } + if (port < 0 || port >= static_cast(cfg_.ifs_.size()) || dst_host == INADDR_ANY || + dst_host == INADDR_BROADCAST || IN_MULTICAST(dst_host) || IN_BADCLASS(dst_host) || + timeout_ms == 0) { + return Status::INVALID_PARAMETER; + } + if (cfg_.common_.loopback_ != LoopbackType::DISABLED) { + return Status::NOT_SUPPORTED; + } + const std::string netdev = port_netdev(port); + if (netdev.empty()) { + DAQIRI_LOG_ERROR("ARP: no kernel netdev found for ibverbs port {}", port); + return Status::CONNECT_FAILURE; + } + if (!cfg_.ifs_[port].rx_.queues_.empty() && !cfg_.ifs_[port].rx_.flow_isolation_) { + DAQIRI_LOG_WARN( + "ARP: port {} does not enable rx.flow_isolation; catch-all application steering may " + "consume ARP traffic before Linux receives it", + port); + } + return resolve_ipv4_neighbor(netdev, dst_host, mac, timeout_ms); +} + Status IbverbsEngine::set_ipv4_header(BurstParams* burst, int idx, int ip_len, uint8_t proto, unsigned int src_host, unsigned int dst_host) { auto* pkt = static_cast(burst->pkts[0][idx]); diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.h b/src/engines/ibverbs/daqiri_ibverbs_engine.h index 923f8886..c90e2011 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.h +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.h @@ -498,6 +498,7 @@ class IbverbsEngine : public Engine { // TX construction / header fill (implemented in the TX milestone) Status get_tx_packet_burst(BurstParams* burst) override; Status get_tx_packet_burst_checked(BurstParams* burst) override; + Status resolve_ipv4_mac(int port, uint32_t dst_host, char* mac, uint32_t timeout_ms) override; Status set_eth_header(BurstParams* burst, int idx, char* dst_addr) override; Status set_ipv4_header(BurstParams* burst, int idx, int ip_len, uint8_t proto, unsigned int src_host, unsigned int dst_host) override; diff --git a/src/net_neighbor.cpp b/src/net_neighbor.cpp new file mode 100644 index 00000000..2bd2227d --- /dev/null +++ b/src/net_neighbor.cpp @@ -0,0 +1,605 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. + * All rights reserved. SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "src/net_neighbor.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace daqiri { +namespace { + +constexpr uint16_t kConfirmedNeighborStates = NUD_REACHABLE | NUD_PERMANENT | NUD_NOARP; +constexpr uint16_t kRefreshNeighborStates = NUD_STALE | NUD_DELAY | NUD_PROBE; + +struct NetlinkRequest { + std::array storage{}; + + nlmsghdr* header() { + return reinterpret_cast(storage.data()); + } + + bool add_attr(uint16_t type, const void* data, size_t size) { + nlmsghdr* hdr = header(); + const size_t offset = NLMSG_ALIGN(hdr->nlmsg_len); + const size_t attr_size = RTA_LENGTH(size); + const size_t next = offset + RTA_ALIGN(attr_size); + if (next > storage.size()) { + return false; + } + auto* attr = reinterpret_cast(storage.data() + offset); + attr->rta_type = type; + attr->rta_len = static_cast(attr_size); + std::memcpy(RTA_DATA(attr), data, size); + hdr->nlmsg_len = static_cast(next); + return true; + } +}; + +class NetlinkSocket { + public: + NetlinkSocket() { + fd_ = socket(AF_NETLINK, SOCK_RAW | SOCK_CLOEXEC, NETLINK_ROUTE); + if (fd_ < 0) { + error_ = errno; + return; + } + + sockaddr_nl local{}; + local.nl_family = AF_NETLINK; + local.nl_groups = RTMGRP_NEIGH; + if (bind(fd_, reinterpret_cast(&local), sizeof(local)) != 0) { + error_ = errno; + close(fd_); + fd_ = -1; + return; + } + + socklen_t len = sizeof(local); + if (getsockname(fd_, reinterpret_cast(&local), &len) != 0) { + error_ = errno; + close(fd_); + fd_ = -1; + return; + } + pid_ = local.nl_pid; + } + + ~NetlinkSocket() { + if (fd_ >= 0) { + close(fd_); + } + } + + NetlinkSocket(const NetlinkSocket&) = delete; + NetlinkSocket& operator=(const NetlinkSocket&) = delete; + + bool valid() const { + return fd_ >= 0; + } + int fd() const { + return fd_; + } + uint32_t next_sequence() { + return ++sequence_; + } + uint32_t pid() const { + return pid_; + } + int error() const { + return error_; + } + + bool send(const nlmsghdr* hdr) const { + sockaddr_nl kernel{}; + kernel.nl_family = AF_NETLINK; + iovec iov{const_cast(hdr), hdr->nlmsg_len}; + msghdr msg{}; + msg.msg_name = &kernel; + msg.msg_namelen = sizeof(kernel); + msg.msg_iov = &iov; + msg.msg_iovlen = 1; + return sendmsg(fd_, &msg, 0) == static_cast(hdr->nlmsg_len); + } + + private: + int fd_ = -1; + int error_ = 0; + uint32_t pid_ = 0; + uint32_t sequence_ = 0; +}; + +struct RouteResult { + int ifindex = 0; + uint32_t next_hop_network = 0; +}; + +enum class NeighborState { MISSING, PENDING, REFRESH_NEEDED, USABLE, FAILED }; + +struct NeighborResult { + NeighborState state = NeighborState::MISSING; + std::array mac{}; +}; + +bool is_terminal(const NeighborResult& result) { + return result.state == NeighborState::USABLE || result.state == NeighborState::FAILED; +} + +using Deadline = std::chrono::steady_clock::time_point; +using NetlinkConsumer = std::function; + +void parse_attrs(rtattr** attrs, size_t count, rtattr* attr, int len) { + std::fill(attrs, attrs + count, nullptr); + while (RTA_OK(attr, len)) { + if (attr->rta_type < count) { + attrs[attr->rta_type] = attr; + } + attr = RTA_NEXT(attr, len); + } +} + +int receive_for_sequence(NetlinkSocket& sock, uint32_t sequence, const Deadline& deadline, + const NetlinkConsumer& consume, + const NetlinkConsumer& consume_notification = {}) { + std::array buffer{}; + while (true) { + const auto now = std::chrono::steady_clock::now(); + if (now >= deadline) { + return -ETIMEDOUT; + } + const auto time_remaining = + std::chrono::duration_cast(deadline - now); + pollfd descriptor{sock.fd(), POLLIN, 0}; + const int wait_ms = static_cast(std::min(time_remaining.count() + 1, INT_MAX)); + const int ready = poll(&descriptor, 1, wait_ms); + if (ready < 0) { + if (errno == EINTR) { + continue; + } + return -errno; + } + if (ready == 0) { + return -ETIMEDOUT; + } + + const ssize_t size = recv(sock.fd(), buffer.data(), buffer.size(), 0); + if (size < 0) { + if (errno == EINTR) { + continue; + } + return -errno; + } + int remaining = static_cast(size); + for (auto* hdr = reinterpret_cast(buffer.data()); NLMSG_OK(hdr, remaining); + hdr = NLMSG_NEXT(hdr, remaining)) { + if (hdr->nlmsg_seq != sequence) { + if (consume_notification) { + consume_notification(hdr); + } + continue; + } + if (hdr->nlmsg_type == NLMSG_ERROR) { + const auto* error = static_cast(NLMSG_DATA(hdr)); + return error->error; + } + if (consume(hdr)) { + return 0; + } + if (hdr->nlmsg_type == NLMSG_DONE) { + return 0; + } + } + } +} + +int lookup_route(NetlinkSocket& sock, int requested_ifindex, uint32_t src_network, + uint32_t dst_network, const Deadline& deadline, RouteResult* result) { + NetlinkRequest request; + nlmsghdr* hdr = request.header(); + hdr->nlmsg_len = NLMSG_LENGTH(sizeof(rtmsg)); + hdr->nlmsg_type = RTM_GETROUTE; + hdr->nlmsg_flags = NLM_F_REQUEST; + hdr->nlmsg_seq = sock.next_sequence(); + hdr->nlmsg_pid = sock.pid(); + auto* route = static_cast(NLMSG_DATA(hdr)); + route->rtm_family = AF_INET; + route->rtm_dst_len = 32; + route->rtm_src_len = 32; + route->rtm_table = RT_TABLE_UNSPEC; + + if (!request.add_attr(RTA_DST, &dst_network, sizeof(dst_network)) || + !request.add_attr(RTA_SRC, &src_network, sizeof(src_network)) || + !request.add_attr(RTA_OIF, &requested_ifindex, sizeof(requested_ifindex))) { + return -EMSGSIZE; + } + if (!sock.send(hdr)) { + return -errno; + } + + bool found = false; + const int status = + receive_for_sequence(sock, hdr->nlmsg_seq, deadline, [&](const nlmsghdr* response) { + if (response->nlmsg_type != RTM_NEWROUTE) { + return false; + } + const auto* message = static_cast(NLMSG_DATA(response)); + if (message->rtm_family != AF_INET || message->rtm_type != RTN_UNICAST) { + return false; + } + int attr_len = RTM_PAYLOAD(response); + std::array attrs{}; + parse_attrs(attrs.data(), attrs.size(), RTM_RTA(message), attr_len); + if (attrs[RTA_OIF] == nullptr) { + return false; + } + std::memcpy(&result->ifindex, RTA_DATA(attrs[RTA_OIF]), sizeof(result->ifindex)); + result->next_hop_network = dst_network; + if (attrs[RTA_GATEWAY] != nullptr) { + std::memcpy(&result->next_hop_network, RTA_DATA(attrs[RTA_GATEWAY]), + sizeof(result->next_hop_network)); + } + found = true; + return true; + }); + if (status != 0) { + return status; + } + return found ? 0 : -ENETUNREACH; +} + +bool parse_neighbor(const nlmsghdr* hdr, int ifindex, uint32_t address_network, + NeighborResult* result) { + if (hdr->nlmsg_type != RTM_NEWNEIGH && hdr->nlmsg_type != RTM_GETNEIGH) { + return false; + } + const auto* message = static_cast(NLMSG_DATA(hdr)); + if (message->ndm_family != AF_INET || message->ndm_ifindex != ifindex) { + return false; + } + + int attr_len = NLMSG_PAYLOAD(hdr, sizeof(*message)); + std::array attrs{}; + auto* first_attr = reinterpret_cast( + reinterpret_cast(const_cast(message)) + NLMSG_ALIGN(sizeof(*message))); + parse_attrs(attrs.data(), attrs.size(), first_attr, attr_len); + if (attrs[NDA_DST] == nullptr || RTA_PAYLOAD(attrs[NDA_DST]) < sizeof(address_network)) { + return false; + } + uint32_t candidate = 0; + std::memcpy(&candidate, RTA_DATA(attrs[NDA_DST]), sizeof(candidate)); + if (candidate != address_network) { + return false; + } + + if ((message->ndm_state & NUD_FAILED) != 0) { + result->state = NeighborState::FAILED; + } else if (attrs[NDA_LLADDR] != nullptr && RTA_PAYLOAD(attrs[NDA_LLADDR]) == result->mac.size() && + (message->ndm_state & (kConfirmedNeighborStates | kRefreshNeighborStates)) != 0) { + std::memcpy(result->mac.data(), RTA_DATA(attrs[NDA_LLADDR]), result->mac.size()); + result->state = (message->ndm_state & kConfirmedNeighborStates) != 0 + ? NeighborState::USABLE + : NeighborState::REFRESH_NEEDED; + } else { + result->state = NeighborState::PENDING; + } + return true; +} + +int lookup_neighbor(NetlinkSocket& sock, int ifindex, uint32_t address_network, + const Deadline& deadline, NeighborResult* result, + bool consume_notifications = false) { + NetlinkRequest request; + nlmsghdr* hdr = request.header(); + hdr->nlmsg_len = NLMSG_LENGTH(sizeof(ndmsg)); + hdr->nlmsg_type = RTM_GETNEIGH; + hdr->nlmsg_flags = NLM_F_REQUEST; + hdr->nlmsg_seq = sock.next_sequence(); + hdr->nlmsg_pid = sock.pid(); + auto* message = static_cast(NLMSG_DATA(hdr)); + message->ndm_family = AF_INET; + message->ndm_ifindex = ifindex; + if (!request.add_attr(NDA_DST, &address_network, sizeof(address_network))) { + return -EMSGSIZE; + } + if (!sock.send(hdr)) { + return -errno; + } + + bool found = false; + NeighborResult queried; + const auto notification = [&](const nlmsghdr* response) { + return parse_neighbor(response, ifindex, address_network, result); + }; + const int status = receive_for_sequence( + sock, hdr->nlmsg_seq, deadline, + [&](const nlmsghdr* response) { + found = parse_neighbor(response, ifindex, address_network, &queried); + return found; + }, + consume_notifications ? NetlinkConsumer(notification) : NetlinkConsumer{}); + if (status == -ENOENT) { + if (!is_terminal(*result)) { + result->state = NeighborState::MISSING; + } + return 0; + } + if (status != 0) { + return status; + } + if (found && !is_terminal(*result)) { + *result = queried; + } else if (!found && !is_terminal(*result)) { + result->state = NeighborState::MISSING; + } + return 0; +} + +int update_neighbor(NetlinkSocket& sock, int ifindex, uint32_t address_network, + uint16_t request_flags, uint16_t neighbor_state, uint8_t neighbor_flags, + const Deadline& deadline, NeighborResult* result) { + NetlinkRequest request; + nlmsghdr* hdr = request.header(); + hdr->nlmsg_len = NLMSG_LENGTH(sizeof(ndmsg)); + hdr->nlmsg_type = RTM_NEWNEIGH; + hdr->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK | request_flags; + hdr->nlmsg_seq = sock.next_sequence(); + hdr->nlmsg_pid = sock.pid(); + auto* message = static_cast(NLMSG_DATA(hdr)); + message->ndm_family = AF_INET; + message->ndm_ifindex = ifindex; + message->ndm_state = neighbor_state; + message->ndm_flags = neighbor_flags; + if (!request.add_attr(NDA_DST, &address_network, sizeof(address_network))) { + return -EMSGSIZE; + } + if (!sock.send(hdr)) { + return -errno; + } + return receive_for_sequence( + sock, hdr->nlmsg_seq, deadline, [](const nlmsghdr*) { return false; }, + [&](const nlmsghdr* response) { + return parse_neighbor(response, ifindex, address_network, result); + }); +} + +int trigger_neighbor(NetlinkSocket& sock, int ifindex, uint32_t address_network, + const Deadline& deadline, NeighborResult* result) { + if (result->state == NeighborState::REFRESH_NEEDED) { + // NTF_USE transitions a stale entry only to NUD_DELAY, and NUD_PROBE sends + // unicast solicitations to the cached MAC. Neither can discover a peer + // whose MAC changed while raw TX bypassed Linux's neighbor-use path. First + // invalidate the entry; NTF_USE below then starts a fresh broadcast ARP. + const int error = update_neighbor(sock, ifindex, address_network, NLM_F_REPLACE, NUD_NONE, 0, + deadline, result); + if (error != 0) { + return error; + } + } + return update_neighbor(sock, ifindex, address_network, NLM_F_CREATE, NUD_NONE, NTF_USE, deadline, + result); +} + +const char* ipv4_text(uint32_t network, char* buffer, size_t size) { + in_addr addr{network}; + return inet_ntop(AF_INET, &addr, buffer, static_cast(size)); +} + +} // namespace + +Status resolve_ipv4_neighbor(const std::string& netdev, uint32_t dst_host, char* mac, + uint32_t timeout_ms) { + if (mac == nullptr) { + return Status::NULL_PTR; + } + if (netdev.empty() || dst_host == INADDR_ANY || dst_host == INADDR_BROADCAST || + IN_MULTICAST(dst_host) || IN_BADCLASS(dst_host) || timeout_ms == 0) { + return Status::INVALID_PARAMETER; + } + const Deadline deadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms); + + const unsigned int ifindex = if_nametoindex(netdev.c_str()); + if (ifindex == 0) { + DAQIRI_LOG_ERROR("ARP: kernel netdev '{}' does not exist: {}", netdev, strerror(errno)); + return Status::CONNECT_FAILURE; + } + + const int control_fd = socket(AF_INET, SOCK_DGRAM | SOCK_CLOEXEC, 0); + if (control_fd < 0) { + DAQIRI_LOG_ERROR("ARP: failed to inspect kernel netdev '{}': {}", netdev, strerror(errno)); + return Status::CONNECT_FAILURE; + } + ifreq request{}; + std::snprintf(request.ifr_name, sizeof(request.ifr_name), "%s", netdev.c_str()); + if (ioctl(control_fd, SIOCGIFFLAGS, &request) != 0) { + const int error = errno; + close(control_fd); + DAQIRI_LOG_ERROR("ARP: failed to read state for kernel netdev '{}': {}", netdev, + strerror(error)); + return Status::CONNECT_FAILURE; + } + if ((request.ifr_flags & IFF_UP) == 0) { + close(control_fd); + DAQIRI_LOG_ERROR("ARP: kernel netdev '{}' is down; bring it up before resolving neighbors", + netdev); + return Status::CONNECT_FAILURE; + } + + ifreq address_request{}; + std::snprintf(address_request.ifr_name, sizeof(address_request.ifr_name), "%s", netdev.c_str()); + if (ioctl(control_fd, SIOCGIFADDR, &address_request) != 0) { + const int error = errno; + close(control_fd); + DAQIRI_LOG_ERROR( + "ARP: kernel netdev '{}' has no usable primary IPv4 address: {}. Configure the source " + "address before resolving neighbors", + netdev, strerror(error)); + return Status::CONNECT_FAILURE; + } + const auto* source = reinterpret_cast(&address_request.ifr_addr); + const uint32_t src_network = source->sin_addr.s_addr; + close(control_fd); + + NetlinkSocket sock; + if (!sock.valid()) { + DAQIRI_LOG_ERROR("ARP: failed to open rtnetlink socket: {}", strerror(sock.error())); + return Status::CONNECT_FAILURE; + } + + const uint32_t dst_network = htonl(dst_host); + RouteResult route; + // Supplying the netdev's primary IPv4 source makes Linux apply source-based + // policy routing instead of selecting a gateway from destination + oif alone. + int error = + lookup_route(sock, static_cast(ifindex), src_network, dst_network, deadline, &route); + if (error != 0) { + if (error == -ETIMEDOUT) { + DAQIRI_LOG_ERROR("ARP: IPv4 route lookup on '{}' exceeded the {} ms deadline", netdev, + timeout_ms); + } else { + DAQIRI_LOG_ERROR( + "ARP: no IPv4 route to destination from the primary address on netdev " + "'{}': {}", + netdev, strerror(-error)); + } + return Status::CONNECT_FAILURE; + } + if (route.ifindex != static_cast(ifindex)) { + DAQIRI_LOG_ERROR("ARP: route selected ifindex {} instead of netdev '{}' (ifindex {})", + route.ifindex, netdev, ifindex); + return Status::CONNECT_FAILURE; + } + + NeighborResult neighbor; + error = lookup_neighbor(sock, route.ifindex, route.next_hop_network, deadline, &neighbor); + if (error != 0) { + if (error == -ETIMEDOUT) { + DAQIRI_LOG_ERROR("ARP: neighbor lookup on '{}' exceeded the {} ms deadline", netdev, + timeout_ms); + return Status::NOT_READY; + } + DAQIRI_LOG_ERROR("ARP: failed to query the neighbor table on '{}': {}", netdev, + strerror(-error)); + return Status::CONNECT_FAILURE; + } + if (neighbor.state == NeighborState::USABLE) { + std::memcpy(mac, neighbor.mac.data(), neighbor.mac.size()); + return Status::SUCCESS; + } + + error = trigger_neighbor(sock, route.ifindex, route.next_hop_network, deadline, &neighbor); + if (neighbor.state == NeighborState::USABLE) { + std::memcpy(mac, neighbor.mac.data(), neighbor.mac.size()); + return Status::SUCCESS; + } + if (error != 0) { + if (error == -ETIMEDOUT) { + DAQIRI_LOG_ERROR("ARP: neighbor trigger on '{}' exceeded the {} ms deadline", netdev, + timeout_ms); + return Status::NOT_READY; + } + DAQIRI_LOG_ERROR( + "ARP: failed to trigger neighbor resolution on '{}': {}. Check CAP_NET_ADMIN/root access", + netdev, strerror(-error)); + return Status::CONNECT_FAILURE; + } + + // Query once after the trigger to close the ACK/notification race, then wait + // on RTMGRP_NEIGH for state changes until the caller's deadline. + error = lookup_neighbor(sock, route.ifindex, route.next_hop_network, deadline, &neighbor, true); + if (neighbor.state == NeighborState::USABLE) { + std::memcpy(mac, neighbor.mac.data(), neighbor.mac.size()); + return Status::SUCCESS; + } + if (error != 0) { + if (error != -ETIMEDOUT) { + DAQIRI_LOG_ERROR("ARP: failed to recheck the neighbor table on '{}': {}", netdev, + strerror(-error)); + return Status::CONNECT_FAILURE; + } + } + std::array buffer{}; + while (neighbor.state != NeighborState::USABLE && neighbor.state != NeighborState::FAILED) { + const auto now = std::chrono::steady_clock::now(); + if (now >= deadline) { + break; + } + const auto remaining = std::chrono::duration_cast(deadline - now); + pollfd descriptor{sock.fd(), POLLIN, 0}; + const int wait_ms = static_cast(std::min(remaining.count() + 1, INT_MAX)); + const int ready = poll(&descriptor, 1, wait_ms); + if (ready < 0) { + if (errno == EINTR) { + continue; + } + DAQIRI_LOG_ERROR("ARP: poll failed on '{}': {}", netdev, strerror(errno)); + return Status::CONNECT_FAILURE; + } + if (ready == 0) { + break; + } + const ssize_t size = recv(sock.fd(), buffer.data(), buffer.size(), 0); + if (size < 0) { + if (errno == EINTR) { + continue; + } + DAQIRI_LOG_ERROR("ARP: failed to receive neighbor update on '{}': {}", netdev, + strerror(errno)); + return Status::CONNECT_FAILURE; + } + int bytes = static_cast(size); + for (auto* hdr = reinterpret_cast(buffer.data()); NLMSG_OK(hdr, bytes); + hdr = NLMSG_NEXT(hdr, bytes)) { + if (hdr->nlmsg_seq == 0) { + parse_neighbor(hdr, route.ifindex, route.next_hop_network, &neighbor); + } + } + } + + if (neighbor.state == NeighborState::USABLE) { + std::memcpy(mac, neighbor.mac.data(), neighbor.mac.size()); + return Status::SUCCESS; + } + + char next_hop[INET_ADDRSTRLEN] = {}; + DAQIRI_LOG_ERROR( + "ARP: neighbor {} on '{}' did not resolve within {} ms; ensure the peer is reachable and " + "if this port has DAQIRI RX queues, ensure rx.flow_isolation is true so ARP remains on the " + "kernel path", + ipv4_text(route.next_hop_network, next_hop, sizeof(next_hop)), netdev, timeout_ms); + return Status::NOT_READY; +} + +} // namespace daqiri diff --git a/src/net_neighbor.h b/src/net_neighbor.h new file mode 100644 index 00000000..8adc1d0e --- /dev/null +++ b/src/net_neighbor.h @@ -0,0 +1,33 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. + * All rights reserved. SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include + +#include + +namespace daqiri { + +/// Resolve the IPv4 next-hop MAC selected by Linux for `dst_host` on `netdev`. +/// `dst_host` is in host byte order. Linux routing and neighbour tables remain +/// the source of truth; this helper does not maintain a private cache. +Status resolve_ipv4_neighbor(const std::string& netdev, uint32_t dst_host, char* mac, + uint32_t timeout_ms); + +} // namespace daqiri