diff --git a/crates/irtt-cli/src/lib.rs b/crates/irtt-cli/src/lib.rs index 6cd8e0c..dbf50a3 100644 --- a/crates/irtt-cli/src/lib.rs +++ b/crates/irtt-cli/src/lib.rs @@ -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, pub receive_ipdv: Option, @@ -352,8 +352,8 @@ impl From for HumanEventStats { impl From 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) { +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, logical_seq: Option) -> 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) -> 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, diff --git a/crates/irtt-cli/src/main.rs b/crates/irtt-cli/src/main.rs index 829ced5..cdf6091 100644 --- a/crates/irtt-cli/src/main.rs +++ b/crates/irtt-cli/src/main.rs @@ -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, diff --git a/crates/irtt-cli/src/summary.rs b/crates/irtt-cli/src/summary.rs index 9107b49..edd80f2 100644 --- a/crates/irtt-cli/src/summary.rs +++ b/crates/irtt-cli/src/summary.rs @@ -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, diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index f89d86d..4a76239 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -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, diff --git a/crates/irtt-client/src/client/tests/probes_replies.rs b/crates/irtt-client/src/client/tests/probes_replies.rs index 89e3d0c..c4daaca 100644 --- a/crates/irtt-client/src/client/tests/probes_replies.rs +++ b/crates/irtt-client/src/client/tests/probes_replies.rs @@ -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, ¶ms)); } @@ -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, ¶ms)); 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, ¶ms)); } 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(); diff --git a/crates/irtt-client/src/client/tests/sequence.rs b/crates/irtt-client/src/client/tests/sequence.rs index d546cb5..1703ff2 100644 --- a/crates/irtt-client/src/client/tests/sequence.rs +++ b/crates/irtt-client/src/client/tests/sequence.rs @@ -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(), diff --git a/crates/irtt-client/src/event.rs b/crates/irtt-client/src/event.rs index 4f705e4..7b38bdd 100644 --- a/crates/irtt-client/src/event.rs +++ b/crates/irtt-client/src/event.rs @@ -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, highest_seen: u32, remote: SocketAddr, sent_at: Option, diff --git a/crates/irtt-client/src/managed/runner.rs b/crates/irtt-client/src/managed/runner.rs index bf7c448..3fc15db 100644 --- a/crates/irtt-client/src/managed/runner.rs +++ b/crates/irtt-client/src/managed/runner.rs @@ -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(_), .. diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 5f18212..5bfb931 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -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, + set: HashSet, insertion_order: VecDeque, 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); diff --git a/crates/irtt-client/src/session.rs b/crates/irtt-client/src/session.rs index b52b4ad..f5c3cda 100644 --- a/crates/irtt-client/src/session.rs +++ b/crates/irtt-client/src/session.rs @@ -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, pub packets_sent: u64, pub start_mono: Instant, diff --git a/crates/irtt-client/tests/hmac.rs b/crates/irtt-client/tests/hmac.rs index e21f57d..42ddcb4 100644 --- a/crates/irtt-client/tests/hmac.rs +++ b/crates/irtt-client/tests/hmac.rs @@ -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, ¶ms)); assert!(server_timing.is_some()); } diff --git a/crates/irtt-client/tests/protocol_options.rs b/crates/irtt-client/tests/protocol_options.rs index e1c738e..6e1c12b 100644 --- a/crates/irtt-client/tests/protocol_options.rs +++ b/crates/irtt-client/tests/protocol_options.rs @@ -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, ¶ms)); 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(), diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index ee35698..f800068 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -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, pub receive_ipdv: Option, @@ -297,9 +297,9 @@ struct CoreStats { send_delay: MetricU64, receive_delay: MetricU64, server_processing: MetricU64, - samples: HashMap, - sample_order: VecDeque, - ipdv_pairs: HashSet, + samples: HashMap, + sample_order: VecDeque, + ipdv_pairs: HashSet, } 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 { + fn try_ipdv_pair(&mut self, current_seq: u32) -> Option { 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;