| 1 | //! Process-wide shared state. |
| 2 | |
| 3 | use std::path::PathBuf; |
| 4 | use std::sync::Arc; |
| 5 | use std::sync::atomic::{AtomicU64, Ordering}; |
| 6 | use std::time::Duration; |
| 7 | |
| 8 | use arc_swap::ArcSwap; |
| 9 | use tokio::sync::{OwnedSemaphorePermit, Semaphore}; |
| 10 | |
| 11 | use crate::cache::Cache; |
| 12 | use crate::config::{Config, ConfigError, Listen}; |
| 13 | use crate::index::Indexer; |
| 14 | use crate::index::languages::Warmer; |
| 15 | use crate::server::assets::Assets; |
| 16 | |
| 17 | pub struct AppState { |
| 18 | /// Current config; swapped atomically on reload. |
| 19 | pub config: ArcSwap<Config>, |
| 20 | /// Bumped on every successful reload; part of page ETags. |
| 21 | pub config_generation: AtomicU64, |
| 22 | pub assets: Assets, |
| 23 | pub cache: Cache, |
| 24 | /// Bounds concurrent git work (`limits.git_parallel`). |
| 25 | pub git_permits: Arc<Semaphore>, |
| 26 | /// Bounds expensive requests: blame, archives, clones, log searches |
| 27 | /// (`limits.heavy_parallel`). Held for a whole response stream. |
| 28 | pub heavy_permits: Arc<Semaphore>, |
| 29 | /// Bounds requests in flight (`limits.max_requests`). |
| 30 | pub request_permits: Arc<Semaphore>, |
| 31 | pub index: Indexer, |
| 32 | /// Background language statistics. |
| 33 | pub warmer: Warmer, |
| 34 | /// Wakes the index watcher for an immediate full rescan. |
| 35 | pub rescan: tokio::sync::Notify, |
| 36 | /// Command-line overrides re-applied after every reload. |
| 37 | pub overrides: Overrides, |
| 38 | } |
| 39 | |
| 40 | #[derive(Debug, Clone, Default)] |
| 41 | pub struct Overrides { |
| 42 | pub config_path: Option<PathBuf>, |
| 43 | pub listen: Option<Listen>, |
| 44 | } |
| 45 | |
| 46 | impl Overrides { |
| 47 | pub fn load(&self) -> Result<Config, ConfigError> { |
| 48 | let mut config = Config::load(self.config_path.as_deref())?; |
| 49 | if let Some(listen) = &self.listen { |
| 50 | config.server.listen = listen.clone(); |
| 51 | } |
| 52 | Ok(config) |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | impl AppState { |
| 57 | pub fn new(config: Config, overrides: Overrides) -> Arc<Self> { |
| 58 | Arc::new(AppState { |
| 59 | cache: Cache::new(config.cache.size_mb), |
| 60 | git_permits: Arc::new(Semaphore::new(config.git_parallel())), |
| 61 | heavy_permits: Arc::new(Semaphore::new(config.limits.heavy_parallel)), |
| 62 | request_permits: Arc::new(Semaphore::new(config.limits.max_requests)), |
| 63 | config_generation: AtomicU64::new(0), |
| 64 | config: ArcSwap::from_pointee(config), |
| 65 | assets: Assets::new(), |
| 66 | index: Indexer::new(), |
| 67 | warmer: Warmer::default(), |
| 68 | rescan: tokio::sync::Notify::new(), |
| 69 | overrides, |
| 70 | }) |
| 71 | } |
| 72 | |
| 73 | /// Re-read the config file and swap it in. On error the running config |
| 74 | /// stays in place. Returns the new config. |
| 75 | pub fn reload(&self) -> Result<Arc<Config>, ConfigError> { |
| 76 | let mut new = self.overrides.load()?; |
| 77 | let old = self.config.load(); |
| 78 | // Settings baked into the listener and middleware stack keep their |
| 79 | // running values so the config always describes the live server. |
| 80 | if new.server.listen != old.server.listen { |
| 81 | tracing::warn!( |
| 82 | "server.listen changed to {}; restart required to take effect", |
| 83 | new.server.listen |
| 84 | ); |
| 85 | new.server.listen = old.server.listen.clone(); |
| 86 | } |
| 87 | if new.server.base_path != old.server.base_path { |
| 88 | tracing::warn!("server.base_path changed; restart required to take effect"); |
| 89 | new.server.base_path = old.server.base_path.clone(); |
| 90 | } |
| 91 | if new.server.compression != old.server.compression { |
| 92 | tracing::warn!("server.compression changed; restart required"); |
| 93 | new.server.compression = old.server.compression; |
| 94 | } |
| 95 | if new.cache.size_mb != old.cache.size_mb |
| 96 | || new.git_parallel() != old.git_parallel() |
| 97 | || new.limits.heavy_parallel != old.limits.heavy_parallel |
| 98 | || new.limits.max_requests != old.limits.max_requests |
| 99 | { |
| 100 | tracing::warn!( |
| 101 | "cache.size_mb / limits.git_parallel / limits.heavy_parallel / limits.max_requests \ |
| 102 | changed; restart required" |
| 103 | ); |
| 104 | new.limits.max_requests = old.limits.max_requests; |
| 105 | new.cache.size_mb = old.cache.size_mb; |
| 106 | new.limits.git_parallel = old.limits.git_parallel; |
| 107 | new.limits.heavy_parallel = old.limits.heavy_parallel; |
| 108 | } |
| 109 | let new = Arc::new(new); |
| 110 | self.config.store(new.clone()); |
| 111 | self.config_generation.fetch_add(1, Ordering::Relaxed); |
| 112 | Ok(new) |
| 113 | } |
| 114 | |
| 115 | /// A heavy-request permit, waiting at most `limits.heavy_wait_secs`. |
| 116 | /// `None` when the server is too busy (or shutting down). |
| 117 | pub async fn heavy_permit(&self) -> Option<OwnedSemaphorePermit> { |
| 118 | let wait = Duration::from_secs(self.config.load().limits.heavy_wait_secs); |
| 119 | match tokio::time::timeout(wait, self.heavy_permits.clone().acquire_owned()).await { |
| 120 | Ok(Ok(permit)) => Some(permit), |
| 121 | _ => None, |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | /// Run blocking git work on the blocking pool, at most |
| 126 | /// `limits.git_parallel` jobs at a time. |
| 127 | pub async fn blocking<T: Send + 'static>( |
| 128 | self: &Arc<Self>, |
| 129 | work: impl FnOnce(&AppState) -> T + Send + 'static, |
| 130 | ) -> anyhow::Result<T> { |
| 131 | // The permit moves into the job: a timed-out request stops waiting, |
| 132 | // but the job keeps its slot until it really finishes. |
| 133 | let permit = self.git_permits.clone().acquire_owned().await?; |
| 134 | let state = self.clone(); |
| 135 | Ok(tokio::task::spawn_blocking(move || { |
| 136 | let _permit = permit; |
| 137 | work(&state) |
| 138 | }) |
| 139 | .await?) |
| 140 | } |
| 141 | } |