diff --git a/src/controller.rs b/src/controller.rs index 258c414..6a131bf 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -50,6 +50,7 @@ impl Controller { active_streams: Arc>>, endpoint_streams: Arc>>, source: MeasurementSource, + period: Duration, ) -> Self { Self { reflector_rx, @@ -60,7 +61,7 @@ impl Controller { active_streams, endpoint_streams, source, - period: Duration::from_millis(500), + period, next_stream_id: 0, stream_ids: HashMap::new(), } diff --git a/src/main.rs b/src/main.rs index cef3c66..8cfa5e6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -174,16 +174,55 @@ fn resolve_reflectors( } else { 0 }; - for endpoint in (definition.host.as_str(), port).to_socket_addrs()? { + let endpoints = match (definition.host.as_str(), port).to_socket_addrs() { + Ok(endpoints) => endpoints, + Err(error) => { + warn!("Failed to resolve reflector {}: {error}", definition.host); + continue; + } + }; + for endpoint in endpoints { if protocol != ReflectorProtocol::Irtt && endpoint.is_ipv6() { continue; } - resolved.insert(endpoint, definition.clone()); + insert_reflector(&mut resolved, endpoint, definition.clone())?; } } Ok(resolved) } +fn insert_reflector( + reflectors: &mut HashMap, + endpoint: SocketAddr, + definition: ReflectorDefinition, +) -> anyhow::Result<()> { + if let Some(existing) = reflectors.get(&endpoint) { + let same_country = match (&existing.country, &definition.country) { + (Some(a), Some(b)) => a.eq_ignore_ascii_case(b), + (None, None) => true, + _ => false, + }; + let same_protocols = existing + .protocols + .iter() + .all(|p| definition.protocols.contains(p)) + && definition + .protocols + .iter() + .all(|p| existing.protocols.contains(p)); + if !same_country + || !same_protocols + || existing.irtt_port.unwrap_or(2112) != definition.irtt_port.unwrap_or(2112) + || existing.irtt_hmac != definition.irtt_hmac + { + anyhow::bail!("conflicting reflector definitions for resolved endpoint {endpoint}"); + } + } else { + reflectors.insert(endpoint, definition); + } + Ok(()) +} + fn setup_reflectors(settings: &Settings) -> anyhow::Result { let protocol = settings.advanced_settings.measurement_type; let configured = resolve_reflectors( @@ -194,11 +233,12 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result { )?; let bootstrap = resolve_reflectors(&settings.reflectors.bootstrap, protocol, &[], false)?; let mut reflectors = bootstrap.clone(); - reflectors.extend( - configured - .iter() - .map(|(endpoint, definition)| (*endpoint, definition.clone())), - ); + for (endpoint, definition) in &configured { + insert_reflector(&mut reflectors, *endpoint, definition.clone())?; + } + if reflectors.is_empty() { + anyhow::bail!("no usable reflectors for measurement protocol {protocol}"); + } let mut pool = configured.keys().copied().collect::>(); pool.sort(); let mut initial = bootstrap.keys().copied().collect::>(); @@ -332,6 +372,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { Arc::clone(&active_streams), Arc::clone(&endpoint_streams), source, + Duration::from_secs_f64(settings.advanced_settings.tick_interval), ); let icmp_engine = { let _guard = rt.enter(); diff --git a/src/measurement/irtt.rs b/src/measurement/irtt.rs index 665a33d..226e77e 100644 --- a/src/measurement/irtt.rs +++ b/src/measurement/irtt.rs @@ -5,10 +5,9 @@ use irtt_client::{ ClientConfig, ClientEvent, NegotiationPolicy, OpenPolicy, managed::{ ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, - ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError, - ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent, - ManagedLifecycle, ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason, - TargetAuth, TargetId, + ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError, + ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError, + ManagedTargetConfig, ManagedTargetEndReason, TargetAuth, TargetId, }, }; use log::{debug, trace, warn}; @@ -139,17 +138,34 @@ impl IrttEngine { pub async fn run(mut self, task: ManagedClientTask) -> Result<(), IrttError> { let mut task = Box::pin(task); - let mut events = self.client.subscribe()?; + let client = self.client.clone(); + // Keep driving the client even while the event loop awaits command receipts. + let result = tokio::select! { + biased; + outcome = task.as_mut() => { + return match outcome.end_reason { + ManagedEndReason::StopRequested => Ok(()), + end_reason => Err(IrttError::ManagedFailure(end_reason)), + }; + } + result = self.run_events() => result, + }; + + // Every event-loop exit stops and joins the driver, including channel errors. + let stop = client.stop(); + let (outcome, _) = tokio::join!(task, stop); + result?; + match outcome.end_reason { + ManagedEndReason::StopRequested => Ok(()), + end_reason => Err(IrttError::ManagedFailure(end_reason)), + } + } + + async fn run_events(&mut self) -> Result<(), IrttError> { + let mut events = self.client.subscribe()?; loop { tokio::select! { - outcome = task.as_mut() => { - return match outcome.end_reason { - ManagedEndReason::StopRequested => Ok(()), - end_reason => Err(IrttError::ManagedFailure(end_reason)), - }; - } - bindings = self.binding_rx.recv_async() => { let bindings = bindings .map_err(|_| IrttError::BindingChannelClosed)?; @@ -344,22 +360,7 @@ impl IrttEngine { _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => { if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) { - let stop = self.client.stop(); - tokio::pin!(stop); - tokio::select! { - _ = &mut stop => {}, - outcome = task.as_mut() => { - return match outcome.end_reason { - ManagedEndReason::StopRequested => Ok(()), - end_reason => Err(IrttError::ManagedFailure(end_reason)), - }; - } - } - let outcome = task.await; - return match outcome.end_reason { - ManagedEndReason::StopRequested => Ok(()), - end_reason => Err(IrttError::ManagedFailure(end_reason)), - }; + return Ok(()); } } } diff --git a/src/settings/mod.rs b/src/settings/mod.rs index a59aabd..5bc4ab7 100644 --- a/src/settings/mod.rs +++ b/src/settings/mod.rs @@ -178,7 +178,22 @@ impl Settings { let builder = Config::builder(); #[cfg(all(feature = "uci", unix))] - let builder = builder.add_source(UciSource::from_schema::().required(false)); + let builder = builder.add_source( + UciSource::from_schema::() + .with_repeated_schema(&[ + uci_schema::UciRepeatedSectionMapping { + uci_section: "reflector", + config_path: "reflectors.entries", + options: ::UCI_OPTIONS, + }, + uci_schema::UciRepeatedSectionMapping { + uci_section: "bootstrap_reflector", + config_path: "reflectors.bootstrap", + options: ::UCI_OPTIONS, + }, + ]) + .required(false), + ); #[cfg(feature = "toml")] let builder = builder.add_source(