diff --git a/Cargo.toml b/Cargo.toml index dc8e2ff..978bb15 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,6 @@ license = "Apache-2.0" rust-version = "1.88" repository = "https://github.com/Lochnair/irtt-rs" homepage = "https://github.com/Lochnair/irtt-rs" -readme = "README.md" [workspace.dependencies] irtt-proto = { version = "0.5.0", path = "crates/irtt-proto" } diff --git a/crates/irtt-app/Cargo.toml b/crates/irtt-app/Cargo.toml index 2c30424..14687e7 100644 --- a/crates/irtt-app/Cargo.toml +++ b/crates/irtt-app/Cargo.toml @@ -7,7 +7,7 @@ license.workspace = true rust-version.workspace = true repository.workspace = true homepage.workspace = true -readme.workspace = true +readme = "README.md" keywords = ["irtt", "networking", "latency", "rtt"] categories = ["command-line-utilities", "network-programming"] diff --git a/crates/irtt-app/README.md b/crates/irtt-app/README.md new file mode 100644 index 0000000..2e2afda --- /dev/null +++ b/crates/irtt-app/README.md @@ -0,0 +1,33 @@ +# irtt-rs + +The IRTT-compatible client, TUI, and server applications, packaged as Cargo +package `irtt-rs`. + +Installing this crate builds: + +- `irtt-rs` — a multicall dispatcher binary that runs the client, TUI, or + server applet by subcommand. +- `irtt-client` — the dedicated client binary (feature `client`). +- `irtt-tui` — the interactive terminal UI (feature `tui`). +- `irtt-server` — the dedicated server binary (feature `server`). + +All three features are enabled by default, so a plain `cargo install irtt-rs` +installs everything. + +## This is an application, not a library + +This package's library target is undocumented (`doc = false`) and exists only +to share code between its binaries. If you're looking for the client/session +or server API to embed in your own Rust program, use +[`irtt-client`](https://crates.io/crates/irtt-client) or +[`irtt-server`](https://crates.io/crates/irtt-server) instead. + +## Usage + +See the [project repository](https://github.com/Lochnair/irtt-rs) for +installation instructions, command-line usage, and configuration. + +## Project + +Part of [irtt-rs](https://github.com/Lochnair/irtt-rs), an independent Rust +implementation of an IRTT-compatible protocol stack. diff --git a/crates/irtt-client/Cargo.toml b/crates/irtt-client/Cargo.toml index 2de2a58..a01740f 100644 --- a/crates/irtt-client/Cargo.toml +++ b/crates/irtt-client/Cargo.toml @@ -7,7 +7,7 @@ license.workspace = true rust-version.workspace = true repository.workspace = true homepage.workspace = true -readme.workspace = true +readme = "README.md" keywords = ["irtt", "networking", "latency", "rtt"] categories = ["network-programming"] @@ -27,6 +27,21 @@ tokio = ["dep:tokio"] [package.metadata.docs.rs] all-features = true +cargo-args = ["-Zunstable-options", "-Zrustdoc-scrape-examples"] + +[[example]] +name = "blocking_client" +doc-scrape-examples = true + +[[example]] +name = "tokio_client" +required-features = ["tokio"] +doc-scrape-examples = true + +[[example]] +name = "managed_client" +required-features = ["tokio"] +doc-scrape-examples = true [dev-dependencies] irtt-server.workspace = true diff --git a/crates/irtt-client/README.md b/crates/irtt-client/README.md new file mode 100644 index 0000000..c388e60 --- /dev/null +++ b/crates/irtt-client/README.md @@ -0,0 +1,37 @@ +# irtt-client + +Reusable client/session library for IRTT-compatible round-trip-time probing: +socket lifecycle, open negotiation, probe send/receive, loss/duplicate/late +classification, and a typed event stream. + +## Looking for the CLI? + +If you just want to run probes from the command line, use the +[`irtt-rs`](https://crates.io/crates/irtt-rs) application package instead, +which installs the `irtt-client` binary. This crate is the library the +binaries are built on. + +## API tiers + +- `Client` — runtime-free blocking adapter. No Tokio dependency; this is the + default build. +- `AsyncClient` (feature `tokio`) — low-level Tokio adapter for callers that + own a runtime and drive readiness directly. +- `managed` (feature `tokio`) — a unified managed driver: `ManagedClientTask` + / `ManagedClientHandle` for multi-target orchestration under a Tokio + runtime, and `BlockingManagedClient` for synchronous callers, which owns its + own dedicated current-thread runtime. + +Tokio stays optional: the default build has no runtime and no Tokio +dependency at all. + +See `examples/` in the repository for runnable examples of each tier. + +## Documentation + +Full API documentation is on [docs.rs/irtt-client](https://docs.rs/irtt-client). + +## Project + +Part of [irtt-rs](https://github.com/Lochnair/irtt-rs), an independent Rust +implementation of an IRTT-compatible protocol stack. diff --git a/crates/irtt-client/examples/blocking_client.rs b/crates/irtt-client/examples/blocking_client.rs new file mode 100644 index 0000000..d5f103f --- /dev/null +++ b/crates/irtt-client/examples/blocking_client.rs @@ -0,0 +1,110 @@ +//! Runtime-free blocking client example. +//! +//! Opens a session against a local IRTT-compatible server, sends a few +//! probes, and prints the events observed. Uses only [`Client`], which has no +//! Tokio dependency and works in a default (`--no-default-features`) build. +//! +//! Run a server first, then run this example: +//! +//! ```text +//! cargo run -p irtt-rs --bin irtt-server --features server +//! cargo run -p irtt-client --example blocking_client +//! ``` +//! +//! Without a reachable server this exits quickly with an open-timeout error +//! instead of hanging, because the example shortens `open_timeouts` for a +//! fast demonstration; a real caller would normally keep the default retry +//! schedule. + +use std::time::{Duration, Instant}; + +use irtt_client::{Client, ClientConfig, ClientEvent}; + +const RUN_DURATION: Duration = Duration::from_secs(2); + +fn main() { + let config = ClientConfig { + server_addr: "127.0.0.1:2112".to_owned(), + duration: Some(RUN_DURATION), + interval: Duration::from_millis(200), + // A single short attempt keeps this example fast when no server is + // reachable; production callers should generally keep the default. + open_timeouts: vec![Duration::from_millis(300)], + socket_config: irtt_client::SocketConfig { + recv_timeout: Some(Duration::from_millis(100)), + ..Default::default() + }, + ..Default::default() + }; + + let mut client = match Client::connect(config) { + Ok(client) => client, + Err(err) => { + eprintln!("failed to prepare client socket: {err}"); + return; + } + }; + + match client.open() { + Ok(outcome) => println!("session opened: {outcome:?}"), + Err(err) => { + eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}"); + return; + } + } + + // The run itself lasts `duration`, but a lost probe isn't classified as + // `EchoLoss` until `probe_timeout()` after it was sent; wait at least + // that much past the run's own end so a loss near the end can still be + // observed and printed before this example gives up and closes. + let deadline = Instant::now() + RUN_DURATION + client.probe_timeout(); + while Instant::now() < deadline && !client.is_run_complete() { + // send_probe is caller-paced: only call it once its own schedule + // says a probe is due, rather than as fast as the loop spins. + if client + .next_send_deadline() + .is_some_and(|send_at| Instant::now() >= send_at) + { + // A failed send retains its prepared probe without advancing the + // schedule, so a persistent error (e.g. the interface going + // down) would otherwise make this loop retry silently until the + // safety deadline instead of reporting it. + match client.send_probe() { + Ok(events) => events.iter().for_each(print_event), + Err(err) => { + eprintln!("send failed: {err}"); + break; + } + } + } + match client.recv_once() { + Ok(events) => events.iter().for_each(print_event), + Err(err) => { + eprintln!("receive failed: {err}"); + break; + } + } + match client.poll_timeouts() { + Ok(events) => events.iter().for_each(print_event), + Err(err) => { + eprintln!("timeout polling failed: {err}"); + break; + } + } + } + + match client.close() { + Ok(events) => events.iter().for_each(print_event), + Err(err) => eprintln!("close failed: {err}"), + } +} + +fn print_event(event: &ClientEvent) { + match event { + ClientEvent::EchoReply { seq, rtt, .. } => { + println!("seq {seq}: rtt {:?}", rtt.effective); + } + ClientEvent::EchoLoss { seq, .. } => println!("seq {seq}: lost"), + other => println!("{other:?}"), + } +} diff --git a/crates/irtt-client/examples/managed_client.rs b/crates/irtt-client/examples/managed_client.rs new file mode 100644 index 0000000..1377442 --- /dev/null +++ b/crates/irtt-client/examples/managed_client.rs @@ -0,0 +1,73 @@ +//! Managed multi-target client example (`tokio` feature). +//! +//! [`BlockingManagedClient`] runs one or more targets under a dedicated +//! current-thread Tokio runtime and exposes a synchronous handle, so a caller +//! that does not want to touch Tokio directly can still get managed +//! multi-target orchestration, retries, and a lossy event stream. +//! +//! Run a server first, then run this example: +//! +//! ```text +//! cargo run -p irtt-rs --bin irtt-server --features server +//! cargo run -p irtt-client --example managed_client --features tokio +//! ``` +//! +//! This example does not wait for the managed run to finish naturally: it +//! requests a stop after a fixed short window so it exits promptly even +//! without a reachable server. + +use std::time::Duration; + +use irtt_client::managed::{ + BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEvent, + ManagedTargetConfig, +}; +use irtt_client::ClientConfig; + +fn main() { + let config = ManagedClientConfig { + client: ClientConfig { + duration: None, + interval: Duration::from_millis(200), + open_timeouts: vec![Duration::from_millis(300)], + ..Default::default() + }, + completion: ManagedCompletionPolicy::ExplicitStop, + ..Default::default() + }; + let targets = vec![ManagedTargetConfig::new("local", "127.0.0.1:2112")]; + + let (owner, mut events) = match BlockingManagedClient::start_with_subscription(config, targets) + { + Ok(started) => started, + Err(err) => { + eprintln!("failed to start managed client: {err}"); + return; + } + }; + + let stop_after = Duration::from_secs(1); + let deadline = std::time::Instant::now() + stop_after; + while std::time::Instant::now() < deadline { + match events.try_recv() { + Ok(event) => print_event(&event), + Err(_) => std::thread::sleep(Duration::from_millis(50)), + } + } + + // `stop()` requests the stop synchronously; the returned receipt is only + // useful for awaiting durable acknowledgement, which this blocking + // example does not need. + std::mem::drop(owner.handle().stop()); + match owner.join() { + Ok(outcome) => println!("managed run ended: {:?}", outcome.end_reason), + Err(err) => eprintln!("managed worker did not join cleanly: {err}"), + } +} + +fn print_event(event: &ManagedEvent) { + match event { + ManagedEvent::Client { target, event } => println!("{}: {event:?}", target.id), + other => println!("{other:?}"), + } +} diff --git a/crates/irtt-client/examples/tokio_client.rs b/crates/irtt-client/examples/tokio_client.rs new file mode 100644 index 0000000..f46c6b3 --- /dev/null +++ b/crates/irtt-client/examples/tokio_client.rs @@ -0,0 +1,133 @@ +//! Low-level Tokio client example (`tokio` feature). +//! +//! Drives [`AsyncClient`] directly on a caller-owned runtime. This is the +//! adapter to reach for when you already have a Tokio runtime and want to +//! await sends/receives yourself, rather than use the higher-level `managed` +//! driver (see the `managed_client` example). +//! +//! Run a server first, then run this example: +//! +//! ```text +//! cargo run -p irtt-rs --bin irtt-server --features server +//! cargo run -p irtt-client --example tokio_client --features tokio +//! ``` +//! +//! Without a reachable server this exits quickly with an open-timeout error +//! instead of hanging, because the example shortens `open_timeouts` for a +//! fast demonstration. + +use std::time::{Duration, Instant}; + +use irtt_client::{AsyncClient, ClientConfig, ClientEvent}; + +/// Upper bound on how long the receive wait can go without also calling +/// `poll_timeouts()`. `AsyncClient` does not expose the earliest pending +/// probe's own timeout deadline, so this example polls on a short fixed +/// cadence instead of waiting for the full `probe_timeout()` from whenever +/// this loop last ran — that would let poll_timeouts() classify a lost +/// probe (and its dependent statistics/session state) far later than it +/// actually became lost. +const TIMEOUT_POLL_INTERVAL: Duration = Duration::from_millis(50); + +fn main() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .expect("failed to build Tokio runtime"); + + runtime.block_on(run()); +} + +async fn run() { + let config = ClientConfig { + server_addr: "127.0.0.1:2112".to_owned(), + duration: Some(Duration::from_secs(2)), + interval: Duration::from_millis(200), + // A single short attempt keeps this example fast when no server is + // reachable; production callers should generally keep the default. + open_timeouts: vec![Duration::from_millis(300)], + ..Default::default() + }; + + let mut client = match AsyncClient::connect(config).await { + Ok(client) => client, + Err(err) => { + eprintln!("failed to prepare client socket: {err}"); + return; + } + }; + + match client.open().await { + Ok(outcome) => println!("session opened: {outcome:?}"), + Err(err) => { + eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}"); + return; + } + } + + while !client.is_run_complete() { + // Bound the receive wait by the next send/timeout deadline, rather + // than awaiting recv() unconditionally: a lost or rate-limited reply + // must not stall pacing or timeout classification indefinitely. + let wake_at = client + .next_send_deadline() + .into_iter() + .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL)) + .min() + .expect("the fixed poll-interval deadline is always present"); + + let recv_result = tokio::select! { + events = client.recv() => Some(events), + () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None, + }; + if let Some(result) = recv_result { + match result { + Ok(events) => events.iter().for_each(print_event), + Err(err) => { + eprintln!("receive failed: {err}"); + break; + } + } + } + + if client + .next_send_deadline() + .is_some_and(|send_at| Instant::now() >= send_at) + { + // A failed send retains its prepared probe without advancing the + // schedule, so a persistent error (e.g. the interface going + // down) would otherwise make this loop retry forever instead of + // reporting it. + match client.send_probe().await { + Ok(events) => events.iter().for_each(print_event), + Err(err) => { + eprintln!("send failed: {err}"); + break; + } + } + } + match client.poll_timeouts() { + Ok(events) => events.iter().for_each(print_event), + Err(err) => { + eprintln!("timeout polling failed: {err}"); + break; + } + } + } + + match client.close().await { + Ok(events) => events.iter().for_each(print_event), + Err(err) => eprintln!("close failed: {err}"), + } +} + +fn print_event(event: &ClientEvent) { + match event { + ClientEvent::EchoReply { seq, rtt, .. } => { + println!("seq {seq}: rtt {:?}", rtt.effective); + } + ClientEvent::EchoLoss { seq, .. } => println!("seq {seq}: lost"), + other => println!("{other:?}"), + } +} diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index 58471f8..83a6b4b 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -13,10 +13,19 @@ //! also signed when the required wall-clock timestamps are available; negative //! values usually indicate client/server clock skew. //! -//! [`Client`] is the runtime-free low-level blocking adapter. With the -//! `tokio` feature, `AsyncClient` provides the corresponding Tokio adapter; -//! [`managed`] contains the unified managed task/handle implementation, -//! including the dedicated-runtime blocking frontend for synchronous callers. +//! [`Client`] is the runtime-free low-level blocking adapter and requires no +//! Tokio dependency at all; it is what a caller without a runtime, or that +//! wants full control over its own event loop, should start from. With the +//! `tokio` feature, `AsyncClient` provides the corresponding low-level Tokio +//! adapter for a caller that owns a runtime and drives socket readiness +//! itself. [`managed`] sits above both: `ManagedClientTask` / +//! `ManagedClientHandle` (`tokio` feature) run and control one or more +//! targets under a Tokio runtime, and `BlockingManagedClient` (`tokio` +//! feature) wraps that in a synchronous owner backed by its own dedicated +//! current-thread runtime for callers that want managed multi-target +//! behavior without touching Tokio themselves. +//! +//! See `examples/` in the repository for a runnable example of each tier. //! #![forbid(unsafe_code)] diff --git a/crates/irtt-proto/Cargo.toml b/crates/irtt-proto/Cargo.toml index 143b1cc..0838601 100644 --- a/crates/irtt-proto/Cargo.toml +++ b/crates/irtt-proto/Cargo.toml @@ -7,7 +7,7 @@ license.workspace = true rust-version.workspace = true repository.workspace = true homepage.workspace = true -readme.workspace = true +readme = "README.md" keywords = ["irtt", "networking", "latency", "protocol"] categories = ["network-programming"] diff --git a/crates/irtt-proto/README.md b/crates/irtt-proto/README.md new file mode 100644 index 0000000..fc0a4a2 --- /dev/null +++ b/crates/irtt-proto/README.md @@ -0,0 +1,30 @@ +# irtt-proto + +Low-level wire encoding and decoding for the IRTT-compatible protocol: request +and reply structs, flags, negotiable parameters, varint/zigzag primitives, and +HMAC-MD5 packet authentication. No sockets, no runtime, no session state. + +## Who should use this + +Most applications should not depend on this crate directly. Use +[`irtt-client`](https://docs.rs/irtt-client) to run client sessions or +[`irtt-server`](https://docs.rs/irtt-server) to run a server; both build on +`irtt-proto` and provide session lifecycle, negotiation, and validated +configuration on top of it. + +Reach for `irtt-proto` directly only if you are implementing another +IRTT-compatible client or server from scratch and need the raw packet layer. + +Field values mirror the wire format closely rather than expressing policy: +`Params::decode` rejects malformed input, but direct construction plus +`Params::encode` performs no additional validation — callers are responsible +for validating values before sending them. + +## Documentation + +Full API documentation is on [docs.rs/irtt-proto](https://docs.rs/irtt-proto). + +## Project + +Part of [irtt-rs](https://github.com/Lochnair/irtt-rs), an independent Rust +implementation of an IRTT-compatible protocol stack. diff --git a/crates/irtt-proto/src/lib.rs b/crates/irtt-proto/src/lib.rs index b1ebc17..3a8e7f1 100644 --- a/crates/irtt-proto/src/lib.rs +++ b/crates/irtt-proto/src/lib.rs @@ -6,6 +6,10 @@ //! crates such as `irtt-client` are responsible for session behavior and //! user-facing validation. //! +//! Most applications should depend on `irtt-client` or `irtt-server` instead +//! of this crate directly; use `irtt-proto` only when implementing another +//! IRTT-compatible client or server that needs the raw packet layer itself. +//! //! [`Params::decode`] rejects malformed or incompatible incoming parameter //! values, including invalid enum discriminants, malformed UTF-8, and oversized //! `server_fill` values. Direct construction followed by [`Params::encode`] diff --git a/crates/irtt-server/Cargo.toml b/crates/irtt-server/Cargo.toml index 0d1027e..f77bc13 100644 --- a/crates/irtt-server/Cargo.toml +++ b/crates/irtt-server/Cargo.toml @@ -7,7 +7,7 @@ license.workspace = true rust-version.workspace = true repository.workspace = true homepage.workspace = true -readme.workspace = true +readme = "README.md" keywords = ["irtt", "networking", "latency", "server"] categories = ["network-programming"] @@ -23,3 +23,13 @@ tokio = { version = "1.53.1", default-features = false, features = ["rt", "net", # source-selection path here; elsewhere a wildcard bind is refused. [target.'cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))'.dependencies] nix = { version = "0.31.3", default-features = false, features = ["net", "uio"] } + +[package.metadata.docs.rs] +all-features = true +cargo-args = ["-Zunstable-options", "-Zrustdoc-scrape-examples"] + +[[example]] +name = "basic_server" + +[[example]] +name = "multi_listener" diff --git a/crates/irtt-server/README.md b/crates/irtt-server/README.md new file mode 100644 index 0000000..669e5b8 --- /dev/null +++ b/crates/irtt-server/README.md @@ -0,0 +1,41 @@ +# irtt-server + +Reusable, Tokio-native library for running an IRTT-compatible UDP server: +open negotiation, the bounded session table, echo processing, per-session +rate limiting, idle and max-duration expiry, and reply construction. + +## Looking for the CLI? + +If you just want to run a server from the command line, use the +[`irtt-rs`](https://crates.io/crates/irtt-rs) application package instead, +which installs the `irtt-server` binary. This crate is the library that +binary is built on. + +## Structure + +- `Server` — the one-listener primitive: one UDP socket, one deterministic + `ServerCore`, one sequential receive/reply loop. +- `ServerSet` — supervises one or more `Server`s as a single service: binds + them all or none, runs each in its own task, and fans one shutdown out to + all of them. + +Unlike `irtt-client`, Tokio is not optional here — this crate is intentionally +Tokio-native, with no blocking or alternate-runtime variant. + +A wildcard listener (`0.0.0.0` / `[::]`) recovers each request's local +destination and replies from it, so a client on a second local address is +answered from the address it actually contacted. That path is implemented +for Linux, macOS, and FreeBSD; a wildcard bind is refused at construction on +other targets rather than served incorrectly. Explicit-address listeners are +unaffected everywhere. + +See `examples/` in the repository for runnable examples. + +## Documentation + +Full API documentation is on [docs.rs/irtt-server](https://docs.rs/irtt-server). + +## Project + +Part of [irtt-rs](https://github.com/Lochnair/irtt-rs), an independent Rust +implementation of an IRTT-compatible protocol stack. diff --git a/crates/irtt-server/examples/basic_server.rs b/crates/irtt-server/examples/basic_server.rs new file mode 100644 index 0000000..e6b00dc --- /dev/null +++ b/crates/irtt-server/examples/basic_server.rs @@ -0,0 +1,42 @@ +//! Basic single-listener server example. +//! +//! Binds one [`Server`] on an ephemeral loopback port, serves it for a short +//! fixed window, and shuts down. A real deployment would normally bind a +//! fixed port and run until an external shutdown signal. +//! +//! Run with: +//! +//! ```text +//! cargo run -p irtt-server --example basic_server +//! ``` + +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::time::Duration; + +use irtt_server::{Server, ServerConfig}; + +#[tokio::main(flavor = "current_thread")] +async fn main() { + let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + let config = ServerConfig::default(); + + let mut server = match Server::bind(bind_addr, config).await { + Ok(server) => server, + Err(err) => { + eprintln!("failed to bind server: {err}"); + return; + } + }; + + println!( + "listening on {}", + server + .local_addr() + .expect("bound socket has a local address") + ); + + let result = server.run(tokio::time::sleep(Duration::from_secs(2))).await; + if let Err(err) = result { + eprintln!("server stopped with an error: {err}"); + } +} diff --git a/crates/irtt-server/examples/multi_listener.rs b/crates/irtt-server/examples/multi_listener.rs new file mode 100644 index 0000000..6238050 --- /dev/null +++ b/crates/irtt-server/examples/multi_listener.rs @@ -0,0 +1,43 @@ +//! Multi-listener example using [`ServerSet`]. +//! +//! `ServerSet` binds several independent [`Server`] listeners as one service: +//! all-or-nothing bind, one shutdown fanned out to every listener, and the +//! whole set fails if any listener does. Each listener still gets its own +//! socket, session table, and tokens — a token issued by one is unknown at +//! the other. This is what the `irtt-rs` server CLI always runs through, even +//! for a single bind. +//! +//! Run with: +//! +//! ```text +//! cargo run -p irtt-server --example multi_listener +//! ``` + +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::time::Duration; + +use irtt_server::{ServerConfig, ServerSet}; + +#[tokio::main(flavor = "current_thread")] +async fn main() { + let addrs = [ + SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), + SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), + ]; + let config = ServerConfig::default(); + + let set = match ServerSet::bind(addrs, config).await { + Ok(set) => set, + Err(err) => { + eprintln!("failed to bind listener set: {err}"); + return; + } + }; + + println!("listening on {:?}", set.local_addrs()); + + let result = set.run(tokio::time::sleep(Duration::from_secs(2))).await; + if let Err(err) = result { + eprintln!("listener set stopped with an error: {err}"); + } +} diff --git a/crates/irtt-stats/Cargo.toml b/crates/irtt-stats/Cargo.toml index be164d3..216ae0c 100644 --- a/crates/irtt-stats/Cargo.toml +++ b/crates/irtt-stats/Cargo.toml @@ -7,12 +7,15 @@ license.workspace = true rust-version.workspace = true repository.workspace = true homepage.workspace = true -readme.workspace = true +readme = "README.md" keywords = ["irtt", "networking", "latency", "statistics"] categories = ["network-programming"] [dependencies] irtt-client.workspace = true +[[example]] +name = "finite_vs_continuous" + [dev-dependencies] proptest = "1.11.0" diff --git a/crates/irtt-stats/README.md b/crates/irtt-stats/README.md new file mode 100644 index 0000000..ee04402 --- /dev/null +++ b/crates/irtt-stats/README.md @@ -0,0 +1,38 @@ +# irtt-stats + +Statistics aggregation over [`irtt-client`](https://crates.io/crates/irtt-client) +events: loss/duplicate/late accounting, RTT and one-way-delay timing, and +inter-packet delay variation (IPDV), as cumulative and rolling snapshots. + +## Relationship to irtt-client + +This crate does not open sessions or drive sockets. Feed it the `ClientEvent` +stream produced by an `irtt-client` `Client`, `AsyncClient`, or managed +session via `StatsCollector::process`, and read back a `Snapshot`. + +A `StatsCollector` has one sequence/IPDV namespace and one packet count, both +scoped to a single target. A managed session covering multiple targets needs +one collector per target — as the `irtt-rs` CLI does — since a target's +sequence numbers start over at zero and feeding two targets into one +collector would pair up unrelated probes. + +## Retention modes + +`StatsConfig::finite()` retains exact timing samples for its cumulative +snapshot, so every metric with samples reports an exact median there — +retention grows with the probe count. `StatsConfig::continuous()` keeps +bounded running statistics instead (no exact median) plus a bounded +4096-entry adjacent-sequence IPDV store per target, for long-running or +unbounded sessions. Rolling-window snapshots are always running-only and +report no medians, whichever `StatsConfig` produced them. + +See `examples/` in the repository for a runnable comparison of both modes. + +## Documentation + +Full API documentation is on [docs.rs/irtt-stats](https://docs.rs/irtt-stats). + +## Project + +Part of [irtt-rs](https://github.com/Lochnair/irtt-rs), an independent Rust +implementation of an IRTT-compatible protocol stack. diff --git a/crates/irtt-stats/examples/finite_vs_continuous.rs b/crates/irtt-stats/examples/finite_vs_continuous.rs new file mode 100644 index 0000000..463be44 --- /dev/null +++ b/crates/irtt-stats/examples/finite_vs_continuous.rs @@ -0,0 +1,102 @@ +//! Contrasts finite and continuous retention modes. +//! +//! Feeds the same synthetic stream of `irtt-client` events into two +//! collectors — one configured with [`StatsConfig::finite`], one with +//! [`StatsConfig::continuous`] — and prints how their cumulative snapshots +//! differ. This uses synthetic events rather than a live session so the +//! example runs deterministically with no network or server involved. +//! +//! Run with: +//! +//! ```text +//! cargo run -p irtt-stats --example finite_vs_continuous +//! ``` + +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::time::{Duration, Instant, SystemTime}; + +use irtt_client::{ClientEvent, ClientTimestamp, PacketMeta, RttSample, SignedDuration}; +use irtt_stats::{StatsCollector, StatsConfig}; + +fn main() { + let mut finite = StatsCollector::new(StatsConfig::finite()); + let mut continuous = StatsCollector::new(StatsConfig::continuous()); + + for event in synthetic_probe_stream(10) { + finite.process(&event); + continuous.process(&event); + } + + let finite_snapshot = finite.snapshot(); + let continuous_snapshot = continuous.snapshot(); + + println!( + "finite: replies={}, rtt median_ns={:?}", + finite_snapshot.events.echo_replies, finite_snapshot.rtt.primary.median_ns + ); + println!( + "continuous: replies={}, rtt median_ns={:?}", + continuous_snapshot.events.echo_replies, continuous_snapshot.rtt.primary.median_ns + ); + println!( + "continuous mode keeps no exact samples, so its median is always None, \ + while finite mode's median is available once it has samples." + ); +} + +/// Builds a short synthetic stream of `EchoSent`/`EchoReply` pairs, plus one +/// loss, standing in for what a real `irtt-client` session would emit. +fn synthetic_probe_stream(count: u32) -> Vec { + let remote = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 2112); + let mut events = Vec::new(); + + for seq in 0..count { + let sent_at = ClientTimestamp { + wall: SystemTime::now(), + mono: Instant::now(), + }; + events.push(ClientEvent::EchoSent { + seq, + remote, + scheduled_at: sent_at.mono, + sent_at, + bytes: 64, + send_call: Duration::from_micros(50), + timer_error: Duration::from_micros(20), + }); + + if seq == count - 1 { + // The last probe times out instead of receiving a reply. + events.push(ClientEvent::EchoLoss { + seq, + sent_at, + timeout_at: sent_at.mono + Duration::from_secs(1), + }); + continue; + } + + let rtt = Duration::from_millis(10 + u64::from(seq)); + let received_at = ClientTimestamp { + wall: sent_at.wall + rtt, + mono: sent_at.mono + rtt, + }; + events.push(ClientEvent::EchoReply { + seq, + remote, + sent_at, + received_at, + rtt: RttSample { + raw: rtt, + adjusted: None, + effective: SignedDuration::from_duration(rtt), + }, + server_timing: None, + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: PacketMeta::default(), + }); + } + + events +} diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index dbde3aa..ee7e373 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -1,7 +1,24 @@ //! Statistics aggregation for `irtt-client` events. //! -//! The crate consumes `irtt-client` events and produces cumulative or rolling -//! snapshots for reporting and integration code. +//! [`StatsCollector`] consumes [`irtt_client::ClientEvent`] values through +//! [`StatsCollector::process`] and produces a cumulative [`Snapshot`], plus +//! optional count- and time-based rolling snapshots, for reporting and +//! integration code. +//! +//! # Retention modes +//! +//! [`StatsConfig`] selects between two retention modes via [`SampleMode`]. +//! [`StatsConfig::finite`] ([`SampleMode::Exact`]) retains an exact sample per +//! timing metric, so every metric reports an exact median once it has +//! samples; retention grows with the number of probes processed. This is the +//! default and suits a run with a known, bounded duration. +//! [`StatsConfig::continuous`] ([`SampleMode::RunningOnly`]) keeps only +//! running statistics — no exact median — and bounds its adjacent-sequence +//! IPDV tracking, so memory stays bounded for a long-running or unbounded +//! session. Rolling snapshots always use running statistics regardless of +//! this setting. +//! +//! See `examples/` in the repository for a runnable comparison of both modes. #![forbid(unsafe_code)] #![warn(missing_docs)]