perf(cache): check cache entries off the scheduler thread

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)).

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 }))
         }

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> {

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 }

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<dyn std::error::Error>> {
-    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<dyn std::error::Error>> {
         );
     }
 
-    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 `<prefix>` in `dir`: create `dir/<prefix>_<pid>`, then
+/// wait until `count` files matching `<prefix>_*` exist in `dir`.
+pub fn wait(
+    dir: &std::path::Path,
+    prefix: &str,
+    count: usize,
+) -> Result<(), Box<dyn std::error::Error>> {
+    use notify::Watcher as _;
 
     std::fs::create_dir_all(dir)?;
 
@@ -64,26 +94,5 @@ pub fn run(args: &[String]) -> Result<(), Box<dyn std::error::Error>> {
             }
         }
     }
-
-    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(())
 }

crates/vt_bin/src/vtt/hold_reads.rs

@@ -0,0 +1,108 @@
+/// hold-reads `<file>` `<dir>`
+///
+/// Holds other processes' reads of `<file>` until another process releases
+/// them, for testing that a blocked read doesn't stop other work:
+///
+/// 1. Make opening `<file>` wait, then join the barrier `<dir>` `held` 2,
+///    so that its other participant knows a read of `<file>` would wait.
+/// 2. Wait until another process opens `<file>`.
+/// 3. Join the barrier `<dir>` `released` 2, then let the waiting read finish
+///    with the file's content and exit.
+///
+/// On Unix, `<file>` 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 `<file>`. Another
+/// process's open breaks it and waits until the oplock's handle is closed.
+pub fn run(args: &[String]) -> Result<(), Box<dyn std::error::Error>> {
+    let [file, dir] = args else {
+        return Err("Usage: vtt hold-reads <file> <dir>".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<dyn std::error::Error>> {
+    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<dyn std::error::Error>> {
+    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(())
+}

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 <subcommand> [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..]),

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

crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/input.txt

@@ -0,0 +1 @@
+input

crates/vt_bin/tests/e2e_snapshots/fixtures/cache_check_concurrency/package.json

@@ -0,0 +1,4 @@
+{
+  "name": "cache-check-concurrency",
+  "private": true
+}