a3a0b34668
`submit` used to complete the container node by hand, right after inserting it, so it would park in `Finishing` and its children unblock. That was the last caller of `Scheduler::complete` outside the crate, and the justification was that the container "never needs claiming or executing". It does, though, in the sense that matters: it is a node with no logic of its own, and the scheduler already knows what to do with one. It declares no resources, so it is claimable the moment it is inserted; `run_node`'s `Dag` arm already returns `Ok(())`, exactly as it does for `DeployWindow`, which is the same shape and was never special-cased. Deleting the inline completion costs one claim round-trip and removes the only reason the crate had to expose completion at all. `complete` is `pub(crate)` now. Completion is reachable only from inside the future `claim_next` hands back, so a node cannot be finished without the claim it answers, and cannot be claimed without the future that finishes it. That was the point of the seam.
1330 lines
59 KiB
Rust
1330 lines
59 KiB
Rust
//! The settle loop — drives a [`Graph`] to completion over a resource pool the
|
|
//! scheduler owns directly.
|
|
//!
|
|
//! [`Scheduler::claim_next`] claims one currently-runnable pending node (its
|
|
//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired
|
|
//! atomically), marks it `Running`, records the units it holds, and hands back
|
|
//! a future that executes the node **and completes it**, so "forgot to finish
|
|
//! the node" is not expressible. One at a time is the primitive on purpose: it
|
|
//! lets the caller choose between claiming again and backing off, which a batch
|
|
//! return can't express. A running node may grow more work by declaring into
|
|
//! the builder it was handed. Concurrency is emergent from resource capacity.
|
|
//!
|
|
//! Single-threaded by design: the scheduler is the only driver, holds the
|
|
//! [`ResourceTable`] as a plain owned field, mutating it through `&mut self` —
|
|
//! no interior mutability, no guard objects.
|
|
//!
|
|
//! Completion rolls up the parent tree: a node with children parks in
|
|
//! [`State::Finishing`] until they finish; a child is gated on its parent there.
|
|
//!
|
|
//! Resource holding follows the [`crate::Node::parent`] tree, not the deps.
|
|
//! For each resource a node needs, the scheduler walks its parent-ancestors:
|
|
//! no ancestor holds it → acquire fresh units (this node *owns* them, held for
|
|
//! its whole subtree); an ancestor owns it but its grant is lent to a different
|
|
//! branch → acquire an additional unit if one is free, else wait; an ancestor
|
|
//! owns it and the grant is free (or lent to a branch this node is inside) →
|
|
//! *borrow* it, no new unit. A grant is lent to one branch at a time; nodes
|
|
//! inside a branch are covered by its borrow (dep-sequenced, so no concurrent
|
|
//! work under a cap-1 lease). An owner's unit releases only once the owner and
|
|
//! its whole subtree are terminal. (Early release once no subtree node needs it
|
|
//! is a deferred optimization — unsafe under dynamically-appended subnodes.)
|
|
|
|
use std::collections::HashMap;
|
|
use std::future::Future;
|
|
use std::hash::Hash;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use crate::builder::{BuildError, JobBuilder, NodeGuid};
|
|
use crate::resources::ResourceTable;
|
|
use crate::{Dep, Graph, GraphError, NodeId, State, TerminalState};
|
|
|
|
/// 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, PartialEq, Eq)]
|
|
pub enum Outcome {
|
|
/// The node's work succeeded.
|
|
Done,
|
|
/// The node's work failed, carrying the failure reason — recorded on
|
|
/// [`crate::Node::error`] for the failed node itself (a node that rolls up
|
|
/// `Failed` from a child, or is cancelled, carries no error of its own).
|
|
Failed(String),
|
|
}
|
|
|
|
/// Drives a [`Graph`] over an owned resource pool: claim runnable nodes, record
|
|
/// the units each *owns*, track which node is currently *borrowing* each grant,
|
|
/// and release an owner's grant once its whole subtree is terminal.
|
|
pub struct Scheduler<N, R: Clone + Eq + Hash> {
|
|
graph: Graph<N, R>,
|
|
resources: ResourceTable<R>,
|
|
/// Fresh units each owner node acquired: `owner → [(resource, count)]`.
|
|
/// Recorded against the node that *acquired* the units (never a borrower);
|
|
/// released back to the table once the owner and its whole [`Node::parent`]
|
|
/// subtree are terminal.
|
|
owned: HashMap<NodeId, Vec<(R, u32)>>,
|
|
/// Which branch currently borrows a given owner's grant: `(owner, resource)
|
|
/// → branch-root node`. A grant is lent to one branch at a time; nodes
|
|
/// inside that branch are covered by it. Cleared when the branch leaves
|
|
/// (its subtree terminal), freeing the grant for a waiting sibling.
|
|
borrowed: HashMap<(NodeId, R), NodeId>,
|
|
}
|
|
|
|
impl<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,
|
|
owned: HashMap::new(),
|
|
borrowed: HashMap::new(),
|
|
}
|
|
}
|
|
|
|
/// The graph, for inspection (state, hierarchy, UI rendering).
|
|
#[must_use]
|
|
pub fn graph(&self) -> &Graph<N, R> {
|
|
&self.graph
|
|
}
|
|
|
|
/// Append a node under `parent` — e.g. a running node growing more work into
|
|
/// its own subtree. Delegates to [`Graph::insert`]; claim again afterwards
|
|
/// to start it once it is runnable.
|
|
///
|
|
/// # Errors
|
|
/// Propagates [`GraphError`] for a dangling dependency or parent id.
|
|
pub fn append(
|
|
&mut self,
|
|
payload: N,
|
|
deps: Vec<Dep<R>>,
|
|
parent: Option<NodeId>,
|
|
) -> Result<NodeId, GraphError> {
|
|
self.graph.insert(payload, deps, parent)
|
|
}
|
|
|
|
/// Insert a whole job under `root_parent`, returning the id each handle's
|
|
/// node was minted as.
|
|
///
|
|
/// `declare` receives a fresh [`JobBuilder`], names the job's nodes on it,
|
|
/// and returns the handles whose ids it wants back — they come back in
|
|
/// that order. The builder never leaves this call. That is the whole
|
|
/// insertion API — a caller cannot construct a builder, hold one, or
|
|
/// insert one itself, so there is no way to end up with a job-shaped value
|
|
/// being passed around as a spec.
|
|
///
|
|
/// The one insertion entry point for a job. Nodes go straight into
|
|
/// [`Graph::insert_unchecked`]: [`crate::builder::check_job_shape`] has
|
|
/// already decided every rejection the graph could raise, so re-validating
|
|
/// per node could only report a problem *after* the earlier nodes were
|
|
/// inserted. Claim again afterwards to start whatever became runnable.
|
|
///
|
|
/// **Atomic in the job's own shape.** A forward edge, a forward parent, or
|
|
/// a request for a handle this job never declared is rejected *before* the
|
|
/// first node is inserted, so a malformed job leaves the graph untouched
|
|
/// rather than half-built.
|
|
///
|
|
/// # Errors
|
|
/// Propagates [`BuildError`] — a forward reference in the job's own
|
|
/// declarations, a handle from a different job, or a graph rejection.
|
|
pub fn insert_job(
|
|
&mut self,
|
|
root_parent: Option<NodeId>,
|
|
declare: impl FnOnce(&JobBuilder<N, R>) -> Vec<NodeGuid>,
|
|
) -> Result<Vec<NodeId>, BuildError> {
|
|
let job = JobBuilder::new();
|
|
let wanted = declare(&job);
|
|
let graph = &mut self.graph;
|
|
job.insert_with(root_parent, &wanted, |payload, deps, parent| {
|
|
graph.insert_unchecked(payload, deps, parent)
|
|
})
|
|
}
|
|
|
|
/// Claim **one** currently-runnable pending node and start it: node-deps
|
|
/// satisfied and all resource-deps acquired atomically (all-or-nothing).
|
|
/// The node is marked `Running`, its acquired units recorded, and its id
|
|
/// returned for the caller to execute. `None` means nothing is runnable
|
|
/// right now — which is a different statement from "nothing is pending".
|
|
///
|
|
/// **Private**: [`Self::claim_next`] is the only way out of this crate.
|
|
/// Claiming without the future that completes the node is the sequence the
|
|
/// seam exists to make inexpressible, so the primitive stays in here.
|
|
#[must_use]
|
|
fn claim_one(&mut self) -> Option<NodeId> {
|
|
let pending: Vec<NodeId> = self
|
|
.graph
|
|
.nodes()
|
|
.filter(|n| n.state == State::Pending)
|
|
.map(|n| n.id)
|
|
.collect();
|
|
pending
|
|
.into_iter()
|
|
.find(|&id| self.node_deps_satisfied(id) && self.try_start(id))
|
|
}
|
|
|
|
/// Claim one runnable node and return **the work that runs it**, or `None`
|
|
/// when nothing is runnable right now.
|
|
///
|
|
/// This is the seam: the caller supplies how to execute a node and spawns
|
|
/// the returned future, but never touches claiming or completion. The
|
|
/// future runs the node **and completes it**, so "forgot to finish the
|
|
/// node" is not expressible — completion is inside the thing you spawn.
|
|
///
|
|
/// The `Option` is answered *synchronously*, before anything is awaited, so
|
|
/// the caller can decide "claim again immediately" vs "back off" without
|
|
/// waiting on the node it just started.
|
|
///
|
|
/// ## Locking
|
|
/// The lock is taken twice, briefly, and **never held across the await**:
|
|
/// once here to claim, once inside the future to complete. That is what
|
|
/// keeps the returned future `Send` — a guard alive across an await point
|
|
/// would poison it — and it is why the node itself runs unlocked, for
|
|
/// however many minutes it needs.
|
|
///
|
|
/// ## Why the payload is cloned
|
|
/// `run` gets an owned `N` rather than a borrow: a `&N` parameter is live
|
|
/// for the whole future, which both borrows the graph across the await and
|
|
/// makes the future non-`Send`.
|
|
///
|
|
/// The output carries the insert result rather than swallowing it — this
|
|
/// crate has no logger, so a malformed grown job is reported to the caller,
|
|
/// who is the one that can log it. The node is completed either way: its
|
|
/// own work already happened.
|
|
pub fn claim_next<F, Fut>(
|
|
sched: &Arc<Mutex<Self>>,
|
|
run: F,
|
|
) -> Option<impl Future<Output = (NodeId, Result<(), BuildError>)> + use<F, Fut, N, R>>
|
|
where
|
|
N: Clone,
|
|
F: FnOnce(NodeId, N, JobBuilder<N, R>) -> Fut,
|
|
Fut: Future<Output = (JobBuilder<N, R>, Outcome)>,
|
|
{
|
|
let (id, payload) = {
|
|
let mut guard = sched.lock().expect("jobq scheduler mutex poisoned");
|
|
let id = guard.claim_one()?;
|
|
let payload = guard.graph.node(id)?.payload.clone();
|
|
(id, payload)
|
|
};
|
|
let sched = Arc::clone(sched);
|
|
Some(async move {
|
|
let (grown, outcome) = run(id, payload, JobBuilder::new()).await;
|
|
let grew = sched
|
|
.lock()
|
|
.expect("jobq scheduler mutex poisoned")
|
|
.complete_growing(id, outcome, grown);
|
|
(id, grew)
|
|
})
|
|
}
|
|
|
|
/// Try to start node `id`. For each resource it needs, decide per the parent
|
|
/// tree (see the module docs): acquire fresh units (owner), acquire an extra
|
|
/// unit (grant lent elsewhere), or borrow an ancestor's grant. The fresh set
|
|
/// is taken atomically (all-or-nothing via [`ResourceTable::try_acquire_all`])
|
|
/// and recorded as owned; borrows are recorded only once the fresh set is
|
|
/// secured. If the fresh set doesn't fit, nothing is taken, no borrow is
|
|
/// recorded, and the node stays pending. Returns whether it started.
|
|
fn try_start(&mut self, id: NodeId) -> bool {
|
|
let mut to_acquire: Vec<(R, u32)> = Vec::new();
|
|
let mut to_borrow: Vec<(NodeId, R)> = Vec::new();
|
|
for (name, count) in self.resource_reqs(id) {
|
|
match self.parent_ancestor_owning(id, &name) {
|
|
// Case 1: no ancestor holds it → this node acquires + owns it.
|
|
None => to_acquire.push((name, count)),
|
|
Some(owner) => match self.borrowed.get(&(owner, name.clone())).copied() {
|
|
// Case 3: the grant is free → borrow it, no new unit.
|
|
None => to_borrow.push((owner, name)),
|
|
// Covered: already lent to a branch this node is inside.
|
|
Some(branch) if self.parent_chain_contains(id, branch) => {}
|
|
// Case 2: lent to a different branch → take an extra unit.
|
|
Some(_) => to_acquire.push((name, count)),
|
|
},
|
|
}
|
|
}
|
|
if !to_acquire.is_empty() && !self.resources.try_acquire_all(&to_acquire) {
|
|
return false;
|
|
}
|
|
if !to_acquire.is_empty() {
|
|
self.owned.entry(id).or_default().extend(to_acquire);
|
|
}
|
|
for (owner, name) in to_borrow {
|
|
self.borrowed.insert((owner, name), id);
|
|
}
|
|
self.graph.set_state(id, State::Running);
|
|
true
|
|
}
|
|
|
|
/// The nearest [`Node::parent`] ancestor of `id` that *owns* (holds real
|
|
/// units of) `name`, or `None` if none does (⇒ `id` must acquire it fresh).
|
|
fn parent_ancestor_owning(&self, id: NodeId, name: &R) -> Option<NodeId> {
|
|
let mut cur = self.graph.node(id).and_then(|n| n.parent);
|
|
while let Some(p) = cur {
|
|
if self.node_owns(p, name) {
|
|
return Some(p);
|
|
}
|
|
cur = self.graph.node(p).and_then(|n| n.parent);
|
|
}
|
|
None
|
|
}
|
|
|
|
/// Whether `ancestor` lies on `id`'s [`Node::parent`] chain (i.e. `id` is in
|
|
/// `ancestor`'s subtree). `id` itself does not count as its own ancestor.
|
|
fn parent_chain_contains(&self, id: NodeId, ancestor: NodeId) -> bool {
|
|
let mut cur = self.graph.node(id).and_then(|n| n.parent);
|
|
while let Some(p) = cur {
|
|
if p == ancestor {
|
|
return true;
|
|
}
|
|
cur = self.graph.node(p).and_then(|n| n.parent);
|
|
}
|
|
false
|
|
}
|
|
|
|
/// Whether node `holder` holds real units of resource `name`.
|
|
fn node_owns(&self, holder: NodeId, name: &R) -> bool {
|
|
self.owned
|
|
.get(&holder)
|
|
.is_some_and(|units| units.iter().any(|(n, _)| n == name))
|
|
}
|
|
|
|
/// Whether `root` and every node in its [`Node::parent`] subtree are
|
|
/// terminal — the condition for releasing `root`'s owned grants (and for
|
|
/// giving back a borrow whose branch-root is `root`).
|
|
fn subtree_terminal(&self, root: NodeId) -> bool {
|
|
self.graph.node(root).is_none_or(|n| n.state.is_terminal())
|
|
&& !self
|
|
.graph
|
|
.nodes()
|
|
.any(|n| !n.state.is_terminal() && self.parent_chain_contains(n.id, root))
|
|
}
|
|
|
|
/// Report a running node's own logic result. On success the node is *not*
|
|
/// terminal until its sub-nodes ([`Node::parent`] children) all finish — it
|
|
/// rests in [`State::Finishing`] until then, rolling up to [`State::Done`]
|
|
/// (every child `Done`) or [`State::Failed`] (any child `Failed`/`Cancelled`).
|
|
/// On failure it is `Failed` at once and its pending sub-nodes are cancelled
|
|
/// (gated on a `Finishing` the parent never reached). Terminality then
|
|
/// propagates up the parent chain. Claim again afterwards to start
|
|
/// newly-unblocked work.
|
|
///
|
|
/// `pub(crate)`: completion is reachable only from inside the future
|
|
/// [`Self::claim_next`] hands back, so it is not expressible without the
|
|
/// claim it answers.
|
|
pub(crate) fn complete(&mut self, id: NodeId, outcome: Outcome) {
|
|
match outcome {
|
|
Outcome::Failed(error) => {
|
|
// Record the reason before the terminal transition so it's set
|
|
// by the time `set_state` stamps `finished_at`.
|
|
self.graph.set_error(id, error);
|
|
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();
|
|
}
|
|
|
|
/// [`Scheduler::complete`], plus whatever the node declared into the builder
|
|
/// it was handed while running.
|
|
///
|
|
/// `grown`'s nodes are inserted **under `id`** and *before* the completion,
|
|
/// so the node cannot roll terminal with its own appended work still
|
|
/// pending — the same ordering the caller previously had to arrange by
|
|
/// hand. A job that declares nothing costs nothing: the insert is skipped
|
|
/// outright, which is the overwhelmingly common case (most nodes grow no
|
|
/// work at all).
|
|
///
|
|
/// **A failed node grows nothing**, whatever it declared. Failure
|
|
/// cancel-cascades to every pending child of `id`, so work inserted here
|
|
/// would be `Skipped` by the very next statement — the insert is not wrong,
|
|
/// it is provably pointless. This lives here rather than in the caller
|
|
/// because it is a consequence of *this crate's* cascade rule; a host that
|
|
/// had to remember it could forget it.
|
|
///
|
|
/// # Errors
|
|
/// [`BuildError`] if `grown` is malformed — **and the node is still
|
|
/// completed**. Its own work already happened; refusing to complete it
|
|
/// would misreport that, and leaving it `Running` forever would wedge the
|
|
/// DAG. So the error is returned for the caller to log, not used to abort
|
|
/// the completion. This crate has no logger of its own; the caller does.
|
|
pub(crate) fn complete_growing(
|
|
&mut self,
|
|
id: NodeId,
|
|
outcome: Outcome,
|
|
grown: JobBuilder<N, R>,
|
|
) -> Result<(), BuildError> {
|
|
// A node that is no longer in the graph grows nothing. The insert below
|
|
// is *unchecked* — rooting on a departed parent would plant a dangling
|
|
// `parent` edge rather than being rejected. The host used to carry this
|
|
// guard itself, as a lookup before a separate append call; it belongs
|
|
// here, where the graph is and where it cannot be skipped.
|
|
//
|
|
// ⚠️ Deliberately untested, and untestable today: nothing removes a node
|
|
// from the graph yet (eviction only stops *retaining* a DAG; its nodes
|
|
// linger), and `NodeId` cannot be fabricated, so a test would have to
|
|
// fake the very condition it checks. This guard is defensive against the
|
|
// bounded prune that does not exist yet — when that lands, it needs a
|
|
// test, and this comment is the reminder.
|
|
let grew = if grown.is_empty()
|
|
|| matches!(outcome, Outcome::Failed(_))
|
|
|| self.graph.node(id).is_none()
|
|
{
|
|
Ok(())
|
|
} else {
|
|
let graph = &mut self.graph;
|
|
grown
|
|
.insert_with(Some(id), &[], |payload, deps, parent| {
|
|
graph.insert_unchecked(payload, deps, parent)
|
|
})
|
|
.map(|_ids| ())
|
|
};
|
|
self.complete(id, outcome);
|
|
grew
|
|
}
|
|
|
|
/// 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())
|
|
}
|
|
|
|
/// What `id` rolls up to once all its children are terminal: `Failed` if any
|
|
/// child failed, else `Cancelled` if any was cancelled, else `Done`.
|
|
///
|
|
/// `Failed` outranks `Cancelled` because the failure is the actionable fact —
|
|
/// a group where one step broke and the rest were dropped in response is a
|
|
/// failure, not a cancellation. A group whose children were *all* dropped
|
|
/// never failed at anything, and says so.
|
|
///
|
|
/// `Skipped` children are ignored: being ruled out by an edge is the expected
|
|
/// fate of every branch not taken, so counting it would make any group that
|
|
/// branches on outcome roll up non-`Done` however the run went.
|
|
fn rolled_up_state(&self, id: NodeId) -> State {
|
|
let mut any_cancelled = false;
|
|
for child in self.graph.nodes().filter(|n| n.parent == Some(id)) {
|
|
match child.state {
|
|
State::Failed => return State::Failed,
|
|
State::Cancelled => any_cancelled = true,
|
|
_ => {}
|
|
}
|
|
}
|
|
if any_cancelled {
|
|
State::Cancelled
|
|
} else {
|
|
State::Done
|
|
}
|
|
}
|
|
|
|
/// 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`.
|
|
///
|
|
/// Cascades on **any** terminal outcome, `Done` included. Since an edge names
|
|
/// the set of outcomes it accepts, success can rule a dependent out just as
|
|
/// failure can — a `{Failed}` compensation branch is unsatisfiable the moment
|
|
/// its dependency succeeds, and leaving it `Pending` would wedge the subtree
|
|
/// non-terminal forever.
|
|
fn settle_terminal(&mut self, id: NodeId) {
|
|
let state = if self.all_children_terminal(id) {
|
|
self.rolled_up_state(id)
|
|
} else {
|
|
State::Finishing
|
|
};
|
|
self.graph.set_state(id, state);
|
|
if state.is_terminal() {
|
|
self.cascade_cancel(id);
|
|
}
|
|
}
|
|
|
|
/// After `start` became terminal, roll up every ancestor that was parked in
|
|
/// `Finishing` awaiting its children: once all of an ancestor's children are
|
|
/// terminal it transitions (Done / Failed), which may let *its* parent roll
|
|
/// up too, and so on up the [`Node::parent`] chain.
|
|
fn roll_up_ancestors(&mut self, start: NodeId) {
|
|
let mut cur = self.graph.node(start).and_then(|n| n.parent);
|
|
while let Some(a) = cur {
|
|
if self.graph.node(a).map(|n| n.state) != Some(State::Finishing)
|
|
|| !self.all_children_terminal(a)
|
|
{
|
|
break;
|
|
}
|
|
let state = self.rolled_up_state(a);
|
|
self.graph.set_state(a, state);
|
|
// Any terminal outcome can rule a dependent out — see `settle_terminal`.
|
|
self.cascade_cancel(a);
|
|
cur = self.graph.node(a).and_then(|n| n.parent);
|
|
}
|
|
}
|
|
|
|
/// Cancel the not-yet-started work at `id`: mark it [`State::Cancelled`]
|
|
/// and cascade to the dependents that rules out. Cancelling a node cancels
|
|
/// what hangs under it — a group is abandoned by abandoning its root.
|
|
/// Reports whether anything was cancelled.
|
|
///
|
|
/// **All-or-nothing.** A subtree with any node already `Running` or terminal
|
|
/// is left completely untouched: an in-flight node's work is not
|
|
/// interruptible, and cancelling only the pending half would leave the group
|
|
/// half-executed with no way to finish it.
|
|
///
|
|
/// **A node whose edge accepts [`TerminalState::Cancelled`] is spared.** Such
|
|
/// a node is asking to run precisely when the work it follows is dropped,
|
|
/// which is what lets a reporting tail settle whatever it reports to instead
|
|
/// of leaving it dangling forever. Nothing is special-cased by payload — the
|
|
/// node's own declared edges decide. Note `DepWhen::AFTER_ANY` deliberately
|
|
/// does *not* accept `Cancelled`, so an ordinary weak-edged step is cancelled
|
|
/// along with the rest; there is nothing to do when no node ever ran.
|
|
///
|
|
/// Cancelled nodes were pending, so they hold no resources and none are
|
|
/// released here.
|
|
pub fn cancel_node(&mut self, id: NodeId) -> bool {
|
|
// The unit is the work *under* `id`; a node with no children is its own
|
|
// work. A group root's state is its subtree's roll-up rather than a step
|
|
// that ran, so the root itself is not part of the gate.
|
|
let under: Vec<NodeId> = self.graph.descendants(id).map(|n| n.id).collect();
|
|
let is_group = !under.is_empty();
|
|
let targets = if is_group { under } else { vec![id] };
|
|
if !targets.iter().all(|&n| {
|
|
self.graph
|
|
.node(n)
|
|
.is_some_and(|n| n.state == State::Pending)
|
|
}) {
|
|
return false;
|
|
}
|
|
for n in targets {
|
|
if self.observes_cancellation(n) {
|
|
continue;
|
|
}
|
|
self.graph.set_state(n, State::Cancelled);
|
|
self.cascade_cancel(n);
|
|
}
|
|
if is_group {
|
|
// Re-run the root's roll-up now its children are terminal. With a
|
|
// spared node still pending this is a deliberate no-op: the root
|
|
// still has a non-terminal child, so it parks back in `Finishing`
|
|
// and rolls up for real once that node finishes.
|
|
self.complete(id, Outcome::Done);
|
|
}
|
|
true
|
|
}
|
|
|
|
/// Whether `id` has an edge that accepts a **dropped** dependency — i.e. it
|
|
/// exists to report on work that may never run.
|
|
fn observes_cancellation(&self, id: NodeId) -> bool {
|
|
self.graph.node(id).is_some_and(|n| {
|
|
n.deps.iter().any(
|
|
|d| matches!(d, Dep::Node { when, .. } if when.accepts(TerminalState::Cancelled)),
|
|
)
|
|
})
|
|
}
|
|
|
|
/// 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 elimination out from a just-terminal `origin`: every
|
|
/// still-`Pending` node that can no longer run gets marked
|
|
/// [`State::Skipped`], transitively. Skipped nodes were `Pending`, so they
|
|
/// hold no resources.
|
|
///
|
|
/// `Skipped`, not `Cancelled`: these nodes were *ruled out by their edges*,
|
|
/// which is a normal outcome, not a dropped job. `Cancelled` is reserved for
|
|
/// work the caller abandoned before it started ([`Scheduler::cancel_node`]),
|
|
/// and the two are distinguished precisely so a parent's roll-up can ignore
|
|
/// the former while still treating the latter as not-success.
|
|
///
|
|
/// One rule decides it: **a node is doomed once any edge it names can never
|
|
/// be satisfied** — the dep settled on an outcome that edge does not accept.
|
|
/// `AFTER_OK` on a `Failed` dep dooms (a strong dependency failed);
|
|
/// `AFTER_ANY` never dooms, which is what lets a tail node survive the
|
|
/// cancellation of the work it reports on. That falls out of the edge's own
|
|
/// set rather than being a special case for particular node kinds.
|
|
///
|
|
/// Plus the structural edge: being a *child* of a doomed node. Its parent
|
|
/// will never reach `Finishing`, so it was gated from ever starting, and
|
|
/// leaving it pending would wedge the subtree non-terminal.
|
|
fn cascade_cancel(&mut self, origin: NodeId) {
|
|
let mut stack = vec![origin];
|
|
while let Some(cur) = stack.pop() {
|
|
let Some(outcome) = self.graph.node(cur).and_then(|n| n.state.terminal()) else {
|
|
continue;
|
|
};
|
|
let doomed: Vec<NodeId> = self
|
|
.graph
|
|
.nodes()
|
|
.filter(|n| {
|
|
n.state == State::Pending
|
|
&& (n.parent == Some(cur)
|
|
|| n.deps.iter().any(|d| {
|
|
matches!(d, Dep::Node { id, when } if *id == cur && !when.accepts(outcome))
|
|
}))
|
|
})
|
|
.map(|n| n.id)
|
|
.collect();
|
|
for d in doomed {
|
|
self.graph.set_state(d, State::Skipped);
|
|
stack.push(d);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Give back any borrow whose branch has fully left (freeing the grant for a
|
|
/// waiting sibling), then release every owner's grant whose whole subtree is
|
|
/// terminal (dropping the units back into the table).
|
|
fn release_ready(&mut self) {
|
|
// 1. Return borrows whose branch-root subtree is now terminal.
|
|
let returned: Vec<(NodeId, R)> = self
|
|
.borrowed
|
|
.iter()
|
|
.filter(|&(_, &branch)| self.subtree_terminal(branch))
|
|
.map(|((owner, name), _)| (*owner, name.clone()))
|
|
.collect();
|
|
for key in returned {
|
|
self.borrowed.remove(&key);
|
|
}
|
|
// 2. Release owner grants whose whole subtree is terminal.
|
|
let owners: Vec<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);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Whether `id` is clear to start: its parent's own logic is done *and* every
|
|
/// [`Dep::Node`] edge is satisfied. The parent gate (a sub-node runs only
|
|
/// after its parent reaches [`State::Finishing`] — the parent can't be
|
|
/// terminal while this child is pending, so `Finishing` is the exact "logic
|
|
/// done" signal) supplies parent→child ordering; `Dep::Node` edges (which the
|
|
/// graph restricts to the same parent group) supply sibling ordering.
|
|
/// `Dep::Resource` edges are handled by the atomic acquire in
|
|
/// [`Scheduler::try_start`], not here.
|
|
fn node_deps_satisfied(&self, id: NodeId) -> bool {
|
|
let Some(node) = self.graph.node(id) else {
|
|
return false;
|
|
};
|
|
if let Some(parent) = node.parent
|
|
&& self.graph.node(parent).map(|n| n.state) != Some(State::Finishing)
|
|
{
|
|
return false;
|
|
}
|
|
node.deps.iter().all(|dep| match dep {
|
|
Dep::Resource { .. } => true,
|
|
Dep::Node { id, when } => self
|
|
.graph
|
|
.node(*id)
|
|
.is_some_and(|n| when.satisfied_by(n.state)),
|
|
})
|
|
}
|
|
|
|
/// The `(name, count)` resource units `id` must hold to run.
|
|
fn resource_reqs(&self, id: NodeId) -> Vec<(R, u32)> {
|
|
self.graph.node(id).map_or_else(Vec::new, |node| {
|
|
node.deps
|
|
.iter()
|
|
.filter_map(|dep| match dep {
|
|
Dep::Resource { name, count } => Some((name.clone(), *count)),
|
|
Dep::Node { .. } => None,
|
|
})
|
|
.collect()
|
|
})
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::{DepWhen, TerminalState};
|
|
|
|
fn res(name: &str) -> String {
|
|
name.to_owned()
|
|
}
|
|
|
|
/// Claim every currently-runnable node, as arrangement for the assertions
|
|
/// below. Equivalent to calling [`Scheduler::claim_one`] until it yields
|
|
/// `None`: a node started by an earlier iteration is `Running`, not
|
|
/// terminal, so it cannot satisfy another node's dependency here — it only
|
|
/// consumes resources.
|
|
///
|
|
/// **Was `Scheduler::settle`, a public method.** It was a `claim_one` loop
|
|
/// returning a `Vec`, and production never wanted the batch: the run loop
|
|
/// takes one node at a time through [`Scheduler::claim_next`] so it can
|
|
/// choose between claiming again and backing off, which a batch return
|
|
/// can't express. The only callers were tests, so it lives with them.
|
|
fn settle<N, R: Clone + Eq + Hash>(s: &mut Scheduler<N, R>) -> Vec<NodeId> {
|
|
let mut started = Vec::new();
|
|
while let Some(id) = s.claim_one() {
|
|
started.push(id);
|
|
}
|
|
started
|
|
}
|
|
|
|
/// 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)
|
|
}
|
|
|
|
/// A single-unit resource dep on `name`.
|
|
fn res_dep(name: &str) -> Vec<Dep<String>> {
|
|
vec![Dep::Resource {
|
|
name: res(name),
|
|
count: 1,
|
|
}]
|
|
}
|
|
|
|
fn after_ok(on: NodeId) -> Dep<String> {
|
|
Dep::Node {
|
|
id: on,
|
|
when: DepWhen::AFTER_OK,
|
|
}
|
|
}
|
|
|
|
fn avail(s: &Scheduler<&str, String>, name: &str) -> u32 {
|
|
s.resources.available(&res(name))
|
|
}
|
|
|
|
/// Children of `id`, by payload, in insertion order.
|
|
fn children_of(s: &Scheduler<&'static str, String>, id: NodeId) -> Vec<&'static str> {
|
|
s.graph()
|
|
.nodes()
|
|
.filter(|n| n.parent == Some(id))
|
|
.map(|n| n.payload)
|
|
.collect()
|
|
}
|
|
|
|
/// A node completing `Done` gets the work it declared while running,
|
|
/// inserted **under itself** — so the DAG cannot roll terminal with the
|
|
/// appended work still pending.
|
|
#[test]
|
|
fn a_completing_node_grows_the_work_it_declared() {
|
|
let mut s = scheduler_with_slots(1);
|
|
let n = s.append("emitter", vec![], None).expect("insert");
|
|
assert_eq!(settle(&mut s), vec![n]);
|
|
|
|
let grown = JobBuilder::new();
|
|
grown.node("child-a");
|
|
grown.node("child-b");
|
|
s.complete_growing(n, Outcome::Done, grown)
|
|
.expect("well-formed growth");
|
|
|
|
assert_eq!(children_of(&s, n), vec!["child-a", "child-b"]);
|
|
// The emitter parks in `Finishing` rather than going terminal: its own
|
|
// appended work is still pending under it. That ordering is the whole
|
|
// point of growing *as part of* the completion.
|
|
assert_eq!(s.graph().node(n).unwrap().state, State::Finishing);
|
|
}
|
|
|
|
/// A **failed** node grows nothing, whatever it declared.
|
|
///
|
|
/// The companion to the test above, and the reason this rule lives in the
|
|
/// crate rather than in a caller: failure cancel-cascades to every pending
|
|
/// child of the completing node, so anything inserted here would be
|
|
/// `Skipped` by the very next statement. Enforcing it host-side means every
|
|
/// host has to remember it; enforcing it here means none can forget.
|
|
#[test]
|
|
fn a_failed_node_grows_nothing() {
|
|
let mut s = scheduler_with_slots(1);
|
|
let n = s.append("emitter", vec![], None).expect("insert");
|
|
assert_eq!(settle(&mut s), vec![n]);
|
|
|
|
let grown = JobBuilder::new();
|
|
grown.node("never-runs");
|
|
s.complete_growing(n, Outcome::Failed("boom".to_owned()), grown)
|
|
.expect("growth is dropped, not rejected");
|
|
|
|
assert!(
|
|
children_of(&s, n).is_empty(),
|
|
"a failed node must not append work, got {:?}",
|
|
children_of(&s, n)
|
|
);
|
|
assert_eq!(s.graph().node(n).unwrap().state, State::Failed);
|
|
}
|
|
|
|
/// A contended resource goes to the oldest waiter.
|
|
///
|
|
/// [`Scheduler::claim_one`] scans [`Graph::nodes`] — insertion order — and
|
|
/// takes the first node whose deps are satisfied and whose resources it can
|
|
/// acquire. That *is* the fairness guarantee: there is no queue, no
|
|
/// priority, just the scan order.
|
|
///
|
|
/// Load-bearing for any host that submits work over time, because without
|
|
/// it a steady arrival rate could starve the earliest waiter indefinitely.
|
|
/// It was previously only covered downstream, by a host test driving its own
|
|
/// templates — which meant the property this crate provides was asserted
|
|
/// everywhere except in this crate.
|
|
#[test]
|
|
fn a_contended_resource_goes_to_the_oldest_waiter() {
|
|
let mut s = scheduler_with_slots(1);
|
|
let a = s.append("a", res_dep("build-slot"), None).expect("a");
|
|
let b = s.append("b", res_dep("build-slot"), None).expect("b");
|
|
let c = s.append("c", res_dep("build-slot"), None).expect("c");
|
|
|
|
assert_eq!(settle(&mut s), vec![a], "cap 1: only the first can start");
|
|
s.complete(a, Outcome::Done);
|
|
// b and c are both satisfiable now; b was inserted first.
|
|
assert_eq!(settle(&mut s), vec![b], "the freed unit goes to b, not c");
|
|
s.complete(b, Outcome::Done);
|
|
assert_eq!(settle(&mut s), vec![c]);
|
|
}
|
|
|
|
#[test]
|
|
fn leaf_owner_goes_done_directly_and_releases() {
|
|
let mut s = scheduler_with_slots(1);
|
|
let n = s
|
|
.append("build", res_dep("build-slot"), None)
|
|
.expect("insert");
|
|
assert_eq!(settle(&mut s), vec![n]);
|
|
assert_eq!(s.graph().node(n).unwrap().state, State::Running);
|
|
assert_eq!(avail(&s, "build-slot"), 0);
|
|
// No children → completing it goes straight to Done (skips Finishing).
|
|
s.complete(n, Outcome::Done);
|
|
assert_eq!(s.graph().node(n).unwrap().state, State::Done);
|
|
assert_eq!(avail(&s, "build-slot"), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn lifecycle_timestamps_and_error_are_stamped() {
|
|
let mut s = scheduler_with_slots(2);
|
|
let ok = s.append("ok", vec![], None).expect("insert");
|
|
let bad = s.append("bad", vec![], None).expect("insert");
|
|
let downstream = s.append("down", vec![after_ok(bad)], None).expect("insert");
|
|
|
|
// Before running: no timestamps.
|
|
assert!(s.graph().node(ok).unwrap().started_at.is_none());
|
|
assert!(s.graph().node(ok).unwrap().finished_at.is_none());
|
|
|
|
let started = settle(&mut s);
|
|
assert!(started.contains(&ok) && started.contains(&bad));
|
|
// Running → started_at stamped, finished_at still none.
|
|
assert!(s.graph().node(ok).unwrap().started_at.is_some());
|
|
assert!(s.graph().node(ok).unwrap().finished_at.is_none());
|
|
|
|
// Done → finished_at stamped, no error.
|
|
s.complete(ok, Outcome::Done);
|
|
let n = s.graph().node(ok).unwrap();
|
|
assert!(n.finished_at.is_some());
|
|
assert_eq!(n.error, None);
|
|
|
|
// Failed → the reason rides `Outcome::Failed`, finished_at stamped.
|
|
s.complete(bad, Outcome::Failed("boom".to_owned()));
|
|
let n = s.graph().node(bad).unwrap();
|
|
assert_eq!(n.state, State::Failed);
|
|
assert_eq!(n.error.as_deref(), Some("boom"));
|
|
assert!(n.finished_at.is_some());
|
|
|
|
// The `AFTER_OK` dependent: ruled out by its edge, so finished_at is set,
|
|
// but it never ran (no started_at) and carries no error of its own.
|
|
let n = s.graph().node(downstream).unwrap();
|
|
assert_eq!(n.state, State::Skipped);
|
|
assert!(n.started_at.is_none());
|
|
assert!(n.finished_at.is_some());
|
|
assert_eq!(n.error, None);
|
|
}
|
|
|
|
#[test]
|
|
fn build_slot_cap_limits_concurrency_and_release_unblocks() {
|
|
let mut s = scheduler_with_slots(2);
|
|
// Three independent (unparented) nodes each own a fresh slot unit.
|
|
let a = s.append("a", res_dep("build-slot"), None).expect("a");
|
|
let b = s.append("b", res_dep("build-slot"), None).expect("b");
|
|
let c = s.append("c", res_dep("build-slot"), None).expect("c");
|
|
// cap 2 → a + b start, c blocks on the exhausted slot.
|
|
assert_eq!(settle(&mut s), 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!(settle(&mut s), vec![c]);
|
|
assert_eq!(s.graph().node(c).unwrap().state, State::Running);
|
|
}
|
|
|
|
#[test]
|
|
fn parent_parks_in_finishing_until_children_roll_up() {
|
|
// `root` (a group node) runs, then its two sub-nodes run. `root` is not
|
|
// terminal until both children are — it waits in `Finishing`.
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let c1 = s.append("c1", vec![], Some(root)).expect("c1");
|
|
let c2 = s.append("c2", vec![], Some(root)).expect("c2");
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
// Children can't start yet — parent still Running (logic not done).
|
|
assert!(settle(&mut s).is_empty(), "children gated on parent logic");
|
|
s.complete(root, Outcome::Done);
|
|
assert_eq!(
|
|
s.graph().node(root).unwrap().state,
|
|
State::Finishing,
|
|
"logic done, children pending → Finishing"
|
|
);
|
|
let mut started = settle(&mut s);
|
|
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"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn failed_child_rolls_parent_up_to_failed() {
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let child = s.append("child", vec![], Some(root)).expect("child");
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Done);
|
|
assert_eq!(settle(&mut s), vec![child]);
|
|
s.complete(child, Outcome::Failed(String::new()));
|
|
assert_eq!(
|
|
s.graph().node(root).unwrap().state,
|
|
State::Failed,
|
|
"a failed child rolls the parent up to Failed"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn owner_holds_grant_for_its_whole_subtree() {
|
|
// Group root R owns the slot; c1 (its child) borrows it; c2 (its child,
|
|
// needs no slot, ordered after c1) doesn't. The slot is held until R's
|
|
// WHOLE subtree is terminal — not freed after the last needer (c1).
|
|
let mut s = scheduler_with_slots(1);
|
|
let r = s.append("R", res_dep("build-slot"), None).expect("R");
|
|
let c1 = s.append("c1", res_dep("build-slot"), Some(r)).expect("c1");
|
|
let c2 = s.append("c2", vec![after_ok(c1)], Some(r)).expect("c2");
|
|
assert_eq!(settle(&mut s), vec![r]);
|
|
s.complete(r, Outcome::Done); // → Finishing (children pending)
|
|
assert_eq!(avail(&s, "build-slot"), 0, "held: subtree not terminal");
|
|
assert_eq!(settle(&mut s), 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!(settle(&mut s), 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!(settle(&mut s), 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!(settle(&mut s), 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!(settle(&mut s), vec![r]);
|
|
s.complete(r, Outcome::Done);
|
|
assert_eq!(settle(&mut s), vec![child], "child borrows R's grant");
|
|
s.complete(child, Outcome::Done);
|
|
assert_eq!(settle(&mut s), vec![grand], "grand covered, no deadlock");
|
|
s.complete(grand, Outcome::Done);
|
|
assert_eq!(settle(&mut s), 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!(
|
|
settle(&mut s),
|
|
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!(settle(&mut s), 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!(settle(&mut s), vec![owner]);
|
|
let c1 = s
|
|
.append("c1", res_dep("agent/foo"), Some(owner))
|
|
.expect("c1");
|
|
let c2 = s
|
|
.append("c2", res_dep("agent/foo"), Some(owner))
|
|
.expect("c2");
|
|
s.complete(owner, Outcome::Done); // → Finishing
|
|
assert_eq!(settle(&mut s), vec![c1], "c1 borrows; c2 can't (cap 1)");
|
|
assert_eq!(s.graph().node(c2).unwrap().state, State::Pending);
|
|
s.complete(c1, Outcome::Done);
|
|
assert_eq!(settle(&mut s), vec![c2], "borrow returned → c2 borrows");
|
|
assert_eq!(avail(&s, "agent/foo"), 0, "still just the owner's unit");
|
|
}
|
|
|
|
#[test]
|
|
fn sibling_borrowers_run_concurrently_when_capacity_allows() {
|
|
// build-slot cap 2: owner holds one unit; c1 borrows it, c2 (grant lent
|
|
// to a sibling branch) takes the *second* unit — both run at once.
|
|
let mut s = scheduler_with_slots(2);
|
|
let owner = s
|
|
.append("owner", res_dep("build-slot"), None)
|
|
.expect("owner");
|
|
assert_eq!(settle(&mut s), 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 = settle(&mut s);
|
|
started.sort();
|
|
let mut expected = vec![c1, c2];
|
|
expected.sort();
|
|
assert_eq!(started, expected, "c1 borrows, c2 takes the 2nd unit");
|
|
assert_eq!(avail(&s, "build-slot"), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn failed_after_ok_dep_cancels_dependents_but_after_any_still_runs() {
|
|
// A group of top-level siblings ordered by `AfterOk`; the failure of
|
|
// `root` cancels its strong-dependent chain, an `AfterAny` still runs.
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let strong1 = s
|
|
.append("strong1", vec![after_ok(root)], None)
|
|
.expect("strong1");
|
|
let strong2 = s
|
|
.append("strong2", vec![after_ok(strong1)], None)
|
|
.expect("strong2");
|
|
let weak = s
|
|
.append(
|
|
"weak",
|
|
vec![Dep::Node {
|
|
id: root,
|
|
when: DepWhen::AFTER_ANY,
|
|
}],
|
|
None,
|
|
)
|
|
.expect("weak");
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Failed(String::new()));
|
|
assert_eq!(s.graph().node(strong1).unwrap().state, State::Skipped);
|
|
assert_eq!(s.graph().node(strong2).unwrap().state, State::Skipped);
|
|
assert_eq!(settle(&mut s), vec![weak]);
|
|
}
|
|
|
|
/// The direction only a *set* edge can express: a branch that runs solely on
|
|
/// failure. Success has to rule it out, which means the cascade must fire on
|
|
/// `Done` too — otherwise it sits `Pending` forever and wedges the graph.
|
|
#[test]
|
|
fn success_cancels_a_failure_only_branch() {
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let on_fail = s
|
|
.append(
|
|
"compensate",
|
|
vec![Dep::Node {
|
|
id: root,
|
|
when: DepWhen::of(&[TerminalState::Failed]),
|
|
}],
|
|
None,
|
|
)
|
|
.expect("compensate");
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Done);
|
|
assert_eq!(
|
|
s.graph().node(on_fail).unwrap().state,
|
|
State::Skipped,
|
|
"a Failed-only branch is unsatisfiable once its dep succeeds — and it is \
|
|
`Skipped`, not `Cancelled`, so the parent roll-up ignores it"
|
|
);
|
|
assert!(settle(&mut s).is_empty(), "and nothing is left runnable");
|
|
}
|
|
|
|
/// The mirror: the same branch is exactly what *does* run on failure, while
|
|
/// an `AFTER_OK` sibling is cancelled. One edge set, both directions.
|
|
#[test]
|
|
fn failure_runs_the_failure_only_branch_and_cancels_the_ok_one() {
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let on_ok = s
|
|
.append("on_ok", vec![after_ok(root)], None)
|
|
.expect("on_ok");
|
|
let on_fail = s
|
|
.append(
|
|
"on_fail",
|
|
vec![Dep::Node {
|
|
id: root,
|
|
when: DepWhen::of(&[TerminalState::Failed]),
|
|
}],
|
|
None,
|
|
)
|
|
.expect("on_fail");
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Failed("boom".to_owned()));
|
|
assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped);
|
|
assert_eq!(settle(&mut s), vec![on_fail]);
|
|
}
|
|
|
|
/// A weak edge accepts a dependency that was *ruled out*, so a tail still
|
|
/// runs when the work it reports on never happened — held by the edge itself
|
|
/// rather than by any node-kind special case.
|
|
#[test]
|
|
fn eliminated_dep_still_satisfies_a_weak_edge() {
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let mid = s.append("mid", vec![after_ok(root)], None).expect("mid");
|
|
let tail = s
|
|
.append(
|
|
"tail",
|
|
vec![Dep::Node {
|
|
id: mid,
|
|
when: DepWhen::AFTER_ANY,
|
|
}],
|
|
None,
|
|
)
|
|
.expect("tail");
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Failed("boom".to_owned()));
|
|
assert_eq!(s.graph().node(mid).unwrap().state, State::Skipped);
|
|
assert_eq!(
|
|
settle(&mut s),
|
|
vec![tail],
|
|
"the tail runs off a cancelled dependency"
|
|
);
|
|
}
|
|
|
|
/// Edges are **conjunctive**, so "any of these several nodes failed" is not
|
|
/// directly expressible — a `{Failed}` edge on each would mean *all* failed.
|
|
/// The composition that does work: the success branch depends `AFTER_OK` on
|
|
/// every node (so it runs only if all succeeded, and is ruled out the moment
|
|
/// one doesn't), and the failure branch hangs off *it* with `{Skipped}` —
|
|
/// "run when the success branch was ruled out". The success branch is the
|
|
/// aggregator, and exactly one of the two runs.
|
|
#[test]
|
|
fn ok_branch_aggregates_and_failure_branch_hangs_off_its_elimination() {
|
|
let build = || {
|
|
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![], None).expect("b");
|
|
let on_ok = s
|
|
.append("on_ok", vec![after_ok(a), after_ok(b)], None)
|
|
.expect("on_ok");
|
|
let on_fail = s
|
|
.append(
|
|
"on_fail",
|
|
vec![Dep::Node {
|
|
id: on_ok,
|
|
when: DepWhen::of(&[TerminalState::Skipped]),
|
|
}],
|
|
None,
|
|
)
|
|
.expect("on_fail");
|
|
(s, a, b, on_ok, on_fail)
|
|
};
|
|
|
|
// Everything succeeds: the ok branch runs, the failure branch is ruled out.
|
|
let (mut s, a, b, on_ok, on_fail) = build();
|
|
assert_eq!(
|
|
settle(&mut s),
|
|
vec![a, b],
|
|
"both roots start; neither tail can"
|
|
);
|
|
s.complete(a, Outcome::Done);
|
|
s.complete(b, Outcome::Done);
|
|
assert_eq!(settle(&mut s), vec![on_ok]);
|
|
s.complete(on_ok, Outcome::Done);
|
|
assert_eq!(s.graph().node(on_fail).unwrap().state, State::Skipped);
|
|
assert!(settle(&mut s).is_empty());
|
|
|
|
// One of them fails: the ok branch is ruled out, which is precisely the
|
|
// signal the failure branch waits on.
|
|
let (mut s, a, b, on_ok, on_fail) = build();
|
|
assert_eq!(settle(&mut s), vec![a, b]);
|
|
s.complete(a, Outcome::Failed("boom".to_owned()));
|
|
s.complete(b, Outcome::Done);
|
|
assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped);
|
|
assert_eq!(settle(&mut s), vec![on_fail]);
|
|
}
|
|
|
|
#[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!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Failed(String::new()));
|
|
assert_eq!(s.graph().node(child).unwrap().state, State::Skipped);
|
|
assert_eq!(s.graph().node(grandchild).unwrap().state, State::Skipped);
|
|
}
|
|
|
|
#[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));
|
|
// `a` was dropped by the caller; `b` was merely ruled out by its edge.
|
|
assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled);
|
|
assert_eq!(s.graph().node(b).unwrap().state, State::Skipped);
|
|
let c = s.append("c", vec![], None).expect("c");
|
|
assert_eq!(settle(&mut s), vec![c]);
|
|
assert!(!s.cancel_node(c));
|
|
assert_eq!(s.graph().node(c).unwrap().state, State::Running);
|
|
}
|
|
|
|
/// Cancelling a group root abandons the work under it. The root's own state
|
|
/// is a roll-up (it parks in `Finishing`), never `Pending`, so gating on the
|
|
/// root instead of its children would refuse every group.
|
|
#[test]
|
|
fn cancel_node_cancels_the_work_under_a_group_root() {
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let a = s.append("a", vec![], Some(root)).expect("a");
|
|
let b = s.append("b", vec![after_ok(a)], Some(root)).expect("b");
|
|
// The root runs first and parks in `Finishing` while its children are
|
|
// outstanding — the state a group root is actually in when cancelled.
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Done);
|
|
assert_eq!(s.graph().node(root).unwrap().state, State::Finishing);
|
|
|
|
assert!(s.cancel_node(root));
|
|
assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled);
|
|
assert_eq!(s.graph().node(b).unwrap().state, State::Cancelled);
|
|
// With every child terminal the root leaves `Finishing` and rolls up —
|
|
// as `Cancelled`, not `Failed`: nothing under it failed at anything, the
|
|
// work was dropped.
|
|
assert_eq!(s.graph().node(root).unwrap().state, State::Cancelled);
|
|
}
|
|
|
|
/// All-or-nothing: an in-flight node's work is not interruptible, and
|
|
/// cancelling only the pending half would strand the group half-executed.
|
|
#[test]
|
|
fn cancel_node_refuses_a_group_with_anything_running() {
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let a = s.append("a", vec![], Some(root)).expect("a");
|
|
let b = s.append("b", vec![after_ok(a)], Some(root)).expect("b");
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Done);
|
|
assert_eq!(settle(&mut s), vec![a], "a is claimed and running");
|
|
|
|
assert!(!s.cancel_node(root), "refused while a runs");
|
|
assert_eq!(s.graph().node(a).unwrap().state, State::Running);
|
|
assert_eq!(
|
|
s.graph().node(b).unwrap().state,
|
|
State::Pending,
|
|
"the pending half is left alone too — nothing partial"
|
|
);
|
|
}
|
|
|
|
/// A node whose edge accepts `Cancelled` asked to run when the work it
|
|
/// follows is dropped. Sparing it is what lets a reporting tail settle the
|
|
/// thing it reports to instead of leaving it dangling forever.
|
|
#[test]
|
|
fn cancel_node_spares_a_node_that_observes_cancellation() {
|
|
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
|
let root = s.append("root", vec![], None).expect("root");
|
|
let work = s.append("work", vec![], Some(root)).expect("work");
|
|
let tail = s
|
|
.append(
|
|
"tail",
|
|
vec![Dep::Node {
|
|
id: work,
|
|
when: DepWhen::of(&[TerminalState::Cancelled]),
|
|
}],
|
|
Some(root),
|
|
)
|
|
.expect("tail");
|
|
assert_eq!(settle(&mut s), vec![root]);
|
|
s.complete(root, Outcome::Done);
|
|
|
|
assert!(s.cancel_node(root));
|
|
assert_eq!(s.graph().node(work).unwrap().state, State::Cancelled);
|
|
assert_eq!(
|
|
s.graph().node(tail).unwrap().state,
|
|
State::Pending,
|
|
"spared, and now runnable since its dep is Cancelled"
|
|
);
|
|
assert_eq!(settle(&mut s), vec![tail], "the tail still gets to report");
|
|
}
|
|
|
|
#[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!(settle(&mut s).len(), 2);
|
|
let state = s.resource_state();
|
|
assert!(state.contains(&(res("agent/foo"), g)));
|
|
assert!(state.contains(&(res("build-slot"), b)));
|
|
}
|
|
}
|