feat(client): emit EchoSent event with scheduling metadata
This commit is contained in:
3 files changed
+115
-4
No files matched your search
@@ -152,6 +152,9 @@ pub fn parse_length(input: &str) -> Result<u32, String> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn format_event(event: &ClientEvent, mode: OutputMode) -> Option<String> {
|
pub fn format_event(event: &ClientEvent, mode: OutputMode) -> Option<String> {
|
||||||
|
if matches!(event, ClientEvent::EchoSent { .. }) {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
match mode {
|
match mode {
|
||||||
OutputMode::RttUs => format_rtt_us(event),
|
OutputMode::RttUs => format_rtt_us(event),
|
||||||
OutputMode::Machine => Some(format_machine(event)),
|
OutputMode::Machine => Some(format_machine(event)),
|
||||||
@@ -197,6 +200,7 @@ fn format_machine(event: &ClientEvent) -> String {
|
|||||||
write_token(&mut out, *token);
|
write_token(&mut out, *token);
|
||||||
write_wall(&mut out, "at_ns", at.wall);
|
write_wall(&mut out, "at_ns", at.wall);
|
||||||
}
|
}
|
||||||
|
ClientEvent::EchoSent { .. } => {}
|
||||||
ClientEvent::EchoReply {
|
ClientEvent::EchoReply {
|
||||||
seq,
|
seq,
|
||||||
logical_seq,
|
logical_seq,
|
||||||
@@ -300,12 +304,14 @@ fn format_simple(event: &ClientEvent) -> String {
|
|||||||
ClientEvent::SessionClosed { remote, token, .. } => {
|
ClientEvent::SessionClosed { remote, token, .. } => {
|
||||||
format!("session closed remote={remote} token={token:#x}")
|
format!("session closed remote={remote} token={token:#x}")
|
||||||
}
|
}
|
||||||
|
ClientEvent::EchoSent { .. } => String::new(),
|
||||||
ClientEvent::EchoReply {
|
ClientEvent::EchoReply {
|
||||||
seq,
|
seq,
|
||||||
logical_seq,
|
logical_seq,
|
||||||
remote,
|
remote,
|
||||||
rtt,
|
rtt,
|
||||||
server_timing,
|
server_timing,
|
||||||
|
bytes: _,
|
||||||
..
|
..
|
||||||
} => {
|
} => {
|
||||||
let mut out = format!(
|
let mut out = format!(
|
||||||
@@ -667,6 +673,25 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn echo_sent_is_not_formatted_for_stream_outputs() {
|
||||||
|
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,
|
||||||
|
bytes: 64,
|
||||||
|
send_call: Duration::from_micros(10),
|
||||||
|
timer_error: Duration::ZERO,
|
||||||
|
};
|
||||||
|
|
||||||
|
assert!(format_event(&event, OutputMode::RttUs).is_none());
|
||||||
|
assert!(format_event(&event, OutputMode::Machine).is_none());
|
||||||
|
assert!(format_event(&event, OutputMode::Simple).is_none());
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn warning_and_loss_variants_format_without_panicking() {
|
fn warning_and_loss_variants_format_without_panicking() {
|
||||||
let ts = test_timestamp(Duration::from_secs(1));
|
let ts = test_timestamp(Duration::from_secs(1));
|
||||||
|
|||||||
@@ -196,6 +196,7 @@ impl Client {
|
|||||||
|
|
||||||
let wire_seq = session.next_wire_seq;
|
let wire_seq = session.next_wire_seq;
|
||||||
let logical_seq = session.next_logical_seq;
|
let logical_seq = session.next_logical_seq;
|
||||||
|
let scheduled_at = session.next_send_at;
|
||||||
|
|
||||||
let request = EchoRequest {
|
let request = EchoRequest {
|
||||||
token,
|
token,
|
||||||
@@ -205,7 +206,10 @@ impl Client {
|
|||||||
};
|
};
|
||||||
let packet = encode_echo_request(&request, self.config.hmac_key.as_deref())?;
|
let packet = encode_echo_request(&request, self.config.hmac_key.as_deref())?;
|
||||||
let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now);
|
let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now);
|
||||||
self.socket.send(&packet)?;
|
let send_call_start = Instant::now();
|
||||||
|
let bytes = self.socket.send(&packet)?;
|
||||||
|
let send_call = send_call_start.elapsed();
|
||||||
|
let timer_error = instant_abs_diff(sent_at.mono, scheduled_at);
|
||||||
|
|
||||||
let session = self
|
let session = self
|
||||||
.session
|
.session
|
||||||
@@ -242,7 +246,16 @@ impl Client {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok(vec![])
|
Ok(vec![ClientEvent::EchoSent {
|
||||||
|
seq: wire_seq,
|
||||||
|
logical_seq,
|
||||||
|
remote: self.remote,
|
||||||
|
scheduled_at,
|
||||||
|
sent_at,
|
||||||
|
bytes,
|
||||||
|
send_call,
|
||||||
|
timer_error,
|
||||||
|
}])
|
||||||
}
|
}
|
||||||
|
|
||||||
fn recv_once_inner(
|
fn recv_once_inner(
|
||||||
@@ -606,6 +619,12 @@ fn update_highest_received(session: &mut ActiveSession, wire_seq: u32) {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn instant_abs_diff(left: Instant, right: Instant) -> Duration {
|
||||||
|
left.checked_duration_since(right)
|
||||||
|
.or_else(|| right.checked_duration_since(left))
|
||||||
|
.unwrap_or(Duration::ZERO)
|
||||||
|
}
|
||||||
|
|
||||||
fn compute_rtt(
|
fn compute_rtt(
|
||||||
sent_at: &ClientTimestamp,
|
sent_at: &ClientTimestamp,
|
||||||
received_at: &ClientTimestamp,
|
received_at: &ClientTimestamp,
|
||||||
@@ -1541,7 +1560,23 @@ mod tests {
|
|||||||
};
|
};
|
||||||
let mut client = Client::connect(config).unwrap();
|
let mut client = Client::connect(config).unwrap();
|
||||||
assert_open_started(client.open(ClientTimestamp::now()).unwrap());
|
assert_open_started(client.open(ClientTimestamp::now()).unwrap());
|
||||||
client.send_probe().unwrap();
|
let events = client.send_probe().unwrap();
|
||||||
|
assert_eq!(events.len(), 1);
|
||||||
|
match &events[0] {
|
||||||
|
ClientEvent::EchoSent {
|
||||||
|
seq,
|
||||||
|
logical_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));
|
||||||
|
}
|
||||||
|
other => panic!("expected EchoSent, got {other:?}"),
|
||||||
|
}
|
||||||
thread::sleep(Duration::from_millis(30));
|
thread::sleep(Duration::from_millis(30));
|
||||||
let packets: Vec<_> = server.rx.try_iter().collect();
|
let packets: Vec<_> = server.rx.try_iter().collect();
|
||||||
let echo_reqs: Vec<_> = packets
|
let echo_reqs: Vec<_> = packets
|
||||||
@@ -1559,6 +1594,43 @@ mod tests {
|
|||||||
server.join();
|
server.join();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn echo_sent_reports_schedule_and_timer_error() {
|
||||||
|
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();
|
||||||
|
let start = ClientTimestamp {
|
||||||
|
mono: Instant::now(),
|
||||||
|
wall: SystemTime::now(),
|
||||||
|
};
|
||||||
|
assert_open_started(client.open(start).unwrap());
|
||||||
|
|
||||||
|
let events = client.send_probe_at(start).unwrap();
|
||||||
|
match &events[0] {
|
||||||
|
ClientEvent::EchoSent {
|
||||||
|
scheduled_at,
|
||||||
|
sent_at,
|
||||||
|
timer_error,
|
||||||
|
..
|
||||||
|
} => {
|
||||||
|
assert_eq!(*scheduled_at, start.mono);
|
||||||
|
assert_eq!(*sent_at, start);
|
||||||
|
assert_eq!(*timer_error, Duration::ZERO);
|
||||||
|
}
|
||||||
|
other => panic!("expected EchoSent, got {other:?}"),
|
||||||
|
}
|
||||||
|
|
||||||
|
client.close(ClientTimestamp::now()).unwrap();
|
||||||
|
server.join();
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn send_probe_starts_seq_at_zero_and_increments() {
|
fn send_probe_starts_seq_at_zero_and_increments() {
|
||||||
let params = default_params();
|
let params = default_params();
|
||||||
|
|||||||
@@ -1,4 +1,7 @@
|
|||||||
use std::{net::SocketAddr, time::Duration};
|
use std::{
|
||||||
|
net::SocketAddr,
|
||||||
|
time::{Duration, Instant},
|
||||||
|
};
|
||||||
|
|
||||||
use crate::{session::NegotiatedParams, timing::ClientTimestamp};
|
use crate::{session::NegotiatedParams, timing::ClientTimestamp};
|
||||||
|
|
||||||
@@ -36,6 +39,17 @@ pub enum ClientEvent {
|
|||||||
at: ClientTimestamp,
|
at: ClientTimestamp,
|
||||||
},
|
},
|
||||||
|
|
||||||
|
EchoSent {
|
||||||
|
seq: u32,
|
||||||
|
logical_seq: u64,
|
||||||
|
remote: SocketAddr,
|
||||||
|
scheduled_at: Instant,
|
||||||
|
sent_at: ClientTimestamp,
|
||||||
|
bytes: usize,
|
||||||
|
send_call: Duration,
|
||||||
|
timer_error: Duration,
|
||||||
|
},
|
||||||
|
|
||||||
EchoReply {
|
EchoReply {
|
||||||
seq: u32,
|
seq: u32,
|
||||||
logical_seq: u64,
|
logical_seq: u64,
|
||||||
|
|||||||
Reference in new issue
Block a user