This commit brings kill switch initialization and monitoring to the TaskQueue. This helps to better synchronize the kill switch event and avoid adding/executing tasks after the switch has been set.

This approach is tighter than my previous draft, but a race condition still exists where a task might be started after the kill switch has been set and announced. If this happens then the leaked job's monitor function will kill it and the subprocess' lifespan will be a maximum of the monitor's timer interval (currently 1 second). So when this rare event happens, the console output will be potentially confusing since it will show a new job starting after the kill switch has been announced.

Author: AlexTate

cwltool/executors.py

@@ -439,7 +439,7 @@ def run_jobs(
         logger: logging.Logger,
         runtime_context: RuntimeContext,
     ) -> None:
-        self.taskqueue: TaskQueue = TaskQueue(threading.Lock(), int(math.ceil(self.max_cores)))
+        self.taskqueue: TaskQueue = TaskQueue(threading.Lock(), int(math.ceil(self.max_cores)), runtime_context)
         try:
             jobiter = process.job(job_order_object, self.output_callback, runtime_context)
 

cwltool/task_queue.py

@@ -8,6 +8,7 @@
 from typing import Callable, Optional
 
 from .errors import WorkflowKillSwitch
+from .context import RuntimeContext
 from .loghandler import _logger
 
 
@@ -34,7 +35,7 @@ class TaskQueue:
     in_flight: int = 0
     """The number of tasks in the queue."""
 
-    def __init__(self, lock: threading.Lock, thread_count: int):
+    def __init__(self, lock: threading.Lock, thread_count: int, runtime_context: RuntimeContext):
         """Create a new task queue using the specified lock and number of threads."""
         self.thread_count = thread_count
         self.task_queue: queue.Queue[Optional[Callable[[], None]]] = queue.Queue(
@@ -44,6 +45,11 @@ def __init__(self, lock: threading.Lock, thread_count: int):
         self.lock = lock
         self.error: Optional[BaseException] = None
 
+        if runtime_context.kill_switch is None:
+            self.kill_switch = runtime_context.kill_switch = threading.Event()
+        else:
+            self.kill_switch = runtime_context.kill_switch
+
         for _r in range(0, self.thread_count):
             t = threading.Thread(target=self._task_queue_func)
             self.task_queue_threads.append(t)
@@ -52,11 +58,12 @@ def __init__(self, lock: threading.Lock, thread_count: int):
     def _task_queue_func(self) -> None:
         while True:
             task = self.task_queue.get()
-            if task is None:
+            if task is None or self.kill_switch.is_set():
                 return
             try:
                 task()
             except WorkflowKillSwitch:
+                self.kill_switch.set()
                 self.drain()
                 break
             except BaseException as e:  # noqa: B036
@@ -96,7 +103,10 @@ def add(
             try:
                 if unlock is not None:
                     unlock.release()
-                if check_done is not None and check_done.is_set():
+                if (
+                        (check_done is not None and check_done.is_set())
+                        or self.kill_switch.is_set()
+                ):
                     with self.lock:
                         self.in_flight -= 1
                     return