vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(#2591): hive-jobq parent-axis grouping + borrow + roll-up scheduler

Browse source 
Rework the crate's scheduling model onto an explicit parent (grouping)
axis, separate from the dep (ordering) axis.

- Node gains a structural `parent: Option<NodeId>`, 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.
This commit is contained in:
atlas 2026-07-20 19:45:53 +02:00 • committed by mara
commit 5906cc2f2b
3 changed files with 734 additions and 481 deletions
Download patch file Download diff file Expand all files Collapse all files

130
hive-jobq/src/guard.rs
View file

@ -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<RefCell<…>>` (single-threaded interior mutability), not
//! `Arc<Mutex<…>>` — 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<R>(Rc<RefCell<ResourceTable<R>>>);
impl<R: Clone + Eq + Hash> Default for SharedResources<R> {
fn default() -> Self {
Self::new(ResourceTable::new())
}
}
impl<R: Clone + Eq + Hash> SharedResources<R> {
/// Wrap an existing table so guards can release into it.
#[must_use]
pub fn new(table: ResourceTable<R>) -> 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<ResourceGuard<R>> {
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<T>(&self, f: impl FnOnce(&ResourceTable<R>) -> 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<R: Clone + Eq + Hash> {
table: SharedResources<R>,
reqs: Vec<(R, u32)>,
}
impl<R: Clone + Eq + Hash> ResourceGuard<R> {
/// The `(name, count)` units this guard releases on drop.
#[must_use]
pub fn held(&self) -> &[(R, u32)] {
&self.reqs
}
}
impl<R: Clone + Eq + Hash> Drop for ResourceGuard<R> {
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<String> {
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());
}
}

343
hive-jobq/src/lib.rs
View file

@ -4,42 +4,35 @@
//! # Model (v2) //! # Model (v2)
//! //!
//! One **persistent graph** for the whole system, not a DAG per job. Enqueuing //! 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 //! inserts a self-contained sub-DAG of nodes and returns their ids; the
//! runs a continuous loop, starting every node whose [`Dep`]s are satisfied: //! scheduler runs a continuous loop, starting every node whose [`Dep`]s are
//! satisfied:
//! //!
//! - **Resource** deps are named counting semaphores over a caller-chosen //! - **Resource** deps are named counting semaphores over a caller-chosen
//! type `R` (a `String` or an enum): `build-slot` (cap N), `agent/<name>` //! type `R`: `build-slot` (cap N), `agent/<name>` (cap 1), or any name
//! (cap 1), or any name (cap 1, created on use). A node acquires *all* its //! (cap 1, created on use). A node acquires *all* its resource deps
//! resource deps atomically at start (all-or-nothing) — no hold-and-wait, //! atomically at start (all-or-nothing) — no hold-and-wait, no deadlock.
//! so no deadlock and no cycle detection needed. //! - **Node** deps wait on another node per [`DepWhen`]: `AfterOk` needs
//! - **Node** deps wait on a node/group per [`DepWhen`]: `AfterOk` needs
//! success (a failed dep cancels the dependent), `AfterAny` only terminal. //! 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 //! A node carries two independent axes: its [`Dep`]s (ordering + resource
//! whole (done = every inner node terminal), never on an inner node. Groups //! needs) and its [`Node::parent`] (structural grouping) — the parent chain,
//! nest; a running node may grow its own group but not reach outside it. //! 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 //! A resource unit is held for the acquiring node + its whole [`Node::parent`]
//! restarts. Group membership is a parent edge ([`Node::parent`]), *not* in the //! subtree; a node needing a resource an ancestor holds re-uses that grant (a
//! id; the `1/1/2` hierarchy is a derived UI label. The node payload is generic //! re-entrant borrow, one branch at a time). Single-threaded — the scheduler
//! (`N`) so the library stays container-agnostic — the caller supplies its own //! owns the resource table and mutates it directly. See [`scheduler`].
//! 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`.
pub(crate) mod guard;
pub mod resources; pub mod resources;
pub mod scheduler; pub mod scheduler;
/// Opaque, stable, monotonic node identifier. /// Opaque, stable, monotonic node identifier.
/// ///
/// Assigned by the [`Graph`] on insert and persisted, so it is stable across /// Assigned by the [`Graph`] on insert and persisted, so it is stable across
/// restarts. Group membership is a separate parent edge ([`Node::parent`]) — it /// restarts.
/// 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.
/// ///
/// The inner field is crate-private: an id can only originate from the graph's /// 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 /// monotonic counter (or deserialization of a persisted graph), never be
@ -49,6 +42,18 @@ pub mod scheduler;
)] )]
pub struct NodeId(pub(crate) u64); 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 /// When a [`Dep::Node`] edge is satisfied — the strong/weak distinction the
/// current queue carries as `DepWhen`, load-bearing for failure safety. /// current queue carries as `DepWhen`, load-bearing for failure safety.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[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). /// it names can be acquired (all of them, atomically).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum Dep<R> { pub enum Dep<R> {
/// Depend on another node (or a group, by its group node's id). Whether a /// Depend on another node. Whether a *failed* dependency satisfies the edge
/// *failed* dependency satisfies the edge is decided by `when`: `AfterOk` /// is decided by `when`: `AfterOk` requires success (and cancels this node
/// requires success (and cancels this node if the dep fails), `AfterAny` /// if the dep fails), `AfterAny` only requires the dep to be terminal.
/// only requires the dep to be terminal.
Node { Node {
/// The node (or group) depended on. /// The node depended on.
id: NodeId, id: NodeId,
/// Strong (`AfterOk`) vs weak (`AfterAny`). /// Strong (`AfterOk`) vs weak (`AfterAny`).
when: DepWhen, when: DepWhen,
}, },
/// Hold `count` units of a named resource for the duration of this node's /// Need `count` units of a named resource to run. Declared on every node
/// run. Acquired atomically with the node's other resource deps at start, /// that needs it, even when a [`Node::parent`]-ancestor already holds it.
/// released when the node completes. /// 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 { Resource {
/// The resource to acquire. /// The resource to acquire.
name: R, name: R,
@ -106,11 +114,17 @@ pub enum Dep<R> {
pub enum State { pub enum State {
/// Waiting on dependencies (node or resource). /// Waiting on dependencies (node or resource).
Pending, Pending,
/// Dependencies satisfied, resources held, currently executing. /// Dependencies satisfied, resources held, currently executing its own logic.
Running, 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, Done,
/// Completed unsuccessfully. /// Completed unsuccessfully — own logic failed, or a sub-node did.
Failed, Failed,
/// Never ran: an `AfterOk` dependency failed, so this node (and the rest of /// Never ran: an `AfterOk` dependency failed, so this node (and the rest of
/// its strong-dependent chain) is cancelled rather than run. /// 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 /// 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 /// 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)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Node<N, R> { pub struct Node<N, R> {
/// Stable identity, assigned on insert. /// Stable identity, assigned on insert.
pub id: NodeId, pub id: NodeId,
/// The group this node belongs to, if any. `None` for a top-level group /// Structural grouping: the node this one is a sub-node of, or `None` for a
/// node. Group membership lives here, not in the id. /// 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<NodeId>, pub parent: Option<NodeId>,
/// Caller-defined payload (the node's kind / work description). /// Caller-defined payload (the node's kind / work description).
pub payload: N, pub payload: N,
@ -158,9 +178,20 @@ pub enum GraphError {
/// A node's dependency named an id not present in the graph. /// A node's dependency named an id not present in the graph.
#[error("dependency references unknown node {0:?}")] #[error("dependency references unknown node {0:?}")]
UnknownDep(NodeId), 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:?}")] #[error("parent references unknown node {0:?}")]
UnknownParent(NodeId), 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<NodeId>,
},
/// A loaded graph's `next_id` counter is not past the largest existing id, /// 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. /// 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}")] #[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. /// The single persistent graph of all nodes.
/// ///
/// New jobs are inserted as node groups; the scheduler (added in a follow-up) /// New jobs are inserted as sub-DAGs of nodes; the scheduler walks this graph
/// walks this graph filling open slots. Completed groups are retained (no /// filling open slots. Completed nodes are retained (no pruning in v1).
/// pruning in v1).
#[derive(Debug, serde::Serialize, serde::Deserialize)] #[derive(Debug, serde::Serialize, serde::Deserialize)]
#[serde( #[serde(
try_from = "GraphData<N, R>", try_from = "GraphData<N, R>",
@ -238,33 +268,37 @@ impl<N, R> Graph<N, R> {
id id
} }
/// Insert a node with the given payload, deps, and parent group, returning /// Insert a node with the given payload, deps, and `parent`, returning its
/// its freshly-minted id. The node starts [`State::Pending`]. /// freshly-minted id. The node starts [`State::Pending`].
/// ///
/// Every [`Dep::Node`] id and the `parent` id (if any) must already resolve /// Every [`Dep::Node`] id and the `parent` id (when `Some`) must already
/// to a node in the graph — an id is only meaningful against the graph that /// resolve to a node in the graph — an id is only meaningful against the
/// minted it, so a dangling reference is rejected here rather than surfacing /// graph that minted it, so a dangling reference is rejected here rather than
/// as a broken edge later. /// surfacing as a broken edge later.
/// ///
/// # Errors /// # Errors
/// Returns [`GraphError::UnknownParent`] / [`GraphError::UnknownDep`] if the /// Returns [`GraphError::UnknownDep`] / [`GraphError::UnknownParent`] for a
/// parent or a dependency references a node not in the graph. /// dangling dependency or parent id, or [`GraphError::DepOutsideParent`] if a
/// `Dep::Node` edge points outside the node's own parent group.
pub fn insert( pub fn insert(
&mut self, &mut self,
payload: N, payload: N,
deps: Vec<Dep<R>>, deps: Vec<Dep<R>>,
parent: Option<NodeId>, parent: Option<NodeId>,
) -> Result<NodeId, GraphError> { ) -> Result<NodeId, GraphError> {
if let Some(parent_id) = parent if let Some(p) = parent
&& self.node(parent_id).is_none() && self.node(p).is_none()
{ {
return Err(GraphError::UnknownParent(parent_id)); return Err(GraphError::UnknownParent(p));
} }
for dep in &deps { for dep in &deps {
if let Dep::Node { id, .. } = dep if let Dep::Node { id, .. } = dep {
&& self.node(*id).is_none() if self.node(*id).is_none() {
{ return Err(GraphError::UnknownDep(*id));
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(); let id = self.mint_id();
@ -284,17 +318,39 @@ impl<N, R> Graph<N, R> {
self.nodes.iter().find(|n| n.id == id) 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<Item = &Node<N, R>> {
self.nodes.iter().filter(move |n| n.parent == Some(id))
}
/// Every node in the graph, in insertion order. The scheduler iterates /// Every node in the graph, in insertion order. The scheduler iterates
/// this to find runnable pending nodes. /// this to find runnable pending nodes.
pub fn nodes(&self) -> impl Iterator<Item = &Node<N, R>> { pub fn nodes(&self) -> impl Iterator<Item = &Node<N, R>> {
self.nodes.iter() 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<NodeId>, 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 /// Set a node's lifecycle state, returning `false` for an unknown id. The
/// scheduler drives every state transition — nothing else mutates state, /// scheduler drives every state transition — nothing else mutates state,
/// which is what keeps the resource guards + terminality in sync. /// which is what keeps the resource guards + terminality in sync.
@ -307,40 +363,32 @@ impl<N, R> Graph<N, R> {
} }
} }
/// A group is terminal once the group node itself is terminal *and* every /// Check that every id the graph holds resolves: every [`Dep::Node`] id
/// node inside it (recursively) is terminal. The node's own state matters: /// names a node present in the graph, and `next_id` is past the largest
/// a group node still `Pending`/`Running` is not terminal even with no /// existing id. Deserialization runs this, so a loaded graph is internally
/// children yet, since a running node may still append some. (Deciding when /// consistent and internal iteration can trust its ids.
/// 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.
/// ///
/// # Errors /// # Errors
/// Returns [`GraphError`] on a dangling parent / dependency reference, or a /// Returns [`GraphError`] on a dangling dependency reference, or a `next_id`
/// `next_id` that would remint an id already in the graph. /// that would remint an id already in the graph.
pub fn validate(&self) -> Result<(), GraphError> { pub fn validate(&self) -> Result<(), GraphError> {
for node in &self.nodes { for node in &self.nodes {
if let Some(parent_id) = node.parent if let Some(p) = node.parent
&& self.node(parent_id).is_none() && self.node(p).is_none()
{ {
return Err(GraphError::UnknownParent(parent_id)); return Err(GraphError::UnknownParent(p));
} }
for dep in &node.deps { for dep in &node.deps {
if let Dep::Node { id, .. } = dep if let Dep::Node { id, .. } = dep {
&& self.node(*id).is_none() if self.node(*id).is_none() {
{ return Err(GraphError::UnknownDep(*id));
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, id: a,
when: DepWhen::AfterOk, when: DepWhen::AfterOk,
}], }],
Some(a), None,
) )
.unwrap(); .unwrap();
assert_eq!(a, NodeId(0)); assert_eq!(a, NodeId(0));
assert_eq!(b, NodeId(1)); assert_eq!(b, NodeId(1));
// Membership is the parent edge, not the id. // The dep edge references the earlier node; ids are stable + monotonic.
assert_eq!(g.node(b).unwrap().parent, Some(a)); assert!(matches!(
assert_eq!(g.node(a).unwrap().parent, None); g.node(b).unwrap().deps.first(),
} Some(Dep::Node { id, .. }) if *id == a
));
#[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));
} }
#[test] #[test]
@ -417,6 +438,8 @@ mod tests {
assert!(State::Cancelled.is_terminal()); assert!(State::Cancelled.is_terminal());
assert!(!State::Pending.is_terminal()); assert!(!State::Pending.is_terminal());
assert!(!State::Running.is_terminal()); assert!(!State::Running.is_terminal());
// Finishing (logic done, children still running) is NOT terminal.
assert!(!State::Finishing.is_terminal());
} }
#[test] #[test]
@ -432,16 +455,10 @@ mod tests {
assert!(DepWhen::AfterAny.satisfied_by(State::Failed)); assert!(DepWhen::AfterAny.satisfied_by(State::Failed));
assert!(DepWhen::AfterAny.satisfied_by(State::Cancelled)); assert!(DepWhen::AfterAny.satisfied_by(State::Cancelled));
assert!(!DepWhen::AfterAny.satisfied_by(State::Pending)); assert!(!DepWhen::AfterAny.satisfied_by(State::Pending));
} // Finishing satisfies neither — a dependent waits until the node rolls
// up to a terminal state (all its sub-nodes done).
#[test] assert!(!DepWhen::AfterOk.satisfied_by(State::Finishing));
fn insert_rejects_unknown_parent() { assert!(!DepWhen::AfterAny.satisfied_by(State::Finishing));
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)
);
} }
#[test] #[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] #[test]
fn valid_graph_round_trips_through_serde() { fn valid_graph_round_trips_through_serde() {
let mut g: Graph<String, String> = Graph::new(); let mut g: Graph<String, String> = 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(); let a = g.insert("a".to_owned(), vec![], None).unwrap();
g.insert( g.insert(
"b".to_owned(), "b".to_owned(),
@ -468,13 +501,57 @@ mod tests {
id: a, id: a,
when: DepWhen::AfterAny, when: DepWhen::AfterAny,
}], }],
Some(a), None,
) )
.unwrap(); .unwrap();
let c = g.insert("c".to_owned(), vec![], Some(a)).unwrap();
let json = serde_json::to_string(&g).unwrap(); let json = serde_json::to_string(&g).unwrap();
let back: Graph<String, String> = serde_json::from_str(&json).unwrap(); let back: Graph<String, String> = serde_json::from_str(&json).unwrap();
assert!(back.validate().is_ok()); assert!(back.validate().is_ok());
assert_eq!(back.node(a).unwrap().payload, "a"); 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<String> {
Dep::Node {
id: on,
when: DepWhen::AfterOk,
}
} }
#[test] #[test]

748
hive-jobq/src/scheduler.rs
View file

@ -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 //! [`Scheduler::settle`] claims every currently-runnable pending node (its
//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired //! [`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 //! 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 //! node's result back with [`Scheduler::complete`]; a running node may grow more
//! own sub-group first via [`Scheduler::append`]. Concurrency is emergent from //! work first via [`Scheduler::append`]. Concurrency is emergent from resource
//! resource capacity — there is no separate active-node cap. //! capacity — there is no separate active-node cap.
//! //!
//! A resource is held for the acquiring node's *entire subtree* lifetime: the //! Single-threaded by design: the scheduler is the only driver, holds the
//! owned guard is released only when that node and every descendant is terminal //! [`ResourceTable`] as a plain owned field, mutating it through `&mut self` —
//! (`group_terminal`), not when the node's own work finishes. Single-owner and //! no interior mutability, no guard objects.
//! 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 //! Completion rolls up the parent tree: a node with children parks in
//! the eager `AfterOk` failure cascade are layered on top of this owned core. //! [`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::collections::HashMap;
use std::hash::Hash; use std::hash::Hash;
use crate::guard::{ResourceGuard, SharedResources};
use crate::resources::ResourceTable; use crate::resources::ResourceTable;
use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State}; use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State};
@ -36,19 +46,22 @@ pub enum Outcome {
Failed, Failed,
} }
/// Drives a [`Graph`] over a shared resource pool: claim runnable nodes, hold /// Drives a [`Graph`] over an owned resource pool: claim runnable nodes, record
/// their resources for the subtree's lifetime, release on subtree-terminal. /// 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<N, R: Clone + Eq + Hash> { pub struct Scheduler<N, R: Clone + Eq + Hash> {
graph: Graph<N, R>, graph: Graph<N, R>,
resources: SharedResources<R>, resources: ResourceTable<R>,
/// Owned resource guards, keyed by the node that acquired them. Dropped /// Fresh units each owner node acquired: `owner → [(resource, count)]`.
/// (releasing the units) when that node's whole subtree is terminal. /// Recorded against the node that *acquired* the units (never a borrower);
owned: HashMap<NodeId, Vec<ResourceGuard<R>>>, /// released back to the table once the owner and its whole [`Node::parent`]
/// The single re-entrancy slot per `(ancestor-holder, resource)`: the id of /// subtree are terminal.
/// the descendant currently *borrowing* that ancestor's lock. Present ⇒ the owned: HashMap<NodeId, Vec<(R, u32)>>,
/// slot is taken, so no other descendant may re-enter the same lock until /// Which branch currently borrows a given owner's grant: `(owner, resource)
/// the borrower's subtree is terminal — "only one node at a time within". /// → branch-root node`. A grant is lent to one branch at a time; nodes
borrow_slots: HashMap<(NodeId, R), NodeId>, /// 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<N, R: Clone + Eq + Hash> Scheduler<N, R> { impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
@ -57,9 +70,9 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
pub fn new(graph: Graph<N, R>, resources: ResourceTable<R>) -> Self { pub fn new(graph: Graph<N, R>, resources: ResourceTable<R>) -> Self {
Self { Self {
graph, graph,
resources: SharedResources::new(resources), resources,
owned: HashMap::new(), owned: HashMap::new(),
borrow_slots: HashMap::new(), borrowed: HashMap::new(),
} }
} }
@ -69,12 +82,12 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
&self.graph &self.graph
} }
/// Append a node — e.g. a running node growing its own sub-group. Delegates /// Append a node under `parent` — e.g. a running node growing more work into
/// to [`Graph::insert`]; call [`Scheduler::settle`] afterwards to start it /// its own subtree. Delegates to [`Graph::insert`]; call [`Scheduler::settle`]
/// once it is runnable. /// afterwards to start it once it is runnable.
/// ///
/// # Errors /// # Errors
/// Propagates [`GraphError`] for a dangling parent or dependency id. /// Propagates [`GraphError`] for a dangling dependency or parent id.
pub fn append( pub fn append(
&mut self, &mut self,
payload: N, payload: N,
@ -86,9 +99,9 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// Claim every currently-runnable pending node and start it: node-deps /// Claim every currently-runnable pending node and start it: node-deps
/// satisfied and all resource-deps acquired atomically (all-or-nothing). /// satisfied and all resource-deps acquired atomically (all-or-nothing).
/// Each claimed node is marked `Running`, its owned guards held, and its id /// Each claimed node is marked `Running`, its acquired units recorded, and
/// returned for the runner to execute. A single pass suffices — a node /// its id returned for the runner to execute. A single pass suffices — a
/// started here is `Running`, not terminal, so it cannot satisfy another /// node started here is `Running`, not terminal, so it cannot satisfy another
/// node's dependency in the same pass; it only consumes resources. /// node's dependency in the same pass; it only consumes resources.
#[must_use] #[must_use]
pub fn settle(&mut self) -> Vec<NodeId> { pub fn settle(&mut self) -> Vec<NodeId> {
@ -107,163 +120,288 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
started started
} }
/// Try to start node `id`: classify each resource dep as *owned* (no /// Try to start node `id`. For each resource it needs, decide per the parent
/// ancestor holds it → acquire real units) or *borrowed* (an ancestor group /// tree (see the module docs): acquire fresh units (owner), acquire an extra
/// already holds it → re-enter, gated by the one re-entrancy slot), then /// unit (grant lent elsewhere), or borrow an ancestor's grant. The fresh set
/// take everything atomically or nothing. Returns whether it started. /// 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 { fn try_start(&mut self, id: NodeId) -> bool {
let mut owned_reqs = Vec::new(); let mut to_acquire: Vec<(R, u32)> = Vec::new();
let mut borrows = Vec::new(); let mut to_borrow: Vec<(NodeId, R)> = Vec::new();
for (name, count) in self.resource_reqs(id) { for (name, count) in self.resource_reqs(id) {
if let Some(ancestor) = self.ancestor_owning(id, &name) { match self.parent_ancestor_owning(id, &name) {
// Re-entrant reuse: allowed only if the slot is free. // Case 1: no ancestor holds it → this node acquires + owns it.
if self.borrow_slots.contains_key(&(ancestor, name.clone())) { None => to_acquire.push((name, count)),
return false; Some(owner) => match self.borrowed.get(&(owner, name.clone())).copied() {
} // Case 3: the grant is free → borrow it, no new unit.
borrows.push((ancestor, name)); None => to_borrow.push((owner, name)),
} else { // Covered: already lent to a branch this node is inside.
owned_reqs.push((name, count)); 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 if !to_acquire.is_empty() && !self.resources.try_acquire_all(&to_acquire) {
// confirmed free above, so acquiring them is the only fallible step. return false;
// 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);
} }
for slot in borrows { if !to_acquire.is_empty() {
self.borrow_slots.insert(slot, id); 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); self.graph.set_state(id, State::Running);
true true
} }
/// The nearest ancestor of `id` that *owns* (holds real units of) `name`, /// The nearest [`Node::parent`] ancestor of `id` that *owns* (holds real
/// or `None` if no ancestor holds it (⇒ `id` must own-acquire it itself). /// units of) `name`, or `None` if none does (⇒ `id` must acquire it fresh).
fn ancestor_owning(&self, id: NodeId, name: &R) -> Option<NodeId> { fn parent_ancestor_owning(&self, id: NodeId, name: &R) -> Option<NodeId> {
let mut cursor = self.graph.node(id)?.parent; let mut cur = self.graph.node(id).and_then(|n| n.parent);
while let Some(ancestor) = cursor { while let Some(p) = cur {
if self.node_owns(ancestor, name) { if self.node_owns(p, name) {
return Some(ancestor); return Some(p);
} }
cursor = self.graph.node(ancestor)?.parent; cur = self.graph.node(p).and_then(|n| n.parent);
} }
None None
} }
/// Whether node `holder` holds an owned guard covering resource `name`. /// Whether `ancestor` lies on `id`'s [`Node::parent`] chain (i.e. `id` is in
fn node_owns(&self, holder: NodeId, name: &R) -> bool { /// `ancestor`'s subtree). `id` itself does not count as its own ancestor.
self.owned.get(&holder).is_some_and(|guards| { fn parent_chain_contains(&self, id: NodeId, ancestor: NodeId) -> bool {
guards let mut cur = self.graph.node(id).and_then(|n| n.parent);
.iter() while let Some(p) = cur {
.any(|g| g.held().iter().any(|(n, _)| n == name)) 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 /// Whether node `holder` holds real units of resource `name`.
/// releases the owned guards of every node whose whole subtree has become fn node_owns(&self, holder: NodeId, name: &R) -> bool {
/// terminal — a parent keeps its lock until its last descendant finishes. self.owned
/// Call [`Scheduler::settle`] again afterwards to start newly-unblocked work. .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) { pub fn complete(&mut self, id: NodeId, outcome: Outcome) {
let state = match outcome { match outcome {
Outcome::Done => State::Done, Outcome::Failed => {
Outcome::Failed => State::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); self.graph.set_state(id, state);
if outcome == Outcome::Failed { if state == State::Failed {
self.cascade_cancel(id); self.cascade_cancel(id);
} }
self.release_settled_subtrees();
} }
/// Eagerly cancel the transitive `AfterOk` dependents of a just-failed node: /// After `start` became terminal, roll up every ancestor that was parked in
/// they can never run (a strong dependency failed), so mark them `Cancelled` /// `Finishing` awaiting its children: once all of an ancestor's children are
/// now — before they could claim resources. A dependent is always still /// terminal it transitions (Done / Failed), which may let *its* parent roll
/// `Pending` here (a `Running` node's `AfterOk` deps were `Done` when it /// up too, and so on up the [`Node::parent`] chain.
/// started, and `Done` is terminal), so no resources need releasing. fn roll_up_ancestors(&mut self, start: NodeId) {
fn cascade_cancel(&mut self, failed: NodeId) { let mut cur = self.graph.node(start).and_then(|n| n.parent);
let mut stack = vec![failed]; while let Some(a) = cur {
while let Some(dep) = stack.pop() { if self.graph.node(a).map(|n| n.state) != Some(State::Finishing)
let dependents: Vec<NodeId> = self || !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<NodeId> = self
.graph .graph
.nodes() .nodes()
.filter(|n| { .filter(|n| {
n.state == State::Pending n.state == State::Pending
&& n.deps.iter().any( && (n.parent == Some(cur)
|d| matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == dep), || n.deps.iter().any(|d| {
) matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == cur)
}))
}) })
.map(|n| n.id) .map(|n| n.id)
.collect(); .collect();
for d in dependents { for d in doomed {
self.graph.set_state(d, State::Cancelled); self.graph.set_state(d, State::Cancelled);
stack.push(d); stack.push(d);
} }
} }
} }
/// Release everything whose subtree has become terminal: drop the owned /// Give back any borrow whose branch has fully left (freeing the grant for a
/// guards of any holder (→ frees its units) and free any re-entrancy slot /// waiting sibling), then release every owner's grant whose whole subtree is
/// held by a borrower — both are held for the whole subtree lifetime. /// terminal (dropping the units back into the table).
fn release_settled_subtrees(&mut self) { fn release_ready(&mut self) {
let holders: Vec<NodeId> = self.owned.keys().copied().collect(); // 1. Return borrows whose branch-root subtree is now terminal.
for holder in holders { let returned: Vec<(NodeId, R)> = self
if self.graph.group_terminal(holder) { .borrowed
self.owned.remove(&holder); // drops guards → releases the units .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<NodeId> = 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` /// Whether `id` is clear to start: its parent's own logic is done *and* every
/// edges are handled by the atomic acquire in [`Scheduler::settle`], not here. /// [`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 { fn node_deps_satisfied(&self, id: NodeId) -> bool {
let Some(node) = self.graph.node(id) else { let Some(node) = self.graph.node(id) else {
return false; 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 { node.deps.iter().all(|dep| match dep {
Dep::Resource { .. } => true, 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. /// The `(name, count)` resource units `id` must hold to run.
fn resource_reqs(&self, id: NodeId) -> Vec<(R, u32)> { fn resource_reqs(&self, id: NodeId) -> Vec<(R, u32)> {
let Some(node) = self.graph.node(id) else { self.graph.node(id).map_or_else(Vec::new, |node| {
return Vec::new(); node.deps
}; .iter()
node.deps .filter_map(|dep| match dep {
.iter() Dep::Resource { name, count } => Some((name.clone(), *count)),
.filter_map(|dep| match dep { Dep::Node { .. } => None,
Dep::Resource { name, count } => Some((name.clone(), *count)), })
Dep::Node { .. } => None, .collect()
}) })
.collect()
} }
} }
@ -282,44 +420,49 @@ mod tests {
Scheduler::new(Graph::new(), table) Scheduler::new(Graph::new(), table)
} }
fn slot_dep() -> Vec<Dep<String>> { /// A single-unit resource dep on `name`.
vec![Dep::Resource { fn res_dep(name: &str) -> Vec<Dep<String>> {
name: res("build-slot"),
count: 1,
}]
}
fn resource_dep(name: &str) -> Vec<Dep<String>> {
vec![Dep::Resource { vec![Dep::Resource {
name: res(name), name: res(name),
count: 1, count: 1,
}] }]
} }
fn after_ok(on: NodeId) -> Dep<String> {
Dep::Node {
id: on,
when: DepWhen::AfterOk,
}
}
fn avail(s: &Scheduler<&str, String>, name: &str) -> u32 {
s.resources.available(&res(name))
}
#[test] #[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 mut s = scheduler_with_slots(1);
let n = s.append("build", slot_dep(), None).expect("insert"); let n = s
// settle claims it (a slot is free) and marks it Running. .append("build", res_dep("build-slot"), None)
.expect("insert");
assert_eq!(s.settle(), vec![n]); assert_eq!(s.settle(), vec![n]);
assert_eq!(s.graph().node(n).unwrap().state, State::Running); assert_eq!(s.graph().node(n).unwrap().state, State::Running);
// Slot is held. assert_eq!(avail(&s, "build-slot"), 0);
assert!(s.resources.with(|t| t.available(&res("build-slot")) == 0)); // No children → completing it goes straight to Done (skips Finishing).
// Completing it releases the slot (subtree is just this node).
s.complete(n, Outcome::Done); s.complete(n, Outcome::Done);
assert_eq!(s.graph().node(n).unwrap().state, State::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] #[test]
fn build_slot_cap_limits_concurrency_and_release_unblocks() { fn build_slot_cap_limits_concurrency_and_release_unblocks() {
let mut s = scheduler_with_slots(2); let mut s = scheduler_with_slots(2);
let a = s.append("a", slot_dep(), None).expect("a"); // Three independent (unparented) nodes each own a fresh slot unit.
let b = s.append("b", slot_dep(), None).expect("b"); let a = s.append("a", res_dep("build-slot"), None).expect("a");
let c = s.append("c", slot_dep(), None).expect("c"); 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. // cap 2 → a + b start, c blocks on the exhausted slot.
let started = s.settle(); assert_eq!(s.settle(), vec![a, b]);
assert_eq!(started, vec![a, b]);
assert_eq!(s.graph().node(c).unwrap().state, State::Pending); assert_eq!(s.graph().node(c).unwrap().state, State::Pending);
// a finishes → its slot frees → c can now start. // a finishes → its slot frees → c can now start.
s.complete(a, Outcome::Done); s.complete(a, Outcome::Done);
@ -328,85 +471,212 @@ mod tests {
} }
#[test] #[test]
fn parent_holds_resource_until_child_subtree_done() { fn parent_parks_in_finishing_until_children_roll_up() {
let mut s = scheduler_with_slots(1); // `root` (a group node) runs, then its two sub-nodes run. `root` is not
// Parent grabs the single build-slot and runs. // terminal until both children are — it waits in `Finishing`.
let parent = s.append("parent", slot_dep(), None).expect("parent"); let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
assert_eq!(s.settle(), vec![parent]); let root = s.append("root", vec![], None).expect("root");
// Parent grows a child (no resource dep of its own) and finishes its let c1 = s.append("c1", vec![], Some(root)).expect("c1");
// OWN work — but its subtree is not terminal, so it keeps the slot. let c2 = s.append("c2", vec![], Some(root)).expect("c2");
let child = s.append("child", vec![], Some(parent)).expect("child"); assert_eq!(s.settle(), vec![root]);
s.complete(parent, Outcome::Done); // Children can't start yet — parent still Running (logic not done).
assert!( assert!(s.settle().is_empty(), "children gated on parent logic");
s.resources.with(|t| t.available(&res("build-slot")) == 0), s.complete(root, Outcome::Done);
"parent must keep its lock while a child is still pending/running" 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] #[test]
fn recursive_lock_serializes_re_entrant_descendants() { fn failed_child_rolls_parent_up_to_failed() {
// `agent/foo` is unconfigured → default capacity 1. let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
let mut s = Scheduler::new(Graph::new(), ResourceTable::new()); let root = s.append("root", vec![], None).expect("root");
let agent = res("agent/foo"); let child = s.append("child", vec![], Some(root)).expect("child");
// A group node owns agent/foo and runs. assert_eq!(s.settle(), vec![root]);
let group = s s.complete(root, Outcome::Done);
.append("group", resource_dep("agent/foo"), None) assert_eq!(s.settle(), vec![child]);
.expect("group"); s.complete(child, Outcome::Failed);
assert_eq!(s.settle(), vec![group]); assert_eq!(
assert!(s.resources.with(|t| t.available(&agent) == 0)); s.graph().node(root).unwrap().state,
// Two sub-nodes each need agent/foo → they re-enter the group's lock, State::Failed,
// but only ONE at a time (the single re-entrancy slot). "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 let c1 = s
.append("c1", resource_dep("agent/foo"), Some(group)) .append("c1", res_dep("agent/foo"), Some(owner))
.expect("c1"); .expect("c1");
let c2 = s let c2 = s
.append("c2", resource_dep("agent/foo"), Some(group)) .append("c2", res_dep("agent/foo"), Some(owner))
.expect("c2"); .expect("c2");
let started = s.settle(); s.complete(owner, Outcome::Done); // → Finishing
assert_eq!( assert_eq!(s.settle(), vec![c1], "c1 borrows; c2 can't (cap 1)");
started,
vec![c1],
"only one descendant may borrow at a time"
);
assert_eq!(s.graph().node(c2).unwrap().state, State::Pending); 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); s.complete(c1, Outcome::Done);
assert_eq!(s.settle(), vec![c2]); assert_eq!(s.settle(), vec![c2], "borrow returned → c2 borrows");
assert_eq!(s.graph().node(c2).unwrap().state, State::Running); assert_eq!(avail(&s, "agent/foo"), 0, "still just the owner's unit");
// Still no double-acquire; the group's single unit is the only hold. }
assert!(s.resources.with(|t| t.available(&agent) == 0));
#[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] #[test]
fn failed_after_ok_dep_cancels_dependents_but_after_any_still_runs() { 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 mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
let root = s.append("root", vec![], None).expect("root"); let root = s.append("root", vec![], None).expect("root");
let strong1 = s let strong1 = s
.append( .append("strong1", vec![after_ok(root)], None)
"strong1",
vec![Dep::Node {
id: root,
when: DepWhen::AfterOk,
}],
None,
)
.expect("strong1"); .expect("strong1");
let strong2 = s let strong2 = s
.append( .append("strong2", vec![after_ok(strong1)], None)
"strong2",
vec![Dep::Node {
id: strong1,
when: DepWhen::AfterOk,
}],
None,
)
.expect("strong2"); .expect("strong2");
let weak = s let weak = s
.append( .append(
@ -420,11 +690,47 @@ mod tests {
.expect("weak"); .expect("weak");
assert_eq!(s.settle(), vec![root]); assert_eq!(s.settle(), vec![root]);
s.complete(root, Outcome::Failed); 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(strong1).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(strong2).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]); 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)));
}
} }