feat(client): add Linux kernel receive timestamp metadata
This commit is contained in:
1 parent
231021fd40
commit
9b5bb46210
2 files changed
+160
-10
No files matched your search
@@ -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");
|
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]
|
#[test]
|
||||||
fn send_probe_fails_before_open() {
|
fn send_probe_fails_before_open() {
|
||||||
let server = start_fake_server(|_socket, _tx| {});
|
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.ecn,
|
||||||
packet_meta.traffic_class.map(|tc| tc & 0b11)
|
packet_meta.traffic_class.map(|tc| tc & 0b11)
|
||||||
);
|
);
|
||||||
assert_eq!(packet_meta.kernel_rx_timestamp, None);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
other => panic!("expected EchoReply, got {other:?}"),
|
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!(traffic_class, 184);
|
||||||
assert_eq!(packet_meta.dscp, Some(46));
|
assert_eq!(packet_meta.dscp, Some(46));
|
||||||
assert_eq!(packet_meta.ecn, Some(0));
|
assert_eq!(packet_meta.ecn, Some(0));
|
||||||
assert_eq!(packet_meta.kernel_rx_timestamp, None);
|
|
||||||
}
|
}
|
||||||
other => panic!("expected EchoReply, got {other:?}"),
|
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!(traffic_class, 0);
|
||||||
assert_eq!(packet_meta.dscp, Some(0));
|
assert_eq!(packet_meta.dscp, Some(0));
|
||||||
assert_eq!(packet_meta.ecn, 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:?}"),
|
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!(traffic_class, 184);
|
||||||
assert_eq!(packet_meta.dscp, Some(46));
|
assert_eq!(packet_meta.dscp, Some(46));
|
||||||
assert_eq!(packet_meta.ecn, Some(0));
|
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:?}"),
|
other => panic!("expected LateReply, got {other:?}"),
|
||||||
}
|
}
|
||||||
@@ -1347,7 +1385,7 @@ fn process_received_packet_uses_supplied_receive_metadata() {
|
|||||||
recv_ts,
|
recv_ts,
|
||||||
ReceiveMeta {
|
ReceiveMeta {
|
||||||
traffic_class: Some(186),
|
traffic_class: Some(186),
|
||||||
kernel_rx_timestamp: None,
|
kernel_rx_timestamp: Some(SystemTime::UNIX_EPOCH + Duration::new(12, 34)),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
.unwrap();
|
.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.traffic_class, Some(186));
|
||||||
assert_eq!(packet_meta.dscp, Some(46));
|
assert_eq!(packet_meta.dscp, Some(46));
|
||||||
assert_eq!(packet_meta.ecn, Some(2));
|
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:?}"),
|
other => panic!("expected EchoReply, got {other:?}"),
|
||||||
}
|
}
|
||||||
@@ -1395,7 +1436,7 @@ fn process_received_packet_uses_supplied_receive_metadata_for_late_reply() {
|
|||||||
recv_ts,
|
recv_ts,
|
||||||
ReceiveMeta {
|
ReceiveMeta {
|
||||||
traffic_class: Some(0),
|
traffic_class: Some(0),
|
||||||
kernel_rx_timestamp: None,
|
kernel_rx_timestamp: Some(SystemTime::UNIX_EPOCH + Duration::new(56, 78)),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
.unwrap();
|
.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.traffic_class, Some(0));
|
||||||
assert_eq!(packet_meta.dscp, Some(0));
|
assert_eq!(packet_meta.dscp, Some(0));
|
||||||
assert_eq!(packet_meta.ecn, 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:?}"),
|
other => panic!("expected LateReply, got {other:?}"),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,13 +4,15 @@ use std::{
|
|||||||
net::{SocketAddr, UdpSocket},
|
net::{SocketAddr, UdpSocket},
|
||||||
os::fd::AsRawFd,
|
os::fd::AsRawFd,
|
||||||
ptr,
|
ptr,
|
||||||
|
time::{Duration, SystemTime, UNIX_EPOCH},
|
||||||
};
|
};
|
||||||
|
|
||||||
use crate::{metadata::ReceiveMeta, receive::ReceivedDatagram, timing::ClientTimestamp};
|
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<()> {
|
pub(crate) fn configure_receive_metadata(socket: &UdpSocket, remote: SocketAddr) -> io::Result<()> {
|
||||||
|
enable_kernel_rx_timestamps(socket)?;
|
||||||
let socket = socket2::SockRef::from(socket);
|
let socket = socket2::SockRef::from(socket);
|
||||||
if remote.is_ipv4() {
|
if remote.is_ipv4() {
|
||||||
socket.set_recv_tos(true)
|
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(
|
pub(crate) fn recv_datagram(
|
||||||
socket: &UdpSocket,
|
socket: &UdpSocket,
|
||||||
buf: &mut [u8],
|
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;
|
let data_len = cmsg_ref.cmsg_len - libc::CMSG_LEN(0) as usize;
|
||||||
if is_traffic_class_cmsg(cmsg_ref) {
|
if is_traffic_class_cmsg(cmsg_ref) {
|
||||||
meta.traffic_class = read_int_cmsg_low_byte(cmsg, data_len);
|
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);
|
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)
|
|| (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<u8> {
|
unsafe fn read_u8_cmsg(cmsg: *mut libc::cmsghdr, data_len: usize) -> Option<u8> {
|
||||||
if data_len >= mem::size_of::<u8>() {
|
if data_len >= mem::size_of::<u8>() {
|
||||||
Some(*libc::CMSG_DATA(cmsg).cast::<u8>())
|
Some(*libc::CMSG_DATA(cmsg).cast::<u8>())
|
||||||
@@ -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<SystemTime> {
|
||||||
|
if data_len < mem::size_of::<libc::timespec>() {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
let timespec = ptr::read_unaligned(libc::CMSG_DATA(cmsg).cast::<libc::timespec>());
|
||||||
|
system_time_from_timespec(timespec)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn system_time_from_timespec(timespec: libc::timespec) -> Option<SystemTime> {
|
||||||
|
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))]
|
#[repr(align(8))]
|
||||||
struct ControlBuffer([u8; CONTROL_LEN]);
|
struct ControlBuffer([u8; CONTROL_LEN]);
|
||||||
|
|
||||||
@@ -118,7 +164,7 @@ mod tests {
|
|||||||
use std::{
|
use std::{
|
||||||
io,
|
io,
|
||||||
net::{SocketAddr, UdpSocket},
|
net::{SocketAddr, UdpSocket},
|
||||||
time::Duration,
|
time::{Duration, UNIX_EPOCH},
|
||||||
};
|
};
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
@@ -206,6 +252,27 @@ mod tests {
|
|||||||
assert!(datagram.received_at.mono <= after.mono);
|
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]
|
#[test]
|
||||||
fn ipv4_traffic_class_metadata_is_observed_when_kernel_provides_it() {
|
fn ipv4_traffic_class_metadata_is_observed_when_kernel_provides_it() {
|
||||||
let (sender, receiver) = connected_ipv4_loopback_pair();
|
let (sender, receiver) = connected_ipv4_loopback_pair();
|
||||||
@@ -270,6 +337,45 @@ mod tests {
|
|||||||
assert_eq!(packet_meta.ecn, Some(0));
|
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 {
|
fn is_unavailable_ipv6_loopback(error: &io::Error) -> bool {
|
||||||
matches!(
|
matches!(
|
||||||
error.kind(),
|
error.kind(),
|
||||||
|
|||||||
Reference in new issue
Block a user