jobq: a job asks for the ids it wants back
The operator's instruction on the issue was "the closure returns an array of guids, and enqueue_job returns the node ids in that order". What was here instead returned a HashMap of everything inserted, and no caller used the keys: submit dropped the return, insert_group did into_values(), and the scheduler ignored what append_subgraph handed back. The guid-keyed lookup was dead weight, and into_values() made that Vec arbitrarily ordered -- harmless only because nothing read it. insert_job now takes FnOnce(&JobBuilder) -> Vec<NodeGuid> and returns the matching ids positionally. A handle from another job is UnknownNode rather than a silent omission: the return is positional, so a short vector would misalign every id after it. c0re's Declare stays FnOnce(&Job) and the wrapper names no handles in one place, rather than ending seven templates in an empty vector -- a DAG is addressed by its container node, which submit inserts itself. That frees insert_group from needing every id, so the node_rt pre-seeding goes too: NodeRuntime is one Option field and every reader already tolerated a missing entry (entry().or_default(), get().and_then(), iter().find()). The tests are the argument for the shape: capturing a handle through a mutable binding to look it up in the map afterwards collapses into returning it and destructuring the result.
This commit is contained in:
parent
c82853af5a
commit
bf138ae79a
7 changed files with 147 additions and 76 deletions
|
|
@ -181,15 +181,18 @@ fn insert_group(
|
|||
inner: &mut QueueInner,
|
||||
declare: Declare,
|
||||
group_parent: Option<NodeId>,
|
||||
) -> anyhow::Result<Vec<NodeId>> {
|
||||
let ids = inner
|
||||
) -> anyhow::Result<()> {
|
||||
inner
|
||||
.sched
|
||||
.insert_job(group_parent, declare)
|
||||
.insert_job(group_parent, |b| {
|
||||
declare(b);
|
||||
// c0re names no handles: a DAG is addressed by its container node,
|
||||
// which `submit` inserts itself, and nothing downstream looks an
|
||||
// individual step up by id.
|
||||
Vec::new()
|
||||
})
|
||||
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
|
||||
for &id in ids.values() {
|
||||
inner.node_rt.insert(id, NodeRuntime::default());
|
||||
}
|
||||
Ok(ids.into_values().collect())
|
||||
Ok(())
|
||||
}
|
||||
|
||||
impl JobQueue {
|
||||
|
|
@ -255,12 +258,11 @@ impl JobQueue {
|
|||
/// gate — the children run once `dep_on` reaches `Finishing`. Because the
|
||||
/// emitting node stays `Finishing` until this appended subtree is terminal and
|
||||
/// the DAG's terminal node deps on the top root, roll-up keeps the DAG from
|
||||
/// settling early with no explicit wiring. Returns the new node ids; empty if
|
||||
/// the DAG is gone or `nodes` is empty.
|
||||
pub fn append_subgraph(&self, dag_id: u64, declare: Declare, dep_on: NodeId) -> Vec<NodeId> {
|
||||
/// settling early with no explicit wiring. A no-op if the DAG is gone.
|
||||
pub fn append_subgraph(&self, dag_id: u64, declare: Declare, dep_on: NodeId) {
|
||||
let mut inner = self.lock();
|
||||
if inner.container(dag_id).is_none() {
|
||||
return Vec::new();
|
||||
return;
|
||||
}
|
||||
// Insert the subgraph as a group rooted under the emitting node: the
|
||||
// subgraph's own root becomes a child of `dep_on`, its steps children of
|
||||
|
|
@ -268,20 +270,16 @@ impl JobQueue {
|
|||
// emitter stays `Finishing` until this appended subtree settles, and the
|
||||
// container node rolls up terminal only once its whole subtree (incl. this
|
||||
// appended work) has settled, so the DAG hook waits for free.
|
||||
let ids = match insert_group(&mut inner, declare, Some(dep_on)) {
|
||||
Ok(ids) => ids,
|
||||
Err(e) => {
|
||||
tracing::error!(
|
||||
dag = dag_id,
|
||||
error = %e,
|
||||
"job_queue: append_subgraph insert failed"
|
||||
);
|
||||
return Vec::new();
|
||||
}
|
||||
};
|
||||
if let Err(e) = insert_group(&mut inner, declare, Some(dep_on)) {
|
||||
tracing::error!(
|
||||
dag = dag_id,
|
||||
error = %e,
|
||||
"job_queue: append_subgraph insert failed"
|
||||
);
|
||||
return;
|
||||
}
|
||||
drop(inner);
|
||||
self.notify.notify_one();
|
||||
ids
|
||||
}
|
||||
|
||||
/// Claim every currently-runnable node, acquiring its resources, and mark it
|
||||
|
|
|
|||
|
|
@ -1252,12 +1252,11 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
|
|||
// BEFORE the emitting node is completed. Completing first would settle the
|
||||
// apply node `Done` with nothing under it, opening the tail's `AfterAny`
|
||||
// gate immediately and letting the deploy "finish" before it had built.
|
||||
let grown = q.append_subgraph(
|
||||
q.append_subgraph(
|
||||
id,
|
||||
templates::deploy_rebuild_nodes("agent-a", 11),
|
||||
apply.node_id,
|
||||
);
|
||||
assert!(!grown.is_empty(), "subgraph grafted onto the apply node");
|
||||
q.complete_node(apply.node_id, Ok(()));
|
||||
|
||||
// The grafted chain runs in rebuild order. `claim_one` asserts exactly one
|
||||
|
|
|
|||
Loading…
Reference in a new issue