From b15c81c84d7304f924e2373912d1fa38a6b45b21 Mon Sep 17 00:00:00 2001 From: calvin-archastro Date: Tue, 10 Nov 2026 17:40:37 -0700 Subject: [PATCH] fix(affected): narrow pnpm-workspace.yaml edits; admit targets by longest dependent chain A packages:-only edit to pnpm-workspace.yaml reaches the root and the packages whose membership changed; other setting edits reach the packages that resolve from the root. Waiting targets take free slots by the longest chain they gate, then by request order, instead of whichever lock poll came first. Co-Authored-By: Claude Opus 5.5 --- src/executor/admission.rs | 246 ++++++++++++++++--------- src/executor/admission_order.rs | 207 +++++++++++++++++++++ src/executor/admission_unsupported.rs | 11 ++ src/executor/mod.rs | 1 + src/executor/runner.rs | 256 ++++++++++++++++++++++++-- src/git/affected.rs | 5 +- src/plugins/nodejs.rs | 2 +- src/related/engine.rs | 157 ++++++++++++++++ tests/execution_capacity.rs | 61 ++++++ tests/related.rs | 76 ++++++++ 10 files changed, 921 insertions(+), 101 deletions(-) create mode 100644 src/executor/admission_order.rs diff --git a/src/executor/admission.rs b/src/executor/admission.rs index f22f8e79..3efd8119 100644 --- a/src/executor/admission.rs +++ b/src/executor/admission.rs @@ -12,6 +12,8 @@ use std::process::Command; use std::sync::{Arc, OnceLock}; use std::time::Duration; +use super::admission_order::Ticket; + const CONTEXT_ENV: &str = "ASTER_EXECUTION_CONTEXT"; const SPEC_ENV: &str = "ASTER_EXECUTION_SPEC"; const INTERNAL_COMMAND: &str = "__aster_execute"; @@ -56,12 +58,14 @@ fn try_lock(file: &File) -> Result { } } +const POLL: Duration = Duration::from_millis(20); + fn wait() -> Result<()> { if super::shutdown_requested() { bail!("Cancelled: shutdown requested while waiting for execution admission"); } // Poll OS lock availability, never infer ownership from elapsed time. - std::thread::sleep(Duration::from_millis(20)); + std::thread::sleep(POLL); Ok(()) } @@ -162,6 +166,8 @@ impl Host { }))) } + /// Wait for admission, polling the OS locks. Waiters in this process race + /// each other; use [`Host::acquire_ordered`] to choose who goes first. pub(super) fn acquire( &self, resources: &[String], @@ -169,6 +175,64 @@ impl Host { address: &str, on_wait: impl FnOnce(String), ) -> Result { + let request = self.request(resources, workspace)?; + let mut on_wait = Some(on_wait); + loop { + if super::shutdown_requested() { + bail!("Cancelled: shutdown requested while waiting for execution admission"); + } + if let Attempt::Granted(permit) = self.attempt(&request)? { + return Ok(permit); + } + self.report_wait(&mut on_wait, address); + wait()?; + } + } + + /// Wait for admission in the order [`Gate`] gives: the waiter on the + /// longest remaining chain of dependents first, then the earliest request. + /// The ticket is released as soon as the permit is granted, so the next + /// waiter is not held behind a target that is already running. + pub(super) fn acquire_ordered( + &self, + resources: &[String], + workspace: &Path, + address: &str, + ticket: Ticket, + on_wait: impl FnOnce(String), + ) -> Result { + let request = self.request(resources, workspace)?; + let mut on_wait = Some(on_wait); + loop { + if super::shutdown_requested() { + bail!("Cancelled: shutdown requested while waiting for execution admission"); + } + if ticket.wait_turn(POLL) { + match self.attempt(&request)? { + Attempt::Granted(permit) => return Ok(permit), + // The slots are full: stay at the front of the line. + Attempt::Busy(Blocker::Capacity) => ticket.unpark(), + // A resource held elsewhere says nothing about the slots, + // so waiters behind this one may try them. + Attempt::Busy(Blocker::Resource) => ticket.park(), + } + self.report_wait(&mut on_wait, address); + wait()?; + } + } + } + + fn report_wait(&self, on_wait: &mut Option, address: &str) { + if let Some(on_wait) = on_wait.take() { + on_wait(format!( + "[aster] Waiting for execution admission: {address} (host capacity {})", + self.capacity + )); + } + } + + /// The resource keys this acquisition needs, minus those an ancestor owns. + fn request(&self, resources: &[String], workspace: &Path) -> Result { let workspace = fs::canonicalize(workspace)?; let mut keys = resources .iter() @@ -203,98 +267,114 @@ impl Host { } keys.retain(|k| !inherited_resources.contains(k)); } - let mut on_wait = Some(on_wait); - loop { - if super::shutdown_requested() { - bail!("Cancelled: shutdown requested while waiting for execution admission"); + Ok(Request { + keys, + inherited_resources, + }) + } + + /// One nonblocking try at the whole set of locks. + fn attempt(&self, request: &Request) -> Result { + let keys = &request.keys; + let mut files = vec![self.epoch.try_clone()?]; + let mut blocker = None; + // Acquire the whole set nonblocking. Never hold a resource while + // waiting for capacity, or a partial set while another caller runs. + for key in keys { + use sha2::{Digest, Sha256}; + let digest = Sha256::digest(key.as_bytes()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + let name = format!("resource-{digest}.lock"); + let file = open_lock(&self.directory.join(name))?; + if !try_lock(&file)? { + blocker = Some(Blocker::Resource); + break; } - let mut files = vec![self.epoch.try_clone()?]; - let mut ready = true; - // Acquire the whole set nonblocking. Never hold a resource while - // waiting for capacity, or a partial set while another caller runs. - for key in &keys { - use sha2::{Digest, Sha256}; - let digest = Sha256::digest(key.as_bytes()) - .iter() - .map(|byte| format!("{byte:02x}")) - .collect::(); - let name = format!("resource-{digest}.lock"); - let file = open_lock(&self.directory.join(name))?; - if !try_lock(&file)? { - ready = false; - break; + files.push(file); + } + if blocker.is_none() { + if let Some(family) = &self.inherited { + let owner = OpenOptions::new() + .read(true) + .write(true) + .open(family.directory.join("owner.lock")) + .context("inherited execution owner is gone")?; + if try_lock(&owner)? { + bail!("inherited execution owner is gone"); } - files.push(file); - } - if ready { - if let Some(family) = &self.inherited { - let owner = OpenOptions::new() - .read(true) - .write(true) - .open(family.directory.join("owner.lock")) - .context("inherited execution owner is gone")?; - if try_lock(&owner)? { - bail!("inherited execution owner is gone"); - } - let borrow = open_lock(&family.directory.join("borrow.lock"))?; - if try_lock(&borrow)? { - files.push(borrow); - } else { - ready = false; - } + let borrow = open_lock(&family.directory.join("borrow.lock"))?; + if try_lock(&borrow)? { + files.push(borrow); } else { - ready = false; - for n in 0..self.capacity { - let file = open_lock(&self.directory.join(format!("slot-{n}.lock")))?; - if try_lock(&file)? { - files.push(file); - ready = true; - break; - } + blocker = Some(Blocker::Capacity); + } + } else { + blocker = Some(Blocker::Capacity); + for n in 0..self.capacity { + let file = open_lock(&self.directory.join(format!("slot-{n}.lock")))?; + if try_lock(&file)? { + files.push(file); + blocker = None; + break; } } } - if ready { - let family_dir = tempfile::Builder::new() - .prefix("family-") - .tempdir_in(&self.directory)?; - let owner = open_lock(&family_dir.path().join("owner.lock"))?; - owner.lock_exclusive()?; - files.push(owner); - open_lock(&family_dir.path().join("borrow.lock"))?; - inherited_resources.extend(keys); - let family = Family { - directory: family_dir.path().to_path_buf(), - resources: inherited_resources, - ancestors: self - .inherited - .as_ref() - .map(|parent| { - let mut ancestors = parent.ancestors.clone(); - ancestors.push(parent.directory.clone()); - ancestors - }) - .unwrap_or_default(), - }; - return Ok(Permit { - files, - family, - _directory: family_dir, - executable: self.executable.clone(), - }); - } - drop(files); - if let Some(on_wait) = on_wait.take() { - on_wait(format!( - "[aster] Waiting for execution admission: {address} (host capacity {})", - self.capacity - )); - } - wait()?; } + if let Some(blocker) = blocker { + return Ok(Attempt::Busy(blocker)); + } + let family_dir = tempfile::Builder::new() + .prefix("family-") + .tempdir_in(&self.directory)?; + let owner = open_lock(&family_dir.path().join("owner.lock"))?; + owner.lock_exclusive()?; + files.push(owner); + open_lock(&family_dir.path().join("borrow.lock"))?; + let mut inherited_resources = request.inherited_resources.clone(); + inherited_resources.extend(keys.iter().cloned()); + let family = Family { + directory: family_dir.path().to_path_buf(), + resources: inherited_resources, + ancestors: self + .inherited + .as_ref() + .map(|parent| { + let mut ancestors = parent.ancestors.clone(); + ancestors.push(parent.directory.clone()); + ancestors + }) + .unwrap_or_default(), + }; + Ok(Attempt::Granted(Permit { + files, + family, + _directory: family_dir, + executable: self.executable.clone(), + })) } } +/// What one admission needs, worked out once per acquisition. +struct Request { + keys: Vec, + inherited_resources: Vec, +} + +enum Attempt { + Granted(Permit), + Busy(Blocker), +} + +/// What one failed attempt was stopped by. +enum Blocker { + /// A named resource is held by another target or invocation. + Resource, + /// Every execution slot is taken. + Capacity, +} + pub(super) struct Permit { files: Vec, family: Family, diff --git a/src/executor/admission_order.rs b/src/executor/admission_order.rs new file mode 100644 index 00000000..8f7d4d54 --- /dev/null +++ b/src/executor/admission_order.rs @@ -0,0 +1,207 @@ +//! The order in which one invocation's ready targets take execution slots. +//! +//! Every ready target runs on its own thread, and each thread polls the host +//! locks until a slot is free. Left alone, whichever thread polls first after +//! a slot frees wins, so a target that gates a long chain of dependents can +//! sit behind targets that gate nothing. A [`Gate`] gives the waiting threads +//! a line: the target with the longest remaining chain of dependents goes +//! first, and requests with equal chains go in the order they were made. +//! +//! The line orders one invocation's own targets. Other invocations on the +//! host still compete through the OS locks. + +#[cfg(unix)] +use std::cmp::Reverse; +use std::sync::{Arc, Condvar, Mutex, MutexGuard}; +#[cfg(unix)] +use std::time::Duration; + +/// Waiting targets of one invocation. +#[derive(Default)] +pub(crate) struct Gate { + waiting: Mutex, + changed: Condvar, +} + +#[derive(Default)] +struct Waiting { + next: u64, + entries: Vec, +} + +struct Entry { + seq: u64, + /// Targets on the longest chain that starts at this one, itself included. + #[cfg_attr(not(unix), allow(dead_code))] + chain: usize, + /// Held up by something other than the slots (an exclusive resource), so + /// waiters behind it need not wait for it. + parked: bool, +} + +impl Entry { + /// Smaller sorts first: longer chain, then earlier request. + #[cfg(unix)] + fn priority(&self) -> (Reverse, u64) { + (Reverse(self.chain), self.seq) + } +} + +impl Gate { + pub(crate) fn new() -> Arc { + Arc::new(Self::default()) + } + + /// Join the line. Call from the thread that launches targets, in launch + /// order, before the target's own thread starts: requests with the same + /// chain length then keep that order. + pub(crate) fn enter(self: &Arc, chain: usize) -> Ticket { + let mut waiting = self.lock(); + let seq = waiting.next; + waiting.next += 1; + waiting.entries.push(Entry { + seq, + chain, + parked: false, + }); + Ticket { + gate: Arc::clone(self), + seq, + } + } + + fn lock(&self) -> MutexGuard<'_, Waiting> { + self.waiting + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +/// A place in the line. Dropping it leaves the line. +pub(crate) struct Ticket { + gate: Arc, + seq: u64, +} + +impl Ticket { + fn with_entry(&self, change: impl FnOnce(&mut Entry)) { + let mut waiting = self.gate.lock(); + if let Some(entry) = waiting.entries.iter_mut().find(|e| e.seq == self.seq) { + change(entry); + } + drop(waiting); + self.gate.changed.notify_all(); + } + + /// Stop holding up the targets behind this one while something other + /// than a free slot is awaited. + pub(crate) fn park(&self) { + self.with_entry(|entry| entry.parked = true); + } + + /// Hold up the targets behind this one again. + pub(crate) fn unpark(&self) { + self.with_entry(|entry| entry.parked = false); + } + + /// Whether it is this target's turn to try for a slot, waiting up to + /// `limit` for it to become so. It is not while a waiter ahead of it + /// (longer chain, or the same chain requested earlier) still wants a slot. + #[cfg(unix)] + pub(crate) fn wait_turn(&self, limit: Duration) -> bool { + let waiting = self.gate.lock(); + let (waiting, _) = self + .gate + .changed + .wait_timeout_while(waiting, limit, |waiting| !self.is_turn(waiting)) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + self.is_turn(&waiting) + } + + #[cfg(unix)] + fn is_turn(&self, waiting: &Waiting) -> bool { + let Some(mine) = waiting.entries.iter().find(|e| e.seq == self.seq) else { + return true; + }; + !waiting + .entries + .iter() + .any(|other| !other.parked && other.priority() < mine.priority()) + } +} + +impl Drop for Ticket { + fn drop(&mut self) { + let mut waiting = self.gate.lock(); + waiting.entries.retain(|e| e.seq != self.seq); + drop(waiting); + self.gate.changed.notify_all(); + } +} + +#[cfg(all(test, unix))] +mod tests { + use super::*; + + const NOW: Duration = Duration::ZERO; + + #[test] + fn the_longest_chain_goes_first_whatever_the_request_order() { + let gate = Gate::new(); + let independent = gate.enter(1); + let gating = gate.enter(3); + let middle = gate.enter(2); + assert!(!independent.wait_turn(NOW)); + assert!(!middle.wait_turn(NOW)); + assert!(gating.wait_turn(NOW)); + + drop(gating); + assert!(!independent.wait_turn(NOW)); + assert!(middle.wait_turn(NOW)); + + drop(middle); + assert!(independent.wait_turn(NOW)); + } + + #[test] + fn equal_chains_go_in_request_order() { + let gate = Gate::new(); + let first = gate.enter(1); + let second = gate.enter(1); + let third = gate.enter(1); + assert!(first.wait_turn(NOW)); + assert!(!second.wait_turn(NOW)); + assert!(!third.wait_turn(NOW)); + drop(first); + assert!(second.wait_turn(NOW)); + assert!(!third.wait_turn(NOW)); + } + + #[test] + fn a_target_waiting_on_a_resource_does_not_hold_up_the_rest() { + let gate = Gate::new(); + let blocked = gate.enter(5); + let other = gate.enter(1); + assert!(!other.wait_turn(NOW)); + blocked.park(); + assert!(other.wait_turn(NOW)); + // Parked, it still waits for any waiter that is ahead and not parked. + let ahead = gate.enter(9); + assert!(!blocked.wait_turn(NOW)); + drop(ahead); + blocked.unpark(); + assert!(blocked.wait_turn(NOW)); + assert!(!other.wait_turn(NOW)); + } + + #[test] + fn a_waiter_wakes_when_the_one_ahead_leaves() { + let gate = Gate::new(); + let ahead = gate.enter(2); + let behind = gate.enter(1); + let waiter = std::thread::spawn(move || behind.wait_turn(Duration::from_secs(10))); + std::thread::sleep(Duration::from_millis(50)); + drop(ahead); + assert!(waiter.join().unwrap()); + } +} diff --git a/src/executor/admission_unsupported.rs b/src/executor/admission_unsupported.rs index 90c21224..13189de8 100644 --- a/src/executor/admission_unsupported.rs +++ b/src/executor/admission_unsupported.rs @@ -30,4 +30,15 @@ impl Host { ) -> Result { unreachable!("host admission is not enabled on this platform") } + + pub(super) fn acquire_ordered( + &self, + _resources: &[String], + _workspace: &Path, + _address: &str, + _ticket: super::admission_order::Ticket, + _on_wait: impl FnOnce(String), + ) -> Result { + unreachable!("host admission is not enabled on this platform") + } } diff --git a/src/executor/mod.rs b/src/executor/mod.rs index b35afb55..2840b123 100644 --- a/src/executor/mod.rs +++ b/src/executor/mod.rs @@ -10,6 +10,7 @@ pub mod admission; #[cfg(not(unix))] #[path = "admission_unsupported.rs"] pub mod admission; +pub(crate) mod admission_order; pub(crate) mod command; pub mod logs; pub mod runner; diff --git a/src/executor/runner.rs b/src/executor/runner.rs index 02f9204b..8e2e1ee4 100644 --- a/src/executor/runner.rs +++ b/src/executor/runner.rs @@ -10,6 +10,7 @@ //! tie-break between targets that become ready together use crate::{errln, out, outln}; +use std::cmp::Reverse; use std::collections::{BTreeSet, HashMap, HashSet}; use std::io::Read; use std::path::{Path, PathBuf}; @@ -48,6 +49,7 @@ mod chain; mod retry; use super::admission::{Host, Permit}; +use super::admission_order::Gate; use super::command::parse_command; use super::{force_shutdown_requested, shutdown_requested}; #[cfg(any(unix, windows))] @@ -743,6 +745,7 @@ impl<'a> Executor<'a> { // One lock per exclusive resource for the whole run: targets that // share a resource serialize even when they sit at different levels. let resource_mutexes = exclusive_resource_locks(targets_to_run, project_map); + let gate = Gate::new(); let (tx, rx) = mpsc::channel(); let ctx = LaunchContext { project_map, @@ -755,6 +758,7 @@ impl<'a> Executor<'a> { runtime_inputs: &runtime_inputs, resource_mutexes: &resource_mutexes, host: &host, + gate: &gate, tx: &tx, }; @@ -871,6 +875,7 @@ impl<'a> Executor<'a> { &mut progress, show_progress, &mut handles, + queue.chain(&target_addr), ) { running += 1; } else { @@ -964,6 +969,7 @@ impl<'a> Executor<'a> { progress: &mut ProgressDisplay, show_progress: bool, handles: &mut Vec>, + gated_chain: usize, ) -> bool { let LaunchContext { project_map, @@ -976,6 +982,7 @@ impl<'a> Executor<'a> { runtime_inputs, resource_mutexes, host, + gate, tx: _, } = *ctx; let target_addr = &target_addr.to_string(); @@ -1346,6 +1353,9 @@ impl<'a> Executor<'a> { }; let host = host.clone(); + // Join the line for a slot now, on the launching thread, so requests + // keep launch order whatever order the threads start in. + let ticket = host.as_ref().map(|_| gate.enter(gated_chain)); let admission_wait_hook = self.admission_wait_hook.clone(); let resource_names = project .targets @@ -1360,11 +1370,26 @@ impl<'a> Executor<'a> { if test_hooks::should_panic(&addr) { panic!("injected panic for {addr}"); } - // Acquire exclusive resource locks (sorted to prevent deadlocks) + // Acquire exclusive resource locks (sorted to prevent deadlocks). + // A target that has to wait for one steps out of the line for + // a slot, so it cannot hold up targets behind it, or wait on + // a resource whose holder waits behind it. let _resource_guards: Vec<_> = target_resource_locks .iter() - .map(|m| m.lock().unwrap_or_else(|poisoned| poisoned.into_inner())) + .map(|m| match m.try_lock() { + Ok(guard) => guard, + Err(std::sync::TryLockError::Poisoned(poisoned)) => poisoned.into_inner(), + Err(std::sync::TryLockError::WouldBlock) => { + if let Some(ticket) = &ticket { + ticket.park(); + } + m.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) + } + }) .collect(); + if let Some(ticket) = &ticket { + ticket.unpark(); + } if shutdown_requested() { let result = ExecutionResult { @@ -1387,10 +1412,17 @@ impl<'a> Executor<'a> { } let permit = match host .as_ref() - .map(|host| { - host.acquire(&resource_names, &workspace_root_path, &addr, |line| { - report_admission_wait(&admission_wait_hook, line); - }) + .zip(ticket) + .map(|(host, ticket)| { + host.acquire_ordered( + &resource_names, + &workspace_root_path, + &addr, + ticket, + |line| { + report_admission_wait(&admission_wait_hook, line); + }, + ) }) .transpose() { @@ -2060,6 +2092,8 @@ struct LaunchContext<'a> { runtime_inputs: &'a Arc, resource_mutexes: &'a HashMap>>, host: &'a Option>, + /// Orders the targets waiting for an execution slot. + gate: &'a Arc, tx: &'a mpsc::Sender, } @@ -2089,13 +2123,18 @@ fn exclusive_resource_locks( /// /// A target becomes ready when every dependency inside the run has /// completed (passed, failed or been blocked). Ready targets come out ordered -/// by display level, then address, so launch order is deterministic. +/// by the longest chain of dependents they gate (longest first), then display +/// level, then address, so launch order is deterministic and a target that +/// gates a long chain asks for an execution slot before one that gates nothing. struct ReadyQueue<'a> { /// Dependencies of each target that have not completed yet. unmet: HashMap>, dependents: HashMap>, - ready: BTreeSet<(usize, String)>, + ready: BTreeSet<(Reverse, usize, String)>, level_of: &'a HashMap, + /// Targets on the longest chain of dependents that starts at each target, + /// itself included. + chain: HashMap, } impl<'a> ReadyQueue<'a> { @@ -2109,6 +2148,7 @@ impl<'a> ReadyQueue<'a> { dependents: HashMap::new(), ready: BTreeSet::new(), level_of, + chain: HashMap::new(), }; for addr in targets { let deps: HashSet = target_dependencies(addr, project_map) @@ -2122,24 +2162,42 @@ impl<'a> ReadyQueue<'a> { .or_default() .push(addr.clone()); } - if deps.is_empty() { - queue.ready.insert((queue.level(addr), addr.clone())); - } else { + if !deps.is_empty() { queue.unmet.insert(addr.clone(), deps); } } + queue.chain = chain_lengths(targets, &queue.dependents); + for addr in targets { + if !queue.unmet.contains_key(addr) { + queue.ready.insert(queue.ready_key(addr)); + } + } queue } + fn ready_key(&self, addr: &str) -> (Reverse, usize, String) { + ( + Reverse(self.chain(addr)), + self.level(addr), + addr.to_string(), + ) + } + + /// Targets on the longest chain of dependents starting at `addr`. + fn chain(&self, addr: &str) -> usize { + self.chain.get(addr).copied().unwrap_or(1) + } + fn level(&self, addr: &str) -> usize { self.level_of.get(addr).copied().unwrap_or(0) } - /// Take every ready target, lowest level and address first. + /// Take every ready target: longest dependent chain first, then lowest + /// level and address. fn take_ready(&mut self) -> Vec { std::mem::take(&mut self.ready) .into_iter() - .map(|(_, addr)| addr) + .map(|(_, _, addr)| addr) .collect() } @@ -2166,12 +2224,51 @@ impl<'a> ReadyQueue<'a> { unmet.remove(addr); if unmet.is_empty() { self.unmet.remove(&dependent); - self.ready.insert((self.level(&dependent), dependent)); + self.ready.insert(self.ready_key(&dependent)); } } } } +/// For each target, the number of targets on the longest chain of +/// dependents that starts at it, itself included: the work that cannot begin +/// until it finishes. A dependency cycle contributes nothing past its entry +/// point. +fn chain_lengths( + targets: &HashSet, + dependents: &HashMap>, +) -> HashMap { + fn visit( + addr: &str, + dependents: &HashMap>, + lengths: &mut HashMap, + visiting: &mut HashSet, + ) -> usize { + if let Some(length) = lengths.get(addr) { + return *length; + } + if !visiting.insert(addr.to_string()) { + return 0; + } + let longest = dependents + .get(addr) + .into_iter() + .flatten() + .map(|dependent| visit(dependent, dependents, lengths, visiting)) + .max() + .unwrap_or(0); + visiting.remove(addr); + lengths.insert(addr.to_string(), longest + 1); + longest + 1 + } + let mut lengths = HashMap::new(); + let mut visiting = HashSet::new(); + for addr in targets { + visit(addr, dependents, &mut lengths, &mut visiting); + } + lengths +} + /// Compute DAG levels for target execution /// /// Level 0 = targets with no dependencies (in our set) @@ -4645,7 +4742,8 @@ mod tests { .collect(); let mut queue = ReadyQueue::new(&set, &project_map, &level_of); - assert_eq!(queue.take_ready(), vec!["//p:a", "//p:b"]); + // b gates c and d (three targets deep); a gates only d. + assert_eq!(queue.take_ready(), vec!["//p:b", "//p:a"]); assert!(queue.take_ready().is_empty()); queue.complete("//p:b"); assert_eq!(queue.take_ready(), vec!["//p:c"]); @@ -4656,6 +4754,134 @@ mod tests { assert!(queue.take_stuck().is_empty()); } + /// Level and address of every target of `set`, as the runner computes them. + fn levels_of( + set: &HashSet, + project_map: &HashMap, + ) -> HashMap { + compute_target_levels(set, project_map) + .iter() + .enumerate() + .flat_map(|(level, addrs)| addrs.iter().map(move |addr| (addr.clone(), level))) + .collect() + } + + #[test] + fn ready_targets_come_out_longest_dependent_chain_first() { + // z-gate gates chain1 and chain2; the independents gate nothing. + let mut targets = HashMap::new(); + targets.insert("z-gate".to_string(), target("true", vec![])); + targets.insert("chain1".to_string(), target("true", vec!["//p:z-gate"])); + targets.insert("chain2".to_string(), target("true", vec!["//p:chain1"])); + targets.insert("a-solo".to_string(), target("true", vec![])); + targets.insert("b-solo".to_string(), target("true", vec![])); + targets.insert("diamond".to_string(), target("true", vec!["//p:a-solo"])); + let project = make_project_with_targets("p", targets); + let project_map = HashMap::from([("//p".to_string(), &project)]); + let set: HashSet = ["z-gate", "chain1", "chain2", "a-solo", "b-solo", "diamond"] + .iter() + .map(|t| format!("//p:{t}")) + .collect(); + let level_of = levels_of(&set, &project_map); + + let mut queue = ReadyQueue::new(&set, &project_map, &level_of); + assert_eq!(queue.chain("//p:z-gate"), 3); + assert_eq!(queue.chain("//p:a-solo"), 2); + assert_eq!(queue.chain("//p:b-solo"), 1); + // Longest chain first; equal chains by level, then address. + assert_eq!( + queue.take_ready(), + vec!["//p:z-gate", "//p:a-solo", "//p:b-solo"] + ); + // A target released later takes the chain it gates, not its release order. + queue.complete("//p:z-gate"); + assert_eq!(queue.take_ready(), vec!["//p:chain1"]); + } + + #[test] + fn chain_length_survives_a_dependency_cycle() { + let mut targets = HashMap::new(); + targets.insert("a".to_string(), target("true", vec!["//p:b"])); + targets.insert("b".to_string(), target("true", vec!["//p:a"])); + let project = make_project_with_targets("p", targets); + let project_map = HashMap::from([("//p".to_string(), &project)]); + let set: HashSet = ["a", "b"].iter().map(|t| format!("//p:{t}")).collect(); + let level_of = levels_of(&set, &project_map); + let queue = ReadyQueue::new(&set, &project_map, &level_of); + assert!(queue.chain("//p:a") >= 1 && queue.chain("//p:b") >= 1); + } + + /// Entry 334 in miniature: one short target (`z-gate`, last by address) + /// gates a chain of longer targets, while twelve independent targets + /// compete for four slots. Runs the real ready queue against a simulated + /// clock and returns when the last target finishes. + fn makespan(prioritise_chains: bool) -> u32 { + let mut targets = HashMap::new(); + let mut seconds: HashMap = HashMap::new(); + let mut add = |name: &str, deps: Vec<&str>, duration: u32| { + targets.insert(name.to_string(), target("true", deps)); + seconds.insert(format!("//p:{name}"), duration); + }; + add("z-gate", vec![], 1); + add("chain1", vec!["//p:z-gate"], 10); + add("chain2", vec!["//p:chain1"], 10); + for n in 0..12 { + add(&format!("crate-{n:02}"), vec![], 5); + } + let project = make_project_with_targets("p", targets); + let project_map = HashMap::from([("//p".to_string(), &project)]); + let set: HashSet = seconds.keys().cloned().collect(); + let level_of = levels_of(&set, &project_map); + + let capacity = 4; + let mut queue = ReadyQueue::new(&set, &project_map, &level_of); + // Waiting for a slot, with the request order they arrived in. + let mut waiting: Vec<(usize, String)> = Vec::new(); + let mut running: Vec<(u32, String)> = Vec::new(); + let (mut now, mut requests) = (0, 0); + loop { + for addr in queue.take_ready() { + waiting.push((requests, addr)); + requests += 1; + } + while running.len() < capacity && !waiting.is_empty() { + let next = (0..waiting.len()) + .min_by_key(|&i| { + let (request, addr) = &waiting[i]; + if prioritise_chains { + (Reverse(queue.chain(addr)), *request, String::new()) + } else { + // Without priorities: level, then address. + (Reverse(0), queue.level(addr), addr.clone()) + } + }) + .unwrap(); + let (_, addr) = waiting.remove(next); + running.push((now + seconds[&addr], addr)); + } + let Some(end) = running.iter().map(|(end, _)| *end).min() else { + return now; + }; + now = end; + for (_, addr) in running.iter().filter(|(e, _)| *e == end) { + queue.complete(addr); + } + running.retain(|(e, _)| *e != end); + } + } + + #[test] + fn starting_the_gating_chain_first_shortens_the_run() { + let by_address = makespan(false); + let by_chain = makespan(true); + // Address order starts z-gate after three rounds of the independents + // (15 s), then needs 1 + 10 + 10 s for the chain. Starting it first + // runs the chain beside the independents. The chain's last target + // gates nothing, so it still queues in request order behind them. + assert_eq!(by_address, 36); + assert_eq!(by_chain, 25); + } + #[test] fn test_execution_result_struct() { let result = ExecutionResult { diff --git a/src/git/affected.rs b/src/git/affected.rs index 84135de0..bf2bc723 100644 --- a/src/git/affected.rs +++ b/src/git/affected.rs @@ -296,10 +296,11 @@ impl AffectedDetector { if change .path .file_name() - .is_some_and(|n| n == "pnpm-lock.yaml") + .is_some_and(|n| n == "pnpm-lock.yaml" || n == "pnpm-workspace.yaml") { // The related engine diffs both lockfiles to find the - // importers and packages that changed. + // importers and packages that changed, and reads both + // workspace files to find the packages they list. if !change.deleted { change.old_source = blob(&base_tree, &change.path); change.new_source = match &head_tree { diff --git a/src/plugins/nodejs.rs b/src/plugins/nodejs.rs index 39e20b38..f9299379 100644 --- a/src/plugins/nodejs.rs +++ b/src/plugins/nodejs.rs @@ -897,7 +897,7 @@ fn workspace_patterns_at(dir: &Path) -> Vec { /// /// This is a deliberately minimal parser (the project has no YAML dependency); it /// reads just the one field aster needs and ignores everything else. -fn parse_pnpm_workspace_packages(content: &str) -> Vec { +pub(crate) fn parse_pnpm_workspace_packages(content: &str) -> Vec { let mut packages = Vec::new(); let mut in_block = false; diff --git a/src/related/engine.rs b/src/related/engine.rs index ccd5c4f9..2d575d80 100644 --- a/src/related/engine.rs +++ b/src/related/engine.rs @@ -686,6 +686,128 @@ fn lockfile_plugin(name: &str) -> Option<&'static str> { } } +/// Whether a `pnpm-workspace.yaml` change can reach `candidate`, a Node +/// project under the file's directory `dir`. +/// +/// The root is always reached, and so is any project when a side of the file +/// cannot be read. When only the `packages:` list changed, the packages +/// whose membership changed are reached (their lockfile importers are judged +/// separately) and nothing else installs differently. When any other +/// setting changed (overrides, patches, catalogs, the linker, hoisting), +/// every package either version lists is reached, and so is a project the +/// list does not name unless it is installed on its own: a project with no +/// `node_modules` or lockfile of its own resolves modules from the root's by +/// directory walk-up. +fn pnpm_workspace_reaches(change: &Change, candidate: &DiscoveredProject, dir: &Path) -> bool { + let member = candidate + .relative_path + .strip_prefix(dir) + .unwrap_or(Path::new("")); + if member.as_os_str().is_empty() { + return true; + } + let (Some(old), Some(new)) = (change.old_source.as_deref(), change.new_source.as_deref()) + else { + return true; + }; + let member = member.to_string_lossy().replace('\\', "/"); + let (Some(was), Some(is)) = ( + pnpm_workspace_has(old, &member), + pnpm_workspace_has(new, &member), + ) else { + return true; + }; + if pnpm_workspace_settings(old) == pnpm_workspace_settings(new) { + was != is + } else { + was || is || !installed_on_its_own(candidate, dir) + } +} + +/// Whether a project, or a directory between it and the workspace root +/// `dir`, holds its own install: a pnpm workspace file, lockfile or +/// `node_modules`. +fn installed_on_its_own(candidate: &DiscoveredProject, dir: &Path) -> bool { + let depth = candidate + .relative_path + .strip_prefix(dir) + .map_or(0, |rest| rest.components().count()); + candidate.root.ancestors().take(depth).any(|path| { + ["pnpm-workspace.yaml", "pnpm-lock.yaml", "node_modules"] + .iter() + .any(|name| path.join(name).exists()) + }) +} + +/// Whether a `pnpm-workspace.yaml` lists the package at `member`; `None` +/// when the file has no readable `packages:` list and could list anything. +fn pnpm_workspace_has(text: &str, member: &str) -> Option { + use globset::GlobBuilder; + let patterns = crate::plugins::nodejs::parse_pnpm_workspace_packages(text); + if patterns.is_empty() { + return None; + } + let matches = |pattern: &str| { + let pattern = pattern.trim_start_matches("./").trim_end_matches('/'); + GlobBuilder::new(pattern) + .literal_separator(true) + .build() + // A pattern that does not parse might list anything. + .map_or(true, |glob| glob.compile_matcher().is_match(member)) + }; + let (excluded, included): (Vec<&String>, Vec<&String>) = + patterns.iter().partition(|p| p.starts_with('!')); + Some(included.iter().any(|p| matches(p)) && !excluded.iter().any(|p| matches(&p[1..]))) +} + +/// The lines of a `pnpm-workspace.yaml` outside its `packages:` list, +/// without comments and blank lines. +fn pnpm_workspace_settings(text: &str) -> Vec { + let mut settings = Vec::new(); + let mut in_packages = false; + for raw in text.lines() { + let line = without_yaml_comment(raw).trim_end(); + if line.trim().is_empty() { + continue; + } + if line.starts_with("packages:") { + in_packages = true; + continue; + } + if in_packages + && line.starts_with(char::is_whitespace) + && line.trim_start().starts_with('-') + { + continue; + } + in_packages = false; + settings.push(line.to_string()); + } + settings +} + +/// A YAML line without its comment: `#` starts one only at the start of the +/// line or after whitespace, and outside quotes. +fn without_yaml_comment(line: &str) -> &str { + let mut quote: Option = None; + let mut previous = ' '; + for (index, c) in line.char_indices() { + match quote { + Some(open) if c == open => quote = None, + Some(_) => {} + None if (c == '\'' || c == '"') + && (previous.is_whitespace() || ":[,{".contains(previous)) => + { + quote = Some(c) + } + None if c == '#' && previous.is_whitespace() => return &line[..index], + None => {} + } + previous = c; + } + line +} + fn spans(lines: &Option>) -> String { match lines { Some(lines) if !lines.is_empty() => { @@ -1878,6 +2000,14 @@ impl<'a> Index<'a> { .flatten(); for (i, candidate) in self.projects.iter().enumerate() { if candidate.plugin_name == plugin && candidate.relative_path.starts_with(dir) { + // A workspace file governs the root and the packages + // it lists. A project it does not list is installed + // on its own and cannot see the edit. + if name == "pnpm-workspace.yaml" + && !pnpm_workspace_reaches(change, candidate, dir) + { + continue; + } let mut reason = format!("{} changed", change.path.display()); if let Some(pnpm_lock::LockImpact::Importers(hit)) = &pnpm { let importer = candidate @@ -3675,3 +3805,30 @@ mod config_change_tests { assert_eq!(change("targets = 1\n"), ConfigChange::Other); } } + +#[cfg(test)] +mod pnpm_workspace_tests { + use super::{pnpm_workspace_settings, without_yaml_comment}; + + #[test] + fn yaml_comments_start_only_after_whitespace_outside_quotes() { + assert_eq!(without_yaml_comment("a: b # note"), "a: b "); + assert_eq!(without_yaml_comment("# whole line"), ""); + assert_eq!(without_yaml_comment("a: b#c"), "a: b#c"); + assert_eq!(without_yaml_comment("a: 'b # c'"), "a: 'b # c'"); + assert_eq!(without_yaml_comment("a: \"b # c\" # d"), "a: \"b # c\" "); + assert_eq!(without_yaml_comment("a: don't # d"), "a: don't "); + } + + #[test] + fn a_hash_inside_a_value_is_still_a_settings_change() { + let old = "packages:\n - 'a'\nregistry: https://x/#one\n"; + let new = "packages:\n - 'a'\nregistry: https://x/#two\n"; + assert_ne!(pnpm_workspace_settings(old), pnpm_workspace_settings(new)); + let commented = "packages:\n - 'a'\n# note\nregistry: https://x/#one # why\n"; + assert_eq!( + pnpm_workspace_settings(old), + pnpm_workspace_settings(commented) + ); + } +} diff --git a/tests/execution_capacity.rs b/tests/execution_capacity.rs index 10217462..c66b321b 100644 --- a/tests/execution_capacity.rs +++ b/tests/execution_capacity.rs @@ -632,6 +632,67 @@ fn cached_work_bypasses_capacity_and_affected_dependencies_make_progress() { assert!(affected.finish()); } +/// Entry 334 of the firstlanding CI audit: a short target that gates a chain +/// of dependents waited for a slot behind targets that gate nothing, because +/// every ready target polled the host locks and the first poll after a slot +/// freed won. The target with the longest chain of dependents must take a +/// slot first, whatever its name sorts like or which thread polls first. +#[test] +fn the_target_gating_the_longest_chain_takes_a_slot_before_independent_targets() { + // Setup: four slots; `z-gate` sorts after eight independent targets and + // gates `chain1`. Every target holds its slot until the test releases it. + let f = Fixture::new(); + let root = f.root.path().join("sched"); + fs::create_dir(&root).unwrap(); + fs::write(root.join("package.json"), r#"{"name":"sched"}"#).unwrap(); + let probe = |label: &str| { + format!( + "python3 {} {} {label} normal", + aster::executor::quote_command_argument( + &f.root.path().join("probe.py").display().to_string() + ), + f.listener.local_addr().unwrap().port() + ) + }; + let mut config = String::from("name = 'sched'\n"); + let mut requested = vec!["//:chain1".to_string()]; + for n in 0..8 { + config += &format!( + "[targets.i{n}]\ncommand = {:?}\ncache = {{ enabled = false }}\n", + probe(&format!("i{n}")) + ); + requested.push(format!("//:i{n}")); + } + config += &format!( + "[targets.z-gate]\ncommand = {:?}\ncache = {{ enabled = false }}\n", + probe("z-gate") + ); + config += &format!( + "[targets.chain1]\ncommand = {:?}\ndepends_on = ['//:z-gate']\ncache = {{ enabled = false }}\n", + probe("chain1") + ); + fs::write(root.join("aster.toml"), config).unwrap(); + + // Action: run all ten targets at once on four slots. + let mut command = f.command(&root, 4); + command.arg("run").args(&requested).arg("--quiet"); + let mut run = Run::spawn(command); + + // Outcome: the first four admitted include the gate, which an unordered + // poll would admit last of the nine ready targets about half the time. + let first_four: Vec = (0..4).map(|_| f.started()).collect(); + let mut labels: Vec<&str> = first_four.iter().map(|t| t.label.as_str()).collect(); + labels.sort(); + assert_eq!(labels, ["i0", "i1", "i2", "z-gate"]); + + // Cleanup: release everything as it starts so the run completes. + first_four.into_iter().for_each(Target::release); + for _ in 0..6 { + f.started().release(); + } + assert!(run.finish()); +} + #[test] fn multiple_ready_targets_share_two_slots_across_invocations() { let f = Fixture::new(); diff --git a/tests/related.rs b/tests/related.rs index 26eeb0fa..56c3c609 100644 --- a/tests/related.rs +++ b/tests/related.rs @@ -4447,6 +4447,82 @@ snapshots: ), ]; +/// An edit to `pnpm-workspace.yaml` reaches the root and the packages it +/// lists. A project it does not list resolves modules from the root's +/// `node_modules`, so a setting change reaches it too unless it is installed +/// on its own (its own lockfile); a change to the package list alone does not. +#[test] +fn a_pnpm_workspace_edit_runs_only_the_packages_it_can_reach() { + // Setup: the workspace lists `pkg-*`; `wrapper` has a package.json so + // Aster discovers it, but no pattern lists it and it has no install of + // its own; `standalone` is unlisted with its own lockfile. + let ws = Workspace::new(PNPM_WORKSPACE); + ws.write( + "wrapper/package.json", + r#"{"name":"wrapper","scripts":{"test":"vitest run"}}"#, + ); + ws.write( + "standalone/package.json", + r#"{"name":"standalone","scripts":{"test":"vitest run"}}"#, + ); + ws.write("standalone/pnpm-lock.yaml", "lockfileVersion: '9.0'\n"); + ws.git(&["add", "-A"]); + ws.git(&[ + "-c", + "user.name=t", + "-c", + "user.email=t@example.com", + "commit", + "-q", + "-m", + "wrapper", + ]); + + // Action 1: a setting changes (an override applies to every package). + ws.write( + "pnpm-workspace.yaml", + "packages:\n - 'pkg-*'\noverrides:\n left-pad: 1.0.1\n", + ); + let plan = ws.plan("test"); + // Outcome: both listed packages run in full, and so does the unlisted + // wrapper that resolves from the root install; the standalone one does not. + assert!(plan.targets.contains_key("//pkg-a:test"), "{plan:#?}"); + assert!(plan.targets.contains_key("//pkg-b:test"), "{plan:#?}"); + assert!( + plan.notes("//pkg-a:test") + .contains("pnpm-workspace.yaml changed; running in full"), + "{plan:#?}" + ); + assert!(plan.targets.contains_key("//wrapper:test"), "{plan:#?}"); + assert!(!plan.targets.contains_key("//standalone:test"), "{plan:#?}"); + + // Action 2: only the package list changes, by adding a package. + ws.git(&["checkout", "-q", "pnpm-workspace.yaml"]); + ws.write( + "pnpm-workspace.yaml", + "packages:\n - 'pkg-*'\n - 'extra'\n", + ); + ws.write( + "extra/package.json", + r#"{"name":"extra","scripts":{"test":"vitest run"}}"#, + ); + let plan = ws.plan("test"); + // Outcome: the new member runs; members whose membership did not change + // keep their selection (the lockfile decides whether they install anew). + assert!(plan.targets.contains_key("//extra:test"), "{plan:#?}"); + assert!(!plan.targets.contains_key("//pkg-a:test"), "{plan:#?}"); + assert!(!plan.targets.contains_key("//pkg-b:test"), "{plan:#?}"); + assert!(!plan.targets.contains_key("//wrapper:test"), "{plan:#?}"); + assert!(!plan.targets.contains_key("//standalone:test"), "{plan:#?}"); + + // Fail closed: a workspace file this parser finds no list in could list + // anything, so every project runs. + ws.write("pnpm-workspace.yaml", "# emptied\n"); + let plan = ws.plan("test"); + assert!(plan.targets.contains_key("//pkg-a:test"), "{plan:#?}"); + assert!(plan.targets.contains_key("//wrapper:test"), "{plan:#?}"); +} + /// Bumping a package that one workspace project resolves must run that /// project and what depends on it, not the whole workspace. #[test]