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

Compare commits

base: vinzenz:f64ab47de00e0d2cfe46f121a22a8a486cc24b4b
Branches Tags
vinzenz:main
..
compare: vinzenz:06bc7e31c029fc060eef1002022cb54b64e52a95
Branches Tags
vinzenz:main
4 changed files with 71 additions and 638 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());
}
}

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

@ -7,11 +7,11 @@
//! inserts a self-contained **node group** and returns its id; the scheduler //! inserts a self-contained **node group** and returns its id; the scheduler
//! runs a continuous loop, starting every node whose [`Dep`]s are satisfied: //! 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 ([`ResourceName`]):
//! type `R` (a `String` or an enum): `build-slot` (cap N), `agent/<name>` //! `build-slot` (capacity N), `agent/<name>` (capacity 1), or any name
//! (cap 1), or any name (cap 1, created on use). A node acquires *all* its //! (capacity 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, so no deadlock
//! so no deadlock and no cycle detection needed. //! and no cycle detection needed.
//! - **Node** deps wait on a node/group 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.
//! //!
@ -26,12 +26,10 @@
//! node kind. Resources are held by the acquiring node and released on //! node kind. Resources are held by the acquiring node and released on
//! completion via guard objects, recursive within a group. //! completion via guard objects, recursive within a group.
//! //!
//! The [`scheduler`] settle loop drives execution; the resource machinery //! The guards and the scheduler loop are follow-ups; the named-counter resource
//! lives in [`resources`] and the RAII lock guards over it in `guard`. //! machinery lives in [`resources`], the base this data model builds on.
pub(crate) mod guard;
pub mod resources; pub mod resources;
pub mod scheduler;
/// Opaque, stable, monotonic node identifier. /// Opaque, stable, monotonic node identifier.
/// ///
@ -49,6 +47,16 @@ pub mod scheduler;
)] )]
pub struct NodeId(pub(crate) u64); pub struct NodeId(pub(crate) u64);
/// A named counting semaphore.
///
/// Examples: `build-slot` (capacity configured to the number of build slots),
/// `agent/<name>` (capacity 1 — the per-agent lifecycle lock), or any other
/// name, which is assumed to have capacity 1 and is created on first use.
#[derive(
Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
)]
pub struct ResourceName(pub String);
/// 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)]
@ -79,7 +87,7 @@ impl DepWhen {
/// edge it names is satisfied (per its [`DepWhen`]) *and* every [`Dep::Resource`] /// edge it names is satisfied (per its [`DepWhen`]) *and* every [`Dep::Resource`]
/// 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 {
/// Depend on another node (or a group, by its group node's id). Whether a /// 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` /// *failed* dependency satisfies the edge is decided by `when`: `AfterOk`
/// requires success (and cancels this node if the dep fails), `AfterAny` /// requires success (and cancels this node if the dep fails), `AfterAny`
@ -95,7 +103,7 @@ pub enum Dep<R> {
/// released when the node completes. /// released when the node completes.
Resource { Resource {
/// The resource to acquire. /// The resource to acquire.
name: R, name: ResourceName,
/// How many units to hold (usually 1). /// How many units to hold (usually 1).
count: u32, count: u32,
}, },
@ -133,7 +141,7 @@ impl State {
/// 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.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Node<N, R> { pub struct Node<N> {
/// 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 /// The group this node belongs to, if any. `None` for a top-level group
@ -142,7 +150,7 @@ pub struct Node<N, R> {
/// Caller-defined payload (the node's kind / work description). /// Caller-defined payload (the node's kind / work description).
pub payload: N, pub payload: N,
/// What must hold before this node runs (other nodes + resources). /// What must hold before this node runs (other nodes + resources).
pub deps: Vec<Dep<R>>, pub deps: Vec<Dep>,
/// Lifecycle state. /// Lifecycle state.
pub state: State, pub state: State,
} }
@ -178,15 +186,9 @@ pub enum GraphError {
/// walks this graph filling open slots. Completed groups are retained (no /// walks this graph filling open slots. Completed groups 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>")]
try_from = "GraphData<N, R>", pub struct Graph<N> {
bound( nodes: Vec<Node<N>>,
serialize = "N: serde::Serialize, R: serde::Serialize",
deserialize = "N: serde::Deserialize<'de>, R: serde::Deserialize<'de>"
)
)]
pub struct Graph<N, R> {
nodes: Vec<Node<N, R>>,
next_id: u64, next_id: u64,
} }
@ -194,16 +196,15 @@ pub struct Graph<N, R> {
// below — which runs [`Graph::validate`], so a loaded graph can never carry a // below — which runs [`Graph::validate`], so a loaded graph can never carry a
// dangling id reference (Serialize does not validate; Deserialize always does). // dangling id reference (Serialize does not validate; Deserialize always does).
#[derive(serde::Deserialize)] #[derive(serde::Deserialize)]
#[serde(bound(deserialize = "N: serde::Deserialize<'de>, R: serde::Deserialize<'de>"))] struct GraphData<N> {
struct GraphData<N, R> { nodes: Vec<Node<N>>,
nodes: Vec<Node<N, R>>,
next_id: u64, next_id: u64,
} }
impl<N, R> TryFrom<GraphData<N, R>> for Graph<N, R> { impl<N> TryFrom<GraphData<N>> for Graph<N> {
type Error = GraphError; type Error = GraphError;
fn try_from(data: GraphData<N, R>) -> Result<Self, Self::Error> { fn try_from(data: GraphData<N>) -> Result<Self, Self::Error> {
let graph = Graph { let graph = Graph {
nodes: data.nodes, nodes: data.nodes,
next_id: data.next_id, next_id: data.next_id,
@ -215,13 +216,13 @@ impl<N, R> TryFrom<GraphData<N, R>> for Graph<N, R> {
// A `derive(Default)` would wrongly require `N: Default` (an empty graph holds // A `derive(Default)` would wrongly require `N: Default` (an empty graph holds
// no payload); an empty `Vec<Node<N>>` needs no such bound, so impl it directly. // no payload); an empty `Vec<Node<N>>` needs no such bound, so impl it directly.
impl<N, R> Default for Graph<N, R> { impl<N> Default for Graph<N> {
fn default() -> Self { fn default() -> Self {
Self::new() Self::new()
} }
} }
impl<N, R> Graph<N, R> { impl<N> Graph<N> {
/// An empty graph. /// An empty graph.
#[must_use] #[must_use]
pub fn new() -> Self { pub fn new() -> Self {
@ -252,7 +253,7 @@ impl<N, R> Graph<N, R> {
pub fn insert( pub fn insert(
&mut self, &mut self,
payload: N, payload: N,
deps: Vec<Dep<R>>, deps: Vec<Dep>,
parent: Option<NodeId>, parent: Option<NodeId>,
) -> Result<NodeId, GraphError> { ) -> Result<NodeId, GraphError> {
if let Some(parent_id) = parent if let Some(parent_id) = parent
@ -280,33 +281,15 @@ impl<N, R> Graph<N, R> {
/// Borrow a node by id. /// Borrow a node by id.
#[must_use] #[must_use]
pub fn node(&self, id: NodeId) -> Option<&Node<N, R>> { pub fn node(&self, id: NodeId) -> Option<&Node<N>> {
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`). /// The direct children of a group node (nodes whose `parent` is `id`).
pub fn children(&self, id: NodeId) -> impl Iterator<Item = &Node<N, R>> { pub fn children(&self, id: NodeId) -> impl Iterator<Item = &Node<N>> {
self.nodes.iter().filter(move |n| n.parent == Some(id)) 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<Item = &Node<N, R>> {
self.nodes.iter()
}
/// Set a node's lifecycle state, returning `false` for an unknown id. The
/// scheduler drives every state transition — nothing else mutates state,
/// which is what keeps the resource guards + terminality in sync.
pub(crate) fn set_state(&mut self, id: NodeId, state: State) -> bool {
if let Some(node) = self.nodes.iter_mut().find(|n| n.id == id) {
node.state = state;
true
} else {
false
}
}
/// A group is terminal once the group node itself is terminal *and* every /// 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: /// node inside it (recursively) is terminal. The node's own state matters:
/// a group node still `Pending`/`Running` is not terminal even with no /// a group node still `Pending`/`Running` is not terminal even with no
@ -362,7 +345,7 @@ mod tests {
#[test] #[test]
fn insert_mints_stable_monotonic_ids() { fn insert_mints_stable_monotonic_ids() {
let mut g: Graph<&str, String> = Graph::new(); let mut g: Graph<&str> = Graph::new();
let a = g.insert("sweep", vec![], None).unwrap(); let a = g.insert("sweep", vec![], None).unwrap();
let b = g let b = g
.insert( .insert(
@ -381,19 +364,24 @@ mod tests {
assert_eq!(g.node(a).unwrap().parent, None); assert_eq!(g.node(a).unwrap().parent, None);
} }
fn set_state<N>(g: &mut Graph<N>, id: NodeId, state: State) {
let idx = g.nodes.iter().position(|n| n.id == id).unwrap();
g.nodes[idx].state = state;
}
#[test] #[test]
fn group_terminal_requires_the_group_node_and_all_children_terminal() { fn group_terminal_requires_the_group_node_and_all_children_terminal() {
let mut g: Graph<&str, String> = Graph::new(); let mut g: Graph<&str> = Graph::new();
let group = g.insert("group", vec![], None).unwrap(); let group = g.insert("group", vec![], None).unwrap();
let child = g.insert("child", vec![], Some(group)).unwrap(); let child = g.insert("child", vec![], Some(group)).unwrap();
// Both pending → not terminal. // Both pending → not terminal.
assert!(!g.group_terminal(group)); assert!(!g.group_terminal(group));
// Child done, but the group node itself is still pending → NOT terminal: // 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. // the group node's own state is load-bearing, not just its children.
g.set_state(child, State::Done); set_state(&mut g, child, State::Done);
assert!(!g.group_terminal(group)); assert!(!g.group_terminal(group));
// Group node terminal too → the whole group is terminal. // Group node terminal too → the whole group is terminal.
g.set_state(group, State::Done); set_state(&mut g, group, State::Done);
assert!(g.group_terminal(group)); assert!(g.group_terminal(group));
} }
@ -401,12 +389,12 @@ mod tests {
fn empty_running_group_is_not_terminal() { fn empty_running_group_is_not_terminal() {
// A running node with no children yet may still append some, so it must // 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. // not read as terminal just because its child set is currently empty.
let mut g: Graph<&str, String> = Graph::new(); let mut g: Graph<&str> = Graph::new();
let group = g.insert("group", vec![], None).unwrap(); let group = g.insert("group", vec![], None).unwrap();
g.set_state(group, State::Running); set_state(&mut g, group, State::Running);
assert!(!g.group_terminal(group)); assert!(!g.group_terminal(group));
// Once it finishes (having grown no children), it is terminal. // Once it finishes (having grown no children), it is terminal.
g.set_state(group, State::Done); set_state(&mut g, group, State::Done);
assert!(g.group_terminal(group)); assert!(g.group_terminal(group));
} }
@ -436,7 +424,7 @@ mod tests {
#[test] #[test]
fn insert_rejects_unknown_parent() { fn insert_rejects_unknown_parent() {
let mut g: Graph<&str, String> = Graph::new(); let mut g: Graph<&str> = Graph::new();
let bogus = NodeId(7); let bogus = NodeId(7);
assert_eq!( assert_eq!(
g.insert("x", vec![], Some(bogus)).unwrap_err(), g.insert("x", vec![], Some(bogus)).unwrap_err(),
@ -446,7 +434,7 @@ mod tests {
#[test] #[test]
fn insert_rejects_unknown_dep() { fn insert_rejects_unknown_dep() {
let mut g: Graph<&str, String> = Graph::new(); let mut g: Graph<&str> = Graph::new();
let bogus = NodeId(42); let bogus = NodeId(42);
let deps = vec![Dep::Node { let deps = vec![Dep::Node {
id: bogus, id: bogus,
@ -460,7 +448,7 @@ mod tests {
#[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> = Graph::new();
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(),
@ -472,7 +460,7 @@ mod tests {
) )
.unwrap(); .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> = 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");
} }
@ -481,7 +469,7 @@ mod tests {
fn deserialize_rejects_a_dangling_dependency() { fn deserialize_rejects_a_dangling_dependency() {
// Build a graph whose only node depends on a non-existent id, serialize // Build a graph whose only node depends on a non-existent id, serialize
// it (Serialize does not validate), and confirm deserialize rejects it. // it (Serialize does not validate), and confirm deserialize rejects it.
let bad = Graph::<String, String> { let bad = Graph::<String> {
nodes: vec![Node { nodes: vec![Node {
id: NodeId(0), id: NodeId(0),
parent: None, parent: None,
@ -495,13 +483,13 @@ mod tests {
next_id: 1, next_id: 1,
}; };
let json = serde_json::to_string(&bad).unwrap(); let json = serde_json::to_string(&bad).unwrap();
let err = serde_json::from_str::<Graph<String, String>>(&json).unwrap_err(); let err = serde_json::from_str::<Graph<String>>(&json).unwrap_err();
assert!(err.to_string().contains("unknown node")); assert!(err.to_string().contains("unknown node"));
} }
#[test] #[test]
fn validate_rejects_next_id_that_would_remint() { fn validate_rejects_next_id_that_would_remint() {
let bad = Graph::<&str, String> { let bad = Graph::<&str> {
nodes: vec![Node { nodes: vec![Node {
id: NodeId(5), id: NodeId(5),
parent: None, parent: None,

38
hive-jobq/src/resources.rs
View file

@ -20,8 +20,8 @@
//! //!
//! [`Dep::Resource`]: crate::Dep::Resource //! [`Dep::Resource`]: crate::Dep::Resource
use crate::ResourceName;
use std::collections::HashMap; use std::collections::HashMap;
use std::hash::Hash;
/// A set of named counting semaphores. /// A set of named counting semaphores.
/// ///
@ -30,22 +30,22 @@ use std::hash::Hash;
/// units atomically via [`ResourceTable::try_acquire_all`] / /// units atomically via [`ResourceTable::try_acquire_all`] /
/// [`ResourceTable::release_all`]. /// [`ResourceTable::release_all`].
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct ResourceTable<R> { pub struct ResourceTable {
/// Configured capacities, keyed by resource. Missing ⇒ `default_capacity`. /// Configured capacities, keyed by name. Missing ⇒ `default_capacity`.
capacities: HashMap<R, u32>, capacities: HashMap<ResourceName, u32>,
/// Units currently held, keyed by resource. Missing ⇒ 0. /// Units currently held, keyed by name. Missing ⇒ 0.
held: HashMap<R, u32>, held: HashMap<ResourceName, u32>,
/// Capacity assumed for a resource with no configured entry. /// Capacity assumed for a name with no configured entry.
default_capacity: u32, default_capacity: u32,
} }
impl<R: Clone + Eq + Hash> Default for ResourceTable<R> { impl Default for ResourceTable {
fn default() -> Self { fn default() -> Self {
Self::new() Self::new()
} }
} }
impl<R: Clone + Eq + Hash> ResourceTable<R> { impl ResourceTable {
/// An empty table whose unconfigured names default to capacity 1. /// An empty table whose unconfigured names default to capacity 1.
#[must_use] #[must_use]
pub fn new() -> Self { pub fn new() -> Self {
@ -61,14 +61,14 @@ impl<R: Clone + Eq + Hash> ResourceTable<R> {
/// Overwrites any previous capacity for that name. Lowering capacity below /// Overwrites any previous capacity for that name. Lowering capacity below
/// the currently-held count is allowed — the table simply reports zero /// the currently-held count is allowed — the table simply reports zero
/// available until enough is released; it never rejects a config change. /// available until enough is released; it never rejects a config change.
pub fn set_capacity(&mut self, name: R, capacity: u32) { pub fn set_capacity(&mut self, name: ResourceName, capacity: u32) {
self.capacities.insert(name, capacity); self.capacities.insert(name, capacity);
} }
/// The capacity of a name — its configured value, or the default (1) if it /// The capacity of a name — its configured value, or the default (1) if it
/// was never configured. /// was never configured.
#[must_use] #[must_use]
pub fn capacity(&self, name: &R) -> u32 { pub fn capacity(&self, name: &ResourceName) -> u32 {
self.capacities self.capacities
.get(name) .get(name)
.copied() .copied()
@ -77,13 +77,13 @@ impl<R: Clone + Eq + Hash> ResourceTable<R> {
/// Units of a name currently held (0 if none). /// Units of a name currently held (0 if none).
#[must_use] #[must_use]
pub fn held(&self, name: &R) -> u32 { pub fn held(&self, name: &ResourceName) -> u32 {
self.held.get(name).copied().unwrap_or(0) self.held.get(name).copied().unwrap_or(0)
} }
/// Units of a name available to acquire right now (`capacity - held`). /// Units of a name available to acquire right now (`capacity - held`).
#[must_use] #[must_use]
pub fn available(&self, name: &R) -> u32 { pub fn available(&self, name: &ResourceName) -> u32 {
self.capacity(name).saturating_sub(self.held(name)) self.capacity(name).saturating_sub(self.held(name))
} }
@ -95,7 +95,7 @@ impl<R: Clone + Eq + Hash> ResourceTable<R> {
/// `false` and leaves the table completely untouched. A request for more /// `false` and leaves the table completely untouched. A request for more
/// units than a name's capacity can therefore never succeed — the scheduler /// units than a name's capacity can therefore never succeed — the scheduler
/// should reject such a node at insert time so it does not wait forever. /// should reject such a node at insert time so it does not wait forever.
pub(crate) fn try_acquire_all(&mut self, reqs: &[(R, u32)]) -> bool { pub fn try_acquire_all(&mut self, reqs: &[(ResourceName, u32)]) -> bool {
let wanted = aggregate(reqs); let wanted = aggregate(reqs);
// All-or-nothing: bail before mutating if any request cannot be met. // All-or-nothing: bail before mutating if any request cannot be met.
for (name, &count) in &wanted { for (name, &count) in &wanted {
@ -113,7 +113,7 @@ impl<R: Clone + Eq + Hash> ResourceTable<R> {
/// ///
/// Duplicate names are summed. Releasing more than is held saturates at zero /// Duplicate names are summed. Releasing more than is held saturates at zero
/// rather than underflowing, so a double release is harmless. /// rather than underflowing, so a double release is harmless.
pub(crate) fn release_all(&mut self, reqs: &[(R, u32)]) { pub fn release_all(&mut self, reqs: &[(ResourceName, u32)]) {
for (name, count) in aggregate(reqs) { for (name, count) in aggregate(reqs) {
if let Some(h) = self.held.get_mut(name) { if let Some(h) = self.held.get_mut(name) {
*h = h.saturating_sub(count); *h = h.saturating_sub(count);
@ -123,8 +123,8 @@ impl<R: Clone + Eq + Hash> ResourceTable<R> {
} }
/// Sum a request list into per-name totals so duplicate names are one entry. /// Sum a request list into per-name totals so duplicate names are one entry.
fn aggregate<R: Eq + Hash>(reqs: &[(R, u32)]) -> HashMap<&R, u32> { fn aggregate(reqs: &[(ResourceName, u32)]) -> HashMap<&ResourceName, u32> {
let mut wanted: HashMap<&R, u32> = HashMap::new(); let mut wanted: HashMap<&ResourceName, u32> = HashMap::new();
for (name, count) in reqs { for (name, count) in reqs {
*wanted.entry(name).or_insert(0) += *count; *wanted.entry(name).or_insert(0) += *count;
} }
@ -135,8 +135,8 @@ fn aggregate<R: Eq + Hash>(reqs: &[(R, u32)]) -> HashMap<&R, u32> {
mod tests { mod tests {
use super::*; use super::*;
fn res(name: &str) -> String { fn res(name: &str) -> ResourceName {
name.to_owned() ResourceName(name.to_owned())
} }
#[test] #[test]

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

@ -1,425 +0,0 @@
//! The settle loop — drives a [`Graph`] to completion over the resource pool.
//!
//! [`Scheduler::settle`] claims every currently-runnable pending node (its
//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired
//! atomically), marks it `Running`, holds its resource guards, and returns the
//! newly-started ids for the caller's runner to execute. The runner reports each
//! node's result back with [`Scheduler::complete`]; a running node may grow its
//! own sub-group first via [`Scheduler::append`]. Concurrency is emergent from
//! resource capacity — there is no separate active-node cap.
//!
//! A resource is held for the acquiring node's *entire subtree* lifetime: the
//! owned guard is released only when that node and every descendant is terminal
//! (`group_terminal`), not when the node's own work finishes. Single-owner and
//! synchronous — the caller drives `settle` / `complete`; no async or locking
//! lives here (that's the runner's job, one layer up).
//!
//! Recursive-lock re-entrancy (a sub-node reusing an ancestor group's lock) and
//! the eager `AfterOk` failure cascade are layered on top of this owned core.
use std::collections::HashMap;
use std::hash::Hash;
use crate::guard::{ResourceGuard, SharedResources};
use crate::resources::ResourceTable;
use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State};
/// The result of a node's own execution, reported to [`Scheduler::complete`].
///
/// `Cancelled` is not an outcome a runner reports — it is scheduler-driven (an
/// `AfterOk` dependency failed), so a runner only ever says `Done` or `Failed`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Outcome {
/// The node's work succeeded.
Done,
/// The node's work failed.
Failed,
}
/// Drives a [`Graph`] over a shared resource pool: claim runnable nodes, hold
/// their resources for the subtree's lifetime, release on subtree-terminal.
pub struct Scheduler<N, R: Clone + Eq + Hash> {
graph: Graph<N, R>,
resources: SharedResources<R>,
/// Owned resource guards, keyed by the node that acquired them. Dropped
/// (releasing the units) when that node's whole subtree is terminal.
owned: HashMap<NodeId, Vec<ResourceGuard<R>>>,
/// 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>,
}
impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// A scheduler over `graph` with `resources` as the capacity pool.
#[must_use]
pub fn new(graph: Graph<N, R>, resources: ResourceTable<R>) -> Self {
Self {
graph,
resources: SharedResources::new(resources),
owned: HashMap::new(),
borrow_slots: HashMap::new(),
}
}
/// The graph, for inspection (state, hierarchy, UI rendering).
#[must_use]
pub fn graph(&self) -> &Graph<N, R> {
&self.graph
}
/// Append a node — e.g. a running node growing its own sub-group. Delegates
/// to [`Graph::insert`]; call [`Scheduler::settle`] afterwards to start it
/// once it is runnable.
///
/// # Errors
/// Propagates [`GraphError`] for a dangling parent or dependency id.
pub fn append(
&mut self,
payload: N,
deps: Vec<Dep<R>>,
parent: Option<NodeId>,
) -> Result<NodeId, GraphError> {
self.graph.insert(payload, deps, parent)
}
/// Claim every currently-runnable pending node and start it: node-deps
/// satisfied and all resource-deps acquired atomically (all-or-nothing).
/// Each claimed node is marked `Running`, its owned guards held, and its id
/// returned for the runner to execute. A single pass suffices — a node
/// started here is `Running`, not terminal, so it cannot satisfy another
/// node's dependency in the same pass; it only consumes resources.
#[must_use]
pub fn settle(&mut self) -> Vec<NodeId> {
let pending: Vec<NodeId> = self
.graph
.nodes()
.filter(|n| n.state == State::Pending)
.map(|n| n.id)
.collect();
let mut started = Vec::new();
for id in pending {
if self.node_deps_satisfied(id) && self.try_start(id) {
started.push(id);
}
}
started
}
/// Try to start node `id`: classify each resource dep as *owned* (no
/// ancestor holds it → acquire real units) or *borrowed* (an ancestor group
/// already holds it → re-enter, gated by the one re-entrancy slot), then
/// take everything atomically or nothing. Returns whether it started.
fn try_start(&mut self, id: NodeId) -> bool {
let mut owned_reqs = Vec::new();
let mut borrows = Vec::new();
for (name, count) in self.resource_reqs(id) {
if let Some(ancestor) = self.ancestor_owning(id, &name) {
// Re-entrant reuse: allowed only if the slot is free.
if self.borrow_slots.contains_key(&(ancestor, name.clone())) {
return false;
}
borrows.push((ancestor, name));
} else {
owned_reqs.push((name, count));
}
}
// Owned units are all-or-nothing; borrow slots were all confirmed free
// above, so this is the only fallible step. Nothing mutated until here.
let Some(guard) = self.resources.acquire(owned_reqs) else {
return false;
};
self.owned.entry(id).or_default().push(guard);
for slot in borrows {
self.borrow_slots.insert(slot, id);
}
self.graph.set_state(id, State::Running);
true
}
/// The nearest ancestor of `id` that *owns* (holds real units of) `name`,
/// or `None` if no ancestor holds it (⇒ `id` must own-acquire it itself).
fn ancestor_owning(&self, id: NodeId, name: &R) -> Option<NodeId> {
let mut cursor = self.graph.node(id)?.parent;
while let Some(ancestor) = cursor {
if self.node_owns(ancestor, name) {
return Some(ancestor);
}
cursor = self.graph.node(ancestor)?.parent;
}
None
}
/// Whether node `holder` holds an owned guard covering resource `name`.
fn node_owns(&self, holder: NodeId, name: &R) -> bool {
self.owned.get(&holder).is_some_and(|guards| {
guards
.iter()
.any(|g| g.held().iter().any(|(n, _)| n == name))
})
}
/// Report a running node's own execution result. Sets its state, then
/// releases the owned guards of every node whose whole subtree has become
/// terminal — a parent keeps its lock until its last descendant finishes.
/// Call [`Scheduler::settle`] again afterwards to start newly-unblocked work.
pub fn complete(&mut self, id: NodeId, outcome: Outcome) {
let state = match outcome {
Outcome::Done => State::Done,
Outcome::Failed => State::Failed,
};
self.graph.set_state(id, state);
if outcome == Outcome::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<NodeId> = self
.graph
.nodes()
.filter(|n| {
n.state == State::Pending
&& n.deps.iter().any(
|d| matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == dep),
)
})
.map(|n| n.id)
.collect();
for d in dependents {
self.graph.set_state(d, State::Cancelled);
stack.push(d);
}
}
}
/// Release everything whose subtree has become terminal: drop the owned
/// guards of any holder (→ frees its units) and free any re-entrancy slot
/// held by a borrower — both are held for the whole subtree lifetime.
fn release_settled_subtrees(&mut self) {
let holders: Vec<NodeId> = self.owned.keys().copied().collect();
for holder in holders {
if self.graph.group_terminal(holder) {
self.owned.remove(&holder); // drops guards → releases the units
}
}
self.borrow_slots
.retain(|_, borrower| !self.graph.group_terminal(*borrower));
}
/// Whether every [`Dep::Node`] edge of `id` is satisfied. `Dep::Resource`
/// edges are handled by the atomic acquire in [`Scheduler::settle`], not here.
fn node_deps_satisfied(&self, id: NodeId) -> bool {
let Some(node) = self.graph.node(id) else {
return false;
};
node.deps.iter().all(|dep| match dep {
Dep::Resource { .. } => true,
Dep::Node { id, when } => self.dep_node_satisfied(*id, *when),
})
}
/// Whether a node/group dependency `id` satisfies edge kind `when`. A group
/// is depended on as a whole: `AfterAny` needs its subtree terminal (any
/// outcome), `AfterOk` needs its whole subtree to have succeeded.
fn dep_node_satisfied(&self, id: NodeId, when: DepWhen) -> bool {
match when {
DepWhen::AfterAny => self.graph.group_terminal(id),
DepWhen::AfterOk => self.subtree_all_done(id),
}
}
/// Whether `id` and every descendant reached [`State::Done`] — the success
/// condition for an `AfterOk` edge onto a (possibly group) node.
fn subtree_all_done(&self, id: NodeId) -> bool {
let Some(node) = self.graph.node(id) else {
return false;
};
node.state == State::Done && self.graph.children(id).all(|c| self.subtree_all_done(c.id))
}
/// The `(name, count)` resource units `id` must hold to run.
fn resource_reqs(&self, id: NodeId) -> Vec<(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()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn res(name: &str) -> String {
name.to_owned()
}
/// A graph + a resource table with `build-slot` set to `slots`.
fn scheduler_with_slots(slots: u32) -> Scheduler<&'static str, String> {
let mut table = ResourceTable::new();
table.set_capacity(res("build-slot"), slots);
Scheduler::new(Graph::new(), table)
}
fn slot_dep() -> Vec<Dep<String>> {
vec![Dep::Resource {
name: res("build-slot"),
count: 1,
}]
}
fn resource_dep(name: &str) -> Vec<Dep<String>> {
vec![Dep::Resource {
name: res(name),
count: 1,
}]
}
#[test]
fn resource_node_starts_then_releases_on_complete() {
let mut s = scheduler_with_slots(1);
let n = s.append("build", slot_dep(), None).expect("insert");
// settle claims it (a slot is free) and marks it Running.
assert_eq!(s.settle(), vec![n]);
assert_eq!(s.graph().node(n).unwrap().state, State::Running);
// Slot is held.
assert!(s.resources.with(|t| t.available(&res("build-slot")) == 0));
// Completing it releases the slot (subtree is just this node).
s.complete(n, Outcome::Done);
assert_eq!(s.graph().node(n).unwrap().state, State::Done);
assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1));
}
#[test]
fn build_slot_cap_limits_concurrency_and_release_unblocks() {
let mut s = scheduler_with_slots(2);
let a = s.append("a", slot_dep(), None).expect("a");
let b = s.append("b", slot_dep(), None).expect("b");
let c = s.append("c", slot_dep(), None).expect("c");
// cap 2 → a + b start, c blocks on the exhausted slot.
let started = s.settle();
assert_eq!(started, vec![a, b]);
assert_eq!(s.graph().node(c).unwrap().state, State::Pending);
// a finishes → its slot frees → c can now start.
s.complete(a, Outcome::Done);
assert_eq!(s.settle(), vec![c]);
assert_eq!(s.graph().node(c).unwrap().state, State::Running);
}
#[test]
fn parent_holds_resource_until_child_subtree_done() {
let mut s = scheduler_with_slots(1);
// Parent grabs the single build-slot and runs.
let parent = s.append("parent", slot_dep(), None).expect("parent");
assert_eq!(s.settle(), vec![parent]);
// Parent grows a child (no resource dep of its own) and finishes its
// OWN work — but its subtree is not terminal, so it keeps the slot.
let child = s.append("child", vec![], Some(parent)).expect("child");
s.complete(parent, Outcome::Done);
assert!(
s.resources.with(|t| t.available(&res("build-slot")) == 0),
"parent must keep its lock while a child is still pending/running"
);
// The child starts and completes → now the whole subtree is terminal →
// the parent's slot is released exactly once.
assert_eq!(s.settle(), vec![child]);
s.complete(child, Outcome::Done);
assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1));
}
#[test]
fn recursive_lock_serializes_re_entrant_descendants() {
// `agent/foo` is unconfigured → default capacity 1.
let mut s = Scheduler::new(Graph::new(), ResourceTable::new());
let agent = res("agent/foo");
// A group node owns agent/foo and runs.
let group = s
.append("group", resource_dep("agent/foo"), None)
.expect("group");
assert_eq!(s.settle(), vec![group]);
assert!(s.resources.with(|t| t.available(&agent) == 0));
// Two sub-nodes each need agent/foo → they re-enter the group's lock,
// but only ONE at a time (the single re-entrancy slot).
let c1 = s
.append("c1", resource_dep("agent/foo"), Some(group))
.expect("c1");
let c2 = s
.append("c2", resource_dep("agent/foo"), Some(group))
.expect("c2");
let started = s.settle();
assert_eq!(
started,
vec![c1],
"only one descendant may borrow at a time"
);
assert_eq!(s.graph().node(c2).unwrap().state, State::Pending);
// The lock was NOT re-acquired — still just the group's one unit held.
assert!(s.resources.with(|t| t.available(&agent) == 0));
// c1 finishes → its borrow slot frees → c2 can now re-enter.
s.complete(c1, Outcome::Done);
assert_eq!(s.settle(), vec![c2]);
assert_eq!(s.graph().node(c2).unwrap().state, State::Running);
// Still no double-acquire; the group's single unit is the only hold.
assert!(s.resources.with(|t| t.available(&agent) == 0));
}
#[test]
fn failed_after_ok_dep_cancels_dependents_but_after_any_still_runs() {
let mut s: Scheduler<&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,
)
.expect("strong1");
let strong2 = s
.append(
"strong2",
vec![Dep::Node {
id: strong1,
when: DepWhen::AfterOk,
}],
None,
)
.expect("strong2");
let weak = s
.append(
"weak",
vec![Dep::Node {
id: root,
when: DepWhen::AfterAny,
}],
None,
)
.expect("weak");
assert_eq!(s.settle(), vec![root]);
s.complete(root, Outcome::Failed);
// The AfterOk chain strong1→strong2 is eagerly cancelled (a strong dep
// failed)…
assert_eq!(s.graph().node(strong1).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled);
// …but the AfterAny dependent still runs — it converges regardless.
assert_eq!(s.settle(), vec![weak]);
}
}