From 0c073951f4a49da165cba9a341fa4d4aec1611a7 Mon Sep 17 00:00:00 2001 From: wan9chi Date: Mon, 9 Nov 2026 15:48:48 +0800 Subject: [PATCH] perf(cache): check cache entries off the scheduler thread The scheduler polls every task's future on one thread, so the synchronous work in a task's cache lookup held up every other task: hashing its explicit inputs, decoding its local entry or the entries fetched from the remote cache, and validating them, which stats and hashes every inferred input. Tasks that became ready together had their lookups run one after another. In a CI run where 26 cached Vite builds became ready at once, all remote hits, they started over about 1 second on Linux and up to 2.7 seconds on Windows instead of together. That work now runs on rayon's thread pool, in the task's span, so the lookups of ready tasks proceed concurrently, with the same results and miss reasons. The pool is rayon's rather than tokio's blocking threads: validation already hashes inputs in parallel on it, and its size keeps the checks from crowding out downloads and extraction. On a 4-core Windows runner, running all 26 checks at once on blocking threads made those builds finish later than before. A lookup still stops when the run is cancelled: work that hasn't started doesn't run, and work that's already running finishes in the background with its result dropped. Tasks that become ready together now start in the order their checks finish. The e2e harness runs vp with RAYON_NUM_THREADS=1, so the checks finish in the order they're submitted, and those tasks start in scheduling order in snapshots. The new e2e cases hold the reads of a task's inferred input while its local or remote entry is checked, and release them from a task that the scheduler can only start once it sees another task finish, so a check on the scheduler's thread would never finish. vtt hold-reads holds the reads with a FIFO on Unix and with an exclusive oplock on Windows. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + Cargo.lock | 1 + crates/vt/src/session/cache/mod.rs | 259 ++++++++++++------ crates/vt/src/session/execute/mod.rs | 49 ++-- crates/vt/src/session/execute/scheduler.rs | 5 +- crates/vt_bin/Cargo.toml | 8 + crates/vt_bin/src/vtt/barrier.rs | 59 ++-- crates/vt_bin/src/vtt/hold_reads.rs | 108 ++++++++ crates/vt_bin/src/vtt/main.rs | 4 +- crates/vt_bin/tests/e2e_snapshots/README.md | 2 +- .../cache_check_concurrency/input.txt | 1 + .../cache_check_concurrency/package.json | 4 + .../cache_check_concurrency/snapshots.toml | 75 +++++ ...ng_local_entry_doesnt_block_other_tasks.md | 28 ++ ...te_cache_entry_doesnt_block_other_tasks.md | 47 ++++ .../cache_check_concurrency/vite-task.json | 31 +++ crates/vt_bin/tests/e2e_snapshots/main.rs | 6 + crates/vt_glob/src/path.rs | 13 + 18 files changed, 577 insertions(+), 124 deletions(-) create mode 100644 crates/vt_bin/src/vtt/hold_reads.rs create mode 100644 crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/input.txt create mode 100644 crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/package.json create mode 100644 crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots.toml create mode 100644 crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots/checking_local_entry_doesnt_block_other_tasks.md create mode 100644 crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots/checking_remote_cache_entry_doesnt_block_other_tasks.md create mode 100644 crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/vite-task.json diff --git a/CHANGELOG.md b/CHANGELOG.md index eb0954b90..86d9e3289 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # Changelog +- **Perf** When several tasks become ready at once, `vp run` now checks their cache entries at the same time instead of one after another, so cache hits start sooner. Hashing a task's input files for its cache lookup no longer holds up other tasks ([#982](https://github.com/voidzero-dev/vite-task/pull/982)). - **Fixed** On Windows, the files a task deletes or moves, such as a build cleaning its output folder before writing it again, are no longer among its inputs, as on Linux and macOS. Before, when a later step changed one of those files, for example by stamping a version into the rebuilt output, the next run missed the cache with a reason like `'dist/index.js' modified`, though the task never read the file ([#980](https://github.com/voidzero-dev/vite-task/pull/980)). - **Changed** In `@voidzero-dev/vite-task-client`, a `getEnvs` query that matches every env name, `{ prefix: '' }` or `*`, now returns nothing instead of every environment variable, and no longer adds them to the cache key ([#976](https://github.com/voidzero-dev/vite-task/pull/976)). - **Added** `vp run --json` prints every package and task in the workspace, with each task's `dependsOn` resolved to task IDs, as one JSON document for scripts, editors and coding agents. It runs nothing and, like `vp run` without a task, loads every package's config. It prints the same document from any directory in the workspace: indented in a terminal, on one line when captured. Its `$schema` field points to a JSON Schema that describes each field, which `vp run` writes to the system temp directory ([#948](https://github.com/voidzero-dev/vite-task/pull/948)). diff --git a/Cargo.lock b/Cargo.lock index 8e8ccd7b5..45cdea086 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5056,6 +5056,7 @@ dependencies = [ "vt_str", "vt_workspace", "which", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/vt/src/session/cache/mod.rs b/crates/vt/src/session/cache/mod.rs index e76ef7aad..810ffddc3 100644 --- a/crates/vt/src/session/cache/mod.rs +++ b/crates/vt/src/session/cache/mod.rs @@ -7,9 +7,11 @@ mod validation; use std::{ collections::BTreeMap, + ffi::OsStr, fmt::Display, fs::File, io::Write, + panic::{self, AssertUnwindSafe}, sync::{Arc, OnceLock}, time::{Duration, Instant}, }; @@ -21,9 +23,11 @@ pub use display::{ format_spawn_change, }; use rusqlite::{Connection, OptionalExtension as _}; +use rustc_hash::FxHashMap; use serde::{Deserialize, Serialize}; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, oneshot}; use tokio_util::sync::CancellationToken; +use vt_casefold::EnvName; use vt_glob::path::PathGlobSet; use vt_graph::config::ResolvedGlobConfig; use vt_path::{AbsolutePath, RelativePathBuf}; @@ -307,6 +311,77 @@ fn enable_wal(conn: &Connection) -> rusqlite::Result<()> { Ok(()) } +/// Run `work` on rayon's global thread pool, in the current span, and wait for +/// it until `cancel_token` is cancelled. +/// +/// The scheduler polls every task's future on one thread, so file and CPU work +/// done there, such as hashing a task's inputs or checking a cache entry, +/// holds up every other task until it's done. Run there, the cache lookups of +/// tasks that become ready together would run one after another. +/// +/// The pool is rayon's rather than tokio's blocking threads because checking +/// an entry already hashes its inferred inputs in parallel on it, and a pool +/// thread takes part in that work instead of waiting for it. With one thread +/// (`RAYON_NUM_THREADS=1`, as in the e2e tests), work also finishes in the +/// order it's submitted, so tasks that become ready together start in the +/// same order every run. +/// +/// Returns `None` if `cancel_token` is cancelled first. Work that hasn't +/// started then doesn't run, and work that's already running finishes in the +/// background with its result dropped, so it must have no effects that +/// outlive it. If `work` panics, the panic is resumed here. +pub async fn run_off_thread( + cancel_token: &CancellationToken, + work: impl FnOnce() -> T + Send + 'static, +) -> Option { + if cancel_token.is_cancelled() { + return None; + } + let span = tracing::Span::current(); + let (sender, receiver) = oneshot::channel(); + rayon::spawn_fifo(move || { + if sender.is_closed() { + return; + } + // A panic in a job would abort the process, so it's sent to the + // waiting task instead. + let _ = sender.send(panic::catch_unwind(AssertUnwindSafe(|| span.in_scope(work)))); + }); + match cancel_token.run_until_cancelled(receiver).await? { + Ok(Ok(output)) => Some(output), + Ok(Err(panic)) => panic::resume_unwind(panic), + // The job sends a result unless the receiver is dropped first. + Err(oneshot::error::RecvError { .. }) => None, + } +} + +/// What a lookup validates entries against: the execution's current envs and +/// explicit inputs, and the files under the workspace root. It owns all of +/// them, so that entries can be checked off the scheduler's thread. +#[derive(Clone)] +struct Validation { + /// The execution's `CacheMetadata::unfiltered_envs`. + unfiltered_envs: Arc>, Arc>>, + globbed_inputs: Arc>, + input_negative_globs: Arc>, + workspace_root: Arc, +} + +impl Validation { + /// Validate `cache_value` as [`CacheEntryValue::validate`] does. + fn validate( + &self, + cache_value: &CacheEntryValue, + ) -> anyhow::Result> { + cache_value.validate( + &self.unfiltered_envs, + &self.globbed_inputs, + &self.input_negative_globs, + &self.workspace_root, + ) + } +} + impl ExecutionCache { #[tracing::instrument(level = "debug", skip_all)] pub fn load_from_path(path: &AbsolutePath) -> anyhow::Result { @@ -347,37 +422,47 @@ impl ExecutionCache { } /// Try to hit cache by looking up the cache entry key and validating inputs. - /// Returns `Ok(Ok(cache_hit))` on cache hit, `Ok(Err(cache_miss))` on miss. + /// Returns `Ok(Some(Ok(cache_hit)))` on cache hit, `Ok(Some(Err(cache_miss)))` + /// on miss. /// /// After a local miss, the remote cache is queried if the task has one. A /// remote hit has its output archive downloaded into `cache_dir`, is /// recorded locally by [`Self::restore`], and is never uploaded. If the /// local cache has an entry for the task, its miss reason is kept. - /// Otherwise the reason comes from the remote cache. Remote requests stop - /// when `cancel_token` is cancelled. + /// Otherwise the reason comes from the remote cache. + /// + /// Entries are decoded and validated on rayon's pool, as + /// [`run_off_thread`] describes, so other tasks' lookups don't wait for + /// them. The lookup stops when `cancel_token` is cancelled. A cancelled + /// remote request or check is a [`ReadError::Cancelled`] miss, and a + /// cancelled check of the local entry returns `Ok(None)`. Either way the + /// task doesn't start, so nothing is reported. #[tracing::instrument(level = "debug", skip_all)] pub async fn try_hit( &self, cache_metadata: &CacheMetadata, - globbed_inputs: &BTreeMap, - workspace_root: &AbsolutePath, + globbed_inputs: &Arc>, + workspace_root: &Arc, cache_dir: &AbsolutePath, cancel_token: &CancellationToken, - ) -> anyhow::Result> { + ) -> anyhow::Result>> { let cache_key = CacheEntryKey::from_metadata(cache_metadata); - // Part of the key, so the same as when the entry was created. - let input_negative_globs = PathGlobSet::new(&cache_metadata.input_config.negative_globs)?; + let validation = Validation { + unfiltered_envs: Arc::clone(&cache_metadata.unfiltered_envs), + globbed_inputs: Arc::clone(globbed_inputs), + // Part of the key, so the same as when the entry was created. + input_negative_globs: Arc::new( + PathGlobSet::new(&cache_metadata.input_config.negative_globs)?.into_owned(), + ), + workspace_root: Arc::clone(workspace_root), + }; let start = Instant::now(); - let local_lookup = self - .try_hit_local( - cache_metadata, - &cache_key, - globbed_inputs, - &input_negative_globs, - workspace_root, - ) - .await?; + let Some(local_lookup) = + self.try_hit_local(cache_metadata, &cache_key, &validation, cancel_token).await? + else { + return Ok(None); + }; tracing::debug!( outcome = match &local_lookup { Ok(_) => "hit", @@ -388,7 +473,7 @@ impl ExecutionCache { "looked up the local cache" ); let local_miss = match local_lookup { - Ok(value) => return Ok(Ok(CacheHit { value, source: CacheHitSource::Local })), + Ok(value) => return Ok(Some(Ok(CacheHit { value, source: CacheHitSource::Local }))), Err(miss) => miss, }; #[expect( @@ -402,7 +487,7 @@ impl ExecutionCache { .. }, ) => remote_config, - None => return Ok(Err(local_miss)), + None => return Ok(Some(Err(local_miss))), }; let start = Instant::now(); let remote_lookup = self @@ -410,9 +495,7 @@ impl ExecutionCache { remote_config, cache_metadata, &cache_key, - globbed_inputs, - &input_negative_globs, - workspace_root, + &validation, cache_dir, cancel_token, ) @@ -428,39 +511,47 @@ impl ExecutionCache { "looked up the remote cache" ); let remote_miss = match remote_lookup { - Ok(value) => return Ok(Ok(CacheHit { value, source: CacheHitSource::Remote })), + Ok(value) => return Ok(Some(Ok(CacheHit { value, source: CacheHitSource::Remote }))), Err(miss) => miss, }; - Ok(Err(match local_miss { + Ok(Some(Err(match local_miss { CacheMiss::NotFound => remote_miss, local_miss => local_miss, - })) + }))) } + /// Look up the local entry under `cache_key` and validate it with + /// `validation`. Returns `None` if `cancel_token` is cancelled while the + /// entry is checked. async fn try_hit_local( &self, cache_metadata: &CacheMetadata, cache_key: &CacheEntryKey, - globbed_inputs: &BTreeMap, - input_negative_globs: &PathGlobSet<'_>, - workspace_root: &AbsolutePath, - ) -> anyhow::Result> { + validation: &Validation, + cancel_token: &CancellationToken, + ) -> anyhow::Result>> { let execution_cache_key = &cache_metadata.execution_cache_key; // Try to find the cache entry by key (spawn fingerprint + input config) - if let Some(cache_value) = self.get_by_cache_key(cache_key).await? { - if let Some(mismatch) = cache_value.validate( - &cache_metadata.unfiltered_envs, - globbed_inputs, - input_negative_globs, - workspace_root, - )? { - return Ok(Err(CacheMiss::FingerprintMismatch(mismatch))); + if let Some(value_blob) = self.get_cache_entry_blob(cache_key).await? { + let validation = validation.clone(); + let Some(checked) = run_off_thread(cancel_token, move || { + let cache_value: CacheEntryValue = deserialize_cache(&value_blob)?; + let mismatch = validation.validate(&cache_value)?; + anyhow::Ok((cache_value, mismatch)) + }) + .await + else { + return Ok(None); + }; + let (cache_value, mismatch) = checked?; + if let Some(mismatch) = mismatch { + return Ok(Some(Err(CacheMiss::FingerprintMismatch(mismatch)))); } // Associate the execution key to the cache entry key if not already, // so that next time we can find it and report what changed self.upsert_task_fingerprint(execution_cache_key, cache_key).await?; - return Ok(Ok(cache_value)); + return Ok(Some(Ok(cache_value))); } // No cache found with the current cache entry key, @@ -468,13 +559,13 @@ impl ExecutionCache { if let Some(old_cache_key) = self.get_cache_key_by_execution_key(execution_cache_key).await? { - // `get_by_cache_key` above returned None for the *current* cache key, - // so the associated key must differ. + // `get_cache_entry_blob` above returned None for the *current* + // cache key, so the associated key must differ. let mismatch = old_cache_key.into_mismatch(cache_key); - return Ok(Err(CacheMiss::FingerprintMismatch(mismatch))); + return Ok(Some(Err(CacheMiss::FingerprintMismatch(mismatch)))); } - Ok(Err(CacheMiss::NotFound)) + Ok(Some(Err(CacheMiss::NotFound))) } /// Fetch the entry from the remote cache that `remote_config` configures. @@ -482,16 +573,14 @@ impl ExecutionCache { /// is downloaded. /// A fallback entry, a failed validation, or a failed read is a miss. An /// error while validating counts as a failed read, so the remote entry - /// never fails the task. - #[expect(clippy::too_many_arguments, reason = "forwarded from `try_hit`")] + /// never fails the task. The fetched entries are checked off the + /// scheduler's thread. async fn try_hit_remote( &self, remote_config: &ResolvedRemoteCacheConfig, cache_metadata: &CacheMetadata, cache_key: &CacheEntryKey, - globbed_inputs: &BTreeMap, - input_negative_globs: &PathGlobSet<'_>, - workspace_root: &AbsolutePath, + validation: &Validation, cache_dir: &AbsolutePath, cancel_token: &CancellationToken, ) -> anyhow::Result> { @@ -513,19 +602,20 @@ impl ExecutionCache { elapsed = ?start.elapsed(), "fetched from the remote cache" ); - let validate = |cache_value: &CacheEntryValue| { - cache_value.validate( - &cache_metadata.unfiltered_envs, - globbed_inputs, - input_negative_globs, - workspace_root, - ) + let (cache_key, validation) = (cache_key.clone(), validation.clone()); + #[expect( + clippy::result_large_err, + reason = "`CacheMiss` is intentionally large, and a lookup returns it once" + )] + let check = move || { + remote::resolve(fetched, &cache_key, |cache_value| validation.validate(cache_value)) + }; + let resolved = run_off_thread(cancel_token, check).await; + let Restore { value: cache_value, blob_id } = match resolved { + Some(Ok(restore)) => restore, + Some(Err(miss)) => return Ok(Err(miss)), + None => return Ok(Err(ReadError::Cancelled.into_miss())), }; - let Restore { value: cache_value, blob_id } = - match remote::resolve(fetched, cache_key, validate) { - Ok(restore) => restore, - Err(miss) => return Ok(Err(miss)), - }; let output_archive = match blob_id { Some(blob_id) => { @@ -684,10 +774,6 @@ impl ExecutionCache { // Basic database operations impl ExecutionCache { - #[expect( - clippy::significant_drop_tightening, - reason = "lock guard cannot be dropped earlier because prepared statement borrows connection" - )] async fn get_key_by_value< K: SchemaWrite, V: SchemaReadOwned, @@ -696,26 +782,34 @@ impl ExecutionCache { table: &str, key: &K, ) -> anyhow::Result> { - let key_blob = serialize_cache(key)?; - let value_blob = { - let conn = self.conn.lock().await; - #[expect( - clippy::disallowed_macros, - reason = "SQL query string for rusqlite requires String" - )] - let mut select_stmt = - conn.prepare_cached(&format!("SELECT value FROM {table} WHERE key=?"))?; - let value_blob: Option> = - select_stmt.query_row::, _, _>([key_blob], |row| row.get(0)).optional()?; - value_blob - }; - let Some(value_blob) = value_blob else { + let Some(value_blob) = self.get_value_blob(table, key).await? else { return Ok(None); }; let value: V = deserialize_cache(&value_blob)?; Ok(Some(value)) } + /// The encoded value stored under `key` in `table`. + #[expect( + clippy::significant_drop_tightening, + reason = "lock guard cannot be dropped earlier because prepared statement borrows connection" + )] + async fn get_value_blob>( + &self, + table: &str, + key: &K, + ) -> anyhow::Result>> { + let key_blob = serialize_cache(key)?; + let conn = self.conn.lock().await; + #[expect( + clippy::disallowed_macros, + reason = "SQL query string for rusqlite requires String" + )] + let mut select_stmt = + conn.prepare_cached(&format!("SELECT value FROM {table} WHERE key=?"))?; + Ok(select_stmt.query_row::, _, _>([key_blob], |row| row.get(0)).optional()?) + } + async fn get_by_cache_key( &self, cache_key: &CacheEntryKey, @@ -723,6 +817,15 @@ impl ExecutionCache { self.get_key_by_value("cache_entries", cache_key).await } + /// The entry under `cache_key`, still encoded, so that it can be decoded + /// off the scheduler's thread. + async fn get_cache_entry_blob( + &self, + cache_key: &CacheEntryKey, + ) -> anyhow::Result>> { + self.get_value_blob("cache_entries", cache_key).await + } + async fn get_cache_key_by_execution_key( &self, execution_cache_key: &ExecutionCacheKey, diff --git a/crates/vt/src/session/execute/mod.rs b/crates/vt/src/session/execute/mod.rs index 2d3b91b42..751162f8d 100644 --- a/crates/vt/src/session/execute/mod.rs +++ b/crates/vt/src/session/execute/mod.rs @@ -33,7 +33,7 @@ use self::{ spawn::{ChildHandle, ChildOutcome, SpawnStdio, spawn}, }; use super::{ - cache::{CacheHit, CacheHitSource, CacheMiss, ExecutionCache}, + cache::{CacheHit, CacheHitSource, CacheMiss, ExecutionCache, run_off_thread}, event::{ CacheDisabledReason, CacheErrorKind, CacheNotUpdatedReason, CacheStatus, CacheUpdateStatus, ExecutionError, @@ -547,8 +547,10 @@ enum CacheLookup<'a> { } /// Phase 1: compute the globbed inputs and try to hit the cache. A remote hit -/// downloads its output archive into `cache_dir`. Remote requests stop when -/// `cancel_token` is cancelled. +/// downloads its output archive into `cache_dir`. The inputs are hashed, and +/// entries checked, off the scheduler's thread (see [`run_off_thread`]). The +/// lookup stops when `cancel_token` is cancelled, and is then +/// [`Report::Cancelled`] or a miss that isn't reported. async fn lookup_cache<'a>( cache_metadata: Option<&'a CacheMetadata>, cache: &ExecutionCache, @@ -562,27 +564,38 @@ async fn lookup_cache<'a>( // Compute globbed inputs from positive globs at execution time. // Globs are already workspace-root-relative (resolved at task graph stage). - let start = Instant::now(); - let globbed_inputs = compute_globbed_inputs( - workspace_root, - &cache_metadata.input_config.positive_globs, - &cache_metadata.input_config.negative_globs, - ) - .map_err(|err| { + let (root, input_config) = (Arc::clone(workspace_root), cache_metadata.input_config.clone()); + let hashed = run_off_thread(cancel_token, move || { + let start = Instant::now(); + let globbed_inputs = compute_globbed_inputs( + &root, + &input_config.positive_globs, + &input_config.negative_globs, + )?; + tracing::debug!( + files = globbed_inputs.len(), + elapsed = ?start.elapsed(), + "hashed the globbed inputs" + ); + anyhow::Ok(globbed_inputs) + }) + .await + .ok_or(Report::Cancelled)?; + let globbed_inputs = Arc::new(hashed.map_err(|err| { Report::failed(ExecutionError::Cache { kind: CacheErrorKind::Lookup, source: err }) - })?; - tracing::debug!( - files = globbed_inputs.len(), - elapsed = ?start.elapsed(), - "hashed the globbed inputs" - ); + })?); match cache .try_hit(cache_metadata, &globbed_inputs, workspace_root, cache_dir, cancel_token) .await { - Ok(Ok(hit)) => Ok(CacheLookup::Hit { hit, metadata: cache_metadata }), - Ok(Err(miss)) => Ok(CacheLookup::Miss { miss, globbed_inputs }), + Ok(Some(Ok(hit))) => Ok(CacheLookup::Hit { hit, metadata: cache_metadata }), + // Unless the lookup was cancelled, its checks have finished and + // dropped their references, so the inputs aren't copied. + Ok(Some(Err(miss))) => { + Ok(CacheLookup::Miss { miss, globbed_inputs: Arc::unwrap_or_clone(globbed_inputs) }) + } + Ok(None) => Err(Report::Cancelled), Err(err) => { Err(Report::failed(ExecutionError::Cache { kind: CacheErrorKind::Lookup, source: err })) } diff --git a/crates/vt/src/session/execute/scheduler.rs b/crates/vt/src/session/execute/scheduler.rs index 9dcc7b588..d23ed512a 100644 --- a/crates/vt/src/session/execute/scheduler.rs +++ b/crates/vt/src/session/execute/scheduler.rs @@ -38,7 +38,10 @@ const HIDE_PENDING_UPLOADS_ENV: &str = "VP_RUN_INTERNAL_HIDE_PENDING_UPLOADS"; /// The `reporter` field is wrapped in `RefCell` because concurrent futures /// (via `FuturesUnordered`) need shared access to create leaf reporters. /// Since all futures run on a single thread (no `tokio::spawn`), `RefCell` -/// is sufficient for interior mutability. +/// is sufficient for interior mutability. That thread also runs any blocking +/// work done inside them, so a cache lookup hashes inputs and checks entries +/// on rayon's pool instead (see +/// [`run_off_thread`](crate::session::cache::run_off_thread)). /// /// Cache fields are passed through to [`execute_spawn`] for cache-aware execution. struct ExecutionContext<'a> { diff --git a/crates/vt_bin/Cargo.toml b/crates/vt_bin/Cargo.toml index 0b39cb66a..3ad2c7986 100644 --- a/crates/vt_bin/Cargo.toml +++ b/crates/vt_bin/Cargo.toml @@ -34,6 +34,14 @@ which = { workspace = true } [target.'cfg(target_os = "linux")'.dependencies] nix = { workspace = true, features = ["mount", "sched", "user"] } +[target.'cfg(windows)'.dependencies] +windows-sys = { workspace = true, features = [ + "Win32_Foundation", + "Win32_Storage_FileSystem", + "Win32_System_IO", + "Win32_System_Ioctl", +] } + [dev-dependencies] cargo-platform = { workspace = true } cow-utils = { workspace = true } diff --git a/crates/vt_bin/src/vtt/barrier.rs b/crates/vt_bin/src/vtt/barrier.rs index c79db0279..00fa2b4a1 100644 --- a/crates/vt_bin/src/vtt/barrier.rs +++ b/crates/vt_bin/src/vtt/barrier.rs @@ -9,7 +9,6 @@ /// - `--hang`: Keep process alive after the barrier (for kill tests). /// - `--daemonize`: Close stdout/stderr but keep process alive (for daemon kill tests). pub fn run(args: &[String]) -> Result<(), Box> { - use notify::Watcher as _; let mut positional: Vec<&str> = Vec::new(); let mut exit_code: i32 = 0; let mut hang = false; @@ -34,9 +33,40 @@ pub fn run(args: &[String]) -> Result<(), Box> { ); } - let dir = std::path::Path::new(positional[0]); - let prefix = positional[1]; let count: usize = positional[2].parse()?; + wait(std::path::Path::new(positional[0]), positional[1], count)?; + + if daemonize { + // Close stdout/stderr but keep the process alive. Simulates a daemon that + // detaches from stdio — tests that the runner can still kill such processes. + // Closing the fds gives the parent's pipe an EOF. + // SAFETY: fds 1 and 2 are always valid (stdout/stderr). + unsafe { + libc::close(1); + libc::close(2); + } + loop { + std::thread::park(); + } + } + + if hang { + loop { + std::thread::park(); + } + } + + std::process::exit(exit_code); +} + +/// Join the barrier `` in `dir`: create `dir/_`, then +/// wait until `count` files matching `_*` exist in `dir`. +pub fn wait( + dir: &std::path::Path, + prefix: &str, + count: usize, +) -> Result<(), Box> { + use notify::Watcher as _; std::fs::create_dir_all(dir)?; @@ -64,26 +94,5 @@ pub fn run(args: &[String]) -> Result<(), Box> { } } } - - if daemonize { - // Close stdout/stderr but keep the process alive. Simulates a daemon that - // detaches from stdio — tests that the runner can still kill such processes. - // Closing the fds gives the parent's pipe an EOF. - // SAFETY: fds 1 and 2 are always valid (stdout/stderr). - unsafe { - libc::close(1); - libc::close(2); - } - loop { - std::thread::park(); - } - } - - if hang { - loop { - std::thread::park(); - } - } - - std::process::exit(exit_code); + Ok(()) } diff --git a/crates/vt_bin/src/vtt/hold_reads.rs b/crates/vt_bin/src/vtt/hold_reads.rs new file mode 100644 index 000000000..63e564f35 --- /dev/null +++ b/crates/vt_bin/src/vtt/hold_reads.rs @@ -0,0 +1,108 @@ +/// hold-reads `` `` +/// +/// Holds other processes' reads of `` until another process releases +/// them, for testing that a blocked read doesn't stop other work: +/// +/// 1. Make opening `` wait, then join the barrier `` `held` 2, +/// so that its other participant knows a read of `` would wait. +/// 2. Wait until another process opens ``. +/// 3. Join the barrier `` `released` 2, then let the waiting read finish +/// with the file's content and exit. +/// +/// On Unix, `` is replaced by a FIFO, whose `open` waits for a writer. +/// The file is put back as soon as a reader opens the FIFO, so later opens +/// don't wait. On Windows, an exclusive oplock is taken on ``. Another +/// process's open breaks it and waits until the oplock's handle is closed. +pub fn run(args: &[String]) -> Result<(), Box> { + let [file, dir] = args else { + return Err("Usage: vtt hold-reads ".into()); + }; + hold(std::path::Path::new(file), std::path::Path::new(dir)) +} + +#[cfg(unix)] +fn hold(file: &std::path::Path, dir: &std::path::Path) -> Result<(), Box> { + use std::{io::Write as _, os::unix::ffi::OsStrExt as _}; + + let content = std::fs::read(file)?; + let mut held = file.as_os_str().to_owned(); + held.push(".held"); + std::fs::write(&held, &content)?; + std::fs::remove_file(file)?; + let fifo_path = std::ffi::CString::new(file.as_os_str().as_bytes())?; + // SAFETY: `fifo_path` is a valid NUL-terminated path. + if unsafe { libc::mkfifo(fifo_path.as_ptr(), 0o644) } != 0 { + return Err(std::io::Error::last_os_error().into()); + } + crate::barrier::wait(dir, "held", 2)?; + + // Opening a FIFO to write waits until a reader opens it. + let mut fifo = std::fs::OpenOptions::new().write(true).open(file)?; + // The reader keeps the FIFO open, and later opens get the file. + std::fs::rename(&held, file)?; + crate::barrier::wait(dir, "released", 2)?; + // The reader reads the content, then the end of the file once `fifo` is + // closed. + fifo.write_all(&content)?; + Ok(()) +} + +#[cfg(windows)] +fn hold(file: &std::path::Path, dir: &std::path::Path) -> Result<(), Box> { + use std::os::windows::{fs::OpenOptionsExt as _, io::AsRawHandle as _}; + + use windows_sys::Win32::{ + Foundation::ERROR_IO_PENDING, + Storage::FileSystem::FILE_FLAG_OVERLAPPED, + System::{ + IO::{DeviceIoControl, GetOverlappedResult, OVERLAPPED}, + Ioctl::FSCTL_REQUEST_OPLOCK_LEVEL_1, + }, + }; + + // The oplock request stays pending while the oplock is held, and is + // completed when it breaks. Leaked so that it outlives the request on + // every path. + // SAFETY: `OVERLAPPED` is a plain C struct, valid when zeroed. + let overlapped: *mut OVERLAPPED = Box::into_raw(Box::new(unsafe { std::mem::zeroed() })); + // The default share mode lets the other process open the file, which + // breaks the oplock. + let held = + std::fs::OpenOptions::new().read(true).custom_flags(FILE_FLAG_OVERLAPPED).open(file)?; + let handle = held.as_raw_handle(); + // SAFETY: `handle` is an open file handle for overlapped I/O, and + // `overlapped` lives until the process exits. A level 1 oplock request + // takes no buffers. + let requested = unsafe { + DeviceIoControl( + handle, + FSCTL_REQUEST_OPLOCK_LEVEL_1, + std::ptr::null(), + 0, + std::ptr::null_mut(), + 0, + std::ptr::null_mut(), + overlapped, + ) + }; + let error = std::io::Error::last_os_error(); + if requested != 0 { + return Err("the oplock request completed without holding the oplock".into()); + } + if error.raw_os_error() != i32::try_from(ERROR_IO_PENDING).ok() { + return Err(std::format!("failed to take an oplock: {error}").into()); + } + crate::barrier::wait(dir, "held", 2)?; + + let mut transferred = 0; + // SAFETY: `overlapped` is the pending request on `handle`. With no event + // in it, the wait is on `handle`, which has no other I/O. + if unsafe { GetOverlappedResult(handle, overlapped, &raw mut transferred, 1) } == 0 { + return Err(std::io::Error::last_os_error().into()); + } + crate::barrier::wait(dir, "released", 2)?; + // Closing the handle acknowledges the break, and the waiting open + // proceeds. + drop(held); + Ok(()) +} diff --git a/crates/vt_bin/src/vtt/main.rs b/crates/vt_bin/src/vtt/main.rs index 1707704f3..652835535 100644 --- a/crates/vt_bin/src/vtt/main.rs +++ b/crates/vt_bin/src/vtt/main.rs @@ -13,6 +13,7 @@ mod exit; mod exit_on_ctrlc; mod fs; mod grep_file; +mod hold_reads; mod list_dir; mod mkdir; mod pipe_stdin; @@ -40,7 +41,7 @@ fn main() { if args.len() < 2 { eprintln!("Usage: vtt [args...]"); eprintln!( - "Subcommands: barrier, check-tty, cp, exit, exit-on-ctrlc, fs, grep-file, list-dir, mkdir, pipe-stdin, pipe-stdout, print, print-color, print-cwd, print-env, print-file, read-link, read-stdin, replace-file-content, rm, small_dev_shm, stalled-remote-cache, stat-file, stat-many, symlink, touch-file, write-file" + "Subcommands: barrier, check-tty, cp, exit, exit-on-ctrlc, fs, grep-file, hold-reads, list-dir, mkdir, pipe-stdin, pipe-stdout, print, print-color, print-cwd, print-env, print-file, read-link, read-stdin, replace-file-content, rm, small_dev_shm, stalled-remote-cache, stat-file, stat-many, symlink, touch-file, write-file" ); std::process::exit(1); } @@ -59,6 +60,7 @@ fn main() { grep_file::run(&args[2..]); Ok(()) } + "hold-reads" => hold_reads::run(&args[2..]), "list-dir" => list_dir::run(&args[2..]), "mkdir" => mkdir::run(&args[2..]), "pipe-stdin" => pipe_stdin::run(&args[2..]), diff --git a/crates/vt_bin/tests/e2e_snapshots/README.md b/crates/vt_bin/tests/e2e_snapshots/README.md index 742f5cf30..06b9ead0e 100644 --- a/crates/vt_bin/tests/e2e_snapshots/README.md +++ b/crates/vt_bin/tests/e2e_snapshots/README.md @@ -47,7 +47,7 @@ Notes: The test runner: 1. Copies the fixture to a temp directory -2. Starts each step's `argv` directly, without a shell, in an environment holding only `PATH`, `TERM`, and the step's `envs` (plus `PLAYWRIGHT_HOST_PLATFORM_OVERRIDE` when set, and a few system variables on Windows) +2. Starts each step's `argv` directly, without a shell, in an environment holding only `PATH`, `TERM`, `RAYON_NUM_THREADS=1` (so that tasks that become ready together start in the same order every run), and the step's `envs` (plus `PLAYWRIGHT_HOST_PLATFORM_OVERRIDE` when set, and a few system variables on Windows) 3. Runs each step in PTY mode (`TestTerminal`), with a timeout of 20 seconds (60 seconds on Windows and x86_64 macOS) 4. Applies configured interactions in order for PTY steps 5. Captures output and exit codes diff --git a/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/input.txt b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/input.txt new file mode 100644 index 000000000..3f9177e45 --- /dev/null +++ b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/input.txt @@ -0,0 +1 @@ +input diff --git a/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/package.json b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/package.json new file mode 100644 index 000000000..fe25552c7 --- /dev/null +++ b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/package.json @@ -0,0 +1,4 @@ +{ + "name": "cache-check-concurrency", + "private": true +} diff --git a/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots.toml b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots.toml new file mode 100644 index 000000000..848ffd515 --- /dev/null +++ b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots.toml @@ -0,0 +1,75 @@ +# Each case holds the reads of input.txt while read's cache entry is checked. +# read depends on reads-held, which finishes once hold-reads holds them, so +# the check opens input.txt and waits. hold-reads releases them only after +# release-reads starts, which depends on other, and the scheduler can only +# start release-reads once it sees other finish. If the check ran on the +# scheduler's thread, it would wait there for release-reads forever, and the +# step would time out. Once the reads are released, read hits while +# hold-reads and release-reads finish, in any order. Each task that finishes +# ends its output with a blank line, so those steps leave blank lines out. + +[[e2e]] +name = "checking_local_entry_doesnt_block_other_tasks" +steps = [ + { argv = [ + "vt", + "run", + "--ignore-depends-on", + "read", + ], comment = "read reads input.txt, and its entry is recorded locally." }, + { argv = [ + "vt", + "run", + "test", + ], comment = "Other tasks run while read's local entry is checked, and then it hits.", redact = { '\n\n+' = "\n" } }, +] + +[[e2e]] +name = "checking_remote_cache_entry_doesnt_block_other_tasks" +cfg = 'not(target_env = "musl")' +ignore = true +steps = [ + [ + "remote-cache-server", + "start", + ], + { argv = [ + "remote-cache-server", + "run", + "--github-actions", + "vt", + "run", + "--ignore-depends-on", + "read", + ], envs = [ + [ + "VP_REMOTE_CACHE", + "read-write", + ], + [ + "VP_RUN_INTERNAL_HIDE_PENDING_UPLOADS", + "1", + ], + ], comment = "read reads input.txt, and its entry is uploaded." }, + [ + "vt", + "cache", + "clean", + ], + { argv = [ + "remote-cache-server", + "run", + "vt", + "run", + "test", + ], envs = [ + [ + "VP_REMOTE_CACHE", + "read", + ], + ], comment = "Other tasks run while read's remote entry is checked, and then it hits.", redact = { '\n\n+' = "\n" } }, + [ + "remote-cache-server", + "stop", + ], +] diff --git a/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots/checking_local_entry_doesnt_block_other_tasks.md b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots/checking_local_entry_doesnt_block_other_tasks.md new file mode 100644 index 000000000..53a47ae83 --- /dev/null +++ b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots/checking_local_entry_doesnt_block_other_tasks.md @@ -0,0 +1,28 @@ +# checking_local_entry_doesnt_block_other_tasks + +## `vt run --ignore-depends-on read` + +read reads input.txt, and its entry is recorded locally. + +``` +$ vtt print-file input.txt +input +``` + +## `vt run test` + +Other tasks run while read's local entry is checked, and then it hits. + +``` +$ vtt hold-reads input.txt .barrier ⊘ cache disabled +$ vtt barrier .barrier held 2 ⊘ cache disabled +$ vtt print other ⊘ cache disabled +other +$ vtt barrier .barrier released 2 ⊘ cache disabled +$ vtt print-file input.txt ◉ cache hit, replaying +input +$ vtt print done ⊘ cache disabled +done +--- +vt run: 1/1 cache hit (100%), 6/6 successful. (Run `vt run --last-details` for full details) +``` diff --git a/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots/checking_remote_cache_entry_doesnt_block_other_tasks.md b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots/checking_remote_cache_entry_doesnt_block_other_tasks.md new file mode 100644 index 000000000..110995be0 --- /dev/null +++ b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/snapshots/checking_remote_cache_entry_doesnt_block_other_tasks.md @@ -0,0 +1,47 @@ +# checking_remote_cache_entry_doesnt_block_other_tasks + +## `remote-cache-server start` + +``` +``` + +## `VP_REMOTE_CACHE=read-write VP_RUN_INTERNAL_HIDE_PENDING_UPLOADS=1 remote-cache-server run --github-actions vt run --ignore-depends-on read` + +read reads input.txt, and its entry is uploaded. + +``` +$ vtt print-file input.txt +input + +[remote-cache] POST /fetch 404 +[remote-cache] POST /store 200 +``` + +## `vt cache clean` + +``` +``` + +## `VP_REMOTE_CACHE=read remote-cache-server run vt run test` + +Other tasks run while read's remote entry is checked, and then it hits. + +``` +$ vtt hold-reads input.txt .barrier ⊘ cache disabled +$ vtt barrier .barrier held 2 ⊘ cache disabled +$ vtt print other ⊘ cache disabled +other +$ vtt barrier .barrier released 2 ⊘ cache disabled +$ vtt print-file input.txt ◉ remote cache hit, replaying +input +$ vtt print done ⊘ cache disabled +done +--- +vt run: 1/1 cache hit (100%, 1 remote), 6/6 successful. (Run `vt run --last-details` for full details) +[remote-cache] POST /fetch 200 exact [-] +``` + +## `remote-cache-server stop` + +``` +``` diff --git a/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/vite-task.json b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/vite-task.json new file mode 100644 index 000000000..84ad19f9d --- /dev/null +++ b/crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/vite-task.json @@ -0,0 +1,31 @@ +{ + "tasks": { + "test": { + "command": "vtt print done", + "dependsOn": ["hold-reads", "read", "release-reads"], + "cache": false + }, + "hold-reads": { + "command": "vtt hold-reads input.txt .barrier", + "cache": false + }, + "reads-held": { + "command": "vtt barrier .barrier held 2", + "cache": false + }, + "read": { + "command": "vtt print-file input.txt", + "dependsOn": ["reads-held"] + }, + "other": { + "command": "vtt print other", + "dependsOn": ["reads-held"], + "cache": false + }, + "release-reads": { + "command": "vtt barrier .barrier released 2", + "dependsOn": ["other"], + "cache": false + } + } +} diff --git a/crates/vt_bin/tests/e2e_snapshots/main.rs b/crates/vt_bin/tests/e2e_snapshots/main.rs index 36f954b44..29f60bb36 100644 --- a/crates/vt_bin/tests/e2e_snapshots/main.rs +++ b/crates/vt_bin/tests/e2e_snapshots/main.rs @@ -529,6 +529,12 @@ fn run_case( // by vt100's `Screen::contents()` when the snapshot is rendered, // so this does not introduce colour-noise into existing snapshots. cmd.env("TERM", "xterm-256color"); + // `vp run` checks cache entries on rayon's pool, and a task starts + // once its check finishes. With one thread, the checks finish in + // the order they're submitted, so tasks that become ready + // together start, and print their commands, in the same order + // every run. + cmd.env("RAYON_NUM_THREADS", "1"); // The macOS x64 CI shard needs Playwright to select an x64 browser // while running under Rosetta on Apple Silicon. if let Some(value) = env::var_os("PLAYWRIGHT_HOST_PLATFORM_OVERRIDE") { diff --git a/crates/vt_glob/src/path.rs b/crates/vt_glob/src/path.rs index 3a718da42..f93ec1160 100644 --- a/crates/vt_glob/src/path.rs +++ b/crates/vt_glob/src/path.rs @@ -50,6 +50,19 @@ impl<'a> PathGlobSet<'a> { Ok(Self { patterns, has_negated }) } + /// The same set with its patterns copied, so it borrows nothing. + #[must_use] + pub fn into_owned(self) -> PathGlobSet<'static> { + let Self { patterns, has_negated } = self; + PathGlobSet { + patterns: patterns + .into_iter() + .map(|(glob, match_or_not)| (glob.into_owned(), match_or_not)) + .collect(), + has_negated, + } + } + /// Whether the set has no patterns, so it matches nothing. #[must_use] pub const fn is_empty(&self) -> bool {