refactor(client): use SignedDuration for effective RTT to support negative values

Simplify RttSample by merging signed and unsigned fields into a single SignedDuration for adjusted and effective RTT. This ensures accurate reporting and statistics when server processing time exceeds raw RTT.
This commit is contained in:
Nils Andreas Svee committed 2026-05-17 21:30:51 +07:00
1 parent 37de7c96a1
commit bec56d4114
14 files changed
+207 -86

No files matched your search

+78 -15
View File
@@ -12,8 +12,8 @@ use std::{
use clap::{Parser, ValueEnum};
use irtt_client::{
ClientConfig, ClientEvent, NegotiatedParams, NegotiationPolicy, OneWayDelaySample, PacketMeta,
ReceivedStatsSample, RttSample, ServerTiming, SocketConfig, WarningKind, MAX_DSCP_CODEPOINT,
MAX_SERVER_FILL_BYTES, MAX_TTL, MAX_UDP_PAYLOAD_LENGTH,
ReceivedStatsSample, RttSample, ServerTiming, SignedDuration, SocketConfig, WarningKind,
MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_TTL, MAX_UDP_PAYLOAD_LENGTH,
};
use irtt_proto::{Clock, ReceivedStats, StampAt};
@@ -397,7 +397,7 @@ pub fn format_human_event_with_options(
..
} => {
let mut out = format!("seq={seq}");
write!(out, " rtt={}", format_duration(rtt.effective)).unwrap();
write!(out, " rtt={}", format_signed_duration(rtt.effective)).unwrap();
write_human_one_way(&mut out, *one_way);
write!(out, " ipdv={}", format_human_ipdv(stats, *seq)).unwrap();
if let Some(processing) = server_timing.and_then(|timing| timing.processing) {
@@ -425,7 +425,7 @@ pub fn format_human_event_with_options(
} => {
let mut out = format!("late seq={seq} highest_seen={highest_seen} remote={remote}",);
if let Some(rtt) = rtt {
write!(out, " rtt={}", format_duration(rtt.effective)).unwrap();
write!(out, " rtt={}", format_signed_duration(rtt.effective)).unwrap();
write_human_one_way(&mut out, *one_way);
write!(out, " ipdv={}", format_human_ipdv(stats, *seq)).unwrap();
}
@@ -440,7 +440,7 @@ pub fn format_human_event_with_options(
fn format_rtt_us(event: &ClientEvent) -> Option<String> {
match event {
ClientEvent::EchoReply { rtt, .. } => Some(duration_us(rtt.effective).to_string()),
ClientEvent::EchoReply { rtt, .. } => Some(signed_duration_us(rtt.effective).to_string()),
_ => None,
}
}
@@ -585,10 +585,10 @@ fn format_simple(event: &ClientEvent) -> String {
} => {
let mut out = format!(
"reply seq={seq} remote={remote} rtt_us={}",
duration_us(rtt.effective)
signed_duration_us(rtt.effective)
);
if let Some(raw) = rtt.adjusted.map(|_| rtt.raw) {
write!(out, " raw_rtt_us={}", duration_us(raw)).unwrap();
if rtt.adjusted.is_some() {
write!(out, " raw_rtt_us={}", duration_us(rtt.raw)).unwrap();
}
if let Some(processing) = server_timing.and_then(|timing| timing.processing) {
write!(out, " server_processing_us={}", duration_us(processing)).unwrap();
@@ -610,7 +610,7 @@ fn format_simple(event: &ClientEvent) -> String {
} => {
let mut out = format!("late seq={seq} highest_seen={highest_seen} remote={remote}",);
if let Some(rtt) = rtt {
write!(out, " rtt_us={}", duration_us(rtt.effective)).unwrap();
write!(out, " rtt_us={}", signed_duration_us(rtt.effective)).unwrap();
}
out
}
@@ -656,11 +656,11 @@ fn write_rtt(out: &mut String, rtt: &RttSample) {
out,
" raw_rtt_us={} effective_rtt_us={}",
duration_us(rtt.raw),
duration_us(rtt.effective)
signed_duration_us(rtt.effective)
)
.unwrap();
if let Some(adjusted) = rtt.adjusted {
write!(out, " adjusted_rtt_us={}", duration_us(adjusted)).unwrap();
write!(out, " adjusted_rtt_us={}", signed_duration_us(adjusted)).unwrap();
}
}
@@ -773,6 +773,10 @@ fn duration_us(duration: Duration) -> u128 {
duration.as_micros()
}
fn signed_duration_us(duration: SignedDuration) -> i128 {
duration.ns / 1_000
}
fn format_optional_duration(duration: Option<Duration>) -> String {
duration
.map(format_duration)
@@ -783,6 +787,18 @@ fn format_duration(duration: Duration) -> String {
format_ns(duration.as_nanos() as f64)
}
fn format_signed_duration(duration: SignedDuration) -> String {
format_signed_ns(duration.ns as f64)
}
fn format_signed_ns(ns: f64) -> String {
if ns < 0.0 {
format!("-{}", format_ns(-ns))
} else {
format_ns(ns)
}
}
fn format_ns(ns: f64) -> String {
if ns < 1_000.0 {
format!("{ns:.0}ns")
@@ -853,10 +869,8 @@ mod tests {
received_at: test_timestamp(Duration::from_secs(1) + Duration::from_micros(1500)),
rtt: RttSample {
raw: Duration::from_micros(1500),
adjusted: Some(Duration::from_micros(1200)),
effective: Duration::from_micros(1200),
adjusted_signed: Some(SignedDuration { ns: 1_200_000 }),
effective_signed: SignedDuration { ns: 1_200_000 },
adjusted: Some(SignedDuration { ns: 1_200_000 }),
effective: SignedDuration { ns: 1_200_000 },
},
server_timing: Some(ServerTiming {
receive_wall_ns: Some(1_000),
@@ -880,6 +894,33 @@ mod tests {
}
}
fn negative_adjusted_reply_event() -> ClientEvent {
ClientEvent::EchoReply {
seq: 8,
remote: test_remote(),
sent_at: test_timestamp(Duration::from_secs(1)),
received_at: test_timestamp(Duration::from_secs(1) + Duration::from_micros(500)),
rtt: RttSample {
raw: Duration::from_micros(500),
adjusted: Some(SignedDuration { ns: -1_500_000 }),
effective: SignedDuration { ns: -1_500_000 },
},
server_timing: Some(ServerTiming {
receive_wall_ns: Some(1_000),
receive_mono_ns: Some(2_000),
send_wall_ns: Some(2_001_000),
send_mono_ns: Some(2_002_000),
midpoint_wall_ns: None,
midpoint_mono_ns: None,
processing: Some(Duration::from_micros(2_000)),
}),
one_way: None,
received_stats: None,
bytes: 64,
packet_meta: PacketMeta::default(),
}
}
fn late_event_with_meta(packet_meta: PacketMeta) -> ClientEvent {
ClientEvent::LateReply {
seq: 4,
@@ -1157,6 +1198,28 @@ mod tests {
);
}
#[test]
#[cfg(feature = "stats")]
fn streamed_and_summary_rtt_use_signed_effective_policy() {
let event = negative_adjusted_reply_event();
assert_eq!(
format_event(&event, OutputMode::RttUs),
Some("-1500".to_owned())
);
let machine = format_event(&event, OutputMode::Machine).unwrap();
assert!(machine.contains("raw_rtt_us=500"));
assert!(machine.contains("effective_rtt_us=-1500"));
assert!(machine.contains("adjusted_rtt_us=-1500"));
let mut collector = irtt_stats::StatsCollector::new(irtt_stats::StatsConfig::finite());
collector.process(&event);
let summary = collector.snapshot();
assert_eq!(summary.rtt.primary.min_ns, Some(-1_500_000));
assert_eq!(summary.rtt.adjusted.min_ns, Some(-1_500_000));
}
#[test]
fn machine_prints_stable_key_value_fields() {
let line = format_event(&reply_event(), OutputMode::Machine).unwrap();
+3 -9
View File
@@ -426,9 +426,7 @@ mod tests {
rtt: RttSample {
raw: Duration::from_micros(rtt_us),
adjusted: None,
effective: Duration::from_micros(rtt_us),
adjusted_signed: None,
effective_signed: SignedDuration {
effective: SignedDuration {
ns: i128::from(rtt_us) * 1_000,
},
},
@@ -468,9 +466,7 @@ mod tests {
rtt: RttSample {
raw: Duration::from_micros(1200),
adjusted: None,
effective: Duration::from_micros(1200),
adjusted_signed: None,
effective_signed: SignedDuration { ns: 1_200_000 },
effective: SignedDuration { ns: 1_200_000 },
},
server_timing: None,
one_way: None,
@@ -583,9 +579,7 @@ mod tests {
rtt: RttSample {
raw: Duration::from_micros(1),
adjusted: None,
effective: Duration::from_micros(1),
adjusted_signed: None,
effective_signed: SignedDuration { ns: 1_000 },
effective: SignedDuration { ns: 1_000 },
},
server_timing: None,
one_way: None,
+4 -8
View File
@@ -206,10 +206,8 @@ mod tests {
received_at,
rtt: RttSample {
raw: Duration::from_micros(1500),
adjusted: Some(Duration::from_micros(1200)),
effective: Duration::from_micros(1200),
adjusted_signed: Some(SignedDuration { ns: 1_200_000 }),
effective_signed: SignedDuration { ns: 1_200_000 },
adjusted: Some(SignedDuration { ns: 1_200_000 }),
effective: SignedDuration { ns: 1_200_000 },
},
server_timing: Some(ServerTiming {
receive_wall_ns: None,
@@ -254,10 +252,8 @@ mod tests {
received_at,
rtt: RttSample {
raw: Duration::from_micros(1500),
adjusted: Some(Duration::from_micros(1200)),
effective: Duration::from_micros(1200),
adjusted_signed: Some(SignedDuration { ns: 1_200_000 }),
effective_signed: SignedDuration { ns: 1_200_000 },
adjusted: Some(SignedDuration { ns: 1_200_000 }),
effective: SignedDuration { ns: 1_200_000 },
},
server_timing: None,
one_way: None,
+2 -9
View File
@@ -727,22 +727,15 @@ fn compute_rtt(
let server_processing = compute_server_processing(ts);
let adjusted = server_processing.and_then(|sp| raw.checked_sub(sp));
let effective = adjusted.unwrap_or(raw);
let adjusted_signed = server_processing.map(|sp| SignedDuration {
let adjusted = server_processing.map(|sp| SignedDuration {
ns: duration_ns_i128(raw) - duration_ns_i128(sp),
});
let effective_signed = adjusted_signed.unwrap_or(SignedDuration {
ns: duration_ns_i128(raw),
});
let effective = adjusted.unwrap_or_else(|| SignedDuration::from_duration(raw));
RttSample {
raw,
adjusted,
effective,
adjusted_signed,
effective_signed,
}
}
@@ -41,7 +41,11 @@ fn recv_once_decodes_echo_reply_and_emits_event() {
assert_eq!(*seq, 0);
assert_eq!(*bytes, echo_packet_len(false, &params));
assert!(rtt.raw > Duration::ZERO);
assert_eq!(rtt.effective, rtt.adjusted.unwrap_or(rtt.raw));
assert_eq!(
rtt.effective,
rtt.adjusted
.unwrap_or_else(|| SignedDuration::from_duration(rtt.raw))
);
assert!(received_stats.is_some());
let stats = received_stats.as_ref().unwrap();
assert_eq!(stats.count, Some(42));
@@ -18,13 +18,27 @@ fn server_processing_greater_than_raw_does_not_underflow() {
..Default::default()
},
);
assert!(rtt.adjusted.is_none());
assert_eq!(rtt.effective, rtt.raw);
assert_eq!(
rtt.adjusted_signed,
Some(SignedDuration { ns: -999_999_999 })
assert_eq!(rtt.adjusted, Some(SignedDuration { ns: -999_999_999 }));
assert_eq!(rtt.effective, SignedDuration { ns: -999_999_999 });
}
#[test]
fn effective_rtt_uses_raw_when_adjusted_rtt_is_unavailable() {
let base = Instant::now();
let rtt = compute_rtt(
&ClientTimestamp {
mono: base,
wall: SystemTime::now(),
},
&ClientTimestamp {
mono: base + Duration::from_nanos(42),
wall: SystemTime::now(),
},
&TimestampFields::default(),
);
assert_eq!(rtt.effective_signed, SignedDuration { ns: -999_999_999 });
assert_eq!(rtt.adjusted, None);
assert_eq!(rtt.effective, SignedDuration::from_duration(rtt.raw));
}
#[test]
+22 -4
View File
@@ -106,11 +106,15 @@ pub enum WarningKind {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RttSample {
/// Client-observed RTT from send to receive.
pub raw: Duration,
pub adjusted: Option<Duration>,
pub effective: Duration,
pub adjusted_signed: Option<SignedDuration>,
pub effective_signed: SignedDuration,
/// Signed RTT adjusted for server processing when server processing is available.
pub adjusted: Option<SignedDuration>,
/// Signed effective RTT used by stats and CLI reporting.
///
/// This is `adjusted` when server processing is available, otherwise `raw`
/// converted to a signed duration.
pub effective: SignedDuration,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -118,6 +122,20 @@ pub struct SignedDuration {
pub ns: i128,
}
impl SignedDuration {
pub fn from_duration(duration: Duration) -> Self {
Self {
ns: i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX),
}
}
}
impl From<Duration> for SignedDuration {
fn from(value: Duration) -> Self {
Self::from_duration(value)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ServerTiming {
pub receive_wall_ns: Option<i64>,
+6 -3
View File
@@ -7,7 +7,7 @@ use std::time::Duration;
use irtt_client::{
Client, ClientConfig, ClientError, ClientEvent, OneWayDelaySample, ReceivedStatsSample,
RttSample, ServerTiming,
RttSample, ServerTiming, SignedDuration,
};
use irtt_proto::{Clock, Params, ReceivedStats, StampAt, TimestampFields};
@@ -69,7 +69,10 @@ fn timestamp_modes_drive_reply_event_timing_fields() {
assert!(reply.server_timing.is_none());
assert!(reply.one_way.is_none());
assert!(reply.rtt.adjusted.is_none());
assert_eq!(reply.rtt.effective, reply.rtt.raw);
assert_eq!(
reply.rtt.effective,
SignedDuration::from_duration(reply.rtt.raw)
);
assert!(reply.rtt.raw > Duration::ZERO);
}
StampAt::Send => {
@@ -349,7 +352,7 @@ fn assert_processing_subtracted(reply: ReplyView<'_>) {
assert_eq!(processing, Duration::from_nanos(1));
let adjusted = reply.rtt.adjusted.unwrap();
assert_eq!(reply.rtt.effective, adjusted);
assert!(adjusted < reply.rtt.raw);
assert!(adjusted.ns < SignedDuration::from_duration(reply.rtt.raw).ns);
}
fn assert_one_way_presence(
+16 -7
View File
@@ -69,8 +69,8 @@ impl CoreStats {
StatsEvent::UniqueReply {
is_late, sample, ..
} => self.apply_unique_reply(is_late, *sample),
StatsEvent::DuplicateReply { .. } => {
self.apply_duplicate_reply();
StatsEvent::DuplicateReply { bytes, .. } => {
self.apply_duplicate_reply(bytes);
EventStatsUpdate::default()
}
StatsEvent::Loss { .. } => {
@@ -81,8 +81,8 @@ impl CoreStats {
self.apply_warning();
EventStatsUpdate::default()
}
StatsEvent::UntrackedLate { .. } => {
self.apply_untracked_late();
StatsEvent::UntrackedLate { bytes, .. } => {
self.apply_untracked_late(bytes);
EventStatsUpdate::default()
}
}
@@ -121,7 +121,12 @@ impl CoreStats {
.saturating_add(sample.bytes as u64);
if let Some(count) = sample.received_count {
self.packets.server_packets_received = Some(u64::from(count));
let count = u64::from(count);
self.packets.server_packets_received = Some(
self.packets
.server_packets_received
.map_or(count, |current| current.max(count)),
);
}
if let Some(window) = sample.received_window {
self.packets.server_received_window = Some(window);
@@ -181,10 +186,11 @@ impl CoreStats {
updates
}
fn apply_duplicate_reply(&mut self) {
fn apply_duplicate_reply(&mut self, bytes: usize) {
self.events.duplicate_replies += 1;
self.packets.packets_received += 1;
self.packets.duplicates += 1;
self.packets.bytes_received = self.packets.bytes_received.saturating_add(bytes as u64);
}
fn apply_loss(&mut self) {
@@ -195,8 +201,11 @@ impl CoreStats {
self.events.warning_events += 1;
}
fn apply_untracked_late(&mut self) {
fn apply_untracked_late(&mut self, bytes: usize) {
self.events.untracked_late_replies += 1;
self.packets.packets_received += 1;
self.packets.late_packets += 1;
self.packets.bytes_received = self.packets.bytes_received.saturating_add(bytes as u64);
}
pub(crate) fn snapshot(&self) -> Snapshot {
+15 -5
View File
@@ -198,6 +198,9 @@ pub struct EventCounts {
/// Warning events processed.
pub warning_events: u64,
/// Late replies that could not be matched to retained in-flight state.
///
/// These replies are observed on the socket, but cannot contribute RTT,
/// one-way delay, server-processing, or IPDV samples.
pub untracked_late_replies: u64,
}
@@ -206,19 +209,26 @@ pub struct EventCounts {
pub struct PacketCounts {
/// Probe packets sent by the local client.
pub packets_sent: u64,
/// Reply packets received by the local client, including duplicates and late packets.
/// Reply packets observed by the local client.
///
/// This includes unique measurable replies, duplicate replies, and late
/// replies that cannot be matched to retained in-flight state.
pub packets_received: u64,
/// Unique replies received by the local client.
/// Unique replies that can contribute timing measurements.
pub unique_replies: u64,
/// Duplicate replies received by the local client.
pub duplicates: u64,
/// Late reply packets received by the local client.
/// Late reply packets received by the local client, including untracked-late replies.
pub late_packets: u64,
/// Probe bytes sent by the local client.
pub bytes_sent: u64,
/// Reply bytes received by the local client.
/// Reply bytes observed by the local client.
///
/// This includes bytes from unique measurable replies, duplicate replies,
/// and untracked-late replies. Per-category byte counts are not currently
/// exposed separately.
pub bytes_received: u64,
/// Server-reported packets received, when available.
/// Highest cumulative server-reported packets received, when available.
pub server_packets_received: Option<u64>,
/// Server-reported receive window, when available.
pub server_received_window: Option<u64>,
+14 -6
View File
@@ -22,6 +22,7 @@ pub(crate) enum StatsEvent {
},
DuplicateReply {
at: Instant,
bytes: usize,
},
Loss {
at: Instant,
@@ -31,6 +32,7 @@ pub(crate) enum StatsEvent {
},
UntrackedLate {
at: Instant,
bytes: usize,
},
}
@@ -96,11 +98,17 @@ pub(crate) fn normalize_event(event: &ClientEvent) -> Option<StatsEvent> {
*bytes,
)),
}),
ClientEvent::LateReply { received_at, .. } => Some(StatsEvent::UntrackedLate {
ClientEvent::LateReply {
received_at, bytes, ..
} => Some(StatsEvent::UntrackedLate {
at: received_at.mono,
bytes: *bytes,
}),
ClientEvent::DuplicateReply { received_at, .. } => Some(StatsEvent::DuplicateReply {
ClientEvent::DuplicateReply {
received_at, bytes, ..
} => Some(StatsEvent::DuplicateReply {
at: received_at.mono,
bytes: *bytes,
}),
ClientEvent::EchoLoss { timeout_at, .. } => Some(StatsEvent::Loss { at: *timeout_at }),
ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }),
@@ -115,10 +123,10 @@ impl StatsEvent {
match self {
Self::Sent { at, .. }
| Self::UniqueReply { at, .. }
| Self::DuplicateReply { at }
| Self::DuplicateReply { at, .. }
| Self::Loss { at }
| Self::Warning { at }
| Self::UntrackedLate { at } => *at,
| Self::UntrackedLate { at, .. } => *at,
}
}
}
@@ -148,11 +156,11 @@ impl ReplySample {
received_stats: Option<ReceivedStatsSample>,
bytes: usize,
) -> Self {
let rtt_primary_ns = signed_duration_ns(rtt.effective_signed);
let rtt_primary_ns = signed_duration_ns(rtt.effective);
Self {
bytes,
rtt_raw_ns: duration_ns_i128(rtt.raw),
rtt_adjusted_ns: rtt.adjusted_signed.map(signed_duration_ns),
rtt_adjusted_ns: rtt.adjusted.map(signed_duration_ns),
send_delay_ns: one_way
.and_then(|sample| sample.client_to_server)
.map(duration_ns_i128),
+5 -11
View File
@@ -13,19 +13,13 @@ pub fn ts(ms: u64) -> ClientTimestamp {
}
pub fn rtt(raw_ms: u64, effective_ms: i128) -> RttSample {
let effective = SignedDuration {
ns: effective_ms * 1_000_000,
};
RttSample {
raw: Duration::from_millis(raw_ms),
adjusted: u64::try_from(effective_ms).ok().map(Duration::from_millis),
effective: u64::try_from(effective_ms)
.ok()
.map(Duration::from_millis)
.unwrap_or_else(|| Duration::from_millis(raw_ms)),
adjusted_signed: Some(SignedDuration {
ns: effective_ms * 1_000_000,
}),
effective_signed: SignedDuration {
ns: effective_ms * 1_000_000,
},
adjusted: Some(effective),
effective,
}
}
+16 -1
View File
@@ -49,7 +49,7 @@ fn cumulative_rtt_uses_signed_effective_and_tracks_raw() {
}
#[test]
fn late_unique_counts_and_duplicates_do_not_update_measurements() {
fn late_unique_counts_and_duplicates_do_not_update_duplicate_measurements() {
let mut collector = StatsCollector::new(StatsConfig::finite());
collector.process(&sent(0, ts(0)));
collector.process(&sent(1, ts(10)));
@@ -68,6 +68,7 @@ fn late_unique_counts_and_duplicates_do_not_update_measurements() {
assert_eq!(snapshot.packets.unique_replies, 2);
assert_eq!(snapshot.packets.duplicates, 1);
assert_eq!(snapshot.packets.late_packets, 1);
assert_eq!(snapshot.packets.bytes_received, 64 + 64 + 64);
assert_eq!(snapshot.rtt.primary.count, 2);
assert_eq!(snapshot.loss.lost_packets, 0);
assert_eq!(snapshot.loss.duplicate_percent, 100.0 / 3.0);
@@ -210,6 +211,20 @@ fn server_processing_and_one_way_require_available_samples() {
assert_eq!(snapshot.server_processing.processing.count, 1);
assert_eq!(snapshot.one_way_delay.send_delay.count, 1);
assert_eq!(snapshot.events.untracked_late_replies, 1);
assert_eq!(snapshot.packets.packets_received, 2);
assert_eq!(snapshot.packets.unique_replies, 1);
assert_eq!(snapshot.packets.late_packets, 1);
assert_eq!(snapshot.packets.bytes_received, 128);
}
#[test]
fn older_cumulative_server_receive_count_does_not_regress() {
let mut collector = StatsCollector::new(StatsConfig::finite());
collector.process(&reply(1, 10, 10));
collector.process(&late_reply(0, 10, 10));
let snapshot = collector.snapshot();
assert_eq!(snapshot.packets.server_packets_received, Some(2));
}
#[test]