Lochnair-Labs/irtt-rs
2
0 Fork 0
Code Issues 1 Pull Requests 4 Actions 4 Packages Projects Releases 6 Wiki Activity

refactor: remove logical_seq from client events, session tracking and stats usage #11

Merged
Lochnair merged 1 commits from remove-logical-seq into main 2026-05-11 07:56:08 +00:00
Conversation 0 Commits 1 Files Changed 13
+117 -273
13 changed files with 117 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/irtt-cli/src/lib.rs
+30 -76
View File
@@ -326,8 +326,8 @@ pub struct HumanEventStats {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct HumanIpdvPair {
pub previous_logical_seq: u64,
pub current_logical_seq: u64,
pub previous_seq: u32,
pub current_seq: u32,
pub rtt_ipdv: Duration,
pub send_ipdv: Option<Duration>,
pub receive_ipdv: Option<Duration>,
@@ -352,8 +352,8 @@ impl From<irtt_stats::EventStatsUpdate> for HumanEventStats {
impl From<irtt_stats::IpdvPairUpdate> for HumanIpdvPair {
fn from(value: irtt_stats::IpdvPairUpdate) -> Self {
Self {
previous_logical_seq: value.previous_logical_seq,
current_logical_seq: value.current_logical_seq,
previous_seq: value.previous_seq,
current_seq: value.current_seq,
rtt_ipdv: value.rtt_ipdv,
send_ipdv: value.send_ipdv,
receive_ipdv: value.receive_ipdv,
@@ -385,7 +385,6 @@ pub fn format_human_event_with_options(
ClientEvent::EchoSent { .. } => String::new(),
ClientEvent::EchoReply {
seq,
logical_seq,
rtt,
server_timing,
one_way,
@@ -393,17 +392,9 @@ pub fn format_human_event_with_options(
..
} => {
let mut out = format!("seq={seq}");
if options.verbose {
write!(out, " logical_seq={logical_seq}").unwrap();
}
write!(out, " rtt={}", format_duration(rtt.effective)).unwrap();
write_human_one_way(&mut out, *one_way);
write!(
out,
" ipdv={}",
format_human_ipdv(stats, Some(*logical_seq))
)
.unwrap();
write!(out, " ipdv={}", format_human_ipdv(stats, *seq)).unwrap();
if let Some(processing) = server_timing.and_then(|timing| timing.processing) {
write!(out, " proc={}", format_duration(processing)).unwrap();
}
@@ -412,17 +403,14 @@ pub fn format_human_event_with_options(
}
out
}
ClientEvent::EchoLoss {
seq, logical_seq, ..
} => {
format!("loss seq={seq} logical_seq={logical_seq}")
ClientEvent::EchoLoss { seq, .. } => {
format!("loss seq={seq}")
}
ClientEvent::DuplicateReply { seq, remote, .. } => {
format!("duplicate seq={seq} remote={remote}")
}
ClientEvent::LateReply {
seq,
logical_seq,
highest_seen,
remote,
rtt,
@@ -430,14 +418,11 @@ pub fn format_human_event_with_options(
received_stats,
..
} => {
let mut out = format!(
"late seq={seq} logical_seq={} highest_seen={highest_seen} remote={remote}",
optional_u64(*logical_seq)
);
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_human_one_way(&mut out, *one_way);
write!(out, " ipdv={}", format_human_ipdv(stats, *logical_seq)).unwrap();
write!(out, " ipdv={}", format_human_ipdv(stats, *seq)).unwrap();
}
write_human_received_stats(&mut out, *received_stats);
out
@@ -489,7 +474,6 @@ fn format_machine(event: &ClientEvent) -> String {
ClientEvent::EchoSent { .. } => {}
ClientEvent::EchoReply {
seq,
logical_seq,
remote,
sent_at,
received_at,
@@ -501,7 +485,7 @@ fn format_machine(event: &ClientEvent) -> String {
packet_meta,
} => {
write_common(&mut out, "echo_reply");
write_seq(&mut out, *seq, Some(*logical_seq));
write_seq(&mut out, *seq);
write_remote(&mut out, *remote);
write_wall(&mut out, "client_send_wall_ns", sent_at.wall);
write_wall(&mut out, "client_receive_wall_ns", received_at.wall);
@@ -511,14 +495,9 @@ fn format_machine(event: &ClientEvent) -> String {
write_received_stats(&mut out, *received_stats);
write_packet_meta(&mut out, *packet_meta);
}
ClientEvent::EchoLoss {
seq,
logical_seq,
sent_at,
..
} => {
ClientEvent::EchoLoss { seq, sent_at, .. } => {
write_common(&mut out, "loss");
write_seq(&mut out, *seq, Some(*logical_seq));
write_seq(&mut out, *seq);
write_wall(&mut out, "client_send_wall_ns", sent_at.wall);
out.push_str(" warning=loss");
}
@@ -529,14 +508,13 @@ fn format_machine(event: &ClientEvent) -> String {
bytes: _,
} => {
write_common(&mut out, "duplicate");
write_seq(&mut out, *seq, None);
write_seq(&mut out, *seq);
write_remote(&mut out, *remote);
write_wall(&mut out, "client_receive_wall_ns", received_at.wall);
out.push_str(" warning=duplicate");
}
ClientEvent::LateReply {
seq,
logical_seq,
highest_seen,
remote,
sent_at,
@@ -549,7 +527,7 @@ fn format_machine(event: &ClientEvent) -> String {
packet_meta,
} => {
write_common(&mut out, "late");
write_seq(&mut out, *seq, *logical_seq);
write_seq(&mut out, *seq);
write_remote(&mut out, *remote);
write!(out, " highest_seen={highest_seen}").unwrap();
if let Some(sent_at) = sent_at {
@@ -594,7 +572,6 @@ fn format_simple(event: &ClientEvent) -> String {
ClientEvent::EchoSent { .. } => String::new(),
ClientEvent::EchoReply {
seq,
logical_seq,
remote,
rtt,
server_timing,
@@ -602,7 +579,7 @@ fn format_simple(event: &ClientEvent) -> String {
..
} => {
let mut out = format!(
"reply seq={seq} logical_seq={logical_seq} remote={remote} rtt_us={}",
"reply seq={seq} remote={remote} rtt_us={}",
duration_us(rtt.effective)
);
if let Some(raw) = rtt.adjusted.map(|_| rtt.raw) {
@@ -613,26 +590,20 @@ fn format_simple(event: &ClientEvent) -> String {
}
out
}
ClientEvent::EchoLoss {
seq, logical_seq, ..
} => {
format!("loss seq={seq} logical_seq={logical_seq}")
ClientEvent::EchoLoss { seq, .. } => {
format!("loss seq={seq}")
}
ClientEvent::DuplicateReply { seq, remote, .. } => {
format!("duplicate seq={seq} remote={remote}")
}
ClientEvent::LateReply {
seq,
logical_seq,
highest_seen,
remote,
rtt,
..
} => {
let mut out = format!(
"late seq={seq} logical_seq={} highest_seen={highest_seen} remote={remote}",
optional_u64(*logical_seq)
);
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();
}
@@ -648,11 +619,8 @@ fn write_common(out: &mut String, event: &str) {
write!(out, "event={event}").unwrap();
}
fn write_seq(out: &mut String, seq: u32, logical_seq: Option<u64>) {
fn write_seq(out: &mut String, seq: u32) {
write!(out, " seq={seq}").unwrap();
if let Some(logical_seq) = logical_seq {
write!(out, " logical_seq={logical_seq}").unwrap();
}
}
fn write_remote(out: &mut String, remote: SocketAddr) {
@@ -763,19 +731,21 @@ fn write_human_received_stats(out: &mut String, stats: Option<ReceivedStatsSampl
}
}
fn format_human_ipdv(stats: Option<&HumanEventStats>, logical_seq: Option<u64>) -> String {
fn format_human_ipdv(stats: Option<&HumanEventStats>, seq: u32) -> String {
let Some(stats) = stats else {
return "n/a".to_owned();
};
let pair = logical_seq
.and_then(|seq| {
let pair = stats
.ipdv_pairs
.iter()
.find(|pair| pair.current_seq == seq)
.or_else(|| {
stats
.ipdv_pairs
.iter()
.find(|pair| pair.current_logical_seq == seq)
})
.or_else(|| stats.ipdv_pairs.last());
.find(|pair| pair.previous_seq == seq)
});
pair.map(|pair| format_duration(pair.rtt_ipdv))
.unwrap_or_else(|| "n/a".to_owned())
@@ -820,12 +790,6 @@ fn format_ns(ns: f64) -> String {
}
}
fn optional_u64(value: Option<u64>) -> String {
value
.map(|value| value.to_string())
.unwrap_or_else(|| "none".to_owned())
}
fn warning_kind(kind: WarningKind) -> &'static str {
match kind {
WarningKind::MalformedOrUnrelatedPacket => "malformed_or_unrelated_packet",
@@ -879,7 +843,6 @@ mod tests {
fn reply_event_with_meta(packet_meta: PacketMeta) -> ClientEvent {
ClientEvent::EchoReply {
seq: 7,
logical_seq: 8,
remote: test_remote(),
sent_at: test_timestamp(Duration::from_secs(1)),
received_at: test_timestamp(Duration::from_secs(1) + Duration::from_micros(1500)),
@@ -915,7 +878,6 @@ mod tests {
fn late_event_with_meta(packet_meta: PacketMeta) -> ClientEvent {
ClientEvent::LateReply {
seq: 4,
logical_seq: None,
highest_seen: 9,
remote: test_remote(),
sent_at: None,
@@ -1220,7 +1182,6 @@ mod tests {
let line = format_event(&reply_event(), OutputMode::Machine).unwrap();
assert!(line.starts_with("event=echo_reply "));
assert!(line.contains("seq=7"));
assert!(line.contains("logical_seq=8"));
assert!(line.contains("remote=127.0.0.1:2112"));
assert!(line.contains("client_send_wall_ns=1000000000"));
assert!(line.contains("client_receive_wall_ns=1001500000"));
@@ -1329,7 +1290,7 @@ mod tests {
let line = format_event(&reply_event(), OutputMode::Simple).unwrap();
assert_eq!(
line,
"reply seq=7 logical_seq=8 remote=127.0.0.1:2112 rtt_us=1200 raw_rtt_us=1500 server_processing_us=300"
"reply seq=7 remote=127.0.0.1:2112 rtt_us=1200 raw_rtt_us=1500 server_processing_us=300"
);
}
@@ -1341,7 +1302,6 @@ mod tests {
assert!(line.contains("sd=400.0µs"));
assert!(line.contains("ipdv=n/a"));
assert!(line.contains("proc=300.0µs"));
assert!(!line.contains("logical_seq="));
assert!(!line.contains("server_received="));
assert!(!line.contains("server_window="));
assert!(!line.contains("rtt_us="));
@@ -1363,7 +1323,6 @@ mod tests {
HumanOutputOptions { verbose: true },
);
assert!(line.contains("logical_seq=8"));
assert!(line.contains("server_received=9"));
assert!(line.contains("server_window=0x7"));
}
@@ -1375,8 +1334,8 @@ mod tests {
Some(HumanEventStats {
contributed_sample: true,
ipdv_pairs: vec![HumanIpdvPair {
previous_logical_seq: 7,
current_logical_seq: 8,
previous_seq: 7,
current_seq: 8,
rtt_ipdv: Duration::from_micros(47),
send_ipdv: None,
receive_ipdv: None,
@@ -1391,7 +1350,6 @@ mod tests {
fn human_reply_marks_missing_one_way_delay_unavailable() {
let ClientEvent::EchoReply {
seq,
logical_seq,
remote,
sent_at,
received_at,
@@ -1407,7 +1365,6 @@ mod tests {
};
let event = ClientEvent::EchoReply {
seq,
logical_seq,
remote,
sent_at,
received_at,
@@ -1430,7 +1387,6 @@ mod tests {
let ts = test_timestamp(Duration::from_secs(1));
let event = ClientEvent::EchoSent {
seq: 1,
logical_seq: 2,
remote: test_remote(),
scheduled_at: ts.mono,
sent_at: ts,
@@ -1451,7 +1407,6 @@ mod tests {
let events = [
ClientEvent::EchoLoss {
seq: 1,
logical_seq: 2,
sent_at: ts,
timeout_at: ts.mono + Duration::from_secs(1),
},
@@ -1463,7 +1418,6 @@ mod tests {
},
ClientEvent::LateReply {
seq: 4,
logical_seq: None,
highest_seen: 9,
remote: test_remote(),
sent_at: None,
crates/irtt-cli/src/main.rs
+7 -11
View File
@@ -313,15 +313,14 @@ mod tests {
SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 2112)
}
fn reply_event(logical_seq: u64, rtt_us: u64) -> ClientEvent {
let sent_at = test_timestamp(Duration::from_millis(logical_seq));
fn reply_event(seq: u32, rtt_us: u64) -> ClientEvent {
let sent_at = test_timestamp(Duration::from_millis(seq as u64));
let received_at = ClientTimestamp {
wall: sent_at.wall + Duration::from_micros(rtt_us),
mono: sent_at.mono + Duration::from_micros(rtt_us),
};
ClientEvent::EchoReply {
seq: logical_seq as u32,
logical_seq,
seq,
remote: test_remote(),
sent_at,
received_at,
@@ -349,7 +348,6 @@ mod tests {
let events = [
ClientEvent::EchoSent {
seq: 1,
logical_seq: 1,
remote: test_remote(),
scheduled_at: sent_at.mono,
sent_at,
@@ -359,7 +357,6 @@ mod tests {
},
ClientEvent::EchoReply {
seq: 1,
logical_seq: 1,
remote: test_remote(),
sent_at,
received_at,
@@ -442,12 +439,11 @@ mod tests {
#[test]
fn continuous_mode_uses_continuous_stats_config() {
let mut collector = StatsCollector::new(stats_config(true));
for seq in 0..5000 {
let sent_at = test_timestamp(Duration::from_micros(seq));
let received_at = test_timestamp(Duration::from_micros(seq + 1));
for seq in 0..5000_u32 {
let sent_at = test_timestamp(Duration::from_micros(seq as u64));
let received_at = test_timestamp(Duration::from_micros(seq as u64 + 1));
collector.process(&ClientEvent::EchoReply {
seq: seq as u32,
logical_seq: seq,
seq,
remote: test_remote(),
sent_at,
received_at,
crates/irtt-cli/src/summary.rs
-3
View File
@@ -229,7 +229,6 @@ mod tests {
collector.process(&ClientEvent::EchoSent {
seq: 1,
logical_seq: 1,
remote: test_remote(),
scheduled_at: sent_at.mono,
sent_at,
@@ -239,7 +238,6 @@ mod tests {
});
collector.process(&ClientEvent::EchoReply {
seq: 1,
logical_seq: 1,
remote: test_remote(),
sent_at,
received_at,
@@ -287,7 +285,6 @@ mod tests {
let received_at = test_timestamp(Duration::from_secs(1) + Duration::from_micros(1500));
collector.process(&ClientEvent::EchoReply {
seq: 1,
logical_seq: 1,
remote: test_remote(),
sent_at: test_timestamp(Duration::from_secs(1)),
received_at,
crates/irtt-client/src/client.rs
+2 -18
View File
@@ -210,7 +210,6 @@ impl Client {
.expect("negotiated must exist when Open");
let wire_seq = session.next_wire_seq;
let logical_seq = session.next_logical_seq;
let scheduled_at = session.next_send_at;
let request = EchoRequest {
@@ -232,7 +231,6 @@ impl Client {
.expect("session must exist when phase is Open");
let pending = PendingProbe {
logical_seq,
wire_seq,
sent_at,
timeout_at: sent_at
@@ -243,13 +241,6 @@ impl Client {
session.pending.insert(pending)?;
session.next_wire_seq = session.next_wire_seq.wrapping_add(1);
session.next_logical_seq =
session
.next_logical_seq
.checked_add(1)
.ok_or(ClientError::CounterOverflow {
counter: "next_logical_seq",
})?;
session.packets_sent =
session
.packets_sent
@@ -279,7 +270,6 @@ impl Client {
Ok(vec![ClientEvent::EchoSent {
seq: wire_seq,
logical_seq,
remote: self.remote,
scheduled_at,
sent_at,
@@ -372,7 +362,6 @@ impl Client {
for probe in expired {
events.push(ClientEvent::EchoLoss {
seq: probe.wire_seq,
logical_seq: probe.logical_seq,
sent_at: probe.sent_at,
timeout_at: probe.timeout_at,
});
@@ -468,13 +457,12 @@ impl Client {
let highest_seen = session.highest_received_seq.unwrap_or(wire_seq);
update_highest_received(session, wire_seq);
session.completed.insert(wire_seq, pending.logical_seq);
session.completed.insert(wire_seq);
let mut events = Vec::new();
if is_late {
events.push(ClientEvent::LateReply {
seq: wire_seq,
logical_seq: Some(pending.logical_seq),
highest_seen,
remote: self.remote,
sent_at: Some(pending.sent_at),
@@ -489,7 +477,6 @@ impl Client {
} else {
events.push(ClientEvent::EchoReply {
seq: wire_seq,
logical_seq: pending.logical_seq,
remote: self.remote,
sent_at: pending.sent_at,
received_at: now,
@@ -520,11 +507,10 @@ impl Client {
let received_stats = build_received_stats(&reply);
let highest_seen = session.highest_received_seq.unwrap_or(wire_seq);
update_highest_received(session, wire_seq);
session.completed.insert(wire_seq, timed_out.logical_seq);
session.completed.insert(wire_seq);
let mut events = vec![ClientEvent::LateReply {
seq: wire_seq,
logical_seq: Some(timed_out.logical_seq),
highest_seen,
remote: self.remote,
sent_at: Some(timed_out.sent_at),
@@ -546,7 +532,6 @@ impl Client {
{
Ok(vec![ClientEvent::LateReply {
seq: wire_seq,
logical_seq: None,
highest_seen: session.highest_received_seq.unwrap(),
remote: self.remote,
sent_at: None,
@@ -648,7 +633,6 @@ impl Client {
self.session = Some(ActiveSession {
next_wire_seq: 0,
next_logical_seq: 0,
highest_received_seq: None,
packets_sent: 0,
start_mono: now.mono,
crates/irtt-client/src/client/tests/probes_replies.rs
+2 -43
View File
@@ -43,14 +43,9 @@ fn send_probe_sends_valid_echo_request() {
assert_eq!(events.len(), 1);
match &events[0] {
ClientEvent::EchoSent {
seq,
logical_seq,
remote,
bytes,
..
seq, remote, bytes, ..
} => {
assert_eq!(*seq, 0);
assert_eq!(*logical_seq, 0);
assert_eq!(*remote, server.addr);
assert_eq!(*bytes, echo_packet_len(false, &params));
}
@@ -311,30 +306,6 @@ fn send_probe_reports_schedule_overflow() {
server.join();
}
#[test]
fn send_probe_reports_logical_sequence_counter_overflow() {
let params = default_params();
let server = silent_open_server(params);
let config = ClientConfig {
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(200)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.session.as_mut().unwrap().next_logical_seq = u64::MAX;
assert!(matches!(
client.send_probe(),
Err(ClientError::CounterOverflow {
counter: "next_logical_seq"
})
));
server.join();
}
#[test]
fn send_probe_reports_packets_sent_counter_overflow() {
let params = default_params();
@@ -390,7 +361,6 @@ fn recv_once_decodes_echo_reply_and_emits_event() {
match &events[0] {
ClientEvent::EchoReply {
seq,
logical_seq,
rtt,
received_stats,
server_timing,
@@ -399,7 +369,6 @@ fn recv_once_decodes_echo_reply_and_emits_event() {
..
} => {
assert_eq!(*seq, 0);
assert_eq!(*logical_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));
@@ -487,14 +456,8 @@ fn recv_once_decodes_only_received_bytes_after_longer_datagram() {
let events = client.recv_once().unwrap();
assert_eq!(events.len(), 1);
match &events[0] {
ClientEvent::EchoReply {
seq,
logical_seq,
bytes,
..
} => {
ClientEvent::EchoReply { seq, bytes, .. } => {
assert_eq!(*seq, 1);
assert_eq!(*logical_seq, 1);
assert_eq!(*bytes, echo_packet_len(false, &params));
}
other => panic!("expected EchoReply after shorter datagram, got {other:?}"),
@@ -980,7 +943,6 @@ fn late_reply_after_timeout_preserves_measurement_metadata() {
match &late[0] {
ClientEvent::LateReply {
seq,
logical_seq,
sent_at,
rtt,
server_timing,
@@ -990,7 +952,6 @@ fn late_reply_after_timeout_preserves_measurement_metadata() {
..
} => {
assert_eq!(*seq, 0);
assert_eq!(*logical_seq, Some(0));
assert!(sent_at.is_some());
assert!(rtt.is_some());
assert!(server_timing.is_some());
@@ -1758,7 +1719,6 @@ fn send_probe_wraps_wire_sequence_at_u32_max() {
let session = client.session.as_mut().unwrap();
session.next_wire_seq = u32::MAX;
session.next_logical_seq = 41;
client.send_probe().unwrap();
client.send_probe().unwrap();
@@ -1814,7 +1774,6 @@ fn wrapped_reply_after_u32_max_is_not_late() {
assert_open_started(client.open().unwrap());
let session = client.session.as_mut().unwrap();
session.next_wire_seq = u32::MAX;
session.next_logical_seq = 41;
client.send_probe().unwrap();
client.send_probe().unwrap();
crates/irtt-client/src/client/tests/sequence.rs
-1
View File
@@ -26,7 +26,6 @@ fn sequence_ordering_handles_normal_and_wrapped_values() {
fn highest_received_seq_updates_across_wrap() {
let mut session = ActiveSession {
next_wire_seq: 0,
next_logical_seq: 0,
highest_received_seq: Some(u32::MAX),
packets_sent: 0,
start_mono: Instant::now(),
crates/irtt-client/src/event.rs
-4
View File
@@ -41,7 +41,6 @@ pub enum ClientEvent {
EchoSent {
seq: u32,
logical_seq: u64,
remote: SocketAddr,
scheduled_at: Instant,
sent_at: ClientTimestamp,
@@ -52,7 +51,6 @@ pub enum ClientEvent {
EchoReply {
seq: u32,
logical_seq: u64,
remote: SocketAddr,
sent_at: ClientTimestamp,
received_at: ClientTimestamp,
@@ -66,7 +64,6 @@ pub enum ClientEvent {
EchoLoss {
seq: u32,
logical_seq: u64,
sent_at: ClientTimestamp,
timeout_at: std::time::Instant,
},
@@ -80,7 +77,6 @@ pub enum ClientEvent {
LateReply {
seq: u32,
logical_seq: Option<u64>,
highest_seen: u32,
remote: SocketAddr,
sent_at: Option<ClientTimestamp>,
crates/irtt-client/src/managed/runner.rs
+3 -9
View File
@@ -594,19 +594,13 @@ mod tests {
server.join();
assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
assert!(events.iter().any(|event| matches!(
event,
ClientEvent::EchoLoss {
seq: 0,
logical_seq: 0,
..
}
)));
assert!(events
.iter()
.any(|event| matches!(event, ClientEvent::EchoLoss { seq: 0, .. })));
assert!(events.iter().any(|event| matches!(
event,
ClientEvent::LateReply {
seq: 0,
logical_seq: Some(0),
sent_at: Some(_),
rtt: Some(_),
..
crates/irtt-client/src/probe.rs
+27 -44
View File
@@ -1,5 +1,5 @@
use std::{
collections::{HashMap, VecDeque},
collections::{HashMap, HashSet, VecDeque},
time::Instant,
};
@@ -7,7 +7,6 @@ use crate::{error::ClientError, timing::ClientTimestamp};
#[derive(Debug, Clone)]
pub(crate) struct PendingProbe {
pub logical_seq: u64,
pub wire_seq: u32,
pub sent_at: ClientTimestamp,
pub timeout_at: Instant,
@@ -59,7 +58,7 @@ impl PendingMap {
expired.push(probe);
}
}
expired.sort_by_key(|p| p.logical_seq);
expired.sort_by_key(|p| p.sent_at.mono);
expired
}
@@ -134,7 +133,7 @@ impl TimedOutMap {
#[derive(Debug)]
pub(crate) struct CompletedSet {
set: HashMap<u32, u64>,
set: HashSet<u32>,
insertion_order: VecDeque<u32>,
max_capacity: usize,
}
@@ -142,31 +141,32 @@ pub(crate) struct CompletedSet {
impl CompletedSet {
pub fn new(max_capacity: usize) -> Self {
Self {
set: HashMap::new(),
set: HashSet::new(),
insertion_order: VecDeque::new(),
max_capacity,
}
}
pub fn insert(&mut self, wire_seq: u32, logical_seq: u64) {
if let Some(existing) = self.set.get_mut(&wire_seq) {
*existing = logical_seq;
pub fn insert(&mut self, seq: u32) {
if self.set.contains(&seq) {
return;
}
if self.set.len() >= self.max_capacity {
self.evict_oldest();
}
self.insertion_order.push_back(wire_seq);
self.set.insert(wire_seq, logical_seq);
self.insertion_order.push_back(seq);
self.set.insert(seq);
}
pub fn contains(&self, wire_seq: u32) -> bool {
self.set.contains_key(&wire_seq)
pub fn contains(&self, seq: u32) -> bool {
self.set.contains(&seq)
}
fn evict_oldest(&mut self) {
while let Some(oldest_key) = self.insertion_order.pop_front() {
if self.set.remove(&oldest_key).is_some() {
while let Some(oldest_seq) = self.insertion_order.pop_front() {
if self.set.remove(&oldest_seq) {
break;
}
}
@@ -185,9 +185,8 @@ mod tests {
}
}
fn pending(seq: u32, logical: u64, timeout_at: Instant) -> PendingProbe {
fn pending(seq: u32, timeout_at: Instant) -> PendingProbe {
PendingProbe {
logical_seq: logical,
wire_seq: seq,
sent_at: ts(timeout_at - Duration::from_secs(1)),
timeout_at,
@@ -198,7 +197,7 @@ mod tests {
fn pending_map_insert_and_remove() {
let mut map = PendingMap::new(10);
let now = Instant::now();
let probe = pending(0, 0, now + Duration::from_secs(4));
let probe = pending(0, now + Duration::from_secs(4));
map.insert(probe).unwrap();
assert_eq!(map.len(), 1);
assert!(map.remove(0).is_some());
@@ -210,42 +209,26 @@ mod tests {
fn pending_map_rejects_over_capacity() {
let mut map = PendingMap::new(2);
let now = Instant::now();
map.insert(pending(0, 0, now + Duration::from_secs(4)))
map.insert(pending(0, now + Duration::from_secs(4)))
.unwrap();
map.insert(pending(1, 1, now + Duration::from_secs(4)))
map.insert(pending(1, now + Duration::from_secs(4)))
.unwrap();
assert!(matches!(
map.insert(pending(2, 2, now + Duration::from_secs(4))),
map.insert(pending(2, now + Duration::from_secs(4))),
Err(ClientError::PendingLimitExceeded { limit: 2 })
));
}
#[test]
fn drain_expired_returns_in_logical_order() {
let mut map = PendingMap::new(10);
let now = Instant::now();
let past = now - Duration::from_secs(1);
map.insert(pending(5, 5, past)).unwrap();
map.insert(pending(2, 2, past)).unwrap();
map.insert(pending(8, 8, now + Duration::from_secs(10)))
.unwrap();
let expired = map.drain_expired(now);
assert_eq!(expired.len(), 2);
assert_eq!(expired[0].wire_seq, 2);
assert_eq!(expired[1].wire_seq, 5);
assert_eq!(map.len(), 1);
}
#[test]
fn completed_set_tracks_and_evicts() {
let mut set = CompletedSet::new(3);
set.insert(0, 0);
set.insert(1, 1);
set.insert(2, 2);
set.insert(0);
set.insert(1);
set.insert(2);
assert!(set.contains(0));
assert!(set.contains(1));
assert!(set.contains(2));
set.insert(3, 3);
set.insert(3);
assert_eq!(set.set.len(), 3);
assert!(!set.contains(0));
assert!(set.contains(3));
@@ -255,9 +238,9 @@ mod tests {
fn timed_out_map_tracks_and_evicts() {
let mut map = TimedOutMap::new(2);
let now = Instant::now();
map.insert(pending(0, 0, now));
map.insert(pending(1, 1, now));
map.insert(pending(2, 2, now));
map.insert(pending(0, now));
map.insert(pending(1, now));
map.insert(pending(2, now));
assert_eq!(map.len(), 2);
assert!(map.remove(0).is_none());
@@ -271,7 +254,7 @@ mod tests {
let now = Instant::now();
for i in 0..20 {
map.insert(pending(i, u64::from(i), now));
map.insert(pending(i, now));
assert!(map.remove(i).is_some());
assert_eq!(map.len(), 0);
assert_eq!(map.insertion_order_len(), 0);
crates/irtt-client/src/session.rs
-1
View File
@@ -24,7 +24,6 @@ pub(crate) enum ClientPhase {
#[derive(Debug)]
pub(crate) struct ActiveSession {
pub next_wire_seq: u32,
pub next_logical_seq: u64,
pub highest_received_seq: Option<u32>,
pub packets_sent: u64,
pub start_mono: Instant,
crates/irtt-client/tests/hmac.rs
-2
View File
@@ -73,13 +73,11 @@ fn hmac_echo_success_verifies_request_and_accepts_reply() {
match &run.reply {
ClientEvent::EchoReply {
seq,
logical_seq,
bytes,
server_timing,
..
} => {
assert_eq!(*seq, 0);
assert_eq!(*logical_seq, 0);
assert_eq!(*bytes, echo_packet_len(true, &params));
assert!(server_timing.is_some());
}
crates/irtt-client/tests/protocol_options.rs
+1 -11
View File
@@ -16,7 +16,6 @@ use support::{
struct ReplyView<'a> {
seq: u32,
logical_seq: u64,
rtt: &'a RttSample,
server_timing: Option<&'a ServerTiming>,
one_way: Option<&'a OneWayDelaySample>,
@@ -38,7 +37,6 @@ fn received_stats_modes_drive_negotiated_echo_and_reply_events() {
let reply = expect_echo_reply(&run.reply);
assert_eq!(reply.seq, 0);
assert_eq!(reply.logical_seq, 0);
assert_eq!(reply.bytes, echo_packet_len(false, &params));
assert!(reply.server_timing.is_none());
assert!(reply.one_way.is_none());
@@ -251,14 +249,8 @@ fn assert_echo_uses_params(run: &OneProbeRun, params: &Params, hmac: bool) {
assert_eq!(sequence, 0);
match &run.sent {
ClientEvent::EchoSent {
seq,
logical_seq,
bytes,
..
} => {
ClientEvent::EchoSent { seq, bytes, .. } => {
assert_eq!(*seq, 0);
assert_eq!(*logical_seq, 0);
assert_eq!(*bytes, echo_packet_len(hmac, params));
}
other => panic!("expected EchoSent, got {other:?}"),
@@ -288,7 +280,6 @@ fn expect_echo_reply(event: &ClientEvent) -> ReplyView<'_> {
match event {
ClientEvent::EchoReply {
seq,
logical_seq,
rtt,
server_timing,
one_way,
@@ -297,7 +288,6 @@ fn expect_echo_reply(event: &ClientEvent) -> ReplyView<'_> {
..
} => ReplyView {
seq: *seq,
logical_seq: *logical_seq,
rtt,
server_timing: server_timing.as_ref(),
one_way: one_way.as_ref(),
crates/irtt-stats/src/lib.rs
+45 -50
View File
@@ -142,8 +142,8 @@ pub struct EventStatsUpdate {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IpdvPairUpdate {
pub previous_logical_seq: u64,
pub current_logical_seq: u64,
pub previous_seq: u32,
pub current_seq: u32,
pub rtt_ipdv: Duration,
pub send_ipdv: Option<Duration>,
pub receive_ipdv: Option<Duration>,
@@ -297,9 +297,9 @@ struct CoreStats {
send_delay: MetricU64,
receive_delay: MetricU64,
server_processing: MetricU64,
samples: HashMap<u64, UniqueSample>,
sample_order: VecDeque<u64>,
ipdv_pairs: HashSet<u64>,
samples: HashMap<u32, UniqueSample>,
sample_order: VecDeque<u32>,
ipdv_pairs: HashSet<u32>,
}
impl CoreStats {
@@ -381,7 +381,7 @@ impl CoreStats {
self.receive_delay.push(delay);
}
let seq = sample.logical_seq;
let seq = sample.seq;
if self.samples.insert(seq, sample).is_none() {
self.sample_order.push_back(seq);
self.enforce_sequence_limit();
@@ -430,7 +430,7 @@ impl CoreStats {
}
}
fn try_ipdv_pair(&mut self, current_seq: u64) -> Option<IpdvPairUpdate> {
fn try_ipdv_pair(&mut self, current_seq: u32) -> Option<IpdvPairUpdate> {
let previous_seq = current_seq.checked_sub(1)?;
if !self.ipdv_pairs.insert(current_seq) {
@@ -464,8 +464,8 @@ impl CoreStats {
}
Some(IpdvPairUpdate {
previous_logical_seq: previous_seq,
current_logical_seq: current_seq,
previous_seq,
current_seq,
rtt_ipdv: Duration::from_nanos(rtt_ipdv),
send_ipdv: send_ipdv.map(Duration::from_nanos),
receive_ipdv: receive_ipdv.map(Duration::from_nanos),
@@ -561,7 +561,7 @@ impl WindowEvent {
timer_error_ns: duration_ns_u64(*timer_error),
}),
ClientEvent::EchoReply {
logical_seq,
seq,
sent_at,
received_at,
rtt,
@@ -574,7 +574,7 @@ impl WindowEvent {
at: received_at.mono,
is_late: false,
sample: Box::new(UniqueSample::new(
*logical_seq,
*seq,
*sent_at,
*received_at,
*rtt,
@@ -585,7 +585,7 @@ impl WindowEvent {
)),
}),
ClientEvent::LateReply {
logical_seq: Some(logical_seq),
seq,
sent_at: Some(sent_at),
received_at,
rtt: Some(rtt),
@@ -598,7 +598,7 @@ impl WindowEvent {
at: received_at.mono,
is_late: true,
sample: Box::new(UniqueSample::new(
*logical_seq,
*seq,
*sent_at,
*received_at,
*rtt,
@@ -636,7 +636,7 @@ impl WindowEvent {
#[derive(Debug, Clone, PartialEq)]
struct UniqueSample {
logical_seq: u64,
seq: u32,
bytes: usize,
rtt_primary_ns: i128,
rtt_raw_ns: u64,
@@ -659,7 +659,7 @@ struct UniqueSample {
impl UniqueSample {
#[allow(clippy::too_many_arguments)]
fn new(
logical_seq: u64,
seq: u32,
sent_at: ClientTimestamp,
received_at: ClientTimestamp,
rtt: RttSample,
@@ -669,7 +669,7 @@ impl UniqueSample {
bytes: usize,
) -> Self {
Self {
logical_seq,
seq,
bytes,
rtt_primary_ns: signed_duration_ns(rtt.effective_signed),
rtt_raw_ns: duration_ns_u64(rtt.raw),
@@ -1070,10 +1070,9 @@ mod tests {
}
}
fn sent(seq: u32, logical_seq: u64, sent_at: ClientTimestamp) -> ClientEvent {
fn sent(seq: u32, sent_at: ClientTimestamp) -> ClientEvent {
ClientEvent::EchoSent {
seq,
logical_seq,
remote: "127.0.0.1:2112".parse().unwrap(),
scheduled_at: sent_at.mono,
sent_at,
@@ -1083,24 +1082,23 @@ mod tests {
}
}
fn reply(logical_seq: u64, raw_ms: u64, effective_ms: i128) -> ClientEvent {
let sent_at = ts(logical_seq * 10);
fn reply(seq: u32, raw_ms: u64, effective_ms: i128) -> ClientEvent {
let sent_at = ts(seq as u64 * 10);
let received_at = ClientTimestamp {
mono: sent_at.mono + Duration::from_millis(raw_ms),
wall: sent_at.wall + Duration::from_millis(raw_ms),
};
ClientEvent::EchoReply {
seq: logical_seq as u32,
logical_seq,
seq,
remote: "127.0.0.1:2112".parse().unwrap(),
sent_at,
received_at,
rtt: rtt(raw_ms, effective_ms),
server_timing: Some(ServerTiming {
receive_wall_ns: Some(system_time_ns(sent_at.wall).unwrap() as i64 + 1_000_000),
receive_mono_ns: Some(logical_seq as i64 * 10_000_000 + 1_000_000),
receive_mono_ns: Some(seq as i64 * 10_000_000 + 1_000_000),
send_wall_ns: Some(system_time_ns(sent_at.wall).unwrap() as i64 + 2_000_000),
send_mono_ns: Some(logical_seq as i64 * 10_000_000 + 2_000_000),
send_mono_ns: Some(seq as i64 * 10_000_000 + 2_000_000),
midpoint_wall_ns: None,
midpoint_mono_ns: None,
processing: Some(Duration::from_millis(1)),
@@ -1110,7 +1108,7 @@ mod tests {
server_to_client: Some(Duration::from_millis(2)),
}),
received_stats: Some(ReceivedStatsSample {
count: Some((logical_seq + 1) as u32),
count: Some(seq + 1),
window: Some(0xff),
}),
bytes: 64,
@@ -1118,7 +1116,7 @@ mod tests {
}
}
fn late_reply(logical_seq: u64, raw_ms: u64, effective_ms: i128) -> ClientEvent {
fn late_reply(seq: u32, raw_ms: u64, effective_ms: i128) -> ClientEvent {
let ClientEvent::EchoReply {
seq,
remote,
@@ -1131,13 +1129,12 @@ mod tests {
bytes,
packet_meta,
..
} = reply(logical_seq, raw_ms, effective_ms)
} = reply(seq, raw_ms, effective_ms)
else {
unreachable!();
};
ClientEvent::LateReply {
seq,
logical_seq: Some(logical_seq),
highest_seen: seq + 1,
remote,
sent_at: Some(sent_at),
@@ -1187,7 +1184,7 @@ mod tests {
#[test]
fn continuous_mode_bounds_sequence_tracking() {
let mut collector = StatsCollector::new(StatsConfig::continuous());
for seq in 0..(CONTINUOUS_SEQUENCE_LIMIT as u64 + 8) {
for seq in 0..(CONTINUOUS_SEQUENCE_LIMIT as u32 + 8) {
collector.process(&reply(seq, 10, 10));
}
@@ -1215,8 +1212,8 @@ mod tests {
#[test]
fn late_unique_counts_and_duplicates_do_not_update_measurements() {
let mut collector = StatsCollector::new(StatsConfig::finite());
collector.process(&sent(0, 0, ts(0)));
collector.process(&sent(1, 1, ts(10)));
collector.process(&sent(0, ts(0)));
collector.process(&sent(1, ts(10)));
collector.process(&reply(1, 10, 9));
collector.process(&late_reply(0, 20, 19));
collector.process(&ClientEvent::DuplicateReply {
@@ -1240,11 +1237,10 @@ mod tests {
#[test]
fn final_loss_uses_sent_minus_unique_replies_not_echo_loss_events() {
let mut collector = StatsCollector::new(StatsConfig::finite());
collector.process(&sent(0, 0, ts(0)));
collector.process(&sent(1, 1, ts(10)));
collector.process(&sent(0, ts(0)));
collector.process(&sent(1, ts(10)));
collector.process(&ClientEvent::EchoLoss {
seq: 0,
logical_seq: 0,
sent_at: ts(0),
timeout_at: Instant::now(),
});
@@ -1263,14 +1259,14 @@ mod tests {
fn assert_one_ipdv_pair(
update: &EventStatsUpdate,
previous_logical_seq: u64,
current_logical_seq: u64,
previous_seq: u32,
current_seq: u32,
rtt_ipdv: Duration,
) -> &IpdvPairUpdate {
assert_eq!(update.ipdv_pairs.len(), 1, "{update:?}");
let pair = &update.ipdv_pairs[0];
assert_eq!(pair.previous_logical_seq, previous_logical_seq);
assert_eq!(pair.current_logical_seq, current_logical_seq);
assert_eq!(pair.previous_seq, previous_seq);
assert_eq!(pair.current_seq, current_seq);
assert_eq!(pair.rtt_ipdv, rtt_ipdv);
pair
}
@@ -1340,12 +1336,12 @@ mod tests {
assert!(fill.contributed_sample);
assert_eq!(fill.ipdv_pairs.len(), 2);
assert_eq!(fill.ipdv_pairs[0].previous_logical_seq, 0);
assert_eq!(fill.ipdv_pairs[0].current_logical_seq, 1);
assert_eq!(fill.ipdv_pairs[0].previous_seq, 0);
assert_eq!(fill.ipdv_pairs[0].current_seq, 1);
assert_eq!(fill.ipdv_pairs[0].rtt_ipdv, Duration::from_millis(3));
assert_eq!(fill.ipdv_pairs[1].previous_logical_seq, 1);
assert_eq!(fill.ipdv_pairs[1].current_logical_seq, 2);
assert_eq!(fill.ipdv_pairs[1].previous_seq, 1);
assert_eq!(fill.ipdv_pairs[1].current_seq, 2);
assert_eq!(fill.ipdv_pairs[1].rtt_ipdv, Duration::from_millis(7));
let snapshot = collector.cumulative();
@@ -1359,7 +1355,6 @@ mod tests {
collector.process(&reply(0, 10, 9));
collector.process(&ClientEvent::LateReply {
seq: 9,
logical_seq: None,
highest_seen: 10,
remote: "127.0.0.1:2112".parse().unwrap(),
sent_at: None,
@@ -1385,7 +1380,7 @@ mod tests {
rolling_count: Some(2),
rolling_time: None,
});
collector.process(&sent(0, 0, ts(0)));
collector.process(&sent(0, ts(0)));
collector.process(&reply(0, 10, 10));
collector.process(&reply(1, 20, 20));
@@ -1402,9 +1397,9 @@ mod tests {
rolling_count: None,
rolling_time: Some(Duration::from_millis(15)),
});
collector.process(&sent(0, 0, ts(0)));
collector.process(&sent(1, 1, ts(10)));
collector.process(&sent(2, 2, ts(30)));
collector.process(&sent(0, ts(0)));
collector.process(&sent(1, ts(10)));
collector.process(&sent(2, ts(30)));
let rolling = collector.rolling_time().unwrap();
assert_eq!(rolling.packets.packets_sent, 1);
@@ -1416,7 +1411,7 @@ mod tests {
assert_eq!(empty.loss.packet_loss_percent, 0.0);
let mut collector = StatsCollector::new(StatsConfig::finite());
collector.process(&sent(0, 0, ts(0)));
collector.process(&sent(0, ts(0)));
let all_lost = collector.summary();
assert_eq!(all_lost.loss.lost_packets, 1);
assert_eq!(all_lost.loss.packet_loss_percent, 100.0);
@@ -1425,8 +1420,8 @@ mod tests {
#[test]
fn directional_loss_uses_server_received_count_when_available() {
let mut collector = StatsCollector::new(StatsConfig::finite());
collector.process(&sent(0, 0, ts(0)));
collector.process(&sent(1, 1, ts(10)));
collector.process(&sent(0, ts(0)));
collector.process(&sent(1, ts(10)));
collector.process(&reply(0, 10, 10));
let loss = collector.summary().loss;