From 69950e2d83516b8e3f534e318194f1a41752ba4d Mon Sep 17 00:00:00 2001 From: Juhana Helovuo Date: Wed, 1 Jul 2026 09:51:29 +0300 Subject: [PATCH 1/5] Make interop_matrix test script run in the background. --- scripts/interop_matrix.sh | 31 ++++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) diff --git a/scripts/interop_matrix.sh b/scripts/interop_matrix.sh index 8cdab2fb..22ab5328 100755 --- a/scripts/interop_matrix.sh +++ b/scripts/interop_matrix.sh @@ -20,13 +20,21 @@ # /executables/rustdds-0.12.0_shape_main_linux # # Usage: -# scripts/interop_matrix.sh +# scripts/interop_matrix.sh # detaches into its own session and returns +# +# By default the script re-execs itself under `setsid`, fully detached from the +# launching terminal, so the multi-hour run survives terminal close / launch +# interruption. It prints the background PID and the console log path, then +# returns immediately. Progress is in /MATRIX.log and SUMMARY.csv. # # Environment overrides (defaults in parentheses): -# DDS_RTPS_DIR path to the dds-rtps checkout (/home/juhe/cursor/dds-rtps) -# RUST_EXE RustDDS shape_main, relative to DDS_RTPS_DIR -# (executables/rustdds-0.12.0_shape_main_linux) -# PAIR_TIMEOUT hard cap per pair, seconds (2400 = 40 min) +# DDS_RTPS_DIR path to the dds-rtps checkout (/home/juhe/cursor/dds-rtps) +# RUST_EXE RustDDS shape_main, relative to DDS_RTPS_DIR +# (executables/rustdds-0.12.0_shape_main_linux) +# PAIR_TIMEOUT hard cap per pair, seconds (2400 = 40 min) +# MATRIX_NO_DETACH set to 1 to run in the foreground (no setsid re-exec) +# MATRIX_CONSOLE_LOG console log path for the detached run +# (/tmp/interop_matrix_.log) # # Results are written under /results/matrix_/: # MATRIX.log high-level progress log @@ -36,6 +44,19 @@ set -u +# Survive the launching shell/terminal. Re-exec once in a brand-new session +# (setsid) detached from the controlling terminal, so closing the terminal or +# interrupting the launch command cannot reap the (long) run. Opt out with +# MATRIX_NO_DETACH=1 to run in the foreground. +if [ -z "${MATRIX_DETACHED:-}" ] && [ -z "${MATRIX_NO_DETACH:-}" ]; then + export MATRIX_DETACHED=1 + CONSOLE_LOG="${MATRIX_CONSOLE_LOG:-/tmp/interop_matrix_$(date +%Y%m%d_%H%M%S).log}" + setsid bash "$0" "$@" "$CONSOLE_LOG" 2>&1 & + echo "interop matrix detached into its own session (PID $!)" + echo "console log: $CONSOLE_LOG" + exit 0 +fi + DDS_RTPS_DIR="${DDS_RTPS_DIR:-/home/juhe/cursor/dds-rtps}" RUST="${RUST_EXE:-executables/rustdds-0.12.0_shape_main_linux}" PAIR_TIMEOUT="${PAIR_TIMEOUT:-2400}" # 40 min hard cap per pair From 99bca8bef060e0baaa97bf78ad89e9ae4f66e187 Mon Sep 17 00:00:00 2001 From: Juhana Helovuo Date: Wed, 1 Jul 2026 10:52:06 +0300 Subject: [PATCH 2/5] Add design for new transmit algorithm --- src/rtps/transmit_design.md | 193 ++++++++++++++++++++++++++++++++++++ 1 file changed, 193 insertions(+) create mode 100644 src/rtps/transmit_design.md diff --git a/src/rtps/transmit_design.md b/src/rtps/transmit_design.md new file mode 100644 index 00000000..30488109 --- /dev/null +++ b/src/rtps/transmit_design.md @@ -0,0 +1,193 @@ +# Smarter Transmit: Interface-Aware Locator Selection + +Status: design proposal (no implementation yet). + +This document describes how to replace RustDDS's current "send to everything" +transmit strategy with a precomputed, interface-aware locator-selection +algorithm that avoids unnecessary duplicate sends while preserving reachability +and correctness. + +The relevant code today lives in +[`writer.rs`](writer.rs) (`send_message_to_readers`), +[`rtps_reader_proxy.rs`](rtps_reader_proxy.rs) (per-reader locator lists), +[`../network/udp_sender.rs`](../network/udp_sender.rs), +[`../network/udp_listener.rs`](../network/udp_listener.rs), and +[`../discovery/spdp_participant_data.rs`](../discovery/spdp_participant_data.rs). + +## 1. Purpose and scope + +Replace the self-labelled "stupid transmit algorithm" (TODO comment in +[`writer.rs`](writer.rs) around lines 1203-1205) with a design that: + +- picks one multicast route and one unicast route per remote, instead of + blasting every advertised locator; +- makes multicast interface-aware, so reaching one remote does not require + sending on every local interface; +- precomputes routes on discovery updates, not on every transmit; +- keeps cross-remote de-duplication (one datagram serves all remotes reachable + behind the same interface + address); +- leaves a clean seam for a route-selection heuristic when a remote is + reachable via multiple interfaces; +- reuses the existing multicast-vs-unicast preference logic. + +Scope of this document is design only. No behavior changes are implied by +merging it. + +## 2. Current behavior (baseline) + +- `send_message_to_readers` ([`writer.rs`](writer.rs) ~1197-1268): for each + message, for each `RtpsReaderProxy`, it dynamically chooses multicast-or-unicast + and then sends to *every* locator in that list. `already_sent_to: + BTreeSet` de-duplicates only within a single send call. +- Reader-proxy locators (`unicast_locator_list`, `multicast_locator_list`) come + only from the advertised SEDP/SPDP payload + ([`rtps_reader_proxy.rs`](rtps_reader_proxy.rs) + `from_discovered_reader_data` / `update`; participant `default_*` / + `metatraffic_*` locators in + [`../discovery/spdp_participant_data.rs`](../discovery/spdp_participant_data.rs)). +- Outbound multicast replicates over *all* interface sockets (`multicast_sockets` + in [`../network/udp_sender.rs`](../network/udp_sender.rs) ~140-148). There is + no API to target a single interface. +- The receive path discards origin information: `UDPListener::messages` uses + `socket.recv()` (not `recv_from`) in + [`../network/udp_listener.rs`](../network/udp_listener.rs) ~266; the four + listeners (discovery/user x unicast/multicast) are not per-interface (the + multicast listener joins all interfaces). As a result, the local interface and + the source address on which a remote's discovery arrived are *not available + today*. +- Delivery mode is chosen per call: DATA / HEARTBEAT prefer Multicast; repair and + directed sends use Unicast ([`writer.rs`](writer.rs) ~512, 564, 568, 652, 796, + 997, 1060). + +## 3. Problems + +- Duplicate unicast sends to every advertised address of a remote. +- Multicast fan-out over every local interface even when the remote is on one. +- The route is recomputed on every transmit. +- No knowledge of which interface actually reaches a given remote. + +## 4. Proposed design + +### 4.1 Core concept: a precomputed `SendRoute` per reader proxy + +Introduce a `SendRoute` (resolved destination) computed once per discovery +update and cached on the `RtpsReaderProxy`: + +- `unicast: Option` -- the single chosen unicast destination. +- `multicast: Option<(Locator, InterfaceSelector)>` -- the chosen multicast group + plus the local egress interface to use. +- `alternates: Vec<...>` (optional) -- for the multi-path heuristic (Section 4.5). + +`send_message_to_readers` then only iterates precomputed routes and de-duplicates +by a `RouteKey` (Section 4.4). It performs no list scanning or mode logic beyond +honoring `preferred_mode`. + +### 4.2 Interface-aware multicast + +- Model a local multicast egress as an `InterfaceSelector` (interface IP or OS + interface index). +- `UDPSender` today keeps an unkeyed `Vec` of per-interface multicast sockets. + The design keys them by interface so the sender can emit on exactly one + interface, e.g. `send_to_multicast(buffer, group, interface)`. +- Route resolution assigns each remote the interface on which its discovery was + observed (Section 4.3). When that is unknown, it falls back to "all interfaces" + (preserving current reachability). + +### 4.3 Prerequisite capability: learn the receiving interface / source + +This is the enabling change the design depends on, and it is currently missing. + +- Capture, per received datagram, the source `SocketAddr` and the local receiving + interface. On Linux and most Unix systems this is `recvmsg` + + `IP_PKTINFO` / `IPV6_PKTINFO` (control message `ipi_ifindex` / `ipi_spec_dst`). + A portable fallback is `recv_from` for the source address plus best-effort + interface inference. +- Thread this origin metadata from `UDPListener` into `MessageReceiver` and then + into SPDP handling, and record it per remote participant in the discovery DB: + the set of `(local_interface, remote_source_addr)` observations from that + participant's SPDP. This is a new, dedicated metadata channel and must be added + to the receive path. +- Important distinction: the existing `unicast_reply_locator_list` / + `multicast_reply_locator_list` fields in `MessageReceiverState` + ([`message_receiver.rs`](message_receiver.rs) ~71-77) are **not** related to + this feature. They carry the reply-locator lists announced via the RTPS + `InfoReply` submessage, whose purpose is to tell the receiver a (possibly + different) locator list to reply to -- for example to relay or forward RTPS + submessages on behalf of another participant. They are transport-independent, + submessage-level data and must not be conflated with the observed local + interface / source address introduced here. +- The observed interface is the primary input for choosing the multicast egress + and for ranking unicast candidates. Degrade gracefully: if it is unavailable, + behavior falls back to today's all-interfaces semantics. + +### 4.4 De-duplication across remotes + +- Compute a `RouteKey`: + - unicast: the destination `Locator` (address + port); + - multicast: `(group Locator, InterfaceSelector)`. +- A single `send_message_to_readers` pass collects the `RouteKey`s of all target + readers into a set and emits each datagram once. Multiple remotes sharing a + multicast group on the same interface produce one packet; remotes sharing a + unicast endpoint produce one packet. + +### 4.5 Multiple reachable interfaces (heuristic seam) + +- When discovery for a remote is observed on more than one local interface (both + hosts multi-homed), store all observed routes and select one via a pluggable + `RouteSelector` policy. +- Document a conservative default heuristic, e.g. prefer the interface with the + most recent / most frequent SPDP, prefer non-loopback, prefer a matching subnet + or lowest metric. Leave the trait / function seam for future policies. +- Record the decision on the proxy so transmit stays O(1). + +### 4.6 Reuse of multicast-vs-unicast preference + +- Keep the existing `DeliveryMode` preference and the current match precedence + (multicast when a multicast route exists in multicast mode; otherwise unicast; + repair and directed sends forced to unicast). The only change is *which* + concrete locator / interface each mode resolves to (a single chosen route + rather than the whole list). + +### 4.7 Recompute triggers (not per transmit) + +Recompute `SendRoute` on: + +- reader / participant discovery add or update; +- a new interface observation for a remote; +- local interface-set changes. + +Hook points: `update_reader_proxy` / `matched_reader_update` +([`writer.rs`](writer.rs) ~1286) and the discovery propagation flow in +[`dp_event_loop.rs`](dp_event_loop.rs) (~636-685 and ~800-802). + +### 4.8 Correctness, fallbacks, compatibility + +- Never send to fewer places than needed for reachability: when the interface or + source is unknown, fall back to all-interfaces multicast and (optionally) all + advertised unicast, matching today's behavior. +- Retain loopback filtering (`not_loopback`) and `Locator::is_udp` gating. +- Interop risks to document and handle conservatively: peers only reachable via a + non-observed interface, asymmetric routing, and NAT. + +## 5. Data flow + +```mermaid +flowchart TD + spdp["Remote SPDP/SEDP packet"] --> listener["UDPListener recvmsg (src addr + ifindex)"] + listener --> mr["MessageReceiver / discovery"] + mr --> db["DiscoveryDB: per-remote observed interfaces + advertised locators"] + db --> resolve["Route resolution (RouteSelector heuristic)"] + resolve --> proxy["RtpsReaderProxy.SendRoute (cached)"] + proxy --> tx["send_message_to_readers: dedup by RouteKey"] + tx --> sender["UDPSender: unicast socket / per-interface multicast socket"] +``` + +## 6. Non-goals and open questions + +- No code changes accompany this document. +- Open questions: + - exact `InterfaceSelector` representation (interface IP vs OS index) and + Windows parity for `IP_PKTINFO`; + - whether to also prune redundant unicast when multicast already covers a + remote; + - the specifics of the default `RouteSelector` policy. From f70b25fa03d4d697de1a55f34b406704fde97bef Mon Sep 17 00:00:00 2001 From: Juhana Helovuo Date: Wed, 1 Jul 2026 11:46:20 +0300 Subject: [PATCH 3/5] First iteration of a smarter transmit algorithm --- Cargo.toml | 5 + src/network/udp_listener.rs | 158 +++++++++++-- src/network/udp_sender.rs | 85 ++++++- src/network/util.rs | 61 ++++- src/rtps.rs | 1 + src/rtps/dp_event_loop.rs | 31 ++- src/rtps/message_receiver.rs | 54 ++++- src/rtps/rtps_reader_proxy.rs | 93 +++++++- src/rtps/transmit.rs | 406 ++++++++++++++++++++++++++++++++++ src/rtps/transmit_design.md | 45 +++- src/rtps/writer.rs | 145 ++++++++---- 11 files changed, 1013 insertions(+), 71 deletions(-) create mode 100644 src/rtps/transmit.rs diff --git a/Cargo.toml b/Cargo.toml index 122fd8da..6ead6790 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -89,6 +89,11 @@ pnet_sys = { version = "0.35.0", default-features = false } [target.'cfg(windows)'.dependencies] local-ip-address = "0.6.1" +[target.'cfg(unix)'.dependencies] +# Used to capture the receiving interface of incoming UDP datagrams via +# recvmsg()/IP_PKTINFO for interface-aware transmit-locator selection. +nix = { version = "0.29", features = ["net", "socket", "uio"] } + [dev-dependencies] serde_repr = {version = "0.1" } log = "0.4" diff --git a/src/network/udp_listener.rs b/src/network/udp_listener.rs index 8a7b23f5..0c63f170 100644 --- a/src/network/udp_listener.rs +++ b/src/network/udp_listener.rs @@ -1,4 +1,5 @@ use std::{ + collections::HashMap, io, net::{IpAddr, Ipv4Addr, SocketAddr}, }; @@ -9,13 +10,34 @@ use bytes::{Bytes, BytesMut}; use crate::{ network::util::{ - get_local_multicast_ip_addrs_filtered, get_local_multicast_locators, - get_local_unicast_locators_filtered, + build_ifindex_to_interface_map, get_local_multicast_ip_addrs_filtered, + get_local_multicast_locators, get_local_unicast_locators_filtered, }, + rtps::transmit::InterfaceSelector, serialization::padding_needed_for_alignment_4, structure::locator::Locator, }; +/// Metadata captured about where a received datagram came from and how it +/// reached us. +#[derive(Clone, Copy, Debug)] +pub struct PacketOrigin { + /// Remote source socket address, if it could be determined. + pub source: Option, + /// Local interface the datagram was received on, if it could be determined + /// (requires `IP_PKTINFO`; `None` on platforms/paths where it is unavailable). + pub local_if: Option, +} + +impl PacketOrigin { + /// An origin with no captured metadata (forces the legacy send fallback). + #[allow(dead_code)] // Used by tests and the loopback path; harmless if unused in a given build. + pub const UNKNOWN: Self = Self { + source: None, + local_if: None, + }; +} + const MAX_MESSAGE_SIZE: usize = 64 * 1024; // This is max we can get from UDP. const MESSAGE_BUFFER_ALLOCATION_CHUNK: usize = 256 * 1024; // must be >= MAX_MESSAGE_SIZE static_assertions::const_assert!(MESSAGE_BUFFER_ALLOCATION_CHUNK > MAX_MESSAGE_SIZE); @@ -29,6 +51,9 @@ pub struct UDPListener { receive_buffer: BytesMut, multicast_group: Option, has_multicast_join: bool, + // Cached OS interface-index -> local interface map, used to resolve the + // receiving interface reported by IP_PKTINFO. Built once at construction. + ifindex_map: HashMap, } impl Drop for UDPListener { @@ -77,6 +102,18 @@ impl UDPListener { } } + // Ask the kernel to attach IP_PKTINFO to received datagrams so we can learn + // which local interface each one arrived on. Best-effort: if it fails we + // simply lose interface metadata and fall back to the legacy send path. + #[cfg(unix)] + { + if let Err(e) = + nix::sys::socket::setsockopt(&raw_socket, nix::sys::socket::sockopt::Ipv4PacketInfo, &true) + { + warn!("Could not enable IP_PKTINFO on listener socket: {e}. Interface-aware transmit disabled for this socket."); + } + } + let address = SocketAddr::new(host.parse().map_err(io::Error::other)?, port); if let Err(e) = raw_socket.bind(&SockAddr::from(address)) { @@ -129,6 +166,7 @@ impl UDPListener { receive_buffer: BytesMut::with_capacity(MESSAGE_BUFFER_ALLOCATION_CHUNK), multicast_group: None, has_multicast_join: false, + ifindex_map: build_ifindex_to_interface_map(), }) } @@ -193,6 +231,7 @@ impl UDPListener { receive_buffer: BytesMut::with_capacity(MESSAGE_BUFFER_ALLOCATION_CHUNK), multicast_group: Some(multicast_group), has_multicast_join: joined_multicast, + ifindex_map: build_ifindex_to_interface_map(), }) } @@ -239,8 +278,9 @@ impl UDPListener { panic!("test helper didn't recv message after ten attempts."); } - /// Get all messages waiting in the socket. - pub fn messages(&mut self) -> Vec { + /// Get all messages waiting in the socket, each paired with its + /// [`PacketOrigin`] (source address + receiving interface, when available). + pub fn messages(&mut self) -> Vec<(Bytes, PacketOrigin)> { let mut messages = Vec::with_capacity(4); loop { @@ -263,16 +303,16 @@ impl UDPListener { "ensure_receive_buffer_capacity - {} bytes left", self.receive_buffer.capacity() ); - let nbytes = match self.socket.recv(&mut self.receive_buffer) { - Ok(n) => n, + let (nbytes, origin) = match self.recv_one() { + Ok(Some(received)) => received, + Ok(None) => { + // WouldBlock: nothing (more) to read. + self.receive_buffer.clear(); + return messages; + } Err(e) => { - self.receive_buffer.clear(); // since nothing was received - if e.kind() == io::ErrorKind::WouldBlock { - // This is the normal case. - } else { - warn!("socket recv() error: {e:?}"); - } - // In any case, we stop trying and return. + self.receive_buffer.clear(); + warn!("socket recv() error: {e:?}"); return messages; } }; @@ -293,13 +333,84 @@ impl UDPListener { // Now split away the used portion. let mut message = self.receive_buffer.split_to(self.receive_buffer.len()); message.truncate(nbytes); // discard (hide) padding - messages.push(Bytes::from(message)); // freeze bytes and push + messages.push((Bytes::from(message), origin)); // freeze bytes and push } // loop // unreachable!(); // But why does this cause a warning? (rustc 1.66.0) // Answer: https://github.com/rust-lang/rust/issues/46500 } + /// Receive a single datagram into `self.receive_buffer`, capturing its + /// [`PacketOrigin`]. Returns `Ok(None)` when the socket would block. + #[cfg(unix)] + fn recv_one(&mut self) -> io::Result> { + use std::{io::IoSliceMut, os::unix::io::AsRawFd}; + + use nix::{ + errno::Errno, + sys::socket::{recvmsg, ControlMessageOwned, MsgFlags, SockaddrStorage}, + }; + + let fd = self.socket.as_raw_fd(); + let mut cmsg_space = nix::cmsg_space!(nix::libc::in_pktinfo); + + // Read the datagram and pull out the Copy metadata; the borrow of + // `receive_buffer` (through `iov`) ends when this block ends. + let (nbytes, source, ifindex, spec_dst) = { + let mut iov = [IoSliceMut::new(&mut self.receive_buffer[..MAX_MESSAGE_SIZE])]; + let msg = match recvmsg::( + fd, + &mut iov, + Some(&mut cmsg_space), + MsgFlags::empty(), + ) { + Ok(m) => m, + Err(Errno::EAGAIN) => return Ok(None), + Err(e) => return Err(io::Error::from_raw_os_error(e as i32)), + }; + + let nbytes = msg.bytes; + let source = msg.address.and_then(sockaddr_storage_to_socketaddr); + + let mut ifindex = 0u32; + let mut spec_dst: Option = None; + for cmsg in msg.cmsgs()? { + if let ControlMessageOwned::Ipv4PacketInfo(info) = cmsg { + ifindex = info.ipi_ifindex as u32; + let addr = Ipv4Addr::from(u32::from_be(info.ipi_spec_dst.s_addr)); + if !addr.is_unspecified() { + spec_dst = Some(IpAddr::V4(addr)); + } + } + } + (nbytes, source, ifindex, spec_dst) + }; + + // Prefer the exact local destination address (matches sender interface + // keys directly); otherwise resolve the interface index. + let local_if = spec_dst + .map(InterfaceSelector::Ip) + .or_else(|| self.ifindex_map.get(&ifindex).copied()); + + Ok(Some((nbytes, PacketOrigin { source, local_if }))) + } + + /// Non-Unix fallback: capture the source address only (no interface info). + #[cfg(not(unix))] + fn recv_one(&mut self) -> io::Result> { + match self.socket.recv_from(&mut self.receive_buffer[..MAX_MESSAGE_SIZE]) { + Ok((nbytes, source)) => Ok(Some(( + nbytes, + PacketOrigin { + source: Some(source), + local_if: None, + }, + ))), + Err(e) if e.kind() == io::ErrorKind::WouldBlock => Ok(None), + Err(e) => Err(e), + } + } + #[cfg(test)] // normally done in .drop() pub fn leave_multicast(&self, address: &Ipv4Addr) -> io::Result<()> { if address.is_multicast() { @@ -311,6 +422,25 @@ impl UDPListener { } } +#[cfg(unix)] +fn sockaddr_storage_to_socketaddr( + addr: nix::sys::socket::SockaddrStorage, +) -> Option { + use std::net::{SocketAddrV4, SocketAddrV6}; + if let Some(v4) = addr.as_sockaddr_in() { + Some(SocketAddr::V4(SocketAddrV4::new(v4.ip(), v4.port()))) + } else { + addr.as_sockaddr_in6().map(|v6| { + SocketAddr::V6(SocketAddrV6::new( + v6.ip(), + v6.port(), + v6.flowinfo(), + v6.scope_id(), + )) + }) + } +} + #[cfg(test)] mod tests { // use std::os::unix::io::AsRawFd; diff --git a/src/network/udp_sender.rs b/src/network/udp_sender.rs index 2a8971fe..756a850a 100644 --- a/src/network/udp_sender.rs +++ b/src/network/udp_sender.rs @@ -11,14 +11,20 @@ use socket2::{Domain, Protocol, SockAddr, Socket, Type}; #[cfg(windows)] use local_ip_address::list_afinet_netifas; -use crate::{network::util::get_local_multicast_ip_addrs_filtered, structure::locator::Locator}; +use crate::{ + network::util::get_local_multicast_ip_addrs_filtered, rtps::transmit::InterfaceSelector, + structure::locator::Locator, +}; // We need one multicast sender socket per interface #[derive(Debug)] pub struct UDPSender { unicast_socket: mio_08::net::UdpSocket, - multicast_sockets: Vec, + // One multicast sender socket per local interface, keyed by the interface it + // was bound to (its `InterfaceSelector`). This lets us target a single + // interface instead of sending on all of them. + multicast_sockets: Vec<(InterfaceSelector, mio_08::net::UdpSocket)>, } impl UDPSender { @@ -88,7 +94,10 @@ impl UDPSender { } }; - multicast_sockets.push(mio_08::net::UdpSocket::from_std(mc_socket)); + multicast_sockets.push(( + InterfaceSelector::Ip(multicast_if_ipaddr), + mio_08::net::UdpSocket::from_std(mc_socket), + )); } // end for let sender = Self { @@ -139,7 +148,7 @@ impl UDPSender { } let send = |socket_address: SocketAddr| { if socket_address.ip().is_multicast() { - for socket in &self.multicast_sockets { + for (_iface, socket) in &self.multicast_sockets { self.send_to_udp_socket(buffer, socket, &socket_address); } } else { @@ -162,6 +171,72 @@ impl UDPSender { } } + /// The set of local interfaces on which this sender can emit multicast. + /// Used by route resolution to validate an observed interface is usable. + pub fn multicast_interfaces(&self) -> Vec { + self + .multicast_sockets + .iter() + .map(|(iface, _)| *iface) + .collect() + } + + /// Send a multicast datagram out of a single, specific local interface. + /// + /// Falls back to sending on all multicast interfaces (like + /// [`Self::send_to_locator`]) if the requested interface is not one of ours, + /// so a stale/misresolved interface never silently drops traffic. + pub fn send_to_multicast_locator_via( + &self, + buffer: &[u8], + locator: &Locator, + interface: &InterfaceSelector, + ) { + if buffer.len() > 1500 { + warn!( + "send_to_multicast_locator_via: Message size = {}", + buffer.len() + ); + } + + let socket_address = match locator { + Locator::UdpV4(sa) => SocketAddr::from(*sa), + Locator::UdpV6(sa) => SocketAddr::from(*sa), + Locator::Invalid | Locator::Reserved => { + error!("send_to_multicast_locator_via: Cannot send to {locator:?}"); + return; + } + Locator::Other { kind, .. } => { + trace!("send_to_multicast_locator_via: Unknown LocatorKind: {kind:?}"); + return; + } + }; + + if !socket_address.ip().is_multicast() { + // Not a multicast destination; treat as a plain unicast send. + self.send_to_udp_socket(buffer, &self.unicast_socket, &socket_address); + return; + } + + match self + .multicast_sockets + .iter() + .find(|(iface, _)| iface == interface) + { + Some((_, socket)) => self.send_to_udp_socket(buffer, socket, &socket_address), + None => { + // Requested interface unknown: preserve reachability by falling back to + // all interfaces. + trace!( + "send_to_multicast_locator_via: interface {interface:?} not found, sending on all" + ); + for (_iface, socket) in &self.multicast_sockets { + self.send_to_udp_socket(buffer, socket, &socket_address); + } + } + } + } + #[cfg(test)] pub fn send_to_all(&self, buffer: &[u8], addresses: &[SocketAddr]) { let buf_len = buffer.len(); @@ -189,7 +264,7 @@ impl UDPSender { if address.is_multicast() { let address = SocketAddr::new(IpAddr::V4(address), port); let mut size = 0; - for s in self.multicast_sockets { + for (_iface, s) in self.multicast_sockets { size = s.send_to(buffer, address)?; } Ok(size) diff --git a/src/network/util.rs b/src/network/util.rs index f8f8da8a..ed510f9a 100644 --- a/src/network/util.rs +++ b/src/network/util.rs @@ -1,11 +1,12 @@ use std::{ + collections::HashMap, io, net::{IpAddr, SocketAddr}, }; use log::{error, warn}; -use crate::structure::locator::Locator; +use crate::{rtps::transmit::InterfaceSelector, structure::locator::Locator}; pub fn get_local_multicast_locators(port: u16) -> Vec { let saddr = SocketAddr::new("239.255.0.1".parse().unwrap(), port); @@ -82,6 +83,36 @@ fn get_local_multicast_ip_addrs_inner( .collect() } +/// Builds a mapping from OS interface index to an [`InterfaceSelector`]. +/// +/// Used to resolve the `ipi_ifindex` reported by `IP_PKTINFO`/`IPV6_PKTINFO` +/// into the interface identity our sender uses (its IP). Prefers an IPv4 +/// address so the result matches the keys of the multicast sender sockets. +pub fn build_ifindex_to_interface_map() -> HashMap { + build_ifindex_map_inner(pnet::datalink::interfaces()) +} + +fn build_ifindex_map_inner( + interfaces: Vec, +) -> HashMap { + let mut map = HashMap::new(); + for iface in interfaces { + if iface.index == 0 { + continue; + } + let ip = iface + .ips + .iter() + .map(|n| n.ip()) + .find(IpAddr::is_ipv4) + .or_else(|| iface.ips.first().map(|n| n.ip())); + if let Some(ip) = ip { + map.insert(iface.index, InterfaceSelector::Ip(ip)); + } + } + map +} + #[cfg(test)] mod tests { use std::{ @@ -226,6 +257,34 @@ mod tests { ); } + #[test] + fn ifindex_map_prefers_ipv4_and_skips_index_zero() { + use super::InterfaceSelector; + + let interfaces = vec![ + loopback(), // index 0 -> skipped + interface( + "eth0", + 3, + &[ + IpNetwork::V6( + Ipv6Network::new(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 1), 64).unwrap(), + ), + IpNetwork::V4(Ipv4Network::new(Ipv4Addr::new(10, 0, 0, 7), 24).unwrap()), + ], + &[pnet_sys::IFF_MULTICAST], + ), + ]; + + let map = super::build_ifindex_map_inner(interfaces); + assert!(!map.contains_key(&0), "index 0 must be skipped"); + assert_eq!( + map.get(&3), + Some(&InterfaceSelector::Ip(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 7)))), + "should prefer the IPv4 address" + ); + } + fn loopback() -> NetworkInterface { NetworkInterface { name: "lo".to_string(), diff --git a/src/rtps.rs b/src/rtps.rs index 8c6cf22f..d8906a45 100644 --- a/src/rtps.rs +++ b/src/rtps.rs @@ -8,6 +8,7 @@ pub(crate) mod reader; pub(crate) mod rtps_reader_proxy; pub(crate) mod rtps_writer_proxy; pub(crate) mod timed_event; +pub(crate) mod transmit; pub(crate) mod writer; pub(crate) mod writer_send_buffer; diff --git a/src/rtps/dp_event_loop.rs b/src/rtps/dp_event_loop.rs index ee647d1f..39b5fad1 100644 --- a/src/rtps/dp_event_loop.rs +++ b/src/rtps/dp_event_loop.rs @@ -1,4 +1,5 @@ use std::{ + cell::RefCell, collections::HashMap, net::IpAddr, rc::Rc, @@ -32,6 +33,7 @@ use crate::{ rtps_reader_proxy::RtpsReaderProxy, rtps_writer_proxy::RtpsWriterProxy, timed_event::DpTimerEvent, + transmit::InterfaceObservations, writer::{Writer, WriterIngredients}, }, structure::{ @@ -91,6 +93,10 @@ pub struct DPEventLoop { writers: HashMap, udp_sender: Rc, + // Interface-aware transmit: per-remote observed receive interfaces/addresses, + // shared (intra-thread) with the MessageReceiver that populates it. + interface_observations: Rc>, + // One timer shared by all Readers, Writers and the periodic loop tasks. // Endpoints hold cloned handles to schedule timeouts; the loop owns it, // registers it once and drains it. This replaces one OS thread per timer. @@ -225,6 +231,8 @@ impl DPEventLoop { #[cfg(not(feature = "security"))] let security_plugins_opt = security_plugins_opt.and(None); // make sure it is None an consume value + let interface_observations = Rc::new(RefCell::new(InterfaceObservations::new())); + Self { domain_info, poll, @@ -237,7 +245,9 @@ impl DPEventLoop { acknack_sender, spdp_liveness_sender, security_plugins_opt.clone(), + Rc::clone(&interface_observations), ), + interface_observations, #[cfg(feature = "security")] security_plugins_opt, add_reader_receiver, @@ -329,8 +339,10 @@ impl DPEventLoop { }, UDPListener::messages, ); - for packet in udp_messages { - ev_wrapper.message_receiver.handle_received_packet(&packet); + for (packet, origin) in udp_messages { + ev_wrapper + .message_receiver + .handle_received_packet(&packet, origin); } } ADD_READER_TOKEN | REMOVE_READER_TOKEN => { @@ -691,6 +703,14 @@ impl DPEventLoop { } } // for + // Fresh SPDP traffic from this participant may have updated our interface + // observations; refresh the interface-aware send routes of any writers that + // already have matched readers behind this participant. Access `writers` + // directly (not via &mut self) so it stays disjoint from the `db` borrow. + for writer in self.writers.values_mut() { + writer.recompute_routes_for(participant_guid_prefix); + } + debug!("update_participant - finished for {participant_guid_prefix:?}"); } @@ -711,6 +731,12 @@ impl DPEventLoop { reader.participant_lost(participant_guid_prefix); } + // Forget interface observations for the departed participant. + self + .interface_observations + .borrow_mut() + .remove(participant_guid_prefix); + #[cfg(feature = "security")] if let Some(security_plugins_handle) = &self.security_plugins_opt { security_plugins_handle @@ -962,6 +988,7 @@ impl DPEventLoop { self.udp_sender.clone(), self.shared_timer.clone(), self.participant_status_sender.clone(), + Rc::clone(&self.interface_observations), ); self diff --git a/src/rtps/message_receiver.rs b/src/rtps/message_receiver.rs index a4f9a624..3951f9e0 100644 --- a/src/rtps/message_receiver.rs +++ b/src/rtps/message_receiver.rs @@ -1,4 +1,8 @@ -use std::collections::{btree_map::Entry, BTreeMap}; +use std::{ + cell::RefCell, + collections::{btree_map::Entry, BTreeMap}, + rc::Rc, +}; use enumflags2::BitFlags; use mio_extras::{channel as mio_channel, channel::TrySendError}; @@ -8,7 +12,8 @@ use bytes::Bytes; use crate::{ messages::{protocol_version::ProtocolVersion, submessages::submessages::*, vendor_id::VendorId}, - rtps::{reader::Reader, Message, Submessage, SubmessageBody}, + network::udp_listener::PacketOrigin, + rtps::{reader::Reader, transmit::InterfaceObservations, Message, Submessage, SubmessageBody}, structure::{ entity::RTPSEntity, guid::{EntityId, GuidPrefix, GUID}, @@ -113,6 +118,11 @@ pub(crate) struct MessageReceiver { spdp_liveness_sender: mio_channel::SyncSender, security_plugins: Option, + // Per-remote-participant record of which local interface / source address we + // have observed their traffic on. Shared (intra-thread) with DPEventLoop, + // which consumes it to resolve interface-aware send routes. + interface_observations: Rc>, + own_guid_prefix: GuidPrefix, pub source_version: ProtocolVersion, pub source_vendor_id: VendorId, @@ -138,12 +148,14 @@ impl MessageReceiver { acknack_sender: mio_channel::SyncSender<(GuidPrefix, AckSubmessage)>, spdp_liveness_sender: mio_channel::SyncSender, security_plugins: Option, + interface_observations: Rc>, ) -> Self { Self { available_readers: BTreeMap::new(), acknack_sender, spdp_liveness_sender, security_plugins, + interface_observations, own_guid_prefix: participant_guid_prefix, source_version: ProtocolVersion::THIS_IMPLEMENTATION, @@ -221,7 +233,7 @@ impl MessageReceiver { self.available_readers.get_mut(&reader_id) } - pub fn handle_received_packet(&mut self, msg_bytes: &Bytes) { + pub fn handle_received_packet(&mut self, msg_bytes: &Bytes, origin: PacketOrigin) { // Check for RTPS ping message. At least RTI implementation sends these. // What should we do with them? The spec does not say. if msg_bytes.len() < RTPS_MESSAGE_HEADER_SIZE { @@ -267,10 +279,30 @@ impl MessageReceiver { } }; + // Record how this remote participant's traffic reaches us, so route + // resolution can later narrow sends to the observed interface/address. + self.record_observation(rtps_message.header.guid_prefix, origin); + // And process message self.handle_parsed_message(rtps_message); } + /// Note the interface/source a packet from `source_prefix` arrived on. + /// Ignores our own traffic and packets without any usable origin metadata. + fn record_observation(&self, source_prefix: GuidPrefix, origin: PacketOrigin) { + if source_prefix == GuidPrefix::UNKNOWN || source_prefix == self.own_guid_prefix { + return; + } + let Some(source) = origin.source else { + // Without a source address there is nothing actionable to record. + return; + }; + self + .interface_observations + .borrow_mut() + .record(source_prefix, origin.local_if, source); + } + // This is also called directly from dp_event_loop in case of loopback messages. pub fn handle_parsed_message(&mut self, rtps_message: Message) { self.reset(); @@ -1147,6 +1179,7 @@ mod tests { acknack_sender, spdp_liveness_sender, None, + Rc::new(RefCell::new(InterfaceObservations::new())), ); // Create a reader to process the message @@ -1208,7 +1241,7 @@ mod tests { // Add reader to message reader and process the bytes message message_receiver.add_reader(new_reader); - message_receiver.handle_received_packet(&udp_bits1); + message_receiver.handle_received_packet(&udp_bits1, PacketOrigin::UNKNOWN); // Verify the message reader has recorded the right amount of submessages assert_eq!(message_receiver.submessage_count, 4); @@ -1269,13 +1302,18 @@ mod tests { let (acknack_sender, _acknack_receiver) = mio_channel::sync_channel::<(GuidPrefix, AckSubmessage)>(10); let (spdp_liveness_sender, _spdp_liveness_receiver) = mio_channel::sync_channel(8); - let mut message_receiver = - MessageReceiver::new(guid_new.prefix, acknack_sender, spdp_liveness_sender, None); + let mut message_receiver = MessageReceiver::new( + guid_new.prefix, + acknack_sender, + spdp_liveness_sender, + None, + Rc::new(RefCell::new(InterfaceObservations::new())), + ); - message_receiver.handle_received_packet(&udp_bits1); + message_receiver.handle_received_packet(&udp_bits1, PacketOrigin::UNKNOWN); assert_eq!(message_receiver.submessage_count, 4); - message_receiver.handle_received_packet(&udp_bits2); + message_receiver.handle_received_packet(&udp_bits2, PacketOrigin::UNKNOWN); assert_eq!(message_receiver.submessage_count, 2); } diff --git a/src/rtps/rtps_reader_proxy.rs b/src/rtps/rtps_reader_proxy.rs index 72af9595..35a4cad9 100644 --- a/src/rtps/rtps_reader_proxy.rs +++ b/src/rtps/rtps_reader_proxy.rs @@ -11,7 +11,10 @@ use crate::{ dds::{participant::DomainParticipant, qos::QosPolicies}, discovery::sedp_messages::DiscoveredReaderData, messages::submessages::submessage::AckSubmessage, - rtps::constant::*, + rtps::{ + constant::*, + transmit::{InterfaceObservations, InterfaceSelector, RouteSelector, SendRoute}, + }, structure::{ guid::{EntityId, GUID}, locator::Locator, @@ -61,6 +64,11 @@ pub(crate) struct RtpsReaderProxy { pub repair_mode: bool, qos: QosPolicies, frags_requested: BTreeMap, + + // Interface-aware transmit: the resolved send destination for this reader, + // recomputed when locators or observations change. Defaults to the fallback + // (legacy all-locators) route until resolved. + send_route: SendRoute, } impl RtpsReaderProxy { @@ -78,6 +86,7 @@ impl RtpsReaderProxy { repair_mode: false, qos, frags_requested: BTreeMap::new(), + send_route: SendRoute::default(), } } @@ -187,6 +196,7 @@ impl RtpsReaderProxy { repair_mode: false, qos: reader.qos_policy.clone(), frags_requested: BTreeMap::new(), + send_route: SendRoute::default(), } } @@ -238,9 +248,32 @@ impl RtpsReaderProxy { repair_mode: false, qos: discovered_reader_data.subscription_topic_data.qos(), frags_requested: BTreeMap::new(), + send_route: SendRoute::default(), } } + /// The currently resolved [`SendRoute`] for this reader. + pub fn send_route(&self) -> SendRoute { + self.send_route + } + + /// Recompute this reader's [`SendRoute`] from its advertised locators and the + /// per-participant [`InterfaceObservations`], using `selector`. + pub fn resolve_send_route( + &mut self, + observations: &InterfaceObservations, + local_multicast_ifaces: &[InterfaceSelector], + selector: &dyn RouteSelector, + ) { + let observed = observations.get(self.remote_reader_guid.prefix); + self.send_route = selector.select( + &self.unicast_locator_list, + &self.multicast_locator_list, + observed, + local_multicast_ifaces, + ); + } + pub fn handle_ack_nack( &mut self, ack_submessage: &AckSubmessage, @@ -521,3 +554,61 @@ mod acknack_tests { assert_eq!(rp.all_acked_before, SequenceNumber::from(50)); } } + +#[cfg(test)] +mod route_tests { + use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4}; + + use super::*; + use crate::{ + rtps::transmit::{DefaultRouteSelector, InterfaceObservations, InterfaceSelector}, + structure::guid::GuidPrefix, + }; + + fn udp(ip: [u8; 4], port: u16) -> Locator { + Locator::UdpV4(SocketAddrV4::new(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]), port)) + } + + fn iface(ip: [u8; 4]) -> InterfaceSelector { + InterfaceSelector::Ip(IpAddr::V4(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]))) + } + + fn proxy_with_prefix(prefix: GuidPrefix) -> RtpsReaderProxy { + let guid = GUID::new(prefix, EntityId::UNKNOWN); + RtpsReaderProxy::new(guid, QosPolicies::default(), false) + } + + #[test] + fn resolve_falls_back_without_observation() { + let mut rp = proxy_with_prefix(GuidPrefix::UNKNOWN); + rp.unicast_locator_list = vec![udp([10, 0, 0, 5], 7410)]; + let observations = InterfaceObservations::new(); + rp.resolve_send_route(&observations, &[iface([10, 0, 0, 1])], &DefaultRouteSelector); + assert!(rp.send_route().fallback); + } + + #[test] + fn resolve_narrows_with_observation() { + let prefix = GuidPrefix::new(&[9; 12]); + let mut rp = proxy_with_prefix(prefix); + rp.unicast_locator_list = vec![udp([10, 0, 0, 5], 7410)]; + rp.multicast_locator_list = vec![udp([239, 255, 0, 1], 7401)]; + + let mut observations = InterfaceObservations::new(); + observations.record( + prefix, + Some(iface([10, 0, 0, 1])), + SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 5)), 7410), + ); + + rp.resolve_send_route(&observations, &[iface([10, 0, 0, 1])], &DefaultRouteSelector); + + let route = rp.send_route(); + assert!(!route.fallback); + assert_eq!(route.unicast, Some(udp([10, 0, 0, 5], 7410))); + assert_eq!( + route.multicast, + Some((udp([239, 255, 0, 1], 7401), iface([10, 0, 0, 1]))) + ); + } +} diff --git a/src/rtps/transmit.rs b/src/rtps/transmit.rs new file mode 100644 index 00000000..64ac1dd3 --- /dev/null +++ b/src/rtps/transmit.rs @@ -0,0 +1,406 @@ +//! Interface-aware transmit-locator selection. +//! +//! See `src/rtps/transmit_design.md` for the full design rationale. +//! +//! The goal of these types is to let a writer send exactly one datagram per +//! distinct destination "route" instead of blindly sending to every locator a +//! remote advertised on every local interface. +//! +//! Terminology: +//! - [`InterfaceSelector`] identifies a *local* egress interface. +//! - [`SendRoute`] is the resolved destination for one remote reader: at most +//! one unicast locator and at most one interface-tagged multicast locator. +//! - [`RouteKey`] is the de-duplication key: two readers that resolve to the +//! same `RouteKey` are served by a single datagram. +//! - [`InterfaceObservations`] records, per remote participant, on which local +//! interface(s) we have actually seen its traffic arrive. This is the primary +//! input to route selection. +//! +//! Safety guardrail: when we do not have enough information to narrow a remote's +//! route confidently, [`SendRoute::fallback`] is set and the writer falls back +//! to the legacy "send to every advertised locator on every interface" path, so +//! reachability is never reduced. + +use std::{ + collections::BTreeMap, + net::{IpAddr, SocketAddr}, + time::Instant, +}; + +use crate::structure::{guid::GuidPrefix, locator::Locator}; + +/// Identifies a local network interface to use as the egress for multicast. +/// +/// Modelled as the interface's IP address for now; the enum leaves room for an +/// OS interface index variant should the IP prove insufficient (e.g. multiple +/// interfaces sharing an address, or IPv6 scope handling). +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum InterfaceSelector { + Ip(IpAddr), +} + +/// The resolved send destination for a single remote reader. +/// +/// `fallback == true` means "we could not narrow this confidently; use the +/// legacy all-locators/all-interfaces path". In that case `unicast`/`multicast` +/// should be ignored by the sender. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SendRoute { + pub unicast: Option, + pub multicast: Option<(Locator, InterfaceSelector)>, + pub fallback: bool, +} + +impl SendRoute { + /// A route that instructs the sender to use the legacy behavior. + pub fn fallback() -> Self { + Self { + unicast: None, + multicast: None, + fallback: true, + } + } +} + +impl Default for SendRoute { + fn default() -> Self { + // Until a route is resolved, behave exactly like today. + Self::fallback() + } +} + +/// De-duplication key for a concrete outbound datagram destination. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum RouteKey { + Unicast(Locator), + Multicast(Locator, InterfaceSelector), +} + +/// One local interface's observation record for a remote participant. +#[derive(Clone, Debug)] +pub struct InterfaceObservation { + pub last_seen: Instant, + pub count: u64, + pub source: Option, +} + +/// What we have observed about how to reach a single remote participant. +#[derive(Clone, Debug, Default)] +pub struct ObservedRoutes { + /// Local interfaces on which we have seen this participant's traffic. + by_iface: BTreeMap, + /// Most recent source socket address seen for this participant, regardless of + /// whether the receiving interface could be determined. + last_source: Option, +} + +impl ObservedRoutes { + fn record(&mut self, iface: Option, source: SocketAddr) { + self.last_source = Some(source); + if let Some(iface) = iface { + let now = Instant::now(); + self + .by_iface + .entry(iface) + .and_modify(|o| { + o.last_seen = now; + o.count = o.count.saturating_add(1); + o.source = Some(source); + }) + .or_insert(InterfaceObservation { + last_seen: now, + count: 1, + source: Some(source), + }); + } + } + + /// The most recently observed source address, if any. + pub fn last_source(&self) -> Option { + self.last_source + } + + /// The "best" local interface to reach this participant: the most recently + /// observed one, breaking ties by observation count. `None` if we have never + /// determined a receiving interface for it. + pub fn best_interface(&self) -> Option { + self + .by_iface + .iter() + .max_by(|(_, a), (_, b)| { + a.last_seen + .cmp(&b.last_seen) + .then_with(|| a.count.cmp(&b.count)) + }) + .map(|(iface, _)| *iface) + } + + /// Number of distinct local interfaces this participant has been seen on. + #[cfg(test)] + pub fn interface_count(&self) -> usize { + self.by_iface.len() + } +} + +/// Per-remote-participant record of observed receive interfaces / source +/// addresses. Populated by the message receiver, consumed by route resolution. +#[derive(Debug, Default)] +pub struct InterfaceObservations { + by_participant: BTreeMap, +} + +impl InterfaceObservations { + pub fn new() -> Self { + Self::default() + } + + /// Record that a packet from `prefix` arrived from `source`, on local + /// interface `iface` (if it could be determined). + pub fn record(&mut self, prefix: GuidPrefix, iface: Option, source: SocketAddr) { + self + .by_participant + .entry(prefix) + .or_default() + .record(iface, source); + } + + pub fn get(&self, prefix: GuidPrefix) -> Option<&ObservedRoutes> { + self.by_participant.get(&prefix) + } + + pub fn remove(&mut self, prefix: GuidPrefix) { + self.by_participant.remove(&prefix); + } +} + +/// Strategy for turning advertised locators + observations into a [`SendRoute`]. +/// +/// Kept as a trait so the heuristic can evolve (or be swapped) without touching +/// the transmit path. +pub trait RouteSelector { + fn select( + &self, + advertised_unicast: &[Locator], + advertised_multicast: &[Locator], + observed: Option<&ObservedRoutes>, + local_multicast_ifaces: &[InterfaceSelector], + ) -> SendRoute; +} + +/// Conservative default policy. +/// +/// - Without any observation for the remote, returns [`SendRoute::fallback`]. +/// - With an observation, chooses the observed interface for multicast (only if +/// it is one of our local multicast interfaces) and the advertised unicast +/// locator that matches the observed source address. +/// - Whenever narrowing would risk dropping reachability (e.g. a multicast +/// locator is advertised but its interface cannot be determined, or several +/// unicast candidates cannot be disambiguated), it returns the fallback route +/// rather than guessing. +#[derive(Clone, Copy, Debug, Default)] +pub struct DefaultRouteSelector; + +fn first_reachable_udp(locators: &[Locator]) -> Option { + locators + .iter() + .copied() + .find(|l| l.is_udp() && !l.is_loopback()) +} + +impl RouteSelector for DefaultRouteSelector { + fn select( + &self, + advertised_unicast: &[Locator], + advertised_multicast: &[Locator], + observed: Option<&ObservedRoutes>, + local_multicast_ifaces: &[InterfaceSelector], + ) -> SendRoute { + // No origin knowledge -> cannot narrow safely. + let Some(obs) = observed else { + return SendRoute::fallback(); + }; + + let mc_advertised = first_reachable_udp(advertised_multicast); + + // Pick the egress interface only if it is genuinely one of ours. + let chosen_iface = obs + .best_interface() + .filter(|iface| local_multicast_ifaces.contains(iface)); + + let multicast = match (mc_advertised, chosen_iface) { + (Some(mc), Some(iface)) => Some((mc, iface)), + _ => None, + }; + + // If the remote advertises multicast but we cannot bind it to a local + // interface, do not silently drop it: fall back to the legacy path. + if mc_advertised.is_some() && multicast.is_none() { + return SendRoute::fallback(); + } + + let unicast = select_unicast(advertised_unicast, obs); + + // Nothing we can confidently send to -> fallback (also covers the case of + // several ambiguous unicast candidates and no multicast). + if unicast.is_none() && multicast.is_none() { + return SendRoute::fallback(); + } + + SendRoute { + unicast, + multicast, + fallback: false, + } + } +} + +/// Choose a single unicast locator, or `None` if the choice is ambiguous. +fn select_unicast(advertised_unicast: &[Locator], obs: &ObservedRoutes) -> Option { + let candidates: Vec = advertised_unicast + .iter() + .copied() + .filter(|l| l.is_udp() && !l.is_loopback()) + .collect(); + + match candidates.len() { + 0 => None, + 1 => Some(candidates[0]), + _ => { + // Multiple advertised addresses: only pick one if the observed source + // address disambiguates it. Otherwise stay ambiguous (caller falls back). + let source_ip = obs.last_source().map(|sa| sa.ip()); + source_ip.and_then(|ip| { + candidates + .iter() + .copied() + .find(|l| SocketAddr::from(*l).ip() == ip) + }) + } + } +} + +#[cfg(test)] +mod tests { + use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; + + use super::*; + use crate::structure::guid::GuidPrefix; + + fn udp(ip: [u8; 4], port: u16) -> Locator { + Locator::UdpV4(SocketAddrV4::new(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]), port)) + } + + fn sockaddr(ip: [u8; 4], port: u16) -> SocketAddr { + SocketAddr::new(IpAddr::V4(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3])), port) + } + + fn iface(ip: [u8; 4]) -> InterfaceSelector { + InterfaceSelector::Ip(IpAddr::V4(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]))) + } + + #[test] + fn no_observation_is_fallback() { + let sel = DefaultRouteSelector; + let route = sel.select(&[udp([10, 0, 0, 5], 7410)], &[], None, &[]); + assert!(route.fallback); + } + + #[test] + fn single_unicast_with_observation_narrows() { + let sel = DefaultRouteSelector; + let mut obs = ObservedRoutes::default(); + obs.record(Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410)); + let route = sel.select( + &[udp([10, 0, 0, 5], 7410)], + &[], + Some(&obs), + &[iface([10, 0, 0, 1])], + ); + assert!(!route.fallback); + assert_eq!(route.unicast, Some(udp([10, 0, 0, 5], 7410))); + assert_eq!(route.multicast, None); + } + + #[test] + fn multicast_tagged_with_observed_interface() { + let sel = DefaultRouteSelector; + let mut obs = ObservedRoutes::default(); + obs.record(Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410)); + let mc = udp([239, 255, 0, 1], 7401); + let route = sel.select( + &[udp([10, 0, 0, 5], 7410)], + &[mc], + Some(&obs), + &[iface([10, 0, 0, 1])], + ); + assert_eq!(route.multicast, Some((mc, iface([10, 0, 0, 1])))); + assert!(!route.fallback); + } + + #[test] + fn advertised_multicast_but_unknown_interface_falls_back() { + let sel = DefaultRouteSelector; + let mut obs = ObservedRoutes::default(); + // Observation without a resolvable local interface (unicast only source). + obs.record(None, sockaddr([10, 0, 0, 5], 7410)); + let mc = udp([239, 255, 0, 1], 7401); + let route = sel.select(&[udp([10, 0, 0, 5], 7410)], &[mc], Some(&obs), &[iface([10, 0, 0, 1])]); + assert!(route.fallback); + } + + #[test] + fn ambiguous_multi_unicast_without_source_match_falls_back() { + let sel = DefaultRouteSelector; + let mut obs = ObservedRoutes::default(); + obs.record(None, sockaddr([172, 16, 0, 9], 7410)); + // Two advertised addresses, neither matching the observed source IP. + let route = sel.select( + &[udp([10, 0, 0, 5], 7410), udp([192, 168, 1, 5], 7410)], + &[], + Some(&obs), + &[], + ); + assert!(route.fallback); + } + + #[test] + fn multi_unicast_disambiguated_by_source() { + let sel = DefaultRouteSelector; + let mut obs = ObservedRoutes::default(); + obs.record(Some(iface([10, 0, 0, 1])), sockaddr([192, 168, 1, 5], 7410)); + let route = sel.select( + &[udp([10, 0, 0, 5], 7410), udp([192, 168, 1, 5], 7410)], + &[], + Some(&obs), + &[iface([10, 0, 0, 1])], + ); + assert_eq!(route.unicast, Some(udp([192, 168, 1, 5], 7410))); + assert!(!route.fallback); + } + + #[test] + fn route_key_dedup() { + use std::collections::BTreeSet; + let mut set = BTreeSet::new(); + let k1 = RouteKey::Multicast(udp([239, 255, 0, 1], 7401), iface([10, 0, 0, 1])); + let k2 = RouteKey::Multicast(udp([239, 255, 0, 1], 7401), iface([10, 0, 0, 1])); + let k3 = RouteKey::Unicast(udp([10, 0, 0, 5], 7410)); + assert!(set.insert(k1)); + assert!(!set.insert(k2)); // duplicate + assert!(set.insert(k3)); + assert_eq!(set.len(), 2); + } + + #[test] + fn observations_track_best_interface() { + let mut obs = InterfaceObservations::new(); + let p = GuidPrefix::UNKNOWN; + obs.record(p, Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410)); + obs.record(p, Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410)); + obs.record(p, Some(iface([192, 168, 1, 1])), sockaddr([192, 168, 1, 5], 7410)); + let recorded = obs.get(p).unwrap(); + assert_eq!(recorded.interface_count(), 2); + // Most recent wins (the 192.168 one recorded last). + assert_eq!(recorded.best_interface(), Some(iface([192, 168, 1, 1]))); + } +} diff --git a/src/rtps/transmit_design.md b/src/rtps/transmit_design.md index 30488109..07a16dda 100644 --- a/src/rtps/transmit_design.md +++ b/src/rtps/transmit_design.md @@ -184,10 +184,53 @@ flowchart TD ## 6. Non-goals and open questions -- No code changes accompany this document. - Open questions: - exact `InterfaceSelector` representation (interface IP vs OS index) and Windows parity for `IP_PKTINFO`; - whether to also prune redundant unicast when multicast already covers a remote; - the specifics of the default `RouteSelector` policy. + +## 7. Implementation notes (as built) + +- `InterfaceSelector` is the interface IP (`InterfaceSelector::Ip(IpAddr)`), + matching the address the multicast sender sockets are bound to. The enum keeps + room for an OS-index variant. +- Receive origin is captured in [`udp_listener.rs`](../network/udp_listener.rs) + via `recvmsg` + `IP_PKTINFO` (Unix; `nix` crate). For IPv4 the local interface + is taken directly from `ipi_spec_dst`; `ipi_ifindex` is resolved through a + cached index→interface map as a fallback. On non-Unix platforms only the + source address is captured (`local_if = None`), which keeps the fallback path. +- Observations live in `InterfaceObservations`, shared intra-thread via + `Rc>` between `MessageReceiver` (writer) and `DPEventLoop` + (consumer). Cleared on `remote_participant_lost`. +- Routes are resolved in `matched_reader_update` (on discovery add/update) and + refreshed in `Writer::recompute_routes_for`, invoked from `update_participant` + when fresh SPDP traffic may have changed observations. +- The default policy is deliberately conservative: unknown/ambiguous → fallback + (legacy all-locators/all-interfaces path), so the feature is a strict + optimization for peers whose receive interface we have positively observed. + +## 8. Manual multi-homed validation + +CI runs on single-interface hosts, which exercise the fallback and the +narrowed-but-single-interface paths, but cannot prove interface selection across +multiple interfaces. To validate that manually: + +1. Set up two hosts, each connected to two separate subnets (e.g. host A on + `10.0.0.0/24` and `192.168.50.0/24`; host B likewise), so each participant is + reachable via two interfaces. +2. Optionally emulate this on one machine with two network namespaces or two + `veth`/dummy interfaces on different subnets, both multicast-capable, and run + a publisher in one namespace and a subscriber in the other. +3. Run a reliable publisher on A and subscriber on B: + - `shape_main -P -t Square -c BLUE -r` on A, + - `shape_main -S -t Square -r` on B. +4. Capture traffic on each interface (`tcpdump -ni udp`). Expected: after + discovery converges, DATA/HEARTBEAT for a given remote is emitted on a single + interface (the one its SPDP/SEDP arrived on), not duplicated across both. +5. Bring the observed interface down mid-run; the next discovery refresh should + re-resolve the route to the surviving interface (or fall back to all + interfaces until a new observation is made), and delivery should continue. +6. Sanity: with `RUST_LOG=trace`, "Already sent to ..." trace lines confirm + `RouteKey` de-duplication across readers sharing a destination. diff --git a/src/rtps/writer.rs b/src/rtps/writer.rs index 0889cbdb..2a665a91 100644 --- a/src/rtps/writer.rs +++ b/src/rtps/writer.rs @@ -1,4 +1,5 @@ use std::{ + cell::RefCell, cmp::{max, min}, collections::{BTreeMap, BTreeSet}, ops::Bound::Included, @@ -33,6 +34,7 @@ use crate::{ }, rtps_reader_proxy::RtpsReaderProxy, timed_event::DpTimerEvent, + transmit::{DefaultRouteSelector, InterfaceObservations, RouteKey}, writer_send_buffer::WriterSendBuffer, Message, MessageBuilder, }, @@ -158,6 +160,11 @@ pub(crate) struct Writer { // Sending mechanism udp_sender: Rc, + // Interface-aware transmit: per-remote observed receive interfaces/addresses, + // shared (intra-thread) with the MessageReceiver that records them. Consulted + // when (re)resolving each reader proxy's SendRoute. + interface_observations: Rc>, + // By default, this writer is a StatefulWriter (see RTPS spec section 8.4.9) // If like_stateless is true, then the writer mimics the behavior of a Best-Effort // StatelessWriter. This behavior is needed only for a single built-in discovery topic of @@ -202,6 +209,7 @@ impl Writer { udp_sender: Rc, timed_event_timer: SharedTimer, participant_status_sender: StatusChannelSender, + interface_observations: Rc>, ) -> Self { // If writer should behave statelessly, only BestEffort QoS is currently // supported @@ -278,6 +286,7 @@ impl Writer { matched_readers_count_total: 0, requested_incompatible_qos_count: 0, udp_sender, + interface_observations, my_topic_name: i.topic_name, send_buffer: i.send_buffer, last_sent: SequenceNumber::zero(), @@ -1200,13 +1209,12 @@ impl Writer { message: Message, readers: &mut dyn Iterator, ) { - // TODO: This is a stupid transmit algorithm. We should compute a preferred - // unicast and multicast locators for each reader only on every reader update, - // and not find it dynamically on every message. - - // TODO: In addition to Locators found in Readers, we should observe - // the Locators given in MEssageReceiverState, i.e. if there was an - // applicable InfoReply submessage, and we are sending a reply. + // Interface-aware transmit (see src/rtps/transmit_design.md): each reader + // carries a pre-resolved `SendRoute`. When the route is known we emit a + // single datagram per distinct destination (`RouteKey`), targeting one + // interface for multicast. When the route is unknown/ambiguous we fall back + // to the legacy path (send to every advertised locator on every interface) + // so reachability is preserved. let readers = readers.collect::>(); // clone iterator @@ -1218,49 +1226,80 @@ impl Writer { match encoded { Ok(message) => { let buffer = message.write_to_vec_with_ctx(self.endianness).unwrap(); - let mut already_sent_to = BTreeSet::new(); - macro_rules! send_unless_sent_and_mark { + // De-duplication of narrowed (interface-aware) sends across readers. + let mut sent_routes: BTreeSet = BTreeSet::new(); + // De-duplication of legacy (all-interface) sends across readers. + let mut sent_legacy: BTreeSet = BTreeSet::new(); + + macro_rules! emit_multicast { + ($mc:expr, $iface:expr) => { + if sent_routes.insert(RouteKey::Multicast($mc, $iface)) { + self + .udp_sender + .send_to_multicast_locator_via(&buffer, &$mc, &$iface); + } else { + trace!("Already sent to multicast {:?} via {:?}", $mc, $iface); + } + }; + } + macro_rules! emit_unicast { + ($uc:expr) => { + if sent_routes.insert(RouteKey::Unicast($uc)) { + self.udp_sender.send_to_locator(&buffer, &$uc); + } else { + trace!("Already sent to unicast {:?}", $uc); + } + }; + } + macro_rules! send_legacy { ($locs:expr) => { for loc in $locs.iter() { - if already_sent_to.contains(loc) { - trace!("Already sent to {:?}", loc); - } else { + if sent_legacy.insert(*loc) { self.udp_sender.send_to_locator(&buffer, loc); - already_sent_to.insert(loc.clone()); + } else { + trace!("Already sent to {:?}", loc); } } }; } for reader in readers { - match ( - preferred_mode, - reader - .unicast_locator_list - .iter() - .find(|l| Locator::is_udp(l)), - reader - .multicast_locator_list - .iter() - .find(|l| Locator::is_udp(l)), - ) { - (DeliveryMode::Multicast, _, Some(_mc_locator)) => { - send_unless_sent_and_mark!(reader.multicast_locator_list); - } - (DeliveryMode::Unicast, Some(_uc_locator), _) => { - send_unless_sent_and_mark!(reader.unicast_locator_list) + let route = reader.send_route(); + + if route.fallback { + // Unknown/ambiguous route: preserve reachability using the legacy + // all-locators/all-interfaces path with the original precedence. + match ( + preferred_mode, + reader + .unicast_locator_list + .iter() + .find(|l| Locator::is_udp(l)), + reader + .multicast_locator_list + .iter() + .find(|l| Locator::is_udp(l)), + ) { + (DeliveryMode::Multicast, _, Some(_)) => send_legacy!(reader.multicast_locator_list), + (DeliveryMode::Unicast, Some(_), _) => send_legacy!(reader.unicast_locator_list), + (_, _, Some(_)) => send_legacy!(reader.multicast_locator_list), + (_, Some(_), _) => send_legacy!(reader.unicast_locator_list), + (_, None, None) => warn!("send_message_to_readers: No locators for {reader:?}"), } - (_delivery_mode, _, Some(_mc_locator)) => { - send_unless_sent_and_mark!(reader.multicast_locator_list); - } - (_delivery_mode, Some(_uc_locator), _) => { - send_unless_sent_and_mark!(reader.unicast_locator_list) - } - (_delivery_mode, None, None) => { - warn!("send_message_to_readers: No locators for {reader:?}"); + continue; + } + + // Narrowed route: reuse the multicast/unicast preference precedence. + match (preferred_mode, route.multicast, route.unicast) { + (DeliveryMode::Multicast, Some((mc, iface)), _) => emit_multicast!(mc, iface), + (DeliveryMode::Unicast, _, Some(uc)) => emit_unicast!(uc), + (_, _, Some(uc)) => emit_unicast!(uc), + (_, Some((mc, iface)), _) => emit_multicast!(mc, iface), + (_, None, None) => { + warn!("send_message_to_readers: resolved route has no destination for {reader:?}"); } - } // match + } } } Err(e) => error!("Failed to send message to readers. Encoding failed: {e:?}"), @@ -1357,10 +1396,20 @@ impl Writer { fn matched_reader_update(&mut self, updated_reader_proxy: &RtpsReaderProxy) -> bool { let mut is_new = false; let is_volatile = self.qos().is_volatile(); // Get this in advance to work with the borrow checker + // Capture the interface set once; resolution consults current observations. + let multicast_ifaces = self.udp_sender.multicast_interfaces(); self .readers .entry(updated_reader_proxy.remote_reader_guid) - .and_modify(|rp| rp.update(updated_reader_proxy, &self.my_topic_name)) + .and_modify(|rp| { + rp.update(updated_reader_proxy, &self.my_topic_name); + // Locators may have changed; refresh the interface-aware send route. + rp.resolve_send_route( + &self.interface_observations.borrow(), + &multicast_ifaces, + &DefaultRouteSelector, + ); + }) .or_insert_with(|| { is_new = true; let mut new_proxy = updated_reader_proxy.clone(); @@ -1370,11 +1419,29 @@ impl Writer { // for all existing sequence numbers new_proxy.set_pending_gap_up_to(self.send_buffer.last_change_sequence_number()); } + new_proxy.resolve_send_route( + &self.interface_observations.borrow(), + &multicast_ifaces, + &DefaultRouteSelector, + ); new_proxy }); is_new } + /// Refresh the [`SendRoute`](crate::rtps::transmit::SendRoute) of every + /// matched reader belonging to `prefix`. Called when fresh interface + /// observations for that participant may have arrived (e.g. periodic SPDP). + pub fn recompute_routes_for(&mut self, prefix: GuidPrefix) { + let multicast_ifaces = self.udp_sender.multicast_interfaces(); + let observations = self.interface_observations.borrow(); + for rp in self.readers.values_mut() { + if rp.remote_reader_guid.prefix == prefix { + rp.resolve_send_route(&observations, &multicast_ifaces, &DefaultRouteSelector); + } + } + } + fn matched_reader_remove(&mut self, guid: GUID) -> Option { let removed = self.readers.remove(&guid); if let Some(ref removed_reader) = removed { From a85ec556ff5a1a1e1d48e0f5d6147b9d0cf35347 Mon Sep 17 00:00:00 2001 From: Juhana Helovuo Date: Wed, 1 Jul 2026 13:46:10 +0300 Subject: [PATCH 4/5] Add hystereisis to the route selection algorithm to prevent flapping. --- src/rtps/transmit.rs | 148 ++++++++++++++++++++++++++++++++++++++----- 1 file changed, 131 insertions(+), 17 deletions(-) diff --git a/src/rtps/transmit.rs b/src/rtps/transmit.rs index 64ac1dd3..48fa4b4d 100644 --- a/src/rtps/transmit.rs +++ b/src/rtps/transmit.rs @@ -24,11 +24,18 @@ use std::{ collections::BTreeMap, net::{IpAddr, SocketAddr}, - time::Instant, + time::{Duration, Instant}, }; use crate::structure::{guid::GuidPrefix, locator::Locator}; +/// Hysteresis margin for switching a remote participant's chosen multicast +/// egress interface. Once an interface is chosen it stays chosen until we have +/// not heard from it for at least this long *and* the participant's traffic is +/// arriving on a different interface. This keeps the route stable across +/// occasional stray packets on a secondary interface. +pub(crate) const STICKY_SWITCH_MARGIN: Duration = Duration::from_secs(30); + /// Identifies a local network interface to use as the egress for multicast. /// /// Modelled as the interface's IP address for now; the enum leaves room for an @@ -80,6 +87,10 @@ pub enum RouteKey { #[derive(Clone, Debug)] pub struct InterfaceObservation { pub last_seen: Instant, + /// Number of packets observed on this interface. Retained for diagnostics and + /// possible future selection policies; the current sticky heuristic decides + /// switches purely on `last_seen`. + #[allow(dead_code)] pub count: u64, pub source: Option, } @@ -92,13 +103,28 @@ pub struct ObservedRoutes { /// Most recent source socket address seen for this participant, regardless of /// whether the receiving interface could be determined. last_source: Option, + /// The interface currently chosen for multicast egress to this participant. + /// Updated with hysteresis (see [`STICKY_SWITCH_MARGIN`]) so it does not flip + /// on a single stray packet arriving on another interface. + current_interface: Option, } impl ObservedRoutes { fn record(&mut self, iface: Option, source: SocketAddr) { + self.record_at(iface, source, Instant::now(), STICKY_SWITCH_MARGIN); + } + + /// Testable core of [`Self::record`]: `now` and `margin` are injected so the + /// hysteresis can be exercised deterministically. + fn record_at( + &mut self, + iface: Option, + source: SocketAddr, + now: Instant, + margin: Duration, + ) { self.last_source = Some(source); if let Some(iface) = iface { - let now = Instant::now(); self .by_iface .entry(iface) @@ -112,6 +138,25 @@ impl ObservedRoutes { count: 1, source: Some(source), }); + + // Sticky choice: a switch is only ever triggered by receiving on a + // non-current interface (the challenger, whose `last_seen` is `now`). + // Displace the current interface only if we have not heard from it for at + // least `margin`; otherwise the current interface stays chosen. A current + // interface that keeps receiving traffic is thus never displaced. + match self.current_interface { + None => self.current_interface = Some(iface), + Some(cur) if cur == iface => { /* refreshed the current choice; keep */ } + Some(cur) => { + let displace = self + .by_iface + .get(&cur) + .is_none_or(|o| now.saturating_duration_since(o.last_seen) >= margin); + if displace { + self.current_interface = Some(iface); + } + } + } } } @@ -120,19 +165,15 @@ impl ObservedRoutes { self.last_source } - /// The "best" local interface to reach this participant: the most recently - /// observed one, breaking ties by observation count. `None` if we have never - /// determined a receiving interface for it. + /// The local interface currently chosen to reach this participant. + /// + /// This is sticky: it stays on the previously chosen interface until the + /// participant's traffic has been arriving on a different interface and the + /// old one has been silent for at least [`STICKY_SWITCH_MARGIN`] (see + /// [`Self::record_at`]). `None` if we have never determined a receiving + /// interface for this participant. pub fn best_interface(&self) -> Option { - self - .by_iface - .iter() - .max_by(|(_, a), (_, b)| { - a.last_seen - .cmp(&b.last_seen) - .then_with(|| a.count.cmp(&b.count)) - }) - .map(|(iface, _)| *iface) + self.current_interface } /// Number of distinct local interfaces this participant has been seen on. @@ -392,7 +433,9 @@ mod tests { } #[test] - fn observations_track_best_interface() { + fn observations_sticky_within_margin() { + // Recording A, A, then B microseconds apart (default 30s margin) must NOT + // flip the chosen interface: A was seen well within the margin. let mut obs = InterfaceObservations::new(); let p = GuidPrefix::UNKNOWN; obs.record(p, Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410)); @@ -400,7 +443,78 @@ mod tests { obs.record(p, Some(iface([192, 168, 1, 1])), sockaddr([192, 168, 1, 5], 7410)); let recorded = obs.get(p).unwrap(); assert_eq!(recorded.interface_count(), 2); - // Most recent wins (the 192.168 one recorded last). - assert_eq!(recorded.best_interface(), Some(iface([192, 168, 1, 1]))); + // Sticky: the first-chosen interface stays. + assert_eq!(recorded.best_interface(), Some(iface([10, 0, 0, 1]))); + } + + #[test] + fn sticky_first_observation_is_chosen() { + let mut obs = ObservedRoutes::default(); + let t0 = Instant::now(); + obs.record_at( + Some(iface([10, 0, 0, 1])), + sockaddr([10, 0, 0, 5], 7410), + t0, + Duration::from_secs(30), + ); + assert_eq!(obs.best_interface(), Some(iface([10, 0, 0, 1]))); + } + + #[test] + fn sticky_does_not_switch_within_margin() { + let mut obs = ObservedRoutes::default(); + let margin = Duration::from_secs(30); + let t0 = Instant::now(); + obs.record_at(Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410), t0, margin); + // Challenger arrives before the current interface has been silent for the + // full margin -> no switch. + obs.record_at( + Some(iface([192, 168, 1, 1])), + sockaddr([192, 168, 1, 5], 7410), + t0 + Duration::from_secs(29), + margin, + ); + assert_eq!(obs.best_interface(), Some(iface([10, 0, 0, 1]))); + } + + #[test] + fn sticky_switches_past_margin() { + let mut obs = ObservedRoutes::default(); + let margin = Duration::from_secs(30); + let t0 = Instant::now(); + obs.record_at(Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410), t0, margin); + // Current interface has been silent for >= margin when the challenger is + // heard -> switch. + obs.record_at( + Some(iface([192, 168, 1, 1])), + sockaddr([192, 168, 1, 5], 7410), + t0 + Duration::from_secs(30), + margin, + ); + assert_eq!(obs.best_interface(), Some(iface([192, 168, 1, 1]))); + } + + #[test] + fn sticky_busy_current_never_displaced() { + let mut obs = ObservedRoutes::default(); + let margin = Duration::from_secs(30); + let t0 = Instant::now(); + let current = iface([10, 0, 0, 1]); + let challenger = iface([192, 168, 1, 1]); + obs.record_at(Some(current), sockaddr([10, 0, 0, 5], 7410), t0, margin); + // Interleave: the current interface keeps receiving every 10s, while an + // intermittent challenger also shows up. The current interface is never + // silent for a full margin, so it is never displaced. + for k in 1..=6 { + let t = t0 + Duration::from_secs(10 * k); + obs.record_at(Some(challenger), sockaddr([192, 168, 1, 5], 7410), t, margin); + obs.record_at( + Some(current), + sockaddr([10, 0, 0, 5], 7410), + t + Duration::from_secs(1), + margin, + ); + } + assert_eq!(obs.best_interface(), Some(current)); } } From e2281350db4a177bbb8315190d33c37d624e54df Mon Sep 17 00:00:00 2001 From: Juhana Helovuo Date: Wed, 1 Jul 2026 14:29:49 +0300 Subject: [PATCH 5/5] cargo fmt --- src/network/udp_listener.rs | 45 +++++++++++++++------------ src/network/udp_sender.rs | 4 +-- src/network/util.rs | 8 ++--- src/rtps/rtps_reader_proxy.rs | 17 +++++++++-- src/rtps/transmit.rs | 57 +++++++++++++++++++++++++++-------- src/rtps/writer.rs | 2 +- 6 files changed, 90 insertions(+), 43 deletions(-) diff --git a/src/network/udp_listener.rs b/src/network/udp_listener.rs index 0c63f170..1491b5c4 100644 --- a/src/network/udp_listener.rs +++ b/src/network/udp_listener.rs @@ -25,7 +25,8 @@ pub struct PacketOrigin { /// Remote source socket address, if it could be determined. pub source: Option, /// Local interface the datagram was received on, if it could be determined - /// (requires `IP_PKTINFO`; `None` on platforms/paths where it is unavailable). + /// (requires `IP_PKTINFO`; `None` on platforms/paths where it is + /// unavailable). pub local_if: Option, } @@ -107,10 +108,15 @@ impl UDPListener { // simply lose interface metadata and fall back to the legacy send path. #[cfg(unix)] { - if let Err(e) = - nix::sys::socket::setsockopt(&raw_socket, nix::sys::socket::sockopt::Ipv4PacketInfo, &true) - { - warn!("Could not enable IP_PKTINFO on listener socket: {e}. Interface-aware transmit disabled for this socket."); + if let Err(e) = nix::sys::socket::setsockopt( + &raw_socket, + nix::sys::socket::sockopt::Ipv4PacketInfo, + &true, + ) { + warn!( + "Could not enable IP_PKTINFO on listener socket: {e}. Interface-aware transmit disabled \ + for this socket." + ); } } @@ -357,17 +363,15 @@ impl UDPListener { // Read the datagram and pull out the Copy metadata; the borrow of // `receive_buffer` (through `iov`) ends when this block ends. let (nbytes, source, ifindex, spec_dst) = { - let mut iov = [IoSliceMut::new(&mut self.receive_buffer[..MAX_MESSAGE_SIZE])]; - let msg = match recvmsg::( - fd, - &mut iov, - Some(&mut cmsg_space), - MsgFlags::empty(), - ) { - Ok(m) => m, - Err(Errno::EAGAIN) => return Ok(None), - Err(e) => return Err(io::Error::from_raw_os_error(e as i32)), - }; + let mut iov = [IoSliceMut::new( + &mut self.receive_buffer[..MAX_MESSAGE_SIZE], + )]; + let msg = + match recvmsg::(fd, &mut iov, Some(&mut cmsg_space), MsgFlags::empty()) { + Ok(m) => m, + Err(Errno::EAGAIN) => return Ok(None), + Err(e) => return Err(io::Error::from_raw_os_error(e as i32)), + }; let nbytes = msg.bytes; let source = msg.address.and_then(sockaddr_storage_to_socketaddr); @@ -398,7 +402,10 @@ impl UDPListener { /// Non-Unix fallback: capture the source address only (no interface info). #[cfg(not(unix))] fn recv_one(&mut self) -> io::Result> { - match self.socket.recv_from(&mut self.receive_buffer[..MAX_MESSAGE_SIZE]) { + match self + .socket + .recv_from(&mut self.receive_buffer[..MAX_MESSAGE_SIZE]) + { Ok((nbytes, source)) => Ok(Some(( nbytes, PacketOrigin { @@ -423,9 +430,7 @@ impl UDPListener { } #[cfg(unix)] -fn sockaddr_storage_to_socketaddr( - addr: nix::sys::socket::SockaddrStorage, -) -> Option { +fn sockaddr_storage_to_socketaddr(addr: nix::sys::socket::SockaddrStorage) -> Option { use std::net::{SocketAddrV4, SocketAddrV6}; if let Some(v4) = addr.as_sockaddr_in() { Some(SocketAddr::V4(SocketAddrV4::new(v4.ip(), v4.port()))) diff --git a/src/network/udp_sender.rs b/src/network/udp_sender.rs index 756a850a..6e341d2c 100644 --- a/src/network/udp_sender.rs +++ b/src/network/udp_sender.rs @@ -227,9 +227,7 @@ impl UDPSender { None => { // Requested interface unknown: preserve reachability by falling back to // all interfaces. - trace!( - "send_to_multicast_locator_via: interface {interface:?} not found, sending on all" - ); + trace!("send_to_multicast_locator_via: interface {interface:?} not found, sending on all"); for (_iface, socket) in &self.multicast_sockets { self.send_to_udp_socket(buffer, socket, &socket_address); } diff --git a/src/network/util.rs b/src/network/util.rs index ed510f9a..e33339f1 100644 --- a/src/network/util.rs +++ b/src/network/util.rs @@ -267,9 +267,7 @@ mod tests { "eth0", 3, &[ - IpNetwork::V6( - Ipv6Network::new(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 1), 64).unwrap(), - ), + IpNetwork::V6(Ipv6Network::new(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 1), 64).unwrap()), IpNetwork::V4(Ipv4Network::new(Ipv4Addr::new(10, 0, 0, 7), 24).unwrap()), ], &[pnet_sys::IFF_MULTICAST], @@ -280,7 +278,9 @@ mod tests { assert!(!map.contains_key(&0), "index 0 must be skipped"); assert_eq!( map.get(&3), - Some(&InterfaceSelector::Ip(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 7)))), + Some(&InterfaceSelector::Ip(IpAddr::V4(Ipv4Addr::new( + 10, 0, 0, 7 + )))), "should prefer the IPv4 address" ); } diff --git a/src/rtps/rtps_reader_proxy.rs b/src/rtps/rtps_reader_proxy.rs index 35a4cad9..4f4b032f 100644 --- a/src/rtps/rtps_reader_proxy.rs +++ b/src/rtps/rtps_reader_proxy.rs @@ -566,7 +566,10 @@ mod route_tests { }; fn udp(ip: [u8; 4], port: u16) -> Locator { - Locator::UdpV4(SocketAddrV4::new(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]), port)) + Locator::UdpV4(SocketAddrV4::new( + Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]), + port, + )) } fn iface(ip: [u8; 4]) -> InterfaceSelector { @@ -583,7 +586,11 @@ mod route_tests { let mut rp = proxy_with_prefix(GuidPrefix::UNKNOWN); rp.unicast_locator_list = vec![udp([10, 0, 0, 5], 7410)]; let observations = InterfaceObservations::new(); - rp.resolve_send_route(&observations, &[iface([10, 0, 0, 1])], &DefaultRouteSelector); + rp.resolve_send_route( + &observations, + &[iface([10, 0, 0, 1])], + &DefaultRouteSelector, + ); assert!(rp.send_route().fallback); } @@ -601,7 +608,11 @@ mod route_tests { SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 5)), 7410), ); - rp.resolve_send_route(&observations, &[iface([10, 0, 0, 1])], &DefaultRouteSelector); + rp.resolve_send_route( + &observations, + &[iface([10, 0, 0, 1])], + &DefaultRouteSelector, + ); let route = rp.send_route(); assert!(!route.fallback); diff --git a/src/rtps/transmit.rs b/src/rtps/transmit.rs index 48fa4b4d..6918a294 100644 --- a/src/rtps/transmit.rs +++ b/src/rtps/transmit.rs @@ -16,10 +16,10 @@ //! interface(s) we have actually seen its traffic arrive. This is the primary //! input to route selection. //! -//! Safety guardrail: when we do not have enough information to narrow a remote's -//! route confidently, [`SendRoute::fallback`] is set and the writer falls back -//! to the legacy "send to every advertised locator on every interface" path, so -//! reachability is never reduced. +//! Safety guardrail: when we do not have enough information to narrow a +//! remote's route confidently, [`SendRoute::fallback`] is set and the writer +//! falls back to the legacy "send to every advertised locator on every +//! interface" path, so reachability is never reduced. use std::{ collections::BTreeMap, @@ -197,7 +197,12 @@ impl InterfaceObservations { /// Record that a packet from `prefix` arrived from `source`, on local /// interface `iface` (if it could be determined). - pub fn record(&mut self, prefix: GuidPrefix, iface: Option, source: SocketAddr) { + pub fn record( + &mut self, + prefix: GuidPrefix, + iface: Option, + source: SocketAddr, + ) { self .by_participant .entry(prefix) @@ -214,7 +219,8 @@ impl InterfaceObservations { } } -/// Strategy for turning advertised locators + observations into a [`SendRoute`]. +/// Strategy for turning advertised locators + observations into a +/// [`SendRoute`]. /// /// Kept as a trait so the heuristic can evolve (or be swapped) without touching /// the transmit path. @@ -328,7 +334,10 @@ mod tests { use crate::structure::guid::GuidPrefix; fn udp(ip: [u8; 4], port: u16) -> Locator { - Locator::UdpV4(SocketAddrV4::new(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]), port)) + Locator::UdpV4(SocketAddrV4::new( + Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]), + port, + )) } fn sockaddr(ip: [u8; 4], port: u16) -> SocketAddr { @@ -385,7 +394,12 @@ mod tests { // Observation without a resolvable local interface (unicast only source). obs.record(None, sockaddr([10, 0, 0, 5], 7410)); let mc = udp([239, 255, 0, 1], 7401); - let route = sel.select(&[udp([10, 0, 0, 5], 7410)], &[mc], Some(&obs), &[iface([10, 0, 0, 1])]); + let route = sel.select( + &[udp([10, 0, 0, 5], 7410)], + &[mc], + Some(&obs), + &[iface([10, 0, 0, 1])], + ); assert!(route.fallback); } @@ -440,7 +454,11 @@ mod tests { let p = GuidPrefix::UNKNOWN; obs.record(p, Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410)); obs.record(p, Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410)); - obs.record(p, Some(iface([192, 168, 1, 1])), sockaddr([192, 168, 1, 5], 7410)); + obs.record( + p, + Some(iface([192, 168, 1, 1])), + sockaddr([192, 168, 1, 5], 7410), + ); let recorded = obs.get(p).unwrap(); assert_eq!(recorded.interface_count(), 2); // Sticky: the first-chosen interface stays. @@ -465,7 +483,12 @@ mod tests { let mut obs = ObservedRoutes::default(); let margin = Duration::from_secs(30); let t0 = Instant::now(); - obs.record_at(Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410), t0, margin); + obs.record_at( + Some(iface([10, 0, 0, 1])), + sockaddr([10, 0, 0, 5], 7410), + t0, + margin, + ); // Challenger arrives before the current interface has been silent for the // full margin -> no switch. obs.record_at( @@ -482,7 +505,12 @@ mod tests { let mut obs = ObservedRoutes::default(); let margin = Duration::from_secs(30); let t0 = Instant::now(); - obs.record_at(Some(iface([10, 0, 0, 1])), sockaddr([10, 0, 0, 5], 7410), t0, margin); + obs.record_at( + Some(iface([10, 0, 0, 1])), + sockaddr([10, 0, 0, 5], 7410), + t0, + margin, + ); // Current interface has been silent for >= margin when the challenger is // heard -> switch. obs.record_at( @@ -507,7 +535,12 @@ mod tests { // silent for a full margin, so it is never displaced. for k in 1..=6 { let t = t0 + Duration::from_secs(10 * k); - obs.record_at(Some(challenger), sockaddr([192, 168, 1, 5], 7410), t, margin); + obs.record_at( + Some(challenger), + sockaddr([192, 168, 1, 5], 7410), + t, + margin, + ); obs.record_at( Some(current), sockaddr([10, 0, 0, 5], 7410), diff --git a/src/rtps/writer.rs b/src/rtps/writer.rs index 2a665a91..2e7d60a7 100644 --- a/src/rtps/writer.rs +++ b/src/rtps/writer.rs @@ -1396,7 +1396,7 @@ impl Writer { fn matched_reader_update(&mut self, updated_reader_proxy: &RtpsReaderProxy) -> bool { let mut is_new = false; let is_volatile = self.qos().is_volatile(); // Get this in advance to work with the borrow checker - // Capture the interface set once; resolution consults current observations. + // Capture the interface set once; resolution consults current observations. let multicast_ifaces = self.udp_sender.multicast_interfaces(); self .readers