feat(server): capture kernel receive timestamps #101
No files matched your search
@@ -114,8 +114,10 @@ down. Nothing it does is visible on the wire.
|
||||
|
||||
A reply must leave from the address its request was sent to, or a client reading
|
||||
from a connected socket discards it. An explicit-address listener satisfies that
|
||||
by construction and keeps the plain `recv_from`/`send_to` path — do not route it
|
||||
through ancillary machinery merely because the helper exists.
|
||||
by construction and asks for no destination metadata — do not give it any merely
|
||||
because the helper exists. It keeps the plain `recv_from`/`send_to` path
|
||||
everywhere except Linux, where it receives through `recvmsg` for the unrelated
|
||||
reason below and still sends with plain `send_to`.
|
||||
|
||||
A wildcard listener does not, so it recovers each request's local destination
|
||||
and sends that reply from it. The supported targets are Linux, macOS and
|
||||
@@ -161,6 +163,39 @@ address fails invisibly, as loss.
|
||||
- A send failure is per-packet loss, exactly as an ordinary `send_to` failure
|
||||
is. It never terminates the listener and never rolls back core state.
|
||||
|
||||
## Kernel receive timestamps
|
||||
|
||||
A Linux listener asks the kernel for a software receive timestamp
|
||||
(`SO_TIMESTAMPNS`) on every datagram and reports it as transport metadata. It is
|
||||
the mirror image of destination metadata, and the contrast is the point.
|
||||
|
||||
- **Optional, not required.** Enabling it is best-effort and must never fail
|
||||
construction: a listener without it answers every request exactly as it
|
||||
otherwise would. Destination metadata failing a wildcard bind is correct;
|
||||
timestamp setup failing anything is not.
|
||||
- **Its absence is never packet loss.** No timestamp, a timestamp truncated away
|
||||
on an explicit listener, or one that describes no representable instant, all
|
||||
mean the datagram is served with `kernel_rx_timestamp: None`. Only the
|
||||
destination rules a datagram out, and only for the listener that needs one.
|
||||
- **Nothing consumes it yet.** It stops at the runtime boundary. Do not thread it
|
||||
into `ServerCore`, add a timestamp argument or variant to `handle_datagram`,
|
||||
stash it in a "next receive" side channel, or wrap the core's clock. How a
|
||||
kernel-observed arrival time should enter the core is an open design question,
|
||||
to be settled deliberately and not as a side effect of some other change.
|
||||
- **Do not map wall to monotonic.** It is a realtime reading with no monotonic
|
||||
counterpart, and `ClockSample` is a paired userspace instant. Synthesizing the
|
||||
missing half would invent precision the kernel never reported.
|
||||
- **Do not judge plausibility here.** Conversion is structural only. Whether a
|
||||
representable reading is close enough to the server's own sample to measure
|
||||
against belongs with whoever measures.
|
||||
- Linux only, and receive only. No macOS, FreeBSD or Windows implementation, and
|
||||
no transmit timestamping, without a reason and agreement.
|
||||
- Control-buffer capacity is derived from the cmsg payload types and asserted at
|
||||
compile time. Adding a control message means extending that derivation — an
|
||||
undersized buffer sets `MSG_CTRUNC`, which is silent packet loss for a wildcard
|
||||
listener. Never parse cmsgs positionally or stop at the first match: the kernel
|
||||
orders them as it likes, and each kind is collected independently.
|
||||
|
||||
## Rejection is silence
|
||||
|
||||
Malformed, unauthenticated, unsupported and policy-refused input is discarded
|
||||
|
||||
@@ -12,7 +12,7 @@ use thiserror::Error;
|
||||
use tokio::{net::UdpSocket, time::MissedTickBehavior};
|
||||
|
||||
use crate::{
|
||||
socket_io,
|
||||
socket_io::{self, ReceivedDatagram},
|
||||
socket_options::{marks_with_ipv4_option, set_reply_traffic_class},
|
||||
ServerConfig, ServerCore, ServerError,
|
||||
};
|
||||
@@ -122,6 +122,13 @@ impl Server {
|
||||
.map_err(|source| ServerRuntimeError::SourceSelectionSetup { addr, source })?;
|
||||
}
|
||||
|
||||
// Kernel receive timestamps, for every listener and strictly after the
|
||||
// mandatory setup above so a refusal here can neither hide nor stand in
|
||||
// for a wildcard listener's failure to secure its reply sources. This
|
||||
// one is allowed to fail: it buys accuracy, not correctness, and a
|
||||
// server that cannot have it is still a correct server.
|
||||
let _ = socket_io::try_configure_kernel_rx_timestamp(&socket);
|
||||
|
||||
Ok(Self {
|
||||
socket,
|
||||
core: ServerCore::new(config),
|
||||
@@ -218,7 +225,17 @@ impl Server {
|
||||
Err(source) if source.kind() == io::ErrorKind::Interrupted => continue,
|
||||
Err(source) => return Err(ServerRuntimeError::Receive { source }),
|
||||
};
|
||||
let (len, peer) = (received.len, received.peer);
|
||||
let ReceivedDatagram {
|
||||
len,
|
||||
peer,
|
||||
reply_source,
|
||||
// Observed by the transport and deliberately unused.
|
||||
// The core samples its own receive instant, and this
|
||||
// is where a kernel-observed arrival time stops until a
|
||||
// follow-up change decides how it should enter the core
|
||||
// — feeding it in here would change protocol timing.
|
||||
kernel_rx_timestamp: _kernel_rx_timestamp,
|
||||
} = received;
|
||||
if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? {
|
||||
if !self.prepare_reply_traffic_class(reply.traffic_class()) {
|
||||
continue;
|
||||
@@ -227,7 +244,7 @@ impl Server {
|
||||
&self.socket,
|
||||
reply.bytes(),
|
||||
peer,
|
||||
received.reply_source,
|
||||
reply_source,
|
||||
);
|
||||
tokio::pin!(send);
|
||||
let sent = loop {
|
||||
|
||||
@@ -16,8 +16,29 @@
|
||||
//! have one reachable through a safe API. Everything here goes through `nix`'s
|
||||
//! safe `recvmsg`/`sendmsg` wrappers; the crate forbids unsafe code, so there is
|
||||
//! no raw syscall, no `CMSG_*` walking and no hand-written ABI layer.
|
||||
//!
|
||||
//! # Kernel receive timestamps
|
||||
//!
|
||||
//! A Linux listener additionally asks the kernel for a software receive
|
||||
//! timestamp (`SO_TIMESTAMPNS`) on every datagram, wildcard or not, and reports
|
||||
//! it on [`ReceivedDatagram::kernel_rx_timestamp`]. It says when the kernel
|
||||
//! received the datagram, before the socket queue, Tokio readiness and
|
||||
//! scheduling that separate that moment from the one the server would otherwise
|
||||
//! observe.
|
||||
//!
|
||||
//! It is the exact opposite of destination metadata in how much it matters.
|
||||
//! Enabling it is best-effort and never keeps a listener from starting; a
|
||||
//! datagram that arrives without one, or with one describing no representable
|
||||
//! instant, is served normally with no timestamp. Nothing about which datagrams
|
||||
//! reach [`ServerCore`](crate::ServerCore) depends on it.
|
||||
//!
|
||||
//! **Nothing consumes it yet.** It is captured here so a follow-up change can
|
||||
//! decide how a kernel-observed arrival time enters the protocol core; today
|
||||
//! the core still samples its own receive instant and no protocol timing is
|
||||
//! affected. It is a software wall-clock reading only — not a NIC hardware
|
||||
//! timestamp, not monotonic, and not a transmit timestamp.
|
||||
|
||||
use std::{io, net::SocketAddr};
|
||||
use std::{io, net::SocketAddr, time::SystemTime};
|
||||
|
||||
use tokio::net::UdpSocket;
|
||||
|
||||
@@ -59,19 +80,32 @@ pub(crate) enum ReplySource {
|
||||
},
|
||||
}
|
||||
|
||||
/// One received datagram and what answering it requires.
|
||||
/// One received datagram, what answering it requires, and what the kernel
|
||||
/// observed about its arrival.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub(crate) struct ReceivedDatagram {
|
||||
pub(crate) len: usize,
|
||||
pub(crate) peer: SocketAddr,
|
||||
pub(crate) reply_source: ReplySource,
|
||||
/// When the kernel received this datagram, where it said so.
|
||||
///
|
||||
/// Optional accuracy metadata, absent on every target but Linux, on a
|
||||
/// Linux listener whose kernel refused the option, and on any datagram
|
||||
/// whose timestamp did not arrive or did not describe a representable
|
||||
/// instant. A wall-clock reading with no monotonic counterpart: none is
|
||||
/// synthesized, because there is no sound mapping from one to the other.
|
||||
///
|
||||
/// Captured for a follow-up measurement change and consumed by nothing
|
||||
/// today — see the module documentation.
|
||||
pub(crate) kernel_rx_timestamp: Option<SystemTime>,
|
||||
}
|
||||
|
||||
/// Asks the kernel to report each datagram's local destination address.
|
||||
///
|
||||
/// Called once, at construction, and only for a wildcard listener: an
|
||||
/// explicit-address listener needs no ancillary metadata, and adding it would
|
||||
/// widen the platform surface of the common path for nothing.
|
||||
/// explicit-address listener already replies from the one address it is bound
|
||||
/// to, and asking for a destination it cannot use would widen the platform
|
||||
/// surface of the common path for nothing.
|
||||
#[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
|
||||
pub(crate) fn configure_destination_metadata(socket: &UdpSocket, is_ipv4: bool) -> io::Result<()> {
|
||||
ancillary::configure_destination_metadata(socket, is_ipv4)
|
||||
@@ -85,6 +119,27 @@ pub(crate) fn configure_destination_metadata(
|
||||
Err(unsupported())
|
||||
}
|
||||
|
||||
/// Asks the kernel to report when it received each datagram, and reports
|
||||
/// whether it agreed.
|
||||
///
|
||||
/// Called once, at construction, for every listener on a target that has this:
|
||||
/// unlike destination metadata it costs an explicit listener nothing to have,
|
||||
/// and both kinds of listener are equally worth timing accurately.
|
||||
///
|
||||
/// Deliberately not fallible. A listener without kernel timestamps serves every
|
||||
/// request exactly as it otherwise would, so a refusal here is not a reason to
|
||||
/// fail construction, and the returned flag exists to be asserted in tests
|
||||
/// rather than acted on. The datagrams simply carry no timestamp.
|
||||
#[cfg(target_os = "linux")]
|
||||
pub(crate) fn try_configure_kernel_rx_timestamp(socket: &UdpSocket) -> bool {
|
||||
ancillary::try_configure_kernel_rx_timestamp(socket)
|
||||
}
|
||||
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
pub(crate) fn try_configure_kernel_rx_timestamp(_socket: &UdpSocket) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// Receives one datagram, reporting where its reply must come from.
|
||||
///
|
||||
/// `select_reply_source` is the listener's own wildcard-ness, decided once at
|
||||
@@ -97,7 +152,7 @@ pub(crate) fn configure_destination_metadata(
|
||||
/// impossible. `Ok(None)` covers:
|
||||
///
|
||||
/// - payload truncation reported by an ancillary receive (`MSG_TRUNC`);
|
||||
/// - ancillary metadata truncation (`MSG_CTRUNC`);
|
||||
/// - ancillary metadata truncation (`MSG_CTRUNC`) on a wildcard listener;
|
||||
/// - a wildcard listener's missing or unusable local destination;
|
||||
/// - a datagram filling `buffer` exactly, which is conservatively treated as
|
||||
/// potentially truncated.
|
||||
@@ -105,8 +160,13 @@ pub(crate) fn configure_destination_metadata(
|
||||
/// That last rule exists because `recv_from` reports no truncation flag, so an
|
||||
/// exactly-full result cannot be told apart from a larger datagram cut down to
|
||||
/// the supplied capacity. It is expressed against `buffer.len()`, so the policy
|
||||
/// follows whatever receive capacity the caller supplies, and it applies to both
|
||||
/// receive paths alike.
|
||||
/// follows whatever receive capacity the caller supplies, and it applies to
|
||||
/// every receive path alike.
|
||||
///
|
||||
/// A missing or unusable *kernel receive timestamp* is on none of those lists.
|
||||
/// It costs precision, not correctness, so the datagram is accepted without one
|
||||
/// — including on an explicit listener whose control data was truncated, where
|
||||
/// the timestamp is all that can have been lost.
|
||||
///
|
||||
/// Cancel-safe: nothing is consumed until a datagram is fully received.
|
||||
pub(crate) async fn receive(
|
||||
@@ -118,17 +178,42 @@ pub(crate) async fn receive(
|
||||
let received = if select_reply_source {
|
||||
receive_with_destination(socket, buffer).await?
|
||||
} else {
|
||||
let (len, peer) = socket.recv_from(buffer).await?;
|
||||
Some(ReceivedDatagram {
|
||||
len,
|
||||
peer,
|
||||
reply_source: ReplySource::Bound,
|
||||
})
|
||||
receive_bound(socket, buffer).await?
|
||||
};
|
||||
|
||||
Ok(received.filter(|datagram| datagram.len < capacity))
|
||||
}
|
||||
|
||||
/// The explicit-address receive, which needs no ancillary data to answer
|
||||
/// correctly.
|
||||
///
|
||||
/// Linux takes it through `recvmsg` anyway, purely so an enabled
|
||||
/// `SCM_TIMESTAMPNS` can be read off the same call — `recv_from` cannot retrieve
|
||||
/// control messages. Elsewhere there is no timestamp to collect, so the plain
|
||||
/// path stays plain rather than widening the common path's platform surface for
|
||||
/// nothing.
|
||||
#[cfg(target_os = "linux")]
|
||||
async fn receive_bound(
|
||||
socket: &UdpSocket,
|
||||
buffer: &mut [u8],
|
||||
) -> io::Result<Option<ReceivedDatagram>> {
|
||||
ancillary::receive_bound(socket, buffer).await
|
||||
}
|
||||
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
async fn receive_bound(
|
||||
socket: &UdpSocket,
|
||||
buffer: &mut [u8],
|
||||
) -> io::Result<Option<ReceivedDatagram>> {
|
||||
let (len, peer) = socket.recv_from(buffer).await?;
|
||||
Ok(Some(ReceivedDatagram {
|
||||
len,
|
||||
peer,
|
||||
reply_source: ReplySource::Bound,
|
||||
kernel_rx_timestamp: None,
|
||||
}))
|
||||
}
|
||||
|
||||
#[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
|
||||
async fn receive_with_destination(
|
||||
socket: &UdpSocket,
|
||||
@@ -232,4 +317,128 @@ mod tests {
|
||||
let (received, _) = receive_one(4, b"irtt-rs").await;
|
||||
assert!(!matches!(received, Ok(Some(_))));
|
||||
}
|
||||
|
||||
/// A listener that never asked for kernel timestamps receives normally and
|
||||
/// reports none — the field is optional in the plainest sense, and its
|
||||
/// absence is not a defect to be worked around later.
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn a_datagram_with_no_kernel_timestamp_is_served_all_the_same() {
|
||||
let (received, _) = receive_one(16, b"irtt").await;
|
||||
let received = received.unwrap().expect("an untimestamped datagram serves");
|
||||
assert_eq!(received.reply_source, ReplySource::Bound);
|
||||
assert_eq!(received.kernel_rx_timestamp, None);
|
||||
}
|
||||
|
||||
/// Linux-only coverage of the two receive paths, which is where the
|
||||
/// timestamp actually comes from.
|
||||
#[cfg(target_os = "linux")]
|
||||
mod linux {
|
||||
use std::net::SocketAddr;
|
||||
|
||||
use super::*;
|
||||
|
||||
/// Sends `payload` from loopback to `listener` and receives it once.
|
||||
async fn exchange(
|
||||
listener: &UdpSocket,
|
||||
select_reply_source: bool,
|
||||
) -> (Option<ReceivedDatagram>, Vec<u8>) {
|
||||
let payload = b"irtt";
|
||||
let destination =
|
||||
SocketAddr::from((Ipv4Addr::LOCALHOST, listener.local_addr().unwrap().port()));
|
||||
let sender = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
|
||||
sender.send_to(payload, destination).await.unwrap();
|
||||
|
||||
let mut buffer = vec![0; 64];
|
||||
let received = receive(listener, &mut buffer, select_reply_source)
|
||||
.await
|
||||
.unwrap();
|
||||
(received, buffer)
|
||||
}
|
||||
|
||||
/// Reports the timestamp, or says why it is skipping the assertion
|
||||
/// rather than inventing one. Loopback under a normal kernel supplies
|
||||
/// it; a container or emulator that does not must not fail the run.
|
||||
fn timestamp_or_skip(received: &ReceivedDatagram, what: &str) {
|
||||
match received.kernel_rx_timestamp {
|
||||
Some(timestamp) => assert!(
|
||||
timestamp.duration_since(std::time::UNIX_EPOCH).is_ok(),
|
||||
"a kernel receive timestamp is a wall-clock instant"
|
||||
),
|
||||
None => eprintln!(
|
||||
"skipping {what} kernel timestamp assertion: \
|
||||
kernel did not provide SCM_TIMESTAMPNS"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// The explicit-address listener's new `recvmsg` path: it must keep
|
||||
/// reporting exactly what `recv_from` did, and additionally see the
|
||||
/// timestamp `recv_from` had no way to retrieve.
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn an_explicit_listener_receives_a_datagram_and_its_kernel_timestamp() {
|
||||
let listener = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
|
||||
assert!(
|
||||
try_configure_kernel_rx_timestamp(&listener),
|
||||
"Linux loopback accepts SO_TIMESTAMPNS"
|
||||
);
|
||||
|
||||
let (received, buffer) = exchange(&listener, false).await;
|
||||
let received = received.expect("an explicit listener serves a short datagram");
|
||||
|
||||
assert_eq!(received.len, 4);
|
||||
assert_eq!(&buffer[..received.len], b"irtt");
|
||||
assert_eq!(received.peer.ip(), Ipv4Addr::LOCALHOST);
|
||||
assert_eq!(received.reply_source, ReplySource::Bound);
|
||||
timestamp_or_skip(&received, "explicit");
|
||||
}
|
||||
|
||||
/// The regression this whole change risks: asking for a second control
|
||||
/// message must not crowd out the first. A wildcard listener with both
|
||||
/// enabled has to recover its destination *and* its timestamp from one
|
||||
/// datagram, with no `MSG_CTRUNC` drop and no dependence on the order
|
||||
/// the kernel returns them in.
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn a_wildcard_listener_recovers_its_destination_and_timestamp_together() {
|
||||
let listener = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).await.unwrap();
|
||||
configure_destination_metadata(&listener, true).unwrap();
|
||||
assert!(try_configure_kernel_rx_timestamp(&listener));
|
||||
|
||||
let (received, buffer) = exchange(&listener, true).await;
|
||||
let received = received.expect(
|
||||
"a wildcard listener asking for two control messages still serves the datagram",
|
||||
);
|
||||
|
||||
assert_eq!(received.len, 4);
|
||||
assert_eq!(&buffer[..received.len], b"irtt");
|
||||
assert_eq!(received.peer.ip(), Ipv4Addr::LOCALHOST);
|
||||
assert_eq!(
|
||||
received.reply_source,
|
||||
ReplySource::V4 {
|
||||
address: Ipv4Addr::LOCALHOST
|
||||
},
|
||||
"the required destination survives the added timestamp request"
|
||||
);
|
||||
timestamp_or_skip(&received, "wildcard");
|
||||
}
|
||||
|
||||
/// The timestamp is not a precondition for either listener: a wildcard
|
||||
/// one that never enabled it still recovers the destination its reply
|
||||
/// depends on and serves the request.
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn a_wildcard_listener_without_timestamps_still_recovers_its_destination() {
|
||||
let listener = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).await.unwrap();
|
||||
configure_destination_metadata(&listener, true).unwrap();
|
||||
|
||||
let (received, _) = exchange(&listener, true).await;
|
||||
let received = received.expect("no timestamp requested, and none needed");
|
||||
|
||||
assert_eq!(
|
||||
received.reply_source,
|
||||
ReplySource::V4 {
|
||||
address: Ipv4Addr::LOCALHOST
|
||||
}
|
||||
);
|
||||
assert_eq!(received.kernel_rx_timestamp, None);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,21 @@
|
||||
//! Destination-address ancillary data on Linux, macOS and FreeBSD.
|
||||
//! Ancillary receive data on Linux, macOS and FreeBSD.
|
||||
//!
|
||||
//! Three platforms, two mechanisms per address family, all reached through
|
||||
//! `nix`'s safe wrappers:
|
||||
//! Two independent kinds of metadata travel with a received datagram here, and
|
||||
//! they are not equally important:
|
||||
//!
|
||||
//! - the request's **local destination address**, which a wildcard listener
|
||||
//! must have to answer from the endpoint the client contacted. Losing it
|
||||
//! makes the request unanswerable, so the datagram is dropped.
|
||||
//! - the kernel's **software receive timestamp**, an accuracy enhancement that
|
||||
//! Linux listeners request opportunistically. Losing it costs nothing but
|
||||
//! precision, so the datagram is served with no timestamp.
|
||||
//!
|
||||
//! Both are collected from one `recvmsg`, and neither is looked for at a fixed
|
||||
//! position: the kernel may return them in either order, so the whole usable
|
||||
//! control sequence is walked and each kind recorded where it appears.
|
||||
//!
|
||||
//! Three platforms, two destination mechanisms per address family, all reached
|
||||
//! through `nix`'s safe wrappers:
|
||||
//!
|
||||
//! | | receive | send |
|
||||
//! |---|---|---|
|
||||
@@ -17,6 +31,11 @@
|
||||
//! receive of which only two mean anything on send. Each direction is built
|
||||
//! explicitly for that reason.
|
||||
//!
|
||||
//! On Linux the kernel receive timestamp arrives as `SCM_TIMESTAMPNS`, enabled
|
||||
//! by `SO_TIMESTAMPNS`. It is a software timestamp taken by the kernel's
|
||||
//! realtime clock as the datagram is received — not a NIC hardware timestamp,
|
||||
//! not a monotonic reading, and nothing to do with transmission.
|
||||
//!
|
||||
//! Both directions run nonblocking through Tokio readiness, following the same
|
||||
//! `readable()`/`try_io` shape the client's ancillary receive path uses. One
|
||||
//! socket, one task, no blocking call and no spawned thread.
|
||||
@@ -25,23 +44,63 @@ use std::{
|
||||
io::{self, IoSlice, IoSliceMut},
|
||||
net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV6},
|
||||
os::fd::AsRawFd,
|
||||
time::SystemTime,
|
||||
};
|
||||
|
||||
use nix::{
|
||||
libc,
|
||||
sys::socket::{
|
||||
recvmsg, sendmsg, setsockopt, sockopt, ControlMessage, ControlMessageOwned, MsgFlags,
|
||||
RecvMsg, SockaddrStorage,
|
||||
cmsg_space, recvmsg, sendmsg, setsockopt, sockopt, ControlMessage, ControlMessageOwned,
|
||||
MsgFlags, RecvMsg, SockaddrStorage,
|
||||
},
|
||||
};
|
||||
use tokio::{io::Interest, net::UdpSocket};
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
use std::time::{Duration, UNIX_EPOCH};
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
use nix::sys::time::TimeSpec;
|
||||
|
||||
use super::{ReceivedDatagram, ReplySource};
|
||||
|
||||
/// Room for the destination control message. The largest of them is an
|
||||
/// `in6_pktinfo`, and this leaves headroom for a second message the kernel may
|
||||
/// add, so `MSG_CTRUNC` reflects a real problem rather than a tight buffer.
|
||||
const CONTROL_LEN: usize = 128;
|
||||
/// Room for the IPv4 destination control message, whose payload is the
|
||||
/// platform's own: Linux reports an `in_pktinfo`, the BSDs a bare `in_addr`.
|
||||
#[cfg(target_os = "linux")]
|
||||
const DESTINATION_V4_LEN: usize = cmsg_space::<libc::in_pktinfo>();
|
||||
#[cfg(any(target_os = "macos", target_os = "freebsd"))]
|
||||
const DESTINATION_V4_LEN: usize = cmsg_space::<libc::in_addr>();
|
||||
|
||||
/// Room for the IPv6 destination control message, which is an `in6_pktinfo`
|
||||
/// everywhere here.
|
||||
const DESTINATION_V6_LEN: usize = cmsg_space::<libc::in6_pktinfo>();
|
||||
|
||||
/// Room for the kernel receive timestamp, which only Linux is asked for.
|
||||
#[cfg(target_os = "linux")]
|
||||
const TIMESTAMP_LEN: usize = cmsg_space::<libc::timespec>();
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
const TIMESTAMP_LEN: usize = 0;
|
||||
|
||||
/// Capacity for every control message this server asks the kernel for, derived
|
||||
/// from the payload types themselves rather than from a round number believed
|
||||
/// to be large enough.
|
||||
///
|
||||
/// Both address families are counted although one socket only ever receives
|
||||
/// one of them: a few bytes of headroom is cheaper than a per-family buffer
|
||||
/// type, and a dual-stack listener's family is a runtime fact. Undersizing this
|
||||
/// would not fail loudly — it would set `MSG_CTRUNC` and turn valid wildcard
|
||||
/// requests into silent drops — so the sum is stated in full and proved by a
|
||||
/// receive test rather than by inspection.
|
||||
const CONTROL_LEN: usize = DESTINATION_V4_LEN + DESTINATION_V6_LEN + TIMESTAMP_LEN;
|
||||
|
||||
/// Whichever family a socket turns out to be, its destination message and the
|
||||
/// timestamp must fit together. Checked at compile time on every target this
|
||||
/// module builds for, so a future control message that outgrows the buffer
|
||||
/// fails the build rather than the wildcard listeners that depend on it.
|
||||
const _: () = {
|
||||
assert!(CONTROL_LEN >= DESTINATION_V4_LEN + TIMESTAMP_LEN);
|
||||
assert!(CONTROL_LEN >= DESTINATION_V6_LEN + TIMESTAMP_LEN);
|
||||
};
|
||||
|
||||
pub(super) fn configure_destination_metadata(socket: &UdpSocket, is_ipv4: bool) -> io::Result<()> {
|
||||
if is_ipv4 {
|
||||
@@ -63,20 +122,59 @@ fn configure_ipv4_destination(socket: &UdpSocket) -> io::Result<()> {
|
||||
setsockopt(socket, sockopt::Ipv4RecvDstAddr, &true).map_err(io::Error::from)
|
||||
}
|
||||
|
||||
/// Asks the kernel for a software receive timestamp on each datagram, and
|
||||
/// reports whether it agreed.
|
||||
///
|
||||
/// Deliberately infallible for the caller. Unlike destination metadata, this is
|
||||
/// an accuracy enhancement and not a correctness requirement: a listener that
|
||||
/// does not get it answers every request exactly as it otherwise would, just
|
||||
/// without the kernel's view of when each one arrived.
|
||||
#[cfg(target_os = "linux")]
|
||||
pub(super) fn try_configure_kernel_rx_timestamp(socket: &UdpSocket) -> bool {
|
||||
setsockopt(socket, sockopt::ReceiveTimestampns, &true).is_ok()
|
||||
}
|
||||
|
||||
pub(super) async fn receive_with_destination(
|
||||
socket: &UdpSocket,
|
||||
buffer: &mut [u8],
|
||||
) -> io::Result<Option<ReceivedDatagram>> {
|
||||
receive_ancillary(socket, buffer, true).await
|
||||
}
|
||||
|
||||
/// The Linux explicit-address receive: no destination metadata is needed or
|
||||
/// asked for, and the datagram is taken through `recvmsg` only so an enabled
|
||||
/// `SCM_TIMESTAMPNS` can be read off it.
|
||||
#[cfg(target_os = "linux")]
|
||||
pub(super) async fn receive_bound(
|
||||
socket: &UdpSocket,
|
||||
buffer: &mut [u8],
|
||||
) -> io::Result<Option<ReceivedDatagram>> {
|
||||
receive_ancillary(socket, buffer, false).await
|
||||
}
|
||||
|
||||
/// `require_destination` is the listener's own wildcard-ness. It decides both
|
||||
/// what this receive must recover and how severe a truncated control buffer is.
|
||||
async fn receive_ancillary(
|
||||
socket: &UdpSocket,
|
||||
buffer: &mut [u8],
|
||||
require_destination: bool,
|
||||
) -> io::Result<Option<ReceivedDatagram>> {
|
||||
loop {
|
||||
socket.readable().await?;
|
||||
match socket.try_io(Interest::READABLE, || receive_once(socket, buffer)) {
|
||||
match socket.try_io(Interest::READABLE, || {
|
||||
receive_once(socket, buffer, require_destination)
|
||||
}) {
|
||||
Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
|
||||
result => return result,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn receive_once(socket: &UdpSocket, buffer: &mut [u8]) -> io::Result<Option<ReceivedDatagram>> {
|
||||
fn receive_once(
|
||||
socket: &UdpSocket,
|
||||
buffer: &mut [u8],
|
||||
require_destination: bool,
|
||||
) -> io::Result<Option<ReceivedDatagram>> {
|
||||
let mut control = ControlBuffer::new();
|
||||
let mut iov = [IoSliceMut::new(buffer)];
|
||||
let message = recvmsg::<SockaddrStorage>(
|
||||
@@ -86,7 +184,7 @@ fn receive_once(socket: &UdpSocket, buffer: &mut [u8]) -> io::Result<Option<Rece
|
||||
MsgFlags::MSG_DONTWAIT,
|
||||
)?;
|
||||
|
||||
Ok(received_datagram(&message))
|
||||
Ok(received_datagram(&message, require_destination))
|
||||
}
|
||||
|
||||
/// Turns one received message into a datagram the runtime may serve, or `None`
|
||||
@@ -97,21 +195,57 @@ fn receive_once(socket: &UdpSocket, buffer: &mut [u8]) -> io::Result<Option<Rece
|
||||
/// no destination arrived with the datagram, that promise cannot be kept for
|
||||
/// this request — so it is dropped here, before the core sees it and before any
|
||||
/// session's receive, rate or lifetime state could move.
|
||||
fn received_datagram(message: &RecvMsg<'_, '_, SockaddrStorage>) -> Option<ReceivedDatagram> {
|
||||
if message
|
||||
.flags
|
||||
.intersects(MsgFlags::MSG_TRUNC | MsgFlags::MSG_CTRUNC)
|
||||
{
|
||||
///
|
||||
/// A kernel receive timestamp is never part of that judgement. Absent,
|
||||
/// truncated away or structurally unrepresentable, it simply does not appear on
|
||||
/// the datagram.
|
||||
fn received_datagram(
|
||||
message: &RecvMsg<'_, '_, SockaddrStorage>,
|
||||
require_destination: bool,
|
||||
) -> Option<ReceivedDatagram> {
|
||||
if is_unusable(message.flags, require_destination) {
|
||||
return None;
|
||||
}
|
||||
|
||||
let metadata = receive_metadata(message);
|
||||
Some(ReceivedDatagram {
|
||||
len: message.bytes,
|
||||
peer: peer_endpoint(message.address.as_ref()?)?,
|
||||
reply_source: reply_source(message)?,
|
||||
reply_source: resolved_reply_source(metadata.reply_source, require_destination)?,
|
||||
kernel_rx_timestamp: metadata.kernel_rx_timestamp,
|
||||
})
|
||||
}
|
||||
|
||||
/// Whether the kernel's own report of this receive rules the datagram out.
|
||||
///
|
||||
/// A truncated payload always does: what arrived is not what was sent. A
|
||||
/// truncated *control* buffer only does where the destination was required,
|
||||
/// because then the metadata that may have been cut is the metadata the reply
|
||||
/// cannot be built without. For a listener that asked only for a timestamp,
|
||||
/// the same flag means at worst that the timestamp was lost — a reason to serve
|
||||
/// the request without one, not to discard a client's packet.
|
||||
fn is_unusable(flags: MsgFlags, require_destination: bool) -> bool {
|
||||
flags.contains(MsgFlags::MSG_TRUNC)
|
||||
|| (require_destination && flags.contains(MsgFlags::MSG_CTRUNC))
|
||||
}
|
||||
|
||||
/// Where this datagram's reply must leave from, given what the control data
|
||||
/// yielded.
|
||||
///
|
||||
/// An explicit-address listener has nothing to recover: it sends from the one
|
||||
/// address it is bound to. A wildcard listener has everything to recover, and a
|
||||
/// request whose destination did not arrive is one it cannot answer.
|
||||
fn resolved_reply_source(
|
||||
recovered: Option<ReplySource>,
|
||||
require_destination: bool,
|
||||
) -> Option<ReplySource> {
|
||||
if require_destination {
|
||||
recovered
|
||||
} else {
|
||||
Some(ReplySource::Bound)
|
||||
}
|
||||
}
|
||||
|
||||
/// The request's full source endpoint, keeping the IPv6 zone the core compares
|
||||
/// endpoints by.
|
||||
fn peer_endpoint(address: &SockaddrStorage) -> Option<SocketAddr> {
|
||||
@@ -123,34 +257,84 @@ fn peer_endpoint(address: &SockaddrStorage) -> Option<SocketAddr> {
|
||||
.map(|peer| SocketAddr::from(*peer))
|
||||
}
|
||||
|
||||
fn reply_source(message: &RecvMsg<'_, '_, SockaddrStorage>) -> Option<ReplySource> {
|
||||
for control_message in message.cmsgs().ok()? {
|
||||
/// What one receive's control data carried, kind by kind.
|
||||
///
|
||||
/// Independent fields rather than an early return, because the kernel is free
|
||||
/// to order control messages as it likes: stopping at the destination would
|
||||
/// discard a timestamp that followed it, and stopping at the timestamp would
|
||||
/// discard the destination a wildcard reply depends on.
|
||||
#[derive(Default)]
|
||||
struct ReceiveMetadata {
|
||||
reply_source: Option<ReplySource>,
|
||||
/// Only Linux is ever asked for one, so on the BSDs this stays absent and
|
||||
/// the datagram simply carries no kernel view of its arrival.
|
||||
kernel_rx_timestamp: Option<SystemTime>,
|
||||
}
|
||||
|
||||
fn receive_metadata(message: &RecvMsg<'_, '_, SockaddrStorage>) -> ReceiveMetadata {
|
||||
let mut metadata = ReceiveMetadata::default();
|
||||
// `cmsgs` refuses to walk a control buffer the kernel had to truncate
|
||||
// (`MSG_CTRUNC`), because what is there cannot be trusted to be complete
|
||||
// messages. Whether that is fatal was already decided by `is_unusable`;
|
||||
// here it just means no metadata was recovered.
|
||||
let Ok(control_messages) = message.cmsgs() else {
|
||||
return metadata;
|
||||
};
|
||||
|
||||
for control_message in control_messages {
|
||||
match control_message {
|
||||
// Linux reports the header destination in `ipi_addr`; `ipi_spec_dst`
|
||||
// is the route's local address, which is the same for the unicast
|
||||
// traffic a server sees and is not what the client addressed.
|
||||
#[cfg(target_os = "linux")]
|
||||
ControlMessageOwned::Ipv4PacketInfo(info) => {
|
||||
return ipv4_reply_source(ipv4_addr(info.ipi_addr));
|
||||
metadata.reply_source = ipv4_reply_source(ipv4_addr(info.ipi_addr));
|
||||
}
|
||||
#[cfg(any(target_os = "macos", target_os = "freebsd"))]
|
||||
ControlMessageOwned::Ipv4RecvDstAddr(address) => {
|
||||
return ipv4_reply_source(ipv4_addr(address));
|
||||
metadata.reply_source = ipv4_reply_source(ipv4_addr(address));
|
||||
}
|
||||
ControlMessageOwned::Ipv6PacketInfo(info) => {
|
||||
let address = Ipv6Addr::from(info.ipi6_addr.s6_addr);
|
||||
if address.is_unspecified() {
|
||||
return None;
|
||||
}
|
||||
return Some(ReplySource::V6 {
|
||||
address,
|
||||
interface_index: info.ipi6_ifindex,
|
||||
});
|
||||
metadata.reply_source = ipv6_reply_source(info);
|
||||
}
|
||||
#[cfg(target_os = "linux")]
|
||||
ControlMessageOwned::ScmTimestampns(timestamp) => {
|
||||
metadata.kernel_rx_timestamp = system_time_from_timespec(timestamp);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
None
|
||||
metadata
|
||||
}
|
||||
|
||||
/// An unspecified destination is no destination: nothing can be sent from `::`,
|
||||
/// so the datagram is dropped rather than answered from a source the kernel
|
||||
/// would pick instead.
|
||||
fn ipv6_reply_source(info: libc::in6_pktinfo) -> Option<ReplySource> {
|
||||
let address = Ipv6Addr::from(info.ipi6_addr.s6_addr);
|
||||
(!address.is_unspecified()).then_some(ReplySource::V6 {
|
||||
address,
|
||||
interface_index: info.ipi6_ifindex,
|
||||
})
|
||||
}
|
||||
|
||||
/// A kernel `SCM_TIMESTAMPNS` reading as a wall-clock instant, or `None` where
|
||||
/// it is not one.
|
||||
///
|
||||
/// Purely structural: a negative or out-of-range `timespec`, or one past what
|
||||
/// `SystemTime` can represent, describes no instant and is reported as absent
|
||||
/// metadata. Whether a representable reading is *plausible* — close enough to
|
||||
/// the server's own userspace sample to be worth measuring against — is a
|
||||
/// question for whoever eventually measures with it, and is deliberately not
|
||||
/// asked here.
|
||||
#[cfg(target_os = "linux")]
|
||||
fn system_time_from_timespec(timespec: TimeSpec) -> Option<SystemTime> {
|
||||
let seconds = u64::try_from(timespec.tv_sec()).ok()?;
|
||||
let nanos = u32::try_from(timespec.tv_nsec()).ok()?;
|
||||
if nanos >= 1_000_000_000 {
|
||||
return None;
|
||||
}
|
||||
UNIX_EPOCH.checked_add(Duration::new(seconds, nanos))
|
||||
}
|
||||
|
||||
/// An unspecified destination is no destination: nothing can be sent from
|
||||
@@ -373,4 +557,113 @@ mod tests {
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// What the two listener kinds do with the metadata they did or did not
|
||||
/// get, which is the whole distinction this module rests on: a destination
|
||||
/// is correctness-critical for one of them and irrelevant to the other.
|
||||
#[test]
|
||||
fn only_a_listener_that_needs_a_destination_is_stopped_by_a_missing_one() {
|
||||
let recovered = ReplySource::V4 {
|
||||
address: Ipv4Addr::LOCALHOST,
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
resolved_reply_source(Some(recovered), true),
|
||||
Some(recovered),
|
||||
"a wildcard listener replies from the destination it recovered"
|
||||
);
|
||||
assert_eq!(
|
||||
resolved_reply_source(None, true),
|
||||
None,
|
||||
"and cannot answer a request whose destination did not arrive"
|
||||
);
|
||||
assert_eq!(
|
||||
resolved_reply_source(None, false),
|
||||
Some(ReplySource::Bound),
|
||||
"an explicit listener never needed one"
|
||||
);
|
||||
assert_eq!(
|
||||
resolved_reply_source(Some(recovered), false),
|
||||
Some(ReplySource::Bound),
|
||||
"and sends from its bind even if one turned up"
|
||||
);
|
||||
}
|
||||
|
||||
/// Which truncation is fatal to whom. A cut payload always is; cut control
|
||||
/// data is fatal only where it may have taken the destination with it.
|
||||
/// Turning a lost *timestamp* into a lost packet would trade a real client
|
||||
/// request for optional precision.
|
||||
#[test]
|
||||
fn control_truncation_only_condemns_a_datagram_whose_destination_it_may_have_taken() {
|
||||
for (flags, require_destination, unusable, why) in [
|
||||
(MsgFlags::empty(), true, false, "a clean wildcard receive"),
|
||||
(MsgFlags::empty(), false, false, "a clean explicit receive"),
|
||||
(
|
||||
MsgFlags::MSG_TRUNC,
|
||||
false,
|
||||
true,
|
||||
"a cut payload is not what the client sent",
|
||||
),
|
||||
(
|
||||
MsgFlags::MSG_TRUNC,
|
||||
true,
|
||||
true,
|
||||
"and no less so for a wildcard listener",
|
||||
),
|
||||
(
|
||||
MsgFlags::MSG_CTRUNC,
|
||||
true,
|
||||
true,
|
||||
"cut control data may have taken the destination",
|
||||
),
|
||||
(
|
||||
MsgFlags::MSG_CTRUNC,
|
||||
false,
|
||||
false,
|
||||
"but for an explicit listener it can only have cost a timestamp",
|
||||
),
|
||||
(
|
||||
MsgFlags::MSG_TRUNC | MsgFlags::MSG_CTRUNC,
|
||||
false,
|
||||
true,
|
||||
"the payload rule still applies",
|
||||
),
|
||||
] {
|
||||
assert_eq!(is_unusable(flags, require_destination), unusable, "{why}");
|
||||
}
|
||||
}
|
||||
|
||||
/// A representable kernel reading becomes an instant; anything else becomes
|
||||
/// absent metadata. Note what is *not* rejected here: a reading far from
|
||||
/// now is structurally fine, and judging it belongs to whoever measures
|
||||
/// with it.
|
||||
#[cfg(target_os = "linux")]
|
||||
#[test]
|
||||
fn a_kernel_timestamp_converts_only_where_it_describes_an_instant() {
|
||||
assert_eq!(
|
||||
system_time_from_timespec(TimeSpec::new(1, 2))
|
||||
.unwrap()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap(),
|
||||
Duration::new(1, 2)
|
||||
);
|
||||
assert_eq!(
|
||||
system_time_from_timespec(TimeSpec::new(0, 0)),
|
||||
Some(UNIX_EPOCH),
|
||||
"the epoch itself is representable"
|
||||
);
|
||||
|
||||
assert_eq!(system_time_from_timespec(TimeSpec::new(-1, 0)), None);
|
||||
assert_eq!(system_time_from_timespec(TimeSpec::new(0, -1)), None);
|
||||
assert_eq!(
|
||||
system_time_from_timespec(TimeSpec::new(0, 1_000_000_000)),
|
||||
None
|
||||
);
|
||||
assert_eq!(
|
||||
system_time_from_timespec(TimeSpec::new(i64::MIN, i64::MIN)),
|
||||
None,
|
||||
"and the extremes convert rather than panic"
|
||||
);
|
||||
let _ = system_time_from_timespec(TimeSpec::new(i64::MAX, 999_999_999));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user