refactor(client): use SignedDuration for one-way delay to support negative values
Allow one-way delay samples to represent negative values caused by wall-clock skew between client and server. Update CLI output and statistics collection to correctly handle and report these signed durations.
This commit is contained in:
1 parent
0b16ee802d
commit
3b5336a487
7 files changed
+193
-19
No files matched your search
@@ -681,10 +681,10 @@ fn write_server_timing(out: &mut String, timing: Option<ServerTiming>) {
|
|||||||
fn write_one_way(out: &mut String, one_way: Option<OneWayDelaySample>) {
|
fn write_one_way(out: &mut String, one_way: Option<OneWayDelaySample>) {
|
||||||
if let Some(one_way) = one_way {
|
if let Some(one_way) = one_way {
|
||||||
if let Some(value) = one_way.client_to_server {
|
if let Some(value) = one_way.client_to_server {
|
||||||
write!(out, " client_to_server_us={}", duration_us(value)).unwrap();
|
write!(out, " client_to_server_us={}", signed_duration_us(value)).unwrap();
|
||||||
}
|
}
|
||||||
if let Some(value) = one_way.server_to_client {
|
if let Some(value) = one_way.server_to_client {
|
||||||
write!(out, " server_to_client_us={}", duration_us(value)).unwrap();
|
write!(out, " server_to_client_us={}", signed_duration_us(value)).unwrap();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -716,8 +716,8 @@ fn write_human_one_way(out: &mut String, one_way: Option<OneWayDelaySample>) {
|
|||||||
write!(
|
write!(
|
||||||
out,
|
out,
|
||||||
" rd={} sd={}",
|
" rd={} sd={}",
|
||||||
format_optional_duration(one_way.server_to_client),
|
format_optional_signed_duration(one_way.server_to_client),
|
||||||
format_optional_duration(one_way.client_to_server)
|
format_optional_signed_duration(one_way.client_to_server)
|
||||||
)
|
)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
}
|
}
|
||||||
@@ -777,9 +777,9 @@ fn signed_duration_us(duration: SignedDuration) -> i128 {
|
|||||||
duration.as_micros()
|
duration.as_micros()
|
||||||
}
|
}
|
||||||
|
|
||||||
fn format_optional_duration(duration: Option<Duration>) -> String {
|
fn format_optional_signed_duration(duration: Option<SignedDuration>) -> String {
|
||||||
duration
|
duration
|
||||||
.map(format_duration)
|
.map(format_signed_duration)
|
||||||
.unwrap_or_else(|| "n/a".to_owned())
|
.unwrap_or_else(|| "n/a".to_owned())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -882,8 +882,8 @@ mod tests {
|
|||||||
processing: Some(Duration::from_micros(300)),
|
processing: Some(Duration::from_micros(300)),
|
||||||
}),
|
}),
|
||||||
one_way: Some(OneWayDelaySample {
|
one_way: Some(OneWayDelaySample {
|
||||||
client_to_server: Some(Duration::from_micros(400)),
|
client_to_server: Some(SignedDuration::from_nanos(400_000)),
|
||||||
server_to_client: Some(Duration::from_micros(500)),
|
server_to_client: Some(SignedDuration::from_nanos(500_000)),
|
||||||
}),
|
}),
|
||||||
received_stats: Some(ReceivedStatsSample {
|
received_stats: Some(ReceivedStatsSample {
|
||||||
count: Some(9),
|
count: Some(9),
|
||||||
@@ -922,6 +922,40 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn negative_one_way_reply_event() -> ClientEvent {
|
||||||
|
let ClientEvent::EchoReply {
|
||||||
|
seq,
|
||||||
|
remote,
|
||||||
|
sent_at,
|
||||||
|
received_at,
|
||||||
|
rtt,
|
||||||
|
server_timing,
|
||||||
|
received_stats,
|
||||||
|
bytes,
|
||||||
|
packet_meta,
|
||||||
|
..
|
||||||
|
} = reply_event()
|
||||||
|
else {
|
||||||
|
unreachable!();
|
||||||
|
};
|
||||||
|
|
||||||
|
ClientEvent::EchoReply {
|
||||||
|
seq,
|
||||||
|
remote,
|
||||||
|
sent_at,
|
||||||
|
received_at,
|
||||||
|
rtt,
|
||||||
|
server_timing,
|
||||||
|
one_way: Some(OneWayDelaySample {
|
||||||
|
client_to_server: Some(SignedDuration::from_nanos(-400_000)),
|
||||||
|
server_to_client: Some(SignedDuration::from_nanos(-500_000)),
|
||||||
|
}),
|
||||||
|
received_stats,
|
||||||
|
bytes,
|
||||||
|
packet_meta,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn late_event_with_meta(packet_meta: PacketMeta) -> ClientEvent {
|
fn late_event_with_meta(packet_meta: PacketMeta) -> ClientEvent {
|
||||||
ClientEvent::LateReply {
|
ClientEvent::LateReply {
|
||||||
seq: 4,
|
seq: 4,
|
||||||
@@ -1238,6 +1272,14 @@ mod tests {
|
|||||||
assert!(line.contains("server_received_count=9"));
|
assert!(line.contains("server_received_count=9"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn machine_prints_negative_one_way_delay_values() {
|
||||||
|
let line = format_event(&negative_one_way_reply_event(), OutputMode::Machine).unwrap();
|
||||||
|
|
||||||
|
assert!(line.contains("client_to_server_us=-400"));
|
||||||
|
assert!(line.contains("server_to_client_us=-500"));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn machine_echo_reply_metadata_prints_observed_and_unavailable_values() {
|
fn machine_echo_reply_metadata_prints_observed_and_unavailable_values() {
|
||||||
let cases = [
|
let cases = [
|
||||||
@@ -1300,6 +1342,14 @@ mod tests {
|
|||||||
assert!(!OutputMode::RttUs.prints_summary());
|
assert!(!OutputMode::RttUs.prints_summary());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn human_prints_negative_one_way_delay_values() {
|
||||||
|
let line = format_event(&negative_one_way_reply_event(), OutputMode::Human).unwrap();
|
||||||
|
|
||||||
|
assert!(line.contains("rd=-500.0µs"));
|
||||||
|
assert!(line.contains("sd=-400.0µs"));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn verbose_human_reply_includes_extra_fields() {
|
fn verbose_human_reply_includes_extra_fields() {
|
||||||
let line = format_human_event_with_options(
|
let line = format_human_event_with_options(
|
||||||
|
|||||||
@@ -789,11 +789,11 @@ fn compute_one_way(
|
|||||||
let c2s = server_recv_wall
|
let c2s = server_recv_wall
|
||||||
.zip(client_send_ns)
|
.zip(client_send_ns)
|
||||||
.and_then(|(srv, cli)| srv.checked_sub(cli))
|
.and_then(|(srv, cli)| srv.checked_sub(cli))
|
||||||
.and_then(|d| u64::try_from(d).ok().map(Duration::from_nanos));
|
.map(|d| SignedDuration::from_nanos(i128::from(d)));
|
||||||
let s2c = client_recv_ns
|
let s2c = client_recv_ns
|
||||||
.zip(server_send_wall)
|
.zip(server_send_wall)
|
||||||
.and_then(|(cli, srv)| cli.checked_sub(srv))
|
.and_then(|(cli, srv)| cli.checked_sub(srv))
|
||||||
.and_then(|d| u64::try_from(d).ok().map(Duration::from_nanos));
|
.map(|d| SignedDuration::from_nanos(i128::from(d)));
|
||||||
|
|
||||||
if c2s.is_none() && s2c.is_none() {
|
if c2s.is_none() && s2c.is_none() {
|
||||||
return None;
|
return None;
|
||||||
|
|||||||
@@ -90,6 +90,68 @@ fn compute_one_way_returns_available_direction_samples() {
|
|||||||
};
|
};
|
||||||
|
|
||||||
let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap();
|
let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap();
|
||||||
assert_eq!(sample.client_to_server, Some(Duration::from_millis(15)));
|
assert_eq!(
|
||||||
assert_eq!(sample.server_to_client, Some(Duration::from_millis(15)));
|
sample.client_to_server,
|
||||||
|
Some(SignedDuration::from_nanos(15_000_000))
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
sample.server_to_client,
|
||||||
|
Some(SignedDuration::from_nanos(15_000_000))
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn compute_one_way_preserves_negative_client_to_server_delay() {
|
||||||
|
let sent_at = ClientTimestamp {
|
||||||
|
mono: Instant::now(),
|
||||||
|
wall: SystemTime::UNIX_EPOCH + Duration::from_secs(10),
|
||||||
|
};
|
||||||
|
let received_at = ClientTimestamp {
|
||||||
|
mono: sent_at.mono + Duration::from_millis(40),
|
||||||
|
wall: SystemTime::UNIX_EPOCH + Duration::from_secs(10) + Duration::from_millis(40),
|
||||||
|
};
|
||||||
|
let ts = TimestampFields {
|
||||||
|
recv_wall: Some(10_000_000_000 - 5_000_000),
|
||||||
|
send_wall: Some(10_000_000_000 + 25_000_000),
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
|
||||||
|
let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap();
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
sample.client_to_server,
|
||||||
|
Some(SignedDuration::from_nanos(-5_000_000))
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
sample.server_to_client,
|
||||||
|
Some(SignedDuration::from_nanos(15_000_000))
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn compute_one_way_preserves_negative_server_to_client_delay() {
|
||||||
|
let sent_at = ClientTimestamp {
|
||||||
|
mono: Instant::now(),
|
||||||
|
wall: SystemTime::UNIX_EPOCH + Duration::from_secs(10),
|
||||||
|
};
|
||||||
|
let received_at = ClientTimestamp {
|
||||||
|
mono: sent_at.mono + Duration::from_millis(40),
|
||||||
|
wall: SystemTime::UNIX_EPOCH + Duration::from_secs(10) + Duration::from_millis(40),
|
||||||
|
};
|
||||||
|
let ts = TimestampFields {
|
||||||
|
recv_wall: Some(10_000_000_000 + 15_000_000),
|
||||||
|
send_wall: Some(10_000_000_000 + 45_000_000),
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
|
||||||
|
let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap();
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
sample.client_to_server,
|
||||||
|
Some(SignedDuration::from_nanos(15_000_000))
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
sample.server_to_client,
|
||||||
|
Some(SignedDuration::from_nanos(-5_000_000))
|
||||||
|
);
|
||||||
}
|
}
|
||||||
@@ -168,9 +168,27 @@ pub struct ServerTiming {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
|
/// Directional one-way delay samples computed from wall-clock timestamps.
|
||||||
|
///
|
||||||
|
/// Each directional field is an `Option<SignedDuration>`. `None` means the
|
||||||
|
/// required client or server wall-clock timestamp was unavailable, or the
|
||||||
|
/// timestamp arithmetic was out of range. `Some(value)` means the delay was
|
||||||
|
/// computed from available timestamps.
|
||||||
|
///
|
||||||
|
/// Negative computed values are preserved as `Some` instead of being clamped to
|
||||||
|
/// zero or dropped. A negative one-way delay usually indicates wall-clock skew
|
||||||
|
/// between the client and server.
|
||||||
pub struct OneWayDelaySample {
|
pub struct OneWayDelaySample {
|
||||||
pub client_to_server: Option<Duration>,
|
/// Client-to-server delay.
|
||||||
pub server_to_client: Option<Duration>,
|
///
|
||||||
|
/// This is computed from the client send wall time and the server receive
|
||||||
|
/// wall time when both are available.
|
||||||
|
pub client_to_server: Option<SignedDuration>,
|
||||||
|
/// Server-to-client delay.
|
||||||
|
///
|
||||||
|
/// This is computed from the server send wall time and the client receive
|
||||||
|
/// wall time when both are available.
|
||||||
|
pub server_to_client: Option<SignedDuration>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
|
|||||||
@@ -163,10 +163,10 @@ impl ReplySample {
|
|||||||
rtt_adjusted_ns: rtt.adjusted.map(signed_duration_ns),
|
rtt_adjusted_ns: rtt.adjusted.map(signed_duration_ns),
|
||||||
send_delay_ns: one_way
|
send_delay_ns: one_way
|
||||||
.and_then(|sample| sample.client_to_server)
|
.and_then(|sample| sample.client_to_server)
|
||||||
.map(duration_ns_i128),
|
.map(signed_duration_ns),
|
||||||
receive_delay_ns: one_way
|
receive_delay_ns: one_way
|
||||||
.and_then(|sample| sample.server_to_client)
|
.and_then(|sample| sample.server_to_client)
|
||||||
.map(duration_ns_i128),
|
.map(signed_duration_ns),
|
||||||
server_processing_ns: server_timing
|
server_processing_ns: server_timing
|
||||||
.and_then(|timing| timing.processing)
|
.and_then(|timing| timing.processing)
|
||||||
.map(duration_ns_i128),
|
.map(duration_ns_i128),
|
||||||
|
|||||||
@@ -72,8 +72,8 @@ fn reply_with_rtt(seq: u32, raw_ms: u64, rtt: RttSample) -> ClientEvent {
|
|||||||
processing: Some(Duration::from_millis(1)),
|
processing: Some(Duration::from_millis(1)),
|
||||||
}),
|
}),
|
||||||
one_way: Some(OneWayDelaySample {
|
one_way: Some(OneWayDelaySample {
|
||||||
client_to_server: Some(Duration::from_millis(1)),
|
client_to_server: Some(SignedDuration::from_nanos(1_000_000)),
|
||||||
server_to_client: Some(Duration::from_millis(2)),
|
server_to_client: Some(SignedDuration::from_nanos(2_000_000)),
|
||||||
}),
|
}),
|
||||||
received_stats: Some(ReceivedStatsSample {
|
received_stats: Some(ReceivedStatsSample {
|
||||||
count: Some(seq + 1),
|
count: Some(seq + 1),
|
||||||
|
|||||||
@@ -3,7 +3,9 @@ mod common;
|
|||||||
use std::time::{Duration, Instant, UNIX_EPOCH};
|
use std::time::{Duration, Instant, UNIX_EPOCH};
|
||||||
|
|
||||||
use common::{adjusted_reply, sent, ts, unadjusted_late_reply, unadjusted_reply};
|
use common::{adjusted_reply, sent, ts, unadjusted_late_reply, unadjusted_reply};
|
||||||
use irtt_client::{ClientEvent, ClientTimestamp, PacketMeta, RttSample, SignedDuration};
|
use irtt_client::{
|
||||||
|
ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, SignedDuration,
|
||||||
|
};
|
||||||
use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig};
|
use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig};
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -280,6 +282,48 @@ fn server_processing_and_one_way_require_available_samples() {
|
|||||||
assert_eq!(snapshot.packets.bytes_received, 128);
|
assert_eq!(snapshot.packets.bytes_received, 128);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn negative_one_way_delay_samples_are_included_in_stats() {
|
||||||
|
let mut collector = StatsCollector::new(StatsConfig::finite());
|
||||||
|
let sent_at = ts(0);
|
||||||
|
let received_at = ClientTimestamp {
|
||||||
|
mono: sent_at.mono + Duration::from_millis(10),
|
||||||
|
wall: sent_at.wall + Duration::from_millis(10),
|
||||||
|
};
|
||||||
|
|
||||||
|
collector.process(&ClientEvent::EchoReply {
|
||||||
|
seq: 0,
|
||||||
|
remote: "127.0.0.1:2112".parse().unwrap(),
|
||||||
|
sent_at,
|
||||||
|
received_at,
|
||||||
|
rtt: RttSample {
|
||||||
|
raw: Duration::from_millis(10),
|
||||||
|
adjusted: None,
|
||||||
|
effective: SignedDuration::from_nanos(10_000_000),
|
||||||
|
},
|
||||||
|
server_timing: None,
|
||||||
|
one_way: Some(OneWayDelaySample {
|
||||||
|
client_to_server: Some(SignedDuration::from_nanos(-1_000_000)),
|
||||||
|
server_to_client: Some(SignedDuration::from_nanos(-2_000_000)),
|
||||||
|
}),
|
||||||
|
received_stats: None,
|
||||||
|
bytes: 64,
|
||||||
|
packet_meta: PacketMeta::default(),
|
||||||
|
});
|
||||||
|
|
||||||
|
let snapshot = collector.snapshot();
|
||||||
|
|
||||||
|
assert_eq!(snapshot.one_way_delay.send_delay.count, 1);
|
||||||
|
assert_eq!(snapshot.one_way_delay.send_delay.min_ns, Some(-1_000_000));
|
||||||
|
assert_eq!(snapshot.one_way_delay.send_delay.total_ns, -1_000_000);
|
||||||
|
assert_eq!(snapshot.one_way_delay.receive_delay.count, 1);
|
||||||
|
assert_eq!(
|
||||||
|
snapshot.one_way_delay.receive_delay.min_ns,
|
||||||
|
Some(-2_000_000)
|
||||||
|
);
|
||||||
|
assert_eq!(snapshot.one_way_delay.receive_delay.total_ns, -2_000_000);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn older_cumulative_server_receive_count_does_not_regress() {
|
fn older_cumulative_server_receive_count_does_not_regress() {
|
||||||
let mut collector = StatsCollector::new(StatsConfig::finite());
|
let mut collector = StatsCollector::new(StatsConfig::finite());
|
||||||
|
|||||||
Reference in new issue
Block a user