From 7eaf36692ff3a115485ee9290acebec946c6d34f Mon Sep 17 00:00:00 2001 From: rohit Date: Tue, 3 Nov 2026 09:30:05 +0000 Subject: [PATCH] fix(syntax): keep pack loads off the grammar store lock Every lookup held the store's state lock while it resolved a tag, which covers hashing a pack's files, opening its library and compiling its queries. The diff view asks the store for a language's status on the UI thread when a highlight lands, so a frame could wait for a different language's pack to load on the worker. If the fetch thread failed to spawn, queuing re-locked the state that lookup still held. Resolving now runs under a lock of its own and touches the state lock only for map updates, so resolved tags answer at once while one tag still loads once. The diff view reads the resolved status without resolving, the catalog checks installed files outside the lock, and a failed fetch-thread spawn is reported to the caller, which records the tag as unavailable. --- crates/diffy-syntax/PACKS.md | 2 + crates/diffy-syntax/src/download.rs | 80 +++++---- crates/diffy-syntax/src/store.rs | 209 +++++++++++++++++++----- crates/diffy-ui/src/diff_view/syntax.rs | 5 +- 4 files changed, 216 insertions(+), 80 deletions(-) diff --git a/crates/diffy-syntax/PACKS.md b/crates/diffy-syntax/PACKS.md index d61e1843..36d88e07 100644 --- a/crates/diffy-syntax/PACKS.md +++ b/crates/diffy-syntax/PACKS.md @@ -80,6 +80,8 @@ so an unknown language costs one map lookup afterwards. - `HighlightWorker` answers a pending language with a plain result marked `pending`, then sends the real result for the same slot and generation. +- `GrammarStore::resolved` reports what lookups found without resolving, + and never waits for a pack that is loading, so the UI thread may call it. ### Enclosing scopes diff --git a/crates/diffy-syntax/src/download.rs b/crates/diffy-syntax/src/download.rs index 047fda2a..ecd24a8c 100644 --- a/crates/diffy-syntax/src/download.rs +++ b/crates/diffy-syntax/src/download.rs @@ -140,20 +140,23 @@ impl Remote { } } +/// The cached index's packs, each with its directory and whether all its +/// files are there; the files are checked without the state lock. pub(crate) fn packs( downloads: &Downloads, - state: &mut State, + inner: &Inner, ) -> Vec<(PathBuf, Arc, bool)> { - let cached = state + let cached = lock(&inner.state) .remote .cached - .get_or_insert_with(|| read_cached_index(downloads)); + .get_or_insert_with(|| read_cached_index(downloads)) + .clone(); cached - .iter() + .into_iter() .map(|manifest| { - let dir = downloads.pack_dir(manifest); + let dir = downloads.pack_dir(&manifest); let installed = manifest.files().all(|f| dir.join(&f.path).is_file()); - (dir, manifest.clone(), installed) + (dir, manifest, installed) }) .collect() } @@ -172,18 +175,19 @@ pub(crate) fn ships_tags(downloads: &Downloads, state: &mut State, tag: &str) -> find(cached, tag).is_some_and(|manifest| manifest.tags.is_some()) } -/// Resolves `tag` from the cache, or queues it for the fetch thread. -pub(crate) fn resolve( - inner: &Arc, - downloads: &Downloads, - state: &mut State, - tag: &str, -) -> Tag { - let cached = state - .remote - .cached - .get_or_insert_with(|| read_cached_index(downloads)); - let entry = find(cached, tag).cloned(); +/// Resolves `tag` from the cache, or queues it for the fetch thread, and +/// records the outcome. Cached files are checked and loaded without the +/// state lock. +pub(crate) fn resolve(inner: &Arc, downloads: &Downloads, tag: &str) -> Tag { + let (entry, fetched) = { + let mut state = lock(&inner.state); + let cached = state + .remote + .cached + .get_or_insert_with(|| read_cached_index(downloads)); + let entry = find(cached, tag).cloned(); + (entry, state.remote.fetched) + }; if let Some(manifest) = &entry { let dir = downloads.pack_dir(manifest); // Files that are all present and intact decide the outcome: a pack @@ -192,16 +196,26 @@ pub(crate) fn resolve( if manifest.files().all(|f| dir.join(&f.path).is_file()) && pack::verify_files(&dir, manifest).is_ok() { - return load(state, &dir, manifest); + let found = load(inner, &dir, manifest); + return inner.record(tag, found); } } + // Held while queuing, so the tag is recorded as pending before the + // fetch thread can lock the state to record its answer. + let mut state = lock(&inner.state); // After this session's fetch the cached index is the fresh one, so a - // tag it lacks is unknown. - if state.remote.fetched && entry.is_none() { - return Tag::Unavailable; - } - queue(inner, tag); - Tag::Pending + // tag it lacks is unknown. `fetched` is read with `entry`: the fetch + // may have replaced the index since. + let found = if fetched && entry.is_none() { + Tag::Unavailable + } else if queue(inner, tag) { + Tag::Pending + } else { + // Without the fetch thread it would stay pending forever. + Tag::Unavailable + }; + state.tags.insert(tag.to_owned(), found.clone()); + found } fn read_cached_index(downloads: &Downloads) -> Vec> { @@ -217,12 +231,14 @@ fn read_cached_index(downloads: &Downloads) -> Vec> { } } -fn queue(inner: &Arc, tag: &str) { +/// Hands `tag` to the fetch thread, starting it if needed. Returns whether +/// it was queued: `false` when the thread could not be spawned. +fn queue(inner: &Arc, tag: &str) -> bool { let mut fetcher = lock(&inner.fetcher); if let Some(sender) = &*fetcher && sender.send(tag.to_owned()).is_ok() { - return; + return true; } let (sender, receiver) = channel(); let _ = sender.send(tag.to_owned()); @@ -230,15 +246,11 @@ fn queue(inner: &Arc, tag: &str) { let spawned = std::thread::Builder::new() .name("diffy-syntax-fetch".to_owned()) .spawn(move || fetch_loop(&weak, &receiver)); - if spawned.is_ok() { + let queued = spawned.is_ok(); + if queued { *fetcher = Some(sender); - } else { - // Without the thread the tag stays pending forever; give up on it. - drop(fetcher); - lock(&inner.state) - .tags - .insert(tag.to_owned(), Tag::Unavailable); } + queued } /// Serves queued tags until every store handle is gone, which drops the diff --git a/crates/diffy-syntax/src/store.rs b/crates/diffy-syntax/src/store.rs index ce330d21..e60bbe1a 100644 --- a/crates/diffy-syntax/src/store.rs +++ b/crates/diffy-syntax/src/store.rs @@ -93,6 +93,8 @@ pub enum LanguageStatus { /// [`GrammarStore::none`] (the default) has no grammars: everything is /// plain. Lookups may read and load packs from disk, so they belong off the /// UI thread; [`crate::HighlightWorker`] makes them on its own thread. +/// [`GrammarStore::resolved`] only reads what lookups found, so the UI +/// thread may call it. #[derive(Clone, Default)] pub struct GrammarStore { #[cfg(feature = "runtime")] @@ -117,6 +119,7 @@ impl GrammarStore { inner: Some(Arc::new(Inner { config, state: Mutex::default(), + resolving: Mutex::default(), listeners: Mutex::default(), status_listeners: Mutex::default(), #[cfg(feature = "download")] @@ -129,6 +132,8 @@ impl GrammarStore { #[cfg(feature = "runtime")] if let Some(inner) = &self.inner { let tag = tag.to_ascii_lowercase(); + // A tag resolving now is recorded before this clears it. + let _resolving = lock(&inner.resolving); let mut state = lock(&inner.state); #[cfg(feature = "download")] let failed = state.remote.forget_failure(&tag); @@ -220,16 +225,27 @@ impl GrammarStore { pub fn status(&self, language: &LanguageId) -> LanguageStatus { #[cfg(feature = "runtime")] if let Some(inner) = &self.inner { - return match inner.lookup(language.as_str()) { - Tag::Ready(_) => LanguageStatus::Ready, - Tag::Pending => LanguageStatus::Pending, - Tag::Unavailable => LanguageStatus::Unavailable, - }; + return inner.lookup(language.as_str()).status(); } let _ = language; LanguageStatus::Unavailable } + /// [`Self::status`] without resolving: what lookups found for + /// `language`, or `None` while none has resolved it (or after + /// [`Self::retry`] forgot it). It never reads or loads a pack, nor + /// waits for one that is loading. + pub fn resolved(&self, language: &LanguageId) -> Option { + #[cfg(feature = "runtime")] + if let Some(inner) = &self.inner { + return inner + .resolved(language.as_str()) + .map(|found| found.status()); + } + let _ = language; + Some(LanguageStatus::Unavailable) + } + /// Calls `listener` (on a background thread) each time a pending /// language resolves, whether its grammar arrived or failed. Return /// `false` to unsubscribe. Apps use it to wake their event loop. @@ -603,13 +619,30 @@ pub(crate) enum Tag { Unavailable, } +#[cfg(feature = "runtime")] +impl Tag { + fn status(&self) -> LanguageStatus { + match self { + Self::Ready(_) => LanguageStatus::Ready, + Self::Pending => LanguageStatus::Pending, + Self::Unavailable => LanguageStatus::Unavailable, + } + } +} + #[cfg(feature = "runtime")] type Listener = Box bool + Send>; #[cfg(feature = "runtime")] pub(crate) struct Inner { pub(crate) config: StoreConfig, + /// Guards the maps only: nothing reads, verifies, or loads a pack + /// while holding it (apart from the first scan of local roots and read + /// of the cached index), so resolved tags answer at once. pub(crate) state: Mutex, + /// Held while a tag resolves, so one tag (and the pack its aliases + /// share) loads once even when several threads ask for it. + resolving: Mutex<()>, listeners: Mutex>, status_listeners: Mutex>, #[cfg(feature = "download")] @@ -681,19 +714,33 @@ pub(crate) fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { #[cfg(feature = "runtime")] impl Inner { pub(crate) fn lookup(self: &Arc, tag: &str) -> Tag { - let mut state = lock(&self.state); - if let Some(found) = state.tags.get(tag) { - return found.clone(); + if let Some(found) = self.resolved(tag) { + return found; + } + let resolving = lock(&self.resolving); + // Another thread may have resolved it while this one waited. + if let Some(found) = self.resolved(tag) { + return found; } - let found = self.resolve(&mut state, tag); - state.tags.insert(tag.to_owned(), found.clone()); - drop(state); + let found = self.resolve(tag); + drop(resolving); if !matches!(found, Tag::Unavailable) { self.notify_status(); } found } + /// What `tag` resolved to, without resolving it. + pub(crate) fn resolved(&self, tag: &str) -> Option { + lock(&self.state).tags.get(tag).cloned() + } + + /// Remembers what `tag` resolved to. + pub(crate) fn record(&self, tag: &str, found: Tag) -> Tag { + lock(&self.state).tags.insert(tag.to_owned(), found.clone()); + found + } + fn ships_tags(&self, tag: &str) -> bool { // Only pack sources mutate state here. #[cfg_attr(not(feature = "engine"), allow(unused_mut))] @@ -722,46 +769,57 @@ impl Inner { false } - fn resolve(self: &Arc, state: &mut State, tag: &str) -> Tag { + /// Resolves `tag` and records what it resolved to. Called with the + /// resolve lock held; packs load without the state lock. + fn resolve(self: &Arc, tag: &str) -> Tag { #[cfg(feature = "engine")] - if let Some((dir, manifest)) = state - .local - .get_or_insert_with(|| scan_local(&self.config.local_packs)) - .get(tag) - .cloned() { - if !state.released && self.config.deferred.contains(&manifest.language) { - state.deferred.push(tag.to_owned()); - return Tag::Pending; + let mut state = lock(&self.state); + let local = state + .local + .get_or_insert_with(|| scan_local(&self.config.local_packs)) + .get(tag) + .cloned(); + if let Some((dir, manifest)) = local { + if !state.released && self.config.deferred.contains(&manifest.language) { + // Recorded with the deferral, so a release in between + // cannot miss it. + state.deferred.push(tag.to_owned()); + state.tags.insert(tag.to_owned(), Tag::Pending); + return Tag::Pending; + } + drop(state); + let found = load(self, &dir, &manifest); + return self.record(tag, found); } - return load(state, &dir, &manifest); } #[cfg(feature = "bundled")] if self.config.bundled && let Some(bundled) = crate::bundled::find(tag) { - return load_with(state, Origin::Bundled(bundled.name), || { + let found = load_with(self, Origin::Bundled(bundled.name), || { Grammar::bundled(bundled) }); + return self.record(tag, found); } #[cfg(feature = "download")] if let Some(downloads) = &self.config.downloads { - return crate::download::resolve(self, downloads, state, tag); + return crate::download::resolve(self, downloads, tag); } - Tag::Unavailable + self.record(tag, Tag::Unavailable) } fn catalog(&self) -> Catalog { - #[cfg_attr(not(feature = "engine"), allow(unused_mut))] - let mut state = lock(&self.state); let mut known: Vec = Vec::new(); let mut seen: std::collections::HashSet = std::collections::HashSet::new(); #[cfg(feature = "engine")] { - let local = state + let mut local_packs: Vec<_> = lock(&self.state) .local - .get_or_insert_with(|| scan_local(&self.config.local_packs)); - let mut local_packs: Vec<_> = local.values().cloned().collect(); + .get_or_insert_with(|| scan_local(&self.config.local_packs)) + .values() + .cloned() + .collect(); local_packs.sort_by(|a, b| a.0.cmp(&b.0)); for (dir, manifest) in local_packs { if seen.insert(manifest.language.clone()) { @@ -784,12 +842,13 @@ impl Inner { } #[cfg(feature = "download")] if let Some(downloads) = &self.config.downloads { - for (dir, manifest, installed) in crate::download::packs(downloads, &mut state) { + for (dir, manifest, installed) in crate::download::packs(downloads, self) { if seen.insert(manifest.language.clone()) { known.push(Known::pack(dir, &manifest, installed)); } } } + let state = lock(&self.state); let mut catalog = Catalog { languages: known.len(), ..Catalog::default() @@ -846,30 +905,42 @@ impl Inner { /// Loads (once per directory) the pack in `dir`. #[cfg(feature = "engine")] -pub(crate) fn load(state: &mut State, dir: &Path, manifest: &PackManifest) -> Tag { - load_with(state, Origin::Pack(dir.to_path_buf()), || { +pub(crate) fn load(inner: &Inner, dir: &Path, manifest: &PackManifest) -> Tag { + load_with(inner, Origin::Pack(dir.to_path_buf()), || { Grammar::load(dir, manifest) }) } +/// The grammar from `origin`, loaded with `load` the first time. Callers +/// hold the resolve lock, so `load` runs once per origin; it runs without +/// the state lock, which checking files, opening a library, and compiling +/// queries would hold for as long as they take. #[cfg(feature = "runtime")] pub(crate) fn load_with( - state: &mut State, + inner: &Inner, origin: Origin, load: impl FnOnce() -> Result, ) -> Tag { - let grammar = state - .loaded - .entry(origin) - .or_insert_with_key(|origin| match load() { - Ok(grammar) => Some(Arc::new(grammar)), - Err(error) => { - tracing::warn!(?origin, %error, "grammar failed to load"); - None - } - }); + let loaded = lock(&inner.state).loaded.get(&origin).cloned(); + let grammar = match loaded { + Some(grammar) => grammar, + None => { + let grammar = match load() { + Ok(grammar) => Some(Arc::new(grammar)), + Err(error) => { + tracing::warn!(?origin, %error, "grammar failed to load"); + None + } + }; + lock(&inner.state) + .loaded + .entry(origin) + .or_insert(grammar) + .clone() + } + }; match grammar { - Some(grammar) => Tag::Ready(grammar.clone()), + Some(grammar) => Tag::Ready(grammar), None => Tag::Unavailable, } } @@ -904,3 +975,53 @@ fn scan_local(roots: &[PathBuf]) -> HashMap) } tags } + +#[cfg(all(test, feature = "runtime"))] +mod tests { + use std::sync::mpsc::channel; + use std::time::Duration; + + use super::*; + + // Catches a pack load holding the lock every lookup takes, so the UI + // thread's check of a language resolved long ago waits while another + // pack is verified, opened, and compiled. + #[test] + fn resolved_languages_answer_while_a_pack_loads() { + let store = GrammarStore::new(StoreConfig::new()); + let klingon = LanguageId::from_fence("klingon").unwrap(); + store.status(&klingon); + let inner = store.inner.clone().unwrap(); + let (started_tx, started) = channel(); + let (release_tx, release) = channel::<()>(); + let loader = std::thread::spawn(move || { + // As `lookup` resolves a tag. + let _resolving = lock(&inner.resolving); + load_with(&inner, Origin::Bundled("slow"), || { + started_tx.send(()).unwrap(); + release.recv().unwrap(); + Err::("never loads") + }) + }); + started.recv_timeout(Duration::from_secs(60)).unwrap(); + let (answer_tx, answer) = channel(); + let asking = store.clone(); + std::thread::spawn(move || { + let _ = answer_tx.send((asking.resolved(&klingon), asking.status(&klingon))); + }); + let answered = answer.recv_timeout(Duration::from_secs(10)); + release_tx.send(()).unwrap(); + let loaded = loader.join().ok().map(|found| found.status()); + + assert_eq!( + (answered, loaded), + ( + Ok(( + Some(LanguageStatus::Unavailable), + LanguageStatus::Unavailable + )), + Some(LanguageStatus::Unavailable) + ) + ); + } +} diff --git a/crates/diffy-ui/src/diff_view/syntax.rs b/crates/diffy-ui/src/diff_view/syntax.rs index 056f1931..c123a7b8 100644 --- a/crates/diffy-ui/src/diff_view/syntax.rs +++ b/crates/diffy-ui/src/diff_view/syntax.rs @@ -527,8 +527,9 @@ impl DiffSyntax { }; held.status = Some(if done.unresolved.contains(language) { SyntaxStatus::Pending - } else if self.store.status(language) == LanguageStatus::Unavailable { - // Resolved by the worker already, so this is a map lookup. + } else if self.store.resolved(language) == Some(LanguageStatus::Unavailable) { + // Only reads what the worker found: this runs on the UI thread, + // and resolving could load a pack. SyntaxStatus::MissingGrammar } else if streaming { let permille = (u64::from(held.exact_end) * 1000 / len.max(1) as u64) as u16;