diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index c3c9111a..98138622 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -26,11 +26,12 @@ //! node kind. Resources are held by the acquiring node and released on //! completion via guard objects, recursive within a group. //! -//! The scheduler loop is a follow-up; the resource machinery lives in -//! [`resources`] and the RAII lock guards over it in [`guard`]. +//! The [`scheduler`] settle loop drives execution; the resource machinery +//! lives in [`resources`] and the RAII lock guards over it in [`guard`]. pub mod guard; pub mod resources; +pub mod scheduler; /// Opaque, stable, monotonic node identifier. /// @@ -291,6 +292,24 @@ impl Graph { self.nodes.iter().filter(move |n| n.parent == Some(id)) } + /// Every node in the graph, in insertion order. The scheduler iterates + /// this to find runnable pending nodes. + pub fn nodes(&self) -> impl Iterator> { + self.nodes.iter() + } + + /// Set a node's lifecycle state, returning `false` for an unknown id. The + /// scheduler drives every state transition — nothing else mutates state, + /// which is what keeps the resource guards + terminality in sync. + pub fn set_state(&mut self, id: NodeId, state: State) -> bool { + if let Some(node) = self.nodes.iter_mut().find(|n| n.id == id) { + node.state = state; + true + } else { + false + } + } + /// A group is terminal once the group node itself is terminal *and* every /// node inside it (recursively) is terminal. The node's own state matters: /// a group node still `Pending`/`Running` is not terminal even with no diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs new file mode 100644 index 00000000..9c1c55bd --- /dev/null +++ b/hive-jobq/src/scheduler.rs @@ -0,0 +1,249 @@ +//! The settle loop — drives a [`Graph`] to completion over the resource pool. +//! +//! [`Scheduler::settle`] claims every currently-runnable pending node (its +//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired +//! atomically), marks it `Running`, holds its resource guards, and returns the +//! newly-started ids for the caller's runner to execute. The runner reports each +//! node's result back with [`Scheduler::complete`]; a running node may grow its +//! own sub-group first via [`Scheduler::append`]. Concurrency is emergent from +//! resource capacity — there is no separate active-node cap. +//! +//! A resource is held for the acquiring node's *entire subtree* lifetime: the +//! owned guard is released only when that node and every descendant is terminal +//! (`group_terminal`), not when the node's own work finishes. Single-owner and +//! synchronous — the caller drives `settle` / `complete`; no async or locking +//! lives here (that's the runner's job, one layer up). +//! +//! Recursive-lock re-entrancy (a sub-node reusing an ancestor group's lock) and +//! the eager `AfterOk` failure cascade are layered on top of this owned core. + +use std::collections::HashMap; + +use crate::guard::{ResourceGuard, SharedResources}; +use crate::resources::ResourceTable; +use crate::{Dep, DepWhen, Graph, GraphError, NodeId, ResourceName, State}; + +/// The result of a node's own execution, reported to [`Scheduler::complete`]. +/// +/// `Cancelled` is not an outcome a runner reports — it is scheduler-driven (an +/// `AfterOk` dependency failed), so a runner only ever says `Done` or `Failed`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + /// The node's work succeeded. + Done, + /// The node's work failed. + Failed, +} + +/// Drives a [`Graph`] over a shared resource pool: claim runnable nodes, hold +/// their resources for the subtree's lifetime, release on subtree-terminal. +pub struct Scheduler { + graph: Graph, + resources: SharedResources, + /// Owned resource guards, keyed by the node that acquired them. Dropped + /// (releasing the units) when that node's whole subtree is terminal. + owned: HashMap>, +} + +impl Scheduler { + /// A scheduler over `graph` with `resources` as the capacity pool. + #[must_use] + pub fn new(graph: Graph, resources: ResourceTable) -> Self { + Self { + graph, + resources: SharedResources::new(resources), + owned: HashMap::new(), + } + } + + /// The graph, for inspection (state, hierarchy, UI rendering). + #[must_use] + pub fn graph(&self) -> &Graph { + &self.graph + } + + /// Append a node — e.g. a running node growing its own sub-group. Delegates + /// to [`Graph::insert`]; call [`Scheduler::settle`] afterwards to start it + /// once it is runnable. + /// + /// # Errors + /// Propagates [`GraphError`] for a dangling parent or dependency id. + pub fn append( + &mut self, + payload: N, + deps: Vec, + parent: Option, + ) -> Result { + self.graph.insert(payload, deps, parent) + } + + /// Claim every currently-runnable pending node and start it: node-deps + /// satisfied and all resource-deps acquired atomically (all-or-nothing). + /// Each claimed node is marked `Running`, its owned guards held, and its id + /// returned for the runner to execute. A single pass suffices — a node + /// started here is `Running`, not terminal, so it cannot satisfy another + /// node's dependency in the same pass; it only consumes resources. + pub fn settle(&mut self) -> Vec { + let pending: Vec = self + .graph + .nodes() + .filter(|n| n.state == State::Pending) + .map(|n| n.id) + .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); + started.push(id); + } + } + started + } + + /// 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. + /// Call [`Scheduler::settle`] again afterwards to start newly-unblocked work. + pub fn complete(&mut self, id: NodeId, outcome: Outcome) { + let state = match outcome { + Outcome::Done => State::Done, + Outcome::Failed => State::Failed, + }; + self.graph.set_state(id, state); + self.release_settled_subtrees(); + } + + /// Drop the owned guards of every holder whose subtree is now terminal. + fn release_settled_subtrees(&mut self) { + let holders: Vec = self.owned.keys().copied().collect(); + for holder in holders { + if self.graph.group_terminal(holder) { + self.owned.remove(&holder); // drops guards → releases the units + } + } + } + + /// Whether every [`Dep::Node`] edge of `id` is satisfied. `Dep::Resource` + /// edges are handled by the atomic acquire in [`Scheduler::settle`], not here. + fn node_deps_satisfied(&self, id: NodeId) -> bool { + let Some(node) = self.graph.node(id) else { + return false; + }; + node.deps.iter().all(|dep| match dep { + Dep::Resource { .. } => true, + Dep::Node { id, when } => self.dep_node_satisfied(*id, *when), + }) + } + + /// Whether a node/group dependency `id` satisfies edge kind `when`. A group + /// is depended on as a whole: `AfterAny` needs its subtree terminal (any + /// outcome), `AfterOk` needs its whole subtree to have succeeded. + fn dep_node_satisfied(&self, id: NodeId, when: DepWhen) -> bool { + match when { + DepWhen::AfterAny => self.graph.group_terminal(id), + DepWhen::AfterOk => self.subtree_all_done(id), + } + } + + /// Whether `id` and every descendant reached [`State::Done`] — the success + /// condition for an `AfterOk` edge onto a (possibly group) node. + fn subtree_all_done(&self, id: NodeId) -> bool { + let Some(node) = self.graph.node(id) else { + return false; + }; + node.state == State::Done && self.graph.children(id).all(|c| self.subtree_all_done(c.id)) + } + + /// The `(name, count)` resource units `id` must hold to run. + fn resource_reqs(&self, id: NodeId) -> Vec<(ResourceName, u32)> { + let Some(node) = self.graph.node(id) else { + return Vec::new(); + }; + node.deps + .iter() + .filter_map(|dep| match dep { + Dep::Resource { name, count } => Some((name.clone(), *count)), + Dep::Node { .. } => None, + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn res(name: &str) -> ResourceName { + ResourceName(name.to_owned()) + } + + /// A graph + a resource table with `build-slot` set to `slots`. + fn scheduler_with_slots(slots: u32) -> Scheduler<&'static str> { + let mut table = ResourceTable::new(); + table.set_capacity(res("build-slot"), slots); + Scheduler::new(Graph::new(), table) + } + + fn slot_dep() -> Vec { + vec![Dep::Resource { + name: res("build-slot"), + count: 1, + }] + } + + #[test] + fn resource_node_starts_then_releases_on_complete() { + let mut s = scheduler_with_slots(1); + let n = s.append("build", slot_dep(), None).expect("insert"); + // settle claims it (a slot is free) and marks it Running. + assert_eq!(s.settle(), vec![n]); + assert_eq!(s.graph().node(n).unwrap().state, State::Running); + // Slot is held. + assert!(s.resources.with(|t| t.available(&res("build-slot")) == 0)); + // Completing it releases the slot (subtree is just this node). + s.complete(n, Outcome::Done); + assert_eq!(s.graph().node(n).unwrap().state, State::Done); + assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1)); + } + + #[test] + fn build_slot_cap_limits_concurrency_and_release_unblocks() { + let mut s = scheduler_with_slots(2); + let a = s.append("a", slot_dep(), None).expect("a"); + let b = s.append("b", slot_dep(), None).expect("b"); + let c = s.append("c", slot_dep(), None).expect("c"); + // cap 2 → a + b start, c blocks on the exhausted slot. + let started = s.settle(); + assert_eq!(started, vec![a, b]); + assert_eq!(s.graph().node(c).unwrap().state, State::Pending); + // a finishes → its slot frees → c can now start. + s.complete(a, Outcome::Done); + assert_eq!(s.settle(), vec![c]); + assert_eq!(s.graph().node(c).unwrap().state, State::Running); + } + + #[test] + fn parent_holds_resource_until_child_subtree_done() { + let mut s = scheduler_with_slots(1); + // Parent grabs the single build-slot and runs. + let parent = s.append("parent", slot_dep(), None).expect("parent"); + assert_eq!(s.settle(), vec![parent]); + // Parent grows a child (no resource dep of its own) and finishes its + // OWN work — but its subtree is not terminal, so it keeps the slot. + let child = s.append("child", vec![], Some(parent)).expect("child"); + s.complete(parent, Outcome::Done); + assert!( + s.resources.with(|t| t.available(&res("build-slot")) == 0), + "parent must keep its lock while a child is still pending/running" + ); + // The child starts and completes → now the whole subtree is terminal → + // the parent's slot is released exactly once. + assert_eq!(s.settle(), vec![child]); + s.complete(child, Outcome::Done); + assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1)); + } +}