rx: avoid copies in receive hot paths

- 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.
- recvmmsg / recvmsg_x (Linux + macOS) receive loop moves each filled
  slot buffer into ReceivedPacket via mem::replace instead of cloning
  it; a fresh empty buffer is installed for the next syscall.
- TransportAddr is formatted directly from the SocketAddr without
  going through an intermediate String.

Focused decode bench: ref 1.6 ns/op vs owned 34.7 ns/op (21.4x).
End-to-end iperf is neutral as expected for a ~30 ns saving per
packet.

Unit tests added:
  - test_session_datagram_ref_decode_borrows_payload (verifies the
    payload slice pointer equals the input slice's offset 35, a real
    zero-copy invariant guard against accidental future to_vec)
  - bench_session_datagram_decode_owned_vs_ref (ignored, run with
    --ignored --nocapture)
  - test_transport_addr_from_socket_addr
This commit is contained in:
Martti Malmi
2026-05-15 21:35:46 +00:00
committed by Johnathan Corgan
parent 59225ccfe1
commit b1af151aef
6 changed files with 161 additions and 22 deletions
+21 -14
View File
@@ -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,
};