Adding new choice to --on-error

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

@@ -172,7 +172,7 @@ def __init__(self, kwargs: Optional[dict[str, Any]] = None) -> None:
         self.select_resources: Optional[select_resources_callable] = None
         self.eval_timeout: float = 60
         self.postScatterEval: Optional[Callable[[CWLObjectType], Optional[CWLObjectType]]] = None
-        self.on_error: Union[Literal["stop"], Literal["continue"]] = "stop"
+        self.on_error: Union[Literal["stop"], Literal["continue"], Literal["kill"]] = "stop"
         self.strict_memory_limit: bool = False
         self.strict_cpu_limit: bool = False
         self.cidfile_dir: Optional[str] = None
@@ -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: Optional[threading.Event] = None
         super().__init__(kwargs)
         if self.tmp_outdir_prefix == "":
             self.tmp_outdir_prefix = self.tmpdir_prefix

cwltool/errors.py

@@ -18,3 +18,16 @@
 
 class ArgumentException(Exception):
     """Mismatched command line arguments provided."""
+
+
+class WorkflowKillSwitch(Exception):
+    """When processStatus != "success" and on-error=kill, raise this exception."""
+
+    def __init__(self, job_id: str, rcode: int) -> None:
+        """Record the job identifier and the error code."""
+        self.job_id = job_id
+        self.rcode = rcode
+
+    def __str__(self) -> str:
+        """Represent this exception as a string."""
+        return f"[job {self.job_id}] activated kill switch with return code {self.rcode}"

cwltool/executors.py

@@ -20,7 +20,7 @@
 from .context import RuntimeContext, getdefault
 from .cuda import cuda_version_and_device_count
 from .cwlprov.provenance_profile import ProvenanceProfile
-from .errors import WorkflowException
+from .errors import WorkflowException, WorkflowKillSwitch
 from .job import JobBase
 from .loghandler import _logger
 from .mutation import MutationManager
@@ -102,6 +102,7 @@
         runtime_context.mutation_manager = MutationManager()
         runtime_context.toplevel = True
         runtime_context.workflow_eval_lock = threading.Condition(threading.RLock())
+        runtime_context.kill_switch = threading.Event()
 
         job_reqs: Optional[list[CWLObjectType]] = None
         if "https://w3id.org/cwl/cwl#requirements" in job_order_object:
@@ -251,6 +252,11 @@
             WorkflowException,
         ):  # pylint: disable=try-except-raise
             raise
+        except WorkflowKillSwitch as err:
+            _logger.error(
+                f"Workflow kill switch activated by [job {err.job_id}] "
+                f"because on-error={runtime_context.on_error}"
+            )
         except Exception as err:
             logger.exception("Got workflow error")
             raise WorkflowException(str(err)) from err
@@ -323,6 +329,11 @@
         except WorkflowException as err:
             _logger.exception(f"Got workflow error: {err}")
             self.exceptions.append(err)
+        except WorkflowKillSwitch as err:
+            _logger.error(
+                f"Workflow kill switch activated by [job {err.job_id}] "
+                f"because on-error={runtime_context.on_error}"
+            )
         except Exception as err:  # pylint: disable=broad-except
             _logger.exception(f"Got workflow error: {err}")
             self.exceptions.append(WorkflowException(str(err)))
@@ -429,7 +440,13 @@
         logger: logging.Logger,
         runtime_context: RuntimeContext,
     ) -> None:
-        self.taskqueue: TaskQueue = TaskQueue(threading.Lock(), int(math.ceil(self.max_cores)))
+        if runtime_context.kill_switch is None:
+            runtime_context.kill_switch = threading.Event()
+
+        self.taskqueue: TaskQueue = TaskQueue(
+            threading.Lock(), int(math.ceil(self.max_cores)), runtime_context.kill_switch
+        )
+
         try:
             jobiter = process.job(job_order_object, self.output_callback, runtime_context)
 
@@ -457,9 +474,9 @@
             while self.taskqueue.in_flight > 0:
                 self.wait_for_next_completion(runtime_context)
                 self.run_job(None, runtime_context)
-
-            runtime_context.workflow_eval_lock.release()
         finally:
+            if (lock := runtime_context.workflow_eval_lock) is not None:
+                lock.release()
             self.taskqueue.drain()
             self.taskqueue.join()