diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 9d2f6c7..3cb1504 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -12,7 +12,7 @@ use crate::{ error::ClientError, event::{ClientEvent, OpenOutcome}, receive::recv_datagram, - runtime::{recv_buffer_size, SendProbeResult, SessionRuntime, MAX_OPEN_PACKET_SIZE}, + session::machine::{recv_buffer_size, SendProbeResult, SessionMachine, 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, @@ -20,7 +20,7 @@ use crate::{ #[cfg(test)] use crate::{ - runtime::{ + session::machine::{ compute_one_way, compute_rtt, params_from_config, sequence_is_after, sequence_is_before, unix_epoch_ns_i64, update_highest_received, }, @@ -36,7 +36,7 @@ use crate::{ /// instead. #[derive(Debug)] pub struct Client { - runtime: SessionRuntime, + runtime: SessionMachine, socket: UdpSocket, remote: SocketAddr, recv_buffer: Vec, @@ -51,7 +51,7 @@ impl Client { pub fn connect(config: ClientConfig) -> Result { validate_open_timeouts(&config.open_timeouts)?; let remote = resolve_remote(&config)?; - let runtime = SessionRuntime::new(config.clone(), remote)?; + let runtime = SessionMachine::new(config.clone(), remote)?; let socket = connect_udp_socket(&config.socket_config, remote)?; Ok(Self { diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index cbe71b6..0f04c19 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -59,7 +59,6 @@ mod managed; mod metadata; mod probe; mod receive; -mod runtime; mod session; mod socket; mod socket_options; diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 862f609..f621e53 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -17,7 +17,7 @@ use crate::{ event::{ClientEvent, OpenOutcome}, metadata::ReceiveMeta, receive::recv_datagram_from, - runtime::{advance_cadence, params_from_config, SendProbeResult, SessionRuntime}, + session::machine::{advance_cadence, params_from_config, SendProbeResult, SessionMachine}, socket::{bind_unconnected_udp_socket, validate_open_timeouts}, socket_options::apply_dscp_to_socket, timing::ClientTimestamp, @@ -625,7 +625,7 @@ struct TargetState { id: TargetId, remote: SocketAddr, configured_auth: Option, - runtime: SessionRuntime, + runtime: SessionMachine, status: TargetStatus, open_packet: Vec, counters: TargetCounters, @@ -709,7 +709,7 @@ impl TargetState { if let Some(auth) = &target.auth { config.hmac_key = auth.hmac_key.clone(); } - let runtime = SessionRuntime::new(config, target.remote)?; + let runtime = SessionMachine::new(config, target.remote)?; let open_packet = runtime.open_packet()?; Ok(Self { id: target.id, diff --git a/crates/irtt-client/src/runtime.rs b/crates/irtt-client/src/session/machine.rs similarity index 83% rename from crates/irtt-client/src/runtime.rs rename to crates/irtt-client/src/session/machine.rs index 0640100..1c01a7d 100644 --- a/crates/irtt-client/src/runtime.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -17,7 +17,7 @@ use crate::{ }, metadata::ReceiveMeta, probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap}, - session::{negotiate_params, ActiveSession, ClientPhase, CloseSource, NegotiatedParams}, + session::{negotiate_params, NegotiatedParams}, timing::ClientTimestamp, }; @@ -25,14 +25,44 @@ pub(crate) const MAX_OPEN_PACKET_SIZE: usize = 512; const MIN_RECV_BUFFER_SIZE: usize = 2048; #[derive(Debug)] -pub(crate) struct SessionRuntime { +pub(crate) struct SessionMachine { config: ClientConfig, remote: std::net::SocketAddr, requested: Params, - negotiated: Option, - phase: ClientPhase, - session: Option, - final_packets_sent: u64, + state: MachineState, +} + +#[derive(Debug)] +enum MachineState { + Connected, + Open(Box), + NoTestCompleted, + Closed { + source: CloseSource, + packets_sent: u64, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CloseSource { + Local, + Peer, +} + +#[derive(Debug)] +struct ActiveSession { + token: u64, + negotiated: NegotiatedParams, + next_wire_seq: u32, + highest_received_seq: Option, + packets_sent: u64, + start_mono: Instant, + end_mono: Option, + next_send_at: Instant, + pending: PendingMap, + timed_out: TimedOutMap, + completed: CompletedSet, + sending_done: bool, } #[derive(Debug, Clone, Copy)] @@ -42,7 +72,7 @@ pub(crate) struct SendProbeResult { pub(crate) send_call: Duration, } -impl SessionRuntime { +impl SessionMachine { pub(crate) fn new( config: ClientConfig, remote: std::net::SocketAddr, @@ -63,10 +93,7 @@ impl SessionRuntime { config, remote, requested, - negotiated: None, - phase: ClientPhase::Connected, - session: None, - final_packets_sent: 0, + state: MachineState::Connected, }) } @@ -79,11 +106,11 @@ impl SessionRuntime { } 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), + match self.state { + MachineState::Connected => {} + MachineState::Open(_) => return Err(ClientError::AlreadyOpen), + MachineState::Closed { .. } => return Err(ClientError::AlreadyClosed), + MachineState::NoTestCompleted => return Err(ClientError::AlreadyCompleted), } let request = OpenRequest { @@ -133,7 +160,9 @@ impl SessionRuntime { } pub(crate) fn next_send_deadline(&self) -> Option { - let session = self.session.as_ref()?; + let MachineState::Open(session) = &self.state else { + return None; + }; if session.sending_done { return None; } @@ -177,18 +206,16 @@ impl SessionRuntime { 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 (config, state) = (&self.config, &mut self.state); + let probe_timeout = config.probe_timeout; + let hmac_key = config.hmac_key.as_deref(); + let session = match state { + MachineState::Open(session) => session, + MachineState::Closed { .. } => return Err(ClientError::AlreadyClosed), + MachineState::Connected => return Err(ClientError::NotOpen), + MachineState::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![]); } @@ -204,35 +231,25 @@ impl SessionRuntime { 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, + token: session.token, sequence: wire_seq, - params: negotiated.params.clone(), + params: session.negotiated.params.clone(), payload: vec![], }; - let packet = encode_echo_request(&request, self.config.hmac_key.as_deref())?; + let packet = encode_echo_request(&request, hmac_key)?; let send_result = send(&packet)?; - 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) + .checked_add(probe_timeout) .ok_or(ClientError::DurationOverflow)?, }; session.pending.insert(pending)?; @@ -246,11 +263,7 @@ impl SessionRuntime { 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) + let interval_ns = u64::try_from(session.negotiated.params.interval_ns) .expect("validated positive negotiated interval"); let interval = Duration::from_nanos(interval_ns); let (scheduled_at, next_send_at) = if skip_missed_slots { @@ -264,10 +277,6 @@ impl SessionRuntime { ) }; 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"); session.next_send_at = next_send_at; if let Some(end) = session.end_mono { @@ -293,12 +302,7 @@ impl SessionRuntime { 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), - } + self.open_session()?; let Some(reply) = self.decode_received_packet(packet) else { return Ok(vec![ClientEvent::Warning { @@ -314,17 +318,7 @@ impl SessionRuntime { &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 session = self.open_session_mut()?; let expired = session.pending.drain_expired(now); let mut events = Vec::with_capacity(expired.len()); @@ -344,10 +338,10 @@ impl SessionRuntime { 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 => { + let token = match &self.state { + MachineState::Open(session) => session.token, + MachineState::Closed { .. } => return Err(ClientError::AlreadyClosed), + MachineState::Connected | MachineState::NoTestCompleted => { return Err(ClientError::NotOpen) } }; @@ -365,38 +359,40 @@ impl SessionRuntime { } 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 + match &self.state { + MachineState::Open(session) => session.sending_done && session.pending.len() == 0, + MachineState::NoTestCompleted | MachineState::Closed { .. } => true, + MachineState::Connected => false, + } } pub(crate) fn is_peer_closed(&self) -> bool { matches!( - self.phase, - ClientPhase::Closed { - source: CloseSource::Peer + self.state, + MachineState::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) + matches!( + &self.state, + MachineState::Open(session) if session.timed_out.len() > 0 + ) } pub(crate) fn packets_sent(&self) -> u64 { - self.session - .as_ref() - .map_or(self.final_packets_sent, |session| session.packets_sent) + match &self.state { + MachineState::Open(session) => session.packets_sent, + MachineState::Closed { packets_sent, .. } => *packets_sent, + MachineState::Connected | MachineState::NoTestCompleted => 0, + } } pub(crate) fn is_open(&self) -> bool { - matches!(self.phase, ClientPhase::Open { .. }) + matches!(self.state, MachineState::Open(_)) } fn accept_normal_open( @@ -415,8 +411,6 @@ impl SessionRuntime { 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( @@ -427,7 +421,9 @@ impl SessionRuntime { None }; - self.session = Some(ActiveSession { + self.state = MachineState::Open(Box::new(ActiveSession { + token, + negotiated: negotiated.clone(), next_wire_seq: 0, highest_received_seq: None, packets_sent: 0, @@ -438,7 +434,7 @@ impl SessionRuntime { 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, @@ -465,8 +461,7 @@ impl SessionRuntime { reply.params, self.config.negotiation_policy, )?; - self.negotiated = Some(negotiated.clone()); - self.phase = ClientPhase::NoTestCompleted; + self.state = MachineState::NoTestCompleted; let event = ClientEvent::NoTestCompleted { remote: self.remote, negotiated: negotiated.clone(), @@ -480,12 +475,16 @@ impl SessionRuntime { } fn decode_received_packet(&self, packet: &[u8]) -> Option { - let negotiated = self - .negotiated - .as_ref() - .expect("negotiated must exist when Open"); + let session = self + .open_session() + .expect("decode_received_packet is only called for an open session"); - decode_echo_reply(packet, &negotiated.params, self.config.hmac_key.as_deref()).ok() + decode_echo_reply( + packet, + &session.negotiated.params, + self.config.hmac_key.as_deref(), + ) + .ok() } fn process_echo_reply( @@ -495,10 +494,10 @@ impl SessionRuntime { now: ClientTimestamp, meta: ReceiveMeta, ) -> Result, ClientError> { - let token = match self.phase { - ClientPhase::Open { token } => token, - _ => unreachable!(), - }; + let token = self + .open_session() + .expect("process_echo_reply is only called for an open session") + .token; if reply.token != token { return Ok(vec![ClientEvent::Warning { kind: WarningKind::WrongToken, @@ -513,7 +512,9 @@ impl SessionRuntime { let wire_seq = reply.sequence; let should_close = flags::has(reply.flags, flags::FLAG_CLOSE); let mut events = { - let session = self.session.as_mut().expect("session must exist when Open"); + let session = self + .open_session_mut() + .expect("process_echo_reply is only called for an open session"); if let Some(pending) = session.pending.remove(wire_seq) { let rtt = compute_rtt(&pending.sent_at, &now, &reply.timestamps); @@ -630,11 +631,36 @@ impl SessionRuntime { } fn transition_to_closed(&mut self, source: CloseSource) { - if let Some(mut session) = self.session.take() { - session.timed_out.clear(); - self.final_packets_sent = session.packets_sent; + let packets_sent = match &mut self.state { + MachineState::Open(session) => { + session.timed_out.clear(); + session.packets_sent + } + MachineState::Closed { packets_sent, .. } => *packets_sent, + MachineState::Connected | MachineState::NoTestCompleted => 0, + }; + self.state = MachineState::Closed { + source, + packets_sent, + }; + } + + fn open_session(&self) -> Result<&ActiveSession, ClientError> { + match &self.state { + MachineState::Open(session) => Ok(session), + MachineState::Closed { .. } => Err(ClientError::AlreadyClosed), + MachineState::Connected => Err(ClientError::NotOpen), + MachineState::NoTestCompleted => Err(ClientError::AlreadyCompleted), + } + } + + fn open_session_mut(&mut self) -> Result<&mut ActiveSession, ClientError> { + match &mut self.state { + MachineState::Open(session) => Ok(session), + MachineState::Closed { .. } => Err(ClientError::AlreadyClosed), + MachineState::Connected => Err(ClientError::NotOpen), + MachineState::NoTestCompleted => Err(ClientError::AlreadyCompleted), } - self.phase = ClientPhase::Closed { source }; } } diff --git a/crates/irtt-client/src/session/mod.rs b/crates/irtt-client/src/session/mod.rs new file mode 100644 index 0000000..14be3c4 --- /dev/null +++ b/crates/irtt-client/src/session/mod.rs @@ -0,0 +1,5 @@ +pub(crate) mod machine; +mod negotiation; + +pub(crate) use negotiation::negotiate_params; +pub use negotiation::{NegotiatedParams, NegotiationRestriction}; diff --git a/crates/irtt-client/src/session.rs b/crates/irtt-client/src/session/negotiation.rs similarity index 95% rename from crates/irtt-client/src/session.rs rename to crates/irtt-client/src/session/negotiation.rs index 08ec6b3..2427606 100644 --- a/crates/irtt-client/src/session.rs +++ b/crates/irtt-client/src/session/negotiation.rs @@ -1,11 +1,10 @@ -use std::{fmt, time::Instant}; +use std::fmt; use irtt_proto::{Clock, Params, ReceivedStats, StampAt, PROTOCOL_VERSION}; use crate::{ config::{NegotiationPolicy, MAX_DSCP_CODEPOINT}, error::ClientError, - probe::{CompletedSet, PendingMap, TimedOutMap}, }; /// Protocol parameters accepted for a session. @@ -156,34 +155,6 @@ impl fmt::Display for NegotiationRestriction { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum ClientPhase { - Connected, - Open { token: u64 }, - NoTestCompleted, - Closed { source: CloseSource }, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum CloseSource { - Local, - Peer, -} - -#[derive(Debug)] -pub(crate) struct ActiveSession { - pub next_wire_seq: u32, - pub highest_received_seq: Option, - pub packets_sent: u64, - pub start_mono: Instant, - pub end_mono: Option, - pub next_send_at: Instant, - pub pending: PendingMap, - pub timed_out: TimedOutMap, - pub completed: CompletedSet, - pub sending_done: bool, -} - pub(crate) fn negotiate_params( requested: &Params, returned: Params,