diff --git a/CHANGELOG.md b/CHANGELOG.md index dd53edd..3a98929 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,6 +46,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 is not tied to release tags. - Tag-triggered `package-*` release-build workflows remain untouched. +- Receive hot path: removed two per-packet copies. New borrowed + `SessionDatagramRef` decoder is used in the forwarding handler so + local delivery and coordinate-cache warming no longer allocate or + copy the session payload; the owned `SessionDatagram` is materialized + only when re-encoding for the next hop. Owned `SessionDatagram:: + decode` is reimplemented as `Ref::decode + into_owned`, so the two + decoders cannot drift. On Linux + macOS the `recvmmsg` / `recvmsg_x` + receive loop now moves each filled slot buffer into `ReceivedPacket` + via `mem::replace` instead of cloning it, and `TransportAddr` is + formatted directly from the `SocketAddr` without an intermediate + `String`. Focused decode bench: ref 1.6 ns/op vs owned 34.7 ns/op + (21.4x). ### Fixed diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs index 245004b..e96dfc0 100644 --- a/src/node/handlers/forwarding.rs +++ b/src/node/handlers/forwarding.rs @@ -12,7 +12,8 @@ use crate::node::session_wire::{ }; use crate::node::{Node, NodeError}; use crate::protocol::{ - CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionSetup, + CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionDatagramRef, + SessionSetup, }; use std::time::{Duration, Instant}; use tracing::{debug, warn}; @@ -30,7 +31,7 @@ impl Node { ) { self.stats_mut().forwarding.record_received(payload.len()); - let mut datagram = match SessionDatagram::decode(payload) { + let datagram_ref = match SessionDatagramRef::decode(payload) { Ok(dg) => dg, Err(e) => { self.stats_mut() @@ -41,35 +42,41 @@ impl Node { } }; - // TTL enforcement: decrement and drop if exhausted - if !datagram.decrement_ttl() { + // TTL enforcement: decrement for forwarding and drop only if the + // received datagram was already exhausted. + if datagram_ref.ttl == 0 { self.stats_mut() .forwarding .record_ttl_exhausted(payload.len()); debug!( - src = %datagram.src_addr, - dest = %datagram.dest_addr, + src = %datagram_ref.src_addr, + dest = %datagram_ref.dest_addr, "SessionDatagram TTL exhausted, dropping" ); return; } + let forwarded_ttl = datagram_ref.ttl - 1; // Coordinate cache warming from plaintext session-layer headers - self.try_warm_coord_cache(&datagram); + self.try_warm_coord_cache_ref(&datagram_ref); - // Local delivery: dispatch to session layer handlers - if datagram.dest_addr == *self.node_addr() { + // Local delivery: dispatch to session layer handlers without + // materializing an owned SessionDatagram payload Vec. + if datagram_ref.dest_addr == *self.node_addr() { self.stats_mut().forwarding.record_delivered(payload.len()); self.handle_session_payload( - &datagram.src_addr, - &datagram.payload, - datagram.path_mtu, + &datagram_ref.src_addr, + datagram_ref.payload, + datagram_ref.path_mtu, incoming_ce, ) .await; return; } + let mut datagram = datagram_ref.into_owned(); + datagram.ttl = forwarded_ttl; + // Find next hop toward destination let next_hop_addr = match self.find_next_hop(&datagram.dest_addr) { Some(peer) => *peer.node_addr(), @@ -153,8 +160,8 @@ impl Node { /// /// Decode failures are logged and silently ignored — they don't block /// forwarding. - fn try_warm_coord_cache(&mut self, datagram: &SessionDatagram) { - let prefix = match FspCommonPrefix::parse(&datagram.payload) { + fn try_warm_coord_cache_ref(&mut self, datagram: &SessionDatagramRef<'_>) { + let prefix = match FspCommonPrefix::parse(datagram.payload) { Some(p) => p, None => return, }; diff --git a/src/protocol/link.rs b/src/protocol/link.rs index 1c4781d..102f814 100644 --- a/src/protocol/link.rs +++ b/src/protocol/link.rs @@ -297,6 +297,19 @@ pub struct SessionDatagram { pub payload: Vec, } +/// Borrowed view of a session datagram payload. +/// +/// This avoids allocating and copying the inner payload when the caller only +/// needs to inspect or locally deliver it. +#[derive(Clone, Copy, Debug)] +pub struct SessionDatagramRef<'a> { + pub src_addr: NodeAddr, + pub dest_addr: NodeAddr, + pub ttl: u8, + pub path_mtu: u16, + pub payload: &'a [u8], +} + /// SessionDatagram fixed header size: msg_type(1) + ttl(1) + path_mtu(2) + src_addr(16) + dest_addr(16). pub const SESSION_DATAGRAM_HEADER_SIZE: usize = 36; @@ -353,6 +366,14 @@ impl SessionDatagram { /// Decode from link-layer payload (after msg_type byte has been consumed). pub fn decode(payload: &[u8]) -> Result { + let view = SessionDatagramRef::decode(payload)?; + Ok(view.into_owned()) + } +} + +impl<'a> SessionDatagramRef<'a> { + /// Decode a borrowed view from link-layer payload after the msg_type byte. + pub fn decode(payload: &'a [u8]) -> Result { // ttl(1) + path_mtu(2) + src_addr(16) + dest_addr(16) = 35 if payload.len() < 35 { return Err(ProtocolError::MessageTooShort { @@ -366,16 +387,26 @@ impl SessionDatagram { src_bytes.copy_from_slice(&payload[3..19]); let mut dest_bytes = [0u8; 16]; dest_bytes.copy_from_slice(&payload[19..35]); - let inner_payload = payload[35..].to_vec(); Ok(Self { src_addr: NodeAddr::from_bytes(src_bytes), dest_addr: NodeAddr::from_bytes(dest_bytes), ttl, path_mtu, - payload: inner_payload, + payload: &payload[35..], }) } + + /// Materialize an owned datagram for forwarding/re-encoding paths. + pub fn into_owned(self) -> SessionDatagram { + SessionDatagram { + src_addr: self.src_addr, + dest_addr: self.dest_addr, + ttl: self.ttl, + path_mtu: self.path_mtu, + payload: self.payload.to_vec(), + } + } } // Legacy type alias for compatibility during transition @@ -547,6 +578,75 @@ mod tests { assert_eq!(decoded.payload, payload); } + #[test] + fn test_session_datagram_ref_decode_borrows_payload() { + let src = make_node_addr(0xAA); + let dest = make_node_addr(0xBB); + let payload = vec![0x10, 0x00, 0x05, 0x00, 1, 2, 3, 4, 5]; + let dg = SessionDatagram::new(src, dest, payload.clone()) + .with_ttl(32) + .with_path_mtu(1400); + + let encoded = dg.encode(); + let decoded = SessionDatagramRef::decode(&encoded[1..]).unwrap(); + + assert_eq!(decoded.src_addr, src); + assert_eq!(decoded.dest_addr, dest); + assert_eq!(decoded.ttl, 32); + assert_eq!(decoded.path_mtu, 1400); + assert_eq!(decoded.payload, payload.as_slice()); + assert_eq!( + decoded.payload.as_ptr(), + encoded[SESSION_DATAGRAM_HEADER_SIZE..].as_ptr() + ); + } + + #[test] + #[ignore = "performance benchmark; run explicitly with --ignored --nocapture"] + fn bench_session_datagram_decode_owned_vs_ref() { + use std::hint::black_box; + use std::time::Instant; + + const ITERS: usize = 300_000; + + let src = make_node_addr(0xAA); + let dest = make_node_addr(0xBB); + let payload = vec![0x5A; 1200]; + let datagram = SessionDatagram::new(src, dest, payload) + .with_ttl(32) + .with_path_mtu(1400); + let encoded = datagram.encode(); + let link_payload = &encoded[1..]; + + let ref_start = Instant::now(); + let mut ref_bytes = 0usize; + for _ in 0..ITERS { + let decoded = SessionDatagramRef::decode(black_box(link_payload)).unwrap(); + ref_bytes = ref_bytes.wrapping_add(decoded.payload.len()); + black_box(decoded); + } + let ref_elapsed = ref_start.elapsed(); + + let owned_start = Instant::now(); + let mut owned_bytes = 0usize; + for _ in 0..ITERS { + let decoded = SessionDatagram::decode(black_box(link_payload)).unwrap(); + owned_bytes = owned_bytes.wrapping_add(decoded.payload.len()); + black_box(decoded); + } + let owned_elapsed = owned_start.elapsed(); + + assert_eq!(ref_bytes, owned_bytes); + println!( + "SessionDatagram decode: ref={:.1} ns/op owned={:.1} ns/op speedup={:.2}x iters={} payload_bytes={}", + ref_elapsed.as_secs_f64() * 1_000_000_000.0 / ITERS as f64, + owned_elapsed.as_secs_f64() * 1_000_000_000.0 / ITERS as f64, + owned_elapsed.as_secs_f64() / ref_elapsed.as_secs_f64(), + ITERS, + link_payload.len() - 35 + ); + } + #[test] fn test_session_datagram_empty_payload() { let dg = SessionDatagram::new(make_node_addr(1), make_node_addr(2), Vec::new()); diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index 1e90519..99e898c 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -33,7 +33,7 @@ pub use error::ProtocolError; pub use filter::FilterAnnounce; pub use link::{ Disconnect, DisconnectReason, HandshakeMessageType, LinkMessageType, - SESSION_DATAGRAM_HEADER_SIZE, SessionDatagram, + SESSION_DATAGRAM_HEADER_SIZE, SessionDatagram, SessionDatagramRef, }; pub use session::{ COORDS_REQUIRED_SIZE, CoordsRequired, FspFlags, FspInnerFlags, MTU_EXCEEDED_SIZE, MtuExceeded, diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 41d0f8d..3fa2c14 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -399,6 +399,15 @@ impl TransportAddr { Self(s.as_bytes().to_vec()) } + /// Create a UDP/TCP transport address directly from a socket address. + pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self { + use std::io::Write; + + let mut buf = Vec::with_capacity(56); + write!(&mut buf, "{addr}").expect("Vec::write_fmt is infallible"); + Self(buf) + } + /// Get the raw bytes. pub fn as_bytes(&self) -> &[u8] { &self.0 @@ -1311,6 +1320,15 @@ mod tests { assert_eq!(addr2.as_str(), Some("hello")); } + #[test] + fn test_transport_addr_from_socket_addr() { + let addr = TransportAddr::from_socket_addr("127.0.0.1:2121".parse().unwrap()); + assert_eq!(addr.as_str(), Some("127.0.0.1:2121")); + + let addr = TransportAddr::from_socket_addr("[::1]:2121".parse().unwrap()); + assert_eq!(addr.as_str(), Some("[::1]:2121")); + } + #[test] fn test_link_stats_basic() { let mut stats = LinkStats::new(); diff --git a/src/transport/udp/mod.rs b/src/transport/udp/mod.rs index 27acf2d..7fddcbd 100644 --- a/src/transport/udp/mod.rs +++ b/src/transport/udp/mod.rs @@ -430,7 +430,9 @@ async fn udp_receive_loop( { const BATCH: usize = 32; let buf_size = mtu as usize + 100; - // One contiguous backing alloc; slice it for recvmmsg. + // One Vec per recvmmsg / recvmsg_x slot. When a packet lands, move the + // filled buffer directly into ReceivedPacket and install a fresh empty + // buffer for the next syscall, avoiding a per-packet memcpy. let mut backing: Vec> = (0..BATCH).map(|_| vec![0u8; buf_size]).collect(); let mut addrs: [Option; BATCH] = std::array::from_fn(|_| None); let mut lens: [usize; BATCH] = [0; BATCH]; @@ -454,8 +456,7 @@ async fn udp_receive_loop( }; stats.record_recv(len); - let buf = &backing[i][..len]; - if is_punch_packet(buf) { + if is_punch_packet(&backing[i][..len]) { trace!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -465,8 +466,9 @@ async fn udp_receive_loop( continue; } - let data = buf.to_vec(); - let addr = TransportAddr::from_string(&remote_addr.to_string()); + let mut data = std::mem::replace(&mut backing[i], vec![0u8; buf_size]); + data.truncate(len); + let addr = TransportAddr::from_socket_addr(remote_addr); let packet = ReceivedPacket::new(transport_id, addr, data); trace!(