diff options
Diffstat (limited to 'crates/arti/src')
| -rw-r--r-- | crates/arti/src/reload_cfg.rs | 301 | ||||
| -rw-r--r-- | crates/arti/src/subcommands/proxy.rs | 4 |
2 files changed, 175 insertions, 130 deletions
diff --git a/crates/arti/src/reload_cfg.rs b/crates/arti/src/reload_cfg.rs index db714d6a8..f5406329b 100644 --- a/crates/arti/src/reload_cfg.rs +++ b/crates/arti/src/reload_cfg.rs @@ -8,7 +8,9 @@ use arti_client::TorClient; use arti_client::config::Reconfigure; use futures::StreamExt; use futures::{FutureExt as _, Stream, select_biased}; -use tor_config::file_watcher::{self, FileEventSender, FileWatcher, FileWatcherBuilder}; +use tor_config::file_watcher::{ + self, FileEventReceiver, FileEventSender, FileWatcher, FileWatcherBuilder, +}; use tor_config::{ConfigurationSource, ConfigurationSources, sources::FoundConfigFiles}; use tor_rtcompat::Runtime; use tor_rtcompat::SpawnExt; @@ -44,80 +46,155 @@ pub(crate) trait ReconfigurableModule: Send + Sync { fn reconfigure(&self, new: &ArtiCombinedConfig) -> anyhow::Result<()>; } -/// Launch a thread to reload our configuration files. -/// -/// If current configuration requires it, watch for changes in `sources` -/// and try to reload our configuration. On unix platforms, also watch -/// for SIGHUP and reload configuration then. -/// -/// The modules are `Weak` references to prevent this background task -/// from keeping them alive. -/// -/// See the [`FileWatcher`](FileWatcher#Limitations) docs for limitations. +/// Structure to reload configuration as necessary. #[cfg_attr(feature = "experimental-api", visibility::make(pub))] -#[instrument(level = "trace", skip_all)] -pub(crate) fn watch_for_config_changes<R: Runtime>( - runtime: &R, +pub(crate) struct CfgMgr<R> { + /// A runtime that we use when constructing [`FileWatcher`]s. + runtime: R, + + /// The sources from which we read our configuration. sources: ConfigurationSources, - config: &ArtiConfig, + + /// A sender to use when constructing new [`FileWatcher`]s. + tx: FileEventSender, + + /// A list of modules to alert whenever the configuration has changed. modules: Vec<Weak<dyn ReconfigurableModule>>, -) -> anyhow::Result<()> { - let watch_file = config.application().watch_configuration; - cfg_if::cfg_if! { - if #[cfg(target_family = "unix")] { - let sighup_stream = sighup_stream()?; - } else { - let sighup_stream = stream::pending(); + /// Mutable state. + inner: Mutex<CfgMgrInner>, +} + +/// Mutable part of a CfgMgr. +struct CfgMgrInner { + /// If present, a [`FileWatcher`] that is currently watching for changes + /// in the configuration files and directories. + watcher: Option<FileWatcher>, +} + +impl<R: Runtime> CfgMgr<R> { + /// Construct a new CfgMgr, and launch a task to watch for any events + /// that mean we have to reload our configuration. + /// + /// If the provided configuration requires it, watch for changes in `sources` + /// and try to reload our configuration. On unix platforms, also watch + /// for SIGHUP and reload configuration then. + /// + /// The modules are `Weak` references to prevent this background task + /// from keeping them alive. + /// + /// See the [`FileWatcher`](FileWatcher#Limitations) docs for limitations. + #[cfg_attr(feature = "experimental-api", visibility::make(pub))] + #[instrument(level = "trace", skip_all)] + pub(crate) fn launch( + runtime: R, + sources: ConfigurationSources, + config: &ArtiConfig, + modules: Vec<Weak<dyn ReconfigurableModule>>, + ) -> anyhow::Result<Arc<Self>> { + let (tx, rx) = file_watcher::channel(); + let mgr = Arc::new(CfgMgr { + runtime, + sources, + tx, + modules, + inner: Mutex::new(CfgMgrInner { watcher: None }), + }); + + cfg_if::cfg_if! { + if #[cfg(target_family = "unix")] { + let sighup_stream = sighup_stream()?; + } else { + let sighup_stream = stream::pending(); + } } + + let rt = mgr.runtime.clone(); + let weak_mgr = Arc::downgrade(&mgr); + mgr.runtime + .spawn(async move { + let res: anyhow::Result<()> = + run_watcher(rt, rx, sighup_stream, weak_mgr, Some(DEBOUNCE_INTERVAL)).await; + match res { + Ok(()) => debug!("Config watcher task exiting"), + // TODO: warn_report does not work on anyhow::Error. + Err(e) => error!("Config watcher task exiting: {}", tor_error::Report(e)), + } + }) + .context("failed to spawn task")?; + + if config.application().watch_configuration { + let (watcher, _files) = mgr.launch_file_watcher()?; + mgr.inner.lock().expect("lock poisoned").watcher = Some(watcher); + } + + Ok(mgr) + } + + /// Create a new [`FileWatcher`] for the files in this configuration. + /// + /// Return it, along with the set of files we found. + /// + /// The caller is responsible for storing the `FileWatcher`; when it is dropped, + /// it stops watching. + fn launch_file_watcher(&self) -> anyhow::Result<(FileWatcher, FoundConfigFiles<'_>)> { + let mut watcher = FileWatcher::builder(self.runtime.clone()); + let found_files = prepare(&mut watcher, &self.sources)?; + let watcher = watcher.start_watching(self.tx.clone())?; + Ok((watcher, found_files)) } - let rt = runtime.clone(); - let () = runtime - .clone() - .spawn(async move { - let res: anyhow::Result<()> = run_watcher( - rt, - sources, - modules, - watch_file, - sighup_stream, - Some(DEBOUNCE_INTERVAL), - ) - .await; - - match res { - Ok(()) => debug!("Config watcher task exiting"), - // TODO: warn_report does not work on anyhow::Error. - Err(e) => error!("Config watcher task exiting: {}", tor_error::Report(e)), + /// Reload the configuration. + #[instrument(level = "trace", skip_all)] + fn reload_configuration(&self) -> anyhow::Result<()> { + let mut inner = self.inner.lock().expect("Lock poisoned"); + + // TODO RPC: Take 'how' as an argument. + let found_files = if inner.watcher.is_some() { + let (watcher, files) = self + .launch_file_watcher() + .context("Failed to re-scan config")?; + inner.watcher = Some(watcher); + files + } else { + self.sources + .scan() + .context("FS watch: failed to rescan config")? + }; + + match reconfigure(found_files, &self.modules) { + Ok(watch) => { + info!("Successfully reloaded configuration."); + if watch && inner.watcher.is_none() { + info!("Starting watching over configuration."); + let (watcher, _files) = self + .launch_file_watcher() + .context("Starting to watch over config")?; + inner.watcher = Some(watcher); + } else if !watch && inner.watcher.is_some() { + info!("Stopped watching over configuration."); + inner.watcher = None; + } } - }) - .context("failed to spawn task")?; + // TODO: warn_report does not work on anyhow::Error. + Err(e) => warn!("Couldn't reload configuration: {}", tor_error::Report(e)), + } - Ok(()) + Ok(()) + } } /// Start watching for configuration changes. /// -/// Spawned from `watch_for_config_changes`. +/// Spawned from [`CfgMgr::launch`]. #[instrument(level = "trace", skip_all)] async fn run_watcher<R: Runtime>( runtime: R, - sources: ConfigurationSources, - modules: Vec<Weak<dyn ReconfigurableModule>>, - watch_file: bool, + mut rx: FileEventReceiver, mut sighup_stream: impl Stream<Item = ()> + Unpin, + weak_mgr: Weak<CfgMgr<R>>, debounce_interval: Option<Duration>, ) -> anyhow::Result<()> { - let (tx, mut rx) = file_watcher::channel(); - let mut watcher = if watch_file { - let mut watcher = FileWatcher::builder(runtime.clone()); - prepare(&mut watcher, &sources)?; - Some(watcher.start_watching(tx.clone())?) - } else { - None - }; - debug!("Entering FS event loop"); loop { @@ -146,60 +223,16 @@ async fn run_watcher<R: Runtime>( }, } - watcher = - reload_configuration(runtime.clone(), watcher, &sources, &modules, tx.clone()).await?; - } - - Ok(()) -} - -/// Reload the configuration. -#[instrument(level = "trace", skip_all)] -async fn reload_configuration<R: Runtime>( - runtime: R, - mut watcher: Option<FileWatcher>, - sources: &ConfigurationSources, - modules: &[Weak<dyn ReconfigurableModule>], - tx: FileEventSender, -) -> anyhow::Result<Option<FileWatcher>> { - // TODO RPC: Take 'how' as an argument. - let found_files = if watcher.is_some() { - let mut new_watcher = FileWatcher::builder(runtime.clone()); - let found_files = prepare(&mut new_watcher, sources) - .context("FS watch: failed to rescan config and re-establish watch")?; - let new_watcher = new_watcher - .start_watching(tx.clone()) - .context("FS watch: failed to start watching config")?; - watcher = Some(new_watcher); - found_files - } else { - sources - .scan() - .context("FS watch: failed to rescan config")? - }; - - match reconfigure(found_files, modules) { - Ok(watch) => { - info!("Successfully reloaded configuration."); - if watch && watcher.is_none() { - info!("Starting watching over configuration."); - let mut new_watcher = FileWatcher::builder(runtime.clone()); - let _found_files = prepare(&mut new_watcher, sources) - .context("FS watch: failed to rescan config and re-establish watch: {}")?; - let new_watcher = new_watcher - .start_watching(tx.clone()) - .context("FS watch: failed to rescan config and re-establish watch: {}")?; - watcher = Some(new_watcher); - } else if !watch && watcher.is_some() { - info!("Stopped watching over configuration."); - watcher = None; - } + if let Some(mgr) = weak_mgr.upgrade() { + mgr.reload_configuration()?; + drop(mgr); + } else { + debug!("Configuration mgr disappeared; exiting loop"); + break; } - // TODO: warn_report does not work on anyhow::Error. - Err(e) => warn!("Couldn't reload configuration: {}", tor_error::Report(e)), } - Ok(watcher) + Ok(()) } /// A TorClient that we may or may not have told to start bootstrapping. @@ -471,22 +504,28 @@ mod test { config_builder.logging().log_sensitive_information(true); let _: PathBuf = write_config(&temp_dir, CONFIG_NAME1, &config_builder); + let (fw_tx, fw_rx) = file_watcher::channel(); + let mgr = Arc::new(CfgMgr { + runtime: rt.clone(), + sources: cfg_sources, + tx: fw_tx, + modules: vec![Arc::downgrade(&module)], + inner: Mutex::new(CfgMgrInner { watcher: None }), + }); + + let (watcher, _) = mgr.launch_file_watcher().unwrap(); + mgr.inner.lock().unwrap().watcher = Some(watcher); + let weak_mgr = Arc::downgrade(&mgr); + // Use a fake sighup stream to wait until run_watcher()'s select_biased! // loop is entered let (mut sighup_tx, sighup_rx) = mpsc::unbounded(); let runtime = rt.clone(); let () = rt .spawn(async move { - run_watcher( - runtime, - cfg_sources, - vec![Arc::downgrade(&module)], - true, - sighup_rx, - None, - ) - .await - .unwrap(); + run_watcher(runtime.clone(), fw_rx, sighup_rx, weak_mgr, None) + .await + .unwrap(); }) .unwrap(); @@ -494,7 +533,6 @@ mod test { // The reconfigurable modules should've been reloaded in response to sighup let config = rx.next().await.unwrap(); - assert_eq!(config.0, config_builder.build().unwrap()); // Overwrite the config @@ -523,22 +561,29 @@ mod test { ); let (module, mut rx) = create_module().await; + + let (fw_tx, fw_rx) = file_watcher::channel(); + let mgr = Arc::new(CfgMgr { + runtime: rt.clone(), + sources: cfg_sources, + tx: fw_tx, + modules: vec![Arc::downgrade(&module)], + inner: Mutex::new(CfgMgrInner { watcher: None }), + }); + + let (watcher, _) = mgr.launch_file_watcher().unwrap(); + mgr.inner.lock().unwrap().watcher = Some(watcher); + let weak_mgr = Arc::downgrade(&mgr); + // Use a fake sighup stream to wait until run_watcher()'s select_biased! // loop is entered let (mut sighup_tx, sighup_rx) = mpsc::unbounded(); let runtime = rt.clone(); let () = rt .spawn(async move { - run_watcher( - runtime, - cfg_sources, - vec![Arc::downgrade(&module)], - true, - sighup_rx, - None, - ) - .await - .unwrap(); + run_watcher(runtime.clone(), fw_rx, sighup_rx, weak_mgr, None) + .await + .unwrap(); }) .unwrap(); diff --git a/crates/arti/src/subcommands/proxy.rs b/crates/arti/src/subcommands/proxy.rs index f996866ea..95a10cdd4 100644 --- a/crates/arti/src/subcommands/proxy.rs +++ b/crates/arti/src/subcommands/proxy.rs @@ -178,8 +178,8 @@ async fn run_proxy<R: ToplevelRuntime>( // NOTE: reconfigurable_modules stores the only strong references to these modules, // so we must keep the variable alive until the end of the function let weak_modules = reconfigurable_modules.iter().map(Arc::downgrade).collect(); - reload_cfg::watch_for_config_changes( - client.runtime(), + let _cfg_mgr = reload_cfg::CfgMgr::launch( + client.runtime().clone(), config_sources, &arti_config, weak_modules, |