forked from 1b-agent/fieldcraft-upstream
jobs: Cancel and job.cancel interrupt the running command (#371)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UhhZMBTviDQxZvwnyP89fo
This commit is contained in:
1 parent
5fcbee028e
commit
5aace9b5b3
8 files changed
+176
-8
No files matched your search
@@ -0,0 +1,130 @@
|
||||
//! A cancelled job ([`fieldcraft_model::progress::Progress::cancel`]) stops every solver and
|
||||
//! mesher promptly with `cancelled` and leaves the model as it was.
|
||||
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use fieldcraft_geometry::sample::Plate;
|
||||
use fieldcraft_model::progress::{self, Progress};
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use crate::Session;
|
||||
use crate::sample::impact;
|
||||
|
||||
/// How long a cancelled command may keep running: one increment, time step or pass, far
|
||||
/// below the minutes the runs below would take uncancelled.
|
||||
const PROMPT: Duration = Duration::from_secs(5);
|
||||
|
||||
/// Run `command` on `s` the way a job does and cancel it once it reports the stage `stage`
|
||||
/// (mid-run), or before it starts (`None`): it must fail with `cancelled`, soon after, leaving
|
||||
/// the model and the journal as they were.
|
||||
fn cancel(s: &mut Session, command: &str, params: Value, stage: Option<&str>) {
|
||||
let (model, journal) = (s.model.clone(), s.journal.len());
|
||||
let p = Progress::new();
|
||||
if stage.is_none() {
|
||||
p.cancel();
|
||||
}
|
||||
let (r, waited) = std::thread::scope(|scope| {
|
||||
let worker = {
|
||||
let p = p.clone();
|
||||
let s = &mut *s;
|
||||
let params = params.clone();
|
||||
scope.spawn(move || progress::scoped(Some(p), || s.execute(command, ¶ms)))
|
||||
};
|
||||
let mut at = Instant::now();
|
||||
if let Some(stage) = stage {
|
||||
let deadline = Instant::now() + Duration::from_secs(600);
|
||||
while !worker.is_finished() && Instant::now() < deadline {
|
||||
let now = p.snapshot();
|
||||
if now.stage.starts_with(stage) && now.done > 0.0 {
|
||||
at = Instant::now();
|
||||
p.cancel();
|
||||
break;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(1));
|
||||
}
|
||||
assert!(p.is_cancelled(), "{command}: never reached stage `{stage}` ({:?})", p.snapshot());
|
||||
}
|
||||
let r = worker.join().unwrap();
|
||||
(r, at.elapsed())
|
||||
});
|
||||
let e = r.expect_err(&format!("{command}: ran to the end although cancelled"));
|
||||
assert!(e.is_cancelled(), "{command}: {e}");
|
||||
assert_eq!(e.to_string(), "cancelled", "{command}");
|
||||
assert!(waited < PROMPT, "{command}: stopped {waited:?} after the cancel");
|
||||
assert!(s.model == model, "{command}: the model changed");
|
||||
assert_eq!(s.journal.len(), journal, "{command}: journalled");
|
||||
}
|
||||
|
||||
/// A sample's script run on a fresh session.
|
||||
fn session(script: &Value) -> Session {
|
||||
let mut s = Session::default();
|
||||
s.run_script(script).unwrap();
|
||||
s
|
||||
}
|
||||
|
||||
fn plate() -> (crate::tests::TempDir, Session) {
|
||||
let (dir, path) = crate::tests::sample_design_file(&Plate::default());
|
||||
let mut s = Session::default();
|
||||
s.execute("geometry.open", &json!({"path": path})).unwrap();
|
||||
(dir, s)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn explicit_stops_between_time_steps() {
|
||||
// A hundred times as long as the sample: hundreds of thousands of increments.
|
||||
let mut script = impact();
|
||||
for c in script.as_array_mut().unwrap() {
|
||||
if c["command"] == "model.step.create" {
|
||||
let p = &mut c["params"]["procedure"];
|
||||
(p["end_time"], p["output_interval"], p["history_interval"]) = (json!(4e-3), json!(4e-4), json!(2e-5));
|
||||
}
|
||||
}
|
||||
let mut s = session(&script);
|
||||
cancel(&mut s, "solve.explicit", json!({}), Some("Time"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fluid_stops_between_lattice_steps() {
|
||||
let mut s = Session::default();
|
||||
s.execute("file.sample", &json!({"name": "flow-past"})).unwrap();
|
||||
cancel(&mut s, "solve.fluid", json!({}), Some("Fluid steps"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nonlinear_stops_between_increments() {
|
||||
let mut s = Session::default();
|
||||
s.execute("file.sample", &json!({"name": "rubber-seal"})).unwrap();
|
||||
cancel(&mut s, "solve.nonlinear", json!({}), Some("Load factor"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transient_heat_stops_between_increments() {
|
||||
let mut s = Session::default();
|
||||
s.execute("file.sample", &json!({"name": "heated-bracket"})).unwrap();
|
||||
s.execute(
|
||||
"model.step.create",
|
||||
&json!({"name": "Warm", "procedure": {"type": "heat_transfer", "transient": {"end_time": 1e5, "increment": 1}},
|
||||
"boundary_conditions": ["Heater"], "loads": ["Air"]}),
|
||||
)
|
||||
.unwrap();
|
||||
cancel(&mut s, "solve.heat", json!({"step": "Warm"}), Some("Time increment"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn static_frequency_and_solve_all_stop_at_the_factorization() {
|
||||
let mut s = Session::default();
|
||||
s.execute("file.sample", &json!({"name": "cantilever"})).unwrap();
|
||||
cancel(&mut s, "solve.static", json!({}), None);
|
||||
cancel(&mut s, "solve.frequency", json!({}), None);
|
||||
cancel(&mut s, "solve.run", json!({}), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn meshers_stop_and_leave_no_mesh() {
|
||||
let (_dir, mut s) = plate();
|
||||
cancel(&mut s, "mesh.tet.generate", json!({"body": "Body1", "size": 0.4}), Some("Refinement round"));
|
||||
cancel(&mut s, "mesh.tet.generate", json!({"body": "Body1", "size": 2.5}), None);
|
||||
cancel(&mut s, "mesh.hex.generate", json!({"body": "Body1", "size": 2.5}), None);
|
||||
cancel(&mut s, "mesh.hex.octree", json!({"body": "Body1", "size": 5.0, "max_depth": 2}), None);
|
||||
assert!(s.model.nodes().is_empty());
|
||||
}
|
||||
@@ -220,7 +220,10 @@ pub(super) fn hex_blocks(s: &mut Session, p: &Value) -> Result<Value> {
|
||||
seeds: super::seeds::edge_seeds(s)?,
|
||||
..HexOptions::default()
|
||||
};
|
||||
let cm = hex_mesh_cells(&cells, &UniformSize(h), &options).map_err(|e| EngineError::Params(e.to_string()))?;
|
||||
let cm = hex_mesh_cells(&cells, &UniformSize(h), &options).map_err(|e| match e {
|
||||
fieldcraft_mesh::Error::Model(fieldcraft_model::Error::Cancelled) => EngineError::Model(fieldcraft_model::Error::Cancelled),
|
||||
e => EngineError::Params(e.to_string()),
|
||||
})?;
|
||||
let m = &cm.mesh;
|
||||
// Replace the cells' meshes (replace.rs); other bodies and the definitions stay.
|
||||
let model = &mut s.model;
|
||||
@@ -349,7 +352,7 @@ pub(super) fn hex_body(p: &Value, b: &Input, method: HexMethod, quadratic: bool,
|
||||
seeds: b.seeds.clone(),
|
||||
..HexOptions::default()
|
||||
};
|
||||
hex_mesh(&surface, &b.surfaces, &UniformSize(h), &options).map_err(|e| EngineError::Params(format!("body `{}`: {e}", b.name)))
|
||||
hex_mesh(&surface, &b.surfaces, &UniformSize(h), &options).map_err(|e| EngineError::of_body(&b.name, e))
|
||||
}
|
||||
|
||||
/// Add one body's hex mesh after the model's largest labels, tagging the elements with the body
|
||||
|
||||
@@ -271,7 +271,7 @@ pub(super) fn mesh_body(
|
||||
surface_volume: surface.volume(),
|
||||
})
|
||||
};
|
||||
mesh().map_err(|e| if several { EngineError::Params(format!("body `{}`: {e}", input.name)) } else { e })
|
||||
mesh().map_err(|e| if several && !e.is_cancelled() { EngineError::Params(format!("body `{}`: {e}", input.name)) } else { e })
|
||||
}
|
||||
|
||||
fn tet_generate(s: &mut Session, p: &Value) -> Result<Value> {
|
||||
|
||||
@@ -60,7 +60,7 @@ fn hex_octree(s: &mut Session, p: &Value) -> Result<Value> {
|
||||
let surface = BoundarySurface::new(&b.mesh.positions, &b.mesh.triangles, &b.mesh.tri_face, &surface_options)
|
||||
.map_err(|e| EngineError::Params(format!("body `{}`: {e}", b.name)))?;
|
||||
let left = OctreeOptions { max_elements: options.max_elements.saturating_sub(total).max(1), ..options.clone() };
|
||||
let mesh = octree_hex_mesh(&surface, &UniformSize(h), &left).map_err(|e| EngineError::Params(format!("body `{}`: {e}", b.name)))?;
|
||||
let mesh = octree_hex_mesh(&surface, &UniformSize(h), &left).map_err(|e| EngineError::of_body(&b.name, e))?;
|
||||
total += mesh.hexes.len();
|
||||
meshes.push((b, mesh));
|
||||
}
|
||||
|
||||
@@ -201,7 +201,9 @@ fn run_all(s: &mut Session, _: &Value) -> Result<Value> {
|
||||
let count = steps.len();
|
||||
for (index, (name, procedure, solve)) in steps.into_iter().enumerate() {
|
||||
fieldcraft_model::progress::step(&name, index + 1, count);
|
||||
let summary = solve(s, &json!({"step": name})).map_err(|e| EngineError::Params(format!("step `{name}`: {e}")))?;
|
||||
fieldcraft_model::progress::check()?;
|
||||
let summary =
|
||||
solve(s, &json!({"step": name})).map_err(|e| if e.is_cancelled() { e } else { EngineError::Params(format!("step `{name}`: {e}")) })?;
|
||||
done.push(json!({"step": name, "procedure": procedure, "summary": summary}));
|
||||
}
|
||||
Ok(json!({"steps": done}))
|
||||
|
||||
@@ -79,6 +79,29 @@ pub enum EngineError {
|
||||
}
|
||||
|
||||
impl EngineError {
|
||||
/// Whether the command stopped because it was cancelled
|
||||
/// ([`fieldcraft_model::progress::Progress::cancel`]).
|
||||
pub fn is_cancelled(&self) -> bool {
|
||||
use fieldcraft_model::Error::Cancelled;
|
||||
matches!(
|
||||
self,
|
||||
EngineError::Model(Cancelled)
|
||||
| EngineError::Mesh(fieldcraft_mesh::Error::Model(Cancelled))
|
||||
| EngineError::Solver(fieldcraft_solver::Error::Model(Cancelled))
|
||||
| EngineError::Explicit(fieldcraft_explicit::Error::Model(Cancelled))
|
||||
| EngineError::Fluid(fieldcraft_lbm::Error::Model(Cancelled) | fieldcraft_lbm::Error::Mesh(fieldcraft_mesh::Error::Model(Cancelled)))
|
||||
)
|
||||
}
|
||||
|
||||
/// A body's mesher failure as a parameter problem naming the body, or the cancellation as
|
||||
/// it is.
|
||||
pub(crate) fn of_body(body: &str, e: fieldcraft_mesh::Error) -> Self {
|
||||
match e {
|
||||
fieldcraft_mesh::Error::Model(fieldcraft_model::Error::Cancelled) => EngineError::Model(fieldcraft_model::Error::Cancelled),
|
||||
e => EngineError::Params(format!("body `{body}`: {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// A problem with the parameter at `path` (`elastic/young`).
|
||||
pub fn param(path: impl Into<String>, kind: schema::ProblemKind) -> Self {
|
||||
Self::Param(schema::Problem::at(path, kind))
|
||||
@@ -440,6 +463,9 @@ impl Session {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod cancel_tests;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -676,9 +676,12 @@ impl ModelApp {
|
||||
);
|
||||
}
|
||||
|
||||
/// Stop waiting for the background job; the model stays as it was.
|
||||
/// Cancel the background job: its command stops at its next check
|
||||
/// ([`fieldcraft_model::progress::check`]) and what it made is dropped; the model stays as
|
||||
/// it was.
|
||||
pub fn cancel_loading(&mut self) {
|
||||
if let Some(l) = self.loading.take() {
|
||||
l.progress.cancel();
|
||||
self.queued.clear();
|
||||
for w in &l.waiters {
|
||||
let _ = w.send(json!({"ok": false, "error": format!("cancelled: {}", l.title), "cancelled": true}));
|
||||
|
||||
Reference in new issue
Block a user