diff --git a/Cargo.lock b/Cargo.lock index c996483..f057229 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,9 +4,9 @@ version = 4 [[package]] name = "aho-corasick" -version = "1.1.4" +version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" dependencies = [ "memchr", ] @@ -56,9 +56,18 @@ dependencies = [ [[package]] name = "bitflags" -version = "2.13.1" +version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] [[package]] name = "bumpalo" @@ -74,9 +83,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.3.0" +version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8" +checksum = "f74872d07caf508b30a21f6836e7d7016a2eaf7d9ff4f48deaa58cd8a0407630" dependencies = [ "find-msvc-tools", "shlex 2.0.1", @@ -93,9 +102,9 @@ dependencies = [ [[package]] name = "cfg-if" -version = "1.0.4" +version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" [[package]] name = "cfg_aliases" @@ -105,9 +114,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" [[package]] name = "clang-sys" -version = "1.8.1" +version = "1.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" +checksum = "157a8ba7b480713b56f4c09fd13fc3e0a22a5dfab8097ba61cbc5feef950788a" dependencies = [ "glob", "libc", @@ -123,6 +132,12 @@ dependencies = [ "cc", ] +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + [[package]] name = "config" version = "0.15.27" @@ -135,6 +150,21 @@ dependencies = [ "winnow", ] +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + [[package]] name = "csv" version = "1.4.0" @@ -157,10 +187,31 @@ dependencies = [ ] [[package]] -name = "either" -version = "1.16.0" +name = "ctutils" +version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +checksum = "03bb0e1cc970d482d121d9a1744999169b69a07470b3d644a7894e53fcaf4574" +dependencies = [ + "cmov", +] + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", + "ctutils", +] + +[[package]] +name = "either" +version = "1.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be" [[package]] name = "equivalent" @@ -189,9 +240,9 @@ dependencies = [ [[package]] name = "find-msvc-tools" -version = "0.1.9" +version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +checksum = "aedcfb3409746eddb02b9e19ebda1c3394f759a152e48ee875a0844d1b955484" [[package]] name = "flume" @@ -213,9 +264,9 @@ checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" [[package]] name = "futures-sink" -version = "0.3.32" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" [[package]] name = "getrandom" @@ -232,9 +283,9 @@ dependencies = [ [[package]] name = "glob" -version = "0.3.3" +version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280" +checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b" [[package]] name = "hashbrown" @@ -242,6 +293,24 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest", +] + +[[package]] +name = "hybrid-array" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" +dependencies = [ + "typenum", +] + [[package]] name = "icmp-socket2" version = "0.6.4" @@ -257,14 +326,39 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.14.0" +version = "2.14.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855" dependencies = [ "equivalent", "hashbrown", ] +[[package]] +name = "irtt-client" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a59c24704bd94f19b84b2b0190aa9163efede499f8cf4eb02922443cfee08339" +dependencies = [ + "irtt-proto", + "nix", + "socket2", + "thiserror", + "tokio", +] + +[[package]] +name = "irtt-proto" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a56ff307983b8f6477724e08cf726c5e24975b9affeb853d0830d91c60026ab2" +dependencies = [ + "hmac", + "md-5", + "subtle", + "thiserror", +] + [[package]] name = "itertools" version = "0.13.0" @@ -282,9 +376,9 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "js-sys" -version = "0.3.103" +version = "0.3.106" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" +checksum = "7883d941dae510fb2d978fc3fe018c71c9e2892fd38854de3e8b92c2e5ad9cc5" dependencies = [ "cfg-if", "wasm-bindgen", @@ -337,6 +431,16 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest", +] + [[package]] name = "memchr" version = "2.8.3" @@ -507,9 +611,9 @@ dependencies = [ [[package]] name = "regex-automata" -version = "0.4.16" +version = "0.4.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" dependencies = [ "aho-corasick", "memchr", @@ -615,9 +719,9 @@ dependencies = [ [[package]] name = "serde_spanned" -version = "1.1.1" +version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +checksum = "4e7523beb55eece201a2356bee0bbca0d1ab466c14c07703b2e0ee6d42cb0c2c" dependencies = [ "serde_core", ] @@ -679,6 +783,7 @@ dependencies = [ "fastrand", "flume", "icmp-socket2", + "irtt-client", "libc", "log", "netlink-bindings", @@ -697,6 +802,12 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "85ec932792e62b2d3936163eacbd2e09984453bcd114f665c1e1694e36ff59c9" +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.119" @@ -784,9 +895,9 @@ dependencies = [ [[package]] name = "toml" -version = "1.1.3+spec-1.1.0" +version = "1.1.7+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53c96ecdfa941c8fc4fcaed14f99ada8ebed502eef533015095a07e3301d4c3c" +checksum = "7874cd34dd99040fc24e30923aa64a60a81ff3ddfe2b8129ccc84ba24949deed" dependencies = [ "indexmap", "serde_core", @@ -799,27 +910,27 @@ dependencies = [ [[package]] name = "toml_datetime" -version = "1.1.1+spec-1.1.0" +version = "1.1.2+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7" +checksum = "2b86d767906c6c42421dcba507eb9d203e779497710a47782a224bb871653053" dependencies = [ "serde_core", ] [[package]] name = "toml_parser" -version = "1.1.2+spec-1.1.0" +version = "1.1.4+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526" +checksum = "bed810ee007d8e7f9cf20b4774e7198142fa27186e0f805871da7f3c39eaf333" dependencies = [ "winnow", ] [[package]] name = "toml_writer" -version = "1.1.2+spec-1.1.0" +version = "1.1.3+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" +checksum = "06bdbd8cfc056b8d2e2e85f29b56a3bdbecb527cef81eb39e3e7b98af4652770" [[package]] name = "trybuild" @@ -836,6 +947,12 @@ dependencies = [ "toml", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "uci-derive" version = "0.1.0" @@ -848,9 +965,9 @@ dependencies = [ [[package]] name = "unicode-ident" -version = "1.0.24" +version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" [[package]] name = "wasi" @@ -860,9 +977,9 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" [[package]] name = "wasm-bindgen" -version = "0.2.126" +version = "0.2.129" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" +checksum = "9bb54f33acc68fd454578d9820b0bde1a1a3d17aa17bb7b6595806d02886d409" dependencies = [ "cfg-if", "once_cell", @@ -873,9 +990,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.126" +version = "0.2.129" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" +checksum = "2e29d0c35b16e224a7eeb5cd2d25e3e1968fbd65604117b44d3b789d00ee8535" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -883,22 +1000,22 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.126" +version = "0.2.129" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" +checksum = "6f501a8bc3719dba86ef8ae4728879c08001bea749eb1333ac5b91e040e2a6b7" dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.6", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-shared" -version = "0.2.126" +version = "0.2.129" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" +checksum = "23f0c9c52aa7cd7d77769a4cfe2a9adb1b331f489a41d912ce14513d5ab995c6" dependencies = [ "unicode-ident", ] diff --git a/Cargo.toml b/Cargo.toml index cd61932..179e96c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ csv = "1.4.0" fastrand = "2" flume = { version = "0.12.0", default-features = false, features = ["async", "eventual-fairness", "select"] } icmp-socket2 = { version = "0.6.4", features = ["tokio"] } +irtt-client = { version = "0.6.0", optional = true, features = ["ancillary", "tokio"] } libc = "0.2" rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] } serde = { version = "1", features = ["derive"] } @@ -44,7 +45,8 @@ netlink-socket2 = { version = "0.3.3" } rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "fcfedd204fb76928ef617bf9631b041c816e03d4", optional = true, default-features = false, } [features] -default = ["toml", "uci-vendored"] +default = ["irtt", "toml", "uci-vendored"] +irtt = ["dep:irtt-client"] toml = ["config/toml"] uci = ["dep:rust-uci"] uci-vendored = ["uci", "rust-uci/vendored"] diff --git a/src/main.rs b/src/main.rs index 560cab9..f252695 100644 --- a/src/main.rs +++ b/src/main.rs @@ -22,6 +22,7 @@ mod util; use crate::baseliner::Baseliner; use crate::controller::Controller; use crate::measurement::icmp::IcmpEngine; +use crate::measurement::irtt::IrttEngine; use crate::metrics::{Metric, Metrics, MetricsSender}; use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations}; use crate::ratecontroller::Ratecontroller; @@ -236,6 +237,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded(); let (error_tx, error_rx) = flume::unbounded::(); let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded(); + let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded(); let (measurement_tx, measurement_rx) = flume::unbounded(); let (reflector_tx, reflector_rx) = flume::unbounded(); let (reselect_tx, reselect_rx) = flume::bounded(1); @@ -247,17 +249,22 @@ fn run(settings: &Settings) -> anyhow::Result<()> { }; let controller = Controller::new(reflector_rx, icmp_binding_tx); - let engine = { + let icmp_engine = { let _guard = rt.enter(); IcmpEngine::new( start_time, &settings.network, - measurement_tx, + measurement_tx.clone(), icmp_binding_rx, )? }; + let (_irtt_task, irtt_engine) = { + let _guard = rt.enter(); + IrttEngine::new(start_time, measurement_tx, irtt_binding_rx)? + }; - spawn_task(&rt, &error_tx, engine.run()); + spawn_task(&rt, &error_tx, icmp_engine.run()); + spawn_task(&rt, &error_tx, irtt_engine.run()); let dropped = Arc::new(AtomicU32::new(0)); diff --git a/src/measurement/irtt.rs b/src/measurement/irtt.rs new file mode 100644 index 0000000..77c7f8d --- /dev/null +++ b/src/measurement/irtt.rs @@ -0,0 +1,343 @@ +use std::{collections::HashMap, net::SocketAddr, time::Instant}; + +use flume::{Receiver, Sender}; +use irtt_client::{ + ClientConfig, ClientEvent, + managed::{ + ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, + ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError, + ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent, + ManagedLifecycle, ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason, + TargetId, + }, +}; +use log::{debug, trace, warn}; +use thiserror::Error; +use tokio::sync::broadcast::error::RecvError; + +use crate::measurement::model::{ + MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementStream, + MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule, SignedDuration, +}; + +#[derive(Debug, Error)] +pub enum IrttError { + #[error("failed to build IRTT config")] + InvalidConfig(#[from] ManagedConfigError), + + #[error("event channel closed")] + EventChannelClosed, + + #[error("binding channel closed")] + BindingChannelClosed, + + #[error("error while reciving on channel")] + ChannelRecvError(#[from] RecvError), + + #[error("unable to subscribe to irtt events")] + SubscriptionError(#[from] ManagedSubscribeError), + + #[error("failed to submit IRTT target update")] + CommandError(#[from] ManagedCommandError), + + #[error("failed to apply IRTT target update")] + CommandApplyError(#[from] ManagedCommandApplyError), + + #[error("IRTT managed client failed: {0:?}")] + ManagedFailure(ManagedEndReason), + + #[error("IRTT managed task was abandoned")] + ManagedAbandoned, +} + +pub struct IrttBinding { + stream: MeasurementStream, + schedule: ProbeSchedule, + port: u16, +} + +impl IrttBinding { + pub fn new(stream: MeasurementStream, schedule: ProbeSchedule, port: u16) -> Self { + Self { + stream, + schedule, + port, + } + } +} + +pub struct IrttEngine { + client: ManagedClientHandle, + + bindings: HashMap, + + origin: Instant, + + binding_rx: Receiver>, + observation_tx: Sender, +} + +impl IrttEngine { + pub fn new( + origin: Instant, + observation_tx: Sender, + binding_rx: Receiver>, + ) -> Result<(ManagedClientTask, Self), IrttError> { + let config = ManagedClientConfig { + client: ClientConfig { + request: irtt_client::SessionRequest { + duration: None, + ..Default::default() + }, + ..Default::default() + }, + completion: ManagedCompletionPolicy::ExplicitStop, + ..Default::default() + }; + let (task, handle) = ManagedClient::task(config, Vec::new())?; + + Ok(( + task, + Self { + client: handle, + bindings: HashMap::new(), + origin, + binding_rx, + observation_tx, + }, + )) + } + + async fn set_bindings(&mut self, bindings: Vec) -> Result<(), IrttError> { + let targets = bindings + .iter() + .map(|binding| { + ManagedTargetConfig::new( + TargetId::new(binding.stream.id.0.to_string()), + SocketAddr::new(binding.stream.peer, binding.port).to_string(), + ) + }) + .collect(); + + self.client.update_targets(targets)?.await?; + + self.bindings = bindings + .into_iter() + .map(|binding| (binding.stream.id, binding)) + .collect(); + + Ok(()) + } + + pub async fn run(mut self) -> Result<(), IrttError> { + let mut events = self.client.subscribe()?; + + loop { + tokio::select! { + bindings = self.binding_rx.recv_async() => { + let bindings = bindings + .map_err(|_| IrttError::BindingChannelClosed)?; + + self.set_bindings(bindings).await?; + } + + event = events.recv() => { + let event = match event { + Ok(event) => event, + + Err(RecvError::Lagged(n)) => { + warn!("IRTT event receiver dropped {n} events"); + continue; + } + + Err(RecvError::Closed) => { + let status = self.client.status(); + + match status.lifecycle { + ManagedLifecycle::Completed => return Ok(()), + + ManagedLifecycle::Failed => { + if let Some(outcome) = &status.final_outcome { + return Err(IrttError::ManagedFailure( + outcome.end_reason.clone() + )); + } + + return Err(IrttError::ManagedAbandoned); + } + + _ => return Err(IrttError::ManagedAbandoned), + } + } + }; + + match event { + ManagedEvent::Started => { + debug!("IRTT managed client started"); + } + + ManagedEvent::TargetStateChanged { + target, + lifecycle, + } => { + debug!("IRTT target {target:?}: {lifecycle:?}"); + } + + ManagedEvent::Client { target, event } => { + let stream = target.id.as_str() + .parse::() + .ok() + .and_then(|id| { + self.bindings.get(&MeasurementStreamId(id)) + }) + .map(|binding| binding.stream); + + match event { + ClientEvent::SessionStarted(session) => { + debug!( + "IRTT session started for {}: {}", + target.id, + session.remote + ); + } + + ClientEvent::NoTestCompleted(outcome) => { + debug!( + "IRTT no-test completed for {}: {}", + target.id, + outcome.remote + ); + } + + ClientEvent::SessionClosed { .. } => { + debug!("IRTT session closed for {}", target.id); + } + + ClientEvent::EchoSent { seq, .. } => { + trace!("IRTT {} sent probe {seq}", target.id); + } + + ClientEvent::EchoReply { + sent_at, + received_at, + one_way, + .. + } => { + let Some(stream) = stream else { + continue; + }; + + let observation = MeasurementObservation { + stream, + started_at: sent_at.mono, + observed_at: received_at.mono, + one_way: one_way.and_then(map_one_way), + }; + + self.observation_tx + .send_async(MeasurementEvent::Observation(observation)) + .await + .map_err(|_| IrttError::EventChannelClosed)?; + } + + ClientEvent::EchoLoss { + sent_at, + timeout_at, + .. + } => { + let Some(stream) = stream else { + continue; + }; + + let loss = MeasurementLoss { + stream, + started_at: sent_at.mono, + detected_at: timeout_at, + }; + + self.observation_tx + .send_async(MeasurementEvent::Loss(loss)) + .await + .map_err(|_| IrttError::EventChannelClosed)?; + } + + ClientEvent::DuplicateReply { seq, .. } => { + trace!( + "IRTT {} duplicate reply {seq}", + target.id + ); + } + + ClientEvent::LateReply { seq, .. } => { + trace!( + "IRTT {} late reply {seq}", + target.id + ); + } + + ClientEvent::Warning { kind, message, .. } => { + debug!( + "IRTT {} warning ({kind:?}): {message}", + target.id + ); + } + } + } + + ManagedEvent::TargetFinished { outcome } => { + match &outcome.end_reason { + ManagedTargetEndReason::Failed(failure) => { + warn!( + "IRTT target {} failed: {failure:?}", + outcome.target.id + ); + } + + reason => { + debug!( + "IRTT target {} finished: {reason:?}", + outcome.target.id + ); + } + } + } + + ManagedEvent::Stopping => { + debug!("IRTT managed client stopping"); + } + + ManagedEvent::Completed { outcome } => { + debug!( + "IRTT managed client completed: {:?}", + outcome.end_reason + ); + return Ok(()); + } + + ManagedEvent::Failed { outcome } => { + return Err(IrttError::ManagedFailure( + outcome.end_reason.clone() + )); + } + + ManagedEvent::Abandoned => { + return Err(IrttError::ManagedAbandoned); + } + } + } + } + } + } +} + +fn map_one_way(sample: irtt_client::OneWayDelaySample) -> Option { + let uplink = i64::try_from(sample.client_to_server?.as_nanos()).ok()?; + + let downlink = i64::try_from(sample.server_to_client?.as_nanos()).ok()?; + + Some(OneWayLatency { + uplink: SignedDuration::from_nanos(uplink), + downlink: SignedDuration::from_nanos(downlink), + clock: OneWayClock::Uncalibrated, + }) +} diff --git a/src/measurement/mod.rs b/src/measurement/mod.rs index 0ee60f2..836dd96 100644 --- a/src/measurement/mod.rs +++ b/src/measurement/mod.rs @@ -1,2 +1,4 @@ pub(crate) mod icmp; +#[cfg(feature = "irtt")] +pub(crate) mod irtt; pub(crate) mod model;