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

tests(client): cleanup unnecessary tests and hooks
Rust CI / cargo fmt (push) Successful in 14s
Details
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 32s
Details
Rust CI / cargo clippy (push) Successful in 34s
Details
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Failing after 34s
Details
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Failing after 7s
Details
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Failing after 10s
Details
Rust CI / cargo doc (push) Successful in 26s
Details
Fuzz Smoke / cargo fuzz smoke (push) Failing after 32s
Details
Supply chain / cargo-deny advisories (push) Successful in 6s
Details
Supply chain / cargo-deny bans licenses sources (push) Successful in 5s
Details
Supply chain / actionlint (push) Successful in 2s
Details
Netem Interop / netem interop (push) Successful in 1m3s
Details

Browse Source
This commit is contained in:
Nils Andreas Svee committed 2026-10-02 08:33:49 +07:00
1 parent 50fe83a4c8
commit dc5b19059e
27 files changed
+118 -2625
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/irtt-client/Cargo.toml
-1
View File
@@ -47,5 +47,4 @@ doc-scrape-examples = true
# Keep the packaged client testable against the latest published server while
# workspace tests continue to exercise the local source through this path.
irtt-server = { version = "0.5.1", path = "../irtt-server" }
proptest = "1.11.0"
tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] }
crates/irtt-client/src/async_client.rs
+3 -11
View File
@@ -85,7 +85,7 @@ impl AsyncOpenCleanup {
/// Owned state for one asynchronous open transaction.
///
/// This state deliberately contains no reference to [`AsyncClient`], allowing
/// a future managed target to store the client and opening state side by side.
/// a managed target to store the client and opening state side by side.
#[derive(Debug)]
pub(crate) struct AsyncOpenState {
attempt: usize,
@@ -567,7 +567,7 @@ impl AsyncClient {
};
if self.machine.is_peer_closed() {
self.prepared_probe = None;
if self.clear_peer_close_dscp().is_ok() {
if clear_dscp_on_tokio_socket(&self.socket, self.remote).is_ok() {
self.applied_traffic_class = None;
}
}
@@ -607,7 +607,7 @@ impl AsyncClient {
return Poll::Ready(Err(ClientError::Socket(error)));
}
}
if let Err(error) = self.clear_close_dscp() {
if let Err(error) = clear_dscp_on_tokio_socket(&self.socket, self.remote) {
return Poll::Ready(Err(error));
}
let mut rollback =
@@ -798,14 +798,6 @@ impl AsyncClient {
}
}
fn clear_peer_close_dscp(&self) -> Result<(), ClientError> {
clear_dscp_on_tokio_socket(&self.socket, self.remote)
}
fn clear_close_dscp(&self) -> Result<(), ClientError> {
clear_dscp_on_tokio_socket(&self.socket, self.remote)
}
fn restore_dscp_best_effort(&self, previous_traffic_class: Option<u8>) {
let _ = match previous_traffic_class {
Some(traffic_class) => {
crates/irtt-client/src/async_client/tests.rs
+15 -124
View File
@@ -17,7 +17,7 @@ use tokio::runtime::{Builder, Runtime};
mod in_tree_server;
use super::*;
use crate::{socket_options::tokio_socket_traffic_class, Client, RunMode, SocketConfig};
use crate::{Client, RunMode, SocketConfig};
use in_tree_server::InTreeServer;
use irtt_server::ServerConfig;
@@ -68,10 +68,6 @@ async fn opened_client(addr: SocketAddr, dscp: u8) -> AsyncClient {
client
}
fn socket_dscp(client: &AsyncClient) -> u32 {
tokio_socket_traffic_class(&client.socket, client.remote).unwrap() >> 2
}
fn start_server<F>(handler: F) -> TestServer
where
F: FnOnce(UdpSocket, mpsc::Sender<Vec<u8>>) + Send + 'static,
@@ -382,85 +378,24 @@ fn recv_after_local_close_and_no_test_fails_on_first_poll() {
}
#[test]
fn blocking_and_async_hmac_dscp_lifecycle_are_semantically_equivalent() {
let key = b"async-equivalence-key".to_vec();
let blocking_server = InTreeServer::start(ServerConfig::default().with_hmac_key(key.clone()));
let mut blocking =
Client::connect(config(blocking_server.addr, Some(key.clone()), 46)).unwrap();
let blocking_open = blocking.open().unwrap();
let blocking_sent = blocking.send_probe().unwrap();
let blocking_reply = blocking.recv_once().unwrap();
let blocking_close = blocking.close().unwrap();
drop(blocking_server);
let async_server = InTreeServer::start(ServerConfig::default().with_hmac_key(key.clone()));
let (async_open, async_sent, async_reply, async_close) = runtime().block_on(async {
let mut client = AsyncClient::connect(config(async_server.addr, Some(key), 46))
fn authenticated_async_lifecycle_applies_negotiated_dscp() {
let key = b"async-lifecycle-key".to_vec();
let server = InTreeServer::start(ServerConfig::default().with_hmac_key(key.clone()));
runtime().block_on(async {
let mut client = AsyncClient::connect(config(server.addr, Some(key), 46))
.await
.unwrap();
let opened = client.open().await.unwrap();
assert_eq!(socket_dscp(&client), 46);
let sent = client.send_probe().await.unwrap();
let reply = client.recv().await.unwrap();
let closed = client.close().await.unwrap();
(opened, sent, reply, closed)
assert_eq!(open_negotiated(&opened).params.dscp, 184);
#[cfg(not(any(
target_os = "fuchsia", target_os = "redox", target_os = "solaris",
target_os = "illumos", target_os = "haiku",
)))]
assert_eq!(socket2::SockRef::from(&client.socket).tos_v4().unwrap() >> 2, 46);
client.send_probe().await.unwrap();
assert!(matches!(client.recv().await.unwrap().as_slice(), [ClientEvent::EchoReply { seq: 0, .. }]));
assert!(matches!(client.close().await.unwrap().as_slice(), [ClientEvent::SessionClosed { token, .. }] if *token == open_token(&opened)));
});
drop(async_server);
assert_eq!(
open_negotiated(&blocking_open),
open_negotiated(&async_open)
);
assert_eq!(
open_negotiated(&async_open).params.dscp,
184,
"negotiated Params::dscp is the raw wire byte for codepoint 46"
);
assert_matching_event_shape(&blocking_sent[0], &async_sent[0]);
assert_matching_event_shape(&blocking_reply[0], &async_reply[0]);
assert!(matches!(
blocking_close.as_slice(),
[ClientEvent::SessionClosed { token, .. }] if *token == open_token(&blocking_open)
));
assert!(matches!(
async_close.as_slice(),
[ClientEvent::SessionClosed { token, .. }] if *token == open_token(&async_open)
));
assert_matching_event_shape(&blocking_close[0], &async_close[0]);
}
#[test]
fn blocking_and_async_no_test_are_semantically_equivalent() {
let blocking_no_test_server = InTreeServer::start(ServerConfig::default());
let mut blocking_no_test_config = config(blocking_no_test_server.addr, None, 0);
blocking_no_test_config.run_mode = RunMode::NoTest;
let mut blocking_no_test = Client::connect(blocking_no_test_config).unwrap();
let blocking_no_test_open = blocking_no_test.open().unwrap();
assert!(blocking_no_test.negotiated_params().is_none());
drop(blocking_no_test_server);
let async_no_test_server = InTreeServer::start(ServerConfig::default());
let mut async_no_test_config = config(async_no_test_server.addr, None, 0);
async_no_test_config.run_mode = RunMode::NoTest;
let async_no_test_open = runtime().block_on(async {
let mut client = AsyncClient::connect(async_no_test_config).await.unwrap();
let opened = client.open().await.unwrap();
assert!(client.negotiated_params().is_none());
opened
});
drop(async_no_test_server);
assert!(matches!(
blocking_no_test_open,
OpenOutcome::NoTestCompleted { .. }
));
assert!(matches!(
async_no_test_open,
OpenOutcome::NoTestCompleted { .. }
));
assert_eq!(
open_negotiated(&blocking_no_test_open),
open_negotiated(&async_no_test_open)
);
}
#[test]
@@ -724,50 +659,6 @@ fn blocking_and_async_complete_every_caller_paced_probe() {
}
}
#[test]
fn blocking_and_async_reject_probes_before_open_and_after_close() {
let blocking_server = InTreeServer::start(ServerConfig::default());
let mut blocking = Client::connect(config(blocking_server.addr, None, 0)).unwrap();
let blocking_before = blocking.send_probe().unwrap_err();
blocking.open().unwrap();
let blocking_reopen = blocking.open().unwrap_err();
blocking.close().unwrap();
let blocking_after_send = blocking.send_probe().unwrap_err();
let blocking_after_close = blocking.close().unwrap_err();
drop(blocking_server);
let async_server = InTreeServer::start(ServerConfig::default());
let (async_before, async_reopen, async_after_send, async_after_close) =
runtime().block_on(async {
let mut client = AsyncClient::connect(config(async_server.addr, None, 0))
.await
.unwrap();
let before = client.send_probe().await.unwrap_err();
client.open().await.unwrap();
let reopen = client.open().await.unwrap_err();
client.close().await.unwrap();
let after_send = client.send_probe().await.unwrap_err();
let after_close = client.close().await.unwrap_err();
(before, reopen, after_send, after_close)
});
drop(async_server);
assert_eq!(error_name(&blocking_before), "not open");
assert_eq!(error_name(&blocking_reopen), "already open");
assert_eq!(error_name(&blocking_after_send), "already closed");
assert_eq!(error_name(&blocking_after_close), "already closed");
assert_eq!(error_name(&blocking_before), error_name(&async_before));
assert_eq!(error_name(&blocking_reopen), error_name(&async_reopen));
assert_eq!(
error_name(&blocking_after_send),
error_name(&async_after_send)
);
assert_eq!(
error_name(&blocking_after_close),
error_name(&async_after_close)
);
}
#[test]
fn ignored_open_traffic_uses_one_request_and_deadline_per_attempt() {
let server = start_server(move |socket, tx| {
crates/irtt-client/src/client/tests/close.rs
-14
View File
@@ -93,20 +93,6 @@ fn recv_available_stops_after_peer_close() {
server.join();
}
#[test]
fn normal_echo_reply_does_not_close_session() {
let params = default_params();
let (mut client, server) = open_client_with_echo_server(&params);
client.send_probe().unwrap();
let events = client.recv_once().unwrap();
assert!(matches!(events.as_slice(), [ClientEvent::EchoReply { .. }]));
assert!(client.send_probe().is_ok());
client.close().unwrap();
server.join();
}
#[test]
fn close_flagged_duplicate_emits_duplicate_then_closes() {
let params = default_params();
crates/irtt-client/src/client/tests/config.rs
-19
View File
@@ -1,24 +1,5 @@
use super::*;
#[test]
fn client_config_default() {
let config = ClientConfig::default();
assert_eq!(config.duration, Some(Duration::from_secs(3)));
assert_eq!(config.interval, Duration::from_secs(1));
assert_eq!(config.length, 0);
assert_eq!(config.received_stats, ReceivedStats::Both);
assert_eq!(config.stamp_at, StampAt::Both);
assert_eq!(config.clock, Clock::Both);
assert_eq!(config.dscp, 0);
assert_eq!(config.hmac_key, None);
assert_eq!(config.server_fill, None);
assert_eq!(config.open_timeouts, DEFAULT_OPEN_TIMEOUTS);
assert_eq!(config.run_mode, RunMode::Normal);
assert_eq!(config.negotiation_policy, NegotiationPolicy::Strict);
assert_eq!(config.probe_timeout, Duration::from_secs(4));
assert_eq!(config.max_pending_probes, 4096);
}
#[test]
fn params_from_config_maps_compatibility_fields() {
let config = ClientConfig {
crates/irtt-client/src/client/tests/dscp.rs
+5 -16
View File
@@ -1,5 +1,4 @@
use super::*;
use crate::socket_options::socket_traffic_class;
#[test]
#[cfg(not(any(
@@ -25,20 +24,17 @@ fn normal_open_applies_negotiated_traffic_class_after_open_and_close_clears_it()
let mut client = Client::connect(config).unwrap();
assert_eq!(
socket_traffic_class(&client.socket, client.remote).unwrap() & 0xfc,
socket2::SockRef::from(&client.socket).tos_v4().unwrap() & 0xfc,
0
);
assert_open_started(client.open().unwrap());
assert_eq!(
socket_traffic_class(&client.socket, client.remote).unwrap() & 0xfc,
socket2::SockRef::from(&client.socket).tos_v4().unwrap() & 0xfc,
184
);
client.close().unwrap();
assert_eq!(
socket_traffic_class(&client.socket, client.remote).unwrap(),
0
);
assert_eq!(socket2::SockRef::from(&client.socket).tos_v4().unwrap(), 0);
server.join();
}
@@ -69,10 +65,7 @@ fn normal_open_uses_negotiated_traffic_class_not_requested_dscp() {
negotiated: 0,
}]
);
assert_eq!(
socket_traffic_class(&client.socket, client.remote).unwrap(),
0
);
assert_eq!(socket2::SockRef::from(&client.socket).tos_v4().unwrap(), 0);
server.join();
}
@@ -128,10 +121,6 @@ fn authenticated_peer_close_clears_negotiated_traffic_class() {
]
));
assert_eq!(client.applied_traffic_class, None);
assert_eq!(
socket_traffic_class(&client.socket, client.remote).unwrap(),
0
);
assert_eq!(socket2::SockRef::from(&client.socket).tos_v4().unwrap(), 0);
server.join();
}
crates/irtt-client/src/client/tests/hmac.rs
-18
View File
@@ -1,23 +1,5 @@
use super::*;
#[test]
fn hmac_open_success() {
let key = b"secret".to_vec();
let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
config.hmac_key = Some(key.clone());
let params = params_from_config(&config).unwrap();
let server = start_fake_server(move |socket, tx| {
let (request, peer) = recv_request(&socket, &tx);
verify_packet_hmac(&key, &request).unwrap();
let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, Some(&key));
socket.send_to(&reply, peer).unwrap();
});
config.server_addr = server.addr.to_string();
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
server.join();
}
#[test]
fn hmac_open_ignores_missing_hmac_before_valid_reply() {
let key = b"secret".to_vec();
crates/irtt-client/src/client/tests/mod.rs
+1 -1
View File
@@ -1,5 +1,5 @@
use super::*;
use crate::config::{NegotiationPolicy, DEFAULT_OPEN_TIMEOUTS};
use crate::config::NegotiationPolicy;
use crate::{
session::machine::{
compute_one_way, compute_rtt, params_from_config, sequence_is_after, sequence_is_before,
crates/irtt-client/src/client/tests/negotiation.rs
-125
View File
@@ -17,38 +17,6 @@ fn rejection_reason(requested: &Params, returned: &Params, policy: NegotiationPo
}
}
#[test]
fn strict_negotiation_accepts_identical_params() {
let config = ClientConfig::default();
let params = params_from_config(&config).unwrap();
let negotiated = assert_negotiates(&params, &params, NegotiationPolicy::Strict);
assert_eq!(negotiated.params, params);
assert!(negotiated.restrictions.is_empty());
}
#[test]
fn strict_negotiation_rejects_changed_params() {
let config = ClientConfig {
dscp: 46,
..ClientConfig::default()
};
let requested = params_from_config(&config).unwrap();
assert_eq!(
requested.dscp, 184,
"codepoint 46 must become raw wire byte 184"
);
let mut returned = requested.clone();
returned.dscp = 0;
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::DscpChanged {
requested: 46,
negotiated: 0,
}
.message()
);
}
#[test]
fn loose_negotiation_accepts_server_restricted_params() {
let config = ClientConfig::default();
@@ -88,79 +56,6 @@ fn loose_negotiation_rejects_non_positive_returned_interval() {
}
}
#[test]
fn loose_negotiation_accepts_documented_duration_combinations() {
let finite_config = ClientConfig::default();
let requested_finite = params_from_config(&finite_config).unwrap();
let returned_same_finite = requested_finite.clone();
assert!(assert_negotiates(
&requested_finite,
&returned_same_finite,
NegotiationPolicy::Loose
)
.restrictions
.is_empty());
let mut returned_shorter_finite = requested_finite.clone();
returned_shorter_finite.duration_ns /= 2;
assert_eq!(
assert_negotiates(
&requested_finite,
&returned_shorter_finite,
NegotiationPolicy::Loose
)
.restrictions,
vec![NegotiationRestriction::DurationReduced {
requested_ns: requested_finite.duration_ns,
negotiated_ns: returned_shorter_finite.duration_ns,
}]
);
let continuous_config = ClientConfig {
duration: None,
..ClientConfig::default()
};
let requested_continuous = params_from_config(&continuous_config).unwrap();
let returned_continuous = requested_continuous.clone();
assert!(assert_negotiates(
&requested_continuous,
&returned_continuous,
NegotiationPolicy::Loose
)
.restrictions
.is_empty());
let mut returned_finite = requested_continuous.clone();
returned_finite.duration_ns = 1_000_000_000;
assert_eq!(
assert_negotiates(
&requested_continuous,
&returned_finite,
NegotiationPolicy::Loose
)
.restrictions,
vec![NegotiationRestriction::DurationReduced {
requested_ns: requested_continuous.duration_ns,
negotiated_ns: returned_finite.duration_ns,
}]
);
}
#[test]
fn loose_negotiation_rejects_finite_request_returned_continuous() {
let config = ClientConfig::default();
let requested = params_from_config(&config).unwrap();
let mut returned = requested.clone();
returned.duration_ns = 0;
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
"server returned continuous duration for finite request"
);
}
#[test]
fn loose_negotiation_rejects_negative_returned_length() {
let config = ClientConfig::default();
@@ -238,26 +133,6 @@ fn loose_negotiation_records_dscp_disabled_by_server() {
);
}
#[test]
fn strict_negotiation_rejects_dscp_disabled_by_server_as_specific_restriction() {
let config = ClientConfig {
dscp: 46,
..ClientConfig::default()
};
let requested = params_from_config(&config).unwrap();
let mut returned = requested.clone();
returned.dscp = 0;
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::DscpChanged {
requested: 46,
negotiated: 0,
}
.message()
);
}
#[test]
fn negotiation_rejects_unsupported_dscp_changes() {
let config = ClientConfig {
crates/irtt-client/src/client/tests/no_test.rs
-16
View File
@@ -113,19 +113,3 @@ fn open_fails_after_no_test_completed() {
assert!(matches!(client.open(), Err(ClientError::AlreadyCompleted)));
server.join();
}
#[test]
fn no_test_success_validates_params() {
let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
config.run_mode = RunMode::NoTest;
let params = params_from_config(&config).unwrap();
let server = no_test_server(params.clone(), 0);
config.server_addr = server.addr.to_string();
let mut client = Client::connect(config).unwrap();
let negotiated = assert_no_test_completed(client.open().unwrap());
assert_eq!(negotiated.params, params);
assert_eq!(client.applied_traffic_class, None);
server.join();
}
crates/irtt-client/src/client/tests/open.rs
-14
View File
@@ -136,20 +136,6 @@ fn post_token_negotiation_failure_sends_cleanup_close() {
server.join();
}
#[test]
fn successful_open_handshake() {
let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
let params = params_from_config(&config).unwrap();
let server = open_success_server(params.clone());
let mut client = Client::connect(default_test_config(server.addr)).unwrap();
let negotiated = assert_open_started(client.open().unwrap());
assert_eq!(negotiated.params, params);
assert_eq!(client.applied_traffic_class, Some(0));
server.join();
}
#[test]
fn successful_open_restores_configured_receive_timeout() {
let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
crates/irtt-client/src/client/tests/ttl.rs
+2 -4
View File
@@ -1,5 +1,3 @@
use crate::socket_options::socket_ttl;
use super::*;
#[test]
@@ -27,13 +25,13 @@ fn connect_applies_configured_ipv4_ttl_before_open_and_close_preserves_it() {
config.socket_config.ttl = Some(64);
let mut client = Client::connect(config).unwrap();
assert_eq!(socket_ttl(&client.socket, client.remote).unwrap(), 64);
assert_eq!(socket2::SockRef::from(&client.socket).ttl_v4().unwrap(), 64);
client.open().unwrap();
server.join();
client.close().unwrap();
assert_eq!(socket_ttl(&client.socket, client.remote).unwrap(), 64);
assert_eq!(socket2::SockRef::from(&client.socket).ttl_v4().unwrap(), 64);
}
#[test]
crates/irtt-client/src/managed/blocking.rs
+5 -13
View File
@@ -157,22 +157,14 @@ impl BlockingManagedClient {
}
fn join_worker(&mut self) -> Result<ManagedOutcome, BlockingManagedJoinError> {
join_worker(
self.worker
.take()
.expect("blocking managed worker is joined at most once"),
)
self.worker
.take()
.expect("blocking managed worker is joined at most once")
.join()
.map_err(|_| BlockingManagedJoinError::WorkerPanicked)
}
}
pub(super) fn join_worker(
worker: JoinHandle<ManagedOutcome>,
) -> Result<ManagedOutcome, BlockingManagedJoinError> {
worker
.join()
.map_err(|_| BlockingManagedJoinError::WorkerPanicked)
}
impl Drop for BlockingManagedClient {
fn drop(&mut self) {
if self.worker.is_some() {
crates/irtt-client/src/managed/blocking_tests.rs
+1 -11
View File
@@ -12,8 +12,7 @@ use irtt_proto::{
use tokio::{runtime::Builder, time::timeout};
use super::{
blocking::{join_worker, WorkerRuntime},
BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, ManagedCompletionPolicy,
blocking::WorkerRuntime, BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy,
ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedPacing, ManagedTargetConfig,
ManagedTargetEndReason,
};
@@ -309,15 +308,6 @@ fn dropping_owner_stops_and_joins_worker() {
server.finish();
}
#[test]
fn worker_panic_is_a_join_error() {
let worker = thread::spawn(|| -> super::ManagedOutcome { panic!("test worker panic") });
assert_eq!(
join_worker(worker),
Err(BlockingManagedJoinError::WorkerPanicked)
);
}
#[test]
fn dropping_worker_runtime_inside_async_context_does_not_panic() {
test_runtime().block_on(async {
crates/irtt-client/src/managed/model.rs
-196
View File
@@ -449,199 +449,3 @@ pub(crate) fn classify_client_error(
message: Arc::from(error.to_string()),
}
}
#[cfg(test)]
mod tests {
use std::{collections::TryReserveError, io};
use super::*;
fn io_error() -> io::Error {
io::Error::other("boom")
}
fn try_reserve_error() -> TryReserveError {
Vec::<u8>::new().try_reserve(usize::MAX).unwrap_err()
}
fn classification_cases() -> Vec<(ClientError, ManagedTargetFailureKind)> {
vec![
(
ClientError::Resolve {
addr: "example.invalid:2112".to_owned(),
},
ManagedTargetFailureKind::Resolve,
),
(
ClientError::Socket(io_error()),
ManagedTargetFailureKind::Socket,
),
(
ClientError::DatagramLengthMismatch {
expected: 64,
actual: 32,
},
ManagedTargetFailureKind::Socket,
),
(
ClientError::SocketOption {
operation: "set read timeout",
remote: "127.0.0.1:2112".parse().unwrap(),
source: io_error(),
},
ManagedTargetFailureKind::SocketOption,
),
(
ClientError::ReadTimeoutRestore { source: io_error() },
ManagedTargetFailureKind::SocketOption,
),
(
ClientError::Protocol(irtt_proto::ProtoError::BadMagic),
ManagedTargetFailureKind::Protocol,
),
(
ClientError::ProtocolVersionMismatch {
requested: 1,
received: 2,
},
ManagedTargetFailureKind::Protocol,
),
(ClientError::ZeroToken, ManagedTargetFailureKind::Protocol),
(
ClientError::UnexpectedNoTestReply,
ManagedTargetFailureKind::Protocol,
),
(
ClientError::NonZeroNoTestToken { token: 7 },
ManagedTargetFailureKind::Protocol,
),
(
ClientError::ServerRejected,
ManagedTargetFailureKind::Protocol,
),
(
ClientError::NegotiationRejected {
reason: "duration".to_owned(),
},
ManagedTargetFailureKind::Protocol,
),
(ClientError::OpenTimeout, ManagedTargetFailureKind::Timeout),
(
ClientError::InvalidConfig {
reason: "interval".to_owned(),
},
ManagedTargetFailureKind::Configuration,
),
(
ClientError::OpenTimeoutTooSmall {
timeout: Duration::from_millis(1),
minimum: Duration::from_millis(10),
},
ManagedTargetFailureKind::Configuration,
),
(
ClientError::NoOpenTimeouts,
ManagedTargetFailureKind::Configuration,
),
(
ClientError::AllocationFailed {
operation: "pending probes",
source: try_reserve_error(),
},
ManagedTargetFailureKind::ResourceExhausted,
),
(
ClientError::CounterOverflow {
counter: "packets_sent",
},
ManagedTargetFailureKind::ResourceExhausted,
),
(
ClientError::DurationOverflow,
ManagedTargetFailureKind::ResourceExhausted,
),
(
ClientError::PendingLimitExceeded { limit: 8 },
ManagedTargetFailureKind::ResourceExhausted,
),
(
ClientError::NoTokioRuntime,
ManagedTargetFailureKind::InvalidState,
),
(ClientError::NotOpen, ManagedTargetFailureKind::InvalidState),
(
ClientError::AlreadyOpen,
ManagedTargetFailureKind::InvalidState,
),
(
ClientError::AlreadyCompleted,
ManagedTargetFailureKind::InvalidState,
),
(
ClientError::AlreadyClosed,
ManagedTargetFailureKind::InvalidState,
),
(
ClientError::StalePreparedProbe {
prepared_seq: 3,
next_wire_seq: 4,
},
ManagedTargetFailureKind::InvalidState,
),
(
ClientError::PendingSequenceCollision { seq: 9 },
ManagedTargetFailureKind::InvalidState,
),
]
}
#[test]
fn every_client_error_classifies_to_its_documented_kind() {
for (error, expected) in classification_cases() {
let failure = classify_client_error(ManagedTargetFailurePhase::Opening, &error);
assert_eq!(
failure.kind, expected,
"unexpected classification for {error:?}"
);
assert_eq!(failure.phase, ManagedTargetFailurePhase::Opening);
assert_eq!(&*failure.message, error.to_string());
}
}
#[test]
fn classification_preserves_the_reported_phase() {
for phase in [
ManagedTargetFailurePhase::Connecting,
ManagedTargetFailurePhase::Opening,
ManagedTargetFailurePhase::Sending,
ManagedTargetFailurePhase::Receiving,
ManagedTargetFailurePhase::Timing,
ManagedTargetFailurePhase::Closing,
] {
let failure = classify_client_error(phase, &ClientError::OpenTimeout);
assert_eq!(failure.phase, phase);
}
}
#[test]
fn phase_and_kind_display_without_debug_spelling() {
assert_eq!(
ManagedTargetFailurePhase::Connecting.to_string(),
"connecting"
);
assert_eq!(ManagedTargetFailurePhase::Closing.to_string(), "closing");
assert_eq!(
ManagedTargetFailureKind::SocketOption.to_string(),
"socket option"
);
assert_eq!(
ManagedTargetFailureKind::ResourceExhausted.to_string(),
"resource exhausted"
);
assert_eq!(
ManagedTargetFailureKind::InvalidState.to_string(),
"invalid state"
);
assert_eq!(ManagedTargetFailureKind::Other.to_string(), "other");
}
}
crates/irtt-client/src/managed/schedule.rs
+8 -12
View File
@@ -70,7 +70,14 @@ impl ProbeSchedule {
) -> Result<ScheduleCommit, ClientError> {
let (scheduled_at, next_send_at) =
advance_cadence(scheduled_at, self.interval, permission_at)?;
Ok(self.finish_commit(scheduled_at, next_send_at))
Ok(ScheduleCommit {
scheduled_at,
next_send_at: if self.end_at.is_some_and(|end| next_send_at >= end) {
None
} else {
Some(next_send_at)
},
})
}
pub(crate) fn commit(&mut self, commit: ScheduleCommit) {
@@ -90,17 +97,6 @@ impl ProbeSchedule {
pub(crate) fn is_finished(&self) -> bool {
self.next_send_at.is_none()
}
fn finish_commit(&self, scheduled_at: Instant, next_send_at: Instant) -> ScheduleCommit {
ScheduleCommit {
scheduled_at,
next_send_at: if self.end_at.is_some_and(|end| next_send_at >= end) {
None
} else {
Some(next_send_at)
},
}
}
}
pub(crate) fn advance_cadence(
crates/irtt-client/src/managed/task.rs
+16 -82
View File
@@ -181,6 +181,7 @@ pub struct ManagedClientHandle {
update_submission_hook: Option<Arc<UpdateSubmissionTestHook>>,
}
// Pauses between admission and enqueue to exercise stop/seal linearization.
#[cfg(test)]
#[derive(Default)]
struct UpdateSubmissionTestHook {
@@ -402,10 +403,9 @@ impl Future for ManagedStopReceipt {
type ConnectFuture =
Pin<Box<dyn Future<Output = Result<AsyncClient, ClientError>> + Send + 'static>>;
type WakeFuture = Pin<Box<dyn Future<Output = Option<watch::Receiver<()>>> + Send + 'static>>;
// Snapshots at publication verify status is visible before the corresponding event.
#[cfg(test)]
type EventObservations = Arc<std::sync::Mutex<Vec<(ManagedEvent, Arc<ManagedStatus>)>>>;
#[cfg(test)]
type StaggerObservations = Arc<std::sync::Mutex<Vec<(usize, Duration)>>>;
fn arm_wake(mut receiver: watch::Receiver<()>) -> WakeFuture {
Box::pin(async move {
@@ -525,17 +525,7 @@ impl TargetRuntime {
/// The client owns `packets_sent`; this runtime only mirrors it so a
/// terminal outcome can report it after the client is gone.
fn sync_packets_sent(&mut self, client: &AsyncClient) {
self.record_packets_sent(client.packets_sent());
}
/// Adopt an already-read authoritative sent count.
fn record_packets_sent(&mut self, packets_sent: u64) {
self.counters.packets_sent = packets_sent;
}
/// Account for one client event against this target's durable counters.
fn observe(&mut self, event: &ClientEvent) {
self.counters.observe(event);
self.counters.packets_sent = client.packets_sent();
}
/// Build this target's durable outcome.
@@ -723,21 +713,17 @@ pub struct ManagedClientTask {
#[cfg(test)]
event_observations: Option<EventObservations>,
#[cfg(test)]
stagger_observations: Option<StaggerObservations>,
#[cfg(test)]
drain_test_hook: DrainTestHook,
// Counts work that cannot be observed through lossy presentation events.
#[cfg(test)]
timeout_inspections: usize,
#[cfg(test)]
deadline_inspections: usize,
}
// Holds queued replies past the drain deadline to exercise the final receive budget.
#[cfg(test)]
#[derive(Default)]
struct DrainTestHook {
defer_work_until_deadline: bool,
fail_receive: bool,
fail_close: bool,
}
impl ManagedClientTask {
@@ -837,7 +823,7 @@ impl ManagedClientTask {
fn publish_client_events(&mut self, index: usize, events: Vec<ClientEvent>) {
for event in events {
self.targets[index].observe(&event);
self.targets[index].counters.observe(&event);
self.publish_event(ManagedEvent::Client {
target: self.targets[index].instance.clone(),
event,
@@ -1421,10 +1407,6 @@ impl ManagedClientTask {
return true;
}
#[cfg(test)]
let injected_receive_failure = self.take_drain_receive_failure();
#[cfg(not(test))]
let injected_receive_failure = None;
let mut retained_state_changed = false;
let received = match client.poll_recv(cx) {
Poll::Pending => false,
@@ -1441,13 +1423,7 @@ impl ManagedClientTask {
return self.begin_close(index, client, primary_end, cleanup_failure, now);
}
};
if let Some(error) = injected_receive_failure {
self.targets[index].sync_packets_sent(&client);
cleanup_failure.get_or_insert_with(|| {
classify_client_error(ManagedTargetFailurePhase::Receiving, &error)
});
return self.begin_close(index, client, primary_end, cleanup_failure, now);
}
if client.is_peer_closed() {
self.targets[index].sync_packets_sent(&client);
self.finish_target(index, ManagedTargetEndReason::PeerClosed, cleanup_failure);
@@ -1525,17 +1501,7 @@ impl ManagedClientTask {
Poll::Ready(Ok(events)) => {
self.targets[index].sync_packets_sent(&client);
self.publish_client_events(index, events);
#[cfg(test)]
let injected_close_failure = self.take_drain_close_failure().map(|error| {
classify_client_error(ManagedTargetFailurePhase::Closing, &error)
});
#[cfg(not(test))]
let injected_close_failure = None;
self.finish_target(
index,
primary_end,
cleanup_failure.or(injected_close_failure),
);
self.finish_target(index, primary_end, cleanup_failure);
false
}
Poll::Ready(Err(error)) => {
@@ -1616,26 +1582,6 @@ impl ManagedClientTask {
true
}
#[cfg(test)]
fn take_drain_receive_failure(&mut self) -> Option<ClientError> {
if !mem::take(&mut self.drain_test_hook.fail_receive) {
return None;
}
Some(ClientError::Socket(std::io::Error::other(
"injected drain receive failure",
)))
}
#[cfg(test)]
fn take_drain_close_failure(&mut self) -> Option<ClientError> {
if !mem::take(&mut self.drain_test_hook.fail_close) {
return None;
}
Some(ClientError::Socket(std::io::Error::other(
"injected drain close failure",
)))
}
fn active_count(&self) -> usize {
self.targets
.iter()
@@ -1643,7 +1589,7 @@ impl ManagedClientTask {
.count()
}
fn active_stagger_spacing(&self) -> Option<(usize, Duration)> {
fn active_stagger_spacing(&self) -> Option<Duration> {
let mut active = 0;
let mut minimum: Option<Duration> = None;
for target in &self.targets {
@@ -1661,7 +1607,7 @@ impl ManagedClientTask {
.interval();
minimum = minimum.into_iter().chain(Some(interval)).min();
}
minimum.map(|interval| (active, stagger_spacing(interval, active)))
minimum.map(|interval| stagger_spacing(interval, active))
}
fn stagger_target_added(&mut self, now: Instant) {
@@ -1672,7 +1618,7 @@ impl ManagedClientTask {
let candidate = self
.last_stagger_send
.zip(self.active_stagger_spacing())
.and_then(|(last, (_, spacing))| last.checked_add(spacing))
.and_then(|(last, spacing)| last.checked_add(spacing))
.filter(|gate| *gate > now);
self.send_gate = candidate.map(|candidate| existing.min(candidate));
}
@@ -1708,21 +1654,17 @@ impl ManagedClientTask {
fn record_stagger_acceptance(
&mut self,
result: SendResult,
stagger_spacing: Option<(usize, Duration)>,
stagger_spacing: Option<Duration>,
accepted_at: Instant,
) {
if !result.accepted() {
return;
}
let Some((_active, spacing)) = stagger_spacing else {
let Some(spacing) = stagger_spacing else {
return;
};
self.last_stagger_send = Some(accepted_at);
self.send_gate = self.next_stagger_gate(spacing, accepted_at);
#[cfg(test)]
if let Some(observations) = &self.stagger_observations {
observations.lock().unwrap().push((_active, spacing));
}
}
fn poll_one_send(
@@ -1730,7 +1672,7 @@ impl ManagedClientTask {
index: usize,
cx: &mut Context<'_>,
now: Instant,
stagger_spacing: Option<(usize, Duration)>,
stagger_spacing: Option<Duration>,
) -> SendResult {
if !self.targets[index].desired || self.targets[index].retirement.is_some() {
self.targets[index].send_waiting = false;
@@ -1832,7 +1774,7 @@ impl ManagedClientTask {
} else {
None
};
self.targets[index].record_packets_sent(after);
self.targets[index].counters.packets_sent = after;
match result {
Poll::Pending => {
self.targets[index].send_waiting = true;
@@ -2109,11 +2051,7 @@ impl ManagedClientTask {
.all(|target| matches!(target.state, TargetState::Terminal))
}
fn next_deadline(&mut self) -> Option<Instant> {
#[cfg(test)]
{
self.deadline_inspections += self.targets.len();
}
fn next_deadline(&self) -> Option<Instant> {
let mut non_send_deadline = None;
let mut send_deadline = None;
for target in &self.targets {
@@ -2541,13 +2479,9 @@ fn build_task(
#[cfg(test)]
event_observations: None,
#[cfg(test)]
stagger_observations: None,
#[cfg(test)]
drain_test_hook: DrainTestHook::default(),
#[cfg(test)]
timeout_inspections: 0,
#[cfg(test)]
deadline_inspections: 0,
};
let handle = ManagedClientHandle {
stop,
crates/irtt-client/src/managed/tests.rs
+5 -153
View File
@@ -16,7 +16,7 @@ use irtt_proto::{
use tokio::runtime::{Builder, Runtime};
use super::*;
use crate::{socket::resolution_call_counts, ClientAuthConfig, NegotiationPolicy, RunMode};
use crate::{ClientAuthConfig, NegotiationPolicy, RunMode};
const TOKEN: u64 = 0x1234_5678_90ab_cdef;
@@ -388,44 +388,6 @@ fn assert_rotated_without_catchup(sends: &[(TargetInstance, u32, Instant, Instan
}
}
fn run_negotiated_stagger_case(
requested: Duration,
negotiated: &[Duration],
) -> (Vec<(usize, Duration)>, Vec<u64>) {
let servers = negotiated
.iter()
.map(|interval| start_server_negotiating(ServerBehavior::Echo, None, Some(*interval)))
.collect::<Vec<_>>();
let mut managed = config(ManagedPacing::Staggered);
managed.client.duration = Some(Duration::from_millis(190));
managed.client.interval = requested;
managed.client.negotiation_policy = NegotiationPolicy::Loose;
let targets = servers
.iter()
.enumerate()
.map(|(index, server)| target(&format!("target-{index}"), server.addr))
.collect();
let (mut task, _) = ManagedClient::task(managed, targets).unwrap();
let observations = Arc::new(Mutex::new(Vec::new()));
task.stagger_observations = Some(Arc::clone(&observations));
let outcome = runtime().block_on(task);
let packets_sent = (0..servers.len())
.map(|index| {
outcome
.recent_target_outcomes
.iter()
.find(|target| target.target.id.as_ref() == format!("target-{index}"))
.unwrap()
.packets_sent
})
.collect();
for server in servers {
server.finish();
}
let spacings = observations.lock().unwrap().clone();
(spacings, packets_sent)
}
fn run_deferred_drain_burst(packet_count: usize) -> ManagedOutcome {
let server = start_server(ServerBehavior::DeferredBurst(packet_count), None);
let mut managed = config(ManagedPacing::Staggered);
@@ -459,13 +421,11 @@ fn run_deferred_drain_burst(packet_count: usize) -> ManagedOutcome {
#[test]
fn construction_is_runtime_and_io_free() {
let before = resolution_call_counts();
let (task, handle) = ManagedClient::task(
config(ManagedPacing::Staggered),
vec![ManagedTargetConfig::new("dns", "no-such-host.invalid")],
)
.unwrap();
assert_eq!(resolution_call_counts(), before);
assert_eq!(handle.status().lifecycle, ManagedLifecycle::NotStarted);
drop(task);
}
@@ -1400,7 +1360,6 @@ fn first_poll_without_runtime_fails_durably() {
#[test]
fn pre_poll_stop() {
let before = resolution_call_counts();
let (task, handle) = ManagedClient::task(
config(ManagedPacing::Staggered),
vec![ManagedTargetConfig::new("one", "no-such-host.invalid")],
@@ -1413,7 +1372,6 @@ fn pre_poll_stop() {
};
assert!(poll_once(receipt.as_mut()).is_ready());
assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested);
assert_eq!(resolution_call_counts(), before);
}
#[test]
@@ -1604,45 +1562,6 @@ fn staggered_pacing_fairness() {
second.finish();
}
#[test]
fn staggered_pacing_uses_all_active_negotiated_intervals() {
let requested = Duration::from_millis(120);
let (single, single_packets) =
run_negotiated_stagger_case(requested, &[Duration::from_millis(40)]);
assert!(single
.iter()
.all(|entry| *entry == (1, Duration::from_millis(40))));
assert!(single_packets[0] >= 4);
let (different, different_packets) = run_negotiated_stagger_case(
requested,
&[Duration::from_millis(30), Duration::from_millis(90)],
);
let while_both_active = different
.iter()
.filter(|(active, _)| *active == 2)
.collect::<Vec<_>>();
assert!(!while_both_active.is_empty());
assert!(while_both_active
.iter()
.all(|(_, spacing)| *spacing == Duration::from_millis(15)));
assert!(different_packets[0] > different_packets[1]);
let (equal, _) = run_negotiated_stagger_case(
requested,
&[Duration::from_millis(60), Duration::from_millis(60)],
);
let while_both_active = equal
.iter()
.filter(|(active, _)| *active == 2)
.collect::<Vec<_>>();
assert!(!while_both_active.is_empty());
assert!(while_both_active
.iter()
.all(|(_, spacing)| *spacing == Duration::from_millis(30)));
}
#[test]
fn stagger_gate_tracks_active_membership() {
let first =
@@ -1690,7 +1609,7 @@ fn stagger_gate_tracks_active_membership() {
let accepted_at = Instant::now();
task.record_stagger_acceptance(
SendResult::Failed { accepted: true },
Some((2, Duration::from_millis(50))),
Some(Duration::from_millis(50)),
accepted_at,
);
assert_eq!(
@@ -1752,11 +1671,7 @@ fn the_stagger_gate_advances_by_one_spacing_per_send_without_accumulating_latene
let start = Instant::now();
let lateness = Duration::from_millis(3);
task.record_stagger_acceptance(
SendResult::Ready { accepted: true },
Some((1, spacing)),
start,
);
task.record_stagger_acceptance(SendResult::Ready { accepted: true }, Some(spacing), start);
assert_eq!(task.send_gate, Some(start + spacing), "first send anchors");
// Every later send is observed `lateness` after the gate that released it.
@@ -1766,7 +1681,7 @@ fn the_stagger_gate_advances_by_one_spacing_per_send_without_accumulating_latene
let released_at = start + spacing * slot;
task.record_stagger_acceptance(
SendResult::Ready { accepted: true },
Some((1, spacing)),
Some(spacing),
released_at + lateness,
);
assert_eq!(
@@ -1782,7 +1697,7 @@ fn the_stagger_gate_advances_by_one_spacing_per_send_without_accumulating_latene
let resumed_at = stalled_gate + spacing;
task.record_stagger_acceptance(
SendResult::Ready { accepted: true },
Some((1, spacing)),
Some(spacing),
resumed_at,
);
assert_eq!(task.send_gate, Some(resumed_at + spacing));
@@ -1926,69 +1841,6 @@ fn pending_limit_failure_drains_reply_and_closes_session() {
assert!(has_close(&records));
}
#[test]
fn drain_failures_preserve_primary_outcome_and_first_cleanup_failure() {
let completed_server = start_server(ServerBehavior::Echo, None);
let (mut completed_task, _) = ManagedClient::task(
config(ManagedPacing::Staggered),
vec![target("completed", completed_server.addr)],
)
.unwrap();
completed_task.drain_test_hook.fail_receive = true;
completed_task.drain_test_hook.fail_close = true;
let completed = runtime().block_on(completed_task);
let completed_target = &completed.recent_target_outcomes[0];
assert!(matches!(
completed_target.end_reason,
ManagedTargetEndReason::TestComplete
));
assert_eq!(completed.failed_target_outcomes, 0);
assert!(matches!(
completed_target.cleanup_failure,
Some(ManagedTargetFailure {
phase: ManagedTargetFailurePhase::Receiving,
kind: ManagedTargetFailureKind::Socket,
..
})
));
completed_server.finish();
let stopped_server = start_server(ServerBehavior::Echo, None);
let mut managed = config(ManagedPacing::Staggered);
managed.completion = ManagedCompletionPolicy::ExplicitStop;
managed.client.duration = None;
let (mut stopped_task, handle) =
ManagedClient::task(managed, vec![target("stopped", stopped_server.addr)]).unwrap();
stopped_task.drain_test_hook.fail_receive = true;
let seen = Arc::clone(&stopped_server.probe_seen);
let stopper = thread::spawn(move || {
let (flag, ready) = &*seen;
let guard = flag.lock().unwrap();
let (guard, timeout) = ready
.wait_timeout_while(guard, Duration::from_secs(2), |seen| !*seen)
.unwrap();
assert!(*guard && !timeout.timed_out());
drop(handle.stop());
});
let stopped = runtime().block_on(stopped_task);
stopper.join().unwrap();
let stopped_target = &stopped.recent_target_outcomes[0];
assert!(matches!(
stopped_target.end_reason,
ManagedTargetEndReason::Stopped
));
assert_eq!(stopped.failed_target_outcomes, 0);
assert!(matches!(
stopped_target.cleanup_failure,
Some(ManagedTargetFailure {
phase: ManagedTargetFailurePhase::Receiving,
kind: ManagedTargetFailureKind::Socket,
..
})
));
stopped_server.finish();
}
#[test]
fn final_drain_overflow_is_rejected_and_preserves_success() {
let mut invalid = config(ManagedPacing::Staggered);
crates/irtt-client/src/metadata.rs
-44
View File
@@ -135,37 +135,6 @@ mod tests {
);
}
#[test]
fn preferred_receive_wall_rejects_implausibly_old_kernel_timestamp() {
let userspace = userspace_wall();
let kernel = SystemTime::UNIX_EPOCH;
assert_eq!(
meta_with_kernel_rx(kernel).preferred_receive_wall(userspace),
userspace
);
}
#[test]
fn preferred_receive_wall_handles_extreme_timestamps_without_panicking() {
let userspace = userspace_wall();
let far_future = SystemTime::UNIX_EPOCH + Duration::from_secs(u64::from(u32::MAX)) * 4;
let before_epoch = SystemTime::UNIX_EPOCH - Duration::from_secs(u64::from(u32::MAX));
assert_eq!(
meta_with_kernel_rx(far_future).preferred_receive_wall(userspace),
userspace
);
assert_eq!(
meta_with_kernel_rx(before_epoch).preferred_receive_wall(userspace),
userspace
);
assert_eq!(
ReceiveMeta::default().preferred_receive_wall(before_epoch),
before_epoch
);
}
#[test]
fn metadata_preserves_kernel_timestamp_rejected_for_measurement() {
let userspace = userspace_wall();
@@ -180,19 +149,6 @@ mod tests {
);
}
#[test]
fn metadata_observed_traffic_class_zero_preserves_observed_zero() {
let packet_meta = PacketMeta::from(ReceiveMeta {
traffic_class: Some(0),
kernel_rx_timestamp: None,
});
assert_eq!(packet_meta.traffic_class, Some(0));
assert_eq!(packet_meta.dscp, Some(0));
assert_eq!(packet_meta.ecn, Some(0));
assert_eq!(packet_meta.kernel_rx_timestamp, None);
}
#[test]
fn metadata_observed_traffic_class_derives_dscp_and_ecn() {
let packet_meta = PacketMeta::from(ReceiveMeta {
crates/irtt-client/src/probe.rs
+22 -99
View File
@@ -1,5 +1,5 @@
use std::{
collections::{HashMap, HashSet, TryReserveError, VecDeque},
collections::{HashMap, HashSet, VecDeque},
time::{Instant, SystemTime},
};
@@ -85,7 +85,12 @@ impl PendingMap {
});
}
if self.map.len() == self.map.capacity() {
self.map.try_reserve(1).map_err(pending_allocation_failed)?;
self.map
.try_reserve(1)
.map_err(|source| ClientError::AllocationFailed {
operation: "pending probe storage",
source,
})?;
}
Ok(())
}
@@ -186,6 +191,12 @@ impl PendingMap {
ExpiredBatch { probes, more_due }
}
pub fn is_empty(&self) -> bool {
self.map.is_empty()
}
// Verify repeated or abandoned send preflights do not create pending probes.
#[cfg(test)]
pub fn len(&self) -> usize {
self.map.len()
}
@@ -234,34 +245,11 @@ impl PendingMap {
}
}
// Check sequence-reuse and kernel timestamp retention without consuming a probe.
#[cfg(test)]
pub fn contains(&self, wire_seq: u32) -> bool {
self.map.contains_key(&wire_seq)
}
#[cfg(test)]
pub fn capacity(&self) -> usize {
self.map.capacity()
}
#[cfg(test)]
pub fn linkage(&self) -> (Option<u32>, Option<u32>) {
(self.first, self.last)
}
#[cfg(test)]
pub fn entry_links(&self, wire_seq: u32) -> Option<(Option<u32>, Option<u32>)> {
self.map
.get(&wire_seq)
.map(|entry| (entry.previous, entry.next))
}
}
fn pending_allocation_failed(source: TryReserveError) -> ClientError {
ClientError::AllocationFailed {
operation: "pending probe storage",
source,
}
}
#[derive(Debug)]
@@ -312,31 +300,17 @@ impl TimedOutMap {
self.map.get_mut(&wire_seq)
}
pub fn clear(&mut self) {
self.map.clear();
self.insertion_order.clear();
}
#[cfg(test)]
pub fn len(&self) -> usize {
self.map.len()
}
#[cfg(feature = "tokio")]
pub fn latest_timeout_deadline(&self) -> Option<Instant> {
self.map.values().map(|probe| probe.timeout_at).max()
}
// Check sequence-reuse and kernel timestamp retention without consuming a probe.
#[cfg(test)]
pub fn contains(&self, wire_seq: u32) -> bool {
self.map.contains_key(&wire_seq)
}
#[cfg(test)]
fn insertion_order_len(&self) -> usize {
self.insertion_order.len()
}
fn evict_oldest(&mut self) {
while let Some(oldest_key) = self.insertion_order.pop_front() {
if self.map.remove(&oldest_key).is_some() {
@@ -438,23 +412,6 @@ mod tests {
));
}
#[test]
fn pending_map_preflight_reserves_without_changing_contents() {
let mut map = PendingMap::new(2);
let initial_capacity = map.capacity();
map.preflight_insert(7).unwrap();
let reserved_capacity = map.capacity();
assert!(reserved_capacity >= initial_capacity);
assert_eq!(map.len(), 0);
assert!(!map.contains(7));
map.preflight_insert(7).unwrap();
assert_eq!(map.capacity(), reserved_capacity);
assert_eq!(map.len(), 0);
assert!(!map.contains(7));
}
#[test]
fn pending_map_preflight_rejects_sequence_collision() {
let mut map = PendingMap::new(2);
@@ -466,21 +423,7 @@ mod tests {
map.preflight_insert(9),
Err(ClientError::PendingSequenceCollision { seq: 9 })
));
assert_eq!(map.len(), 1);
}
#[test]
fn pending_map_allocation_failure_maps_to_dedicated_error() {
let mut map = HashMap::<u32, PendingProbe>::new();
let source = map.try_reserve(usize::MAX).unwrap_err();
assert!(matches!(
pending_allocation_failed(source),
ClientError::AllocationFailed {
operation: "pending probe storage",
..
}
));
assert_eq!(map.map.len(), 1);
}
#[test]
@@ -499,10 +442,6 @@ mod tests {
map.latest_timeout_deadline(),
Some(now + Duration::from_secs(3))
);
assert_eq!(map.linkage(), (Some(1), Some(3)));
assert_eq!(map.entry_links(1), Some((None, Some(2))));
assert_eq!(map.entry_links(2), Some((Some(1), Some(3))));
assert_eq!(map.entry_links(3), Some((Some(2), None)));
assert!(map.remove(2).is_some());
assert_eq!(
@@ -513,9 +452,6 @@ mod tests {
map.latest_timeout_deadline(),
Some(now + Duration::from_secs(3))
);
assert_eq!(map.linkage(), (Some(1), Some(3)));
assert_eq!(map.entry_links(1), Some((None, Some(3))));
assert_eq!(map.entry_links(3), Some((Some(1), None)));
assert!(map.remove(1).is_some());
assert_eq!(
@@ -526,13 +462,10 @@ mod tests {
map.latest_timeout_deadline(),
Some(now + Duration::from_secs(3))
);
assert_eq!(map.linkage(), (Some(3), Some(3)));
assert_eq!(map.entry_links(3), Some((None, None)));
assert!(map.remove(3).is_some());
assert_eq!(map.next_timeout_deadline(), None);
assert_eq!(map.latest_timeout_deadline(), None);
assert_eq!(map.linkage(), (None, None));
}
#[test]
@@ -554,7 +487,6 @@ mod tests {
let empty = bounded.drain_expired_bounded(now + Duration::from_secs(4), 0);
assert!(empty.probes.is_empty());
assert!(empty.more_due);
assert_eq!(bounded.linkage(), (Some(1), Some(4)));
let mut actual = Vec::new();
loop {
@@ -573,7 +505,6 @@ mod tests {
assert_eq!(actual, expected);
assert_eq!(actual, vec![1, 2, 3, 4]);
assert_eq!(bounded.linkage(), (None, None));
}
#[test]
@@ -585,8 +516,6 @@ mod tests {
insert(&mut map, pending(7, now + Duration::from_secs(1)));
assert!(map.drain_expired_bounded(now, usize::MAX).probes.is_empty());
assert_eq!(map.linkage(), (Some(7), Some(7)));
assert_eq!(map.entry_links(7), Some((None, None)));
assert_eq!(
map.drain_expired_bounded(now + Duration::from_secs(1), usize::MAX)
.probes
@@ -595,7 +524,6 @@ mod tests {
.collect::<Vec<_>>(),
vec![7]
);
assert_eq!(map.linkage(), (None, None));
}
#[test]
@@ -603,13 +531,12 @@ mod tests {
let mut map = PendingMap::new(2);
let now = Instant::now();
map.preflight_insert(7).unwrap();
let capacity = map.capacity();
let capacity = map.map.capacity();
map.commit_insert(pending(7, now));
assert_eq!(map.capacity(), capacity);
assert_eq!(map.len(), 1);
assert_eq!(map.linkage(), (Some(7), Some(7)));
assert_eq!(map.map.capacity(), capacity);
assert_eq!(map.map.len(), 1);
}
#[test]
@@ -624,8 +551,6 @@ mod tests {
}
assert!(map.remove(2).is_some());
assert_eq!(map.entry_links(1), Some((None, Some(3))));
assert_eq!(map.entry_links(3), Some((Some(1), Some(4))));
let expired = map.drain_expired_bounded(now + Duration::from_secs(3), usize::MAX);
assert_eq!(
expired
@@ -635,9 +560,7 @@ mod tests {
.collect::<Vec<_>>(),
vec![1, 3]
);
assert_eq!(map.linkage(), (Some(4), Some(4)));
assert!(map.remove(4).is_some());
assert_eq!(map.linkage(), (None, None));
}
#[test]
@@ -660,7 +583,7 @@ mod tests {
map.insert(pending(1, now));
map.insert(pending(2, now));
assert_eq!(map.len(), 2);
assert_eq!(map.map.len(), 2);
assert!(map.remove(0).is_none());
assert!(map.remove(1).is_some());
assert!(map.remove(2).is_some());
@@ -674,8 +597,8 @@ mod tests {
for i in 0..20 {
map.insert(pending(i, now));
assert!(map.remove(i).is_some());
assert_eq!(map.len(), 0);
assert_eq!(map.insertion_order_len(), 0);
assert_eq!(map.map.len(), 0);
assert_eq!(map.insertion_order.len(), 0);
}
}
}
crates/irtt-client/src/receive.rs
+2 -40
View File
@@ -112,11 +112,11 @@ pub(crate) fn drain_tx_timestamps<S: std::os::fd::AsRawFd>(
linux::drain_tx_timestamps(socket, on_timestamp)
}
#[cfg(test)]
#[cfg(all(test, feature = "tokio"))]
mod tests {
use std::net::UdpSocket;
use crate::{metadata::ReceiveMeta, receive::recv_datagram, timing::ClientTimestamp};
use crate::metadata::ReceiveMeta;
fn connected_loopback_pair() -> (UdpSocket, UdpSocket) {
let a = UdpSocket::bind("127.0.0.1:0").unwrap();
@@ -126,44 +126,6 @@ mod tests {
(a, b)
}
#[test]
fn fallback_receive_returns_length() {
let (sender, receiver) = connected_loopback_pair();
sender.send(b"hello").unwrap();
let mut buf = [0_u8; 16];
let datagram = recv_datagram(&receiver, &mut buf).unwrap();
assert_eq!(datagram.len, 5);
assert_eq!(&buf[..datagram.len], b"hello");
}
#[test]
fn fallback_receive_returns_default_metadata() {
let (sender, receiver) = connected_loopback_pair();
sender.send(b"meta").unwrap();
let mut buf = [0_u8; 16];
let datagram = recv_datagram(&receiver, &mut buf).unwrap();
assert_eq!(datagram.meta, ReceiveMeta::default());
}
#[test]
fn fallback_receive_captures_timestamp_after_successful_receive() {
let (sender, receiver) = connected_loopback_pair();
sender.send(b"time").unwrap();
let before = ClientTimestamp::now();
let mut buf = [0_u8; 16];
let datagram = recv_datagram(&receiver, &mut buf).unwrap();
let after = ClientTimestamp::now();
assert!(datagram.received_at.mono >= before.mono);
assert!(datagram.received_at.mono <= after.mono);
}
#[cfg(feature = "tokio")]
#[test]
fn tokio_try_receive_preserves_fallback_metadata_and_would_block() {
let runtime = tokio::runtime::Builder::new_current_thread()
crates/irtt-client/src/receive/linux.rs
+4 -14
View File
@@ -21,8 +21,8 @@
//! session so later IDs cannot be misattributed. It performs a small bounded,
//! nonblocking read of
//! `MSG_ERRQUEUE` so the adapter can opportunistically collect timestamps
//! without ever waiting for one. The observed kernel TX wall time is retained
//! as metadata only; it is not consumed by any measurement in this change.
//! without ever waiting for one. A plausible kernel TX wall time is used only
//! for upstream one-way delay; RTT retains the userspace send instant.
use std::{
io::{self, IoSliceMut},
@@ -161,12 +161,12 @@ fn recv_datagram_fd(
buf: &mut [u8],
flags: MsgFlags,
) -> Result<ReceivedDatagram, io::Error> {
let mut control = ControlBuffer::new();
let mut control = ControlBuffer([0; CONTROL_LEN]);
let mut iov = [IoSliceMut::new(buf)];
// The socket is connected, so the source address is not needed; `()` skips
// copying it out.
let msg = recvmsg::<()>(socket_fd, &mut iov, Some(control.as_mut_slice()), flags)?;
let msg = recvmsg::<()>(socket_fd, &mut iov, Some(&mut control.0), flags)?;
let received_at = ClientTimestamp::now();
Ok(ReceivedDatagram {
@@ -216,16 +216,6 @@ fn system_time_from_timespec(timespec: TimeSpec) -> Option<SystemTime> {
#[repr(align(8))]
struct ControlBuffer([u8; CONTROL_LEN]);
impl ControlBuffer {
fn new() -> Self {
Self([0; CONTROL_LEN])
}
fn as_mut_slice(&mut self) -> &mut [u8] {
&mut self.0
}
}
#[cfg(test)]
mod tests {
use std::{
crates/irtt-client/src/receive/linux/error_queue.rs
+2 -11
View File
@@ -103,10 +103,7 @@ pub(crate) fn classify(cmsgs: impl Iterator<Item = ControlMessageOwned>) -> Erro
for cmsg in cmsgs {
match cmsg {
ControlMessageOwned::Ipv4RecvErr(err, _) => {
extended_error = Some((err.ee_errno, err.ee_origin, err.ee_info, err.ee_data));
}
ControlMessageOwned::Ipv6RecvErr(err, _) => {
ControlMessageOwned::Ipv4RecvErr(err, _) | ControlMessageOwned::Ipv6RecvErr(err, _) => {
extended_error = Some((err.ee_errno, err.ee_origin, err.ee_info, err.ee_data));
}
ControlMessageOwned::ScmTimestampsns(observed) => {
@@ -145,12 +142,6 @@ pub(crate) fn classify(cmsgs: impl Iterator<Item = ControlMessageOwned>) -> Erro
#[repr(align(8))]
struct ErrorQueueControlBuffer([u8; ERROR_QUEUE_CONTROL_LEN]);
impl ErrorQueueControlBuffer {
fn new() -> Self {
Self([0; ERROR_QUEUE_CONTROL_LEN])
}
}
/// Nonblocking drain of a single record from `fd`'s `MSG_ERRQUEUE`.
///
/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) — the
@@ -171,7 +162,7 @@ impl ErrorQueueControlBuffer {
pub(crate) fn try_recv_error_queue_record(fd: RawFd) -> io::Result<Option<ErrorQueueRecord>> {
let mut payload: [u8; 0] = [];
let mut iov = [IoSliceMut::new(&mut payload)];
let mut control = ErrorQueueControlBuffer::new();
let mut control = ErrorQueueControlBuffer([0; ERROR_QUEUE_CONTROL_LEN]);
match recvmsg::<()>(
fd,
crates/irtt-client/src/session/machine.rs
+23 -1186
View File
File diff suppressed because it is too large. Load diff
crates/irtt-client/src/socket.rs
+1 -43
View File
@@ -36,9 +36,6 @@ pub(crate) fn validate_open_timeouts(timeouts: &[Duration]) -> Result<(), Client
}
pub(crate) fn resolve_remote(config: &ClientConfig) -> Result<SocketAddr, ClientError> {
#[cfg(all(test, feature = "tokio"))]
SYNC_RESOLVER_CALLS.with(|calls| calls.set(calls.get() + 1));
let addr = normalize_server_addr(&config.server_addr);
let mut addrs = addr
.to_socket_addrs()
@@ -57,9 +54,6 @@ pub(crate) async fn resolve_remote_tokio(config: &ClientConfig) -> Result<Socket
.ok_or(ClientError::Resolve { addr });
}
#[cfg(all(test, feature = "tokio"))]
TOKIO_DNS_LOOKUPS.with(|calls| calls.set(calls.get() + 1));
let mut addrs = tokio::net::lookup_host(&addr)
.await
.map_err(|_| ClientError::Resolve { addr: addr.clone() })?;
@@ -73,20 +67,6 @@ fn address_family_allowed(config: &ClientConfig, remote: SocketAddr) -> bool {
&& (!config.socket_config.ipv6_only || remote.is_ipv6())
}
#[cfg(all(test, feature = "tokio"))]
std::thread_local! {
static SYNC_RESOLVER_CALLS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
static TOKIO_DNS_LOOKUPS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
#[cfg(all(test, feature = "tokio"))]
pub(crate) fn resolution_call_counts() -> (usize, usize) {
(
SYNC_RESOLVER_CALLS.with(std::cell::Cell::get),
TOKIO_DNS_LOOKUPS.with(std::cell::Cell::get),
)
}
pub(crate) fn normalize_server_addr(addr: &str) -> String {
if addr.parse::<SocketAddr>().is_ok() {
return addr.to_owned();
@@ -280,7 +260,7 @@ mod tests {
#[cfg(feature = "tokio")]
#[test]
fn tokio_resolver_filters_families_without_using_blocking_resolution() {
fn tokio_resolver_filters_address_families() {
let mut config = ClientConfig {
server_addr: "localhost:2112".to_owned(),
..ClientConfig::default()
@@ -288,13 +268,9 @@ mod tests {
config.socket_config.ipv4_only = true;
tokio_runtime().block_on(async {
let before = resolution_call_counts();
let remote = resolve_remote_tokio(&config).await.unwrap();
let after = resolution_call_counts();
assert!(remote.is_ipv4());
assert_eq!(after.0, before.0);
assert_eq!(after.1, before.1 + 1);
});
}
@@ -313,24 +289,6 @@ mod tests {
});
}
#[cfg(feature = "tokio")]
#[test]
fn tokio_resolver_literal_path_bypasses_all_name_resolution() {
let config = ClientConfig {
server_addr: "127.0.0.1:2112".to_owned(),
..ClientConfig::default()
};
tokio_runtime().block_on(async {
let before = resolution_call_counts();
let remote = resolve_remote_tokio(&config).await.unwrap();
let after = resolution_call_counts();
assert_eq!(remote, SocketAddr::from(([127, 0, 0, 1], 2112)));
assert_eq!(after, before);
});
}
#[cfg(feature = "tokio")]
#[test]
fn tokio_resolver_rejects_disallowed_literal_family() {
crates/irtt-client/src/socket_options.rs
+3 -357
View File
@@ -140,52 +140,15 @@ pub(crate) fn apply_ttl_to_socket(
ttl: u32,
) -> Result<(), ClientError> {
validate_ttl(ttl)?;
set_socket_ttl(socket, remote, ttl).map_err(|source| ClientError::SocketOption {
operation: "set TTL/hop limit",
remote,
source,
})
}
#[cfg(test)]
pub(crate) fn socket_traffic_class(
socket: &UdpSocket,
remote: SocketAddr,
) -> Result<u32, ClientError> {
get_socket_traffic_class(socket, remote).map_err(|source| ClientError::SocketOption {
operation: "read DSCP socket option",
remote,
source,
})
}
#[cfg(all(test, feature = "tokio"))]
pub(crate) fn tokio_socket_traffic_class(
socket: &tokio::net::UdpSocket,
remote: SocketAddr,
) -> Result<u32, ClientError> {
get_socket_traffic_class_ref(SockRef::from(socket), remote).map_err(|source| {
set_socket_ttl_ref(SockRef::from(socket), remote, ttl).map_err(|source| {
ClientError::SocketOption {
operation: "read DSCP socket option",
operation: "set TTL/hop limit",
remote,
source,
}
})
}
#[cfg(test)]
pub(crate) fn socket_ttl(socket: &UdpSocket, remote: SocketAddr) -> Result<u32, ClientError> {
get_socket_ttl(socket, remote).map_err(|source| ClientError::SocketOption {
operation: "read TTL/hop limit socket option",
remote,
source,
})
}
fn set_socket_ttl(socket: &UdpSocket, remote: SocketAddr, ttl: u32) -> io::Result<()> {
set_socket_ttl_ref(SockRef::from(socket), remote, ttl)
}
fn set_socket_ttl_ref(socket: SockRef<'_>, remote: SocketAddr, ttl: u32) -> io::Result<()> {
if remote.is_ipv4() {
socket.set_ttl_v4(ttl)
@@ -194,23 +157,13 @@ fn set_socket_ttl_ref(socket: SockRef<'_>, remote: SocketAddr, ttl: u32) -> io::
}
}
#[cfg(test)]
fn get_socket_ttl(socket: &UdpSocket, remote: SocketAddr) -> io::Result<u32> {
let socket = SockRef::from(socket);
if remote.is_ipv4() {
socket.ttl_v4()
} else {
socket.unicast_hops_v6()
}
}
fn set_socket_traffic_class(
socket: &UdpSocket,
remote: SocketAddr,
traffic_class: u32,
operation: &'static str,
) -> Result<(), ClientError> {
set_socket_traffic_class_io(socket, remote, traffic_class).map_err(|source| {
set_socket_traffic_class_ref(SockRef::from(socket), remote, traffic_class).map_err(|source| {
ClientError::SocketOption {
operation,
remote,
@@ -219,14 +172,6 @@ fn set_socket_traffic_class(
})
}
fn set_socket_traffic_class_io(
socket: &UdpSocket,
remote: SocketAddr,
traffic_class: u32,
) -> io::Result<()> {
set_socket_traffic_class_ref(SockRef::from(socket), remote, traffic_class)
}
#[cfg(feature = "tokio")]
fn set_tokio_socket_traffic_class(
socket: &tokio::net::UdpSocket,
@@ -335,302 +280,3 @@ fn unsupported_traffic_class(traffic_class: u32, feature: &'static str) -> io::R
))
}
}
#[cfg(all(
test,
not(any(
target_os = "fuchsia",
target_os = "redox",
target_os = "solaris",
target_os = "illumos",
target_os = "haiku",
))
))]
fn get_ipv4_traffic_class(socket: SockRef<'_>) -> io::Result<u32> {
socket.tos_v4()
}
#[cfg(all(
test,
any(
target_os = "fuchsia",
target_os = "redox",
target_os = "solaris",
target_os = "illumos",
target_os = "haiku",
)
))]
fn get_ipv4_traffic_class(_socket: SockRef<'_>) -> io::Result<u32> {
unsupported_readback("IPv4 DSCP socket option readback")
}
#[cfg(all(
test,
any(
target_os = "android",
target_os = "dragonfly",
target_os = "freebsd",
target_os = "fuchsia",
target_os = "linux",
target_os = "macos",
target_os = "netbsd",
target_os = "openbsd",
target_os = "cygwin",
)
))]
fn get_ipv6_traffic_class(socket: SockRef<'_>) -> io::Result<u32> {
socket.tclass_v6()
}
#[cfg(all(
test,
not(any(
target_os = "android",
target_os = "dragonfly",
target_os = "freebsd",
target_os = "fuchsia",
target_os = "linux",
target_os = "macos",
target_os = "netbsd",
target_os = "openbsd",
target_os = "cygwin",
))
))]
fn get_ipv6_traffic_class(_socket: SockRef<'_>) -> io::Result<u32> {
unsupported_readback("IPv6 DSCP socket option readback")
}
#[cfg(test)]
fn get_socket_traffic_class(socket: &UdpSocket, remote: SocketAddr) -> io::Result<u32> {
get_socket_traffic_class_ref(SockRef::from(socket), remote)
}
#[cfg(test)]
fn get_socket_traffic_class_ref(socket: SockRef<'_>, remote: SocketAddr) -> io::Result<u32> {
if remote.is_ipv4() {
get_ipv4_traffic_class(socket)
} else {
get_ipv6_traffic_class(socket)
}
}
#[cfg(all(
test,
any(
target_os = "fuchsia",
target_os = "redox",
target_os = "solaris",
target_os = "illumos",
target_os = "haiku",
not(any(
target_os = "android",
target_os = "dragonfly",
target_os = "freebsd",
target_os = "fuchsia",
target_os = "linux",
target_os = "macos",
target_os = "netbsd",
target_os = "openbsd",
target_os = "cygwin",
)),
)
))]
fn unsupported_readback(feature: &'static str) -> io::Result<u32> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
format!("{feature} is unsupported on this target"),
))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn converts_dscp_codepoint_to_traffic_class_byte() {
for (dscp, traffic_class) in [(0, 0), (1, 4), (46, 184), (63, 252)] {
let value = dscp_codepoint_to_traffic_class(dscp).unwrap();
assert_eq!(value, traffic_class);
assert_eq!(value & 0b11, 0);
}
}
#[test]
fn rejects_out_of_range_dscp_codepoint() {
assert!(matches!(
dscp_codepoint_to_traffic_class(64),
Err(ClientError::InvalidConfig { .. })
));
}
#[test]
fn validates_ttl_range() {
assert!(validate_ttl(1).is_ok());
assert!(validate_ttl(64).is_ok());
assert!(validate_ttl(255).is_ok());
assert!(matches!(
validate_ttl(0),
Err(ClientError::InvalidConfig { .. })
));
assert!(matches!(
validate_ttl(256),
Err(ClientError::InvalidConfig { .. })
));
}
#[test]
fn ipv4_socket_option_sets_ttl() {
let remote = SocketAddr::from(([127, 0, 0, 1], 9));
let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
socket.connect(remote).unwrap();
apply_ttl_to_socket(&socket, remote, 64).unwrap();
assert_eq!(socket_ttl(&socket, remote).unwrap(), 64);
apply_ttl_to_socket(&socket, remote, 1).unwrap();
assert_eq!(socket_ttl(&socket, remote).unwrap(), 1);
}
#[test]
fn ipv6_socket_option_sets_unicast_hop_limit() {
let remote = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], 9));
let Some(socket) = bind_connected_ipv6_loopback(remote) else {
return;
};
apply_ttl_to_socket(&socket, remote, 64).unwrap();
let ttl = match socket_ttl(&socket, remote) {
Ok(ttl) => ttl,
Err(error) if is_unsupported_socket_readback(&error) => {
eprintln!("skipping IPv6 hop-limit readback test: {error}");
return;
}
Err(error) => panic!("{error}"),
};
assert_eq!(ttl, 64);
apply_ttl_to_socket(&socket, remote, 1).unwrap();
assert_eq!(socket_ttl(&socket, remote).unwrap(), 1);
}
#[test]
#[cfg(not(any(
target_os = "fuchsia",
target_os = "redox",
target_os = "solaris",
target_os = "illumos",
target_os = "haiku",
)))]
fn ipv4_socket_option_sets_and_clears_traffic_class() {
let remote = SocketAddr::from(([127, 0, 0, 1], 9));
let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
socket.connect(remote).unwrap();
// 184 is the raw wire TOS/Traffic Class byte for DSCP codepoint 46
// (EF); applying it directly must not shift it again.
apply_traffic_class_to_socket(&socket, remote, 184).unwrap();
assert_eq!(socket_traffic_class(&socket, remote).unwrap(), 184);
clear_dscp_on_socket(&socket, remote).unwrap();
assert_eq!(socket_traffic_class(&socket, remote).unwrap(), 0);
}
#[cfg(all(
feature = "tokio",
not(any(
target_os = "fuchsia",
target_os = "redox",
target_os = "solaris",
target_os = "illumos",
target_os = "haiku",
))
))]
#[test]
fn tokio_ipv4_socket_option_sets_and_clears_traffic_class() {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_io()
.build()
.unwrap();
runtime.block_on(async {
let remote = SocketAddr::from(([127, 0, 0, 1], 9));
let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
socket.connect(remote).unwrap();
socket.set_nonblocking(true).unwrap();
let socket = tokio::net::UdpSocket::from_std(socket).unwrap();
apply_traffic_class_to_tokio_socket(&socket, remote, 184).unwrap();
assert_eq!(tokio_socket_traffic_class(&socket, remote).unwrap(), 184);
clear_dscp_on_tokio_socket(&socket, remote).unwrap();
assert_eq!(tokio_socket_traffic_class(&socket, remote).unwrap(), 0);
});
}
#[test]
#[cfg(any(
target_os = "android",
target_os = "dragonfly",
target_os = "freebsd",
target_os = "fuchsia",
target_os = "linux",
target_os = "macos",
target_os = "netbsd",
target_os = "openbsd",
target_os = "cygwin",
))]
fn ipv6_socket_option_sets_and_clears_traffic_class() {
let remote = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], 9));
let Some(socket) = bind_connected_ipv6_loopback(remote) else {
return;
};
apply_traffic_class_to_socket(&socket, remote, 184).unwrap();
let traffic_class = match socket_traffic_class(&socket, remote) {
Ok(traffic_class) => traffic_class,
Err(error) if is_unsupported_socket_readback(&error) => {
eprintln!("skipping IPv6 traffic-class readback test: {error}");
return;
}
Err(error) => panic!("{error}"),
};
assert_eq!(traffic_class, 184);
clear_dscp_on_socket(&socket, remote).unwrap();
assert_eq!(socket_traffic_class(&socket, remote).unwrap(), 0);
}
fn bind_connected_ipv6_loopback(remote: SocketAddr) -> Option<UdpSocket> {
let socket = match UdpSocket::bind(SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], 0))) {
Ok(socket) => socket,
Err(error) if is_unavailable_ipv6_loopback(&error) => {
eprintln!("skipping IPv6 socket option test: IPv6 loopback unavailable: {error}");
return None;
}
Err(error) => panic!("{error}"),
};
match socket.connect(remote) {
Ok(()) => Some(socket),
Err(error) if is_unavailable_ipv6_loopback(&error) => {
eprintln!("skipping IPv6 socket option test: IPv6 loopback unavailable: {error}");
None
}
Err(error) => panic!("{error}"),
}
}
fn is_unavailable_ipv6_loopback(error: &io::Error) -> bool {
matches!(
error.kind(),
io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
)
}
fn is_unsupported_socket_readback(error: &ClientError) -> bool {
matches!(
error,
ClientError::SocketOption { source, .. }
if matches!(source.kind(), io::ErrorKind::Unsupported)
)
}
}