diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 9c1c55bd..761ed2fc 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -43,6 +43,11 @@ pub struct Scheduler { /// Owned resource guards, keyed by the node that acquired them. Dropped /// (releasing the units) when that node's whole subtree is terminal. owned: HashMap>, + /// The single re-entrancy slot per `(ancestor-holder, resource)`: the id of + /// the descendant currently *borrowing* that ancestor's lock. Present ⇒ the + /// slot is taken, so no other descendant may re-enter the same lock until + /// the borrower's subtree is terminal — "only one node at a time within". + borrow_slots: HashMap<(NodeId, ResourceName), NodeId>, } impl Scheduler { @@ -53,6 +58,7 @@ impl Scheduler { graph, resources: SharedResources::new(resources), owned: HashMap::new(), + borrow_slots: HashMap::new(), } } @@ -92,18 +98,66 @@ impl Scheduler { .collect(); let mut started = Vec::new(); for id in pending { - if !self.node_deps_satisfied(id) { - continue; - } - if let Some(guard) = self.resources.acquire(self.resource_reqs(id)) { - self.graph.set_state(id, State::Running); - self.owned.entry(id).or_default().push(guard); + if self.node_deps_satisfied(id) && self.try_start(id) { started.push(id); } } started } + /// Try to start node `id`: classify each resource dep as *owned* (no + /// ancestor holds it → acquire real units) or *borrowed* (an ancestor group + /// already holds it → re-enter, gated by the one re-entrancy slot), then + /// take everything atomically or nothing. Returns whether it started. + fn try_start(&mut self, id: NodeId) -> bool { + let mut owned_reqs = Vec::new(); + let mut borrows = Vec::new(); + for (name, count) in self.resource_reqs(id) { + if let Some(ancestor) = self.ancestor_owning(id, &name) { + // Re-entrant reuse: allowed only if the slot is free. + if self.borrow_slots.contains_key(&(ancestor, name.clone())) { + return false; + } + borrows.push((ancestor, name)); + } else { + owned_reqs.push((name, count)); + } + } + // Owned units are all-or-nothing; borrow slots were all confirmed free + // above, so this is the only fallible step. Nothing mutated until here. + let Some(guard) = self.resources.acquire(owned_reqs) else { + return false; + }; + self.owned.entry(id).or_default().push(guard); + for slot in borrows { + self.borrow_slots.insert(slot, id); + } + self.graph.set_state(id, State::Running); + true + } + + /// The nearest ancestor of `id` that *owns* (holds real units of) `name`, + /// or `None` if no ancestor holds it (⇒ `id` must own-acquire it itself). + fn ancestor_owning(&self, id: NodeId, name: &ResourceName) -> Option { + let mut cursor = self.graph.node(id)?.parent; + while let Some(ancestor) = cursor { + if self.node_owns(ancestor, name) { + return Some(ancestor); + } + cursor = self.graph.node(ancestor)?.parent; + } + None + } + + /// Whether node `holder` holds an owned guard covering resource `name`. + fn node_owns(&self, holder: NodeId, name: &ResourceName) -> bool { + self.owned.get(&holder).is_some_and(|guards| { + guards + .iter() + .any(|g| g.held().iter().any(|(n, _)| n == name)) + }) + } + /// Report a running node's own execution result. Sets its state, then /// releases the owned guards of every node whose whole subtree has become /// terminal — a parent keeps its lock until its last descendant finishes. @@ -114,10 +168,41 @@ impl Scheduler { Outcome::Failed => State::Failed, }; self.graph.set_state(id, state); + if outcome == Outcome::Failed { + self.cascade_cancel(id); + } self.release_settled_subtrees(); } - /// Drop the owned guards of every holder whose subtree is now terminal. + /// Eagerly cancel the transitive `AfterOk` dependents of a just-failed node: + /// they can never run (a strong dependency failed), so mark them `Cancelled` + /// now — before they could claim resources. A dependent is always still + /// `Pending` here (a `Running` node's `AfterOk` deps were `Done` when it + /// started, and `Done` is terminal), so no resources need releasing. + fn cascade_cancel(&mut self, failed: NodeId) { + let mut stack = vec![failed]; + while let Some(dep) = stack.pop() { + let dependents: Vec = self + .graph + .nodes() + .filter(|n| { + n.state == State::Pending + && n.deps.iter().any( + |d| matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == dep), + ) + }) + .map(|n| n.id) + .collect(); + for d in dependents { + self.graph.set_state(d, State::Cancelled); + stack.push(d); + } + } + } + + /// Release everything whose subtree has become terminal: drop the owned + /// guards of any holder (→ frees its units) and free any re-entrancy slot + /// held by a borrower — both are held for the whole subtree lifetime. fn release_settled_subtrees(&mut self) { let holders: Vec = self.owned.keys().copied().collect(); for holder in holders { @@ -125,6 +210,8 @@ impl Scheduler { self.owned.remove(&holder); // drops guards → releases the units } } + self.borrow_slots + .retain(|_, borrower| !self.graph.group_terminal(*borrower)); } /// Whether every [`Dep::Node`] edge of `id` is satisfied. `Dep::Resource` @@ -195,6 +282,13 @@ mod tests { }] } + fn resource_dep(name: &str) -> Vec { + vec![Dep::Resource { + name: res(name), + count: 1, + }] + } + #[test] fn resource_node_starts_then_releases_on_complete() { let mut s = scheduler_with_slots(1); @@ -246,4 +340,84 @@ mod tests { s.complete(child, Outcome::Done); assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1)); } + + #[test] + fn recursive_lock_serializes_re_entrant_descendants() { + // `agent/foo` is unconfigured → default capacity 1. + let mut s = Scheduler::new(Graph::new(), ResourceTable::new()); + let agent = res("agent/foo"); + // A group node owns agent/foo and runs. + let group = s + .append("group", resource_dep("agent/foo"), None) + .expect("group"); + assert_eq!(s.settle(), vec![group]); + assert!(s.resources.with(|t| t.available(&agent) == 0)); + // Two sub-nodes each need agent/foo → they re-enter the group's lock, + // but only ONE at a time (the single re-entrancy slot). + let c1 = s + .append("c1", resource_dep("agent/foo"), Some(group)) + .expect("c1"); + let c2 = s + .append("c2", resource_dep("agent/foo"), Some(group)) + .expect("c2"); + let started = s.settle(); + assert_eq!( + started, + vec![c1], + "only one descendant may borrow at a time" + ); + assert_eq!(s.graph().node(c2).unwrap().state, State::Pending); + // The lock was NOT re-acquired — still just the group's one unit held. + assert!(s.resources.with(|t| t.available(&agent) == 0)); + // c1 finishes → its borrow slot frees → c2 can now re-enter. + s.complete(c1, Outcome::Done); + assert_eq!(s.settle(), vec![c2]); + assert_eq!(s.graph().node(c2).unwrap().state, State::Running); + // Still no double-acquire; the group's single unit is the only hold. + assert!(s.resources.with(|t| t.available(&agent) == 0)); + } + + #[test] + fn failed_after_ok_dep_cancels_dependents_but_after_any_still_runs() { + let mut s = Scheduler::new(Graph::new(), ResourceTable::new()); + let root = s.append("root", vec![], None).expect("root"); + let strong1 = s + .append( + "strong1", + vec![Dep::Node { + id: root, + when: DepWhen::AfterOk, + }], + None, + ) + .expect("strong1"); + let strong2 = s + .append( + "strong2", + vec![Dep::Node { + id: strong1, + when: DepWhen::AfterOk, + }], + None, + ) + .expect("strong2"); + let weak = s + .append( + "weak", + vec![Dep::Node { + id: root, + when: DepWhen::AfterAny, + }], + None, + ) + .expect("weak"); + assert_eq!(s.settle(), vec![root]); + s.complete(root, Outcome::Failed); + // The AfterOk chain strong1→strong2 is eagerly cancelled (a strong dep + // failed)… + assert_eq!(s.graph().node(strong1).unwrap().state, State::Cancelled); + assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled); + // …but the AfterAny dependent still runs — it converges regardless. + assert_eq!(s.settle(), vec![weak]); + } }