From 5906cc2f2b1f21730a9e1f70dd0964f6b70c43c6 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 20 Jul 2026 19:45:53 +0200 Subject: [PATCH] feat(#2591): hive-jobq parent-axis grouping + borrow + roll-up scheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rework the crate's scheduling model onto an explicit parent (grouping) axis, separate from the dep (ordering) axis. - Node gains a structural `parent: Option`, set by the caller independent of its `Dep::Node` edges. Grouping is not ordering. A `Dep::Node` edge must stay inside the depender's own parent group (validated) — never crossing to another group or onto the parent. - Resource holding walks the parent tree: acquire fresh when no ancestor holds it (the acquirer owns it, held for its whole subtree); borrow an ancestor's grant (one branch at a time; nodes inside are covered); take an extra unit when the grant is lent to a sibling branch, else wait. A grant releases only once the owner and its whole subtree are terminal. - Completion rolls up the parent tree: a node's sub-nodes run after its own logic, and it is not terminal until they finish — it parks in `State::Finishing`, rolling up to Done (every child Done) or Failed (any child Failed/Cancelled). A child is gated on its parent reaching Finishing; a downstream dep on a node therefore waits for that node's dynamically-appended children with no explicit edge. A failed node cancels its pending sub-nodes. Deletes the SharedResources/ResourceGuard layer (guard.rs) and the add_dep graph-growth hook (no longer needed). The scheduler stays single-threaded, owning the ResourceTable directly. Early release of a grant once no subtree node still needs it is a deferred optimization (unsafe under dynamically-appended subnodes, #2611). Base for the hive-c0re job_queue port (#2605), split out so that PR can rebase onto it. --- hive-jobq/src/guard.rs | 130 ------- hive-jobq/src/lib.rs | 343 ++++++++++------- hive-jobq/src/scheduler.rs | 748 ++++++++++++++++++++++++++----------- 3 files changed, 737 insertions(+), 484 deletions(-) delete mode 100644 hive-jobq/src/guard.rs diff --git a/hive-jobq/src/guard.rs b/hive-jobq/src/guard.rs deleted file mode 100644 index fb150625..00000000 --- a/hive-jobq/src/guard.rs +++ /dev/null @@ -1,130 +0,0 @@ -//! RAII guard objects over [`ResourceTable`] — owning resource grants. -//! -//! A running node acquires its resources through [`SharedResources::acquire`], -//! which hands back a [`ResourceGuard`] owning those units. Dropping the guard -//! releases exactly what it acquired, so a node's resources are freed when its -//! grant goes out of scope — there is no explicit release call to forget. -//! -//! Re-entrancy (a sub-node reusing a resource its ancestor group already holds) -//! is not expressed here: the scheduler tracks it with a single borrow slot per -//! `(holder, resource)` and never re-acquires, so these guards are always owning. -//! -//! Single-owner by design: the scheduler drives one settle loop, so the shared -//! table is `Rc>` (single-threaded interior mutability), not -//! `Arc>` — there is no cross-thread contention to guard against. - -use std::cell::RefCell; -use std::rc::Rc; - -use std::hash::Hash; - -use crate::resources::ResourceTable; - -/// A [`ResourceTable`] shared between the scheduler and the live guards that -/// release back into it on drop. Cheap to clone — an `Rc` refcount bump. -#[derive(Debug, Clone)] -pub struct SharedResources(Rc>>); - -impl Default for SharedResources { - fn default() -> Self { - Self::new(ResourceTable::new()) - } -} - -impl SharedResources { - /// Wrap an existing table so guards can release into it. - #[must_use] - pub fn new(table: ResourceTable) -> Self { - Self(Rc::new(RefCell::new(table))) - } - - /// Atomically acquire every requested `(name, count)` or none of them. - /// - /// Returns an owning [`ResourceGuard`] (releases on drop) when the whole - /// request fits in what is available right now; returns `None` and leaves - /// the table completely untouched otherwise. Duplicate names are summed and - /// an over-capacity request can never succeed — same all-or-nothing - /// semantics as [`ResourceTable::try_acquire_all`]. - #[must_use] - pub fn acquire(&self, reqs: Vec<(R, u32)>) -> Option> { - if self.0.borrow_mut().try_acquire_all(&reqs) { - Some(ResourceGuard { - table: self.clone(), - reqs, - }) - } else { - None - } - } - - /// Observe the underlying table — test-only (the scheduler's tests assert - /// on resource availability). Gated `#[cfg(test)]` so it is compiled out of - /// the shipped crate: no consumer can reach the raw table through it. - #[cfg(test)] - pub fn with(&self, f: impl FnOnce(&ResourceTable) -> T) -> T { - f(&self.0.borrow()) - } -} - -/// An RAII grant of resources: dropping it releases exactly the units it -/// acquired back into the shared table. -#[derive(Debug)] -pub struct ResourceGuard { - table: SharedResources, - reqs: Vec<(R, u32)>, -} - -impl ResourceGuard { - /// The `(name, count)` units this guard releases on drop. - #[must_use] - pub fn held(&self) -> &[(R, u32)] { - &self.reqs - } -} - -impl Drop for ResourceGuard { - fn drop(&mut self) { - self.table.0.borrow_mut().release_all(&self.reqs); - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn res(name: &str) -> String { - name.to_owned() - } - - fn shared_with(slots: u32) -> SharedResources { - let mut t = ResourceTable::new(); - t.set_capacity(res("build-slot"), slots); - SharedResources::new(t) - } - - #[test] - fn owning_guard_releases_on_drop() { - let sr = shared_with(2); - let slot = res("build-slot"); - { - let g = sr.acquire(vec![(slot.clone(), 2)]).expect("fits"); - assert_eq!(g.held(), &[(slot.clone(), 2)]); - // Both units held → any further acquire fails. - assert!(sr.acquire(vec![(slot.clone(), 1)]).is_none()); - } // guard dropped here → its units are released - // Full capacity is available again. - assert!(sr.acquire(vec![(slot.clone(), 2)]).is_some()); - } - - #[test] - fn acquire_returns_none_and_leaves_table_untouched_when_it_does_not_fit() { - let sr = shared_with(1); - let slot = res("build-slot"); - let held = sr.acquire(vec![(slot.clone(), 1)]).expect("first fits"); - assert!(sr.acquire(vec![(slot.clone(), 1)]).is_none()); - // The failed acquire took nothing extra: dropping the one real grant - // frees exactly one unit, so a single-unit acquire then fits. - drop(held); - assert!(sr.acquire(vec![(slot.clone(), 1)]).is_some()); - } -} diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index 29970de4..c4c05985 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -4,42 +4,35 @@ //! # Model (v2) //! //! One **persistent graph** for the whole system, not a DAG per job. Enqueuing -//! inserts a self-contained **node group** and returns its id; the scheduler -//! runs a continuous loop, starting every node whose [`Dep`]s are satisfied: +//! inserts a self-contained sub-DAG of nodes and returns their ids; the +//! scheduler runs a continuous loop, starting every node whose [`Dep`]s are +//! satisfied: //! //! - **Resource** deps are named counting semaphores over a caller-chosen -//! type `R` (a `String` or an enum): `build-slot` (cap N), `agent/` -//! (cap 1), or any name (cap 1, created on use). A node acquires *all* its -//! resource deps atomically at start (all-or-nothing) — no hold-and-wait, -//! so no deadlock and no cycle detection needed. -//! - **Node** deps wait on a node/group per [`DepWhen`]: `AfterOk` needs +//! type `R`: `build-slot` (cap N), `agent/` (cap 1), or any name +//! (cap 1, created on use). A node acquires *all* its resource deps +//! atomically at start (all-or-nothing) — no hold-and-wait, no deadlock. +//! - **Node** deps wait on another node per [`DepWhen`]: `AfterOk` needs //! success (a failed dep cancels the dependent), `AfterAny` only terminal. //! -//! A **node group** is a self-contained sub-graph; things depend on it as a -//! whole (done = every inner node terminal), never on an inner node. Groups -//! nest; a running node may grow its own group but not reach outside it. +//! A node carries two independent axes: its [`Dep`]s (ordering + resource +//! needs) and its [`Node::parent`] (structural grouping) — the parent chain, +//! not the [`Dep::Node`] edges, is what the [`scheduler`] consults for resource +//! re-entrancy. A [`NodeId`] is opaque, stable, and monotonic (persisted). The +//! payload `N` is generic so the library stays container-agnostic. //! -//! A [`NodeId`] is opaque, stable, and monotonic — persisted, so it survives -//! restarts. Group membership is a parent edge ([`Node::parent`]), *not* in the -//! id; the `1/1/2` hierarchy is a derived UI label. The node payload is generic -//! (`N`) so the library stays container-agnostic — the caller supplies its own -//! node kind. Resources are held by the acquiring node and released on -//! completion via guard objects, recursive within a group. -//! -//! The [`scheduler`] settle loop drives execution; the resource machinery -//! lives in [`resources`] and the RAII lock guards over it in `guard`. +//! A resource unit is held for the acquiring node + its whole [`Node::parent`] +//! subtree; a node needing a resource an ancestor holds re-uses that grant (a +//! re-entrant borrow, one branch at a time). Single-threaded — the scheduler +//! owns the resource table and mutates it directly. See [`scheduler`]. -pub(crate) mod guard; pub mod resources; pub mod scheduler; /// Opaque, stable, monotonic node identifier. /// /// Assigned by the [`Graph`] on insert and persisted, so it is stable across -/// restarts. Group membership is a separate parent edge ([`Node::parent`]) — it -/// is deliberately *not* encoded in the id, so the id never changes as the tree -/// grows or collapses. The hierarchical `1/1/2` path used in the UI is derived -/// from the parent tree at render time. +/// restarts. /// /// The inner field is crate-private: an id can only originate from the graph's /// monotonic counter (or deserialization of a persisted graph), never be @@ -49,6 +42,18 @@ pub mod scheduler; )] pub struct NodeId(pub(crate) u64); +impl NodeId { + /// The underlying monotonic value, for carrying this id across a boundary + /// that cannot hold the opaque `NodeId` type — e.g. serializing it onto a + /// wire protocol. The inverse (fabricating a `NodeId` from a raw value) + /// stays impossible by construction: an id only ever originates from the + /// graph's counter, which is what makes it opaque. + #[must_use] + pub fn get(self) -> u64 { + self.0 + } +} + /// When a [`Dep::Node`] edge is satisfied — the strong/weak distinction the /// current queue carries as `DepWhen`, load-bearing for failure safety. #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -80,19 +85,22 @@ impl DepWhen { /// it names can be acquired (all of them, atomically). #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum Dep { - /// Depend on another node (or a group, by its group node's id). Whether a - /// *failed* dependency satisfies the edge is decided by `when`: `AfterOk` - /// requires success (and cancels this node if the dep fails), `AfterAny` - /// only requires the dep to be terminal. + /// Depend on another node. Whether a *failed* dependency satisfies the edge + /// is decided by `when`: `AfterOk` requires success (and cancels this node + /// if the dep fails), `AfterAny` only requires the dep to be terminal. Node { - /// The node (or group) depended on. + /// The node depended on. id: NodeId, /// Strong (`AfterOk`) vs weak (`AfterAny`). when: DepWhen, }, - /// Hold `count` units of a named resource for the duration of this node's - /// run. Acquired atomically with the node's other resource deps at start, - /// released when the node completes. + /// Need `count` units of a named resource to run. Declared on every node + /// that needs it, even when a [`Node::parent`]-ancestor already holds it. + /// Acquired atomically with the node's other resource deps at start; the + /// acquired unit is held for the acquirer's whole subtree (released only + /// once the acquirer and all its sub-nodes are terminal). A node whose + /// parent-ancestor already holds this resource re-uses that grant (a + /// re-entrant borrow) instead of taking a fresh unit. Resource { /// The resource to acquire. name: R, @@ -106,11 +114,17 @@ pub enum Dep { pub enum State { /// Waiting on dependencies (node or resource). Pending, - /// Dependencies satisfied, resources held, currently executing. + /// Dependencies satisfied, resources held, currently executing its own logic. Running, - /// Completed successfully. + /// Own logic finished successfully, but the node is *not yet terminal*: it + /// waits here until all its sub-nodes ([`Node::parent`] children) are + /// terminal, then rolls up to [`State::Done`] (every child `Done`) or + /// [`State::Failed`] (any child `Failed`/`Cancelled`). A node with no + /// children never rests here — it goes straight to a terminal state. + Finishing, + /// Completed successfully — own logic done *and* every sub-node `Done`. Done, - /// Completed unsuccessfully. + /// Completed unsuccessfully — own logic failed, or a sub-node did. Failed, /// Never ran: an `AfterOk` dependency failed, so this node (and the rest of /// its strong-dependent chain) is cancelled rather than run. @@ -131,13 +145,19 @@ impl State { /// /// The library schedules over `Node`s and resources without interpreting the /// payload; the caller supplies `N` (its own node kind) and a runner to execute -/// a claimed node. +/// a claimed node. A node carries two independent axes: its [`Dep`]s (ordering + +/// resource needs) and its [`Node::parent`] (structural grouping), both set by +/// the caller/submit layer. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct Node { /// Stable identity, assigned on insert. pub id: NodeId, - /// The group this node belongs to, if any. `None` for a top-level group - /// node. Group membership lives here, not in the id. + /// Structural grouping: the node this one is a sub-node of, or `None` for a + /// group root. Independent of [`Node::deps`] — grouping is *not* ordering. + /// The [`scheduler`] uses the parent chain to decide resource re-entrancy: a + /// node needing a resource a parent-ancestor holds re-uses that grant rather + /// than acquiring a fresh unit, and a held unit stays reserved for the + /// acquirer's whole subtree. A node's sub-nodes run *after* its own logic. pub parent: Option, /// Caller-defined payload (the node's kind / work description). pub payload: N, @@ -158,9 +178,20 @@ pub enum GraphError { /// A node's dependency named an id not present in the graph. #[error("dependency references unknown node {0:?}")] UnknownDep(NodeId), - /// A node's parent named an id not present in the graph. + /// A node's `parent` named an id not present in the graph. #[error("parent references unknown node {0:?}")] UnknownParent(NodeId), + /// A node's [`Dep::Node`] edge points outside its own parent group — the + /// target must be a proper descendant of the depender's `parent` (a sibling + /// or a sibling's sub-node), never the parent itself or a node in another + /// group. Top-level nodes may only depend on top-level nodes. + #[error("dependency {dep:?} is outside the depender's parent group {parent:?}")] + DepOutsideParent { + /// The out-of-group dependency target. + dep: NodeId, + /// The depender's parent (the group the target had to be inside). + parent: Option, + }, /// A loaded graph's `next_id` counter is not past the largest existing id, /// so the next minted id would collide with one already in the graph. #[error("next_id {next_id} must exceed the largest existing node id {max_id}")] @@ -174,9 +205,8 @@ pub enum GraphError { /// The single persistent graph of all nodes. /// -/// New jobs are inserted as node groups; the scheduler (added in a follow-up) -/// walks this graph filling open slots. Completed groups are retained (no -/// pruning in v1). +/// New jobs are inserted as sub-DAGs of nodes; the scheduler walks this graph +/// filling open slots. Completed nodes are retained (no pruning in v1). #[derive(Debug, serde::Serialize, serde::Deserialize)] #[serde( try_from = "GraphData", @@ -238,33 +268,37 @@ impl Graph { id } - /// Insert a node with the given payload, deps, and parent group, returning - /// its freshly-minted id. The node starts [`State::Pending`]. + /// Insert a node with the given payload, deps, and `parent`, returning its + /// freshly-minted id. The node starts [`State::Pending`]. /// - /// Every [`Dep::Node`] id and the `parent` id (if any) must already resolve - /// to a node in the graph — an id is only meaningful against the graph that - /// minted it, so a dangling reference is rejected here rather than surfacing - /// as a broken edge later. + /// Every [`Dep::Node`] id and the `parent` id (when `Some`) must already + /// resolve to a node in the graph — an id is only meaningful against the + /// graph that minted it, so a dangling reference is rejected here rather than + /// surfacing as a broken edge later. /// /// # Errors - /// Returns [`GraphError::UnknownParent`] / [`GraphError::UnknownDep`] if the - /// parent or a dependency references a node not in the graph. + /// Returns [`GraphError::UnknownDep`] / [`GraphError::UnknownParent`] for a + /// dangling dependency or parent id, or [`GraphError::DepOutsideParent`] if a + /// `Dep::Node` edge points outside the node's own parent group. pub fn insert( &mut self, payload: N, deps: Vec>, parent: Option, ) -> Result { - if let Some(parent_id) = parent - && self.node(parent_id).is_none() + if let Some(p) = parent + && self.node(p).is_none() { - return Err(GraphError::UnknownParent(parent_id)); + return Err(GraphError::UnknownParent(p)); } for dep in &deps { - if let Dep::Node { id, .. } = dep - && self.node(*id).is_none() - { - return Err(GraphError::UnknownDep(*id)); + if let Dep::Node { id, .. } = dep { + if self.node(*id).is_none() { + return Err(GraphError::UnknownDep(*id)); + } + if !self.dep_target_in_group(parent, *id) { + return Err(GraphError::DepOutsideParent { dep: *id, parent }); + } } } let id = self.mint_id(); @@ -284,17 +318,39 @@ impl Graph { self.nodes.iter().find(|n| n.id == id) } - /// The direct children of a group node (nodes whose `parent` is `id`). - pub fn children(&self, id: NodeId) -> impl Iterator> { - 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() } + /// Whether `ancestor` lies on `node`'s [`Node::parent`] chain (i.e. `node` is + /// in `ancestor`'s subtree). `node` is not its own ancestor. + fn is_descendant(&self, node: NodeId, ancestor: NodeId) -> bool { + let mut cur = self.node(node).and_then(|n| n.parent); + while let Some(p) = cur { + if p == ancestor { + return true; + } + cur = self.node(p).and_then(|n| n.parent); + } + false + } + + /// Whether a node whose parent is `node_parent` may depend on `target` — the + /// grouping rule: a [`Dep::Node`] edge must stay inside the depender's own + /// parent group. `target` must be a proper descendant of `node_parent` (a + /// sibling or a sibling's sub-node), never the parent itself (which would + /// deadlock: the parent stays [`State::Finishing`] until its children finish, + /// so a child that waited on the parent could never run). Top-level nodes + /// (`parent == None`) may only depend on other top-level nodes. + fn dep_target_in_group(&self, node_parent: Option, target: NodeId) -> bool { + match node_parent { + Some(p) => self.is_descendant(target, p), + None => self.node(target).is_some_and(|n| n.parent.is_none()), + } + } + /// 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. @@ -307,40 +363,32 @@ impl Graph { } } - /// 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 - /// children yet, since a running node may still append some. (Deciding when - /// to *settle* a group node to terminal once its children are all done is a - /// separate concern the scheduler owns.) An unknown id is not terminal. - #[must_use] - pub fn group_terminal(&self, id: NodeId) -> bool { - let Some(node) = self.node(id) else { - return false; - }; - node.state.is_terminal() && self.children(id).all(|child| self.group_terminal(child.id)) - } - - /// Check that every id the graph holds resolves: each node's `parent` and - /// every [`Dep::Node`] id names a node present in the graph, and `next_id` - /// is past the largest existing id. Deserialization runs this, so a loaded - /// graph is internally consistent and internal iteration can trust its ids. + /// Check that every id the graph holds resolves: every [`Dep::Node`] id + /// names a node present in the graph, and `next_id` is past the largest + /// existing id. Deserialization runs this, so a loaded graph is internally + /// consistent and internal iteration can trust its ids. /// /// # Errors - /// Returns [`GraphError`] on a dangling parent / dependency reference, or a - /// `next_id` that would remint an id already in the graph. + /// Returns [`GraphError`] on a dangling dependency reference, or a `next_id` + /// that would remint an id already in the graph. pub fn validate(&self) -> Result<(), GraphError> { for node in &self.nodes { - if let Some(parent_id) = node.parent - && self.node(parent_id).is_none() + if let Some(p) = node.parent + && self.node(p).is_none() { - return Err(GraphError::UnknownParent(parent_id)); + return Err(GraphError::UnknownParent(p)); } for dep in &node.deps { - if let Dep::Node { id, .. } = dep - && self.node(*id).is_none() - { - return Err(GraphError::UnknownDep(*id)); + if let Dep::Node { id, .. } = dep { + if self.node(*id).is_none() { + return Err(GraphError::UnknownDep(*id)); + } + if !self.dep_target_in_group(node.parent, *id) { + return Err(GraphError::DepOutsideParent { + dep: *id, + parent: node.parent, + }); + } } } } @@ -371,43 +419,16 @@ mod tests { id: a, when: DepWhen::AfterOk, }], - Some(a), + None, ) .unwrap(); assert_eq!(a, NodeId(0)); assert_eq!(b, NodeId(1)); - // Membership is the parent edge, not the id. - assert_eq!(g.node(b).unwrap().parent, Some(a)); - assert_eq!(g.node(a).unwrap().parent, None); - } - - #[test] - fn group_terminal_requires_the_group_node_and_all_children_terminal() { - let mut g: Graph<&str, String> = Graph::new(); - let group = g.insert("group", vec![], None).unwrap(); - let child = g.insert("child", vec![], Some(group)).unwrap(); - // Both pending → not terminal. - assert!(!g.group_terminal(group)); - // Child done, but the group node itself is still pending → NOT terminal: - // the group node's own state is load-bearing, not just its children. - g.set_state(child, State::Done); - assert!(!g.group_terminal(group)); - // Group node terminal too → the whole group is terminal. - g.set_state(group, State::Done); - assert!(g.group_terminal(group)); - } - - #[test] - fn empty_running_group_is_not_terminal() { - // A running node with no children yet may still append some, so it must - // not read as terminal just because its child set is currently empty. - let mut g: Graph<&str, String> = Graph::new(); - let group = g.insert("group", vec![], None).unwrap(); - g.set_state(group, State::Running); - assert!(!g.group_terminal(group)); - // Once it finishes (having grown no children), it is terminal. - g.set_state(group, State::Done); - assert!(g.group_terminal(group)); + // The dep edge references the earlier node; ids are stable + monotonic. + assert!(matches!( + g.node(b).unwrap().deps.first(), + Some(Dep::Node { id, .. }) if *id == a + )); } #[test] @@ -417,6 +438,8 @@ mod tests { assert!(State::Cancelled.is_terminal()); assert!(!State::Pending.is_terminal()); assert!(!State::Running.is_terminal()); + // Finishing (logic done, children still running) is NOT terminal. + assert!(!State::Finishing.is_terminal()); } #[test] @@ -432,16 +455,10 @@ mod tests { assert!(DepWhen::AfterAny.satisfied_by(State::Failed)); assert!(DepWhen::AfterAny.satisfied_by(State::Cancelled)); assert!(!DepWhen::AfterAny.satisfied_by(State::Pending)); - } - - #[test] - fn insert_rejects_unknown_parent() { - let mut g: Graph<&str, String> = Graph::new(); - let bogus = NodeId(7); - assert_eq!( - g.insert("x", vec![], Some(bogus)).unwrap_err(), - GraphError::UnknownParent(bogus) - ); + // Finishing satisfies neither — a dependent waits until the node rolls + // up to a terminal state (all its sub-nodes done). + assert!(!DepWhen::AfterOk.satisfied_by(State::Finishing)); + assert!(!DepWhen::AfterAny.satisfied_by(State::Finishing)); } #[test] @@ -458,9 +475,25 @@ mod tests { ); } + #[test] + fn insert_rejects_unknown_parent() { + let mut g: Graph<&str, String> = Graph::new(); + let bogus = NodeId(7); + assert_eq!( + g.insert("x", vec![], Some(bogus)).unwrap_err(), + GraphError::UnknownParent(bogus) + ); + // A resolvable parent is accepted and recorded. + let a = g.insert("a", vec![], None).unwrap(); + let b = g.insert("b", vec![], Some(a)).unwrap(); + assert_eq!(g.node(b).unwrap().parent, Some(a)); + } + #[test] fn valid_graph_round_trips_through_serde() { let mut g: Graph = Graph::new(); + // `a` (top-level) and `b` (top-level, depends on its sibling `a`), plus + // `c` — a sub-node of `a` (grouping, no dep on its parent). let a = g.insert("a".to_owned(), vec![], None).unwrap(); g.insert( "b".to_owned(), @@ -468,13 +501,57 @@ mod tests { id: a, when: DepWhen::AfterAny, }], - Some(a), + None, ) .unwrap(); + let c = g.insert("c".to_owned(), vec![], Some(a)).unwrap(); let json = serde_json::to_string(&g).unwrap(); let back: Graph = serde_json::from_str(&json).unwrap(); assert!(back.validate().is_ok()); assert_eq!(back.node(a).unwrap().payload, "a"); + assert_eq!(back.node(c).unwrap().parent, Some(a)); + } + + #[test] + fn insert_rejects_dep_on_parent_and_cross_group() { + let mut g: Graph<&str, String> = Graph::new(); + let root = g.insert("root", vec![], None).unwrap(); + // A child cannot depend on its own parent (would deadlock under the + // roll-up model — the parent stays `Finishing` awaiting its children). + let on_parent = vec![Dep::Node { + id: root, + when: DepWhen::AfterOk, + }]; + assert_eq!( + g.insert("child", on_parent, Some(root)).unwrap_err(), + GraphError::DepOutsideParent { + dep: root, + parent: Some(root), + } + ); + // A sibling dep IS allowed: two children of `root`, the second on the first. + let c1 = g.insert("c1", vec![], Some(root)).unwrap(); + let c2 = g + .insert("c2", vec![after_ok_dep(c1)], Some(root)) + .expect("sibling dep is in-group"); + assert_eq!(g.node(c2).unwrap().parent, Some(root)); + // But a node in another group cannot be depended on across the boundary. + let other = g.insert("other", vec![], None).unwrap(); + assert_eq!( + g.insert("x", vec![after_ok_dep(other)], Some(root)) + .unwrap_err(), + GraphError::DepOutsideParent { + dep: other, + parent: Some(root), + } + ); + } + + fn after_ok_dep(on: NodeId) -> Dep { + Dep::Node { + id: on, + when: DepWhen::AfterOk, + } } #[test] diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 0fe62cd4..98dfd6b3 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -1,26 +1,36 @@ -//! The settle loop — drives a [`Graph`] to completion over the resource pool. +//! The settle loop — drives a [`Graph`] to completion over a resource pool the +//! scheduler owns directly. //! //! [`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 +//! atomically), marks it `Running`, records the units it holds, 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. +//! node's result back with [`Scheduler::complete`]; a running node may grow more +//! work 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). +//! Single-threaded by design: the scheduler is the only driver, holds the +//! [`ResourceTable`] as a plain owned field, mutating it through `&mut self` — +//! no interior mutability, no guard objects. //! -//! 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. +//! Completion rolls up the parent tree: a node with children parks in +//! [`State::Finishing`] until they finish; a child is gated on its parent there. +//! +//! Resource holding follows the [`crate::Node::parent`] tree, not the deps. +//! For each resource a node needs, the scheduler walks its parent-ancestors: +//! no ancestor holds it → acquire fresh units (this node *owns* them, held for +//! its whole subtree); an ancestor owns it but its grant is lent to a different +//! branch → acquire an additional unit if one is free, else wait; an ancestor +//! owns it and the grant is free (or lent to a branch this node is inside) → +//! *borrow* it, no new unit. A grant is lent to one branch at a time; nodes +//! inside a branch are covered by its borrow (dep-sequenced, so no concurrent +//! work under a cap-1 lease). An owner's unit releases only once the owner and +//! its whole subtree are terminal. (Early release once no subtree node needs it +//! is a deferred optimization — unsafe under dynamically-appended subnodes.) use std::collections::HashMap; use std::hash::Hash; -use crate::guard::{ResourceGuard, SharedResources}; use crate::resources::ResourceTable; use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State}; @@ -36,19 +46,22 @@ pub enum Outcome { Failed, } -/// Drives a [`Graph`] over a shared resource pool: claim runnable nodes, hold -/// their resources for the subtree's lifetime, release on subtree-terminal. +/// Drives a [`Graph`] over an owned resource pool: claim runnable nodes, record +/// the units each *owns*, track which node is currently *borrowing* each grant, +/// and release an owner's grant once its whole subtree is 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>>, - /// 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, R), NodeId>, + resources: ResourceTable, + /// Fresh units each owner node acquired: `owner → [(resource, count)]`. + /// Recorded against the node that *acquired* the units (never a borrower); + /// released back to the table once the owner and its whole [`Node::parent`] + /// subtree are terminal. + owned: HashMap>, + /// Which branch currently borrows a given owner's grant: `(owner, resource) + /// → branch-root node`. A grant is lent to one branch at a time; nodes + /// inside that branch are covered by it. Cleared when the branch leaves + /// (its subtree terminal), freeing the grant for a waiting sibling. + borrowed: HashMap<(NodeId, R), NodeId>, } impl Scheduler { @@ -57,9 +70,9 @@ impl Scheduler { pub fn new(graph: Graph, resources: ResourceTable) -> Self { Self { graph, - resources: SharedResources::new(resources), + resources, owned: HashMap::new(), - borrow_slots: HashMap::new(), + borrowed: HashMap::new(), } } @@ -69,12 +82,12 @@ impl Scheduler { &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. + /// Append a node under `parent` — e.g. a running node growing more work into + /// its own subtree. 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. + /// Propagates [`GraphError`] for a dangling dependency or parent id. pub fn append( &mut self, payload: N, @@ -86,9 +99,9 @@ impl Scheduler { /// 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 + /// Each claimed node is marked `Running`, its acquired units recorded, 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. #[must_use] pub fn settle(&mut self) -> Vec { @@ -107,163 +120,288 @@ impl Scheduler { 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. + /// Try to start node `id`. For each resource it needs, decide per the parent + /// tree (see the module docs): acquire fresh units (owner), acquire an extra + /// unit (grant lent elsewhere), or borrow an ancestor's grant. The fresh set + /// is taken atomically (all-or-nothing via [`ResourceTable::try_acquire_all`]) + /// and recorded as owned; borrows are recorded only once the fresh set is + /// secured. If the fresh set doesn't fit, nothing is taken, no borrow is + /// recorded, and the node stays pending. Returns whether it started. fn try_start(&mut self, id: NodeId) -> bool { - let mut owned_reqs = Vec::new(); - let mut borrows = Vec::new(); + let mut to_acquire: Vec<(R, u32)> = Vec::new(); + let mut to_borrow: Vec<(NodeId, R)> = 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)); + match self.parent_ancestor_owning(id, &name) { + // Case 1: no ancestor holds it → this node acquires + owns it. + None => to_acquire.push((name, count)), + Some(owner) => match self.borrowed.get(&(owner, name.clone())).copied() { + // Case 3: the grant is free → borrow it, no new unit. + None => to_borrow.push((owner, name)), + // Covered: already lent to a branch this node is inside. + Some(branch) if self.parent_chain_contains(id, branch) => {} + // Case 2: lent to a different branch → take an extra unit. + Some(_) => to_acquire.push((name, count)), + }, } } - // Owned units (if any) are all-or-nothing; borrow slots were all - // confirmed free above, so acquiring them is the only fallible step. - // Nothing is mutated until here. Skip the acquire + guard entirely when - // every dep was re-entrant (no owned units): an empty guard would just - // be a no-op `Drop` plus a wasted `owned` entry. - if !owned_reqs.is_empty() { - let Some(guard) = self.resources.acquire(owned_reqs) else { - return false; - }; - self.owned.entry(id).or_default().push(guard); + if !to_acquire.is_empty() && !self.resources.try_acquire_all(&to_acquire) { + return false; } - for slot in borrows { - self.borrow_slots.insert(slot, id); + if !to_acquire.is_empty() { + self.owned.entry(id).or_default().extend(to_acquire); + } + for (owner, name) in to_borrow { + self.borrowed.insert((owner, name), 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: &R) -> Option { - let mut cursor = self.graph.node(id)?.parent; - while let Some(ancestor) = cursor { - if self.node_owns(ancestor, name) { - return Some(ancestor); + /// The nearest [`Node::parent`] ancestor of `id` that *owns* (holds real + /// units of) `name`, or `None` if none does (⇒ `id` must acquire it fresh). + fn parent_ancestor_owning(&self, id: NodeId, name: &R) -> Option { + let mut cur = self.graph.node(id).and_then(|n| n.parent); + while let Some(p) = cur { + if self.node_owns(p, name) { + return Some(p); } - cursor = self.graph.node(ancestor)?.parent; + cur = self.graph.node(p).and_then(|n| n.parent); } None } - /// Whether node `holder` holds an owned guard covering resource `name`. - fn node_owns(&self, holder: NodeId, name: &R) -> bool { - self.owned.get(&holder).is_some_and(|guards| { - guards - .iter() - .any(|g| g.held().iter().any(|(n, _)| n == name)) - }) + /// Whether `ancestor` lies on `id`'s [`Node::parent`] chain (i.e. `id` is in + /// `ancestor`'s subtree). `id` itself does not count as its own ancestor. + fn parent_chain_contains(&self, id: NodeId, ancestor: NodeId) -> bool { + let mut cur = self.graph.node(id).and_then(|n| n.parent); + while let Some(p) = cur { + if p == ancestor { + return true; + } + cur = self.graph.node(p).and_then(|n| n.parent); + } + false } - /// 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. + /// Whether node `holder` holds real units of resource `name`. + fn node_owns(&self, holder: NodeId, name: &R) -> bool { + self.owned + .get(&holder) + .is_some_and(|units| units.iter().any(|(n, _)| n == name)) + } + + /// Whether `root` and every node in its [`Node::parent`] subtree are + /// terminal — the condition for releasing `root`'s owned grants (and for + /// giving back a borrow whose branch-root is `root`). + fn subtree_terminal(&self, root: NodeId) -> bool { + self.graph.node(root).is_none_or(|n| n.state.is_terminal()) + && !self + .graph + .nodes() + .any(|n| !n.state.is_terminal() && self.parent_chain_contains(n.id, root)) + } + + /// Report a running node's own logic result. On success the node is *not* + /// terminal until its sub-nodes ([`Node::parent`] children) all finish — it + /// rests in [`State::Finishing`] until then, rolling up to [`State::Done`] + /// (every child `Done`) or [`State::Failed`] (any child `Failed`/`Cancelled`). + /// On failure it is `Failed` at once and its pending sub-nodes are cancelled + /// (gated on a `Finishing` the parent never reached). Terminality then + /// propagates up the parent chain. 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, + match outcome { + Outcome::Failed => { + self.graph.set_state(id, State::Failed); + self.cascade_cancel(id); + } + Outcome::Done => self.settle_terminal(id), + } + self.roll_up_ancestors(id); + self.release_ready(); + } + + /// Whether every direct child of `id` is terminal. + fn all_children_terminal(&self, id: NodeId) -> bool { + self.graph + .nodes() + .all(|n| n.parent != Some(id) || n.state.is_terminal()) + } + + /// Whether any direct child of `id` ended `Failed`/`Cancelled` — the roll-up + /// failure condition for the parent. + fn any_child_failed(&self, id: NodeId) -> bool { + self.graph + .nodes() + .any(|n| n.parent == Some(id) && matches!(n.state, State::Failed | State::Cancelled)) + } + + /// Transition a node whose own logic just *succeeded* to its resulting state: + /// [`State::Finishing`] while any child is still non-terminal, else `Failed` + /// if a child failed, else `Done`. A node with no children skips `Finishing`. + fn settle_terminal(&mut self, id: NodeId) { + let state = if !self.all_children_terminal(id) { + State::Finishing + } else if self.any_child_failed(id) { + State::Failed + } else { + State::Done }; self.graph.set_state(id, state); - if outcome == Outcome::Failed { + if state == State::Failed { self.cascade_cancel(id); } - self.release_settled_subtrees(); } - /// 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 + /// After `start` became terminal, roll up every ancestor that was parked in + /// `Finishing` awaiting its children: once all of an ancestor's children are + /// terminal it transitions (Done / Failed), which may let *its* parent roll + /// up too, and so on up the [`Node::parent`] chain. + fn roll_up_ancestors(&mut self, start: NodeId) { + let mut cur = self.graph.node(start).and_then(|n| n.parent); + while let Some(a) = cur { + if self.graph.node(a).map(|n| n.state) != Some(State::Finishing) + || !self.all_children_terminal(a) + { + break; + } + let state = if self.any_child_failed(a) { + State::Failed + } else { + State::Done + }; + self.graph.set_state(a, state); + if state == State::Failed { + self.cascade_cancel(a); + } + cur = self.graph.node(a).and_then(|n| n.parent); + } + } + + /// Cancel a still-*pending* node (and cascade to its `AfterOk` dependents): + /// mark it [`State::Cancelled`] and report whether it was cancellable. A + /// node that has already started (`Running`) or finished is left untouched — + /// an in-flight node's work is not interruptible. A pending node holds no + /// resources, so nothing is released here; call [`Scheduler::settle`] + /// afterwards to let now-terminal dependents advance (e.g. a weak-edge + /// terminal node observing the cancellation). + pub fn cancel_node(&mut self, id: NodeId) -> bool { + if self + .graph + .node(id) + .is_some_and(|n| n.state == State::Pending) + { + self.graph.set_state(id, State::Cancelled); + self.cascade_cancel(id); + true + } else { + false + } + } + + /// Snapshot the currently-held grants as `(resource, owner)` pairs — one + /// entry per resource each owning node holds. Lets a caller render live + /// ownership (which node holds a given resource) as a pull query, instead of + /// threading release events out of the scheduler. + #[must_use] + pub fn resource_state(&self) -> Vec<(R, NodeId)> { + self.owned + .iter() + .flat_map(|(&holder, units)| units.iter().map(move |(name, _)| (name.clone(), holder))) + .collect() + } + + /// Propagate cancellation out from a just-failed/cancelled `origin`: every + /// still-`Pending` node that can no longer run gets marked `Cancelled`, + /// transitively. Two edges carry it: (a) an `AfterOk` dep on a cancelled node + /// (a strong dependency failed), and (b) being a *child* of one (its parent + /// will never reach `Finishing`, so it was gated from ever starting — and + /// leaving it pending would wedge the subtree non-terminal). Cancelled nodes + /// were `Pending`, so they hold no resources. + fn cascade_cancel(&mut self, origin: NodeId) { + let mut stack = vec![origin]; + while let Some(cur) = stack.pop() { + let doomed: 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), - ) + && (n.parent == Some(cur) + || n.deps.iter().any(|d| { + matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == cur) + })) }) .map(|n| n.id) .collect(); - for d in dependents { + for d in doomed { 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 { - if self.graph.group_terminal(holder) { - self.owned.remove(&holder); // drops guards → releases the units + /// Give back any borrow whose branch has fully left (freeing the grant for a + /// waiting sibling), then release every owner's grant whose whole subtree is + /// terminal (dropping the units back into the table). + fn release_ready(&mut self) { + // 1. Return borrows whose branch-root subtree is now terminal. + let returned: Vec<(NodeId, R)> = self + .borrowed + .iter() + .filter(|&(_, &branch)| self.subtree_terminal(branch)) + .map(|((owner, name), _)| (*owner, name.clone())) + .collect(); + for key in returned { + self.borrowed.remove(&key); + } + // 2. Release owner grants whose whole subtree is terminal. + let owners: Vec = self.owned.keys().copied().collect(); + for owner in owners { + if self.subtree_terminal(owner) + && let Some(units) = self.owned.remove(&owner) + { + self.resources.release_all(&units); } } - self.borrow_slots - .retain(|_, borrower| !self.graph.group_terminal(*borrower)); } - /// Whether every [`Dep::Node`] edge of `id` is satisfied. `Dep::Resource` - /// edges are handled by the atomic acquire in [`Scheduler::settle`], not here. + /// Whether `id` is clear to start: its parent's own logic is done *and* every + /// [`Dep::Node`] edge is satisfied. The parent gate (a sub-node runs only + /// after its parent reaches [`State::Finishing`] — the parent can't be + /// terminal while this child is pending, so `Finishing` is the exact "logic + /// done" signal) supplies parent→child ordering; `Dep::Node` edges (which the + /// graph restricts to the same parent group) supply sibling ordering. + /// `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; }; + if let Some(parent) = node.parent + && self.graph.node(parent).map(|n| n.state) != Some(State::Finishing) + { + return false; + } node.deps.iter().all(|dep| match dep { Dep::Resource { .. } => true, - Dep::Node { id, when } => self.dep_node_satisfied(*id, *when), + Dep::Node { id, when } => self + .graph + .node(*id) + .is_some_and(|n| when.satisfied_by(n.state)), }) } - /// 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<(R, 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() + self.graph.node(id).map_or_else(Vec::new, |node| { + node.deps + .iter() + .filter_map(|dep| match dep { + Dep::Resource { name, count } => Some((name.clone(), *count)), + Dep::Node { .. } => None, + }) + .collect() + }) } } @@ -282,44 +420,49 @@ mod tests { Scheduler::new(Graph::new(), table) } - fn slot_dep() -> Vec> { - vec![Dep::Resource { - name: res("build-slot"), - count: 1, - }] - } - - fn resource_dep(name: &str) -> Vec> { + /// A single-unit resource dep on `name`. + fn res_dep(name: &str) -> Vec> { vec![Dep::Resource { name: res(name), count: 1, }] } + fn after_ok(on: NodeId) -> Dep { + Dep::Node { + id: on, + when: DepWhen::AfterOk, + } + } + + fn avail(s: &Scheduler<&str, String>, name: &str) -> u32 { + s.resources.available(&res(name)) + } + #[test] - fn resource_node_starts_then_releases_on_complete() { + fn leaf_owner_goes_done_directly_and_releases() { 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. + let n = s + .append("build", res_dep("build-slot"), None) + .expect("insert"); 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). + assert_eq!(avail(&s, "build-slot"), 0); + // No children → completing it goes straight to Done (skips Finishing). 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)); + assert_eq!(avail(&s, "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"); + // Three independent (unparented) nodes each own a fresh slot unit. + let a = s.append("a", res_dep("build-slot"), None).expect("a"); + let b = s.append("b", res_dep("build-slot"), None).expect("b"); + let c = s.append("c", res_dep("build-slot"), 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.settle(), 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); @@ -328,85 +471,212 @@ mod tests { } #[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" + fn parent_parks_in_finishing_until_children_roll_up() { + // `root` (a group node) runs, then its two sub-nodes run. `root` is not + // terminal until both children are — it waits in `Finishing`. + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let root = s.append("root", vec![], None).expect("root"); + let c1 = s.append("c1", vec![], Some(root)).expect("c1"); + let c2 = s.append("c2", vec![], Some(root)).expect("c2"); + assert_eq!(s.settle(), vec![root]); + // Children can't start yet — parent still Running (logic not done). + assert!(s.settle().is_empty(), "children gated on parent logic"); + s.complete(root, Outcome::Done); + assert_eq!( + s.graph().node(root).unwrap().state, + State::Finishing, + "logic done, children pending → Finishing" + ); + let mut started = s.settle(); + started.sort(); + let mut expected = vec![c1, c2]; + expected.sort(); + assert_eq!(started, expected, "children run once parent is Finishing"); + s.complete(c1, Outcome::Done); + assert_eq!( + s.graph().node(root).unwrap().state, + State::Finishing, + "still Finishing while c2 runs" + ); + s.complete(c2, Outcome::Done); + assert_eq!( + s.graph().node(root).unwrap().state, + State::Done, + "rolls up to Done once every child is Done" ); - // 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)); } #[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). + fn failed_child_rolls_parent_up_to_failed() { + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let root = s.append("root", vec![], None).expect("root"); + let child = s.append("child", vec![], Some(root)).expect("child"); + assert_eq!(s.settle(), vec![root]); + s.complete(root, Outcome::Done); + assert_eq!(s.settle(), vec![child]); + s.complete(child, Outcome::Failed); + assert_eq!( + s.graph().node(root).unwrap().state, + State::Failed, + "a failed child rolls the parent up to Failed" + ); + } + + #[test] + fn owner_holds_grant_for_its_whole_subtree() { + // Group root R owns the slot; c1 (its child) borrows it; c2 (its child, + // needs no slot, ordered after c1) doesn't. The slot is held until R's + // WHOLE subtree is terminal — not freed after the last needer (c1). + let mut s = scheduler_with_slots(1); + let r = s.append("R", res_dep("build-slot"), None).expect("R"); + let c1 = s.append("c1", res_dep("build-slot"), Some(r)).expect("c1"); + let c2 = s.append("c2", vec![after_ok(c1)], Some(r)).expect("c2"); + assert_eq!(s.settle(), vec![r]); + s.complete(r, Outcome::Done); // → Finishing (children pending) + assert_eq!(avail(&s, "build-slot"), 0, "held: subtree not terminal"); + assert_eq!(s.settle(), vec![c1], "c1 borrows R's slot"); + assert_eq!(avail(&s, "build-slot"), 0, "borrow reuses R's unit"); + s.complete(c1, Outcome::Done); + assert_eq!( + avail(&s, "build-slot"), + 0, + "still held: c2 pending in subtree" + ); + assert_eq!(s.settle(), vec![c2]); + s.complete(c2, Outcome::Done); + assert_eq!( + avail(&s, "build-slot"), + 1, + "released once whole subtree done" + ); + assert_eq!(s.graph().node(r).unwrap().state, State::Done); + } + + #[test] + fn child_borrows_ancestor_grant_released_when_subtree_done() { + // Lease-shaped resource (agent/foo, default cap 1): the group root owns + // it, its sub-node borrows it, released only once the subtree is done. + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let owner = s + .append("owner", res_dep("agent/foo"), None) + .expect("owner"); + assert_eq!(s.settle(), vec![owner]); + assert_eq!(avail(&s, "agent/foo"), 0); + let child = s + .append("child", res_dep("agent/foo"), Some(owner)) + .expect("child"); + s.complete(owner, Outcome::Done); // → Finishing + assert_eq!(avail(&s, "agent/foo"), 0, "held while a borrower pends"); + assert_eq!(s.settle(), vec![child]); + assert_eq!(avail(&s, "agent/foo"), 0, "borrow reuses the one unit"); + s.complete(child, Outcome::Done); + assert_eq!(avail(&s, "agent/foo"), 1); + } + + #[test] + fn nested_borrowers_never_deadlock() { + // R (owns foo) → c1 → gc1 → ggc1, each the child of the previous, all + // needing agent/foo (cap 1). c1 borrows R's grant; gc1 + ggc1 are inside + // c1's borrow-branch so they are *covered* — a deep nest never deadlocks + // on the single unit, and foo is held for the whole nest. + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let r = s.append("R", res_dep("agent/foo"), None).expect("R"); + let child = s + .append("child", res_dep("agent/foo"), Some(r)) + .expect("child"); + let grand = s + .append("grand", res_dep("agent/foo"), Some(child)) + .expect("grand"); + let great = s + .append("great", res_dep("agent/foo"), Some(grand)) + .expect("great"); + assert_eq!(s.settle(), vec![r]); + s.complete(r, Outcome::Done); + assert_eq!(s.settle(), vec![child], "child borrows R's grant"); + s.complete(child, Outcome::Done); + assert_eq!(s.settle(), vec![grand], "grand covered, no deadlock"); + s.complete(grand, Outcome::Done); + assert_eq!(s.settle(), vec![great], "great covered too"); + assert_eq!(avail(&s, "agent/foo"), 0, "held across the whole nest"); + s.complete(great, Outcome::Done); + assert_eq!(s.graph().node(r).unwrap().state, State::Done, "R rolled up"); + assert_eq!(avail(&s, "agent/foo"), 1, "released once the nest is done"); + } + + #[test] + fn unrelated_nodes_needing_the_same_resource_are_serialized() { + // Two unparented nodes need agent/foo (cap 1); neither is in the other's + // subtree, so the second can't borrow — it waits for the first's release. + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let a = s.append("a", res_dep("agent/foo"), None).expect("a"); + let b = s.append("b", res_dep("agent/foo"), None).expect("b"); + assert_eq!(s.settle(), vec![a], "only a acquires; b can't borrow it"); + assert_eq!(s.graph().node(b).unwrap().state, State::Pending); + s.complete(a, Outcome::Done); + assert_eq!(s.settle(), vec![b]); + assert_eq!(s.graph().node(b).unwrap().state, State::Running); + } + + #[test] + fn sibling_borrowers_of_a_cap1_grant_serialize() { + // Two children of the owner both need agent/foo (cap 1): one borrows the + // grant, the other (grant lent to a sibling branch, no free unit) waits + // until the borrow is returned — mutual exclusion within the group. + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let owner = s + .append("owner", res_dep("agent/foo"), None) + .expect("owner"); + assert_eq!(s.settle(), vec![owner]); let c1 = s - .append("c1", resource_dep("agent/foo"), Some(group)) + .append("c1", res_dep("agent/foo"), Some(owner)) .expect("c1"); let c2 = s - .append("c2", resource_dep("agent/foo"), Some(group)) + .append("c2", res_dep("agent/foo"), Some(owner)) .expect("c2"); - let started = s.settle(); - assert_eq!( - started, - vec![c1], - "only one descendant may borrow at a time" - ); + s.complete(owner, Outcome::Done); // → Finishing + assert_eq!(s.settle(), vec![c1], "c1 borrows; c2 can't (cap 1)"); 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)); + assert_eq!(s.settle(), vec![c2], "borrow returned → c2 borrows"); + assert_eq!(avail(&s, "agent/foo"), 0, "still just the owner's unit"); + } + + #[test] + fn sibling_borrowers_run_concurrently_when_capacity_allows() { + // build-slot cap 2: owner holds one unit; c1 borrows it, c2 (grant lent + // to a sibling branch) takes the *second* unit — both run at once. + let mut s = scheduler_with_slots(2); + let owner = s + .append("owner", res_dep("build-slot"), None) + .expect("owner"); + assert_eq!(s.settle(), vec![owner]); + assert_eq!(avail(&s, "build-slot"), 1, "owner took one of two"); + let c1 = s + .append("c1", res_dep("build-slot"), Some(owner)) + .expect("c1"); + let c2 = s + .append("c2", res_dep("build-slot"), Some(owner)) + .expect("c2"); + s.complete(owner, Outcome::Done); // → Finishing + let mut started = s.settle(); + started.sort(); + let mut expected = vec![c1, c2]; + expected.sort(); + assert_eq!(started, expected, "c1 borrows, c2 takes the 2nd unit"); + assert_eq!(avail(&s, "build-slot"), 0); } #[test] fn failed_after_ok_dep_cancels_dependents_but_after_any_still_runs() { + // A group of top-level siblings ordered by `AfterOk`; the failure of + // `root` cancels its strong-dependent chain, an `AfterAny` still runs. let mut s: Scheduler<&str, String> = 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, - ) + .append("strong1", vec![after_ok(root)], None) .expect("strong1"); let strong2 = s - .append( - "strong2", - vec![Dep::Node { - id: strong1, - when: DepWhen::AfterOk, - }], - None, - ) + .append("strong2", vec![after_ok(strong1)], None) .expect("strong2"); let weak = s .append( @@ -420,11 +690,47 @@ mod tests { .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]); } + + #[test] + fn failed_parent_cancels_its_pending_children() { + // A failed group node cancels its sub-nodes (they were gated from ever + // running on a `Finishing` the parent never reached). + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let root = s.append("root", vec![], None).expect("root"); + let child = s.append("child", vec![], Some(root)).expect("child"); + let grandchild = s.append("gc", vec![], Some(child)).expect("gc"); + assert_eq!(s.settle(), vec![root]); + s.complete(root, Outcome::Failed); + assert_eq!(s.graph().node(child).unwrap().state, State::Cancelled); + assert_eq!(s.graph().node(grandchild).unwrap().state, State::Cancelled); + } + + #[test] + fn cancel_node_cancels_pending_and_cascades_but_not_running() { + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let a = s.append("a", vec![], None).expect("a"); + let b = s.append("b", vec![after_ok(a)], None).expect("b"); + assert!(s.cancel_node(a)); + assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled); + assert_eq!(s.graph().node(b).unwrap().state, State::Cancelled); + let c = s.append("c", vec![], None).expect("c"); + assert_eq!(s.settle(), vec![c]); + assert!(!s.cancel_node(c)); + assert_eq!(s.graph().node(c).unwrap().state, State::Running); + } + + #[test] + fn resource_state_reports_owners() { + let mut s = scheduler_with_slots(1); + let g = s.append("g", res_dep("agent/foo"), None).expect("g"); + let b = s.append("b", res_dep("build-slot"), None).expect("b"); + assert_eq!(s.settle().len(), 2); + let state = s.resource_state(); + assert!(state.contains(&(res("agent/foo"), g))); + assert!(state.contains(&(res("build-slot"), b))); + } }