feat(#2450): append_subgraph primitive + sweep MetaLock grows rebuilds in-DAG
First half of making the startup sweep one DAG. Adds the runtime subgraph-append machinery and switches the sweep MetaLock from fanning out child Rebuild DAGs to growing one rebuild subgraph per stale agent into the same DAG: - JobQueue::append_subgraph(dag_id, nodes, dep_on) — the multi-node, multi-agent generalisation of append_node: rebases a subgraph's local deps onto the DAG's id space and roots it on the emitting node. - NodeOutput.append_subgraph: Vec<Vec<NodeSpec>> — the executor→scheduler channel for it; scheduler drains it before completing the emitting node (same ordering as append_nodes). - run_meta_lock sweep branch returns the stale agents' rebuild_nodes subgraphs via append_subgraph instead of fanout. Follow-up commit collapses submit_boot_tree (drop boot_root + per-agent reconcile child DAGs) so the whole boot is one DAG built inline.
This commit is contained in:
parent
44286996ed
commit
b6defdeaaf
3 changed files with 92 additions and 3 deletions
|
|
@ -34,7 +34,8 @@ use hive_sh4re::wire_time::now_unix;
|
|||
use tokio::sync::Notify;
|
||||
|
||||
pub use model::{
|
||||
Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, PermPayload, Source, State, Template,
|
||||
Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, NodeSpec, PermPayload, Source, State,
|
||||
Template,
|
||||
};
|
||||
|
||||
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain
|
||||
|
|
@ -206,6 +207,66 @@ impl JobQueue {
|
|||
Some(new_id)
|
||||
}
|
||||
|
||||
/// Append a whole *subgraph* into a live (non-terminal) DAG at runtime
|
||||
/// — the multi-node, multi-agent generalisation of [`Self::append_node`].
|
||||
/// Each [`NodeSpec`] carries its own `agent` and subgraph-relative `deps`
|
||||
/// (indices into `nodes`); this rebases those onto the DAG's node-id
|
||||
/// space (`id == index`, an invariant `append_node` also maintains) and
|
||||
/// attaches every subgraph *root* — a node with no internal deps — to
|
||||
/// `dep_on` with an `AfterOk` edge. Used by the startup sweep's
|
||||
/// `MetaLock` to grow per-agent rebuild subgraphs into the same boot DAG
|
||||
/// instead of fanning out child DAGs. Same call-*before*-`complete_node`
|
||||
/// contract as `append_node` (so the DAG can't roll terminal with the
|
||||
/// appended work still pending). Returns the new node ids; empty if the
|
||||
/// DAG is gone or `nodes` is empty.
|
||||
pub fn append_subgraph(&self, dag_id: u64, nodes: Vec<NodeSpec>, dep_on: NodeId) -> Vec<NodeId> {
|
||||
if nodes.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
let mut inner = self.inner.lock().expect("job_queue mutex poisoned");
|
||||
let Some(dag) = inner.dags.iter_mut().find(|d| d.id == dag_id) else {
|
||||
return Vec::new();
|
||||
};
|
||||
let base: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX);
|
||||
let mut new_ids = Vec::with_capacity(nodes.len());
|
||||
for (i, spec) in nodes.into_iter().enumerate() {
|
||||
let new_id: NodeId = base + u32::try_from(i).unwrap_or(u32::MAX);
|
||||
// Subgraph roots (no internal deps) hang off the emitting node;
|
||||
// internal deps rebase from subgraph-relative onto the DAG id
|
||||
// space (both start at `base`).
|
||||
let deps = if spec.deps.is_empty() {
|
||||
vec![model::Dep {
|
||||
on: dep_on,
|
||||
when: DepWhen::AfterOk,
|
||||
}]
|
||||
} else {
|
||||
spec.deps
|
||||
.into_iter()
|
||||
.map(|d| model::Dep {
|
||||
on: base + d.on,
|
||||
when: d.when,
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
dag.nodes.push(Node {
|
||||
id: new_id,
|
||||
agent: spec.agent,
|
||||
kind: spec.kind,
|
||||
deps,
|
||||
state: State::Queued,
|
||||
step: None,
|
||||
build_log_id: None,
|
||||
started_at: None,
|
||||
finished_at: None,
|
||||
error: None,
|
||||
});
|
||||
new_ids.push(new_id);
|
||||
}
|
||||
drop(inner);
|
||||
self.notify.notify_one();
|
||||
new_ids
|
||||
}
|
||||
|
||||
fn push_dag(inner: &mut Inner, spec: DagSpec) -> u64 {
|
||||
inner.next_id += 1;
|
||||
let id = inner.next_id;
|
||||
|
|
|
|||
Loading…
Reference in a new issue