refactor(client): make configuration reusable across endpoints
This commit is contained in:
43 files changed
+1012
-723
No files matched your search
@@ -19,7 +19,7 @@ use crate::{
|
||||
recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest,
|
||||
PreparedProbe, SessionMachine, TimeoutBatch, MAX_OPEN_PACKET_SIZE,
|
||||
},
|
||||
socket::{connect_tokio_udp_socket, resolve_remote_tokio, validate_open_timeouts},
|
||||
socket::{connect_tokio_udp_socket, resolve_remote_tokio},
|
||||
socket_options::{apply_traffic_class_to_tokio_socket, clear_dscp_on_tokio_socket},
|
||||
timing::ClientTimestamp,
|
||||
};
|
||||
@@ -170,7 +170,7 @@ impl AsyncOpenState {
|
||||
/// use irtt_client::{AsyncClient, ClientConfig};
|
||||
///
|
||||
/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
/// let mut client = AsyncClient::connect(ClientConfig::default()).await?;
|
||||
/// let mut client = AsyncClient::connect("127.0.0.1:2112", ClientConfig::default()).await?;
|
||||
/// let outcome = client.open().await?;
|
||||
/// println!("opened: {outcome:?}");
|
||||
/// let sent = client.send_probe().await?;
|
||||
@@ -199,19 +199,26 @@ pub struct AsyncClient {
|
||||
}
|
||||
|
||||
impl AsyncClient {
|
||||
/// Resolve the configured server and construct one connected Tokio UDP
|
||||
/// Resolve the endpoint and construct one connected Tokio UDP
|
||||
/// socket.
|
||||
///
|
||||
/// Polling this future without a current Tokio runtime returns
|
||||
/// [`ClientError::NoTokioRuntime`]. A runtime without enabled I/O or time
|
||||
/// drivers is outside this type's runtime contract.
|
||||
pub async fn connect(config: ClientConfig) -> Result<Self, ClientError> {
|
||||
///
|
||||
/// The endpoint accepts a name or address; an omitted port defaults to 2112.
|
||||
/// IPv6 literals may be bracketed or unbracketed with the default port.
|
||||
/// The reusable config contains no endpoint identity.
|
||||
pub async fn connect(
|
||||
endpoint: impl AsRef<str>,
|
||||
config: ClientConfig,
|
||||
) -> Result<Self, ClientError> {
|
||||
tokio::runtime::Handle::try_current().map_err(|_| ClientError::NoTokioRuntime)?;
|
||||
validate_open_timeouts(&config.open_timeouts)?;
|
||||
let remote = resolve_remote_tokio(&config).await?;
|
||||
config.open.validate()?;
|
||||
let remote = resolve_remote_tokio(endpoint.as_ref(), config.address_family).await?;
|
||||
let machine = SessionMachine::new(config.clone(), remote)?;
|
||||
let prepared_open = machine.prepare_open_request()?;
|
||||
let socket = connect_tokio_udp_socket(&config.socket_config, remote)?;
|
||||
let socket = connect_tokio_udp_socket(&config.socket, remote, config.address_family)?;
|
||||
|
||||
Ok(Self {
|
||||
socket,
|
||||
@@ -307,7 +314,7 @@ impl AsyncClient {
|
||||
return self.poll_open_cleanup(state, cx);
|
||||
}
|
||||
|
||||
let attempt_count = self.machine.config().open_timeouts.len();
|
||||
let attempt_count = self.machine.config().open.timeouts.len();
|
||||
if state.attempt >= attempt_count {
|
||||
return Poll::Ready(Err(ClientError::OpenTimeout));
|
||||
}
|
||||
@@ -316,7 +323,7 @@ impl AsyncClient {
|
||||
return Poll::Ready(Err(ClientError::OpenTimeout));
|
||||
}
|
||||
if state.deadline.is_none() {
|
||||
let timeout = self.machine.config().open_timeouts[state.attempt];
|
||||
let timeout = self.machine.config().open.timeouts[state.attempt];
|
||||
if let Err(error) = state.start_attempt(timeout) {
|
||||
return Poll::Ready(Err(error));
|
||||
}
|
||||
|
||||
@@ -13,7 +13,7 @@ use crate::{
|
||||
recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent,
|
||||
SessionMachine, MAX_OPEN_PACKET_SIZE,
|
||||
},
|
||||
socket::{connect_udp_socket, resolve_remote, validate_open_timeouts},
|
||||
socket::{connect_udp_socket, resolve_remote},
|
||||
socket_options::{apply_traffic_class_to_socket, clear_dscp_on_socket},
|
||||
timing::ClientTimestamp,
|
||||
};
|
||||
@@ -50,7 +50,7 @@ struct PreparedClientOpenFailure {
|
||||
/// use irtt_client::{Client, ClientConfig};
|
||||
///
|
||||
/// # fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
/// let mut client = Client::connect(ClientConfig::default())?;
|
||||
/// let mut client = Client::connect("127.0.0.1:2112", ClientConfig::default())?;
|
||||
/// let outcome = client.open()?;
|
||||
/// println!("opened: {outcome:?}");
|
||||
/// let sent = client.send_probe()?;
|
||||
@@ -70,6 +70,7 @@ pub struct Client {
|
||||
socket: UdpSocket,
|
||||
remote: SocketAddr,
|
||||
recv_buffer: Vec<u8>,
|
||||
recv_timeout: Option<Duration>,
|
||||
applied_traffic_class: Option<u8>,
|
||||
/// Whether the socket's `SO_TIMESTAMPING` flags carry `TX_SOFTWARE` +
|
||||
/// `OPT_ID` + `OPT_TSONLY`. Always `false` off Linux, without the
|
||||
@@ -81,32 +82,53 @@ pub struct Client {
|
||||
}
|
||||
|
||||
impl Client {
|
||||
/// Resolve the configured server and create a connected UDP socket.
|
||||
/// Resolve the endpoint and create a connected UDP socket.
|
||||
///
|
||||
/// This validates local configuration and prepares the open request, but it
|
||||
/// does not contact the server. Call [`open`](Self::open) to perform the
|
||||
/// IRTT open exchange.
|
||||
pub fn connect(config: ClientConfig) -> Result<Self, ClientError> {
|
||||
validate_open_timeouts(&config.open_timeouts)?;
|
||||
let remote = resolve_remote(&config)?;
|
||||
///
|
||||
/// The endpoint accepts a name or address; an omitted port defaults to 2112.
|
||||
/// IPv6 literals may be bracketed or unbracketed with the default port.
|
||||
/// The reusable config contains no endpoint identity.
|
||||
pub fn connect(endpoint: impl AsRef<str>, config: ClientConfig) -> Result<Self, ClientError> {
|
||||
config.open.validate()?;
|
||||
let remote = resolve_remote(endpoint.as_ref(), config.address_family)?;
|
||||
let runtime = SessionMachine::new(config.clone(), remote)?;
|
||||
let socket = connect_udp_socket(&config.socket_config, remote)?;
|
||||
let socket = connect_udp_socket(&config.socket, remote, config.address_family)?;
|
||||
|
||||
Ok(Self {
|
||||
runtime,
|
||||
socket,
|
||||
remote,
|
||||
recv_buffer: vec![0_u8; recv_buffer_size(false, None)?],
|
||||
recv_timeout: None,
|
||||
applied_traffic_class: None,
|
||||
tx_timestamping_enabled: false,
|
||||
})
|
||||
}
|
||||
|
||||
/// Set the blocking receive timeout used outside open attempts.
|
||||
///
|
||||
/// `None` (the default) lets `recv_once()` block indefinitely. A nonzero
|
||||
/// timeout bounds one logical receive, including retries after EINTR.
|
||||
/// `recv_available()` retains its bounded drain behavior. Open attempts
|
||||
/// use only [`OpenPolicy::timeouts`](crate::OpenPolicy::timeouts) and
|
||||
/// restore this setting on success or failure.
|
||||
///
|
||||
/// A zero duration is rejected by the socket; the previous setting is
|
||||
/// retained when applying the new timeout fails.
|
||||
pub fn set_recv_timeout(&mut self, timeout: Option<Duration>) -> Result<(), ClientError> {
|
||||
self.socket.set_read_timeout(timeout)?;
|
||||
self.recv_timeout = timeout;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Perform the IRTT open exchange.
|
||||
///
|
||||
/// On success, returns the negotiated open outcome and transitions the
|
||||
/// client into either an open probe session or completed no-test state.
|
||||
/// Open attempts use [`ClientConfig::open_timeouts`]. Malformed or
|
||||
/// Open attempts use [`OpenPolicy::timeouts`](crate::OpenPolicy::timeouts). Malformed or
|
||||
/// unrelated datagrams are ignored until the current attempt's absolute
|
||||
/// deadline, so one attempt may consume several datagrams without
|
||||
/// retransmitting. Silence or ignored traffic eventually produces
|
||||
@@ -121,9 +143,7 @@ impl Client {
|
||||
pub fn open(&mut self) -> Result<OpenOutcome, ClientError> {
|
||||
let result = self.open_transaction();
|
||||
if result.is_err() {
|
||||
let _ = self
|
||||
.socket
|
||||
.set_read_timeout(self.runtime.config().socket_config.recv_timeout);
|
||||
let _ = self.socket.set_read_timeout(self.recv_timeout);
|
||||
}
|
||||
result
|
||||
}
|
||||
@@ -225,8 +245,7 @@ impl Client {
|
||||
/// Requires an open session; lifecycle errors are returned before socket I/O.
|
||||
pub fn recv_once(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
|
||||
self.runtime.ensure_open()?;
|
||||
self.socket
|
||||
.set_read_timeout(self.runtime.config().socket_config.recv_timeout)?;
|
||||
self.socket.set_read_timeout(self.recv_timeout)?;
|
||||
self.recv_once_inner()
|
||||
}
|
||||
|
||||
@@ -267,7 +286,7 @@ impl Client {
|
||||
fn recv_datagram_retrying_interrupted(
|
||||
&mut self,
|
||||
) -> Result<Option<ReceivedDatagram>, ClientError> {
|
||||
let configured_timeout = self.runtime.config().socket_config.recv_timeout;
|
||||
let configured_timeout = self.recv_timeout;
|
||||
let deadline = match configured_timeout {
|
||||
Some(timeout) => Some(
|
||||
Instant::now()
|
||||
@@ -322,8 +341,7 @@ impl Client {
|
||||
/// Requires an open session; lifecycle errors are returned before socket I/O.
|
||||
pub fn recv_available(&mut self, budget: RecvBudget) -> Result<Vec<ClientEvent>, ClientError> {
|
||||
self.runtime.ensure_open()?;
|
||||
self.socket
|
||||
.set_read_timeout(self.runtime.config().socket_config.recv_timeout)?;
|
||||
self.socket.set_read_timeout(self.recv_timeout)?;
|
||||
let mut all_events = Vec::new();
|
||||
for _ in 0..budget.max_packets {
|
||||
let events = self.recv_once_inner()?;
|
||||
@@ -383,10 +401,10 @@ impl Client {
|
||||
fn open_transaction(&mut self) -> Result<OpenOutcome, ClientError> {
|
||||
let request = self.runtime.prepare_open_request()?;
|
||||
let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE];
|
||||
let attempt_count = self.runtime.config().open_timeouts.len();
|
||||
let attempt_count = self.runtime.config().open.timeouts.len();
|
||||
|
||||
for attempt in 0..attempt_count {
|
||||
let timeout = self.runtime.config().open_timeouts[attempt];
|
||||
let timeout = self.runtime.config().open.timeouts[attempt];
|
||||
let deadline = Instant::now()
|
||||
.checked_add(timeout)
|
||||
.ok_or(ClientError::DurationOverflow)?;
|
||||
@@ -470,7 +488,7 @@ impl Client {
|
||||
recv_buffer_len: None,
|
||||
negotiated_traffic_class: None,
|
||||
previous_traffic_class: self.applied_traffic_class,
|
||||
post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout,
|
||||
post_open_recv_timeout: self.recv_timeout,
|
||||
});
|
||||
};
|
||||
let recv_buffer_len = match recv_buffer_size(self.runtime.has_hmac(), Some(negotiated)) {
|
||||
@@ -495,7 +513,7 @@ impl Client {
|
||||
recv_buffer_len: Some(recv_buffer_len),
|
||||
negotiated_traffic_class: Some(negotiated_traffic_class),
|
||||
previous_traffic_class: self.applied_traffic_class,
|
||||
post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout,
|
||||
post_open_recv_timeout: self.recv_timeout,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
use std::{net::SocketAddr, time::Duration};
|
||||
|
||||
use crate::Authentication;
|
||||
use crate::{Authentication, ClientError};
|
||||
|
||||
/// Protocol compatibility bound for a requested `server_fill` value, in UTF-8
|
||||
/// bytes.
|
||||
///
|
||||
/// This is the maximum encoded server-fill string accepted by this client and
|
||||
/// protocol decoder. [`ClientConfig::server_fill`] enforces the same bound
|
||||
/// protocol decoder. [`SessionRequest::server_fill`] enforces the same bound
|
||||
/// before opening a session.
|
||||
pub use irtt_proto::MAX_SERVER_FILL_BYTES;
|
||||
use irtt_proto::{Clock, ReceivedStats, StampAt};
|
||||
@@ -28,7 +28,7 @@ pub const MAX_TTL: u32 = 255;
|
||||
/// Largest UDP payload length accepted by client configuration, in bytes.
|
||||
///
|
||||
/// This is the maximum UDP payload size excluding IP and UDP headers. It caps
|
||||
/// [`ClientConfig::length`] before protocol packets are encoded or sent.
|
||||
/// [`SessionRequest::length`] before protocol packets are encoded or sent.
|
||||
pub const MAX_UDP_PAYLOAD_LENGTH: u32 = 65_507;
|
||||
pub(crate) const DEFAULT_DURATION: Duration = Duration::from_secs(3);
|
||||
pub(crate) const DEFAULT_INTERVAL: Duration = Duration::from_secs(1);
|
||||
@@ -58,29 +58,82 @@ pub(crate) const DEFAULT_MAX_PENDING: usize = 4096;
|
||||
/// ```
|
||||
/// use std::time::Duration;
|
||||
///
|
||||
/// use irtt_client::{ClientConfig, NegotiationPolicy};
|
||||
/// use irtt_client::{ClientConfig, NegotiationPolicy, OpenPolicy, SessionRequest};
|
||||
///
|
||||
/// let config = ClientConfig {
|
||||
/// server_addr: "example.net:2112".into(),
|
||||
/// duration: Some(Duration::from_secs(10)),
|
||||
/// interval: Duration::from_millis(250),
|
||||
/// length: 1200,
|
||||
/// negotiation_policy: NegotiationPolicy::Loose,
|
||||
/// request: SessionRequest {
|
||||
/// duration: Some(Duration::from_secs(10)),
|
||||
/// interval: Duration::from_millis(250),
|
||||
/// length: 1200,
|
||||
/// ..SessionRequest::default()
|
||||
/// },
|
||||
/// open: OpenPolicy {
|
||||
/// negotiation: NegotiationPolicy::Loose,
|
||||
/// ..OpenPolicy::default()
|
||||
/// },
|
||||
/// ..ClientConfig::default()
|
||||
/// };
|
||||
///
|
||||
/// assert_eq!(config.duration, Some(Duration::from_secs(10)));
|
||||
/// assert_eq!(config.interval, Duration::from_millis(250));
|
||||
/// assert_eq!(config.length, 1200);
|
||||
/// assert_eq!(config.request.duration, Some(Duration::from_secs(10)));
|
||||
/// assert_eq!(config.request.interval, Duration::from_millis(250));
|
||||
/// assert_eq!(config.request.length, 1200);
|
||||
/// ```
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct ClientConfig {
|
||||
/// Remote server name or address.
|
||||
/// Remote address families allowed during resolution and socket creation.
|
||||
pub address_family: AddressFamily,
|
||||
/// Local UDP socket properties.
|
||||
pub socket: SocketConfig,
|
||||
/// Protocol and session values requested from the server.
|
||||
pub request: SessionRequest,
|
||||
/// Open retry timeouts and negotiation policy.
|
||||
pub open: OpenPolicy,
|
||||
/// Concrete authentication for open, echo, and close packets.
|
||||
///
|
||||
/// If no port is present, the default IRTT port 2112 is used. IPv6
|
||||
/// literals may be supplied either bracketed or unbracketed when the
|
||||
/// default port should be used.
|
||||
pub server_addr: String,
|
||||
/// HMAC peers must use the same key. An empty HMAC key remains authenticated.
|
||||
pub auth: Authentication,
|
||||
/// Time after sending a probe before the client reports it as lost.
|
||||
///
|
||||
/// This timeout is local client behavior; it is not negotiated with the
|
||||
/// server. It must be greater than zero.
|
||||
pub probe_timeout: Duration,
|
||||
/// Maximum number of probes tracked as pending/timed-out/completed.
|
||||
///
|
||||
/// This bounds memory used for reply classification and must be greater
|
||||
/// than zero. A very small value can reject sends when replies are still
|
||||
/// outstanding.
|
||||
pub max_pending_probes: usize,
|
||||
}
|
||||
|
||||
impl Default for ClientConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
address_family: AddressFamily::default(),
|
||||
socket: SocketConfig::default(),
|
||||
request: SessionRequest::default(),
|
||||
open: OpenPolicy::default(),
|
||||
auth: Authentication::Unauthenticated,
|
||||
probe_timeout: DEFAULT_PROBE_TIMEOUT,
|
||||
max_pending_probes: DEFAULT_MAX_PENDING,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Allowed remote address families. IPv6-only selection also sets IPV6_V6ONLY.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
|
||||
pub enum AddressFamily {
|
||||
/// Accept IPv4 and IPv6 addresses.
|
||||
#[default]
|
||||
Any,
|
||||
/// Accept only IPv4 addresses.
|
||||
Ipv4,
|
||||
/// Accept only IPv6 addresses and use IPv6-only sockets.
|
||||
Ipv6,
|
||||
}
|
||||
|
||||
/// Requested IRTT protocol and session semantics, reusable across endpoints.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct SessionRequest {
|
||||
/// Requested run duration.
|
||||
///
|
||||
/// `Some(duration)` requests a finite test and must be greater than zero.
|
||||
@@ -120,46 +173,20 @@ pub struct ClientConfig {
|
||||
/// This is the six-bit DSCP value, not the full traffic-class byte, and
|
||||
/// must be less than or equal to [`MAX_DSCP_CODEPOINT`].
|
||||
pub dscp: u8,
|
||||
/// Concrete authentication for open, echo, and close packets.
|
||||
///
|
||||
/// HMAC peers must use the same key. An empty HMAC key remains authenticated.
|
||||
pub auth: Authentication,
|
||||
/// Optional server payload fill request.
|
||||
///
|
||||
/// `None` leaves server fill behavior unspecified. `Some(value)` requests
|
||||
/// a non-empty server fill mode/value and must not exceed
|
||||
/// [`MAX_SERVER_FILL_BYTES`] bytes when UTF-8 encoded.
|
||||
pub server_fill: Option<String>,
|
||||
/// Per-attempt receive timeouts used while opening the session.
|
||||
///
|
||||
/// The client sends an open request for each entry until a valid open reply
|
||||
/// is received. The list must not be empty, and each timeout must be at
|
||||
/// least 200 ms.
|
||||
pub open_timeouts: Vec<Duration>,
|
||||
/// Whether opening the session should start a probe test or perform a
|
||||
/// negotiation-only no-test exchange.
|
||||
pub run_mode: RunMode,
|
||||
/// Policy for server changes to negotiable protocol parameters.
|
||||
pub negotiation_policy: NegotiationPolicy,
|
||||
/// Local UDP socket configuration.
|
||||
pub socket_config: SocketConfig,
|
||||
/// Time after sending a probe before the client reports it as lost.
|
||||
///
|
||||
/// This timeout is local client behavior; it is not negotiated with the
|
||||
/// server. It must be greater than zero.
|
||||
pub probe_timeout: Duration,
|
||||
/// Maximum number of probes tracked as pending/timed-out/completed.
|
||||
///
|
||||
/// This bounds memory used for reply classification and must be greater
|
||||
/// than zero. A very small value can reject sends when replies are still
|
||||
/// outstanding.
|
||||
pub max_pending_probes: usize,
|
||||
}
|
||||
|
||||
impl Default for ClientConfig {
|
||||
impl Default for SessionRequest {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
server_addr: format!("127.0.0.1:{DEFAULT_PORT}"),
|
||||
duration: Some(DEFAULT_DURATION),
|
||||
interval: DEFAULT_INTERVAL,
|
||||
length: 0,
|
||||
@@ -167,23 +194,55 @@ impl Default for ClientConfig {
|
||||
stamp_at: StampAt::Both,
|
||||
clock: Clock::Both,
|
||||
dscp: 0,
|
||||
auth: Authentication::Unauthenticated,
|
||||
server_fill: None,
|
||||
open_timeouts: DEFAULT_OPEN_TIMEOUTS.to_vec(),
|
||||
run_mode: RunMode::Normal,
|
||||
negotiation_policy: NegotiationPolicy::Strict,
|
||||
socket_config: SocketConfig::default(),
|
||||
probe_timeout: DEFAULT_PROBE_TIMEOUT,
|
||||
max_pending_probes: DEFAULT_MAX_PENDING,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Opening behavior shared by blocking, async, and managed clients.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct OpenPolicy {
|
||||
/// Per-attempt receive timeouts used while opening the session.
|
||||
///
|
||||
/// The client sends an open request for each entry until a valid open reply
|
||||
/// is received. The list must not be empty, and each timeout must be at
|
||||
/// least 200 ms.
|
||||
pub timeouts: Vec<Duration>,
|
||||
/// Policy for server changes to negotiable protocol parameters.
|
||||
pub negotiation: NegotiationPolicy,
|
||||
}
|
||||
|
||||
impl OpenPolicy {
|
||||
pub(crate) fn validate(&self) -> Result<(), ClientError> {
|
||||
if self.timeouts.is_empty() {
|
||||
return Err(ClientError::NoOpenTimeouts);
|
||||
}
|
||||
for timeout in &self.timeouts {
|
||||
if *timeout < MIN_OPEN_TIMEOUT {
|
||||
return Err(ClientError::OpenTimeoutTooSmall {
|
||||
timeout: *timeout,
|
||||
minimum: MIN_OPEN_TIMEOUT,
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for OpenPolicy {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
timeouts: DEFAULT_OPEN_TIMEOUTS.to_vec(),
|
||||
negotiation: NegotiationPolicy::Strict,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Local UDP socket options used by [`ClientConfig`].
|
||||
///
|
||||
/// These settings affect how the client binds, resolves, and receives from the
|
||||
/// socket. They do not change the IRTT protocol parameters negotiated with the
|
||||
/// server.
|
||||
/// These settings affect how the client binds and sends from the socket. They
|
||||
/// do not change the IRTT protocol parameters negotiated with the server.
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
@@ -192,20 +251,20 @@ impl Default for ClientConfig {
|
||||
/// ```
|
||||
/// use std::net::{Ipv4Addr, SocketAddr};
|
||||
///
|
||||
/// use irtt_client::{ClientConfig, SocketConfig};
|
||||
/// use irtt_client::{AddressFamily, ClientConfig, SocketConfig};
|
||||
///
|
||||
/// let config = ClientConfig {
|
||||
/// socket_config: SocketConfig {
|
||||
/// address_family: AddressFamily::Ipv4,
|
||||
/// socket: SocketConfig {
|
||||
/// bind_addr: Some(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))),
|
||||
/// ttl: Some(32),
|
||||
/// ipv4_only: true,
|
||||
/// ..SocketConfig::default()
|
||||
/// },
|
||||
/// ..ClientConfig::default()
|
||||
/// };
|
||||
///
|
||||
/// assert_eq!(config.socket_config.ttl, Some(32));
|
||||
/// assert!(config.socket_config.ipv4_only);
|
||||
/// assert_eq!(config.socket.ttl, Some(32));
|
||||
/// assert_eq!(config.address_family, AddressFamily::Ipv4);
|
||||
/// ```
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Default)]
|
||||
pub struct SocketConfig {
|
||||
@@ -229,17 +288,6 @@ pub struct SocketConfig {
|
||||
/// platform socket option range; [`MAX_TTL`] is the public configuration
|
||||
/// bound used by this crate.
|
||||
pub ttl: Option<u32>,
|
||||
/// Restrict name resolution to IPv4 addresses.
|
||||
pub ipv4_only: bool,
|
||||
/// Restrict name resolution to IPv6 addresses and set IPV6_V6ONLY for IPv6
|
||||
/// sockets.
|
||||
pub ipv6_only: bool,
|
||||
/// Socket read timeout used after the session is open.
|
||||
///
|
||||
/// `None` leaves reads blocking for APIs that perform a single receive.
|
||||
/// Managed sessions may replace `None` or long timeouts with a short
|
||||
/// timeout so cooperative cancellation can be observed promptly.
|
||||
pub recv_timeout: Option<Duration>,
|
||||
}
|
||||
|
||||
/// How strictly to handle server-side negotiation restrictions.
|
||||
|
||||
@@ -53,8 +53,9 @@ pub use async_client::AsyncClient;
|
||||
pub use auth::{Authentication, HmacKey};
|
||||
pub use client::Client;
|
||||
pub use config::{
|
||||
ClientConfig, NegotiationPolicy, RecvBudget, RunMode, SocketConfig, MAX_DSCP_CODEPOINT,
|
||||
MAX_SERVER_FILL_BYTES, MAX_TTL, MAX_UDP_PAYLOAD_LENGTH,
|
||||
AddressFamily, ClientConfig, NegotiationPolicy, OpenPolicy, RecvBudget, RunMode,
|
||||
SessionRequest, SocketConfig, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_TTL,
|
||||
MAX_UDP_PAYLOAD_LENGTH,
|
||||
};
|
||||
pub use error::ClientError;
|
||||
pub use event::{
|
||||
|
||||
@@ -85,6 +85,8 @@ pub enum ManagedCompletionPolicy {
|
||||
/// Shared configuration for one managed task.
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
pub struct ManagedClientConfig {
|
||||
/// Reusable session/socket configuration shared across target generations.
|
||||
/// Each target supplies its endpoint and resolves its authentication policy.
|
||||
pub client: ClientConfig,
|
||||
pub pacing: ManagedPacing,
|
||||
pub completion: ManagedCompletionPolicy,
|
||||
|
||||
@@ -20,7 +20,6 @@ use tokio::{
|
||||
use crate::{
|
||||
async_client::{AsyncClient, AsyncOpenState},
|
||||
session::machine::SessionMachine,
|
||||
socket::validate_open_timeouts,
|
||||
socket_options::validate_ttl,
|
||||
ClientConfig, ClientError, ClientEvent, OpenOutcome,
|
||||
};
|
||||
@@ -62,7 +61,10 @@ const MAX_BROADCAST_CHANNEL_CAPACITY: usize = usize::MAX >> 1;
|
||||
/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
/// let config = ManagedClientConfig {
|
||||
/// client: ClientConfig {
|
||||
/// duration: Some(std::time::Duration::from_secs(10)),
|
||||
/// request: irtt_client::SessionRequest {
|
||||
/// duration: Some(std::time::Duration::from_secs(10)),
|
||||
/// ..irtt_client::SessionRequest::default()
|
||||
/// },
|
||||
/// ..ClientConfig::default()
|
||||
/// },
|
||||
/// ..ManagedClientConfig::default()
|
||||
@@ -944,7 +946,6 @@ impl ManagedClientTask {
|
||||
return Err(ManagedCommandApplyError::DuplicateTargetId { id: target.id });
|
||||
}
|
||||
let mut client_config = self.config.client.clone();
|
||||
client_config.server_addr.clone_from(&target.server_addr);
|
||||
client_config.auth = target.auth.resolve(&self.config.client.auth);
|
||||
validate_target_config(&client_config).map_err(|source| {
|
||||
ManagedCommandApplyError::InvalidTarget {
|
||||
@@ -1060,7 +1061,8 @@ impl ManagedClientTask {
|
||||
}
|
||||
|
||||
fn start_connecting(&mut self, index: usize, config: ClientConfig) {
|
||||
let future = Box::pin(AsyncClient::connect(config));
|
||||
let endpoint = Arc::clone(&self.targets[index].server_addr);
|
||||
let future = Box::pin(async move { AsyncClient::connect(endpoint, config).await });
|
||||
self.install_target_state(index, TargetState::Connecting { future });
|
||||
}
|
||||
|
||||
@@ -2248,7 +2250,6 @@ fn build_task(
|
||||
.checked_add(1)
|
||||
.ok_or(ManagedConfigError::GenerationExhausted)?;
|
||||
let mut client_config = config.client.clone();
|
||||
client_config.server_addr.clone_from(&target.server_addr);
|
||||
client_config.auth = target.auth.resolve(&config.client.auth);
|
||||
validate_target_config(&client_config).map_err(|source| {
|
||||
ManagedConfigError::InvalidTarget {
|
||||
@@ -2361,14 +2362,9 @@ fn build_task(
|
||||
}
|
||||
|
||||
fn validate_target_config(config: &ClientConfig) -> Result<(), ClientError> {
|
||||
validate_open_timeouts(&config.open_timeouts)?;
|
||||
config.open.validate()?;
|
||||
SessionMachine::validate_config(config)?;
|
||||
if config.socket_config.ipv4_only && config.socket_config.ipv6_only {
|
||||
return Err(ClientError::InvalidConfig {
|
||||
reason: "ipv4_only and ipv6_only cannot both be true".to_owned(),
|
||||
});
|
||||
}
|
||||
if let Some(ttl) = config.socket_config.ttl {
|
||||
if let Some(ttl) = config.socket.ttl {
|
||||
validate_ttl(ttl)?;
|
||||
}
|
||||
Ok(())
|
||||
@@ -2429,12 +2425,9 @@ mod tests {
|
||||
vec![ManagedTargetConfig::new("target", "127.0.0.1:9")],
|
||||
)
|
||||
.unwrap();
|
||||
let client = AsyncClient::connect(ClientConfig {
|
||||
server_addr: "127.0.0.1:9".into(),
|
||||
..ClientConfig::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let client = AsyncClient::connect("127.0.0.1:9", ClientConfig::default())
|
||||
.await
|
||||
.unwrap();
|
||||
task.state = DriverState::Running;
|
||||
task.install_target_state(0, TargetState::Active { client });
|
||||
task.send_gate = Some(end + Duration::from_secs(1));
|
||||
|
||||
@@ -7,7 +7,8 @@ use irtt_proto::{
|
||||
|
||||
use crate::{
|
||||
config::{
|
||||
ClientConfig, RunMode, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_UDP_PAYLOAD_LENGTH,
|
||||
ClientConfig, RunMode, SessionRequest, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES,
|
||||
MAX_UDP_PAYLOAD_LENGTH,
|
||||
},
|
||||
error::ClientError,
|
||||
event::{
|
||||
@@ -172,7 +173,7 @@ impl SessionMachine {
|
||||
remote: std::net::SocketAddr,
|
||||
) -> Result<Self, ClientError> {
|
||||
Self::validate_config(&config)?;
|
||||
let requested = params_from_config(&config)?;
|
||||
let requested = params_from_request(&config.request)?;
|
||||
|
||||
Ok(Self {
|
||||
config,
|
||||
@@ -193,7 +194,7 @@ impl SessionMachine {
|
||||
reason: "probe_timeout must be greater than zero".to_owned(),
|
||||
});
|
||||
}
|
||||
params_from_config(config).map(|_| ())
|
||||
params_from_request(&config.request).map(|_| ())
|
||||
}
|
||||
|
||||
pub(crate) fn config(&self) -> &ClientConfig {
|
||||
@@ -209,7 +210,7 @@ impl SessionMachine {
|
||||
let bytes = encode_request(
|
||||
RequestToEncode::Open {
|
||||
params: &self.requested,
|
||||
no_test: self.config.run_mode == RunMode::NoTest,
|
||||
no_test: self.config.request.run_mode == RunMode::NoTest,
|
||||
},
|
||||
self.config.auth.hmac_key(),
|
||||
)?;
|
||||
@@ -269,7 +270,7 @@ impl SessionMachine {
|
||||
));
|
||||
}
|
||||
|
||||
match self.config.run_mode {
|
||||
match self.config.request.run_mode {
|
||||
RunMode::Normal if reply_is_close => Err(OpenAcceptanceFailure::new(
|
||||
ClientError::ServerRejected,
|
||||
cleanup_close,
|
||||
@@ -602,14 +603,11 @@ impl SessionMachine {
|
||||
cleanup_close: Option<Box<[u8]>>,
|
||||
) -> Result<PreparedOpenAcceptance, OpenAcceptanceFailure> {
|
||||
let token = reply.token;
|
||||
let negotiated = match negotiate_params(
|
||||
&self.requested,
|
||||
reply.params,
|
||||
self.config.negotiation_policy,
|
||||
) {
|
||||
Ok(negotiated) => negotiated,
|
||||
Err(primary) => return Err(OpenAcceptanceFailure::new(primary, cleanup_close)),
|
||||
};
|
||||
let negotiated =
|
||||
match negotiate_params(&self.requested, reply.params, self.config.open.negotiation) {
|
||||
Ok(negotiated) => negotiated,
|
||||
Err(primary) => return Err(OpenAcceptanceFailure::new(primary, cleanup_close)),
|
||||
};
|
||||
let local_close_packet =
|
||||
cleanup_close.expect("normal non-zero-token replies prepare a cleanup close");
|
||||
let next_state = MachineState::Open(Box::new(ActiveSession {
|
||||
@@ -649,12 +647,9 @@ impl SessionMachine {
|
||||
reply: OpenReply,
|
||||
now: ClientTimestamp,
|
||||
) -> Result<PreparedOpenAcceptance, OpenAcceptanceFailure> {
|
||||
let negotiated = negotiate_params(
|
||||
&self.requested,
|
||||
reply.params,
|
||||
self.config.negotiation_policy,
|
||||
)
|
||||
.map_err(OpenAcceptanceFailure::without_cleanup)?;
|
||||
let negotiated =
|
||||
negotiate_params(&self.requested, reply.params, self.config.open.negotiation)
|
||||
.map_err(OpenAcceptanceFailure::without_cleanup)?;
|
||||
let event = ClientEvent::NoTestCompleted {
|
||||
remote: self.remote,
|
||||
negotiated: negotiated.clone(),
|
||||
@@ -888,21 +883,24 @@ pub(crate) fn recv_buffer_size(
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn params_from_config(config: &ClientConfig) -> Result<Params, ClientError> {
|
||||
validate_protocol_config(config)?;
|
||||
pub(crate) fn params_from_request(request: &SessionRequest) -> Result<Params, ClientError> {
|
||||
validate_request(request)?;
|
||||
Ok(Params {
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
duration_ns: match config.duration {
|
||||
duration_ns: match request.duration {
|
||||
Some(duration) => config_duration_to_ns("duration", duration)?,
|
||||
None => 0,
|
||||
},
|
||||
interval_ns: config_duration_to_ns("interval", config.interval)?,
|
||||
length: i64::from(config.length),
|
||||
received_stats: config.received_stats,
|
||||
stamp_at: config.stamp_at,
|
||||
clock: config.clock,
|
||||
dscp: i64::from(dscp_codepoint_to_traffic_class(config.dscp)?),
|
||||
server_fill: config.server_fill.clone().map(|value| ServerFill { value }),
|
||||
interval_ns: config_duration_to_ns("interval", request.interval)?,
|
||||
length: i64::from(request.length),
|
||||
received_stats: request.received_stats,
|
||||
stamp_at: request.stamp_at,
|
||||
clock: request.clock,
|
||||
dscp: i64::from(dscp_codepoint_to_traffic_class(request.dscp)?),
|
||||
server_fill: request
|
||||
.server_fill
|
||||
.clone()
|
||||
.map(|value| ServerFill { value }),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1083,34 +1081,34 @@ fn build_received_stats(reply: &EchoReply) -> Option<ReceivedStatsSample> {
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_protocol_config(config: &ClientConfig) -> Result<(), ClientError> {
|
||||
if config.duration == Some(Duration::ZERO) {
|
||||
fn validate_request(request: &SessionRequest) -> Result<(), ClientError> {
|
||||
if request.duration == Some(Duration::ZERO) {
|
||||
return Err(ClientError::InvalidConfig {
|
||||
reason: "duration must be greater than zero; use None for continuous mode".to_owned(),
|
||||
});
|
||||
}
|
||||
if config.interval == Duration::ZERO {
|
||||
if request.interval == Duration::ZERO {
|
||||
return Err(ClientError::InvalidConfig {
|
||||
reason: "interval must be greater than zero".to_owned(),
|
||||
});
|
||||
}
|
||||
if config.clock == Clock::Unspecified {
|
||||
if request.clock == Clock::Unspecified {
|
||||
return Err(ClientError::InvalidConfig {
|
||||
reason: "clock must be wall, monotonic, or both".to_owned(),
|
||||
});
|
||||
}
|
||||
if config.dscp > MAX_DSCP_CODEPOINT {
|
||||
if request.dscp > MAX_DSCP_CODEPOINT {
|
||||
return Err(ClientError::InvalidConfig {
|
||||
reason: format!("dscp must be <= {MAX_DSCP_CODEPOINT}"),
|
||||
});
|
||||
}
|
||||
if config.length > MAX_UDP_PAYLOAD_LENGTH {
|
||||
if request.length > MAX_UDP_PAYLOAD_LENGTH {
|
||||
return Err(ClientError::InvalidConfig {
|
||||
reason: format!("packet length must be <= {MAX_UDP_PAYLOAD_LENGTH}"),
|
||||
});
|
||||
}
|
||||
|
||||
if let Some(fill) = &config.server_fill {
|
||||
if let Some(fill) = &request.server_fill {
|
||||
let len = fill.len();
|
||||
if len == 0 {
|
||||
return Err(ClientError::InvalidConfig {
|
||||
@@ -1191,9 +1189,12 @@ mod tests {
|
||||
fn opened() -> SessionMachine {
|
||||
let mut machine = SessionMachine::new(
|
||||
ClientConfig {
|
||||
request: crate::SessionRequest {
|
||||
clock: Clock::Wall,
|
||||
stamp_at: irtt_proto::StampAt::Receive,
|
||||
..Default::default()
|
||||
},
|
||||
max_pending_probes: 2,
|
||||
clock: Clock::Wall,
|
||||
stamp_at: irtt_proto::StampAt::Receive,
|
||||
..ClientConfig::default()
|
||||
},
|
||||
"127.0.0.1:2112".parse().unwrap(),
|
||||
|
||||
@@ -13,7 +13,7 @@ use crate::{config::NegotiationPolicy, error::ClientError};
|
||||
/// `params.dscp` is the raw IP TOS / Traffic Class byte, not a DSCP
|
||||
/// codepoint: for a configured codepoint of 46 (EF), an unrestricted
|
||||
/// negotiation leaves `params.dscp == 184`. [`NegotiationRestriction::DscpChanged`]
|
||||
/// reports codepoints instead, for consistency with [`crate::ClientConfig::dscp`].
|
||||
/// reports codepoints instead, for consistency with [`crate::SessionRequest::dscp`].
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct NegotiatedParams {
|
||||
/// Server-returned protocol parameters used for the session.
|
||||
@@ -61,7 +61,7 @@ pub enum NegotiationRestriction {
|
||||
/// Returned DSCP codepoint differs from the request.
|
||||
///
|
||||
/// These values are DSCP codepoints (`0..=63`) for human-facing
|
||||
/// consistency with [`crate::ClientConfig::dscp`], not the raw wire
|
||||
/// consistency with [`crate::SessionRequest::dscp`], not the raw wire
|
||||
/// `Params::dscp` byte carried by [`NegotiatedParams::params`].
|
||||
DscpChanged { requested: i64, negotiated: i64 },
|
||||
/// Returned server payload fill behavior differs from the request.
|
||||
|
||||
@@ -1,12 +1,9 @@
|
||||
use std::{
|
||||
net::{SocketAddr, ToSocketAddrs, UdpSocket},
|
||||
time::Duration,
|
||||
};
|
||||
use std::net::{SocketAddr, ToSocketAddrs, UdpSocket};
|
||||
|
||||
use socket2::{Domain, Protocol, Socket, Type};
|
||||
|
||||
use crate::{
|
||||
config::{ClientConfig, SocketConfig, DEFAULT_PORT, MIN_OPEN_TIMEOUT},
|
||||
config::{AddressFamily, SocketConfig, DEFAULT_PORT},
|
||||
error::ClientError,
|
||||
receive::configure_receive_metadata,
|
||||
socket_options::apply_ttl_to_socket,
|
||||
@@ -20,36 +17,27 @@ use crate::{
|
||||
))]
|
||||
use crate::socket_options::apply_routing_options;
|
||||
|
||||
pub(crate) fn validate_open_timeouts(timeouts: &[Duration]) -> Result<(), ClientError> {
|
||||
if timeouts.is_empty() {
|
||||
return Err(ClientError::NoOpenTimeouts);
|
||||
}
|
||||
for timeout in timeouts {
|
||||
if *timeout < MIN_OPEN_TIMEOUT {
|
||||
return Err(ClientError::OpenTimeoutTooSmall {
|
||||
timeout: *timeout,
|
||||
minimum: MIN_OPEN_TIMEOUT,
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn resolve_remote(config: &ClientConfig) -> Result<SocketAddr, ClientError> {
|
||||
let addr = normalize_server_addr(&config.server_addr);
|
||||
pub(crate) fn resolve_remote(
|
||||
endpoint: &str,
|
||||
family: AddressFamily,
|
||||
) -> Result<SocketAddr, ClientError> {
|
||||
let addr = normalize_server_addr(endpoint);
|
||||
let mut addrs = addr
|
||||
.to_socket_addrs()
|
||||
.map_err(|_| ClientError::Resolve { addr: addr.clone() })?;
|
||||
addrs
|
||||
.find(|addr| address_family_allowed(config, *addr))
|
||||
.find(|addr| address_family_allowed(family, *addr))
|
||||
.ok_or(ClientError::Resolve { addr })
|
||||
}
|
||||
|
||||
#[cfg(feature = "tokio")]
|
||||
pub(crate) async fn resolve_remote_tokio(config: &ClientConfig) -> Result<SocketAddr, ClientError> {
|
||||
let addr = normalize_server_addr(&config.server_addr);
|
||||
pub(crate) async fn resolve_remote_tokio(
|
||||
endpoint: &str,
|
||||
family: AddressFamily,
|
||||
) -> Result<SocketAddr, ClientError> {
|
||||
let addr = normalize_server_addr(endpoint);
|
||||
if let Ok(remote) = addr.parse::<SocketAddr>() {
|
||||
return address_family_allowed(config, remote)
|
||||
return address_family_allowed(family, remote)
|
||||
.then_some(remote)
|
||||
.ok_or(ClientError::Resolve { addr });
|
||||
}
|
||||
@@ -58,13 +46,16 @@ pub(crate) async fn resolve_remote_tokio(config: &ClientConfig) -> Result<Socket
|
||||
.await
|
||||
.map_err(|_| ClientError::Resolve { addr: addr.clone() })?;
|
||||
addrs
|
||||
.find(|remote| address_family_allowed(config, *remote))
|
||||
.find(|remote| address_family_allowed(family, *remote))
|
||||
.ok_or_else(|| ClientError::Resolve { addr: addr.clone() })
|
||||
}
|
||||
|
||||
fn address_family_allowed(config: &ClientConfig, remote: SocketAddr) -> bool {
|
||||
(!config.socket_config.ipv4_only || remote.is_ipv4())
|
||||
&& (!config.socket_config.ipv6_only || remote.is_ipv6())
|
||||
fn address_family_allowed(family: AddressFamily, remote: SocketAddr) -> bool {
|
||||
match family {
|
||||
AddressFamily::Any => true,
|
||||
AddressFamily::Ipv4 => remote.is_ipv4(),
|
||||
AddressFamily::Ipv6 => remote.is_ipv6(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn normalize_server_addr(addr: &str) -> String {
|
||||
@@ -92,17 +83,9 @@ pub(crate) fn normalize_server_addr(addr: &str) -> String {
|
||||
pub(crate) fn connect_udp_socket(
|
||||
config: &SocketConfig,
|
||||
remote: SocketAddr,
|
||||
family: AddressFamily,
|
||||
) -> Result<UdpSocket, ClientError> {
|
||||
let socket = create_connected_udp_socket(config, remote)?;
|
||||
socket.set_read_timeout(config.recv_timeout)?;
|
||||
Ok(socket)
|
||||
}
|
||||
|
||||
fn create_connected_udp_socket(
|
||||
config: &SocketConfig,
|
||||
remote: SocketAddr,
|
||||
) -> Result<UdpSocket, ClientError> {
|
||||
let socket = create_prebind_udp_socket(config, remote)?;
|
||||
let socket = create_prebind_udp_socket(remote, family)?;
|
||||
#[cfg(any(
|
||||
target_os = "android",
|
||||
target_os = "freebsd",
|
||||
@@ -133,8 +116,8 @@ fn create_connected_udp_socket(
|
||||
}
|
||||
|
||||
fn create_prebind_udp_socket(
|
||||
config: &SocketConfig,
|
||||
remote: SocketAddr,
|
||||
family: AddressFamily,
|
||||
) -> Result<Socket, ClientError> {
|
||||
let domain = if remote.is_ipv4() {
|
||||
Domain::IPV4
|
||||
@@ -143,7 +126,7 @@ fn create_prebind_udp_socket(
|
||||
};
|
||||
let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
|
||||
|
||||
if config.ipv6_only && remote.is_ipv6() {
|
||||
if family == AddressFamily::Ipv6 && remote.is_ipv6() {
|
||||
socket.set_only_v6(true)?;
|
||||
}
|
||||
Ok(socket)
|
||||
@@ -153,8 +136,9 @@ fn create_prebind_udp_socket(
|
||||
pub(crate) fn connect_tokio_udp_socket(
|
||||
config: &SocketConfig,
|
||||
remote: SocketAddr,
|
||||
family: AddressFamily,
|
||||
) -> Result<tokio::net::UdpSocket, ClientError> {
|
||||
let socket = create_connected_udp_socket(config, remote)?;
|
||||
let socket = connect_udp_socket(config, remote, family)?;
|
||||
socket.set_nonblocking(true)?;
|
||||
Ok(tokio::net::UdpSocket::from_std(socket)?)
|
||||
}
|
||||
Reference in new issue
Block a user