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

Merge pull request #38 from Lochnair/refactor/session-machine-state

Browse Source 
Make session lifecycle state explicit
This commit was merged in pull request #38.
This commit is contained in:
Nils Andreas Svee authored and GitHub committed 2026-08-03 23:41:59 +07:00
2 parents dfe0534905 4b4adc518c
commit 549efa8330
6 files changed
+147 -146
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

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