From 1e706dfefd415aaff9ba204e95686f5c6c586fda Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 7 May 2026 18:51:06 +0700 Subject: [PATCH] feat(client): add Linux kernel receive timestamp metadata --- .../src/client/tests/probes_replies.rs | 60 ++++++++-- crates/irtt-client/src/receive/linux.rs | 110 +++++++++++++++++- 2 files changed, 160 insertions(+), 10 deletions(-) diff --git a/crates/irtt-client/src/client/tests/probes_replies.rs b/crates/irtt-client/src/client/tests/probes_replies.rs index fbe4516..ace51df 100644 --- a/crates/irtt-client/src/client/tests/probes_replies.rs +++ b/crates/irtt-client/src/client/tests/probes_replies.rs @@ -13,6 +13,11 @@ fn metadata_unavailable_skip(test_name: &str) { eprintln!("{test_name}: skipping metadata assertion because kernel did not provide traffic class metadata"); } +#[cfg(all(target_os = "linux", feature = "ancillary"))] +fn kernel_rx_timestamp_unavailable_skip(test_name: &str) { + eprintln!("{test_name}: skipping kernel timestamp assertion because kernel did not provide SCM_TIMESTAMPNS"); +} + #[test] fn send_probe_fails_before_open() { let server = start_fake_server(|_socket, _tx| {}); @@ -329,7 +334,6 @@ fn recv_once_decodes_echo_reply_and_emits_event() { packet_meta.ecn, packet_meta.traffic_class.map(|tc| tc & 0b11) ); - assert_eq!(packet_meta.kernel_rx_timestamp, None); } } other => panic!("expected EchoReply, got {other:?}"), @@ -406,7 +410,6 @@ fn echo_reply_metadata_propagates_observed_dscp_with_ancillary() { assert_eq!(traffic_class, 184); assert_eq!(packet_meta.dscp, Some(46)); assert_eq!(packet_meta.ecn, Some(0)); - assert_eq!(packet_meta.kernel_rx_timestamp, None); } other => panic!("expected EchoReply, got {other:?}"), } @@ -439,7 +442,36 @@ fn echo_reply_metadata_preserves_observed_zero_with_ancillary() { assert_eq!(traffic_class, 0); assert_eq!(packet_meta.dscp, Some(0)); assert_eq!(packet_meta.ecn, Some(0)); - assert_eq!(packet_meta.kernel_rx_timestamp, None); + } + other => panic!("expected EchoReply, got {other:?}"), + } + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +#[cfg(all(target_os = "linux", feature = "ancillary"))] +fn echo_reply_metadata_propagates_kernel_rx_timestamp_with_ancillary() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let events = client.recv_once().unwrap(); + assert_eq!(events.len(), 1); + match &events[0] { + ClientEvent::EchoReply { packet_meta, .. } => { + let Some(timestamp) = packet_meta.kernel_rx_timestamp else { + kernel_rx_timestamp_unavailable_skip( + "echo_reply_metadata_propagates_kernel_rx_timestamp_with_ancillary", + ); + client.close(ClientTimestamp::now()).unwrap(); + server.join(); + return; + }; + assert!(timestamp.duration_since(SystemTime::UNIX_EPOCH).is_ok()); } other => panic!("expected EchoReply, got {other:?}"), } @@ -925,7 +957,13 @@ fn late_reply_metadata_propagates_observed_dscp_with_ancillary() { assert_eq!(traffic_class, 184); assert_eq!(packet_meta.dscp, Some(46)); assert_eq!(packet_meta.ecn, Some(0)); - assert_eq!(packet_meta.kernel_rx_timestamp, None); + if let Some(timestamp) = packet_meta.kernel_rx_timestamp { + assert!(timestamp.duration_since(SystemTime::UNIX_EPOCH).is_ok()); + } else { + kernel_rx_timestamp_unavailable_skip( + "late_reply_metadata_propagates_observed_dscp_with_ancillary", + ); + } } other => panic!("expected LateReply, got {other:?}"), } @@ -1347,7 +1385,7 @@ fn process_received_packet_uses_supplied_receive_metadata() { recv_ts, ReceiveMeta { traffic_class: Some(186), - kernel_rx_timestamp: None, + kernel_rx_timestamp: Some(SystemTime::UNIX_EPOCH + Duration::new(12, 34)), }, ) .unwrap(); @@ -1358,7 +1396,10 @@ fn process_received_packet_uses_supplied_receive_metadata() { assert_eq!(packet_meta.traffic_class, Some(186)); assert_eq!(packet_meta.dscp, Some(46)); assert_eq!(packet_meta.ecn, Some(2)); - assert_eq!(packet_meta.kernel_rx_timestamp, None); + assert_eq!( + packet_meta.kernel_rx_timestamp, + Some(SystemTime::UNIX_EPOCH + Duration::new(12, 34)) + ); } other => panic!("expected EchoReply, got {other:?}"), } @@ -1395,7 +1436,7 @@ fn process_received_packet_uses_supplied_receive_metadata_for_late_reply() { recv_ts, ReceiveMeta { traffic_class: Some(0), - kernel_rx_timestamp: None, + kernel_rx_timestamp: Some(SystemTime::UNIX_EPOCH + Duration::new(56, 78)), }, ) .unwrap(); @@ -1406,7 +1447,10 @@ fn process_received_packet_uses_supplied_receive_metadata_for_late_reply() { assert_eq!(packet_meta.traffic_class, Some(0)); assert_eq!(packet_meta.dscp, Some(0)); assert_eq!(packet_meta.ecn, Some(0)); - assert_eq!(packet_meta.kernel_rx_timestamp, None); + assert_eq!( + packet_meta.kernel_rx_timestamp, + Some(SystemTime::UNIX_EPOCH + Duration::new(56, 78)) + ); } other => panic!("expected LateReply, got {other:?}"), } diff --git a/crates/irtt-client/src/receive/linux.rs b/crates/irtt-client/src/receive/linux.rs index 441c497..5bf7137 100644 --- a/crates/irtt-client/src/receive/linux.rs +++ b/crates/irtt-client/src/receive/linux.rs @@ -4,13 +4,15 @@ use std::{ net::{SocketAddr, UdpSocket}, os::fd::AsRawFd, ptr, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use crate::{metadata::ReceiveMeta, receive::ReceivedDatagram, timing::ClientTimestamp}; -const CONTROL_LEN: usize = 64; +const CONTROL_LEN: usize = 128; pub(crate) fn configure_receive_metadata(socket: &UdpSocket, remote: SocketAddr) -> io::Result<()> { + enable_kernel_rx_timestamps(socket)?; let socket = socket2::SockRef::from(socket); if remote.is_ipv4() { socket.set_recv_tos(true) @@ -19,6 +21,27 @@ pub(crate) fn configure_receive_metadata(socket: &UdpSocket, remote: SocketAddr) } } +fn enable_kernel_rx_timestamps(socket: &UdpSocket) -> io::Result<()> { + let enabled: libc::c_int = 1; + let result = unsafe { + // SAFETY: The file descriptor is borrowed from a valid `UdpSocket`. + // `enabled` is a properly aligned `c_int`, and the length matches the + // pointed-to value for the `SO_TIMESTAMPNS` boolean socket option. + libc::setsockopt( + socket.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_TIMESTAMPNS, + (&enabled as *const libc::c_int).cast(), + mem::size_of_val(&enabled) as libc::socklen_t, + ) + }; + if result == -1 { + Err(io::Error::last_os_error()) + } else { + Ok(()) + } +} + pub(crate) fn recv_datagram( socket: &UdpSocket, buf: &mut [u8], @@ -67,6 +90,8 @@ unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta { let data_len = cmsg_ref.cmsg_len - libc::CMSG_LEN(0) as usize; if is_traffic_class_cmsg(cmsg_ref) { meta.traffic_class = read_int_cmsg_low_byte(cmsg, data_len); + } else if is_kernel_rx_timestamp_cmsg(cmsg_ref) { + meta.kernel_rx_timestamp = read_timespec_cmsg(cmsg, data_len); } } cmsg = libc::CMSG_NXTHDR(msg, cmsg); @@ -79,6 +104,10 @@ fn is_traffic_class_cmsg(cmsg: &libc::cmsghdr) -> bool { || (cmsg.cmsg_level == libc::IPPROTO_IPV6 && cmsg.cmsg_type == libc::IPV6_TCLASS) } +fn is_kernel_rx_timestamp_cmsg(cmsg: &libc::cmsghdr) -> bool { + cmsg.cmsg_level == libc::SOL_SOCKET && cmsg.cmsg_type == libc::SCM_TIMESTAMPNS +} + unsafe fn read_u8_cmsg(cmsg: *mut libc::cmsghdr, data_len: usize) -> Option { if data_len >= mem::size_of::() { Some(*libc::CMSG_DATA(cmsg).cast::()) @@ -96,6 +125,23 @@ unsafe fn read_int_cmsg_low_byte(cmsg: *mut libc::cmsghdr, data_len: usize) -> O } } +unsafe fn read_timespec_cmsg(cmsg: *mut libc::cmsghdr, data_len: usize) -> Option { + if data_len < mem::size_of::() { + return None; + } + let timespec = ptr::read_unaligned(libc::CMSG_DATA(cmsg).cast::()); + system_time_from_timespec(timespec) +} + +fn system_time_from_timespec(timespec: libc::timespec) -> Option { + if timespec.tv_sec < 0 || timespec.tv_nsec < 0 || timespec.tv_nsec >= 1_000_000_000 { + return None; + } + let seconds = u64::try_from(timespec.tv_sec).ok()?; + let nanos = u32::try_from(timespec.tv_nsec).ok()?; + UNIX_EPOCH.checked_add(Duration::new(seconds, nanos)) +} + #[repr(align(8))] struct ControlBuffer([u8; CONTROL_LEN]); @@ -118,7 +164,7 @@ mod tests { use std::{ io, net::{SocketAddr, UdpSocket}, - time::Duration, + time::{Duration, UNIX_EPOCH}, }; use crate::{ @@ -206,6 +252,27 @@ mod tests { assert!(datagram.received_at.mono <= after.mono); } + #[test] + fn kernel_rx_timestamp_metadata_is_observed_when_kernel_provides_it() { + let (sender, receiver) = connected_ipv4_loopback_pair(); + configure_receive_metadata(&receiver, sender.local_addr().unwrap()).unwrap(); + sender.send(b"stamp").unwrap(); + + let mut buf = [0_u8; 16]; + let datagram = recv_datagram(&receiver, &mut buf).unwrap(); + assert_eq!(datagram.len, 5); + assert_eq!(&buf[..datagram.len], b"stamp"); + let Some(timestamp) = datagram.meta.kernel_rx_timestamp else { + eprintln!( + "skipping kernel timestamp assertion: kernel did not provide SCM_TIMESTAMPNS" + ); + return; + }; + + let duration = timestamp.duration_since(UNIX_EPOCH).unwrap(); + assert!(duration.as_nanos() > 0); + } + #[test] fn ipv4_traffic_class_metadata_is_observed_when_kernel_provides_it() { let (sender, receiver) = connected_ipv4_loopback_pair(); @@ -270,6 +337,45 @@ mod tests { assert_eq!(packet_meta.ecn, Some(0)); } + #[test] + fn timespec_conversion_accepts_valid_unix_timestamp() { + let timestamp = super::system_time_from_timespec(libc::timespec { + tv_sec: 1, + tv_nsec: 2, + }) + .unwrap(); + + assert_eq!( + timestamp.duration_since(UNIX_EPOCH).unwrap(), + Duration::new(1, 2) + ); + } + + #[test] + fn timespec_conversion_rejects_negative_or_invalid_values() { + assert_eq!( + super::system_time_from_timespec(libc::timespec { + tv_sec: -1, + tv_nsec: 0, + }), + None + ); + assert_eq!( + super::system_time_from_timespec(libc::timespec { + tv_sec: 0, + tv_nsec: -1, + }), + None + ); + assert_eq!( + super::system_time_from_timespec(libc::timespec { + tv_sec: 0, + tv_nsec: 1_000_000_000, + }), + None + ); + } + fn is_unavailable_ipv6_loopback(error: &io::Error) -> bool { matches!( error.kind(),