diff --git a/README.md b/README.md index cad6303..47e45fb 100644 --- a/README.md +++ b/README.md @@ -134,7 +134,7 @@ And I stop the "prober" process ## Semantics worth knowing - **Processes live for one scenario.** The scenario boundary kills every process the scenario started, clears the context (environment, working directory, stdin, timeout) and forgets the last command; the end of the feature file does the same. Variables follow the host's own per-file rule. -- **…as long as `bddkit` itself gets to finish.** The cleanup runs through the host's end-of-scenario and end-of-file calls. If `bddkit` is killed by a signal mid-run — Ctrl-C, a CI job timeout — those calls never happen, and because every command runs in its own process group, the signal that reached `bddkit` does not reach them either. Find the leftovers with `ps -o pid,pgid,command` and end each group with `kill -- -`. On Windows this does not happen: the Job Object is closed with the `bddkit` process, and closing it kills the tree. Cleaning up on an interrupted run is tracked in [bddkit/bddkit#62](https://github.com/bddkit/bddkit/issues/62) (the host side) and [#1](https://github.com/bddkit/bddkit-exec/issues/1) (stopping a command that is still running when the interruption comes). +- **…as long as `bddkit` itself gets to finish.** The cleanup runs through the host's end-of-scenario and end-of-file calls. If `bddkit` is killed by a signal mid-run — Ctrl-C, a CI job timeout — those calls never happen, and because every command runs in its own process group, the signal that reached `bddkit` does not reach them either. Find the leftovers with `ps -o pid,pgid,command` and end each group with `kill -- -`. On Windows this does not happen: the Job Object is closed with the `bddkit` process, and closing it kills the tree. Cleaning up on an interrupted run is tracked in [bddkit/bddkit#62](https://github.com/bddkit/bddkit/issues/62) (the host side). Once the host drops the instance on an interruption, the plugin acts on it at once, even in the middle of `I run the command`: every process of the file is asked to stop, gets what is left of `stop_grace_secs`, and the interrupted step fails with "the run was interrupted". - **A one-shot command owns its whole tree.** When it exits, anything it left running in its process group or job is killed: a `sleep 30 &` would otherwise hold the output pipes open and outlive the run. Use `I start` for anything that should keep running. - **A non-zero exit is not a failure by itself**, exactly as an HTTP 500 is not — it is asserted. **A timeout is a failure**, exactly as a transport error is, and kills the whole tree. - **A mismatch waits only while the stream can still change.** On a running process it answers `not_yet`, so an armed eventual assertion looks again; on the last command, or on a process that has ended with both pipes closed, it fails at once instead of spending the whole timeout. diff --git a/src/exec/mod.rs b/src/exec/mod.rs index dfc1b15..ee777c0 100644 --- a/src/exec/mod.rs +++ b/src/exec/mod.rs @@ -178,12 +178,67 @@ fn reader( (capture, handle) } +/// The trees of one instance's live jobs, kept apart from the instance so an +/// interrupt can reach them while a step holds the instance lock. A tree is +/// listed from spawn until `Job::finish`, which removes it before reaping: +/// on Unix a group may only be signalled while its leader is unreaped. +pub struct Registry { + grace: Duration, + live: Mutex, +} + +#[derive(Default)] +struct Live { + trees: Vec>, + /// When `interrupt` asked, which is when the grace period began. + interrupted: Option, +} + +impl Registry { + pub fn new(grace: Duration) -> Arc { + Arc::new(Self { + grace, + live: Mutex::default(), + }) + } + + fn live(&self) -> std::sync::MutexGuard<'_, Live> { + self.live.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// Asks every listed tree to stop (SIGTERM / CTRL_BREAK) and starts the + /// grace clock. Killing after the grace is up to whoever owns the jobs: + /// `run` for the one in flight, `Instance::reset` for the rest. + pub fn interrupt(&self) { + let mut live = self.live(); + live.interrupted.get_or_insert_with(Instant::now); + if !self.grace.is_zero() { + for tree in &live.trees { + tree.interrupt(); + } + } + } + + pub fn interrupted(&self) -> bool { + self.live().interrupted.is_some() + } + + /// What is left of `stop_grace_secs`, counted from the interrupt when + /// there was one: the request already went out, so none is spent twice. + pub fn grace_left(&self) -> Duration { + self.live() + .interrupted + .map_or(self.grace, |at| self.grace.saturating_sub(at.elapsed())) + } +} + /// A started command: a one-shot run once it has finished, or a background /// process for as long as the scenario keeps it. pub struct Job { pub spec: Spec, child: Child, - tree: os::Tree, + tree: Arc, + registry: Arc, pub stdout: Arc>, pub stderr: Arc>, readers: Vec>, @@ -192,7 +247,7 @@ pub struct Job { } impl Job { - pub fn spawn(spec: Spec, cap: usize) -> Result { + pub fn spawn(spec: Spec, cap: usize, registry: &Arc) -> Result { let mut child = spec.command().spawn().map_err(|e| { format!( "cannot start `{}` in {}: {e}", @@ -201,7 +256,7 @@ impl Job { ) })?; let tree = match os::Tree::adopt(&mut child) { - Ok(tree) => tree, + Ok(tree) => Arc::new(tree), Err(e) => { let _ = child.kill(); let _ = child.wait(); @@ -218,10 +273,19 @@ impl Job { } let (stdout, out_reader) = reader(child.stdout.take().expect("piped stdout"), cap); let (stderr, err_reader) = reader(child.stderr.take().expect("piped stderr"), cap); + { + let mut live = registry.live(); + live.trees.push(Arc::clone(&tree)); + // Started after the interrupt: nobody is left to wait on it. + if live.interrupted.is_some() { + tree.kill(); + } + } Ok(Self { spec, child, tree, + registry: Arc::clone(registry), stdout, stderr, readers: vec![out_reader, err_reader], @@ -246,6 +310,11 @@ impl Job { return; } self.tree.kill(); + // Unlisted before the reap, under the lock `interrupt` signals with. + self.registry + .live() + .trees + .retain(|t| !Arc::ptr_eq(t, &self.tree)); if let Ok(status) = self.child.wait() { self.exit.get_or_insert(os::exit_of(status)); } @@ -280,28 +349,47 @@ impl Drop for Job { } } -/// How a one-shot run ended when it did not end on its own. -pub struct TimedOut(pub Job); +/// How a one-shot run ended. +pub enum Ended { + Done(Job), + /// Out of time; its tree was killed. + TimedOut(Job), + /// The instance is being dropped; the tree got its grace period. + Interrupted(Job), +} /// Runs to completion. Whatever the command left behind in its tree is /// killed once it exits: it would hold our pipes open, and nothing else /// would ever stop it. -pub fn run(spec: Spec, cap: usize, timeout: Duration) -> Result, String> { +pub fn run( + spec: Spec, + cap: usize, + timeout: Duration, + registry: &Arc, +) -> Result { let deadline = Instant::now() .checked_add(timeout) .ok_or_else(|| format!("a timeout of {timeout:?} is out of range"))?; - let mut job = Job::spawn(spec, cap)?; + let mut job = Job::spawn(spec, cap, registry)?; while job.poll().is_none() { + if registry.interrupted() { + stop_all([&mut job], registry.grace_left()); + return Ok(Ended::Interrupted(job)); + } if Instant::now() >= deadline { job.finish(); job.drain(DRAIN); - return Ok(Err(TimedOut(job))); + return Ok(Ended::TimedOut(job)); } thread::sleep(POLL); } job.finish(); job.drain(DRAIN); - Ok(Ok(job)) + // The interrupt's own request may be what ended it. + if registry.interrupted() { + return Ok(Ended::Interrupted(job)); + } + Ok(Ended::Done(job)) } /// A graceful request to every tree (SIGTERM on Unix, CTRL_BREAK on @@ -356,14 +444,19 @@ pub fn probe(config: &InstanceConfig) -> Result<(), String> { stdin: None, }; let described = spec.describe(); - match run(spec, 64 * 1024, Duration::from_secs(10))? { - Ok(job) if job.exit() == Some(Exit::Code(0)) => Ok(()), - Ok(job) => Err(format!( + match run( + spec, + 64 * 1024, + Duration::from_secs(10), + &Registry::new(Duration::ZERO), + )? { + Ended::Done(job) if job.exit() == Some(Exit::Code(0)) => Ok(()), + Ended::Done(job) => Err(format!( "`{described}` ended with {}: {}", job.exit().map_or("no exit".into(), |e| e.to_string()), text_of(&job.stderr).trim() )), - Err(_) => Err(format!("`{described}` did not finish within 10 seconds")), + _ => Err(format!("`{described}` did not finish within 10 seconds")), } } @@ -426,10 +519,19 @@ mod tests { lf(text_of(c)) } + fn registry() -> Arc { + Registry::new(Duration::from_secs(2)) + } + + fn spawn(spec: Spec) -> Job { + Job::spawn(spec, 1024, ®istry()).expect("spawn") + } + fn completed(spec: Spec, cap: usize) -> Job { - run(spec, cap, Duration::from_secs(10)) - .expect("spawn") - .unwrap_or_else(|_| panic!("timed out")) + match run(spec, cap, Duration::from_secs(10), ®istry()).expect("spawn") { + Ended::Done(job) => job, + _ => panic!("did not finish"), + } } #[test] @@ -455,9 +557,10 @@ mod tests { shell(script::HANG_WITH_CHILD), 1024, Duration::from_millis(500), + ®istry(), ) .expect("spawn"); - let Err(TimedOut(job)) = outcome else { + let Ended::TimedOut(job) = outcome else { panic!("it finished") }; assert!( @@ -487,7 +590,7 @@ mod tests { #[test] fn stop_all_ends_a_pipeline_within_the_grace_period() { - let mut job = Job::spawn(shell(script::PIPELINE), 1024).expect("spawn"); + let mut job = spawn(shell(script::PIPELINE)); let started = Instant::now(); stop_all([&mut job], Duration::from_secs(2)); assert!(started.elapsed() < Duration::from_secs(4)); @@ -500,7 +603,7 @@ mod tests { #[test] fn a_zero_grace_kills_without_asking() { - let mut job = Job::spawn(shell(script::SLEEP), 1024).expect("spawn"); + let mut job = spawn(shell(script::SLEEP)); let started = Instant::now(); stop_all([&mut job], Duration::ZERO); assert!(started.elapsed() < Duration::from_secs(1)); @@ -516,7 +619,7 @@ mod tests { #[test] fn a_process_that_ignores_sigterm_is_killed_when_its_grace_runs_out() { let spec = shell("trap '' TERM; echo ready; while :; do sleep 0.05; done"); - let mut job = Job::spawn(spec, 1024).expect("spawn"); + let mut job = spawn(spec); let deadline = Instant::now() + Duration::from_secs(5); while !text(&job.stdout).contains("ready") && Instant::now() < deadline { thread::sleep(POLL); @@ -531,7 +634,7 @@ mod tests { #[test] fn poll_sees_the_exit_before_the_process_is_reaped() { - let mut job = Job::spawn(shell(script::EXIT_7), 1024).expect("spawn"); + let mut job = spawn(shell(script::EXIT_7)); let deadline = Instant::now() + Duration::from_secs(5); while job.poll().is_none() && Instant::now() < deadline { thread::sleep(POLL); @@ -546,7 +649,7 @@ mod tests { #[cfg(unix)] #[test] fn on_unix_a_polled_leader_stays_a_zombie_until_finish() { - let mut job = Job::spawn(shell(script::EXIT_7), 1024).expect("spawn"); + let mut job = spawn(shell(script::EXIT_7)); let pid = job.child.id(); let deadline = Instant::now() + Duration::from_secs(5); while job.poll().is_none() && Instant::now() < deadline { @@ -559,7 +662,7 @@ mod tests { #[test] fn dropping_a_job_kills_its_tree() { - let job = Job::spawn(shell(script::SLEEP), 1024).expect("spawn"); + let job = spawn(shell(script::SLEEP)); let pid = job.child.id(); drop(job); assert!(!os::alive(pid)); @@ -571,7 +674,7 @@ mod tests { program: Program::Argv(vec!["no-such-binary-bddkit".into()]), ..shell("") }; - let e = Job::spawn(spec, 1024).err().expect("refused"); + let e = Job::spawn(spec, 1024, ®istry()).err().expect("refused"); assert!(e.contains("no-such-binary-bddkit"), "{e}"); } diff --git a/src/exec/windows.rs b/src/exec/windows.rs index 56a523f..a94874e 100644 --- a/src/exec/windows.rs +++ b/src/exec/windows.rs @@ -86,8 +86,9 @@ pub struct Tree { } // SAFETY: a job handle is a kernel object handle, usable from any thread; -// the plugin only ever touches it from one thread at a time anyway. +// TerminateJobObject and GenerateConsoleCtrlEvent take no exclusive access. unsafe impl Send for Tree {} +unsafe impl Sync for Tree {} impl Tree { /// Puts the suspended child in a fresh job, then lets it run. diff --git a/src/instance.rs b/src/instance.rs index 7982e2e..90f196c 100644 --- a/src/instance.rs +++ b/src/instance.rs @@ -5,7 +5,9 @@ use std::path::{Path, PathBuf}; use std::time::Duration; use crate::config::InstanceConfig; -use crate::exec::{self, Job, Program, Spec}; +use std::sync::Arc; + +use crate::exec::{self, Job, Program, Registry, Spec}; /// Per-scenario preparation, set with `the command … is` steps. #[derive(Default)] @@ -19,6 +21,8 @@ pub struct Context { pub struct Instance { pub config: InstanceConfig, + /// Reachable without the instance lock; see `Registry`. + pub registry: Arc, pub context: Context, pub last: Option, /// In start order, names unique within the scenario. @@ -28,6 +32,7 @@ pub struct Instance { impl Instance { pub fn new(config: InstanceConfig) -> Self { Self { + registry: Registry::new(config.stop_grace), config, context: Context::default(), last: None, @@ -37,7 +42,7 @@ impl Instance { /// Kills every process the scenario started and forgets the rest. pub fn reset(&mut self) { - let grace = self.config.stop_grace; + let grace = self.registry.grace_left(); exec::stop_all(self.processes.iter_mut().map(|(_, job)| job), grace); self.processes.clear(); self.last = None; @@ -133,8 +138,10 @@ mod tests { &std::env::temp_dir(), false, ); - i.processes - .push(("sleeper".into(), Job::spawn(spec, 1024).expect("spawn"))); + i.processes.push(( + "sleeper".into(), + Job::spawn(spec, 1024, &i.registry).expect("spawn"), + )); i.reset(); assert!(i.processes.is_empty() && i.context.env.is_empty() && i.last.is_none()); } diff --git a/src/lib.rs b/src/lib.rs index 696204c..1532540 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -116,13 +116,19 @@ pub extern "C" fn bddkit_probe_config(request: *const c_char) -> *mut c_char { }) } -/// A handle is an index into this table, never a pointer. Instances sit -/// behind `Arc` so a step never runs while this lock is held. -static INSTANCES: LazyLock>>>> = - LazyLock::new(Mutex::default); +/// An instance and the part of it a drop may touch while a step holds the +/// instance lock. +struct Entry { + registry: Arc, + instance: Mutex, +} + +/// A handle is an index into this table, never a pointer. Entries sit behind +/// `Arc` so a step never runs while this lock is held. +static INSTANCES: LazyLock>>> = LazyLock::new(Mutex::default); static NEXT_HANDLE: AtomicU64 = AtomicU64::new(1); -fn instance(handle: u64) -> Option>> { +fn instance(handle: u64) -> Option> { INSTANCES.lock().expect("instances").get(&handle).cloned() } @@ -136,10 +142,15 @@ pub extern "C" fn bddkit_init_instance(request: *const c_char) -> *mut c_char { Err(error) => return reply::err(error), }; let handle = NEXT_HANDLE.fetch_add(1, Ordering::SeqCst); + let instance = Instance::new(config); + let entry = Entry { + registry: Arc::clone(&instance.registry), + instance: Mutex::new(instance), + }; INSTANCES .lock() .expect("instances") - .insert(handle, Arc::new(Mutex::new(Instance::new(config)))); + .insert(handle, Arc::new(entry)); serde_json::json!({"ok": true, "handle": handle}).to_string() }) } @@ -155,12 +166,12 @@ pub extern "C" fn bddkit_dispatch( Ok(v) => v, Err(e) => return reply::fatal(&e.to_string(), &[]), }; - let Some(instance) = instance(handle) else { + let Some(entry) = instance(handle) else { return reply::fatal("unknown handle", &[]); }; // A poisoned lock means an earlier step panicked mid-way; the state // is still coherent enough to report on and to kill. - let mut instance = instance.lock().unwrap_or_else(|e| e.into_inner()); + let mut instance = entry.instance.lock().unwrap_or_else(|e| e.into_inner()); steps::route(&mut instance, step_index, &steps::Request::parse(&value)) }) } @@ -170,8 +181,9 @@ pub extern "C" fn bddkit_dispatch( #[unsafe(no_mangle)] pub extern "C" fn bddkit_reset_scenario(handle: u64) -> *mut c_char { guard("envelope", move || match instance(handle) { - Some(instance) => { - instance.lock().unwrap_or_else(|e| e.into_inner()).reset(); + Some(entry) => { + let mut instance = entry.instance.lock().unwrap_or_else(|e| e.into_inner()); + instance.reset(); reply::ok() } None => reply::err("unknown handle"), @@ -180,13 +192,17 @@ pub extern "C" fn bddkit_reset_scenario(handle: u64) -> *mut c_char { /// When the feature file ends, and swept at the end of the run for a file /// that panicked. The last chance to kill a process before the run exits. +/// Every process is asked to stop before the instance lock is awaited, so a +/// step still running holds the drop up for the grace period at most. #[unsafe(no_mangle)] pub extern "C" fn bddkit_drop_instance(handle: u64) -> *mut c_char { guard("envelope", move || { let removed = INSTANCES.lock().expect("instances").remove(&handle); match removed { - Some(instance) => { - instance.lock().unwrap_or_else(|e| e.into_inner()).reset(); + Some(entry) => { + entry.registry.interrupt(); + let mut instance = entry.instance.lock().unwrap_or_else(|e| e.into_inner()); + instance.reset(); reply::ok() } None => reply::err("unknown handle"), diff --git a/src/steps.rs b/src/steps.rs index 53d55d2..993df9c 100644 --- a/src/steps.rs +++ b/src/steps.rs @@ -16,7 +16,7 @@ use regex::Regex; use serde_json::{Map, Value, json}; use crate::config; -use crate::exec::{self, Exit, Job, Program, TimedOut}; +use crate::exec::{self, Ended, Exit, Job, Program}; use crate::instance::Instance; use crate::matchers::{self, Delimited}; use crate::reply::{self, Diagnostic}; @@ -574,9 +574,18 @@ fn run(instance: &mut Instance, req: &Request, program: Program) -> String { ); instance.last = None; let timeout = instance.timeout(); - match exec::run(spec, instance.config.max_output_bytes, timeout) { + let cap = instance.config.max_output_bytes; + match exec::run(spec, cap, timeout, &instance.registry) { Err(e) => reply::fatal(&e, &[]), - Ok(Err(TimedOut(mut job))) => { + Ok(Ended::Interrupted(mut job)) => { + trace(req, format_args!("interrupted")); + let msg = format!( + "`{}` was stopped: the run was interrupted", + job.spec.describe() + ); + fail(&mut job, req, &msg) + } + Ok(Ended::TimedOut(mut job)) => { trace(req, format_args!("timed out after {timeout:?}")); let msg = format!( "`{}` did not finish within {} seconds; its process tree was killed", @@ -585,7 +594,7 @@ fn run(instance: &mut Instance, req: &Request, program: Program) -> String { ); fail(&mut job, req, &msg) } - Ok(Ok(job)) => { + Ok(Ended::Done(job)) => { trace(req, format_args!("{}", describe_exit(job.exit()))); let code = match job.exit() { Some(Exit::Code(c)) => c.to_string(), @@ -626,7 +635,7 @@ fn start(instance: &mut Instance, req: &Request, name: &str, program: Program) - spec.cwd.display() ), ); - match Job::spawn(spec, instance.config.max_output_bytes) { + match Job::spawn(spec, instance.config.max_output_bytes, &instance.registry) { Ok(job) => { instance.processes.push((name.to_string(), job)); reply::passed() diff --git a/tests/exports.rs b/tests/exports.rs index 2dfa4d9..dac776f 100644 --- a/tests/exports.rs +++ b/tests/exports.rs @@ -298,3 +298,34 @@ fn a_mismatch_on_a_process_that_has_ended_is_fatal_not_not_yet() { } call(bddkit_drop_instance(handle)); } + +#[test] +fn a_drop_stops_a_running_command_at_once_and_the_step_says_why() { + let handle = init(json!({"timeout_secs": 60, "stop_grace_secs": 2})); + let running = std::thread::spawn(move || { + dispatch( + handle, + "^I run the command \"(?P", + &[line::SLEEP], + None, + ) + }); + // Long enough for the command to be spawned and polled. A tree that + // ignores the request (ping under CTRL_BREAK) runs out the 2 s grace. + std::thread::sleep(std::time::Duration::from_millis(500)); + let started = std::time::Instant::now(); + assert_eq!(call(bddkit_drop_instance(handle)), json!({"ok": true})); + let reply = running.join().expect("the dispatch returned"); + assert!( + started.elapsed() < std::time::Duration::from_secs(4), + "the drop waited for the step: {:?}", + started.elapsed() + ); + assert_eq!(reply["status"], "fatal", "{reply}"); + assert!( + reply["error"] + .as_str() + .is_some_and(|e| e.contains("run was interrupted")), + "{reply}" + ); +}