Adding inputs for new "kill" mode of the on-error parameter

Author: AlexTate

cwltool/argparser.py

@@ -596,10 +596,11 @@ def arg_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         "--on-error",
         help="Desired workflow behavior when a step fails.  One of 'stop' (do "
-        "not submit any more steps) or 'continue' (may submit other steps that "
-        "are not downstream from the error). Default is 'stop'.",
+        "not submit any more steps), 'continue' (may submit other steps that "
+        "are not downstream from the error), or kill (same as stop, but also "
+        "terminates running jobs in the active step(s)). Default is 'stop'.",
         default="stop",
-        choices=("stop", "continue"),
+        choices=("stop", "continue", "kill"),
     )
 
     checkgroup = parser.add_mutually_exclusive_group()

cwltool/context.py

@@ -189,6 +189,7 @@ def __init__(self, kwargs: Optional[dict[str, Any]] = None) -> None:
         self.default_stderr: Optional[Union[IO[bytes], TextIO]] = None
         self.validate_only: bool = False
         self.validate_stdout: Optional["SupportsWrite[str]"] = None
+        self.kill_switch = threading.Event()
         super().__init__(kwargs)
         if self.tmp_outdir_prefix == "":
             self.tmp_outdir_prefix = self.tmpdir_prefix

cwltool/job.py

@@ -327,6 +327,7 @@ def stderr_stdout_log_path(
                 env=env,
                 cwd=self.outdir,
                 make_job_dir=lambda: runtimeContext.create_outdir(),
+                kill_switch=runtimeContext.kill_switch,
                 job_script_contents=job_script_contents,
                 timelimit=self.timelimit,
                 name=self.name,
@@ -492,8 +493,8 @@ def prepare_environment(
         # Set on ourselves
         self.environment = env
 
-    def process_monitor(self, sproc: "subprocess.Popen[str]") -> None:
-        """Watch a process, logging its max memory usage."""
+    def process_monitor(self, sproc: "subprocess.Popen[str]", kill_switch: threading.Event) -> None:
+        """Watch a process, logging its max memory usage or terminating it if kill_switch is activated."""
         monitor = psutil.Process(sproc.pid)
         # Value must be list rather than integer to utilise pass-by-reference in python
         memory_usage: MutableSequence[Optional[int]] = [None]
@@ -835,6 +836,7 @@ def docker_monitor(
         cleanup_cidfile: bool,
         docker_exe: str,
         process: "subprocess.Popen[str]",
+        kill_switch: threading.Event,
     ) -> None:
         """Record memory usage of the running Docker container."""
         # Todo: consider switching to `docker create` / `docker start`
@@ -911,6 +913,7 @@ def _job_popen(
     env: Mapping[str, str],
     cwd: str,
     make_job_dir: Callable[[], str],
+    kill_switch: threading.Event,
     job_script_contents: Optional[str] = None,
     timelimit: Optional[int] = None,
     name: Optional[str] = None,