diff --git a/src/executor/admission.rs b/src/executor/admission.rs index f22f8e7..3efd811 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 0000000..8f7d4d5 --- /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 90c2122..13189de 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 b35afb5..2840b12 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 02f9204..8e2e1ee 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 84135de..bf2bc72 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 39e20b3..f929937 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 ccd5c4f..2d575d8 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 1021746..c66b321 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 26eeb0f..56c3c60 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]