From acffc3cd4ed1dd3c0cd6a4b5af761ec573e9c6d8 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 23 Jun 2026 21:02:54 +0700 Subject: [PATCH] feat(client): implement multi-target ManagedClientGroup for shared-socket probing Introduce a managed group runner to support concurrent probing of multiple targets over a single UDP socket with coordinated send pacing. Refactor the existing client logic into a reusable SessionRuntime to support both individual and grouped session management. --- crates/irtt-client/src/client.rs | 834 +--------- crates/irtt-client/src/config.rs | 24 + crates/irtt-client/src/lib.rs | 12 +- crates/irtt-client/src/managed/group.rs | 1912 +++++++++++++++++++++++ crates/irtt-client/src/managed/hub.rs | 114 +- crates/irtt-client/src/managed/mod.rs | 6 + crates/irtt-client/src/receive.rs | 32 + crates/irtt-client/src/receive/linux.rs | 109 +- crates/irtt-client/src/runtime.rs | 864 ++++++++++ crates/irtt-client/src/socket.rs | 38 + 10 files changed, 3127 insertions(+), 818 deletions(-) create mode 100644 crates/irtt-client/src/managed/group.rs create mode 100644 crates/irtt-client/src/runtime.rs diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index c6a0f94..5e8e24e 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -1,36 +1,32 @@ use std::{ io, net::{SocketAddr, UdpSocket}, - time::{Duration, Instant, SystemTime, UNIX_EPOCH}, + time::{Duration, Instant}, }; -use irtt_proto::{ - close::CloseRequest, decode_echo_reply, echo_packet_len, encode_close_request, - encode_echo_request, encode_open_request, flags, EchoReply, EchoRequest, OpenReply, - OpenRequest, Params, ServerFill, TimestampFields, PROTOCOL_VERSION, -}; +#[cfg(test)] +use irtt_proto::{flags, Params, TimestampFields, PROTOCOL_VERSION}; use crate::{ - config::{ - ClientConfig, RecvBudget, RunMode, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, - MAX_UDP_PAYLOAD_LENGTH, - }, + config::{ClientConfig, RecvBudget}, error::ClientError, - event::{ - ClientEvent, OneWayDelaySample, OpenOutcome, ReceivedStatsSample, RttSample, ServerTiming, - SignedDuration, WarningKind, - }, - metadata::ReceiveMeta, - probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap}, + event::{ClientEvent, OpenOutcome}, receive::recv_datagram, - session::{negotiate_params, ActiveSession, ClientPhase, CloseSource, NegotiatedParams}, + runtime::{recv_buffer_size, SendProbeResult, SessionRuntime, MAX_OPEN_PACKET_SIZE}, socket::{connect_udp_socket, resolve_remote, validate_open_timeouts}, socket_options::{apply_dscp_to_socket, clear_dscp_on_socket}, timing::ClientTimestamp, }; -const MAX_OPEN_PACKET_SIZE: usize = 512; -const MIN_RECV_BUFFER_SIZE: usize = 2048; +#[cfg(test)] +use crate::{ + runtime::{ + compute_one_way, compute_rtt, params_from_config, sequence_is_after, sequence_is_before, + unix_epoch_ns_i64, update_highest_received, + }, + session::negotiate_params, + NegotiatedParams, RunMode, SignedDuration, WarningKind, MAX_UDP_PAYLOAD_LENGTH, +}; /// Low-level synchronous IRTT client. /// @@ -40,13 +36,9 @@ const MIN_RECV_BUFFER_SIZE: usize = 2048; /// instead. #[derive(Debug)] pub struct Client { - config: ClientConfig, + runtime: SessionRuntime, socket: UdpSocket, remote: SocketAddr, - requested: Params, - negotiated: Option, - phase: ClientPhase, - session: Option, recv_buffer: Vec, } @@ -58,29 +50,15 @@ impl Client { /// IRTT open exchange. pub fn connect(config: ClientConfig) -> Result { validate_open_timeouts(&config.open_timeouts)?; - if config.max_pending_probes == 0 { - return Err(ClientError::InvalidConfig { - reason: "max_pending_probes must be greater than zero".to_owned(), - }); - } - if config.probe_timeout == Duration::ZERO { - return Err(ClientError::InvalidConfig { - reason: "probe_timeout must be greater than zero".to_owned(), - }); - } let remote = resolve_remote(&config)?; + let runtime = SessionRuntime::new(config.clone(), remote)?; let socket = connect_udp_socket(&config.socket_config, remote)?; - let requested = params_from_config(&config)?; Ok(Self { - config, + runtime, socket, remote, - requested, - negotiated: None, - phase: ClientPhase::Connected, - session: None, - recv_buffer: vec![0_u8; MIN_RECV_BUFFER_SIZE], + recv_buffer: vec![0_u8; recv_buffer_size(false, None)?], }) } @@ -90,34 +68,37 @@ impl Client { /// client into either an open probe session or completed no-test state. /// Open attempts use [`ClientConfig::open_timeouts`]. pub fn open(&mut self) -> Result { - match self.phase { - ClientPhase::Connected => {} - ClientPhase::Open { .. } => return Err(ClientError::AlreadyOpen), - ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), - ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted), - } - let outcome = (|| -> Result { - let request = OpenRequest { - params: self.requested.clone(), - close: self.config.run_mode == RunMode::NoTest, - }; - - let packet = encode_open_request(&request, self.config.hmac_key.as_deref())?; + let packet = self.runtime.open_packet()?; let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE]; - for timeout in &self.config.open_timeouts { + for timeout in &self.runtime.config().open_timeouts { self.socket.set_read_timeout(Some(*timeout))?; self.socket.send(&packet)?; match self.socket.recv(&mut buf) { Ok(size) => { - let reply = irtt_proto::decode_open_reply( - &buf[..size], - self.config.hmac_key.as_deref(), - )?; + let reply = self.runtime.decode_open_reply(&buf[..size])?; + let remote = self.remote; + let has_hmac = self.runtime.has_hmac(); + let socket = &self.socket; + let recv_buffer = &mut self.recv_buffer; - return self.accept_open_reply(reply, ClientTimestamp::now()); + return self.runtime.accept_open_reply( + reply, + ClientTimestamp::now(), + |negotiated| { + let size = recv_buffer_size(has_hmac, Some(negotiated))?; + let negotiated_dscp = u8::try_from(negotiated.params.dscp) + .map_err(|_| ClientError::InvalidConfig { + reason: "negotiated dscp must be in range 0..=63" + .to_owned(), + })?; + apply_dscp_to_socket(socket, remote, negotiated_dscp)?; + recv_buffer.resize(size, 0); + Ok(()) + }, + ); } Err(err) if matches!( @@ -133,7 +114,7 @@ impl Client { let restore = self .socket - .set_read_timeout(self.config.socket_config.recv_timeout); + .set_read_timeout(self.runtime.config().socket_config.recv_timeout); match (outcome, restore) { (Ok(outcome), Ok(())) => Ok(outcome), @@ -148,46 +129,24 @@ impl Client { /// The close event means the client has sent its close packet and stopped /// tracking the session locally; it is not a server acknowledgement. pub fn close(&mut self) -> Result, ClientError> { - let token = match self.phase { - ClientPhase::Open { token } => token, - ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), - ClientPhase::Connected | ClientPhase::NoTestCompleted => { - return Err(ClientError::NotOpen) - } - }; - - clear_dscp_on_socket(&self.socket, self.remote)?; - let packet = - encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?; - self.socket.send(&packet)?; - self.phase = ClientPhase::Closed { - source: CloseSource::Local, - }; - if let Some(session) = self.session.as_mut() { - session.timed_out.clear(); - } - self.session = None; - - Ok(vec![ClientEvent::SessionClosed { - remote: self.remote, - token, - at: ClientTimestamp::now(), - }]) + let socket = &self.socket; + let remote = self.remote; + self.runtime.close_with(|packet| { + clear_dscp_on_socket(socket, remote)?; + socket.send(packet)?; + Ok(()) + }) } /// Return the monotonic deadline for the next probe send, if another probe /// is scheduled. pub fn next_send_deadline(&self) -> Option { - let session = self.session.as_ref()?; - if session.sending_done { - return None; - } - Some(session.next_send_at) + self.runtime.next_send_deadline() } /// Return the local timeout used to classify pending probes as lost. pub fn probe_timeout(&self) -> Duration { - self.config.probe_timeout + self.runtime.probe_timeout() } /// Send one echo probe if the negotiated run is still active. @@ -204,126 +163,6 @@ impl Client { /// Returns an empty event list when the socket read would block or times /// out. Malformed or unrelated datagrams are reported as warning events. pub fn recv_once(&mut self) -> Result, ClientError> { - self.recv_once_inner(None) - } - - fn send_probe_inner( - &mut self, - override_ts: Option, - ) -> Result, ClientError> { - let token = match self.phase { - ClientPhase::Open { token } => token, - ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), - ClientPhase::Connected => return Err(ClientError::NotOpen), - ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted), - }; - - let session = self - .session - .as_mut() - .expect("session must exist when phase is Open"); - - if session.sending_done { - return Ok(vec![]); - } - - let now = override_ts.unwrap_or_else(ClientTimestamp::now); - - if let Some(end) = session.end_mono { - if now.mono >= end { - session.sending_done = true; - return Ok(vec![]); - } - } - - session.pending.check_capacity()?; - - let negotiated = self - .negotiated - .as_ref() - .expect("negotiated must exist when Open"); - - let wire_seq = session.next_wire_seq; - let scheduled_at = session.next_send_at; - - let request = EchoRequest { - token, - sequence: wire_seq, - params: negotiated.params.clone(), - payload: vec![], - }; - let packet = encode_echo_request(&request, self.config.hmac_key.as_deref())?; - let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); - 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 - .session - .as_mut() - .expect("session must exist when phase is Open"); - - let pending = PendingProbe { - wire_seq, - sent_at, - timeout_at: sent_at - .mono - .checked_add(self.config.probe_timeout) - .ok_or(ClientError::DurationOverflow)?, - }; - session.pending.insert(pending)?; - - session.next_wire_seq = session.next_wire_seq.wrapping_add(1); - session.packets_sent = - session - .packets_sent - .checked_add(1) - .ok_or(ClientError::CounterOverflow { - counter: "packets_sent", - })?; - - let negotiated = self - .negotiated - .as_ref() - .expect("negotiated must exist when Open"); - let interval_ns = u64::try_from(negotiated.params.interval_ns) - .expect("validated positive negotiated interval"); - let session = self - .session - .as_mut() - .expect("session must exist when phase is Open"); - session.next_send_at = - next_probe_deadline(session.start_mono, interval_ns, session.packets_sent)?; - - if let Some(end) = session.end_mono { - if session.next_send_at >= end { - session.sending_done = true; - } - } - - Ok(vec![ClientEvent::EchoSent { - seq: wire_seq, - remote: self.remote, - scheduled_at, - sent_at, - bytes, - send_call, - timer_error, - }]) - } - - fn recv_once_inner( - &mut self, - override_ts: Option, - ) -> Result, ClientError> { - match self.phase { - ClientPhase::Open { .. } => {} - ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), - ClientPhase::Connected => return Err(ClientError::NotOpen), - ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted), - } - let datagram = match recv_datagram(&self.socket, &mut self.recv_buffer) { Ok(datagram) => datagram, Err(err) @@ -337,16 +176,11 @@ impl Client { Err(err) => return Err(ClientError::Socket(err)), }; - let now = override_ts.unwrap_or(datagram.received_at); - let packet_len = datagram.len; - let Some(reply) = self.decode_received_packet(&self.recv_buffer[..packet_len]) else { - return Ok(vec![ClientEvent::Warning { - kind: WarningKind::MalformedOrUnrelatedPacket, - message: "dropped malformed or unrelated packet".to_owned(), - at: now, - }]); - }; - self.process_echo_reply(reply, packet_len, now, datagram.meta) + self.runtime.process_received_echo_packet( + &self.recv_buffer[..datagram.len], + datagram.received_at, + datagram.meta, + ) } /// Receive and classify datagrams until a receive produces no events or the @@ -379,30 +213,7 @@ impl Client { /// /// `now` is monotonic time only; wall-clock time is not used for timeout expiry. pub fn poll_timeouts_at(&mut self, now: Instant) -> Result, ClientError> { - match self.phase { - ClientPhase::Open { .. } => {} - ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), - ClientPhase::Connected => return Err(ClientError::NotOpen), - ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted), - } - - let session = self - .session - .as_mut() - .expect("session must exist when phase is Open"); - - let expired = session.pending.drain_expired(now); - let mut events = Vec::with_capacity(expired.len()); - for probe in expired { - events.push(ClientEvent::EchoLoss { - seq: probe.wire_seq, - sent_at: probe.sent_at, - timeout_at: probe.timeout_at, - }); - session.timed_out.insert(probe); - } - - Ok(events) + self.runtime.poll_timeouts_at(now) } /// Return whether the current run has completed. @@ -411,13 +222,7 @@ impl Client { /// pending probes have either replied or timed out. No-test and closed /// sessions are also considered complete. pub fn is_run_complete(&self) -> bool { - let Some(session) = self.session.as_ref() else { - return matches!( - self.phase, - ClientPhase::Closed { .. } | ClientPhase::NoTestCompleted - ); - }; - session.sending_done && session.pending.len() == 0 + self.runtime.is_run_complete() } /// Return whether the session was closed by a peer close-flagged reply. @@ -427,295 +232,32 @@ impl Client { /// avoid treating a successfully observed peer close as a local cleanup /// failure. pub fn is_peer_closed(&self) -> bool { - matches!( - self.phase, - ClientPhase::Closed { - source: CloseSource::Peer - } - ) + self.runtime.is_peer_closed() } pub(crate) fn has_timed_out_metadata(&self) -> bool { - self.session - .as_ref() - .is_some_and(|session| session.timed_out.len() > 0) + self.runtime.has_timed_out_metadata() } pub(crate) fn packets_sent(&self) -> u64 { - self.session - .as_ref() - .map_or(0, |session| session.packets_sent) + self.runtime.packets_sent() } - fn decode_received_packet(&self, packet: &[u8]) -> Option { - let negotiated = self - .negotiated - .as_ref() - .expect("negotiated must exist when Open"); - - decode_echo_reply(packet, &negotiated.params, self.config.hmac_key.as_deref()).ok() - } - - fn process_echo_reply( + fn send_probe_inner( &mut self, - reply: EchoReply, - packet_len: usize, - now: ClientTimestamp, - meta: ReceiveMeta, + override_ts: Option, ) -> Result, ClientError> { - let token = match self.phase { - ClientPhase::Open { token } => token, - _ => unreachable!(), - }; - if reply.token != token { - return Ok(vec![ClientEvent::Warning { - kind: WarningKind::WrongToken, - message: format!( - "dropped reply with wrong token: expected {token:#x}, got {:#x}", - reply.token - ), - at: now, - }]); - } - - let session = self.session.as_mut().expect("session must exist when Open"); - - let wire_seq = reply.sequence; - - if let Some(pending) = session.pending.remove(wire_seq) { - let rtt = compute_rtt(&pending.sent_at, &now, &reply.timestamps); - let server_timing = build_server_timing(&reply.timestamps); - let one_way = compute_one_way(&pending.sent_at, &now, &reply.timestamps); - let received_stats = build_received_stats(&reply); - let is_late = session - .highest_received_seq - .is_some_and(|h| sequence_is_before(wire_seq, h)); - let highest_seen = session.highest_received_seq.unwrap_or(wire_seq); - - update_highest_received(&mut session.highest_received_seq, wire_seq); - session.completed.insert(wire_seq); - - let mut events = Vec::new(); - if is_late { - events.push(ClientEvent::LateReply { - seq: wire_seq, - highest_seen, - remote: self.remote, - sent_at: Some(pending.sent_at), - received_at: now, - rtt: Some(rtt), - server_timing, - one_way, - received_stats, - bytes: packet_len, - packet_meta: meta.into(), - }); - } else { - events.push(ClientEvent::EchoReply { - seq: wire_seq, - remote: self.remote, - sent_at: pending.sent_at, - received_at: now, - rtt, - server_timing, - one_way, - received_stats, - bytes: packet_len, - packet_meta: meta.into(), - }); - } - if flags::has(reply.flags, flags::FLAG_CLOSE) { - self.close_from_peer(token, now, &mut events)?; - } - Ok(events) - } else if session.completed.contains(wire_seq) { - update_highest_received(&mut session.highest_received_seq, wire_seq); - Ok(vec![ClientEvent::DuplicateReply { - seq: wire_seq, - remote: self.remote, - received_at: now, - bytes: packet_len, - }]) - } else if let Some(timed_out) = session.timed_out.remove(wire_seq) { - let rtt = compute_rtt(&timed_out.sent_at, &now, &reply.timestamps); - let server_timing = build_server_timing(&reply.timestamps); - let one_way = compute_one_way(&timed_out.sent_at, &now, &reply.timestamps); - let received_stats = build_received_stats(&reply); - let highest_seen = session.highest_received_seq.unwrap_or(wire_seq); - update_highest_received(&mut session.highest_received_seq, wire_seq); - session.completed.insert(wire_seq); - - let mut events = vec![ClientEvent::LateReply { - seq: wire_seq, - highest_seen, - remote: self.remote, - sent_at: Some(timed_out.sent_at), - received_at: now, - rtt: Some(rtt), - server_timing, - one_way, - received_stats, - bytes: packet_len, - packet_meta: meta.into(), - }]; - if flags::has(reply.flags, flags::FLAG_CLOSE) { - self.close_from_peer(token, now, &mut events)?; - } - Ok(events) - } else if session - .highest_received_seq - .is_some_and(|h| sequence_is_before(wire_seq, h)) - { - Ok(vec![ClientEvent::LateReply { - seq: wire_seq, - highest_seen: session.highest_received_seq.unwrap(), - remote: self.remote, - sent_at: None, - received_at: now, - rtt: None, - server_timing: build_server_timing(&reply.timestamps), - one_way: None, - received_stats: build_received_stats(&reply), - bytes: packet_len, - packet_meta: meta.into(), - }]) - } else { - Ok(vec![ClientEvent::Warning { - kind: WarningKind::UntrackedReply, - message: format!( - "dropped reply with untracked seq {wire_seq} (no pending or completed entry)" - ), - at: now, - }]) - } - } - - fn close_from_peer( - &mut self, - token: u64, - now: ClientTimestamp, - events: &mut Vec, - ) -> Result<(), ClientError> { - clear_dscp_on_socket(&self.socket, self.remote)?; - self.phase = ClientPhase::Closed { - source: CloseSource::Peer, - }; - if let Some(session) = self.session.as_mut() { - session.timed_out.clear(); - } - self.session = None; - events.push(ClientEvent::SessionClosed { - remote: self.remote, - token, - at: now, - }); - Ok(()) - } - - fn accept_open_reply( - &mut self, - reply: OpenReply, - now: ClientTimestamp, - ) -> Result { - if reply.params.protocol_version != PROTOCOL_VERSION { - return Err(ClientError::ProtocolVersionMismatch { - requested: PROTOCOL_VERSION, - received: reply.params.protocol_version, - }); - } - - let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE); - match self.config.run_mode { - RunMode::Normal if reply_is_close => Err(ClientError::ServerRejected), - RunMode::Normal if reply.token == 0 => Err(ClientError::ZeroToken), - RunMode::Normal => self.accept_normal_open(reply, now), - RunMode::NoTest if !reply_is_close => Err(ClientError::UnexpectedNoTestReply), - RunMode::NoTest if reply.token != 0 => { - Err(ClientError::NonZeroNoTestToken { token: reply.token }) - } - RunMode::NoTest => self.accept_no_test_open(reply, now), - } - } - - fn accept_normal_open( - &mut self, - reply: OpenReply, - now: ClientTimestamp, - ) -> Result { - let token = reply.token; - let negotiated = negotiate_params( - &self.requested, - reply.params, - self.config.negotiation_policy, - )?; - let recv_buffer_size = recv_buffer_size(self.config.hmac_key.is_some(), Some(&negotiated))?; - let negotiated_dscp = - u8::try_from(negotiated.params.dscp).map_err(|_| ClientError::InvalidConfig { - reason: "negotiated dscp must be in range 0..=63".to_owned(), - })?; - apply_dscp_to_socket(&self.socket, self.remote, negotiated_dscp)?; - self.negotiated = Some(negotiated.clone()); - self.recv_buffer.resize(recv_buffer_size, 0); - self.phase = ClientPhase::Open { token }; - - let end_mono = if negotiated.params.duration_ns > 0 { - Some(negotiated_end_mono( - now.mono, - negotiated.params.duration_ns, - )?) - } else { - None - }; - - self.session = Some(ActiveSession { - next_wire_seq: 0, - highest_received_seq: None, - packets_sent: 0, - start_mono: now.mono, - end_mono, - next_send_at: now.mono, - pending: PendingMap::new(self.config.max_pending_probes), - timed_out: TimedOutMap::new(self.config.max_pending_probes), - completed: CompletedSet::new(self.config.max_pending_probes), - sending_done: false, - }); - - let event = ClientEvent::SessionStarted { - remote: self.remote, - token, - negotiated: negotiated.clone(), - at: now, - }; - - Ok(OpenOutcome::Started { - remote: self.remote, - token, - negotiated, - event, - }) - } - - fn accept_no_test_open( - &mut self, - reply: OpenReply, - now: ClientTimestamp, - ) -> Result { - let negotiated = negotiate_params( - &self.requested, - reply.params, - self.config.negotiation_policy, - )?; - self.negotiated = Some(negotiated.clone()); - self.phase = ClientPhase::NoTestCompleted; - let event = ClientEvent::NoTestCompleted { - remote: self.remote, - negotiated: negotiated.clone(), - at: now, - }; - Ok(OpenOutcome::NoTestCompleted { - remote: self.remote, - negotiated, - event, + let socket = &self.socket; + self.runtime.send_probe_with(override_ts, |packet| { + let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); + let send_call_start = Instant::now(); + let bytes = socket.send(packet)?; + let send_call = send_call_start.elapsed(); + Ok(SendProbeResult { + sent_at, + bytes, + send_call, + }) }) } } @@ -727,233 +269,5 @@ impl Client { } } -fn update_highest_received(highest_received_seq: &mut Option, wire_seq: u32) { - *highest_received_seq = Some(highest_received_seq.map_or(wire_seq, |h| { - if sequence_is_after(wire_seq, h) { - wire_seq - } else { - h - } - })); -} - -fn next_probe_deadline( - start: Instant, - interval_ns: u64, - packets_sent: u64, -) -> Result { - let offset_ns = interval_ns - .checked_mul(packets_sent) - .ok_or(ClientError::DurationOverflow)?; - start - .checked_add(Duration::from_nanos(offset_ns)) - .ok_or(ClientError::DurationOverflow) -} - -fn sequence_is_after(candidate: u32, current: u32) -> bool { - candidate != current && candidate.wrapping_sub(current) < (1 << 31) -} - -fn sequence_is_before(candidate: u32, current: u32) -> bool { - current != candidate && current.wrapping_sub(candidate) < (1 << 31) -} - -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) -} - -pub(crate) fn recv_buffer_size( - has_hmac: bool, - negotiated: Option<&NegotiatedParams>, -) -> Result { - Ok(match negotiated { - Some(negotiated) => echo_packet_len(has_hmac, &negotiated.params)? - .saturating_add(1) - .max(MIN_RECV_BUFFER_SIZE), - None => MIN_RECV_BUFFER_SIZE, - }) -} - -fn compute_rtt( - sent_at: &ClientTimestamp, - received_at: &ClientTimestamp, - ts: &TimestampFields, -) -> RttSample { - let raw = received_at - .mono - .checked_duration_since(sent_at.mono) - .unwrap_or(Duration::ZERO); - - let server_processing = compute_server_processing(ts); - - let adjusted = server_processing - .map(|sp| SignedDuration::from_nanos(duration_ns_i128(raw) - duration_ns_i128(sp))); - let effective = adjusted.unwrap_or_else(|| SignedDuration::from_duration(raw)); - - RttSample { - raw, - adjusted, - effective, - } -} - -fn duration_ns_i128(duration: Duration) -> i128 { - i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX) -} - -fn compute_server_processing(ts: &TimestampFields) -> Option { - if let (Some(recv_mono), Some(send_mono)) = (ts.recv_mono, ts.send_mono) { - let diff = send_mono.checked_sub(recv_mono)?; - return Some(Duration::from_nanos(u64::try_from(diff).ok()?)); - } - if let (Some(recv_wall), Some(send_wall)) = (ts.recv_wall, ts.send_wall) { - let diff = send_wall.checked_sub(recv_wall)?; - return Some(Duration::from_nanos(u64::try_from(diff).ok()?)); - } - None -} - -fn build_server_timing(ts: &TimestampFields) -> Option { - if ts.recv_wall.is_none() - && ts.recv_mono.is_none() - && ts.send_wall.is_none() - && ts.send_mono.is_none() - && ts.midpoint_wall.is_none() - && ts.midpoint_mono.is_none() - { - return None; - } - Some(ServerTiming { - receive_wall_ns: ts.recv_wall, - receive_mono_ns: ts.recv_mono, - send_wall_ns: ts.send_wall, - send_mono_ns: ts.send_mono, - midpoint_wall_ns: ts.midpoint_wall, - midpoint_mono_ns: ts.midpoint_mono, - processing: compute_server_processing(ts), - }) -} - -fn compute_one_way( - sent_at: &ClientTimestamp, - received_at: &ClientTimestamp, - ts: &TimestampFields, -) -> Option { - let server_recv_wall = ts.recv_wall.or(ts.midpoint_wall); - let server_send_wall = ts.send_wall.or(ts.midpoint_wall); - - let client_send_ns = unix_epoch_ns_i64(sent_at.wall); - let client_recv_ns = unix_epoch_ns_i64(received_at.wall); - - let c2s = server_recv_wall - .zip(client_send_ns) - .and_then(|(srv, cli)| srv.checked_sub(cli)) - .map(|d| SignedDuration::from_nanos(i128::from(d))); - let s2c = client_recv_ns - .zip(server_send_wall) - .and_then(|(cli, srv)| cli.checked_sub(srv)) - .map(|d| SignedDuration::from_nanos(i128::from(d))); - - if c2s.is_none() && s2c.is_none() { - return None; - } - - Some(OneWayDelaySample { - client_to_server: c2s, - server_to_client: s2c, - }) -} - -fn unix_epoch_ns_i64(time: SystemTime) -> Option { - time.duration_since(UNIX_EPOCH) - .ok() - .and_then(|duration| i64::try_from(duration.as_nanos()).ok()) -} - -fn build_received_stats(reply: &EchoReply) -> Option { - if reply.recv_count.is_none() && reply.recv_window.is_none() { - return None; - } - Some(ReceivedStatsSample { - count: reply.recv_count, - window: reply.recv_window, - }) -} - -fn params_from_config(config: &ClientConfig) -> Result { - validate_protocol_config(config)?; - Ok(Params { - protocol_version: PROTOCOL_VERSION, - duration_ns: match config.duration { - Some(duration) => config_duration_to_ns("duration", duration)?, - None => 0, - }, - interval_ns: config_duration_to_ns("interval", config.interval)?, - length: i64::from(config.length), - received_stats: config.received_stats, - stamp_at: config.stamp_at, - clock: config.clock, - dscp: i64::from(config.dscp), - server_fill: config.server_fill.clone().map(|value| ServerFill { value }), - }) -} - -fn validate_protocol_config(config: &ClientConfig) -> Result<(), ClientError> { - if config.duration == Some(Duration::ZERO) { - return Err(ClientError::InvalidConfig { - reason: "duration must be greater than zero; use None for continuous mode".to_owned(), - }); - } - if config.interval == Duration::ZERO { - return Err(ClientError::InvalidConfig { - reason: "interval must be greater than zero".to_owned(), - }); - } - if config.dscp > MAX_DSCP_CODEPOINT { - return Err(ClientError::InvalidConfig { - reason: format!("dscp must be <= {MAX_DSCP_CODEPOINT}"), - }); - } - if config.length > MAX_UDP_PAYLOAD_LENGTH { - return Err(ClientError::InvalidConfig { - reason: format!("packet length must be <= {MAX_UDP_PAYLOAD_LENGTH}"), - }); - } - - if let Some(fill) = &config.server_fill { - let len = fill.len(); - if len == 0 { - return Err(ClientError::InvalidConfig { - reason: "server_fill must not be empty".to_owned(), - }); - } - if len > MAX_SERVER_FILL_BYTES { - return Err(ClientError::InvalidConfig { - reason: format!("server_fill must be <= {MAX_SERVER_FILL_BYTES} bytes, got {len}"), - }); - } - } - - Ok(()) -} - -fn config_duration_to_ns(field: &str, duration: Duration) -> Result { - i64::try_from(duration.as_nanos()).map_err(|_| ClientError::InvalidConfig { - reason: format!("{field} is too large to encode as nanoseconds"), - }) -} - -fn negotiated_end_mono(start: Instant, duration_ns: i64) -> Result { - debug_assert!(duration_ns > 0); - let duration_ns = u64::try_from(duration_ns).expect("validated positive negotiated duration"); - start - .checked_add(Duration::from_nanos(duration_ns)) - .ok_or_else(|| ClientError::NegotiationRejected { - reason: "duration is too large to schedule".to_owned(), - }) -} - #[cfg(test)] mod tests; diff --git a/crates/irtt-client/src/config.rs b/crates/irtt-client/src/config.rs index 8bc6647..44a3215 100644 --- a/crates/irtt-client/src/config.rs +++ b/crates/irtt-client/src/config.rs @@ -118,6 +118,30 @@ pub struct ClientConfig { pub max_pending_probes: usize, } +/// Authentication settings used by client sessions. +/// +/// `ClientConfig` keeps its existing top-level HMAC field for source +/// compatibility. Managed multi-target callers can use this smaller type for +/// per-target auth overrides without carrying a full independent +/// [`ClientConfig`]. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct ClientAuthConfig { + /// Optional HMAC key used to authenticate IRTT packets. + pub hmac_key: Option>, +} + +impl From>> for ClientAuthConfig { + fn from(hmac_key: Option>) -> Self { + Self { hmac_key } + } +} + +impl From for Option> { + fn from(auth: ClientAuthConfig) -> Self { + auth.hmac_key + } +} + impl Default for ClientConfig { fn default() -> Self { Self { diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index fc6d960..f712de5 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -59,6 +59,7 @@ mod managed; mod metadata; mod probe; mod receive; +mod runtime; mod session; mod socket; mod socket_options; @@ -66,8 +67,8 @@ mod timing; pub use client::Client; pub use config::{ - ClientConfig, NegotiationPolicy, RecvBudget, RunMode, SocketConfig, MAX_DSCP_CODEPOINT, - MAX_SERVER_FILL_BYTES, MAX_TTL, MAX_UDP_PAYLOAD_LENGTH, + ClientAuthConfig, ClientConfig, NegotiationPolicy, RecvBudget, RunMode, SocketConfig, + MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_TTL, MAX_UDP_PAYLOAD_LENGTH, }; pub use error::{ClientError, EventSubscriptionError}; pub use event::{ @@ -75,8 +76,11 @@ pub use event::{ ServerTiming, SignedDuration, WarningKind, }; pub use managed::{ - CancellationToken, EventHub, EventSubscription, ManagedClient, ManagedClientSession, - SessionEndReason, SessionOutcome, SubscriberConfig, SubscriberOverflow, + CancellationToken, EventHub, EventSubscription, ManagedClient, ManagedClientGroup, + ManagedClientGroupConfig, ManagedClientGroupSession, ManagedClientSession, + ManagedGroupEndReason, ManagedGroupOutcome, ManagedGroupPacing, ManagedTargetConfig, + ManagedTargetEndReason, ManagedTargetOutcome, SessionEndReason, SessionOutcome, + SubscriberConfig, SubscriberOverflow, TargetEvent, TargetEventSubscription, TargetId, }; pub use session::{NegotiatedParams, NegotiationRestriction}; pub use timing::ClientTimestamp; diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs new file mode 100644 index 0000000..e323a74 --- /dev/null +++ b/crates/irtt-client/src/managed/group.rs @@ -0,0 +1,1912 @@ +use std::{ + collections::{HashMap, HashSet}, + fmt, + hash::{Hash, Hasher}, + net::{SocketAddr, UdpSocket}, + sync::{mpsc, Arc, Mutex}, + thread::{self, JoinHandle}, + time::{Duration, Instant}, +}; + +use crate::{ + config::{ClientAuthConfig, ClientConfig}, + error::ClientError, + event::{ClientEvent, OpenOutcome}, + receive::recv_datagram_from, + runtime::{params_from_config, SendProbeResult, SessionRuntime}, + socket::{bind_unconnected_udp_socket, validate_open_timeouts}, + socket_options::apply_dscp_to_socket, + timing::ClientTimestamp, +}; + +use super::{ + cancellation::CancellationToken, + hub::{EventHub, EventSubscription, SubscriberConfig}, +}; + +const GROUP_RECV_TIMEOUT: Duration = Duration::from_millis(20); +const GROUP_FINAL_DRAIN: Duration = Duration::from_millis(100); +const IDLE_SLEEP: Duration = Duration::from_millis(1); +const MAX_SLEEP: Duration = Duration::from_millis(20); +const RECV_BUFFER_SIZE: usize = 65_536; + +/// Caller-owned target identity for managed multi-target probing. +#[derive(Clone, Eq)] +pub struct TargetId(Arc); + +impl TargetId { + /// Borrow the identifier as a string slice. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Debug for TargetId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_tuple("TargetId").field(&self.0).finish() + } +} + +impl fmt::Display for TargetId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +impl PartialEq for TargetId { + fn eq(&self, other: &Self) -> bool { + self.0 == other.0 + } +} + +impl Hash for TargetId { + fn hash(&self, state: &mut H) { + self.0.hash(state); + } +} + +impl From<&str> for TargetId { + fn from(value: &str) -> Self { + Self(Arc::from(value)) + } +} + +impl From for TargetId { + fn from(value: String) -> Self { + Self(Arc::from(value)) + } +} + +impl AsRef for TargetId { + fn as_ref(&self) -> &str { + self.as_str() + } +} + +/// Per-target configuration for [`ManagedClientGroup`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ManagedTargetConfig { + /// Caller-owned stable target identity. + pub id: TargetId, + /// Resolved UDP remote address. Group v1 does not perform DNS resolution. + pub remote: SocketAddr, + /// Optional auth override for this target. + pub auth: Option, +} + +/// Group send pacing strategy. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum ManagedGroupPacing { + /// Send active targets one at a time, spaced approximately interval / N. + #[default] + Staggered, + /// Send one probe to every active target back-to-back once per interval. + Burst, +} + +/// Configuration for a managed multi-target client group. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct ManagedClientGroupConfig { + /// Shared client configuration template. + /// + /// Most protocol, socket, timing, and negotiation settings are group-wide. + /// `server_addr` is ignored because targets already carry resolved + /// [`SocketAddr`] values. `hmac_key` is the default auth unless a target + /// supplies [`ManagedTargetConfig::auth`]. + pub client: ClientConfig, + /// Coordinated group pacing mode. + pub pacing: ManagedGroupPacing, +} + +/// Target-scoped managed event. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TargetEvent { + /// Target that produced the event. + pub target: TargetId, + /// Client event produced by that target's session runtime. + pub event: ClientEvent, +} + +/// Subscription type for target-scoped group events. +pub type TargetEventSubscription = EventSubscription; + +/// Entry point for running a shared-socket multi-target managed client group. +#[derive(Debug)] +pub struct ManagedClientGroup; + +/// Running managed client group. +/// +/// Dropping the session requests cooperative cancellation. Use +/// [`join`](Self::join) to wait for scheduler and receive threads and obtain +/// the final [`ManagedGroupOutcome`]. +#[must_use = "dropping the session cancels the managed client group; call join() to wait for completion"] +#[derive(Debug)] +pub struct ManagedClientGroupSession { + hub: EventHub, + control_tx: mpsc::Sender, + cancellation: CancellationToken, + scheduler: Option>>, + receiver: Option>>, +} + +/// Outcome returned by a completed managed client group. +#[must_use = "managed group outcomes contain completion status and per-target lifecycle counters"] +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ManagedGroupOutcome { + /// Why the group scheduler stopped. + pub end_reason: ManagedGroupEndReason, + /// Per-target lifecycle records. These are not RTT/loss summaries. + pub targets: Vec, +} + +/// Reason the managed group stopped. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ManagedGroupEndReason { + /// All currently desired targets reached a terminal state. + AllTargetsComplete, + /// Cancellation was requested through stop, drop, or a worker failure. + Cancelled, +} + +/// Per-target lifecycle outcome. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ManagedTargetOutcome { + /// Target identity. + pub id: TargetId, + /// Resolved remote address used for the target. + pub remote: SocketAddr, + /// Why this target stopped. + pub end_reason: ManagedTargetEndReason, + /// Number of echo requests sent to this target. + pub packets_sent: u64, + /// Number of first in-window echo replies received from this target. + pub replies_received: u64, + /// Number of duplicate reply events emitted for this target. + pub duplicates: u64, + /// Number of late reply events emitted for this target. + pub late: u64, + /// Number of warning events emitted for this target. + pub warning_events: u64, +} + +/// Reason an individual target stopped. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ManagedTargetEndReason { + /// The negotiated finite test duration completed. + TestComplete, + /// Cancellation stopped the target. + Cancelled, + /// The target was removed by `update_targets`. + Removed, + /// The target completed a no-test open exchange. + NoTestComplete, + /// The server closed the target session. + PeerClosed, + /// Opening failed before a `ClientEvent` lifecycle event existed. + OpenFailed { message: String }, + /// Runtime send/receive/close handling failed for this target. + Failed { message: String }, +} + +impl ManagedClientGroup { + /// Start a managed group without creating an initial event subscription. + pub fn start( + config: ManagedClientGroupConfig, + targets: Vec, + ) -> Result { + Self::start_inner(config, targets, None).map(|(session, _)| session) + } + + /// Start a managed group and subscribe before worker threads run. + pub fn start_with_subscription( + config: ManagedClientGroupConfig, + targets: Vec, + subscriber_config: SubscriberConfig, + ) -> Result<(ManagedClientGroupSession, TargetEventSubscription), ClientError> { + let (session, subscription) = Self::start_inner(config, targets, Some(subscriber_config))?; + Ok(( + session, + subscription.expect("initial subscription must be present"), + )) + } + + fn start_inner( + mut config: ManagedClientGroupConfig, + targets: Vec, + subscriber_config: Option, + ) -> Result<(ManagedClientGroupSession, Option), ClientError> { + if targets.is_empty() { + return Err(ClientError::InvalidConfig { + reason: "managed client group requires at least one target".to_owned(), + }); + } + if config.client.socket_config.recv_timeout.is_none() + || config.client.socket_config.recv_timeout > Some(GROUP_RECV_TIMEOUT) + { + config.client.socket_config.recv_timeout = Some(GROUP_RECV_TIMEOUT); + } + + validate_open_timeouts(&config.client.open_timeouts)?; + let requested = params_from_config(&config.client)?; + let family_remote = validate_target_configs(&targets, None)?; + validate_group_family_flags(&config.client, family_remote)?; + + let now = Instant::now(); + let mut next_order = 0_u64; + let mut registry = TargetRegistry::default(); + for target in targets { + let state = Arc::new(Mutex::new(TargetState::new( + &config.client, + target, + next_order, + now, + )?)); + next_order = next_order + .checked_add(1) + .ok_or(ClientError::CounterOverflow { + counter: "target_order", + })?; + let (id, remote) = { + let target = state.lock().expect("target mutex poisoned"); + (target.id.clone(), target.remote) + }; + registry.remotes.insert(remote, id.clone()); + registry.targets.insert(id, state); + } + + let socket = bind_unconnected_udp_socket(&config.client.socket_config, family_remote)?; + apply_dscp_to_socket(&socket, family_remote, config.client.dscp)?; + + let hub = EventHub::new(); + let initial_subscription = subscriber_config + .map(|config| hub.subscribe(config)) + .transpose()?; + + let cancellation = CancellationToken::new(); + let (control_tx, control_rx) = mpsc::channel(); + let shared = Arc::new(GroupShared { + registry: Mutex::new(registry), + hub: hub.clone(), + cancellation: cancellation.clone(), + control_tx: control_tx.clone(), + requested_interval_ns: requested.interval_ns, + requested_dscp: requested.dscp, + family_remote, + }); + + let send_socket = socket; + let recv_socket = send_socket.try_clone()?; + let scheduler_shared = shared.clone(); + let scheduler_config = config.clone(); + let scheduler = thread::spawn(move || { + run_group_scheduler( + scheduler_config, + send_socket, + scheduler_shared, + control_rx, + next_order, + ) + }); + + let receiver_shared = shared; + let receiver = thread::spawn(move || run_group_receiver(recv_socket, receiver_shared)); + + Ok(( + ManagedClientGroupSession { + hub, + control_tx, + cancellation, + scheduler: Some(scheduler), + receiver: Some(receiver), + }, + initial_subscription, + )) + } +} + +impl ManagedClientGroupSession { + /// Add another target-event subscriber to this running group. + pub fn subscribe( + &self, + config: SubscriberConfig, + ) -> Result { + self.hub.subscribe(config) + } + + /// Replace the desired target set. + /// + /// The update is authoritative. Existing targets with the same + /// [`TargetId`], remote address, and auth setting remain running. In v1, + /// changing the remote address or auth for an existing target id is + /// rejected instead of mutating the active session in place. + pub fn update_targets(&self, targets: Vec) -> Result<(), ClientError> { + let (reply_tx, reply_rx) = mpsc::channel(); + self.control_tx + .send(ControlMessage::Update { + targets, + reply: reply_tx, + }) + .map_err(|_| ClientError::AlreadyClosed)?; + reply_rx.recv().unwrap_or(Err(ClientError::AlreadyClosed)) + } + + /// Request cooperative cancellation of the group. + pub fn stop(&self) { + self.cancellation.cancel(); + let _ = self.control_tx.send(ControlMessage::Wake); + } + + /// Wait for the scheduler and receive threads to finish. + pub fn join(mut self) -> Result { + let scheduler = self.scheduler.take().expect( + "ManagedClientGroupSession invariant violated: scheduler handle missing before join", + ); + let scheduler_result = match scheduler.join() { + Ok(outcome) => outcome, + Err(_) => { + self.cancellation.cancel(); + self.hub.disconnect_all(); + return Err(ClientError::WorkerPanicked); + } + }; + + self.cancellation.cancel(); + let _ = self.control_tx.send(ControlMessage::Wake); + + let receiver = self.receiver.take().expect( + "ManagedClientGroupSession invariant violated: receiver handle missing before join", + ); + let receiver_result = match receiver.join() { + Ok(outcome) => outcome, + Err(_) => { + self.hub.disconnect_all(); + return Err(ClientError::WorkerPanicked); + } + }; + + self.hub.disconnect_all(); + match (scheduler_result, receiver_result) { + (Ok(outcome), Ok(())) => Ok(outcome), + (Err(err), _) => Err(err), + (Ok(_), Err(err)) => Err(err), + } + } +} + +impl Drop for ManagedClientGroupSession { + fn drop(&mut self) { + self.cancellation.cancel(); + let _ = self.control_tx.send(ControlMessage::Wake); + } +} + +#[derive(Debug)] +enum ControlMessage { + Update { + targets: Vec, + reply: mpsc::Sender>, + }, + Wake, +} + +#[derive(Debug)] +struct GroupShared { + registry: Mutex, + hub: EventHub, + cancellation: CancellationToken, + control_tx: mpsc::Sender, + requested_interval_ns: i64, + requested_dscp: i64, + family_remote: SocketAddr, +} + +#[derive(Debug, Default)] +struct TargetRegistry { + targets: HashMap>>, + remotes: HashMap, +} + +#[derive(Debug)] +struct TargetState { + id: TargetId, + remote: SocketAddr, + configured_auth: Option, + runtime: SessionRuntime, + status: TargetStatus, + open_packet: Vec, + counters: TargetCounters, + order: u64, + final_reason: Option, +} + +#[derive(Debug)] +enum TargetStatus { + Opening { + attempt: usize, + next_send_at: Instant, + }, + Active, + Draining { + deadline: Instant, + }, + Finished, +} + +#[derive(Debug, Default, Clone, PartialEq, Eq)] +struct TargetCounters { + replies_received: u64, + duplicates: u64, + late: u64, + warning_events: u64, +} + +impl TargetState { + fn new( + group_config: &ClientConfig, + target: ManagedTargetConfig, + order: u64, + now: Instant, + ) -> Result { + let mut config = group_config.clone(); + if let Some(auth) = &target.auth { + config.hmac_key = auth.hmac_key.clone(); + } + let runtime = SessionRuntime::new(config, target.remote)?; + let open_packet = runtime.open_packet()?; + Ok(Self { + id: target.id, + remote: target.remote, + configured_auth: target.auth, + runtime, + status: TargetStatus::Opening { + attempt: 0, + next_send_at: now, + }, + open_packet, + counters: TargetCounters::default(), + order, + final_reason: None, + }) + } + + fn same_config(&self, config: &ManagedTargetConfig) -> bool { + self.remote == config.remote && self.configured_auth == config.auth + } + + fn observe(&mut self, event: &ClientEvent) { + match event { + ClientEvent::EchoReply { .. } => self.counters.replies_received += 1, + ClientEvent::DuplicateReply { .. } => self.counters.duplicates += 1, + ClientEvent::LateReply { .. } => self.counters.late += 1, + ClientEvent::Warning { .. } => self.counters.warning_events += 1, + _ => {} + } + } + + fn mark_finished(&mut self, reason: ManagedTargetEndReason) { + self.status = TargetStatus::Finished; + self.final_reason = Some(reason); + } + + fn outcome(&self, reason: ManagedTargetEndReason) -> ManagedTargetOutcome { + ManagedTargetOutcome { + id: self.id.clone(), + remote: self.remote, + end_reason: reason, + packets_sent: self.runtime.packets_sent(), + replies_received: self.counters.replies_received, + duplicates: self.counters.duplicates, + late: self.counters.late, + warning_events: self.counters.warning_events, + } + } +} + +fn run_group_scheduler( + config: ManagedClientGroupConfig, + socket: UdpSocket, + shared: Arc, + control_rx: mpsc::Receiver, + mut next_order: u64, +) -> Result { + let mut records = Vec::new(); + let mut pacing = PacingRuntime::new(config.pacing); + let mut pending_control = None; + + loop { + drain_control_messages( + &config.client, + &socket, + &shared, + &control_rx, + pending_control.take(), + &mut next_order, + &mut records, + ); + + if shared.cancellation.is_cancelled() { + cancel_remaining_targets(&socket, &shared, &mut records); + return Ok(ManagedGroupOutcome { + end_reason: ManagedGroupEndReason::Cancelled, + targets: records, + }); + } + + let now = Instant::now(); + drive_open_attempts(&config.client, &socket, &shared, now); + poll_active_timeouts(&shared, now); + finish_completed_targets(&socket, &shared, now); + collect_finished_targets(&shared, &mut records); + + if registry_is_empty(&shared) { + return Ok(ManagedGroupOutcome { + end_reason: ManagedGroupEndReason::AllTargetsComplete, + targets: records, + }); + } + + let active = active_targets(&shared); + pacing.reconcile(&active, now); + if pacing.send_due(&active, now, config.client.interval) { + match config.pacing { + ManagedGroupPacing::Staggered => { + if let Some((target, scheduled_at)) = + pacing.next_staggered(&active, config.client.interval) + { + send_echo_to_target(&socket, &shared, target, scheduled_at); + } + } + ManagedGroupPacing::Burst => { + if let Some(scheduled_at) = pacing.next_burst(config.client.interval) { + for target in active { + send_echo_to_target(&socket, &shared, target, scheduled_at); + } + } + } + } + continue; + } + + pending_control = wait_for_next_scheduler_wakeup(&control_rx, &shared, &pacing, &active); + } +} + +fn run_group_receiver(socket: UdpSocket, shared: Arc) -> Result<(), ClientError> { + let mut buf = vec![0_u8; RECV_BUFFER_SIZE]; + while !shared.cancellation.is_cancelled() { + let datagram = match recv_datagram_from(&socket, &mut buf) { + Ok(datagram) => datagram, + Err(err) + if matches!( + err.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) => + { + continue; + } + Err(err) => { + shared.cancellation.cancel(); + let _ = shared.control_tx.send(ControlMessage::Wake); + return Err(ClientError::Socket(err)); + } + }; + + let target = { + let registry = shared + .registry + .lock() + .expect("target registry mutex poisoned"); + let Some(id) = registry.remotes.get(&datagram.source) else { + continue; + }; + registry.targets.get(id).cloned() + }; + let Some(target) = target else { + continue; + }; + + let packet = &buf[..datagram.len]; + let mut wake_scheduler = false; + { + let mut target = target.lock().expect("target mutex poisoned"); + match target.status { + TargetStatus::Opening { .. } => { + let result = target.runtime.decode_open_reply(packet).and_then(|reply| { + target.runtime.accept_open_reply( + reply, + datagram.received_at, + |negotiated| validate_group_negotiation(&shared, negotiated), + ) + }); + + match result { + Ok(outcome) => { + publish_open_outcome(&shared.hub, &mut target, outcome); + wake_scheduler = true; + } + Err(err) => { + target.mark_finished(ManagedTargetEndReason::OpenFailed { + message: err.to_string(), + }); + wake_scheduler = true; + } + } + } + TargetStatus::Active | TargetStatus::Draining { .. } => { + match target.runtime.process_received_echo_packet( + packet, + datagram.received_at, + datagram.meta, + ) { + Ok(events) => { + publish_events(&shared.hub, &mut target, events); + if target.runtime.is_peer_closed() { + wake_scheduler = true; + } + } + Err(err) => { + target.mark_finished(ManagedTargetEndReason::Failed { + message: err.to_string(), + }); + wake_scheduler = true; + } + } + } + TargetStatus::Finished => {} + } + } + + if wake_scheduler { + let _ = shared.control_tx.send(ControlMessage::Wake); + } + } + Ok(()) +} + +fn validate_group_negotiation( + shared: &GroupShared, + negotiated: &crate::NegotiatedParams, +) -> Result<(), ClientError> { + if negotiated.params.interval_ns != shared.requested_interval_ns { + return Err(ClientError::NegotiationRejected { + reason: "managed group v1 requires the negotiated interval to match the group interval" + .to_owned(), + }); + } + if negotiated.params.dscp != shared.requested_dscp { + return Err(ClientError::NegotiationRejected { + reason: "managed group v1 requires the negotiated DSCP to match the group DSCP" + .to_owned(), + }); + } + Ok(()) +} + +fn publish_open_outcome( + hub: &EventHub, + target: &mut TargetState, + outcome: OpenOutcome, +) { + match outcome { + OpenOutcome::Started { event, .. } => { + target.status = TargetStatus::Active; + hub.publish(TargetEvent { + target: target.id.clone(), + event, + }); + } + OpenOutcome::NoTestCompleted { event, .. } => { + target.observe(&event); + hub.publish(TargetEvent { + target: target.id.clone(), + event, + }); + target.mark_finished(ManagedTargetEndReason::NoTestComplete); + } + } +} + +fn publish_events(hub: &EventHub, target: &mut TargetState, events: Vec) { + for event in events { + target.observe(&event); + hub.publish(TargetEvent { + target: target.id.clone(), + event, + }); + } +} + +fn drain_control_messages( + config: &ClientConfig, + socket: &UdpSocket, + shared: &GroupShared, + control_rx: &mpsc::Receiver, + first_message: Option, + next_order: &mut u64, + records: &mut Vec, +) { + if let Some(message) = first_message { + handle_control_message(config, socket, shared, next_order, records, message); + } + while let Ok(message) = control_rx.try_recv() { + handle_control_message(config, socket, shared, next_order, records, message); + } +} + +fn handle_control_message( + config: &ClientConfig, + socket: &UdpSocket, + shared: &GroupShared, + next_order: &mut u64, + records: &mut Vec, + message: ControlMessage, +) { + match message { + ControlMessage::Update { targets, reply } => { + let result = apply_target_update(config, socket, shared, targets, next_order, records); + let _ = reply.send(result); + } + ControlMessage::Wake => {} + } +} + +fn apply_target_update( + config: &ClientConfig, + socket: &UdpSocket, + shared: &GroupShared, + targets: Vec, + next_order: &mut u64, + records: &mut Vec, +) -> Result<(), ClientError> { + validate_open_timeouts(&config.open_timeouts)?; + if targets.is_empty() { + remove_all_targets(socket, shared, records, ManagedTargetEndReason::Removed); + return Ok(()); + } + validate_target_configs(&targets, Some(shared.family_remote))?; + + let desired_ids: HashSet = targets.iter().map(|target| target.id.clone()).collect(); + let mut additions = Vec::new(); + + { + let registry = shared + .registry + .lock() + .expect("target registry mutex poisoned"); + for target in &targets { + if let Some(existing) = registry.targets.get(&target.id) { + let existing = existing.lock().expect("target mutex poisoned"); + if !existing.same_config(target) { + return Err(ClientError::InvalidConfig { + reason: format!( + "managed group v1 rejects changed remote/auth for existing target {}", + target.id + ), + }); + } + } else { + additions.push(target.clone()); + } + } + } + + let mut prepared = Vec::new(); + let now = Instant::now(); + for target in additions { + let order = *next_order; + *next_order = next_order + .checked_add(1) + .ok_or(ClientError::CounterOverflow { + counter: "target_order", + })?; + prepared.push(Arc::new(Mutex::new(TargetState::new( + config, target, order, now, + )?))); + } + + let removed = { + let mut registry = shared + .registry + .lock() + .expect("target registry mutex poisoned"); + let removed_ids: Vec = registry + .targets + .keys() + .filter(|id| !desired_ids.contains(*id)) + .cloned() + .collect(); + let mut removed = Vec::with_capacity(removed_ids.len()); + for id in removed_ids { + if let Some(target) = registry.targets.remove(&id) { + let remote = target.lock().expect("target mutex poisoned").remote; + registry.remotes.remove(&remote); + removed.push(target); + } + } + for target in prepared { + let (id, remote) = { + let target = target.lock().expect("target mutex poisoned"); + (target.id.clone(), target.remote) + }; + registry.remotes.insert(remote, id.clone()); + registry.targets.insert(id, target); + } + removed + }; + + for target in removed { + records.push(close_target( + socket, + &shared.hub, + target, + ManagedTargetEndReason::Removed, + )); + } + + Ok(()) +} + +fn validate_target_configs( + targets: &[ManagedTargetConfig], + expected_family: Option, +) -> Result { + let mut ids = HashSet::new(); + let mut remotes = HashSet::new(); + let mut family_remote = expected_family; + + for target in targets { + if !ids.insert(target.id.clone()) { + return Err(ClientError::InvalidConfig { + reason: format!("duplicate managed target id {}", target.id), + }); + } + if !remotes.insert(target.remote) { + return Err(ClientError::InvalidConfig { + reason: format!("duplicate managed target remote {}", target.remote), + }); + } + if let Some(family) = family_remote { + if family.is_ipv4() != target.remote.is_ipv4() { + return Err(ClientError::InvalidConfig { + reason: "managed client group targets must use one address family".to_owned(), + }); + } + } else { + family_remote = Some(target.remote); + } + } + + family_remote.ok_or_else(|| ClientError::InvalidConfig { + reason: "managed client group requires at least one target".to_owned(), + }) +} + +fn validate_group_family_flags( + config: &ClientConfig, + remote: SocketAddr, +) -> Result<(), ClientError> { + if config.socket_config.ipv4_only && config.socket_config.ipv6_only { + return Err(ClientError::InvalidConfig { + reason: "ipv4_only and ipv6_only cannot both be true".to_owned(), + }); + } + if config.socket_config.ipv4_only && remote.is_ipv6() { + return Err(ClientError::InvalidConfig { + reason: "ipv4_only cannot be used with IPv6 group targets".to_owned(), + }); + } + if config.socket_config.ipv6_only && remote.is_ipv4() { + return Err(ClientError::InvalidConfig { + reason: "ipv6_only cannot be used with IPv4 group targets".to_owned(), + }); + } + Ok(()) +} + +fn drive_open_attempts( + config: &ClientConfig, + socket: &UdpSocket, + shared: &GroupShared, + now: Instant, +) { + for target in all_targets(shared) { + let mut target = target.lock().expect("target mutex poisoned"); + let TargetStatus::Opening { + attempt, + next_send_at, + } = target.status + else { + continue; + }; + + if next_send_at > now { + continue; + } + + if attempt >= config.open_timeouts.len() { + target.mark_finished(ManagedTargetEndReason::OpenFailed { + message: ClientError::OpenTimeout.to_string(), + }); + continue; + } + + match socket.send_to(&target.open_packet, target.remote) { + Ok(_) => { + target.status = TargetStatus::Opening { + attempt: attempt + 1, + next_send_at: now + config.open_timeouts[attempt], + }; + } + Err(err) => { + target.mark_finished(ManagedTargetEndReason::Failed { + message: ClientError::Socket(err).to_string(), + }); + } + } + } +} + +fn poll_active_timeouts(shared: &GroupShared, now: Instant) { + for target in all_targets(shared) { + let mut target = target.lock().expect("target mutex poisoned"); + if !matches!( + target.status, + TargetStatus::Active | TargetStatus::Draining { .. } + ) { + continue; + } + match target.runtime.poll_timeouts_at(now) { + Ok(events) => publish_events(&shared.hub, &mut target, events), + Err(err) => target.mark_finished(ManagedTargetEndReason::Failed { + message: err.to_string(), + }), + } + } +} + +fn finish_completed_targets(socket: &UdpSocket, shared: &GroupShared, now: Instant) { + for target in all_targets(shared) { + let mut target = target.lock().expect("target mutex poisoned"); + match target.status { + TargetStatus::Active if target.runtime.is_run_complete() => { + if target.runtime.is_peer_closed() { + target.mark_finished(ManagedTargetEndReason::PeerClosed); + } else if target.runtime.has_timed_out_metadata() { + target.status = TargetStatus::Draining { + deadline: now + GROUP_FINAL_DRAIN, + }; + } else { + close_locked_target( + socket, + &shared.hub, + &mut target, + ManagedTargetEndReason::TestComplete, + ); + } + } + TargetStatus::Draining { deadline } => { + if target.runtime.is_peer_closed() { + target.mark_finished(ManagedTargetEndReason::PeerClosed); + } else if now >= deadline || !target.runtime.has_timed_out_metadata() { + close_locked_target( + socket, + &shared.hub, + &mut target, + ManagedTargetEndReason::TestComplete, + ); + } + } + _ => {} + } + } +} + +fn send_echo_to_target( + socket: &UdpSocket, + shared: &GroupShared, + target: Arc>, + scheduled_at: Instant, +) { + let mut target = target.lock().expect("target mutex poisoned"); + if !matches!(target.status, TargetStatus::Active) { + return; + } + let remote = target.remote; + let result = target + .runtime + .send_probe_for_deadline(scheduled_at, |packet| { + let sent_at = ClientTimestamp::now(); + let send_call_start = Instant::now(); + let bytes = socket.send_to(packet, remote)?; + let send_call = send_call_start.elapsed(); + Ok(SendProbeResult { + sent_at, + bytes, + send_call, + }) + }); + + match result { + Ok(events) => { + // Publish while the target lock is still held. EventHub publish is + // bounded and nonblocking, and this preserves per-target + // EchoSent-before-EchoReply ordering for immediately returned UDP + // replies. + publish_events(&shared.hub, &mut target, events); + } + Err(err) => target.mark_finished(ManagedTargetEndReason::Failed { + message: err.to_string(), + }), + } +} + +fn collect_finished_targets(shared: &GroupShared, records: &mut Vec) { + let finished = { + let mut registry = shared + .registry + .lock() + .expect("target registry mutex poisoned"); + let ids: Vec = registry + .targets + .iter() + .filter_map(|(id, target)| { + let target = target.lock().expect("target mutex poisoned"); + matches!(target.status, TargetStatus::Finished).then(|| id.clone()) + }) + .collect(); + let mut finished = Vec::with_capacity(ids.len()); + for id in ids { + if let Some(target) = registry.targets.remove(&id) { + let remote = target.lock().expect("target mutex poisoned").remote; + registry.remotes.remove(&remote); + finished.push(target); + } + } + finished + }; + + for target in finished { + let target = target.lock().expect("target mutex poisoned"); + let reason = + target + .final_reason + .clone() + .unwrap_or_else(|| ManagedTargetEndReason::Failed { + message: "target finished without a reason".to_owned(), + }); + records.push(target.outcome(reason)); + } +} + +fn cancel_remaining_targets( + socket: &UdpSocket, + shared: &GroupShared, + records: &mut Vec, +) { + let targets = drain_registry(shared); + for target in targets { + records.push(close_target( + socket, + &shared.hub, + target, + ManagedTargetEndReason::Cancelled, + )); + } +} + +fn remove_all_targets( + socket: &UdpSocket, + shared: &GroupShared, + records: &mut Vec, + reason: ManagedTargetEndReason, +) { + let targets = drain_registry(shared); + for target in targets { + records.push(close_target(socket, &shared.hub, target, reason.clone())); + } +} + +fn drain_registry(shared: &GroupShared) -> Vec>> { + let mut registry = shared + .registry + .lock() + .expect("target registry mutex poisoned"); + registry.remotes.clear(); + registry.targets.drain().map(|(_, target)| target).collect() +} + +fn close_target( + socket: &UdpSocket, + hub: &EventHub, + target: Arc>, + reason: ManagedTargetEndReason, +) -> ManagedTargetOutcome { + let mut target = target.lock().expect("target mutex poisoned"); + close_locked_target(socket, hub, &mut target, reason.clone()); + let reason = target.final_reason.clone().unwrap_or(reason); + target.outcome(reason) +} + +fn close_locked_target( + socket: &UdpSocket, + hub: &EventHub, + target: &mut TargetState, + reason: ManagedTargetEndReason, +) { + if target.runtime.is_open() && !target.runtime.is_peer_closed() { + let remote = target.remote; + match target.runtime.close_with(|packet| { + socket.send_to(packet, remote)?; + Ok(()) + }) { + Ok(events) => publish_events(hub, target, events), + Err(err) => { + target.mark_finished(ManagedTargetEndReason::Failed { + message: err.to_string(), + }); + return; + } + } + } + target.mark_finished(reason); +} + +fn active_targets(shared: &GroupShared) -> Vec>> { + let mut targets: Vec<_> = all_targets(shared) + .into_iter() + .filter(|target| { + let target = target.lock().expect("target mutex poisoned"); + matches!(target.status, TargetStatus::Active) + }) + .collect(); + targets.sort_by_key(|target| target.lock().expect("target mutex poisoned").order); + targets +} + +fn all_targets(shared: &GroupShared) -> Vec>> { + shared + .registry + .lock() + .expect("target registry mutex poisoned") + .targets + .values() + .cloned() + .collect() +} + +fn registry_is_empty(shared: &GroupShared) -> bool { + shared + .registry + .lock() + .expect("target registry mutex poisoned") + .targets + .is_empty() +} + +#[derive(Debug)] +struct PacingRuntime { + mode: ManagedGroupPacing, + signature: Vec, + next_slot_at: Option, + slot_index: usize, + next_burst_at: Option, +} + +impl PacingRuntime { + fn new(mode: ManagedGroupPacing) -> Self { + Self { + mode, + signature: Vec::new(), + next_slot_at: None, + slot_index: 0, + next_burst_at: None, + } + } + + fn reconcile(&mut self, active: &[Arc>], now: Instant) { + let signature: Vec = active + .iter() + .map(|target| target.lock().expect("target mutex poisoned").id.clone()) + .collect(); + if signature == self.signature { + return; + } + self.signature = signature; + self.slot_index = 0; + if active.is_empty() { + self.next_slot_at = None; + self.next_burst_at = None; + } else { + match self.mode { + ManagedGroupPacing::Staggered => self.next_slot_at = Some(now), + ManagedGroupPacing::Burst => { + self.next_burst_at.get_or_insert(now); + } + } + } + } + + fn send_due( + &self, + active: &[Arc>], + now: Instant, + _interval: Duration, + ) -> bool { + if active.is_empty() { + return false; + } + match self.mode { + ManagedGroupPacing::Staggered => { + self.next_slot_at.is_some_and(|deadline| deadline <= now) + } + ManagedGroupPacing::Burst => self.next_burst_at.is_some_and(|deadline| deadline <= now), + } + } + + fn next_staggered( + &mut self, + active: &[Arc>], + interval: Duration, + ) -> Option<(Arc>, Instant)> { + let scheduled_at = self.next_slot_at?; + if active.is_empty() { + self.next_slot_at = None; + return None; + } + let target = active[self.slot_index % active.len()].clone(); + self.slot_index = (self.slot_index + 1) % active.len(); + self.next_slot_at = Some(scheduled_at + divide_duration(interval, active.len())); + Some((target, scheduled_at)) + } + + fn next_burst(&mut self, interval: Duration) -> Option { + let scheduled_at = self.next_burst_at?; + self.next_burst_at = Some(scheduled_at + interval); + Some(scheduled_at) + } + + fn next_wakeup(&self) -> Option { + match self.mode { + ManagedGroupPacing::Staggered => self.next_slot_at, + ManagedGroupPacing::Burst => self.next_burst_at, + } + } +} + +fn wait_for_next_scheduler_wakeup( + control_rx: &mpsc::Receiver, + shared: &GroupShared, + pacing: &PacingRuntime, + active: &[Arc>], +) -> Option { + let open_deadline = next_open_deadline(shared); + let pacing_deadline = (!active.is_empty()).then(|| pacing.next_wakeup()).flatten(); + let deadline = [open_deadline, pacing_deadline].into_iter().flatten().min(); + let sleep_for = deadline + .and_then(|deadline| deadline.checked_duration_since(Instant::now())) + .map(|duration| duration.min(MAX_SLEEP)) + .unwrap_or(IDLE_SLEEP); + + match control_rx.recv_timeout(sleep_for) { + Ok(message) => Some(message), + Err(mpsc::RecvTimeoutError::Timeout) => None, + Err(mpsc::RecvTimeoutError::Disconnected) => { + shared.cancellation.cancel(); + None + } + } +} + +fn next_open_deadline(shared: &GroupShared) -> Option { + all_targets(shared) + .into_iter() + .filter_map(|target| { + let target = target.lock().expect("target mutex poisoned"); + match target.status { + TargetStatus::Opening { next_send_at, .. } => Some(next_send_at), + _ => None, + } + }) + .min() +} + +fn divide_duration(duration: Duration, divisor: usize) -> Duration { + if divisor == 0 { + return duration; + } + let nanos = (duration.as_nanos() / divisor as u128).max(1); + Duration::from_nanos(u64::try_from(nanos).unwrap_or(u64::MAX)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + config::NegotiationPolicy, managed::SubscriberOverflow, EventSubscriptionError, WarningKind, + }; + use irtt_proto::{ + echo_packet_len, + flags::{self, FLAG_OPEN, FLAG_REPLY}, + layout::PacketLayout, + Clock, Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION, + }; + use std::sync::mpsc; + + const TOKEN: u64 = 0x1234_5678_90ab_cdef; + + #[derive(Debug)] + struct FakeServer { + addr: SocketAddr, + _observations: mpsc::Receiver, + done: JoinHandle<()>, + } + + impl FakeServer { + fn join(self) { + self.done.join().unwrap(); + } + } + + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum ServerObservation { + Open { at: Instant }, + Echo { seq: u32, at: Instant }, + Close { at: Instant }, + } + + fn test_params(duration: Option, interval: Duration) -> Params { + Params { + protocol_version: PROTOCOL_VERSION, + duration_ns: duration.map_or(0, duration_ns_i64), + interval_ns: duration_ns_i64(interval), + length: 0, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + dscp: 0, + server_fill: None, + } + } + + fn duration_ns_i64(duration: Duration) -> i64 { + i64::try_from(duration.as_nanos()).expect("test duration fits i64 nanoseconds") + } + + fn group_config( + duration: Option, + interval: Duration, + pacing: ManagedGroupPacing, + ) -> ManagedClientGroupConfig { + ManagedClientGroupConfig { + client: ClientConfig { + server_addr: "127.0.0.1:1".to_owned(), + duration, + interval, + negotiation_policy: NegotiationPolicy::Strict, + open_timeouts: vec![Duration::from_millis(200)], + probe_timeout: Duration::from_millis(80), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(20)), + ..Default::default() + }, + ..ClientConfig::default() + }, + pacing, + } + } + + fn target(id: &str, remote: SocketAddr) -> ManagedTargetConfig { + ManagedTargetConfig { + id: TargetId::from(id), + remote, + auth: None, + } + } + + fn start_echo_server(params: Params, open_delay: Duration) -> FakeServer { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let addr = socket.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + let done = thread::spawn(move || { + socket + .set_read_timeout(Some(Duration::from_millis(800))) + .unwrap(); + let mut opened = false; + while let Some((packet, peer)) = recv_request_timeout(&socket) { + if packet[3] & FLAG_OPEN != 0 { + if !opened { + opened = true; + tx.send(ServerObservation::Open { at: Instant::now() }) + .unwrap(); + if open_delay > Duration::ZERO { + thread::sleep(open_delay); + } + } + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) + .unwrap(); + continue; + } + + if packet[3] & flags::FLAG_CLOSE != 0 { + tx.send(ServerObservation::Close { at: Instant::now() }) + .unwrap(); + break; + } + + let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); + tx.send(ServerObservation::Echo { + seq, + at: Instant::now(), + }) + .unwrap(); + let ts = TimestampFields { + recv_wall: Some(1_000_000_000), + recv_mono: Some(100_000), + send_wall: Some(1_000_000_000), + send_mono: Some(100_000), + ..Default::default() + }; + socket + .send_to(&echo_reply_packet(TOKEN, seq, ¶ms, &ts), peer) + .unwrap(); + } + }); + FakeServer { + addr, + _observations: rx, + done, + } + } + + fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec, SocketAddr)> { + let mut buf = [0_u8; 2048]; + socket + .recv_from(&mut buf) + .ok() + .map(|(size, peer)| (buf[..size].to_vec(), peer)) + } + + fn open_reply(flags: u8, token: u64, params: &Params) -> Vec { + let mut packet = Vec::new(); + packet.extend_from_slice(&MAGIC); + packet.push(flags); + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(¶ms.encode()); + packet + } + + fn echo_reply_packet( + token: u64, + seq: u32, + params: &Params, + timestamps: &TimestampFields, + ) -> Vec { + let layout = PacketLayout::echo(false, params); + let packet_len = echo_packet_len(false, params).unwrap(); + let mut packet = Vec::with_capacity(packet_len); + + packet.extend_from_slice(&MAGIC); + packet.push(FLAG_REPLY); + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(&seq.to_le_bytes()); + + if layout.recv_count { + packet.extend_from_slice(&42_u32.to_le_bytes()); + } + if layout.recv_window { + packet.extend_from_slice(&0x07_u64.to_le_bytes()); + } + if layout.recv_wall { + packet.extend_from_slice(×tamps.recv_wall.unwrap_or(0).to_le_bytes()); + } + if layout.recv_mono { + packet.extend_from_slice(×tamps.recv_mono.unwrap_or(0).to_le_bytes()); + } + if layout.midpoint_wall { + packet.extend_from_slice(×tamps.midpoint_wall.unwrap_or(0).to_le_bytes()); + } + if layout.midpoint_mono { + packet.extend_from_slice(×tamps.midpoint_mono.unwrap_or(0).to_le_bytes()); + } + if layout.send_wall { + packet.extend_from_slice(×tamps.send_wall.unwrap_or(0).to_le_bytes()); + } + if layout.send_mono { + packet.extend_from_slice(×tamps.send_mono.unwrap_or(0).to_le_bytes()); + } + + packet.resize(packet_len, 0); + packet + } + + fn recv_event_with_timeout(sub: &TargetEventSubscription) -> TargetEvent { + let deadline = Instant::now() + Duration::from_secs(2); + loop { + match sub.try_recv() { + Ok(Some(event)) => return event, + Ok(None) if Instant::now() < deadline => thread::sleep(Duration::from_millis(1)), + Ok(None) => panic!("timed out waiting for group event"), + Err(err) => panic!("subscription ended while waiting for event: {err}"), + } + } + } + + fn drain_after_join(sub: &TargetEventSubscription) -> Vec { + let mut events = Vec::new(); + loop { + match sub.try_recv() { + Ok(Some(event)) => events.push(event), + Ok(None) => thread::sleep(Duration::from_millis(1)), + Err(EventSubscriptionError::Disconnected) => return events, + } + } + } + + fn sent_events_after_start(events: &[TargetEvent]) -> Vec<(&TargetId, Instant)> { + let both_started_at = events + .iter() + .filter_map(|event| match &event.event { + ClientEvent::SessionStarted { at, .. } => Some(at.mono), + _ => None, + }) + .max() + .expect("expected at least one SessionStarted event"); + + let mut sent: Vec<_> = events + .iter() + .filter_map(|event| match &event.event { + ClientEvent::EchoSent { sent_at, .. } if sent_at.mono >= both_started_at => { + Some((&event.target, sent_at.mono)) + } + _ => None, + }) + .collect(); + sent.sort_by_key(|(_, at)| *at); + sent + } + + #[test] + fn target_event_hub_uses_generic_overflow_behavior() { + let hub = EventHub::::new(); + let sub = hub + .subscribe(SubscriberConfig { + capacity: 1, + overflow: SubscriberOverflow::DropOldest, + }) + .unwrap(); + + hub.publish(TargetEvent { + target: TargetId::from("a"), + event: ClientEvent::Warning { + kind: WarningKind::UntrackedReply, + message: "old".to_owned(), + at: ClientTimestamp::now(), + }, + }); + hub.publish(TargetEvent { + target: TargetId::from("b"), + event: ClientEvent::Warning { + kind: WarningKind::UntrackedReply, + message: "new".to_owned(), + at: ClientTimestamp::now(), + }, + }); + + let event = sub.try_recv().unwrap().unwrap(); + assert_eq!(event.target, TargetId::from("b")); + assert!(sub.try_recv().unwrap().is_none()); + } + + #[test] + fn rejects_duplicate_resolved_remote_targets() { + let remote: SocketAddr = "127.0.0.1:12345".parse().unwrap(); + let err = ManagedClientGroup::start( + group_config( + Some(Duration::from_millis(10)), + Duration::from_millis(10), + ManagedGroupPacing::Staggered, + ), + vec![target("a", remote), target("b", remote)], + ) + .unwrap_err(); + + assert!(matches!(err, ClientError::InvalidConfig { .. })); + assert!(err.to_string().contains("duplicate managed target remote")); + } + + #[test] + fn rejects_mixed_address_families() { + let err = ManagedClientGroup::start( + group_config( + Some(Duration::from_millis(10)), + Duration::from_millis(10), + ManagedGroupPacing::Staggered, + ), + vec![ + target("a", "127.0.0.1:12345".parse().unwrap()), + target("b", "[::1]:12345".parse().unwrap()), + ], + ) + .unwrap_err(); + + assert!(matches!(err, ClientError::InvalidConfig { .. })); + assert!(err.to_string().contains("one address family")); + } + + #[test] + fn shared_socket_smoke_two_targets_emit_scoped_events() { + let duration = Duration::from_millis(70); + let interval = Duration::from_millis(20); + let params = test_params(Some(duration), interval); + let a = start_echo_server(params.clone(), Duration::ZERO); + let b = start_echo_server(params, Duration::ZERO); + + let (session, sub) = ManagedClientGroup::start_with_subscription( + group_config(Some(duration), interval, ManagedGroupPacing::Staggered), + vec![target("a", a.addr), target("b", b.addr)], + SubscriberConfig { + capacity: 256, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + let outcome = session.join().unwrap(); + let events = drain_after_join(&sub); + a.join(); + b.join(); + + assert_eq!( + outcome.end_reason, + ManagedGroupEndReason::AllTargetsComplete + ); + assert_eq!(outcome.targets.len(), 2); + for id in [TargetId::from("a"), TargetId::from("b")] { + assert!(events.iter().any(|event| { + event.target == id && matches!(event.event, ClientEvent::SessionStarted { .. }) + })); + assert!(events.iter().any(|event| { + event.target == id && matches!(event.event, ClientEvent::EchoSent { .. }) + })); + assert!(events.iter().any(|event| { + event.target == id && matches!(event.event, ClientEvent::EchoReply { .. }) + })); + } + } + + #[test] + fn staggered_pacing_does_not_burst_active_targets() { + let duration = Duration::from_millis(170); + let interval = Duration::from_millis(80); + let params = test_params(Some(duration), interval); + let a = start_echo_server(params.clone(), Duration::ZERO); + let b = start_echo_server(params, Duration::ZERO); + + let (session, sub) = ManagedClientGroup::start_with_subscription( + group_config(Some(duration), interval, ManagedGroupPacing::Staggered), + vec![target("a", a.addr), target("b", b.addr)], + SubscriberConfig { + capacity: 256, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + let _ = session.join().unwrap(); + let events = drain_after_join(&sub); + a.join(); + b.join(); + + let sent = sent_events_after_start(&events); + let pair = sent + .windows(2) + .find(|window| window[0].0 != window[1].0) + .expect("expected sends for two targets"); + let delta = pair[1].1.duration_since(pair[0].1); + assert!( + delta >= Duration::from_millis(20), + "staggered sends were too close: {delta:?}" + ); + } + + #[test] + fn burst_pacing_sends_active_targets_back_to_back() { + let duration = Duration::from_millis(190); + let interval = Duration::from_millis(80); + let params = test_params(Some(duration), interval); + let a = start_echo_server(params.clone(), Duration::ZERO); + let b = start_echo_server(params, Duration::ZERO); + + let (session, sub) = ManagedClientGroup::start_with_subscription( + group_config(Some(duration), interval, ManagedGroupPacing::Burst), + vec![target("a", a.addr), target("b", b.addr)], + SubscriberConfig { + capacity: 256, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + let _ = session.join().unwrap(); + let events = drain_after_join(&sub); + a.join(); + b.join(); + + let sent = sent_events_after_start(&events); + let pair = sent + .windows(2) + .find(|window| window[0].0 != window[1].0) + .expect("expected sends for two targets"); + let delta = pair[1].1.duration_since(pair[0].1); + assert!( + delta <= Duration::from_millis(20), + "burst sends were too far apart: {delta:?}" + ); + } + + #[test] + fn delayed_open_target_joins_after_active_target_is_already_sending() { + let duration = Duration::from_millis(240); + let interval = Duration::from_millis(50); + let params = test_params(Some(duration), interval); + let a = start_echo_server(params.clone(), Duration::ZERO); + let b = start_echo_server(params, Duration::from_millis(120)); + + let (session, sub) = ManagedClientGroup::start_with_subscription( + group_config(Some(duration), interval, ManagedGroupPacing::Staggered), + vec![target("a", a.addr), target("b", b.addr)], + SubscriberConfig { + capacity: 512, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + let _ = session.join().unwrap(); + let events = drain_after_join(&sub); + a.join(); + b.join(); + + let b_started_at = events + .iter() + .find_map(|event| match (&event.target, &event.event) { + (id, ClientEvent::SessionStarted { at, .. }) if id.as_str() == "b" => Some(at.mono), + _ => None, + }) + .expect("target b should eventually open"); + let a_sent_before_b = events + .iter() + .filter(|event| { + event.target.as_str() == "a" + && matches!( + &event.event, + ClientEvent::EchoSent { sent_at, .. } if sent_at.mono < b_started_at + ) + }) + .count(); + + assert!( + a_sent_before_b >= 2, + "active target did not continue sending while b was opening" + ); + assert!(events.iter().any(|event| { + event.target.as_str() == "b" && matches!(event.event, ClientEvent::EchoReply { .. }) + })); + } + + #[test] + fn update_targets_removes_and_adds_without_restarting_unchanged_target() { + let interval = Duration::from_millis(30); + let params = test_params(None, interval); + let a = start_echo_server(params.clone(), Duration::ZERO); + let b = start_echo_server(params.clone(), Duration::ZERO); + let c = start_echo_server(params, Duration::ZERO); + + let (session, sub) = ManagedClientGroup::start_with_subscription( + group_config(None, interval, ManagedGroupPacing::Burst), + vec![target("a", a.addr), target("b", b.addr)], + SubscriberConfig { + capacity: 512, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + let mut saw_a_reply = false; + let mut saw_b_reply = false; + while !(saw_a_reply && saw_b_reply) { + let event = recv_event_with_timeout(&sub); + saw_a_reply |= event.target.as_str() == "a" + && matches!(event.event, ClientEvent::EchoReply { .. }); + saw_b_reply |= event.target.as_str() == "b" + && matches!(event.event, ClientEvent::EchoReply { .. }); + } + + session + .update_targets(vec![target("a", a.addr), target("c", c.addr)]) + .unwrap(); + + let deadline = Instant::now() + Duration::from_millis(220); + let mut saw_a_after_update = false; + let mut saw_c_after_update = false; + let mut saw_b_echo_after_update = false; + while Instant::now() < deadline && !(saw_a_after_update && saw_c_after_update) { + match sub.try_recv() { + Ok(Some(event)) => { + saw_a_after_update |= event.target.as_str() == "a" + && matches!(event.event, ClientEvent::EchoReply { .. }); + saw_c_after_update |= event.target.as_str() == "c" + && matches!(event.event, ClientEvent::EchoReply { .. }); + saw_b_echo_after_update |= event.target.as_str() == "b" + && matches!( + event.event, + ClientEvent::EchoSent { .. } | ClientEvent::EchoReply { .. } + ); + } + Ok(None) => thread::sleep(Duration::from_millis(1)), + Err(err) => panic!("subscription ended unexpectedly: {err}"), + } + } + + session.stop(); + let outcome = session.join().unwrap(); + a.join(); + b.join(); + c.join(); + + assert_eq!(outcome.end_reason, ManagedGroupEndReason::Cancelled); + assert!(saw_a_after_update); + assert!(saw_c_after_update); + assert!(!saw_b_echo_after_update); + assert!(outcome + .targets + .iter() + .any(|target| target.id.as_str() == "b" + && target.end_reason == ManagedTargetEndReason::Removed)); + } + + #[test] + fn stop_join_cleans_up_group_threads() { + let interval = Duration::from_millis(20); + let params = test_params(None, interval); + let a = start_echo_server(params.clone(), Duration::ZERO); + let b = start_echo_server(params, Duration::ZERO); + + let session = ManagedClientGroup::start( + group_config(None, interval, ManagedGroupPacing::Staggered), + vec![target("a", a.addr), target("b", b.addr)], + ) + .unwrap(); + + thread::sleep(Duration::from_millis(60)); + session.stop(); + session.stop(); + let outcome = session.join().unwrap(); + a.join(); + b.join(); + + assert_eq!(outcome.end_reason, ManagedGroupEndReason::Cancelled); + assert_eq!(outcome.targets.len(), 2); + } +} diff --git a/crates/irtt-client/src/managed/hub.rs b/crates/irtt-client/src/managed/hub.rs index 30b11cc..2fd4cd3 100644 --- a/crates/irtt-client/src/managed/hub.rs +++ b/crates/irtt-client/src/managed/hub.rs @@ -45,20 +45,20 @@ pub enum SubscriberOverflow { Disconnect, } -/// Publish/subscribe fan-out for [`ClientEvent`] values. +/// Publish/subscribe fan-out for managed event values. /// /// `EventHub` is used by managed sessions internally and is also exported for /// callers that want the same bounded subscription behavior around their own /// event producer. Publishing clones each event once per subscriber. -#[derive(Debug, Clone)] -pub struct EventHub { - inner: Arc, +#[derive(Debug)] +pub struct EventHub { + inner: Arc>, } #[derive(Debug)] -struct HubInner { +struct HubInner { next_id: AtomicU64, - subscribers: Mutex>>, + subscribers: Mutex>>>, } /// Handle for receiving managed client events. @@ -68,26 +68,34 @@ struct HubInner { /// when that queue is full. Dropping the handle unregisters the subscriber. #[must_use = "dropping the subscription unregisters it"] #[derive(Debug)] -pub struct EventSubscription { +pub struct EventSubscription { id: u64, - hub: Weak, - inner: Arc, + hub: Weak>, + inner: Arc>, } #[derive(Debug)] -struct SubscriberInner { - state: Mutex, +struct SubscriberInner { + state: Mutex>, available: Condvar, config: SubscriberConfig, } #[derive(Debug)] -struct SubscriberState { - queue: VecDeque, +struct SubscriberState { + queue: VecDeque, connected: bool, } -impl EventHub { +impl Clone for EventHub { + fn clone(&self) -> Self { + Self { + inner: self.inner.clone(), + } + } +} + +impl EventHub { /// Create an empty event hub with no subscribers. pub fn new() -> Self { Self { @@ -104,7 +112,7 @@ impl EventHub { pub fn subscribe( &self, config: SubscriberConfig, - ) -> Result { + ) -> Result, crate::ClientError> { if config.capacity == 0 { return Err(crate::ClientError::InvalidConfig { reason: "subscriber capacity must be greater than zero".to_owned(), @@ -133,13 +141,41 @@ impl EventHub { }) } + /// Disconnect all subscribers after leaving their already queued events + /// available to drain. + pub fn disconnect_all(&self) { + let subscribers: Vec>> = self + .inner + .subscribers + .lock() + .expect("event hub mutex poisoned") + .drain() + .map(|(_, subscriber)| subscriber) + .collect(); + + for subscriber in subscribers { + subscriber.disconnect(); + } + } + + #[cfg(test)] + fn subscriber_count(&self) -> usize { + self.inner + .subscribers + .lock() + .expect("event hub mutex poisoned") + .len() + } +} + +impl EventHub { /// Publish an event to all currently connected subscribers. /// /// Slow subscribers are handled according to their /// [`SubscriberOverflow`] policy. Publishing does not block waiting for a /// subscriber to consume queued events. - pub fn publish(&self, event: ClientEvent) { - let subscribers: Vec<(u64, Arc)> = self + pub fn publish(&self, event: T) { + let subscribers: Vec<(u64, Arc>)> = self .inner .subscribers .lock() @@ -170,47 +206,21 @@ impl EventHub { subscribers.remove(&id); } } - - /// Disconnect all subscribers after leaving their already queued events - /// available to drain. - pub fn disconnect_all(&self) { - let subscribers: Vec> = self - .inner - .subscribers - .lock() - .expect("event hub mutex poisoned") - .drain() - .map(|(_, subscriber)| subscriber) - .collect(); - - for subscriber in subscribers { - subscriber.disconnect(); - } - } - - #[cfg(test)] - fn subscriber_count(&self) -> usize { - self.inner - .subscribers - .lock() - .expect("event hub mutex poisoned") - .len() - } } -impl Default for EventHub { +impl Default for EventHub { fn default() -> Self { Self::new() } } -impl EventSubscription { +impl EventSubscription { /// Block until the next queued event is available. /// /// If the subscription is disconnected, this returns /// [`EventSubscriptionError::Disconnected`] after any already queued events /// have been drained. - pub fn recv(&self) -> Result { + pub fn recv(&self) -> Result { let mut state = self.inner.state.lock().expect("subscriber mutex poisoned"); loop { if let Some(event) = state.queue.pop_front() { @@ -233,7 +243,7 @@ impl EventSubscription { /// is queued. If the subscription is disconnected, this returns /// [`EventSubscriptionError::Disconnected`] after any already queued events /// have been drained. - pub fn try_recv(&self) -> Result, EventSubscriptionError> { + pub fn try_recv(&self) -> Result, EventSubscriptionError> { let mut state = self.inner.state.lock().expect("subscriber mutex poisoned"); if let Some(event) = state.queue.pop_front() { return Ok(Some(event)); @@ -245,7 +255,7 @@ impl EventSubscription { } } -impl Drop for EventSubscription { +impl Drop for EventSubscription { fn drop(&mut self) { self.inner.disconnect(); if let Some(hub) = self.hub.upgrade() { @@ -257,8 +267,8 @@ impl Drop for EventSubscription { } } -impl SubscriberInner { - fn publish(&self, event: ClientEvent) -> bool { +impl SubscriberInner { + fn publish(&self, event: T) -> bool { let mut state = self.state.lock().expect("subscriber mutex poisoned"); if !state.connected { return false; @@ -411,7 +421,7 @@ mod tests { #[test] fn dropping_subscription_unregisters_from_hub() { - let hub = EventHub::new(); + let hub: EventHub = EventHub::new(); let sub = hub.subscribe(SubscriberConfig::default()).unwrap(); assert_eq!(hub.subscriber_count(), 1); @@ -422,7 +432,7 @@ mod tests { #[test] fn disconnect_all_wakes_blocking_receivers() { - let hub = EventHub::new(); + let hub: EventHub = EventHub::new(); let sub = hub.subscribe(SubscriberConfig::default()).unwrap(); hub.disconnect_all(); diff --git a/crates/irtt-client/src/managed/mod.rs b/crates/irtt-client/src/managed/mod.rs index 458edfd..6db649b 100644 --- a/crates/irtt-client/src/managed/mod.rs +++ b/crates/irtt-client/src/managed/mod.rs @@ -1,7 +1,13 @@ mod cancellation; +mod group; mod hub; mod runner; pub use cancellation::CancellationToken; +pub use group::{ + ManagedClientGroup, ManagedClientGroupConfig, ManagedClientGroupSession, ManagedGroupEndReason, + ManagedGroupOutcome, ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, + ManagedTargetOutcome, TargetEvent, TargetEventSubscription, TargetId, +}; pub use hub::{EventHub, EventSubscription, SubscriberConfig, SubscriberOverflow}; pub use runner::{ManagedClient, ManagedClientSession, SessionEndReason, SessionOutcome}; diff --git a/crates/irtt-client/src/receive.rs b/crates/irtt-client/src/receive.rs index 9bb64d3..f239d42 100644 --- a/crates/irtt-client/src/receive.rs +++ b/crates/irtt-client/src/receive.rs @@ -15,6 +15,14 @@ pub(crate) struct ReceivedDatagram { pub(crate) meta: ReceiveMeta, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct ReceivedDatagramFrom { + pub(crate) len: usize, + pub(crate) source: SocketAddr, + pub(crate) received_at: ClientTimestamp, + pub(crate) meta: ReceiveMeta, +} + #[cfg(not(all(target_os = "linux", feature = "ancillary")))] pub(crate) fn recv_datagram( socket: &UdpSocket, @@ -30,6 +38,22 @@ pub(crate) fn recv_datagram( }) } +#[cfg(not(all(target_os = "linux", feature = "ancillary")))] +pub(crate) fn recv_datagram_from( + socket: &UdpSocket, + buf: &mut [u8], +) -> Result { + let (len, source) = socket.recv_from(buf)?; + let received_at = ClientTimestamp::now(); + + Ok(ReceivedDatagramFrom { + len, + source, + received_at, + meta: ReceiveMeta::default(), + }) +} + #[cfg(all(target_os = "linux", feature = "ancillary"))] pub(crate) fn recv_datagram( socket: &UdpSocket, @@ -38,6 +62,14 @@ pub(crate) fn recv_datagram( linux::recv_datagram(socket, buf) } +#[cfg(all(target_os = "linux", feature = "ancillary"))] +pub(crate) fn recv_datagram_from( + socket: &UdpSocket, + buf: &mut [u8], +) -> Result { + linux::recv_datagram_from(socket, buf) +} + #[cfg(not(all(target_os = "linux", feature = "ancillary")))] pub(crate) fn configure_receive_metadata( _socket: &UdpSocket, diff --git a/crates/irtt-client/src/receive/linux.rs b/crates/irtt-client/src/receive/linux.rs index a410b91..ca0c3a4 100644 --- a/crates/irtt-client/src/receive/linux.rs +++ b/crates/irtt-client/src/receive/linux.rs @@ -1,13 +1,17 @@ #![allow(unsafe_code)] use std::{ io, mem, - net::{SocketAddr, UdpSocket}, + net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, UdpSocket}, os::fd::AsRawFd, ptr, time::{Duration, SystemTime, UNIX_EPOCH}, }; -use crate::{metadata::ReceiveMeta, receive::ReceivedDatagram, timing::ClientTimestamp}; +use crate::{ + metadata::ReceiveMeta, + receive::{ReceivedDatagram, ReceivedDatagramFrom}, + timing::ClientTimestamp, +}; const CONTROL_LEN: usize = 128; @@ -87,6 +91,61 @@ pub(crate) fn recv_datagram( }) } +pub(crate) fn recv_datagram_from( + socket: &UdpSocket, + buf: &mut [u8], +) -> Result { + let mut control = ControlBuffer::new(); + let mut name: libc::sockaddr_storage = unsafe { mem::zeroed() }; + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr().cast(), + iov_len: buf.len(), + }; + let mut msg: libc::msghdr = unsafe { mem::zeroed() }; + msg.msg_name = (&mut name as *mut libc::sockaddr_storage).cast(); + msg.msg_namelen = mem::size_of::() as libc::socklen_t; + msg.msg_iov = &mut iov; + msg.msg_iovlen = 1; + msg.msg_control = control.as_mut_ptr().cast(); + msg.msg_controllen = control.len(); + + let len = unsafe { + // SAFETY: `msg` points to one writable iovec backed by `buf`, a + // writable source-address buffer, and a writable control buffer. All + // buffers live until `recvmsg` returns. The socket file descriptor is + // borrowed from a valid `UdpSocket`. + libc::recvmsg(socket.as_raw_fd(), &mut msg, 0) + }; + if len < 0 { + return Err(io::Error::last_os_error()); + } + let len = usize::try_from(len).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "recvmsg returned an unrepresentable datagram length", + ) + })?; + let received_at = ClientTimestamp::now(); + let source = unsafe { + // SAFETY: `name` and `msg_namelen` were initialized by a successful + // `recvmsg` call above. + socket_addr_from_storage(&name, msg.msg_namelen)? + }; + let meta = unsafe { + // SAFETY: `msg` was initialized by a successful `recvmsg` call. The + // parser only reads cmsghdr entries within `msg_controllen` and ignores + // short or unrelated control messages. + parse_receive_meta(&msg) + }; + + Ok(ReceivedDatagramFrom { + len, + source, + received_at, + meta, + }) +} + unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta { let mut meta = ReceiveMeta::default(); let mut cmsg = libc::CMSG_FIRSTHDR(msg); @@ -108,6 +167,52 @@ unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta { meta } +unsafe fn socket_addr_from_storage( + storage: &libc::sockaddr_storage, + len: libc::socklen_t, +) -> io::Result { + match i32::from(storage.ss_family) { + libc::AF_INET => { + if usize::try_from(len).unwrap_or(0) < mem::size_of::() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "short IPv4 source address from recvmsg", + )); + } + let addr = ptr::read_unaligned( + (storage as *const libc::sockaddr_storage).cast::(), + ); + let ip = Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr)); + let port = u16::from_be(addr.sin_port); + Ok(SocketAddr::V4(SocketAddrV4::new(ip, port))) + } + libc::AF_INET6 => { + if usize::try_from(len).unwrap_or(0) < mem::size_of::() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "short IPv6 source address from recvmsg", + )); + } + let addr = ptr::read_unaligned( + (storage as *const libc::sockaddr_storage).cast::(), + ); + let ip = Ipv6Addr::from(addr.sin6_addr.s6_addr); + let port = u16::from_be(addr.sin6_port); + let flowinfo = u32::from_be(addr.sin6_flowinfo); + Ok(SocketAddr::V6(SocketAddrV6::new( + ip, + port, + flowinfo, + addr.sin6_scope_id, + ))) + } + family => Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unsupported source address family from recvmsg: {family}"), + )), + } +} + fn is_traffic_class_cmsg(cmsg: &libc::cmsghdr) -> bool { (cmsg.cmsg_level == libc::IPPROTO_IP && cmsg.cmsg_type == libc::IP_TOS) || (cmsg.cmsg_level == libc::IPPROTO_IPV6 && cmsg.cmsg_type == libc::IPV6_TCLASS) diff --git a/crates/irtt-client/src/runtime.rs b/crates/irtt-client/src/runtime.rs new file mode 100644 index 0000000..041a5f4 --- /dev/null +++ b/crates/irtt-client/src/runtime.rs @@ -0,0 +1,864 @@ +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use irtt_proto::{ + close::CloseRequest, decode_echo_reply, echo_packet_len, encode_close_request, + encode_echo_request, encode_open_request, flags, EchoReply, EchoRequest, OpenReply, + OpenRequest, Params, ServerFill, TimestampFields, PROTOCOL_VERSION, +}; + +use crate::{ + config::{ + ClientConfig, RunMode, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_UDP_PAYLOAD_LENGTH, + }, + error::ClientError, + event::{ + ClientEvent, OneWayDelaySample, OpenOutcome, ReceivedStatsSample, RttSample, ServerTiming, + SignedDuration, WarningKind, + }, + metadata::ReceiveMeta, + probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap}, + session::{negotiate_params, ActiveSession, ClientPhase, CloseSource, NegotiatedParams}, + timing::ClientTimestamp, +}; + +pub(crate) const MAX_OPEN_PACKET_SIZE: usize = 512; +const MIN_RECV_BUFFER_SIZE: usize = 2048; + +#[derive(Debug)] +pub(crate) struct SessionRuntime { + config: ClientConfig, + remote: std::net::SocketAddr, + requested: Params, + negotiated: Option, + phase: ClientPhase, + session: Option, +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct SendProbeResult { + pub(crate) sent_at: ClientTimestamp, + pub(crate) bytes: usize, + pub(crate) send_call: Duration, +} + +impl SessionRuntime { + pub(crate) fn new( + config: ClientConfig, + remote: std::net::SocketAddr, + ) -> Result { + if config.max_pending_probes == 0 { + return Err(ClientError::InvalidConfig { + reason: "max_pending_probes must be greater than zero".to_owned(), + }); + } + if config.probe_timeout == Duration::ZERO { + return Err(ClientError::InvalidConfig { + reason: "probe_timeout must be greater than zero".to_owned(), + }); + } + let requested = params_from_config(&config)?; + + Ok(Self { + config, + remote, + requested, + negotiated: None, + phase: ClientPhase::Connected, + session: None, + }) + } + + pub(crate) fn config(&self) -> &ClientConfig { + &self.config + } + + pub(crate) fn has_hmac(&self) -> bool { + self.config.hmac_key.is_some() + } + + pub(crate) fn open_packet(&self) -> Result, ClientError> { + match self.phase { + ClientPhase::Connected => {} + ClientPhase::Open { .. } => return Err(ClientError::AlreadyOpen), + ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), + ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted), + } + + let request = OpenRequest { + params: self.requested.clone(), + close: self.config.run_mode == RunMode::NoTest, + }; + Ok(encode_open_request( + &request, + self.config.hmac_key.as_deref(), + )?) + } + + pub(crate) fn decode_open_reply(&self, packet: &[u8]) -> Result { + Ok(irtt_proto::decode_open_reply( + packet, + self.config.hmac_key.as_deref(), + )?) + } + + pub(crate) fn accept_open_reply( + &mut self, + reply: OpenReply, + now: ClientTimestamp, + before_normal_open: F, + ) -> Result + where + F: FnOnce(&NegotiatedParams) -> Result<(), ClientError>, + { + if reply.params.protocol_version != PROTOCOL_VERSION { + return Err(ClientError::ProtocolVersionMismatch { + requested: PROTOCOL_VERSION, + received: reply.params.protocol_version, + }); + } + + let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE); + match self.config.run_mode { + RunMode::Normal if reply_is_close => Err(ClientError::ServerRejected), + RunMode::Normal if reply.token == 0 => Err(ClientError::ZeroToken), + RunMode::Normal => self.accept_normal_open(reply, now, before_normal_open), + RunMode::NoTest if !reply_is_close => Err(ClientError::UnexpectedNoTestReply), + RunMode::NoTest if reply.token != 0 => { + Err(ClientError::NonZeroNoTestToken { token: reply.token }) + } + RunMode::NoTest => self.accept_no_test_open(reply, now), + } + } + + pub(crate) fn next_send_deadline(&self) -> Option { + let session = self.session.as_ref()?; + if session.sending_done { + return None; + } + Some(session.next_send_at) + } + + pub(crate) fn probe_timeout(&self) -> Duration { + self.config.probe_timeout + } + + pub(crate) fn send_probe_with( + &mut self, + override_ts: Option, + send: F, + ) -> Result, ClientError> + where + F: FnOnce(&[u8]) -> Result, + { + self.send_probe_inner(override_ts, None, send) + } + + pub(crate) fn send_probe_for_deadline( + &mut self, + scheduled_at: Instant, + send: F, + ) -> Result, ClientError> + where + F: FnOnce(&[u8]) -> Result, + { + self.send_probe_inner(None, Some(scheduled_at), send) + } + + fn send_probe_inner( + &mut self, + override_ts: Option, + scheduled_at_override: Option, + send: F, + ) -> Result, ClientError> + where + F: FnOnce(&[u8]) -> Result, + { + let token = match self.phase { + ClientPhase::Open { token } => token, + ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), + ClientPhase::Connected => return Err(ClientError::NotOpen), + ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted), + }; + + let session = self + .session + .as_mut() + .expect("session must exist when phase is Open"); + + if session.sending_done { + return Ok(vec![]); + } + + let now = override_ts.unwrap_or_else(ClientTimestamp::now); + + if let Some(end) = session.end_mono { + if now.mono >= end { + session.sending_done = true; + return Ok(vec![]); + } + } + + session.pending.check_capacity()?; + + let negotiated = self + .negotiated + .as_ref() + .expect("negotiated must exist when Open"); + + let wire_seq = session.next_wire_seq; + let scheduled_at = scheduled_at_override.unwrap_or(session.next_send_at); + + let request = EchoRequest { + token, + sequence: wire_seq, + params: negotiated.params.clone(), + payload: vec![], + }; + let packet = encode_echo_request(&request, self.config.hmac_key.as_deref())?; + let send_result = send(&packet)?; + let timer_error = instant_abs_diff(send_result.sent_at.mono, scheduled_at); + + let session = self + .session + .as_mut() + .expect("session must exist when phase is Open"); + + let pending = PendingProbe { + wire_seq, + sent_at: send_result.sent_at, + timeout_at: send_result + .sent_at + .mono + .checked_add(self.config.probe_timeout) + .ok_or(ClientError::DurationOverflow)?, + }; + session.pending.insert(pending)?; + + session.next_wire_seq = session.next_wire_seq.wrapping_add(1); + session.packets_sent = + session + .packets_sent + .checked_add(1) + .ok_or(ClientError::CounterOverflow { + counter: "packets_sent", + })?; + + let negotiated = self + .negotiated + .as_ref() + .expect("negotiated must exist when Open"); + let interval_ns = u64::try_from(negotiated.params.interval_ns) + .expect("validated positive negotiated interval"); + let session = self + .session + .as_mut() + .expect("session must exist when phase is Open"); + session.next_send_at = if let Some(scheduled_at) = scheduled_at_override { + scheduled_at + .checked_add(Duration::from_nanos(interval_ns)) + .ok_or(ClientError::DurationOverflow)? + } else { + next_probe_deadline(session.start_mono, interval_ns, session.packets_sent)? + }; + + if let Some(end) = session.end_mono { + if session.next_send_at >= end { + session.sending_done = true; + } + } + + Ok(vec![ClientEvent::EchoSent { + seq: wire_seq, + remote: self.remote, + scheduled_at, + sent_at: send_result.sent_at, + bytes: send_result.bytes, + send_call: send_result.send_call, + timer_error, + }]) + } + + pub(crate) fn process_received_echo_packet( + &mut self, + packet: &[u8], + now: ClientTimestamp, + meta: ReceiveMeta, + ) -> Result, ClientError> { + match self.phase { + ClientPhase::Open { .. } => {} + ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), + ClientPhase::Connected => return Err(ClientError::NotOpen), + ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted), + } + + let Some(reply) = self.decode_received_packet(packet) else { + return Ok(vec![ClientEvent::Warning { + kind: WarningKind::MalformedOrUnrelatedPacket, + message: "dropped malformed or unrelated packet".to_owned(), + at: now, + }]); + }; + self.process_echo_reply(reply, packet.len(), now, meta) + } + + pub(crate) fn poll_timeouts_at( + &mut self, + now: Instant, + ) -> Result, ClientError> { + match self.phase { + ClientPhase::Open { .. } => {} + ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), + ClientPhase::Connected => return Err(ClientError::NotOpen), + ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted), + } + + let session = self + .session + .as_mut() + .expect("session must exist when phase is Open"); + + let expired = session.pending.drain_expired(now); + let mut events = Vec::with_capacity(expired.len()); + for probe in expired { + events.push(ClientEvent::EchoLoss { + seq: probe.wire_seq, + sent_at: probe.sent_at, + timeout_at: probe.timeout_at, + }); + session.timed_out.insert(probe); + } + + Ok(events) + } + + pub(crate) fn close_with(&mut self, send: F) -> Result, ClientError> + where + F: FnOnce(&[u8]) -> Result<(), ClientError>, + { + let token = match self.phase { + ClientPhase::Open { token } => token, + ClientPhase::Closed { .. } => return Err(ClientError::AlreadyClosed), + ClientPhase::Connected | ClientPhase::NoTestCompleted => { + return Err(ClientError::NotOpen) + } + }; + + let packet = + encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?; + send(&packet)?; + self.phase = ClientPhase::Closed { + source: CloseSource::Local, + }; + if let Some(session) = self.session.as_mut() { + session.timed_out.clear(); + } + self.session = None; + + Ok(vec![ClientEvent::SessionClosed { + remote: self.remote, + token, + at: ClientTimestamp::now(), + }]) + } + + pub(crate) fn is_run_complete(&self) -> bool { + let Some(session) = self.session.as_ref() else { + return matches!( + self.phase, + ClientPhase::Closed { .. } | ClientPhase::NoTestCompleted + ); + }; + session.sending_done && session.pending.len() == 0 + } + + pub(crate) fn is_peer_closed(&self) -> bool { + matches!( + self.phase, + ClientPhase::Closed { + source: CloseSource::Peer + } + ) + } + + pub(crate) fn has_timed_out_metadata(&self) -> bool { + self.session + .as_ref() + .is_some_and(|session| session.timed_out.len() > 0) + } + + pub(crate) fn packets_sent(&self) -> u64 { + self.session + .as_ref() + .map_or(0, |session| session.packets_sent) + } + + pub(crate) fn is_open(&self) -> bool { + matches!(self.phase, ClientPhase::Open { .. }) + } + + fn accept_normal_open( + &mut self, + reply: OpenReply, + now: ClientTimestamp, + before_normal_open: F, + ) -> Result + where + F: FnOnce(&NegotiatedParams) -> Result<(), ClientError>, + { + let token = reply.token; + let negotiated = negotiate_params( + &self.requested, + reply.params, + self.config.negotiation_policy, + )?; + before_normal_open(&negotiated)?; + self.negotiated = Some(negotiated.clone()); + self.phase = ClientPhase::Open { token }; + + let end_mono = if negotiated.params.duration_ns > 0 { + Some(negotiated_end_mono( + now.mono, + negotiated.params.duration_ns, + )?) + } else { + None + }; + + self.session = Some(ActiveSession { + next_wire_seq: 0, + highest_received_seq: None, + packets_sent: 0, + start_mono: now.mono, + end_mono, + next_send_at: now.mono, + pending: PendingMap::new(self.config.max_pending_probes), + timed_out: TimedOutMap::new(self.config.max_pending_probes), + completed: CompletedSet::new(self.config.max_pending_probes), + sending_done: false, + }); + + let event = ClientEvent::SessionStarted { + remote: self.remote, + token, + negotiated: negotiated.clone(), + at: now, + }; + + Ok(OpenOutcome::Started { + remote: self.remote, + token, + negotiated, + event, + }) + } + + fn accept_no_test_open( + &mut self, + reply: OpenReply, + now: ClientTimestamp, + ) -> Result { + let negotiated = negotiate_params( + &self.requested, + reply.params, + self.config.negotiation_policy, + )?; + self.negotiated = Some(negotiated.clone()); + self.phase = ClientPhase::NoTestCompleted; + let event = ClientEvent::NoTestCompleted { + remote: self.remote, + negotiated: negotiated.clone(), + at: now, + }; + Ok(OpenOutcome::NoTestCompleted { + remote: self.remote, + negotiated, + event, + }) + } + + fn decode_received_packet(&self, packet: &[u8]) -> Option { + let negotiated = self + .negotiated + .as_ref() + .expect("negotiated must exist when Open"); + + decode_echo_reply(packet, &negotiated.params, self.config.hmac_key.as_deref()).ok() + } + + fn process_echo_reply( + &mut self, + reply: EchoReply, + packet_len: usize, + now: ClientTimestamp, + meta: ReceiveMeta, + ) -> Result, ClientError> { + let token = match self.phase { + ClientPhase::Open { token } => token, + _ => unreachable!(), + }; + if reply.token != token { + return Ok(vec![ClientEvent::Warning { + kind: WarningKind::WrongToken, + message: format!( + "dropped reply with wrong token: expected {token:#x}, got {:#x}", + reply.token + ), + at: now, + }]); + } + + let session = self.session.as_mut().expect("session must exist when Open"); + + let wire_seq = reply.sequence; + + if let Some(pending) = session.pending.remove(wire_seq) { + let rtt = compute_rtt(&pending.sent_at, &now, &reply.timestamps); + let server_timing = build_server_timing(&reply.timestamps); + let one_way = compute_one_way(&pending.sent_at, &now, &reply.timestamps); + let received_stats = build_received_stats(&reply); + let is_late = session + .highest_received_seq + .is_some_and(|h| sequence_is_before(wire_seq, h)); + let highest_seen = session.highest_received_seq.unwrap_or(wire_seq); + + update_highest_received(&mut session.highest_received_seq, wire_seq); + session.completed.insert(wire_seq); + + let mut events = Vec::new(); + if is_late { + events.push(ClientEvent::LateReply { + seq: wire_seq, + highest_seen, + remote: self.remote, + sent_at: Some(pending.sent_at), + received_at: now, + rtt: Some(rtt), + server_timing, + one_way, + received_stats, + bytes: packet_len, + packet_meta: meta.into(), + }); + } else { + events.push(ClientEvent::EchoReply { + seq: wire_seq, + remote: self.remote, + sent_at: pending.sent_at, + received_at: now, + rtt, + server_timing, + one_way, + received_stats, + bytes: packet_len, + packet_meta: meta.into(), + }); + } + if flags::has(reply.flags, flags::FLAG_CLOSE) { + self.close_from_peer(token, now, &mut events); + } + Ok(events) + } else if session.completed.contains(wire_seq) { + update_highest_received(&mut session.highest_received_seq, wire_seq); + Ok(vec![ClientEvent::DuplicateReply { + seq: wire_seq, + remote: self.remote, + received_at: now, + bytes: packet_len, + }]) + } else if let Some(timed_out) = session.timed_out.remove(wire_seq) { + let rtt = compute_rtt(&timed_out.sent_at, &now, &reply.timestamps); + let server_timing = build_server_timing(&reply.timestamps); + let one_way = compute_one_way(&timed_out.sent_at, &now, &reply.timestamps); + let received_stats = build_received_stats(&reply); + let highest_seen = session.highest_received_seq.unwrap_or(wire_seq); + update_highest_received(&mut session.highest_received_seq, wire_seq); + session.completed.insert(wire_seq); + + let mut events = vec![ClientEvent::LateReply { + seq: wire_seq, + highest_seen, + remote: self.remote, + sent_at: Some(timed_out.sent_at), + received_at: now, + rtt: Some(rtt), + server_timing, + one_way, + received_stats, + bytes: packet_len, + packet_meta: meta.into(), + }]; + if flags::has(reply.flags, flags::FLAG_CLOSE) { + self.close_from_peer(token, now, &mut events); + } + Ok(events) + } else if session + .highest_received_seq + .is_some_and(|h| sequence_is_before(wire_seq, h)) + { + Ok(vec![ClientEvent::LateReply { + seq: wire_seq, + highest_seen: session.highest_received_seq.unwrap(), + remote: self.remote, + sent_at: None, + received_at: now, + rtt: None, + server_timing: build_server_timing(&reply.timestamps), + one_way: None, + received_stats: build_received_stats(&reply), + bytes: packet_len, + packet_meta: meta.into(), + }]) + } else { + Ok(vec![ClientEvent::Warning { + kind: WarningKind::UntrackedReply, + message: format!( + "dropped reply with untracked seq {wire_seq} (no pending or completed entry)" + ), + at: now, + }]) + } + } + + fn close_from_peer(&mut self, token: u64, now: ClientTimestamp, events: &mut Vec) { + self.phase = ClientPhase::Closed { + source: CloseSource::Peer, + }; + if let Some(session) = self.session.as_mut() { + session.timed_out.clear(); + } + self.session = None; + events.push(ClientEvent::SessionClosed { + remote: self.remote, + token, + at: now, + }); + } +} + +pub(crate) fn recv_buffer_size( + has_hmac: bool, + negotiated: Option<&NegotiatedParams>, +) -> Result { + Ok(match negotiated { + Some(negotiated) => echo_packet_len(has_hmac, &negotiated.params)? + .saturating_add(1) + .max(MIN_RECV_BUFFER_SIZE), + None => MIN_RECV_BUFFER_SIZE, + }) +} + +pub(crate) fn params_from_config(config: &ClientConfig) -> Result { + validate_protocol_config(config)?; + Ok(Params { + protocol_version: PROTOCOL_VERSION, + duration_ns: match config.duration { + Some(duration) => config_duration_to_ns("duration", duration)?, + None => 0, + }, + interval_ns: config_duration_to_ns("interval", config.interval)?, + length: i64::from(config.length), + received_stats: config.received_stats, + stamp_at: config.stamp_at, + clock: config.clock, + dscp: i64::from(config.dscp), + server_fill: config.server_fill.clone().map(|value| ServerFill { value }), + }) +} + +pub(crate) fn update_highest_received(highest_received_seq: &mut Option, wire_seq: u32) { + *highest_received_seq = Some(highest_received_seq.map_or(wire_seq, |h| { + if sequence_is_after(wire_seq, h) { + wire_seq + } else { + h + } + })); +} + +fn next_probe_deadline( + start: Instant, + interval_ns: u64, + packets_sent: u64, +) -> Result { + let offset_ns = interval_ns + .checked_mul(packets_sent) + .ok_or(ClientError::DurationOverflow)?; + start + .checked_add(Duration::from_nanos(offset_ns)) + .ok_or(ClientError::DurationOverflow) +} + +pub(crate) fn sequence_is_after(candidate: u32, current: u32) -> bool { + candidate != current && candidate.wrapping_sub(current) < (1 << 31) +} + +pub(crate) fn sequence_is_before(candidate: u32, current: u32) -> bool { + current != candidate && current.wrapping_sub(candidate) < (1 << 31) +} + +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) +} + +pub(crate) fn compute_rtt( + sent_at: &ClientTimestamp, + received_at: &ClientTimestamp, + ts: &TimestampFields, +) -> RttSample { + let raw = received_at + .mono + .checked_duration_since(sent_at.mono) + .unwrap_or(Duration::ZERO); + + let server_processing = compute_server_processing(ts); + + let adjusted = server_processing + .map(|sp| SignedDuration::from_nanos(duration_ns_i128(raw) - duration_ns_i128(sp))); + let effective = adjusted.unwrap_or_else(|| SignedDuration::from_duration(raw)); + + RttSample { + raw, + adjusted, + effective, + } +} + +fn duration_ns_i128(duration: Duration) -> i128 { + i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX) +} + +fn compute_server_processing(ts: &TimestampFields) -> Option { + if let (Some(recv_mono), Some(send_mono)) = (ts.recv_mono, ts.send_mono) { + let diff = send_mono.checked_sub(recv_mono)?; + return Some(Duration::from_nanos(u64::try_from(diff).ok()?)); + } + if let (Some(recv_wall), Some(send_wall)) = (ts.recv_wall, ts.send_wall) { + let diff = send_wall.checked_sub(recv_wall)?; + return Some(Duration::from_nanos(u64::try_from(diff).ok()?)); + } + None +} + +fn build_server_timing(ts: &TimestampFields) -> Option { + if ts.recv_wall.is_none() + && ts.recv_mono.is_none() + && ts.send_wall.is_none() + && ts.send_mono.is_none() + && ts.midpoint_wall.is_none() + && ts.midpoint_mono.is_none() + { + return None; + } + Some(ServerTiming { + receive_wall_ns: ts.recv_wall, + receive_mono_ns: ts.recv_mono, + send_wall_ns: ts.send_wall, + send_mono_ns: ts.send_mono, + midpoint_wall_ns: ts.midpoint_wall, + midpoint_mono_ns: ts.midpoint_mono, + processing: compute_server_processing(ts), + }) +} + +pub(crate) fn compute_one_way( + sent_at: &ClientTimestamp, + received_at: &ClientTimestamp, + ts: &TimestampFields, +) -> Option { + let server_recv_wall = ts.recv_wall.or(ts.midpoint_wall); + let server_send_wall = ts.send_wall.or(ts.midpoint_wall); + + let client_send_ns = unix_epoch_ns_i64(sent_at.wall); + let client_recv_ns = unix_epoch_ns_i64(received_at.wall); + + let c2s = server_recv_wall + .zip(client_send_ns) + .and_then(|(srv, cli)| srv.checked_sub(cli)) + .map(|d| SignedDuration::from_nanos(i128::from(d))); + let s2c = client_recv_ns + .zip(server_send_wall) + .and_then(|(cli, srv)| cli.checked_sub(srv)) + .map(|d| SignedDuration::from_nanos(i128::from(d))); + + if c2s.is_none() && s2c.is_none() { + return None; + } + + Some(OneWayDelaySample { + client_to_server: c2s, + server_to_client: s2c, + }) +} + +pub(crate) fn unix_epoch_ns_i64(time: SystemTime) -> Option { + time.duration_since(UNIX_EPOCH) + .ok() + .and_then(|duration| i64::try_from(duration.as_nanos()).ok()) +} + +fn build_received_stats(reply: &EchoReply) -> Option { + if reply.recv_count.is_none() && reply.recv_window.is_none() { + return None; + } + Some(ReceivedStatsSample { + count: reply.recv_count, + window: reply.recv_window, + }) +} + +fn validate_protocol_config(config: &ClientConfig) -> Result<(), ClientError> { + if config.duration == Some(Duration::ZERO) { + return Err(ClientError::InvalidConfig { + reason: "duration must be greater than zero; use None for continuous mode".to_owned(), + }); + } + if config.interval == Duration::ZERO { + return Err(ClientError::InvalidConfig { + reason: "interval must be greater than zero".to_owned(), + }); + } + if config.dscp > MAX_DSCP_CODEPOINT { + return Err(ClientError::InvalidConfig { + reason: format!("dscp must be <= {MAX_DSCP_CODEPOINT}"), + }); + } + if config.length > MAX_UDP_PAYLOAD_LENGTH { + return Err(ClientError::InvalidConfig { + reason: format!("packet length must be <= {MAX_UDP_PAYLOAD_LENGTH}"), + }); + } + + if let Some(fill) = &config.server_fill { + let len = fill.len(); + if len == 0 { + return Err(ClientError::InvalidConfig { + reason: "server_fill must not be empty".to_owned(), + }); + } + if len > MAX_SERVER_FILL_BYTES { + return Err(ClientError::InvalidConfig { + reason: format!("server_fill must be <= {MAX_SERVER_FILL_BYTES} bytes, got {len}"), + }); + } + } + + Ok(()) +} + +fn config_duration_to_ns(field: &str, duration: Duration) -> Result { + i64::try_from(duration.as_nanos()).map_err(|_| ClientError::InvalidConfig { + reason: format!("{field} is too large to encode as nanoseconds"), + }) +} + +fn negotiated_end_mono(start: Instant, duration_ns: i64) -> Result { + debug_assert!(duration_ns > 0); + let duration_ns = u64::try_from(duration_ns).expect("validated positive negotiated duration"); + start + .checked_add(Duration::from_nanos(duration_ns)) + .ok_or_else(|| ClientError::NegotiationRejected { + reason: "duration is too large to schedule".to_owned(), + }) +} diff --git a/crates/irtt-client/src/socket.rs b/crates/irtt-client/src/socket.rs index 6b8c134..767e825 100644 --- a/crates/irtt-client/src/socket.rs +++ b/crates/irtt-client/src/socket.rs @@ -99,6 +99,44 @@ pub(crate) fn connect_udp_socket( Ok(socket) } +pub(crate) fn bind_unconnected_udp_socket( + config: &SocketConfig, + family_remote: SocketAddr, +) -> Result { + let domain = if family_remote.is_ipv4() { + Domain::IPV4 + } else { + Domain::IPV6 + }; + let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?; + + if config.ipv6_only && family_remote.is_ipv6() { + socket.set_only_v6(true)?; + } + let bind_addr = config.bind_addr.unwrap_or_else(|| { + if family_remote.is_ipv4() { + SocketAddr::from(([0, 0, 0, 0], 0)) + } else { + SocketAddr::from(([0_u16; 8], 0)) + } + }); + socket.bind(&bind_addr.into())?; + + let socket: UdpSocket = socket.into(); + configure_receive_metadata(&socket, family_remote).map_err(|source| { + ClientError::SocketOption { + operation: "enable receive metadata", + remote: family_remote, + source, + } + })?; + if let Some(ttl) = config.ttl { + apply_ttl_to_socket(&socket, family_remote, ttl)?; + } + socket.set_read_timeout(config.recv_timeout)?; + Ok(socket) +} + #[cfg(test)] mod tests { use super::*;