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 {