diff --git a/crates/amalthea/src/kernel.rs b/crates/amalthea/src/kernel.rs index b4c23bf940..7114838868 100644 --- a/crates/amalthea/src/kernel.rs +++ b/crates/amalthea/src/kernel.rs @@ -15,6 +15,7 @@ use crossbeam::channel::Receiver; use crossbeam::channel::Sender; use stdext::debug_panic; use stdext::spawn; +use stdext::spawn_with_stack_size; use stdext::unwrap; use crate::comm::event::CommEvent; @@ -219,9 +220,12 @@ pub fn connect( // Create the thread that handles stdout and stderr, if requested if stream_behavior == StreamBehavior::Capture { let iopub_tx_clone = channels.iopub_tx.clone(); - spawn!(format!("{name}-output-capture"), move || { - output_capture_thread(iopub_tx_clone) - }); + // `poll(2)` in a flat loop, reading into a 1 KiB stack buffer. + spawn_with_stack_size!( + format!("{name}-output-capture"), + stdext::TINY_STACK_SIZE, + move || { output_capture_thread(iopub_tx_clone) } + ); } // Create the Control ROUTER/DEALER socket diff --git a/crates/ark/src/console/console_repl.rs b/crates/ark/src/console/console_repl.rs index 6402ad4a1c..a210ca1cea 100644 --- a/crates/ark/src/console/console_repl.rs +++ b/crates/ark/src/console/console_repl.rs @@ -13,6 +13,7 @@ use std::path::Path; use std::rc::Rc; +use stdext::panic_message; use stdext::DebugRefCell; use super::*; @@ -750,14 +751,7 @@ impl Console { match result { Ok(result) => result, Err(panic) => { - let msg = match panic.downcast_ref::<&str>() { - Some(s) => s.to_string(), - None => match panic.downcast_ref::() { - Some(s) => s.clone(), - None => String::from("(unknown payload)"), - }, - }; - + let msg = panic_message(panic.as_ref()); Err(anyhow!("Panic in Console callback: {msg}")) }, } diff --git a/crates/ark/src/lsp.rs b/crates/ark/src/lsp.rs index 1b7dcc9255..a2054e4104 100644 --- a/crates/ark/src/lsp.rs +++ b/crates/ark/src/lsp.rs @@ -5,6 +5,7 @@ // // +mod analysis; pub mod backend; pub mod capabilities; pub mod code_action; @@ -28,6 +29,7 @@ pub mod hover; pub mod indent; pub mod indexer; pub mod input_boundaries; +mod io_pool; pub mod main_loop; pub mod markdown; pub(crate) mod open_file; @@ -43,6 +45,7 @@ pub mod statement_range; pub mod symbols; pub mod traits; pub mod util; +mod watchdog; #[cfg(test)] mod tests; @@ -69,4 +72,3 @@ pub(crate) use log_error; pub(crate) use log_info; pub(crate) use log_warn; pub(crate) use main_loop::publish_diagnostics; -pub(crate) use main_loop::spawn_blocking; diff --git a/crates/ark/src/lsp/analysis.rs b/crates/ark/src/lsp/analysis.rs new file mode 100644 index 0000000000..6a372852ea --- /dev/null +++ b/crates/ark/src/lsp/analysis.rs @@ -0,0 +1,30 @@ +// +// analysis.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +//! Background analysis, the only place a salsa db handle lives off the main +//! loop. +//! +//! Two things maintain that invariant: [`WorldStateSnapshot`] is built only in +//! this module, and `OakDatabase` isn't `Clone`. + +use std::panic::AssertUnwindSafe; + +mod pool; +mod refresh; +mod snapshot; +mod warmup; + +pub(crate) use pool::AnalysisPool; +pub(crate) use refresh::DiagnosticsReady; +pub(crate) use refresh::DiagnosticsState; +pub(crate) use snapshot::WorldStateSnapshot; +pub(crate) use warmup::warm_workspace_index; + +/// Run `f`, swallowing a salsa cancellation as `None`. Any other panic propagates. +fn catch_cancellation(f: impl FnOnce() -> T) -> Option { + salsa::Cancelled::catch(AssertUnwindSafe(f)).ok() +} diff --git a/crates/ark/src/lsp/analysis/pool.rs b/crates/ark/src/lsp/analysis/pool.rs new file mode 100644 index 0000000000..4181a3decd --- /dev/null +++ b/crates/ark/src/lsp/analysis/pool.rs @@ -0,0 +1,253 @@ +// +// pool.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +use std::collections::VecDeque; +use std::panic::AssertUnwindSafe; +use std::sync::Arc; +use std::sync::Condvar; +use std::sync::Mutex; +use std::sync::MutexGuard; + +use aether_path::FilePath; +use stdext::panic_message; +use stdext::spawn; + +use super::catch_cancellation; +use super::snapshot::WorldStateSnapshot; +use crate::lsp; + +/// Enough threads that a handful of open files all get diagnosed in parallel, +/// few enough that they don't crowd out the main loop or the R session we share +/// a process with. +const MAX_ANALYSIS_THREADS: usize = 4; + +/// A fixed set of OS threads running analysis tasks over a db snapshot. +/// +/// Each task's snapshot is taken at enqueue time on the main loop and sees +/// the state as of that tick. A write waits for those snapshots to drop +/// before it can proceed. This pool doesn't order results across tasks: a +/// diagnostics result carries a generation id and [`DiagnosticsState::accept`] +/// drops staled results. +/// +/// A writer never waits on this pool for longer than the one task it +/// interrupted. Queued tasks get thrown away, and the task currently running +/// unwinds at its next salsa query. +pub(crate) struct AnalysisPool { + shared: Arc, +} + +impl AnalysisPool { + pub(crate) fn new() -> Self { + Self::with_threads(analysis_threads()) + } + + fn with_threads(threads: usize) -> Self { + let shared = Arc::new(Shared { + queue: Mutex::new(Queue { + entries: VecDeque::new(), + closed: false, + }), + ready: Condvar::new(), + }); + + for _ in 0..threads { + let shared = Arc::clone(&shared); + spawn!("oak-analysis", move || work(shared)); + } + + Self { shared } + } + + /// Queue `run` behind everything already queued. + pub(super) fn spawn( + &self, + snapshot: WorldStateSnapshot, + run: impl FnOnce(WorldStateSnapshot) + Send + 'static, + ) { + self.push(Entry { + key: None, + snapshot, + run: Box::new(run), + }); + } + + /// Queue `run`, replacing a queued task with the same `key` that hasn't + /// started yet. Diagnostics key on the file, so a fresh pass supersedes a + /// queued predecessor. + pub(super) fn spawn_keyed( + &self, + key: FilePath, + snapshot: WorldStateSnapshot, + run: impl FnOnce(WorldStateSnapshot) + Send + 'static, + ) { + self.push(Entry { + key: Some(key), + snapshot, + run: Box::new(run), + }); + } + + fn push(&self, entry: Entry) { + let mut queue = self.shared.lock(); + + if entry.key.is_some() { + let queued = queue + .entries + .iter_mut() + .find(|queued| queued.key == entry.key); + + // Reuse the slot, so a file that keeps getting edited can't starve + // the other files behind it. + if let Some(queued) = queued { + *queued = entry; + return; + } + } + + queue.entries.push_back(entry); + drop(queue); + self.shared.ready.notify_one(); + } +} + +/// Analysis tasks are CPU-bound, so don't run more of them than the machine can +/// actually run at once. +fn analysis_threads() -> usize { + match std::thread::available_parallelism() { + Ok(parallelism) => parallelism.get().min(MAX_ANALYSIS_THREADS), + Err(err) => { + log::warn!("Can't determine available parallelism, using one analysis thread: {err}"); + 1 + }, + } +} + +/// Closing the queue is all a worker needs to exit, so shutdown doesn't join +/// (and never blocks the caller). Clearing the backlog here releases the db +/// handles those tasks were holding. +impl Drop for AnalysisPool { + fn drop(&mut self) { + let mut queue = self.shared.lock(); + queue.closed = true; + queue.entries.clear(); + drop(queue); + self.shared.ready.notify_all(); + } +} + +struct Shared { + queue: Mutex, + ready: Condvar, +} + +struct Queue { + entries: VecDeque, + closed: bool, +} + +struct Entry { + /// `Some` for a task that a later task with the same key may replace. + key: Option, + snapshot: WorldStateSnapshot, + run: Box, +} + +fn work(shared: Arc) { + // `run_entry` takes the entry by value, so the snapshot has dropped by the + // time we ask for the next one. A worker parked on `next_entry` doesn't + // hold a db handle and can't block a writer. + while let Some(entry) = shared.next_entry() { + run_entry(entry); + } +} + +impl Shared { + fn next_entry(&self) -> Option { + let mut queue = self.lock(); + loop { + if let Some(entry) = queue.entries.pop_front() { + return Some(entry); + } + if queue.closed { + return None; + } + queue = match self.ready.wait(queue) { + Ok(queue) => queue, + Err(err) => err.into_inner(), + }; + } + } + + /// Tasks never run under this lock, so a poisoned lock still guards a + /// consistent queue. + fn lock(&self) -> MutexGuard<'_, Queue> { + match self.queue.lock() { + Ok(queue) => queue, + Err(err) => err.into_inner(), + } + } +} + +fn run_entry(entry: Entry) { + let Entry { snapshot, run, .. } = entry; + + // A writer parked on this handle would only cancel the task at its first + // query, so go straight to dropping the snapshot. This is what lets a + // backlog drain in one pass while a writer waits. + if snapshot.is_cancelled() { + return; + } + + let task = AssertUnwindSafe(|| catch_cancellation(|| run(snapshot))); + if let Err(err) = std::panic::catch_unwind(task) { + lsp::log_error!( + "An analysis task panicked: {msg}", + msg = panic_message(err.as_ref()) + ); + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::AtomicBool; + use std::sync::atomic::Ordering; + use std::sync::Arc; + + use super::AnalysisPool; + use crate::lsp::state::WorldState; + + /// A queued task whose snapshot is already cancelled must be dropped without + /// running. That is what lets a backlog release its db handles while a writer + /// is parked, instead of each task needing a thread first. + /// + /// One worker, so the barrier task behind it can only run after the + /// cancelled task has been dequeued. + #[test] + fn test_pool_drops_cancelled_task_without_running() { + let state = WorldState::default(); + let pool = AnalysisPool::with_threads(1); + + let cancelled = state.snapshot(); + cancelled.cancellation_token().cancel(); + + let ran = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&ran); + pool.spawn(cancelled, move |_snapshot| { + flag.store(true, Ordering::Release) + }); + + let (barrier_tx, barrier_rx) = std::sync::mpsc::channel(); + pool.spawn(state.snapshot(), move |_snapshot| { + barrier_tx.send(()).unwrap() + }); + + barrier_rx + .recv_timeout(std::time::Duration::from_secs(10)) + .unwrap(); + assert!(!ran.load(Ordering::Acquire)); + } +} diff --git a/crates/ark/src/lsp/analysis/refresh.rs b/crates/ark/src/lsp/analysis/refresh.rs new file mode 100644 index 0000000000..cfab9b3fb2 --- /dev/null +++ b/crates/ark/src/lsp/analysis/refresh.rs @@ -0,0 +1,197 @@ +// +// refresh.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +use std::collections::HashMap; + +use aether_path::FilePath; +use stdext::result::ResultExt; + +use super::pool::AnalysisPool; +use super::snapshot::WorldStateSnapshot; +use crate::lsp; +use crate::lsp::diagnostics::generate_diagnostics; +use crate::lsp::main_loop::DiagnosticsPublication; +use crate::lsp::main_loop::Event; +use crate::lsp::main_loop::TokioUnboundedSender; +use crate::lsp::open_file::OpenFile; +use crate::lsp::state::WorldState; +use crate::url::FilePathExt; + +/// A diagnostics task's result on its way back to the main loop. The generation +/// state enables [`DiagnosticsState::accept`] to distinguish a stale result +/// from a fresh one. +#[derive(Debug)] +pub(crate) struct DiagnosticsReady { + pub(crate) generation: u64, + pub(crate) publication: DiagnosticsPublication, +} + +/// Tracks diagnostics staleness across refresh batches, so an out-of-order +/// result gets dropped instead of published over a newer one. +/// +/// Mirrors rust-analyzer's generation counter in +/// `crates/rust-analyzer/src/diagnostics.rs`. +#[derive(Default)] +pub(crate) struct DiagnosticsState { + /// Bumped once per refresh batch. + generation: u64, + /// Generation of the newest result published per file. + published: HashMap, +} + +impl DiagnosticsState { + /// Queue a diagnostics pass for every open file we diagnose, all tagged + /// with a new generation. + pub(crate) fn refresh_all( + &mut self, + state: &WorldState, + pool: &AnalysisPool, + events_tx: &TokioUnboundedSender, + ) { + self.generation += 1; + let generation = self.generation; + + let files: Vec<(&FilePath, &OpenFile)> = state + .open_files + .iter() + .filter(|(path, _open_file)| path.should_diagnose()) + .collect(); + + tracing::trace!("Refreshing diagnostics for {n} documents", n = files.len()); + lsp::log_info!("Queueing {n} diagnostic tasks", n = files.len()); + + for (path, open_file) in files { + let path = path.clone(); + let file = open_file.clone(); + let events_tx = events_tx.clone(); + + pool.spawn_keyed(path.clone(), state.snapshot(), move |snapshot| { + let publication = refresh_diagnostics(path, file, snapshot); + let ready = DiagnosticsReady { + generation, + publication, + }; + events_tx.send(Event::DiagnosticsReady(ready)).log_err(); + }); + } + } + + /// Whether a diagnostics result for `path` computed at `generation` + /// should be published now, or is stale and should be dropped. + /// + /// Equal generations can't legitimately arrive twice for the same file: + /// we spawn one task per file per batch, and keyed replacement on the + /// pool keeps at most one queued entry per file. + pub(crate) fn accept(&mut self, path: &FilePath, generation: u64) -> bool { + if let Some(published) = self.published.get(path) { + if *published > generation { + return false; + } + } + + self.published.insert(path.clone(), generation); + true + } + + /// Generation of the newest result already published for `path`, for the + /// main loop to log alongside a dropped stale result. + pub(crate) fn published_generation(&self, path: &FilePath) -> Option { + self.published.get(path).copied() + } +} + +fn refresh_diagnostics( + path: FilePath, + file: OpenFile, + state: WorldStateSnapshot, +) -> DiagnosticsPublication { + let uri = file.wire_uri().clone(); + let version = file.version(); + let _span = tracing::info_span!("diagnostics_refresh", uri = %uri.as_str()).entered(); + + // Special case testthat-specific behaviour. This is a simple stopgap + // approach that has some false positives (e.g. when we work on testthat + // itself the flag will always be true), but that shouldn't have much + // practical impact. + let testthat = path + .as_path() + .is_some_and(|path| path.components().any(|c| c.as_str() == "testthat")); + + let now = std::time::Instant::now(); + lsp::log_info!("Generating diagnostics for file: {}", uri.as_str()); + + let diagnostics = generate_diagnostics(file.file(), state, testthat); + + lsp::log_info!( + "Finished diagnostics for file: {} in {:.0?}", + uri.as_str(), + now.elapsed() + ); + + DiagnosticsPublication { + path, + uri, + diagnostics, + version, + } +} + +#[cfg(test)] +mod tests { + use aether_path::FilePath; + use oak_scan::DbScan; + use url::Url; + + use super::refresh_diagnostics; + use super::DiagnosticsState; + use crate::lsp::analysis::catch_cancellation; + use crate::lsp::state::WorldState; + use crate::lsp::traits::url::UrlExt; + + /// `accept` is the staleness gate a refresh batch relies on: a result only + /// publishes if no newer generation for that file already went out. Pins + /// the three cases that can arrive at the main loop: a file seen for the + /// first time, a fresh batch superseding the last one, and a straggler + /// from an old batch arriving after a newer one already landed. Also pins + /// the deliberate choice to accept a repeat of the last generation (see + /// `accept`'s doc comment for why that can't happen in practice but is + /// still safe). + #[test] + fn test_accept_tracks_staleness_per_file() { + let mut diagnostics = DiagnosticsState::default(); + let path = FilePath::from_url(&Url::parse("file:///test.R").unwrap()); + + assert!(diagnostics.accept(&path, 1)); + assert!(diagnostics.accept(&path, 2)); + assert!(!diagnostics.accept(&path, 1)); + assert!(diagnostics.accept(&path, 2)); + } + + /// A salsa cancellation during the pass is swallowed into `None` by + /// `catch_cancellation`, the wrapper the pool applies to every task, rather + /// than unwinding and killing the worker thread. + /// + /// `cancellation_token().cancel()` arms local cancellation on the snapshot's + /// oak, so the first salsa query in `generate_diagnostics` (the `tree_sitter` + /// fetch) unwinds with `salsa::Cancelled`, the same payload a concurrent + /// `set_*` produces. The unwind fires before any R, so no `r_task` here. + #[test] + fn test_cancelled_diagnostics_pass_is_caught() { + let mut state = WorldState::default(); + let uri = Url::parse("file:///test.R").unwrap(); + let path = FilePath::from_url(&uri); + let code = "foo"; + let file = state.db.upsert_editor(path.clone(), code.to_string()); + state.insert_open_file(uri.to_uri().unwrap(), path.clone(), file, None); + + let file = state.open_file(&path).unwrap().clone(); + let snapshot = state.snapshot(); + snapshot.cancellation_token().cancel(); + + assert!(catch_cancellation(|| refresh_diagnostics(path, file, snapshot)).is_none()); + } +} diff --git a/crates/ark/src/lsp/analysis/snapshot.rs b/crates/ark/src/lsp/analysis/snapshot.rs new file mode 100644 index 0000000000..feb49f8424 --- /dev/null +++ b/crates/ark/src/lsp/analysis/snapshot.rs @@ -0,0 +1,92 @@ +// +// snapshot.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +use oak_db::OakDatabase; + +use super::catch_cancellation; +use crate::lsp::config::LspConfig; +use crate::lsp::db::ArkDb; +use crate::lsp::state::Workspace; +use crate::lsp::state::WorldState; + +/// Read-only snapshot of [`WorldState`] handed to a background reader, so a +/// reader thread can't reach salsa input setters. Carries only the fields +/// readers actually use. Mirrors rust-analyzer's `GlobalStateSnapshot`. +#[derive(Debug)] +pub(crate) struct WorldStateSnapshot { + /// Private so readers can only reach it through [`Self::db`]. + db: OakDatabase, + pub(crate) workspace: Workspace, + pub(crate) console_scopes: Vec>, + pub(crate) installed_packages: Vec, + pub(crate) config: LspConfig, +} + +/// Minting lives here rather than in `state.rs` because +/// [`WorldStateSnapshot`]'s db field is private to this module. +impl WorldState { + /// Take a read-only snapshot of the world for a background reader. + /// + /// The snapshot holds a Salsa handle, which parks the next main-loop write + /// until it drops. That's safe on the [`AnalysisPool`], whose tasks unwind + /// on cancellation. The only other caller is `handle_completion()`, which + /// hands the snapshot to `r_task()` and blocks until it returns, so that + /// handle can't outlive the tick that made it. + pub(crate) fn snapshot(&self) -> WorldStateSnapshot { + WorldStateSnapshot { + db: self.db.snapshot(), + console_scopes: self.console_scopes.clone(), + installed_packages: self.installed_packages.clone(), + config: self.config.clone(), + workspace: self.workspace.clone(), + } + } +} + +impl WorldStateSnapshot { + /// Read-only access to the database. Returns `&dyn ArkDb` rather than + /// `&OakDatabase` because `dyn ArkDb` is unsized, so a reader can't + /// `.snapshot()` its way to an owned database and call setters on it. + pub(crate) fn db(&self) -> &dyn ArkDb { + &self.db + } + + /// Whether salsa would unwind this handle's next query, because a writer is + /// parked waiting for it to drop (or, in tests, because the token was armed + /// by hand). `unwind_if_revision_cancelled()` reports by throwing, so we + /// catch it to get an answer. + pub(super) fn is_cancelled(&self) -> bool { + catch_cancellation(|| salsa::Database::unwind_if_revision_cancelled(&self.db)).is_none() + } + + /// The database's salsa cancellation token. Read-side only: it observes and + /// arms cancellation, it doesn't mutate any input. Only cancellation tests + /// arm it by hand. + #[cfg(test)] + pub(crate) fn cancellation_token(&self) -> salsa::CancellationToken { + salsa::Database::cancellation_token(&self.db) + } +} + +#[cfg(test)] +mod tests { + use crate::lsp::state::WorldState; + + /// A snapshot reports itself cancelled once salsa would unwind its next + /// query, which is what the pool checks at dequeue. + #[test] + fn test_cancelled_snapshot_reports_cancelled() { + let state = WorldState::default(); + + let live = state.snapshot(); + assert!(!live.is_cancelled()); + + let cancelled = state.snapshot(); + cancelled.cancellation_token().cancel(); + assert!(cancelled.is_cancelled()); + } +} diff --git a/crates/ark/src/lsp/analysis/warmup.rs b/crates/ark/src/lsp/analysis/warmup.rs new file mode 100644 index 0000000000..38466b415c --- /dev/null +++ b/crates/ark/src/lsp/analysis/warmup.rs @@ -0,0 +1,32 @@ +// +// warmup.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +use super::pool::AnalysisPool; +use crate::lsp; +use crate::lsp::indexer; +use crate::lsp::state::WorldState; + +/// Build the per-file workspace symbol indexes on a background thread so +/// main-loop consumers triggered by the user (workspace symbols, workspace +/// completions) find them already computed. The first run after a workspace +/// scan does the real work, parsing and walking each file. Later runs only +/// revalidate the per-file memos. +/// +/// Mirrors rust-analyzer's cache warming: spawned when a workspace scan +/// settles, the analogue of r-a's transitions to quiescence (initial VFS scan, +/// workspace reload, etc). Unlike r-a we don't restart a warmup that gets +/// cancelled (the pool swallows the unwind). A cancelling write can only come +/// from an editor buffer, so a document is open, and the diagnostics passes +/// spawned by that same write force the same memos and finish the job. +pub(crate) fn warm_workspace_index(state: &WorldState, pool: &AnalysisPool) { + pool.spawn(state.snapshot(), |snapshot| { + let now = std::time::Instant::now(); + lsp::log_info!("Starting workspace index warmup"); + indexer::warm(snapshot.db()); + lsp::log_info!("Finished workspace index warmup ({:.0?})", now.elapsed()); + }) +} diff --git a/crates/ark/src/lsp/completions/completion_context.rs b/crates/ark/src/lsp/completions/completion_context.rs index 6d772a4064..c105f20b25 100644 --- a/crates/ark/src/lsp/completions/completion_context.rs +++ b/crates/ark/src/lsp/completions/completion_context.rs @@ -9,11 +9,11 @@ use std::cell::OnceCell; use tree_sitter::Node; +use crate::lsp::analysis::WorldStateSnapshot; use crate::lsp::completions::function_context::FunctionContext; use crate::lsp::completions::sources::composite::pipe::find_pipe_root; use crate::lsp::completions::sources::composite::pipe::PipeRoot; use crate::lsp::document_context::DocumentContext; -use crate::lsp::state::WorldStateSnapshot; use crate::treesitter::node_find_containing_call; pub(crate) struct CompletionContext<'a> { pub(crate) document_context: &'a DocumentContext<'a>, diff --git a/crates/ark/src/lsp/completions/provide.rs b/crates/ark/src/lsp/completions/provide.rs index 430dd0028b..92937712e9 100644 --- a/crates/ark/src/lsp/completions/provide.rs +++ b/crates/ark/src/lsp/completions/provide.rs @@ -7,11 +7,11 @@ use tower_lsp_server::ls_types::CompletionItem; +use crate::lsp::analysis::WorldStateSnapshot; use crate::lsp::completions::completion_context::CompletionContext; use crate::lsp::completions::sources::composite; use crate::lsp::completions::sources::unique; use crate::lsp::document_context::DocumentContext; -use crate::lsp::state::WorldStateSnapshot; use crate::lsp::traits::node::NodeExt; use crate::treesitter::NodeTypeExt; diff --git a/crates/ark/src/lsp/diagnostics.rs b/crates/ark/src/lsp/diagnostics.rs index c090fe108f..9fc105edb2 100644 --- a/crates/ark/src/lsp/diagnostics.rs +++ b/crates/ark/src/lsp/diagnostics.rs @@ -23,13 +23,13 @@ use tree_sitter::Point; use tree_sitter::Range; use crate::lsp; +use crate::lsp::analysis::WorldStateSnapshot; use crate::lsp::db::ArkDb; use crate::lsp::db::FileArkExt; use crate::lsp::declarations::top_level_declare; use crate::lsp::diagnostics_syntax::syntax_diagnostics; use crate::lsp::indexer; use crate::lsp::open_file::lsp_range_from_tree_sitter_range; -use crate::lsp::state::WorldStateSnapshot; use crate::lsp::traits::node::NodeExt; use crate::treesitter::node_has_error_or_missing; use crate::treesitter::BinaryOperatorType; @@ -1182,7 +1182,7 @@ mod tests { use crate::lsp::state::WorldState; use crate::r_task; - fn generate_diagnostics(code: &str, state: WorldState) -> Vec { + fn generate_diagnostics(code: &str, state: &WorldState) -> Vec { let url = url::Url::parse("file:///test.R").unwrap(); let file = oak_db::File::new( &state.db, @@ -1227,7 +1227,7 @@ mod tests { foo 1 }"; - let diagnostics = generate_diagnostics(text, current_state()); + let diagnostics = generate_diagnostics(text, ¤t_state()); assert_eq!(diagnostics.len(), 2); let diagnostic = diagnostics.first().unwrap(); @@ -1250,7 +1250,7 @@ foo 1, 2 # hi there )"; - let diagnostics = generate_diagnostics(text, current_state()); + let diagnostics = generate_diagnostics(text, ¤t_state()); assert!(diagnostics.is_empty()); }) } @@ -1259,7 +1259,7 @@ foo fn test_missing_namespace_rhs() { r_task(|| { let text = "base::"; - let diagnostics = generate_diagnostics(text, current_state()); + let diagnostics = generate_diagnostics(text, ¤t_state()); assert_eq!(diagnostics.len(), 1); let diagnostic = diagnostics.first().unwrap(); insta::assert_snapshot!(diagnostic.message); @@ -1271,7 +1271,7 @@ foo r_task(|| { let text = "..1 + ..2 + 3"; - let diagnostics = generate_diagnostics(text, current_state()); + let diagnostics = generate_diagnostics(text, ¤t_state()); assert!(diagnostics.is_empty()); }) @@ -1290,11 +1290,11 @@ foo let state = current_state(); let text = "x$foo"; - let diagnostics = generate_diagnostics(text, state.clone()); + let diagnostics = generate_diagnostics(text, &state); assert!(diagnostics.is_empty()); let text = "x@foo"; - let diagnostics = generate_diagnostics(text, state.clone()); + let diagnostics = generate_diagnostics(text, &state); assert!(diagnostics.is_empty()); // Clean up @@ -1311,7 +1311,7 @@ foo z = 3 y + x + z "; - let diagnostics = generate_diagnostics(text, current_state()); + let diagnostics = generate_diagnostics(text, ¤t_state()); assert!(diagnostics.is_empty()); }) } @@ -1324,7 +1324,7 @@ foo 2 ->> y y + x "; - let diagnostics = generate_diagnostics(text, current_state()); + let diagnostics = generate_diagnostics(text, ¤t_state()); assert!(diagnostics.is_empty()); }) } @@ -1338,7 +1338,7 @@ foo x + 1 "; - let diagnostics = generate_diagnostics(text, current_state()); + let diagnostics = generate_diagnostics(text, ¤t_state()); assert_eq!(diagnostics.len(), 1); // Only marks the `x` before the `x <- 1` @@ -1356,7 +1356,7 @@ foo identity(foo ~ bar) identity(~foo) "; - let diagnostics = generate_diagnostics(text, current_state()); + let diagnostics = generate_diagnostics(text, ¤t_state()); assert!(diagnostics.is_empty()); }) } @@ -1371,7 +1371,7 @@ foo cherry "; - let diagnostics = generate_diagnostics(code, current_state()); + let diagnostics = generate_diagnostics(code, ¤t_state()); assert_eq!(diagnostics.len(), 1); let diagnostic = diagnostics.first().unwrap(); @@ -1389,7 +1389,7 @@ foo cherry "; - let diagnostics = generate_diagnostics(code, current_state()); + let diagnostics = generate_diagnostics(code, ¤t_state()); assert_eq!(diagnostics.len(), 1); let diagnostic = diagnostics.first().unwrap(); @@ -1408,7 +1408,7 @@ foo x "; - let diagnostics = generate_diagnostics(code, current_state()); + let diagnostics = generate_diagnostics(code, ¤t_state()); assert_eq!(diagnostics.len(), 1); let diagnostic = diagnostics.first().unwrap(); @@ -1426,7 +1426,7 @@ foo cherry "; - let diagnostics = generate_diagnostics(code, current_state()); + let diagnostics = generate_diagnostics(code, ¤t_state()); assert_eq!(diagnostics.len(), 1); let diagnostic = diagnostics.first().unwrap(); @@ -1442,7 +1442,7 @@ foo apple "; - let diagnostics = generate_diagnostics(code, current_state()); + let diagnostics = generate_diagnostics(code, ¤t_state()); assert_eq!(diagnostics.len(), 0); }) } @@ -1455,7 +1455,7 @@ foo apple "; - let diagnostics = generate_diagnostics(code, current_state()); + let diagnostics = generate_diagnostics(code, ¤t_state()); assert_eq!(diagnostics.len(), 0); }) } @@ -1474,13 +1474,13 @@ foo list(x <- 1) x "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); let code = " list({ x <- 1 }) x "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); }); // Subset @@ -1490,14 +1490,14 @@ foo foo[x <- 1] x "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); let code = " foo <- list() foo[{x <- 1}] x "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); }); // Subset2 @@ -1507,14 +1507,14 @@ foo foo[[x <- 1]] x "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); let code = " foo <- list() foo[[{x <- 1}]] x "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); }); } @@ -1529,7 +1529,7 @@ foo let code = " list(x) "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); // Important to test nested case. We have a dynamic stack of state // variable to keep track of whether we are in a call. The inner @@ -1537,14 +1537,14 @@ foo let code = " list(list(), x) "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); // `in_call_like_arguments` state variable is reset let code = " list() x "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 1); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 1); }); // Subset @@ -1557,7 +1557,7 @@ foo data[x] data[,x] "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); // Imagine this is `data.table()` (we don't necessarily have the package // installed in the test) @@ -1566,7 +1566,7 @@ foo data <- data.frame(x = 1) data[, y := x + 1] "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); }); // Subset2 @@ -1575,7 +1575,7 @@ foo foo <- list() foo[[x]] "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); }); } @@ -1587,7 +1587,7 @@ foo x <- list(a = 1) x |> _$a[1] "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); // Imagine this is a data.table // https://github.com/posit-dev/positron/issues/3749 @@ -1595,14 +1595,14 @@ foo data <- data.frame(a = 1) data |> _[1] "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); // We technically disable diagnostics for this symbol everywhere, even outside // of pipe scope, which is probably fine let code = " _ "; - assert_eq!(generate_diagnostics(code, current_state()).len(), 0); + assert_eq!(generate_diagnostics(code, ¤t_state()).len(), 0); }) } @@ -1630,7 +1630,7 @@ foo foo() bar "; - let diagnostics = generate_diagnostics(code, state.clone()); + let diagnostics = generate_diagnostics(code, &state); assert_eq!(diagnostics.len(), 0); @@ -1641,7 +1641,7 @@ foo also_undefined "; - let diagnostics = generate_diagnostics(code, state.clone()); + let diagnostics = generate_diagnostics(code, &state); assert_eq!(diagnostics.len(), 2); assert!(diagnostics @@ -1662,7 +1662,7 @@ foo foo() bar "; - let diagnostics = generate_diagnostics(code, state.clone()); + let diagnostics = generate_diagnostics(code, &state); assert_eq!(diagnostics.len(), 0); // If the library call includes the `character.only` argument, we bail @@ -1670,7 +1670,7 @@ foo library(mockpkg, character.only = TRUE) foo() "#; - let diagnostics = generate_diagnostics(code, state.clone()); + let diagnostics = generate_diagnostics(code, &state); assert_eq!(diagnostics.len(), 1); // Same if passed `FALSE`, we're not trying to be smart (yet) @@ -1678,7 +1678,7 @@ foo library(mockpkg, character.only = FALSE) foo() "#; - let diagnostics = generate_diagnostics(code, state); + let diagnostics = generate_diagnostics(code, &state); assert_eq!(diagnostics.len(), 1); } @@ -1714,7 +1714,7 @@ foo bar # in scope baz # in scope "; - let diagnostics = generate_diagnostics(code, state.clone()); + let diagnostics = generate_diagnostics(code, &state); let messages: Vec<_> = diagnostics.iter().map(|d| d.message.clone()).collect(); assert!(messages.iter().any(|m| m.contains("No symbol named 'foo'"))); @@ -1745,7 +1745,7 @@ foo bar foo() "; - let diagnostics = generate_diagnostics(code, state.clone()); + let diagnostics = generate_diagnostics(code, &state); assert!(diagnostics .iter() .any(|d| d.message.contains("No symbol named 'foo'"))); @@ -1773,7 +1773,7 @@ foo path_to_file penguins_raw "#; - let diagnostics = generate_diagnostics(code, state.clone()); + let diagnostics = generate_diagnostics(code, &state); assert!(diagnostics.is_empty()); let code = r#" @@ -1782,7 +1782,7 @@ foo penguins_raw library(penguins) "#; - let diagnostics = generate_diagnostics(code, state); + let diagnostics = generate_diagnostics(code, &state); assert_eq!(diagnostics.len(), 3); } } diff --git a/crates/ark/src/lsp/handler.rs b/crates/ark/src/lsp/handler.rs index 04b96ed32f..31370960c6 100644 --- a/crates/ark/src/lsp/handler.rs +++ b/crates/ark/src/lsp/handler.rs @@ -39,11 +39,10 @@ impl Lsp { ) -> Self { let rt = Builder::new_multi_thread() .enable_all() - // Workers serve tower-lsp, the auxiliary loop, and the diagnostics - // queue. The main loop runs on its own thread. + // Workers serve tower-lsp and the auxiliary loop. The main loop has + // its own thread, and background analysis, workspace scans, and + // source fetches have their own pools of OS threads. .worker_threads(2) - // Used for diagnostics - .max_blocking_threads(2) .build() .unwrap(); diff --git a/crates/ark/src/lsp/handlers.rs b/crates/ark/src/lsp/handlers.rs index 846d7d926d..b6f74ca60e 100644 --- a/crates/ark/src/lsp/handlers.rs +++ b/crates/ark/src/lsp/handlers.rs @@ -240,9 +240,9 @@ pub(crate) fn handle_completion( // Snapshot so the closure captures by value. `r_task()` sends the closure // across threads, and `&WorldState` isn't `Send` because `OakDatabase`'s - // salsa storage keeps thread-local query state. `snapshot()` hands the - // reader a `WorldStateSnapshot`, so the background thread can query oak but - // can't call a setter. + // salsa storage keeps thread-local query state. `snapshot()` hands + // the reader a `WorldStateSnapshot`, so the background thread can query oak + // but can't call a setter. // TODO(oak/completions): We don't really need a snapshot here since // completions are serviced from the main loop, it's only needed for the // `r_task()`. diff --git a/crates/ark/src/lsp/io_pool.rs b/crates/ark/src/lsp/io_pool.rs new file mode 100644 index 0000000000..e2c579258a --- /dev/null +++ b/crates/ark/src/lsp/io_pool.rs @@ -0,0 +1,63 @@ +// +// io_pool.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +use std::panic::AssertUnwindSafe; + +use crossbeam::channel::Sender; +use stdext::panic_message; +use stdext::spawn_with_stack_size; + +use crate::lsp; + +type Job = Box; + +/// A fixed set of OS threads running I/O jobs in FIFO order. +/// +/// Jobs here must not own a salsa db handle. A download or an R subprocess +/// can't be interrupted by a Salsa cancellation. A handle sitting in this queue +/// would hold up the next main-loop write for that whole time. +pub(crate) struct IoPool { + /// The pool's only sender. Drop it to disconnect the channel and shut down + /// the workers. + jobs_tx: Sender, +} + +impl IoPool { + /// Start `threads` workers, each named `name` and given `stack_size` bytes + /// of stack. Each lane picks its own size from the deepest call tree its + /// jobs can reach, so use [`stdext::DEFAULT_STACK_SIZE`] unless you've + /// bounded that. + pub(crate) fn new(name: &'static str, threads: usize, stack_size: usize) -> Self { + let (jobs_tx, jobs_rx) = crossbeam::channel::unbounded::(); + + for _ in 0..threads { + let jobs_rx = jobs_rx.clone(); + spawn_with_stack_size!(name, stack_size, move || { + while let Ok(job) = jobs_rx.recv() { + run_job(job); + } + }); + } + + Self { jobs_tx } + } + + pub(crate) fn submit(&self, job: impl FnOnce() + Send + 'static) { + if self.jobs_tx.send(Box::new(job)).is_err() { + lsp::log_error!("No live I/O worker left, dropping job"); + } + } +} + +fn run_job(job: Job) { + if let Err(err) = std::panic::catch_unwind(AssertUnwindSafe(job)) { + lsp::log_error!( + "An I/O job panicked: {msg}", + msg = panic_message(err.as_ref()) + ); + } +} diff --git a/crates/ark/src/lsp/main_loop.rs b/crates/ark/src/lsp/main_loop.rs index 12d3f7a6c1..91ce3fdcec 100644 --- a/crates/ark/src/lsp/main_loop.rs +++ b/crates/ark/src/lsp/main_loop.rs @@ -7,31 +7,26 @@ use std::collections::HashMap; use std::collections::HashSet; -use std::future; use std::path::Path; use std::path::PathBuf; -use std::pin::Pin; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::sync::Arc; -use std::sync::LazyLock; use std::sync::RwLock; use aether_path::FilePath; use anyhow::anyhow; -use futures::StreamExt; use oak_db::OakDatabase; use oak_scan::DbScan; use oak_scan::ScanCompleted; use oak_scan::ScanRequest; use oak_scan::ScanScheduler; +use stdext::panic_message; use stdext::result::ResultExt; use stdext::spawn; use tokio::runtime::Handle; -use tokio::sync::mpsc; use tokio::sync::mpsc::unbounded_channel as tokio_unbounded_channel; use tokio::sync::oneshot; -use tokio::task::JoinHandle; use tower_lsp_server::jsonrpc; use tower_lsp_server::ls_types as lsp_types; use tower_lsp_server::ls_types::Diagnostic; @@ -42,6 +37,10 @@ use tower_lsp_server::Client; use super::backend::RequestResponse; use crate::console::ConsoleNotification; use crate::lsp; +use crate::lsp::analysis; +use crate::lsp::analysis::AnalysisPool; +use crate::lsp::analysis::DiagnosticsReady; +use crate::lsp::analysis::DiagnosticsState; use crate::lsp::backend::LspError; use crate::lsp::backend::LspMessage; use crate::lsp::backend::LspNotification; @@ -49,20 +48,18 @@ use crate::lsp::backend::LspRequest; use crate::lsp::backend::LspResponse; use crate::lsp::backend::LspResult; use crate::lsp::capabilities::Capabilities; -use crate::lsp::diagnostics::generate_diagnostics; use crate::lsp::handlers; -use crate::lsp::indexer; -use crate::lsp::open_file::OpenFile; +use crate::lsp::io_pool::IoPool; use crate::lsp::sources::OakSourceHandler; use crate::lsp::sources::SourceCompleted; use crate::lsp::sources::SourceHandler; +use crate::lsp::sources::SourceResponse; use crate::lsp::sources::SourceScheduler; use crate::lsp::state::WorldState; -use crate::lsp::state::WorldStateSnapshot; use crate::lsp::state_handlers; use crate::lsp::state_handlers::ConsoleInputs; use crate::lsp::traits::url::UriExt; -use crate::url::FilePathExt; +use crate::lsp::watchdog::Watchdog; pub(crate) type TokioUnboundedSender = tokio::sync::mpsc::UnboundedSender; pub(crate) type TokioUnboundedReceiver = tokio::sync::mpsc::UnboundedReceiver; @@ -84,20 +81,6 @@ static AUXILIARY_EVENT_TX: RwLock>> pub static LSP_HAS_CRASHED: AtomicBool = AtomicBool::new(false); -// This is the syntax for trait aliases until an official one is stabilised. -// This alias is for the future of a `JoinHandle>` -trait AnyhowJoinHandleFut: - future::Future, tokio::task::JoinError>> -{ -} -impl AnyhowJoinHandleFut for F where - F: future::Future, tokio::task::JoinError>> -{ -} - -// Alias for a list of join handle futures -type TaskList = futures::stream::FuturesUnordered + Send>>>; - #[derive(Debug)] #[expect(clippy::large_enum_variant)] pub(crate) enum Event { @@ -105,6 +88,7 @@ pub(crate) enum Event { Kernel(KernelNotification), OakScanCompleted(ScanCompleted), SourceCompleted(SourceCompleted), + DiagnosticsReady(DiagnosticsReady), } #[derive(Debug)] @@ -135,7 +119,6 @@ pub(crate) struct DidCloseVirtualDocumentParams { pub(crate) enum AuxiliaryEvent { Log(lsp_types::MessageType, String), PublishDiagnostics(DiagnosticsPublication), - SpawnedTask(JoinHandle>>), Shutdown, } @@ -148,8 +131,8 @@ pub(crate) enum AuxiliaryEvent { /// construction. pub(crate) struct GlobalState { /// The global world state containing all inputs for LSP analysis lives - /// here. The dispatcher provides refs, exclusive refs, or snapshots - /// (clones) to handlers. + /// here. The dispatcher provides refs, exclusive refs, or snapshots to + /// handlers. world: WorldState, /// The non-cloneable, per-session LSP state. Only used in exclusive ref @@ -183,8 +166,8 @@ pub(crate) struct LoopHandles { } /// Non-cloneable, per-session state mutated only by exclusive handlers. -/// Sits alongside [`WorldState`] (which is cloneable for snapshot -/// handlers); state that can't be cloned lives here instead. +/// Sits alongside [`WorldState`], which the main loop owns. State that can't +/// travel with a snapshot lives here instead. pub(crate) struct LspState { /// Capabilities negotiated with the client pub(crate) capabilities: Capabilities, @@ -200,6 +183,27 @@ pub(crate) struct LspState { /// Scheduler of [crate::lsp::sources::SourceRequest]s. Scheduling and source /// consumption all happen from the main loop. pub(crate) source_scheduler: SourceScheduler, + + /// Threads running diagnostics and index warmup. The only executor a db + /// snapshot is allowed on, see [`crate::lsp::analysis`]. + pub(crate) analysis_pool: AnalysisPool, + + /// Per-file generation bookkeeping the main loop uses to drop a + /// diagnostics result superseded by a newer refresh. See + /// [`crate::lsp::analysis::DiagnosticsState`]. + pub(crate) diagnostics: DiagnosticsState, + + /// Threads running workspace scans. + pub(crate) scan_pool: IoPool, + + /// Threads running package source fetches. Separate from [`Self::scan_pool`] + /// so a startup scan never queues behind a download. + pub(crate) source_pool: IoPool, + + /// Detects a tick that arms and never disarms, usually a write parked + /// behind a background task that can't drop its db snapshot. See + /// [`crate::lsp::watchdog`]. + pub(crate) watchdog: Watchdog, } impl LspState { @@ -212,6 +216,17 @@ impl LspState { console_notification_tx, oak_scheduler: ScanScheduler::new(), source_scheduler, + analysis_pool: AnalysisPool::new(), + diagnostics: DiagnosticsState::default(), + // Stack size: `ScanRequest::run()` walks the filesystem with + // `ignore::Walk` and `WalkDir`, both iterative, and parses + // DESCRIPTION line by line. + scan_pool: IoPool::new("oak-scan", 1, stdext::SMALL_STACK_SIZE), + // Two threads, so we never have more than two package downloads in flight. + // Full stack because a fetch runs a rustls handshake, zstd and tar + // decoding, and an R subprocess. + source_pool: IoPool::new("oak-source", 2, stdext::DEFAULT_STACK_SIZE), + watchdog: Watchdog::new(), } } } @@ -224,11 +239,10 @@ impl LspState { /// /// The auxiliary loop currently handles: /// - Log messages. -/// - Joining of spawned blocking tasks to relay any errors or panics to the LSP log. +/// - Diagnostics publication. struct AuxiliaryState { client: Client, auxiliary_event_rx: TokioUnboundedReceiver, - tasks: TaskList>, /// Last non-empty diagnostics published per file. A refresh re-runs every /// open file, but most runs produce the same result, so we skip the publish /// when it matches what the client already has. @@ -376,10 +390,11 @@ impl GlobalState { /// run these concurrently but we run these one handler at a time for simplicity. /// - When concurrent handlers are needed for performance reason (one tick /// of the main loop should be as fast as possible to increase throughput) - /// they are spawned on blocking threads and provided a snapshot (clone) of - /// the state. + /// they run on the [`crate::lsp::analysis`] pool over a snapshot of the + /// state. async fn handle_event(&mut self, event: Event) -> anyhow::Result<()> { let loop_tick = std::time::Instant::now(); + let _tick = self.lsp_state.watchdog.tick(self.world.db.outstanding_holds()); // Diagnostics read the oak database (workspace symbols, imports, // resolved definitions), so any handler that writes to oak invalidates @@ -554,22 +569,48 @@ impl GlobalState { n_holds = self.world.db.outstanding_holds(), ); - dispatch_scan_requests(&self.events_tx, followups); + dispatch_scan_requests(&self.lsp_state.scan_pool, &self.events_tx, followups); // Warm the workspace index once the scan settles. Editor // writes don't need to re-warm: they imply an open document, // and the diagnostics passes they trigger force the same // memos. if !self.lsp_state.oak_scheduler.has_pending_scans() { - warm_workspace_index(self.world.snapshot()); + analysis::warm_workspace_index(&self.world, &self.lsp_state.analysis_pool); } }, Event::SourceCompleted(SourceCompleted { package, response }) => { + let outcome = match &response { + SourceResponse::Success(_) => "completed", + SourceResponse::Failure => "failed", + }; + lsp::log_info!( + "Source fetch for package {name} {outcome}", + name = package.name(&self.world.db) + ); + if let Some(directory) = self.lsp_state.source_scheduler.finish(package, response) { self.world.db.set_package_sources(package, &directory); } }, + + Event::DiagnosticsReady(DiagnosticsReady { generation, publication }) => { + lsp::log_info!( + "Received `DiagnosticsReady` for {}", + publication.uri.as_str() + ); + + if self.lsp_state.diagnostics.accept(&publication.path, generation) { + lsp::publish_diagnostics(publication); + } else { + let path = &publication.path; + let published = self.lsp_state.diagnostics.published_generation(path); + tracing::trace!( + "Dropping stale diagnostics for {path}: generation {generation} is older than published generation {published:?}" + ); + } + }, } lsp::log_info!("Finished handling event in {}ms", loop_tick.elapsed().as_millis()); @@ -580,35 +621,20 @@ impl GlobalState { if salsa::plumbing::current_revision(&self.world.db) != old_revision { lsp::log_info!("World state revision advanced"); - diagnostics_refresh_all(&self.world); - self.lsp_state - .source_scheduler - .schedule(&self.world.db, &self.events_tx); + self.lsp_state.diagnostics.refresh_all( + &self.world, + &self.lsp_state.analysis_pool, + &self.events_tx, + ); + self.lsp_state.source_scheduler.schedule( + &self.world.db, + &self.lsp_state.source_pool, + &self.events_tx, + ); } Ok(()) } - - #[allow(dead_code)] // Currently unused - /// Spawn blocking thread for LSP request handler - /// - /// Use this for handlers that might take too long to handle on the main - /// loop and negatively affect throughput. - /// - /// The LSP protocol allows concurrent handling as long as it doesn't affect - /// correctness of responses. For instance handlers that only inspect the - /// world state could be run concurrently. On the other hand, handlers that - /// manipulate documents (e.g. formatting or refactoring) should not. - fn spawn_handler( - response_tx: TokioUnboundedSender, - handler: Handler, - into_lsp_response: impl FnOnce(T) -> LspResponse + Send + 'static, - ) where - Handler: FnOnce() -> LspResult, - Handler: Send + 'static, - { - lsp::spawn_blocking(move || respond(response_tx, handler, into_lsp_response).and(Ok(None))) - } } /// Build the LSP's [`SourceHandler`], or `None` to disable source fetching @@ -651,25 +677,40 @@ impl GlobalState { } } + /// Run a single `event` through the real `handle_event`, without pumping + /// followups. Tests that gate a subsystem need this because + /// `handle_event_to_quiescence` would wait for the gated work to finish. + pub(crate) async fn handle_event_once(&mut self, event: Event) { + self.handle_event(event).await.unwrap(); + } + + /// Pump events until no oak scan is pending, ignoring pending source + /// requests. + pub(crate) async fn pump_scans_to_quiescence(&mut self) { + while self.lsp_state.oak_scheduler.has_pending_scans() { + let event = self.next_event().await; + self.handle_event(event).await.unwrap(); + } + } + pub(crate) fn world(&self) -> &WorldState { &self.world } } -/// Spawn each [`ScanRequest`] on a blocking task. Each task runs the -/// pure-I/O [`ScanRequest::run`] and ships the [`ScanCompleted`] back -/// to the main loop as [`Event::OakScanCompleted`], where the scheduler -/// then applies it. +/// Run each [`ScanRequest`] on `pool`. Each job runs the pure-I/O +/// [`ScanRequest::run`] and ships the [`ScanCompleted`] back to the main loop as +/// [`Event::OakScanCompleted`], where the scheduler then applies it. pub(super) fn dispatch_scan_requests( + pool: &IoPool, events_tx: &TokioUnboundedSender, requests: Vec, ) { for req in requests { let tx = events_tx.clone(); - spawn_blocking(move || { + pool.submit(move || { let scan = req.run(); tx.send(Event::OakScanCompleted(scan)).log_err(); - Ok(None) }); } } @@ -710,13 +751,7 @@ fn respond( // Set global crash flag to disable the LSP LSP_HAS_CRASHED.store(true, Ordering::Release); - let msg: String = if let Some(msg) = err.downcast_ref::<&str>() { - msg.to_string() - } else if let Some(msg) = err.downcast_ref::() { - msg.clone() - } else { - String::from("Couldn't retrieve the message.") - }; + let msg = panic_message(err.as_ref()); // This creates an uninformative backtrace that is reported in the // LSP logs. Note that the relevant backtrace is the one created by @@ -748,9 +783,6 @@ fn respond( out } -// Needed for spawning the loop -unsafe impl Sync for AuxiliaryState {} - impl AuxiliaryState { fn new(client: Client) -> Self { // Channels for communication with the auxiliary loop @@ -768,22 +800,9 @@ impl AuxiliaryState { *tx = Some(auxiliary_event_tx); } - // List of pending tasks for which we manage the lifecycle (mainly relay - // errors and panics) - let tasks = futures::stream::FuturesUnordered::new(); - - // Prevent the stream from ever being empty so that `tasks.next()` never - // resolves to `None` - let pending = - tokio::task::spawn(future::pending::>>()); - let pending = - Box::pin(pending) as Pin> + Send>>; - tasks.push(pending); - Self { client, auxiliary_event_rx, - tasks, published_diagnostics: HashMap::new(), } } @@ -796,7 +815,6 @@ impl AuxiliaryState { loop { match self.next_event().await { AuxiliaryEvent::Log(level, message) => self.log(level, message).await, - AuxiliaryEvent::SpawnedTask(handle) => self.tasks.push(Box::pin(handle)), AuxiliaryEvent::PublishDiagnostics(publication) => { self.publish_diagnostics(publication).await }, @@ -806,30 +824,16 @@ impl AuxiliaryState { } async fn next_event(&mut self) -> AuxiliaryEvent { - loop { - tokio::select! { - event = self.auxiliary_event_rx.recv() => match event { - // Because of the way we communicate with the auxiliary loop - // via global state, the channel may become closed if a new - // LSP session is started in the process. This normally - // should not happen but for now we have to be defensive - // against this situation, see: - // https://github.com/posit-dev/ark/issues/622 - // https://github.com/posit-dev/positron/issues/5321 - Some(event) => return event, - None => return AuxiliaryEvent::Shutdown, - }, - - handle = self.tasks.next() => match handle.unwrap() { - // A joined task returned an event for us, handle it - Ok(Ok(Some(event))) => return event, - - // Otherwise relay any errors and loop back into select - Err(err) => self.log_error(format!("A task panicked:\n{err:?}")).await, - Ok(Err(err)) => self.log_error(format!("A task failed:\n{err:?}")).await, - _ => (), - }, - } + match self.auxiliary_event_rx.recv().await { + // Because of the way we communicate with the auxiliary loop + // via global state, the channel may become closed if a new + // LSP session is started in the process. This normally + // should not happen but for now we have to be defensive + // against this situation, see: + // https://github.com/posit-dev/ark/issues/622 + // https://github.com/posit-dev/positron/issues/5321 + Some(event) => event, + None => AuxiliaryEvent::Shutdown, } } @@ -869,9 +873,6 @@ impl AuxiliaryState { async fn log(&self, level: MessageType, message: String) { self.client.log_message(level, message).await } - async fn log_error(&self, message: String) { - self.client.log_message(MessageType::ERROR, message).await - } } fn with_auxiliary_tx(f: F) -> T @@ -941,32 +942,6 @@ pub(crate) fn log(level: lsp_types::MessageType, message: String) { }; } -/// Spawn a blocking task -/// -/// This runs tasks that do semantic analysis on a separate thread pool to avoid -/// blocking the main loop. -/// -/// Can optionally return an event for the auxiliary loop (i.e. a log message or -/// diagnostics publication). -/// -/// Salsa cancellation is handled here so callers don't have to. A `set_*` on -/// the main loop cancels concurrent oak queries by unwinding with `Cancelled`. -/// We swallow that into `Ok(None)`, so a cancelled task is a quiet no-op -/// instead of a logged "task panicked". The write that cancelled it enqueues -/// its own follow-up. Any other panic still surfaces on join. -pub(crate) fn spawn_blocking(handler: Handler) -where - Handler: FnOnce() -> anyhow::Result>, - Handler: Send + 'static, -{ - let handle = - tokio::task::spawn_blocking(move || catch_cancellation(handler).unwrap_or(Ok(None))); - - // Send the join handle to the auxiliary loop so it can log any errors - // or panics - send_auxiliary(AuxiliaryEvent::SpawnedTask(handle)); -} - pub(crate) fn publish_diagnostics(publication: DiagnosticsPublication) { send_auxiliary(AuxiliaryEvent::PublishDiagnostics(publication)); } @@ -994,15 +969,6 @@ impl std::fmt::Debug for TraceKernelNotification<'_> { } } -#[derive(Debug)] -pub(crate) struct RefreshDiagnosticsTask { - /// Snapshot carrying the live oak plus the session context the diagnostics - /// walk reads. See [`WorldState::diagnostics_snapshot`]. - state: WorldStateSnapshot, - /// The file to diagnose, built against the live oak at enqueue time. - file: OpenFile, -} - #[derive(Debug)] pub(crate) struct DiagnosticsPublication { /// Identity for the dedup cache. Two spellings of the same document @@ -1014,138 +980,6 @@ pub(crate) struct DiagnosticsPublication { pub(crate) version: Option, } -static DIAGNOSTICS_QUEUE: LazyLock> = - LazyLock::new(|| { - let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); - tokio::spawn(process_diagnostics_queue(rx)); - tx - }); - -/// Process diagnostics refresh tasks. -/// -/// Tasks are batched and deduplicated per file (only the last task per file is -/// processed), so stale-version diagnostics get superseded within a batch. -/// -/// Batches can't publish out of order. Each pass holds a db snapshot, and a -/// salsa write blocks until all snapshots drop, so by the time the write -/// completes and the newer batch is enqueued, any older pass has either unwound -/// with `Cancelled` or already produced its result. Changes to state that lives -/// outside oak (console inputs, diagnostics config) get the same barrier by -/// advancing the revision synthetically (see [`WorldState::bump_revision`]). -async fn process_diagnostics_queue(mut rx: mpsc::UnboundedReceiver) { - while let Some(task) = rx.recv().await { - let mut batch = vec![task]; - while let Ok(task) = rx.try_recv() { - batch.push(task); - } - process_diagnostics_batch(batch); - } - lsp::log_warn!("process_diagnostics_queue: channel closed, task exiting"); -} - -fn process_diagnostics_batch(batch: Vec) { - // Deduplicate tasks by keeping only the last one for each file, which is - // effectively a way of cancelling diagnostics tasks for outdated documents. - let batch: HashMap = batch - .into_iter() - .map(|task| (task.file.file().path(task.state.db()).clone(), task)) - .collect(); - - tracing::trace!("Processing {n} diagnostic tasks", n = batch.len()); - lsp::log_info!("Processing {n} diagnostic tasks", n = batch.len()); - - // Each file is its own blocking task. `spawn_blocking()` catches salsa - // cancellation, so a pass cancelled by a concurrent edit just produces no - // event. The publish happens via the returned [`AuxiliaryEvent`]. - for (_path, task) in batch { - lsp::spawn_blocking(move || { - let publication = refresh_diagnostics(task); - Ok(Some(AuxiliaryEvent::PublishDiagnostics(publication))) - }); - } -} - -fn refresh_diagnostics(task: RefreshDiagnosticsTask) -> DiagnosticsPublication { - let RefreshDiagnosticsTask { file, state } = task; - let path = file.file().path(state.db()).clone(); - let uri = file.wire_uri().clone(); - let version = file.version(); - let _span = tracing::info_span!("diagnostics_refresh", uri = %uri.as_str()).entered(); - - // Special case testthat-specific behaviour. This is a simple stopgap - // approach that has some false positives (e.g. when we work on testthat - // itself the flag will always be true), but that shouldn't have much - // practical impact. - let testthat = path - .as_path() - .is_some_and(|path| path.components().any(|c| c.as_str() == "testthat")); - - let now = std::time::Instant::now(); - lsp::log_info!("Generating diagnostics for file: {}", uri.as_str()); - - let diagnostics = generate_diagnostics(file.file(), state, testthat); - - lsp::log_info!( - "Finished diagnostics for file: {} in {:.0?}", - uri.as_str(), - now.elapsed() - ); - - DiagnosticsPublication { - path, - uri, - diagnostics, - version, - } -} - -/// Run `f`, swallowing a salsa cancellation as `None`. Any other panic propagates. -fn catch_cancellation(f: impl FnOnce() -> T) -> Option { - salsa::Cancelled::catch(std::panic::AssertUnwindSafe(f)).ok() -} - -pub(crate) fn diagnostics_refresh_all(state: &WorldState) { - tracing::trace!( - "Refreshing diagnostics for {n} documents", - n = state.open_files.len() - ); - - for file in state.open_files.values() { - if !file.file().path(&state.db).should_diagnose() { - continue; - } - - DIAGNOSTICS_QUEUE - .send(RefreshDiagnosticsTask { - file: file.clone(), - state: state.diagnostics_snapshot(), - }) - .unwrap_or_else(|err| lsp::log_error!("Failed to queue diagnostics refresh: {err}")); - } -} - -/// Build the per-file workspace symbol indexes on a background thread so -/// main-loop consumers triggered by the user (workspace symbols, workspace -/// completions) find them already computed. The first run after a workspace -/// scan does the real work, parsing and walking each file. Later runs only -/// revalidate the per-file memos. -/// -/// Mirrors rust-analyzer's cache warming: spawned when a workspace scan -/// settles, the analogue of r-a's transitions to quiescence (initial VFS scan, -/// workspace reload, etc). Unlike r-a we don't restart a warmup that gets -/// cancelled (`spawn_blocking()` swallows the unwind). A cancelling write can -/// only come from an editor buffer, so a document is open, and the diagnostics -/// passes spawned by that same write force the same memos and finish the job. -fn warm_workspace_index(state: WorldStateSnapshot) { - spawn_blocking(move || { - let now = std::time::Instant::now(); - lsp::log_info!("Starting workspace index warmup"); - indexer::warm(state.db()); - lsp::log_info!("Finished workspace index warmup ({:.0?})", now.elapsed()); - Ok(None) - }) -} - #[cfg(test)] mod tests { use aether_path::FilePath; @@ -1153,46 +987,14 @@ mod tests { use tower_lsp_server::jsonrpc; use url::Url; - use super::catch_cancellation; - use super::refresh_diagnostics; use super::respond; use super::tokio_unbounded_channel; - use super::RefreshDiagnosticsTask; use crate::lsp::backend::LspError; use crate::lsp::backend::LspResponse; use crate::lsp::backend::RequestResponse; use crate::lsp::state::WorldState; use crate::lsp::traits::url::UrlExt; - /// A salsa cancellation during the pass is swallowed into `None` by - /// `catch_cancellation`, the wrapper `spawn_blocking` applies to every task, - /// rather than unwinding and killing the task. - /// - /// `cancellation_token().cancel()` arms local cancellation on the snapshot's - /// oak, so the first salsa query in `generate_diagnostics` (the `tree_sitter` - /// fetch) unwinds with `salsa::Cancelled`, the same payload a concurrent - /// `set_*` produces. The unwind fires before any R, so no `r_task` here. - #[test] - fn test_cancelled_diagnostics_pass_is_caught() { - let mut state = WorldState::default(); - let uri = Url::parse("file:///test.R").unwrap(); - let code = "foo"; - let file = state - .db - .upsert_editor(FilePath::from_url(&uri), code.to_string()); - state.insert_open_file(uri.to_uri().unwrap(), FilePath::from_url(&uri), file, None); - - let file = state.open_file(&FilePath::from_url(&uri)).unwrap().clone(); - let snapshot = state.diagnostics_snapshot(); - snapshot.cancellation_token().cancel(); - - let task = RefreshDiagnosticsTask { - file, - state: snapshot, - }; - assert!(catch_cancellation(|| refresh_diagnostics(task)).is_none()); - } - /// A `salsa::Cancelled` re-raised out of a request handler (by `r_task`, /// after catching it on the R thread) must not crash the LSP. `respond` /// recognises the payload and answers `ContentModified` so the client @@ -1207,7 +1009,7 @@ mod tests { state.insert_open_file(uri.to_uri().unwrap(), FilePath::from_url(&uri), file, None); let file = state.open_file(&FilePath::from_url(&uri)).unwrap().clone(); - let snapshot = state.diagnostics_snapshot(); + let snapshot = state.snapshot(); snapshot.cancellation_token().cancel(); let (response_tx, mut response_rx) = tokio_unbounded_channel::(); diff --git a/crates/ark/src/lsp/sources.rs b/crates/ark/src/lsp/sources.rs index d8dc4204f4..49b3d57d33 100644 --- a/crates/ark/src/lsp/sources.rs +++ b/crates/ark/src/lsp/sources.rs @@ -11,6 +11,8 @@ use oak_source::SourceCache; use oak_srcref::SrcrefCache; use stdext::result::ResultExt; +use crate::lsp; +use crate::lsp::io_pool::IoPool; use crate::lsp::main_loop::Event; use crate::lsp::main_loop::TokioUnboundedSender; @@ -152,7 +154,19 @@ impl SourceScheduler { } } - pub(crate) fn schedule(&mut self, db: &dyn Db, events_tx: &TokioUnboundedSender) { + /// Run a fetch for each package we haven't seen before on `pool`, shipping + /// each [`SourceResponse`] back to the main loop as + /// [`Event::SourceCompleted`], where [`Self::finish`] then applies it. + /// + /// The job owns no db handle. A download can't be interrupted by a Salsa + /// cancellation, so a handle in `pool`'s queue would hold up the next + /// main-loop write for the whole fetch. + pub(crate) fn schedule( + &mut self, + db: &dyn Db, + pool: &IoPool, + events_tx: &TokioUnboundedSender, + ) { let Some(handler) = &self.handler else { return; }; @@ -177,10 +191,16 @@ impl SourceScheduler { let handler = Arc::clone(handler); let tx = events_tx.clone(); - // Mark as `Pending` just before launching the tokio task + // Mark as `Pending` just before launching the job self.state.insert(package, SourceState::Pending); - crate::lsp::spawn_blocking(move || { + lsp::log_info!( + "Fetching sources for package {name}, {n} pending", + name = request.name(), + n = self.pending_count(), + ); + + pool.submit(move || { let response = handler.handle(&request); tx.send(Event::SourceCompleted(SourceCompleted { @@ -188,8 +208,6 @@ impl SourceScheduler { response, })) .log_err(); - - Ok(None) }); } } @@ -203,6 +221,14 @@ impl SourceScheduler { } } + /// How many source requests are in flight + fn pending_count(&self) -> usize { + self.state + .values() + .filter(|state| matches!(state, SourceState::Pending)) + .count() + } + /// Whether any source request is in flight. Allows tests to deterministically "wait" /// for pending source requests to finish. #[cfg(test)] diff --git a/crates/ark/src/lsp/state.rs b/crates/ark/src/lsp/state.rs index b3c2e61be9..077281d704 100644 --- a/crates/ark/src/lsp/state.rs +++ b/crates/ark/src/lsp/state.rs @@ -9,20 +9,20 @@ use salsa::Database; use tower_lsp_server::ls_types::Uri; use crate::lsp::config::LspConfig; -use crate::lsp::db::ArkDb; use crate::lsp::open_file::OpenFile; use crate::lsp::traits::url::UrlExt; -#[derive(Clone, Default, Debug)] +#[derive(Default, Debug)] /// The world state, i.e. all the inputs necessary for analysing or refactoring /// code. This is a pure value. There is no interior mutability in this data -/// structure. It can be cloned and safely sent to other threads. +/// structure. /// /// The main loop owns and mutates this. Background readers get a -/// [`WorldStateSnapshot`] instead, which only lends its database out as -/// `&dyn ArkDb`. This prevents background threads from reaching a Salsa input -/// setter. See [`Self::diagnostics_snapshot`] and [`Self::snapshot`]. This split -/// mirrors rust-analyzer's `GlobalState` and `GlobalStateSnapshot`. +/// [`crate::lsp::analysis::WorldStateSnapshot`] instead, which only lends its +/// database out as `&dyn ArkDb`, so a background thread can't reach a Salsa +/// input setter. Snapshots are minted in [`crate::lsp::analysis`] and nowhere +/// else. This split mirrors rust-analyzer's `GlobalState` and +/// `GlobalStateSnapshot`. pub(crate) struct WorldState { /// Salsa input tree for Oak queries. pub(crate) db: OakDatabase, @@ -81,32 +81,6 @@ impl WorldState { } } - /// Full read-only snapshot for a background reader that needs more than the - /// db, e.g. completions read the workspace. Same shape as `self.clone()`, - /// but the db is only reachable as `&dyn ArkDb`. - pub(crate) fn snapshot(&self) -> WorldStateSnapshot { - WorldStateSnapshot { - db: self.db.clone(), - console_scopes: self.console_scopes.clone(), - installed_packages: self.installed_packages.clone(), - config: self.config.clone(), - workspace: self.workspace.clone(), - } - } - - /// Trimmed read-only snapshot for the diagnostics worker, which runs off - /// the main loop and queries oak. Drops the workspace map the diagnostics - /// pass doesn't read. - pub(crate) fn diagnostics_snapshot(&self) -> WorldStateSnapshot { - WorldStateSnapshot { - db: self.db.clone(), - console_scopes: self.console_scopes.clone(), - installed_packages: self.installed_packages.clone(), - config: self.config.clone(), - workspace: Workspace::default(), - } - } - /// Advance the oak revision without changing any oak input. /// /// Currently used for state that lives on `WorldState` but not in the Oak @@ -170,37 +144,6 @@ impl WorldState { } } -/// Read-only snapshot of [`WorldState`] handed to background readers (e.g. -/// diagnostics), so a reader thread can't reach Salsa input setters. Carries only -/// the fields readers actually use. Mirrors rust-analyzer's -/// `GlobalStateSnapshot`. -#[derive(Clone, Debug)] -pub(crate) struct WorldStateSnapshot { - /// Private so readers can only reach it through [`Self::db`]. - db: OakDatabase, - pub(crate) workspace: Workspace, - pub(crate) console_scopes: Vec>, - pub(crate) installed_packages: Vec, - pub(crate) config: LspConfig, -} - -impl WorldStateSnapshot { - /// Read-only access to the database. Returns `&dyn ArkDb` rather than - /// `&OakDatabase` because `dyn ArkDb` is unsized, so a reader can't - /// `.clone()` its way to an owned database and call setters on it. - pub(crate) fn db(&self) -> &dyn ArkDb { - &self.db - } - - /// The database's salsa cancellation token. Read-side only: it observes and - /// arms cancellation, it doesn't mutate any input. Only cancellation tests - /// arm it by hand. - #[cfg(test)] - pub(crate) fn cancellation_token(&self) -> salsa::CancellationToken { - salsa::Database::cancellation_token(&self.db) - } -} - /// The wire `Uri` of every open buffer, paired with the [`FilePath`] it's keyed /// on so a caller can look the buffer back up without converting. pub(crate) fn open_file_wire_uris(state: &WorldState) -> Vec<(FilePath, Uri)> { diff --git a/crates/ark/src/lsp/state_handlers.rs b/crates/ark/src/lsp/state_handlers.rs index c2154848c2..9e45454e6b 100644 --- a/crates/ark/src/lsp/state_handlers.rs +++ b/crates/ark/src/lsp/state_handlers.rs @@ -104,7 +104,7 @@ pub(crate) fn initialize( &to_std_paths(&state.workspace.folders), &editor_owned, ); - dispatch_scan_requests(events_tx, requests); + dispatch_scan_requests(&lsp_state.scan_pool, events_tx, requests); let result = InitializeResult { server_info: Some(ServerInfo { @@ -313,7 +313,7 @@ pub(crate) fn did_change_watched_files( lsp_state .oak_scheduler .apply_watcher_events(&mut state.db, events, &editor_owned); - dispatch_scan_requests(events_tx, requests); + dispatch_scan_requests(&lsp_state.scan_pool, events_tx, requests); Ok(()) } @@ -368,7 +368,7 @@ pub(crate) fn did_change_workspace_folders( &to_std_paths(&state.workspace.folders), &editor_owned, ); - dispatch_scan_requests(events_tx, requests); + dispatch_scan_requests(&lsp_state.scan_pool, events_tx, requests); Ok(()) } diff --git a/crates/ark/src/lsp/tests/diagnostics.rs b/crates/ark/src/lsp/tests/diagnostics.rs index 3ba95f8088..77c04ace00 100644 --- a/crates/ark/src/lsp/tests/diagnostics.rs +++ b/crates/ark/src/lsp/tests/diagnostics.rs @@ -24,15 +24,15 @@ fn test_diagnostics_published_through_refresh_snapshot() { .upsert_editor(FilePath::from_url(&uri), code.to_string()); state.insert_open_file(uri.to_uri().unwrap(), FilePath::from_url(&uri), file, None); - // Mirror `diagnostics_refresh_all`: fetch the `File` from the live - // state, then hand the worker the `diagnostics_snapshot`. The snapshot's - // oak must still serve that file. + // Mirror `DiagnosticsState::refresh_all`: fetch the `File` from the + // live state, then hand the worker a snapshot. The snapshot's oak must + // still serve that file. let file = state .open_file(&FilePath::from_url(&uri)) .expect("file is open in live state") .file(); - let snapshot = state.diagnostics_snapshot(); + let snapshot = state.snapshot(); generate_diagnostics(file, snapshot, false) }); diff --git a/crates/ark/src/lsp/tests/main_loop.rs b/crates/ark/src/lsp/tests/main_loop.rs index cc80bb87ca..9199c68445 100644 --- a/crates/ark/src/lsp/tests/main_loop.rs +++ b/crates/ark/src/lsp/tests/main_loop.rs @@ -2,18 +2,28 @@ //! //! Where the handler tests in [`super::state_handlers`] reconstruct the scan //! pump by hand, this one feeds an event through the production `handle_event` -//! and lets the loop dispatch the scan, run it on a blocking task, route the +//! and lets the loop dispatch the scan, run it on the scan pool, route the //! [`Event::OakScanCompleted`] back, and apply it. So it pins the main loop's //! own wiring: which arm calls which handler, and the apply-and-redispatch //! step. The scheduler's policy is unit tested without tokio in `oak_scan`. +use std::collections::HashMap; +use std::sync::Arc; + use oak_db::DbInputs; use oak_db::OakDatabase; +use oak_scan::DbScan; use tower_lsp_server::ls_types::DidChangeWorkspaceFoldersParams; use tower_lsp_server::ls_types::Uri; use tower_lsp_server::ls_types::WorkspaceFolder; use tower_lsp_server::ls_types::WorkspaceFoldersChangeEvent; +use super::source_handler::gate; +use super::source_handler::TestBehavior; +use super::source_handler::TestSourceHandler; +use super::utils::did_change; +use super::utils::did_change_workspace_folders; +use super::utils::did_open; use super::utils::test_client; use super::utils::write_sources; use super::utils::DescriptionWriter; @@ -72,3 +82,88 @@ async fn test_workspace_folder_scan_drives_through_main_loop() { assert_eq!(packages[0].name(db), "pkg"); assert_eq!(packages[0].files(db).len(), 1); } + +/// Db-holding work (diagnostics, index warmup) and unbounded I/O (package source +/// fetches) run on separate executors, so saturating the source pool can't stall a +/// main-loop write: the analysis pool stays free to drain the queued diagnostics +/// snapshot the write is waiting on. Gates five packages, one more than +/// `MAX_ANALYSIS_THREADS`, so a regression that merged the two executors back together +/// would leave every thread parked instead of a few free ones, and the write would park +/// behind the pinned snapshot. If that happens, the watchdog (`crate::lsp::watchdog`) +/// aborts the test with a diagnosis instead of hanging to the harness timeout. +#[tokio::test] +async fn test_main_loop_write_survives_saturated_source_pool() { + let _aux = init_aux_for_test(); + + // One shared "entered" sender: with five gates and a 2-thread source pool, only two + // can ever be inside `handle()` at once, but which two is unpredictable. + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + let donors = ["donor1", "donor2", "donor3", "donor4", "donor5"]; + + let mut behavior = HashMap::new(); + let mut releases = Vec::new(); + for name in donors { + let (gate, release) = gate(entered_tx.clone()); + behavior.insert(name.to_string(), TestBehavior::Gated(gate)); + releases.push(release); + } + let handler = Arc::new(TestSourceHandler::new(behavior)); + + let lib = tempfile::tempdir().unwrap(); + for name in donors { + DescriptionWriter::new() + .package(name) + .version("0.0.0") + .built("dummy") + .write(&lib.path().join(name)); + } + let mut db = OakDatabase::new(); + db.set_library_paths(&[lib.path().to_path_buf()]); + + let mut state = GlobalState::from_parts( + test_client(), + WorldState::new(db), + LspState::new( + tokio::sync::mpsc::unbounded_channel().0, + SourceScheduler::new(Some(handler)), + ), + ); + + // A workspace package using all five library packages via `::`, so the scan hands + // the scheduler five dependencies to fetch. + let workspace = tempfile::tempdir().unwrap(); + let myproj = workspace.path().join("myproj"); + DescriptionWriter::new() + .package("myproj") + .version("0.0.0") + .write(&myproj); + let uses: String = donors + .iter() + .map(|name| format!("{name}::foo()\n")) + .collect(); + write_sources(&myproj.join("R"), &[("use.R", &uses)]); + let script = workspace.path().join("script.R"); + + state + .handle_event_once(did_change_workspace_folders(workspace.path())) + .await; + state.pump_scans_to_quiescence().await; + + // The source pool workers are now parked in a fetch that salsa cancellation can't + // reach. Index warmup went to the analysis pool, so it isn't queued behind them. + for _ in 0..2 { + entered_rx.recv().unwrap(); + } + + // Goes through, no holds outstanding. Ends the tick by queueing a diagnostics + // pass, which needs an analysis thread to run on. + state.handle_event_once(did_open(&script, "x <- 1\n")).await; + + // The write that has to drain that pinned hold. + state + .handle_event_once(did_change(&script, "x <- 2\n", 1)) + .await; + + // Let the still-gated workers finish so the test process can exit cleanly. + drop(releases); +} diff --git a/crates/ark/src/lsp/tests/source_handler.rs b/crates/ark/src/lsp/tests/source_handler.rs index c4ef8dd163..bd60ffe035 100644 --- a/crates/ark/src/lsp/tests/source_handler.rs +++ b/crates/ark/src/lsp/tests/source_handler.rs @@ -1,4 +1,6 @@ use std::collections::HashMap; +use std::sync::mpsc::Receiver; +use std::sync::mpsc::Sender; use std::sync::Mutex; use super::utils::write_sources; @@ -25,6 +27,44 @@ pub(super) enum TestBehavior { /// and return `Success(dir)`. Success(Vec<(&'static str, &'static str)>), Failure, + /// Park on a [`Gate`] until the test releases it, then fail. Lets a test hold a + /// source worker for as long as it wants. + Gated(Gate), +} + +/// The handler side of a rendezvous with the test. `handle()` announces that it has +/// started, then parks until the test lets go. +pub(super) struct Gate { + entered_tx: Sender<()>, + release_rx: Mutex>, +} + +/// Build a gate reporting through `entered_tx`, the test's single shared sender, once +/// its `handle()` call starts running. Callers building several gates for the same test +/// pass clones of the same sender, so the test can count entries across all of them +/// without knowing which one got in first. Each gate still gets its own release channel: +/// a shared release receiver would serialise the waiters instead of letting them all +/// park at once. The gate stays shut for as long as the returned sender is alive. +pub(super) fn gate(entered_tx: Sender<()>) -> (Gate, Sender<()>) { + let (release_tx, release_rx) = std::sync::mpsc::channel(); + + let gate = Gate { + entered_tx, + release_rx: Mutex::new(release_rx), + }; + (gate, release_tx) +} + +impl Gate { + fn wait(&self) { + if self.entered_tx.send(()).is_err() { + return; + } + match self.release_rx.lock().unwrap().recv() { + // `Err` means the test dropped the release end, which also lets us through + Ok(()) | Err(_) => (), + } + } } impl TestSourceHandler { @@ -53,6 +93,10 @@ impl SourceHandler for TestSourceHandler { SourceResponse::Success(dir) }, Some(TestBehavior::Failure) => SourceResponse::Failure, + Some(TestBehavior::Gated(gate)) => { + gate.wait(); + SourceResponse::Failure + }, None => panic!("Unknown test package {}", request.name()), } } diff --git a/crates/ark/src/lsp/tests/state_handlers.rs b/crates/ark/src/lsp/tests/state_handlers.rs index b1eaa4c0f0..c2ce3b4836 100644 --- a/crates/ark/src/lsp/tests/state_handlers.rs +++ b/crates/ark/src/lsp/tests/state_handlers.rs @@ -95,9 +95,9 @@ fn did_change_workspace_folders( /// Drive a production handler that dispatches its scans through `events_tx`, /// then pump the resulting `OakScanCompleted` events to quiescence on a local /// runtime. Production does this pumping in the main loop's event handler; -/// the tests have to stand up the same machinery (tokio runtime so -/// `spawn_blocking` works, aux channel so `send_auxiliary` doesn't panic, an -/// events channel to receive completions). +/// the tests have to stand up the same machinery (tokio runtime to await the +/// completions, aux channel so `send_auxiliary` doesn't panic, an events channel +/// to receive completions). fn run_handler_to_quiescence( state: &mut WorldState, lsp_state: &mut LspState, @@ -123,7 +123,7 @@ where lsp_state .oak_scheduler .apply_scan_completed(&mut state.db, scan, &editor_owned); - dispatch_scan_requests(&events_tx, followups); + dispatch_scan_requests(&lsp_state.scan_pool, &events_tx, followups); } Ok(()) }) @@ -653,7 +653,7 @@ fn test_did_close_releases_orphan_file_to_stale() { // Init the aux channel here, after the workspace-folders churn: the // handler wrapper resets the channel each call (it stands up its own to - // satisfy `spawn_blocking`), so grab the receiver only once that's done. + // satisfy `send_auxiliary`), so grab the receiver only once that's done. let mut aux_rx = init_aux_for_test(); // Now close the buffer. File should move from orphan to stale. diff --git a/crates/ark/src/lsp/tests/utils/events.rs b/crates/ark/src/lsp/tests/utils/events.rs index bda4ea0b74..7934b5d630 100644 --- a/crates/ark/src/lsp/tests/utils/events.rs +++ b/crates/ark/src/lsp/tests/utils/events.rs @@ -1,9 +1,12 @@ use std::path::Path; +use tower_lsp_server::ls_types::DidChangeTextDocumentParams; use tower_lsp_server::ls_types::DidChangeWorkspaceFoldersParams; use tower_lsp_server::ls_types::DidOpenTextDocumentParams; +use tower_lsp_server::ls_types::TextDocumentContentChangeEvent; use tower_lsp_server::ls_types::TextDocumentItem; use tower_lsp_server::ls_types::Uri; +use tower_lsp_server::ls_types::VersionedTextDocumentIdentifier; use tower_lsp_server::ls_types::WorkspaceFolder; use tower_lsp_server::ls_types::WorkspaceFoldersChangeEvent; @@ -25,6 +28,24 @@ pub(crate) fn did_change_workspace_folders(path: &Path) -> Event { )) } +/// A whole-document change at `version`, which must be greater than the version the +/// file was opened at. +pub(crate) fn did_change(path: &Path, contents: &str, version: i32) -> Event { + Event::Lsp(LspMessage::Notification( + LspNotification::DidChangeTextDocument(DidChangeTextDocumentParams { + text_document: VersionedTextDocumentIdentifier { + uri: Uri::from_file_path(path).unwrap(), + version, + }, + content_changes: vec![TextDocumentContentChangeEvent { + range: None, + range_length: None, + text: contents.to_string(), + }], + }), + )) +} + pub(crate) fn did_open(path: &Path, contents: &str) -> Event { Event::Lsp(LspMessage::Notification( LspNotification::DidOpenTextDocument(DidOpenTextDocumentParams { diff --git a/crates/ark/src/lsp/tests/utils/mod.rs b/crates/ark/src/lsp/tests/utils/mod.rs index 8aa2c1d494..f0233dbaaa 100644 --- a/crates/ark/src/lsp/tests/utils/mod.rs +++ b/crates/ark/src/lsp/tests/utils/mod.rs @@ -6,6 +6,7 @@ use std::path::Path; use aether_path::FilePath; pub(super) use description_writer::DescriptionWriter; +pub(super) use events::did_change; pub(super) use events::did_change_workspace_folders; pub(super) use events::did_open; pub(super) use namespace_writer::NamespaceWriter; diff --git a/crates/ark/src/lsp/watchdog.rs b/crates/ark/src/lsp/watchdog.rs new file mode 100644 index 0000000000..fe9b029296 --- /dev/null +++ b/crates/ark/src/lsp/watchdog.rs @@ -0,0 +1,222 @@ +// +// watchdog.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +//! Detects a main-loop tick that arms and never disarms, usually a Salsa write +//! parked behind a background task that can't drop its snapshot. +//! `handle_event()` in `main_loop.rs` arms one [`TickGuard`] per tick, and the +//! poller thread here reports a tick still armed past the deadline. + +use std::time::Duration; +use std::time::Instant; + +use crossbeam::channel::select; +use crossbeam::channel::unbounded; +use crossbeam::channel::Receiver; +use crossbeam::channel::Sender; +use stdext::result::ResultExt; +use stdext::spawn_with_stack_size; + +use crate::lsp; + +const DEADLINE: Duration = Duration::from_secs(5); + +/// A tick arming or disarming, carried from `Watchdog`/`TickGuard` to the +/// poller thread. +enum Tick { + Armed { + tick: u64, + holds: usize, + at: Instant, + }, + Disarmed, +} + +pub(crate) struct Watchdog { + tick_tx: Sender, + next_tick: u64, + /// Causes watchdog to shut down when dropped. + _close_tx: Sender<()>, +} + +impl Watchdog { + pub(crate) fn new() -> Self { + let (tick_tx, tick_rx) = unbounded(); + let (close_tx, close_rx) = unbounded(); + + spawn_with_stack_size!("oak-watchdog", stdext::TINY_STACK_SIZE, move || poll( + tick_rx, close_rx + )); + + Self { + tick_tx, + _close_tx: close_tx, + next_tick: 0, + } + } + + /// Arm the watchdog for one tick. Dropping the returned guard disarms it + /// again, on every exit path out of `handle_event` including `?` early + /// returns and panics. + pub(crate) fn tick(&mut self, holds: usize) -> TickGuard { + self.next_tick += 1; + self.tick_tx + .send(Tick::Armed { + tick: self.next_tick, + holds, + at: Instant::now(), + }) + .log_err(); + + TickGuard { + tick_tx: self.tick_tx.clone(), + } + } +} + +/// Owns the arm for one main-loop tick. +pub(crate) struct TickGuard { + tick_tx: Sender, +} + +impl Drop for TickGuard { + fn drop(&mut self) { + self.tick_tx.send(Tick::Disarmed).log_err(); + } +} + +/// Wait for the next arm, then watch it until it disarms, rearms, or +/// `close_rx` disconnects (the watchdog was dropped). +fn poll(tick_rx: Receiver, close_rx: Receiver<()>) { + 'idle: loop { + let msg = select! { + recv(tick_rx) -> msg => msg, + recv(close_rx) -> _ => return, + }; + + let Ok(Tick::Armed { + mut tick, + mut holds, + mut at, + }) = msg + else { + // A stray `Disarmed`: nothing is armed yet, keep waiting. + continue 'idle; + }; + + // Report once `at` crosses `DEADLINE`, then every `DEADLINE` after that. + loop { + let msg = select! { + recv(tick_rx) -> msg => Some(msg), + recv(close_rx) -> _ => return, + default(DEADLINE) => None, + }; + + match msg { + Some(Ok(Tick::Disarmed)) => continue 'idle, + Some(Err(_)) => return, + None => report(tick, at.elapsed(), holds), + // Two arms without a disarm between them shouldn't happen + // (only one tick runs at a time), but if it does, start + // watching the new one instead of reporting on the stale one. + Some(Ok(Tick::Armed { + tick: new_tick, + holds: new_holds, + at: new_at, + })) => { + log::warn!("Unexpected `Tick::Armed` before `Tick::Disarmed`"); + tick = new_tick; + holds = new_holds; + at = new_at; + }, + } + } + } +} + +fn report(tick: u64, elapsed: Duration, holds: usize) { + let message = format!( + "Main loop tick {tick} has been running for {secs:.1}s with {holds} outstanding Salsa \ + db holds. Likely cause: a write parked waiting for a background reader to drop its \ + snapshot.", + secs = elapsed.as_secs_f64(), + ); + + if stdext::IS_TESTING { + // The stuck thread is the main loop, so panicking here would only unwind this + // poller thread and the test would still hang to the harness timeout with no + // information. nextest runs each test in its own process, so aborting attributes + // the failure to the right test and prints the diagnosis. + eprintln!("{message}"); + std::process::abort(); + } else { + lsp::log_error!("{message}"); + log::error!("{message}"); + } +} + +#[cfg(test)] +mod tests { + use crossbeam::channel::unbounded; + use crossbeam::channel::Receiver; + + use super::Tick; + use super::Watchdog; + + impl Watchdog { + /// Build a watchdog without spawning the poller thread, exposing the + /// tick channel so tests can assert on the messages `tick()` and + /// `TickGuard::drop()` send. + fn new_test() -> (Self, Receiver) { + let (tick_tx, tick_rx) = unbounded(); + let (close_tx, _close_rx) = unbounded(); + + let watchdog = Self { + tick_tx, + _close_tx: close_tx, + next_tick: 0, + }; + (watchdog, tick_rx) + } + } + + #[test] + fn test_tick_arms_and_disarms() { + let (mut watchdog, tick_rx) = Watchdog::new_test(); + + let guard = watchdog.tick(3); + let Ok(Tick::Armed { holds, .. }) = tick_rx.try_recv() else { + panic!("expected `Armed`"); + }; + assert_eq!(holds, 3); + + drop(guard); + assert!(matches!(tick_rx.try_recv(), Ok(Tick::Disarmed))); + } + + #[test] + fn test_consecutive_ticks_increase() { + let (mut watchdog, tick_rx) = Watchdog::new_test(); + + let _first = watchdog.tick(0); + let Ok(Tick::Armed { + tick: first_tick, .. + }) = tick_rx.try_recv() + else { + panic!("expected `Armed`"); + }; + + let _second = watchdog.tick(0); + let Ok(Tick::Armed { + tick: second_tick, .. + }) = tick_rx.try_recv() + else { + panic!("expected `Armed`"); + }; + + assert!(second_tick > first_tick); + } +} diff --git a/crates/ark/src/main.rs b/crates/ark/src/main.rs index 7fd9bd6666..59410b0979 100644 --- a/crates/ark/src/main.rs +++ b/crates/ark/src/main.rs @@ -23,6 +23,7 @@ use ark::traps::register_trap_handlers; use crossbeam::channel::unbounded; use harp::command::r_home_setup; use notify::Watcher; +use stdext::panic_message; use stdext::unwrap; thread_local! { @@ -398,14 +399,7 @@ fn main() -> anyhow::Result<()> { String::from("No location information:") }; - let msg: String; - if let Some(s) = info.downcast_ref::<&str>() { - msg = s.to_string(); - } else if let Some(s) = info.downcast_ref::() { - msg = s.clone(); - } else { - msg = String::from("No contextual information."); - } + let msg = panic_message(info); // Top-level-exec and try-catch errors already contain a backtrace // for the R thread so don't repeat it if we see one. Only perform diff --git a/crates/oak_cache/src/lib.rs b/crates/oak_cache/src/lib.rs index 51d6cdce1e..e6fe9ee70f 100644 --- a/crates/oak_cache/src/lib.rs +++ b/crates/oak_cache/src/lib.rs @@ -97,7 +97,7 @@ impl Cache { /// Runs a best-effort [`Cache::clean`] under the exclusive root lock (skipped if /// another session holds the shared lock), then holds the shared root lock for the /// life of the returned `Cache` so handed-out paths stay valid. - pub fn open(root: &str) -> anyhow::Result { + pub fn open(root: &Path) -> anyhow::Result { Self::open_in(cache_dir()?.join(root)) } diff --git a/crates/oak_db/src/storage.rs b/crates/oak_db/src/storage.rs index 0c14546436..ea42aacd89 100644 --- a/crates/oak_db/src/storage.rs +++ b/crates/oak_db/src/storage.rs @@ -13,7 +13,7 @@ use crate::WorkspaceRoots; /// Holds singleton `WorkspaceRoots` / `LibraryRoots` / `OrphanRoot` / /// `StaleRoot` inputs and lazy-initialises them on first access. #[salsa::db] -#[derive(Clone, Default)] +#[derive(Default)] pub struct OakDatabase { storage: salsa::Storage, workspace_roots: Arc>, @@ -29,6 +29,26 @@ impl OakDatabase { Self::default() } + /// A snapshot handle onto the database for a background reader. + /// + /// When the main loop needs to write to a `&mut OakDatabase`, it gets + /// parked by Salsa until all snapshot handles have been dropped. Only + /// create a snapshot for cancellable CPU-bound tasks that either query + /// Salsa or periodically check for cancellation. + /// + /// Keep `OakDatabase` non-`Clone`, this should be the only way to create a + /// Salsa handle. + pub fn snapshot(&self) -> Self { + Self { + storage: self.storage.clone(), + workspace_roots: Arc::clone(&self.workspace_roots), + library_roots: Arc::clone(&self.library_roots), + orphan_root: Arc::clone(&self.orphan_root), + stale_root: Arc::clone(&self.stale_root), + holds: Arc::clone(&self.holds), + } + } + // Number of live clones of this db (always >= 1, the caller itself). A // write through `&mut db` parks until this reaches 1, so a value > 1 here // means a write right now would block on that many outstanding handles. diff --git a/crates/oak_scan/src/packages.rs b/crates/oak_scan/src/packages.rs index e46bb57684..143efebf73 100644 --- a/crates/oak_scan/src/packages.rs +++ b/crates/oak_scan/src/packages.rs @@ -5,6 +5,7 @@ use std::collections::HashMap; use std::fs; +use std::io; use std::path::Path; use std::path::PathBuf; @@ -219,9 +220,18 @@ pub(crate) fn read_package_sources( directory: &Path, collation: Option<&[String]>, ) -> (Vec, Vec) { - let Ok(entries) = fs::read_dir(directory) else { - log::warn!("Cannot read sources directory: {}", directory.display()); - return (Vec::new(), Vec::new()); + let entries = match fs::read_dir(directory) { + Ok(entries) => entries, + // A data-only package has no `R/`. `datasets` is always on the search + // path, so this is the common case, not an error. + Err(err) if err.kind() == io::ErrorKind::NotFound => return (Vec::new(), Vec::new()), + Err(err) => { + log::warn!( + "Cannot read sources directory {}: {err:?}", + directory.display() + ); + return (Vec::new(), Vec::new()); + }, }; let mut files: Vec<(PathBuf, FileEntry)> = Vec::new(); diff --git a/crates/oak_scan/src/tests.rs b/crates/oak_scan/src/tests.rs index 9d73e94273..867bf92508 100644 --- a/crates/oak_scan/src/tests.rs +++ b/crates/oak_scan/src/tests.rs @@ -1,3 +1,4 @@ +mod packages; mod scheduler; mod sources; mod stale; diff --git a/crates/oak_scan/src/tests/packages.rs b/crates/oak_scan/src/tests/packages.rs new file mode 100644 index 0000000000..299c0326a1 --- /dev/null +++ b/crates/oak_scan/src/tests/packages.rs @@ -0,0 +1,129 @@ +//! Tests for [`read_package_sources`], which decides which R files a package +//! contributes to its loadable namespace and which are standalone scripts. + +use std::fs; +use std::path::Path; + +use crate::inputs::FileEntry; +use crate::packages::read_package_sources; + +/// Write `files` into `dir`, creating it first. +fn write_r_dir(dir: &Path, files: &[(&str, &str)]) { + fs::create_dir_all(dir).unwrap(); + for (basename, contents) in files { + fs::write(dir.join(basename), contents).unwrap(); + } +} + +/// Basenames of `files`, in the order returned. +fn names(files: &[FileEntry]) -> Vec { + files + .iter() + .map(|file| file.path.file_name().unwrap().into_owned()) + .collect() +} + +/// A data-only package has no `R/` at all. `datasets` is on the default search +/// path, so this is a routine state and must yield no files rather than an +/// error. +#[test] +fn test_missing_r_directory_yields_no_files() { + let tmp = tempfile::tempdir().unwrap(); + let missing = tmp.path().join("datasets").join("R"); + + let (files, scripts) = read_package_sources(&missing, None); + + assert!(files.is_empty()); + assert!(scripts.is_empty()); +} + +/// An `R/` that exists but holds no R code is distinct from one that's absent, +/// and lands in the same place. +#[test] +fn test_empty_r_directory_yields_no_files() { + let tmp = tempfile::tempdir().unwrap(); + let r = tmp.path().join("R"); + write_r_dir(&r, &[]); + + let (files, scripts) = read_package_sources(&r, None); + + assert!(files.is_empty()); + assert!(scripts.is_empty()); +} + +/// Without `Collate:`, every R file is loadable, ordered case-insensitively by +/// basename so the result doesn't depend on `read_dir` order. +#[test] +fn test_without_collation_all_files_are_loadable_and_sorted() { + let tmp = tempfile::tempdir().unwrap(); + let r = tmp.path().join("R"); + write_r_dir(&r, &[("zebra.R", "1"), ("Apple.R", "1"), ("mango.R", "1")]); + + let (files, scripts) = read_package_sources(&r, None); + + assert_eq!(names(&files), vec!["Apple.R", "mango.R", "zebra.R"]); + assert!(scripts.is_empty()); +} + +/// Non-R files and subdirectories are skipped. `R/` is loaded as a flat +/// directory, so `R/sub/nested.R` isn't part of the namespace. +#[test] +fn test_non_r_entries_and_subdirectories_are_skipped() { + let tmp = tempfile::tempdir().unwrap(); + let r = tmp.path().join("R"); + write_r_dir(&r, &[ + ("keep.R", "1"), + ("README.md", "x"), + ("data.csv", "x"), + ]); + write_r_dir(&r.join("sub"), &[("nested.R", "1")]); + + let (files, scripts) = read_package_sources(&r, None); + + assert_eq!(names(&files), vec!["keep.R"]); + assert!(scripts.is_empty()); +} + +/// `Collate:` sets the load order, which is not alphabetical. Files it lists +/// become the namespace in exactly that sequence. +#[test] +fn test_collation_sets_load_order() { + let tmp = tempfile::tempdir().unwrap(); + let r = tmp.path().join("R"); + write_r_dir(&r, &[("a.R", "1"), ("b.R", "1"), ("c.R", "1")]); + + let order = ["c.R".to_string(), "a.R".to_string(), "b.R".to_string()]; + let (files, scripts) = read_package_sources(&r, Some(&order)); + + assert_eq!(names(&files), vec!["c.R", "a.R", "b.R"]); + assert!(scripts.is_empty()); +} + +/// A file on disk that `Collate:` omits can't enter the namespace, so it's kept +/// as a standalone script rather than dropped. Leftovers are sorted. +#[test] +fn test_files_absent_from_collation_become_scripts() { + let tmp = tempfile::tempdir().unwrap(); + let r = tmp.path().join("R"); + write_r_dir(&r, &[("listed.R", "1"), ("zebra.R", "1"), ("apple.R", "1")]); + + let order = ["listed.R".to_string()]; + let (files, scripts) = read_package_sources(&r, Some(&order)); + + assert_eq!(names(&files), vec!["listed.R"]); + assert_eq!(names(&scripts), vec!["apple.R", "zebra.R"]); +} + +/// A `Collate:` entry with no file on disk is skipped rather than fabricated. +#[test] +fn test_collation_entry_without_a_file_is_skipped() { + let tmp = tempfile::tempdir().unwrap(); + let r = tmp.path().join("R"); + write_r_dir(&r, &[("present.R", "1")]); + + let order = ["missing.R".to_string(), "present.R".to_string()]; + let (files, scripts) = read_package_sources(&r, Some(&order)); + + assert_eq!(names(&files), vec!["present.R"]); + assert!(scripts.is_empty()); +} diff --git a/crates/oak_source/src/lib.rs b/crates/oak_source/src/lib.rs index 71276c1350..93f7083b09 100644 --- a/crates/oak_source/src/lib.rs +++ b/crates/oak_source/src/lib.rs @@ -14,6 +14,7 @@ mod download; mod extract; mod r; +use std::path::Path; use std::path::PathBuf; use oak_cache::Cache; @@ -36,8 +37,8 @@ pub struct SourceCache { impl SourceCache { pub fn open() -> anyhow::Result { Ok(Self { - cran: Cache::open(&format!("source/{CACHE_VERSION}/cran"))?, - r: RCache::open(&format!("source/{CACHE_VERSION}/r"))?, + cran: Cache::open(&Path::new("source").join(CACHE_VERSION).join("cran"))?, + r: RCache::open(&Path::new("source").join(CACHE_VERSION).join("r"))?, }) } diff --git a/crates/oak_source/src/r.rs b/crates/oak_source/src/r.rs index 6c15f1fb66..dfd2166067 100644 --- a/crates/oak_source/src/r.rs +++ b/crates/oak_source/src/r.rs @@ -49,7 +49,7 @@ pub(crate) struct RCache { } impl RCache { - pub(crate) fn open(root: &str) -> anyhow::Result { + pub(crate) fn open(root: &Path) -> anyhow::Result { Ok(Self { cache: Cache::open(root)?, }) diff --git a/crates/oak_srcref/src/lib.rs b/crates/oak_srcref/src/lib.rs index 67f322a479..7c9e98f905 100644 --- a/crates/oak_srcref/src/lib.rs +++ b/crates/oak_srcref/src/lib.rs @@ -28,7 +28,7 @@ impl SrcrefCache { pub fn open(r: PathBuf) -> anyhow::Result { Ok(Self { r, - cache: Cache::open(&format!("srcref/{CACHE_VERSION}"))?, + cache: Cache::open(&Path::new("srcref").join(CACHE_VERSION))?, }) } diff --git a/crates/stdext/src/lib.rs b/crates/stdext/src/lib.rs index 8924459c0e..a59a586c60 100644 --- a/crates/stdext/src/lib.rs +++ b/crates/stdext/src/lib.rs @@ -13,6 +13,7 @@ pub mod event; pub mod join; pub mod local; pub mod ok; +pub mod panic; pub mod push; pub mod result; pub mod sorted_vec; @@ -23,8 +24,12 @@ pub mod unwrap; pub use crate::cell::DebugRefCell; pub use crate::join::Joined; pub use crate::ok::Ok; +pub use crate::panic::panic_message; pub use crate::push::Push; pub use crate::sorted_vec::SortedVec; +pub use crate::spawn::DEFAULT_STACK_SIZE; +pub use crate::spawn::SMALL_STACK_SIZE; +pub use crate::spawn::TINY_STACK_SIZE; pub use crate::testing::assert_testing; pub use crate::testing::IS_TESTING; pub use crate::unwrap::IntoOption; diff --git a/crates/stdext/src/panic.rs b/crates/stdext/src/panic.rs new file mode 100644 index 0000000000..fda04837e1 --- /dev/null +++ b/crates/stdext/src/panic.rs @@ -0,0 +1,41 @@ +// +// panic.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +use std::any::Any; + +/// The message out of a panic payload, for logging. Takes `&dyn Any` so it +/// serves both a `catch_unwind()` payload and `PanicHookInfo::payload()`. +/// +/// `panic!()` boxes its message as `&str` when it has no arguments to format +/// and as `String` when it does, so both need handling. Anything else comes +/// from `panic_any()`. +pub fn panic_message(payload: &(dyn Any + Send)) -> String { + if let Some(msg) = payload.downcast_ref::<&str>() { + msg.to_string() + } else if let Some(msg) = payload.downcast_ref::() { + msg.clone() + } else { + String::from("(unknown panic payload)") + } +} + +#[cfg(test)] +mod tests { + use super::panic_message; + + #[test] + fn test_panic_message_handles_both_payload_types() { + let unformatted = std::panic::catch_unwind(|| panic!("plain")).unwrap_err(); + assert_eq!(panic_message(unformatted.as_ref()), "plain"); + + let formatted = std::panic::catch_unwind(|| panic!("formatted {}", 1)).unwrap_err(); + assert_eq!(panic_message(formatted.as_ref()), "formatted 1"); + + let other = std::panic::catch_unwind(|| std::panic::panic_any(42u8)).unwrap_err(); + assert_eq!(panic_message(other.as_ref()), "(unknown panic payload)"); + } +} diff --git a/crates/stdext/src/spawn.rs b/crates/stdext/src/spawn.rs index ca8cdff221..41417a8ce4 100644 --- a/crates/stdext/src/spawn.rs +++ b/crates/stdext/src/spawn.rs @@ -21,3 +21,31 @@ macro_rules! spawn { .unwrap() }; } + +/// Rust's own default. For a thread whose call tree reaches into dependencies +/// too deep to bound by inspection. +pub const DEFAULT_STACK_SIZE: usize = 2 * 1024 * 1024; + +/// For a thread that walks data structures and does I/O, with no recursion over +/// user input. +pub const SMALL_STACK_SIZE: usize = 512 * 1024; + +/// For a poll or dispatch loop whose deepest frame is a fixed-size buffer. +pub const TINY_STACK_SIZE: usize = 256 * 1024; + +/// Like [`spawn!`], with an explicit stack size instead of +/// [`DEFAULT_STACK_SIZE`]. +/// +/// Budget for more than the thread call tree: a panic hook that captures a +/// backtrace runs on the panicking thread's stack. We don't go below 256kb to +/// remain on the safe side. +#[macro_export] +macro_rules! spawn_with_stack_size { + ($name:expr, $stack_size:expr, $body:expr) => { + std::thread::Builder::new() + .name($name.to_string()) + .stack_size($stack_size) + .spawn($body) + .unwrap() + }; +}