Fix future hang when a queuing-system job dies without output

src/executorlib/standalone/command_pysqa.py

@@ -29,6 +29,31 @@ def pysqa_terminate(
             qa.delete_job(process_id=queue_id)
 
 
+def pysqa_get_status_of_job(
+    queue_id: int,
+    config_directory: Optional[str] = None,
+    backend: Optional[str] = None,
+) -> Optional[str]:
+    """
+    Query the status of a job from the queuing system.
+
+    Args:
+        queue_id (int): Queuing system ID of the job to check.
+        config_directory (str, optional): path to the config directory.
+        backend (str, optional): name of the backend used to spawn tasks ["slurm", "flux"].
+
+    Returns:
+        str: status of the job as reported by the queuing system, None if the queuing system no
+             longer knows about the job (e.g. it timed out, was killed or already finished).
+    """
+    qa = QueueAdapter(
+        directory=config_directory,
+        queue_type=backend,
+        execute_command=pysqa_execute_command,
+    )
+    return qa.get_status_of_job(process_id=queue_id)
+
+
 def pysqa_execute_command(
     commands: str,
     working_directory: Optional[str] = None,

src/executorlib/task_scheduler/file/shared.py

@@ -2,14 +2,20 @@
 import os
 import queue
 from concurrent.futures import Future
-from time import sleep
+from time import monotonic, sleep
 from typing import Any, Callable, Optional
 
 from executorlib.standalone.command import get_cache_execute_command
 from executorlib.standalone.hdf import get_cache_files, get_output, get_queue_id
 from executorlib.standalone.serialize import serialize_funct
 from executorlib.task_scheduler.file.spawner_subprocess import subprocess_terminate
 
+# Minimum time between two queries of the queuing system for the status of a task whose output
+# file has not appeared yet. Detecting a dead job (timeout, OOM, node failure, scancel, ...) relies
+# on this status query, but it must not be issued on every poll of the (much faster) refresh_rate
+# loop, as that would flood the queuing system commands (e.g. squeue/sacct) with requests.
+_JOB_STATUS_CHECK_INTERVAL = 30.0
+
 
 class FutureItem:
     def __init__(self, file_name: str, selector: Optional[int | str] = None):
@@ -92,6 +98,7 @@ def execute_tasks_h5(
     cache_dir_dict: dict = {}
     file_name_dict: dict = {}
     duplicate_dict: dict = {}
+    status_check_dict: dict = {}
     while True:
         task_dict = None
         with contextlib.suppress(queue.Empty):
@@ -104,6 +111,7 @@ def execute_tasks_h5(
                 process_dict=process_dict,
                 duplicate_dict=duplicate_dict,
                 cache_dir_dict=cache_dir_dict,
+                status_check_dict=status_check_dict,
                 terminate_function=terminate_function,
                 pysqa_config_directory=pysqa_config_directory,
                 backend=backend,
@@ -188,6 +196,7 @@ def execute_tasks_h5(
                 cache_dir_dict=cache_dir_dict,
                 process_dict=process_dict,
                 duplicate_dict=duplicate_dict,
+                status_check_dict=status_check_dict,
                 terminate_function=terminate_function,
                 pysqa_config_directory=pysqa_config_directory,
                 backend=backend,
@@ -199,24 +208,57 @@ def _check_task_output(
     task_key: str,
     future_obj: Future,
     cache_directory: str,
+    queue_id: Optional[int] = None,
+    pysqa_config_directory: Optional[str] = None,
+    backend: Optional[str] = None,
+    status_check_dict: Optional[dict] = None,
     duplicate_dict: Optional[dict] = None,
 ) -> Future:
     """
     Check the output of a task and set the result of the future object if available.
 
+    If the output file is missing and the task is running on a queuing system backend, this also
+    detects jobs which died without producing output (e.g. walltime TIMEOUT, OOM, NODE_FAIL or an
+    external scancel) by periodically querying the job status via pysqa and fails the future
+    instead of leaving it pending forever.
+
     Args:
         task_key (str): The key of the task.
         future_obj (Future): The future object associated with the task.
         cache_directory (str): The directory where the HDF5 files are stored.
+        queue_id (int, optional): The queuing system ID of the task, if submitted via pysqa.
+        pysqa_config_directory (str, optional): path to the pysqa config directory.
+        backend (str, optional): name of the backend used to spawn tasks ["slurm", "flux"].
+        status_check_dict (dict): Dictionary tracking when each task's job status was last queried,
+            used to throttle calls to the queuing system.
         duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
     Returns:
         Future: The updated future object.
 
     """
     file_name = os.path.join(cache_directory, task_key + "_o.h5")
     if not os.path.exists(file_name):
-        return future_obj
-    exec_flag, no_error_flag, result = get_output(file_name=file_name)
+        if not _job_died_without_output(
+            task_key=task_key,
+            file_name=file_name,
+            queue_id=queue_id,
+            pysqa_config_directory=pysqa_config_directory,
+            backend=backend,
+            status_check_dict=status_check_dict,
+        ):
+            return future_obj
+        exec_flag, no_error_flag, result = (
+            True,
+            False,
+            RuntimeError(
+                f"executorlib: queue job {queue_id} for task {task_key} terminated without "
+                "producing output (timeout, out-of-memory, node failure or cancellation)."
+            ),
+        )
+    else:
+        exec_flag, no_error_flag, result = get_output(file_name=file_name)
+    if status_check_dict is not None:
+        status_check_dict.pop(task_key, None)
     _update_future(
         future_obj=future_obj,
         exec_flag=exec_flag,
@@ -235,6 +277,56 @@ def _check_task_output(
     return future_obj
 
 
+def _job_died_without_output(
+    task_key: str,
+    file_name: str,
+    queue_id: Optional[int],
+    pysqa_config_directory: Optional[str],
+    backend: Optional[str],
+    status_check_dict: Optional[dict],
+) -> bool:
+    """
+    Check whether the queuing system job backing a task has died without ever writing its output
+    file. Only applies to queuing system backends (pysqa) and is throttled to at most once every
+    ``_JOB_STATUS_CHECK_INTERVAL`` seconds per task, to avoid flooding the queuing system with
+    status queries on every poll of the (much faster) refresh_rate loop.
+
+    Args:
+        task_key (str): The key of the task.
+        file_name (str): Path of the expected output HDF5 file.
+        queue_id (int, optional): The queuing system ID of the task.
+        pysqa_config_directory (str, optional): path to the pysqa config directory.
+        backend (str, optional): name of the backend used to spawn tasks ["slurm", "flux"].
+        status_check_dict (dict): Dictionary tracking when each task's job status was last queried.
+
+    Returns:
+        bool: True if the job is no longer known to the queuing system and still has no output.
+    """
+    if backend is None or queue_id is None:
+        return False
+    try:
+        # Imported lazily so subprocess-only (non-pysqa) task submissions - including every
+        # cache_serial.py backend subprocess spawned for local execution - never pay the cost of
+        # importing pysqa.
+        from executorlib.standalone.command_pysqa import pysqa_get_status_of_job
+    except ImportError:
+        return False
+    now = monotonic()
+    last_checked = (
+        status_check_dict.get(task_key, 0.0) if status_check_dict is not None else 0.0
+    )
+    if now - last_checked < _JOB_STATUS_CHECK_INTERVAL:
+        return False
+    if status_check_dict is not None:
+        status_check_dict[task_key] = now
+    status = pysqa_get_status_of_job(
+        queue_id=queue_id,
+        config_directory=pysqa_config_directory,
+        backend=backend,
+    )
+    return status is None and not os.path.exists(file_name)
+
+
 def _update_future(
     future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
 ) -> None:
@@ -321,6 +413,7 @@ def _refresh_memory_dict(
     cache_dir_dict: dict,
     process_dict: dict,
     duplicate_dict: Optional[dict] = None,
+    status_check_dict: Optional[dict] = None,
     terminate_function: Optional[Callable] = None,
     pysqa_config_directory: Optional[str] = None,
     backend: Optional[str] = None,
@@ -334,6 +427,8 @@ def _refresh_memory_dict(
         cache_dir_dict (dict): dictionary with task keys and cache directories
         process_dict (dict): dictionary with task keys and process reference.
         duplicate_dict (dict): dictionary with task keys and duplicate future objects.
+        status_check_dict (dict): dictionary with task keys and the last time their queuing system
+            job status was queried, used to throttle detection of jobs that died without output.
         terminate_function (callable): The function to terminate the tasks.
         pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
         backend (str): name of the backend used to spawn tasks.
@@ -351,11 +446,18 @@ def _refresh_memory_dict(
         pysqa_config_directory=pysqa_config_directory,
         backend=backend,
     )
+    if status_check_dict is not None:
+        for key in cancelled_lst:
+            status_check_dict.pop(key, None)
     memory_updated_dict = {
         key: _check_task_output(
             task_key=key,
             future_obj=value,
             cache_directory=cache_dir_dict[key],
+            queue_id=process_dict.get(key),
+            pysqa_config_directory=pysqa_config_directory,
+            backend=backend,
+            status_check_dict=status_check_dict,
             duplicate_dict=duplicate_dict,
         )
         for key, value in memory_dict.items()
@@ -443,6 +545,7 @@ def _shutdown_executor(
     process_dict: dict,
     cache_dir_dict: dict,
     duplicate_dict: Optional[dict] = None,
+    status_check_dict: Optional[dict] = None,
     terminate_function: Optional[Callable] = None,
     pysqa_config_directory: Optional[str] = None,
     backend: Optional[str] = None,
@@ -466,6 +569,8 @@ def _shutdown_executor(
         process_dict (dict): Mapping of task keys to process handles or queue IDs.
         duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
         cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
+        status_check_dict (dict): Mapping of task keys to the last time their queuing system job
+            status was queried, used to throttle detection of jobs that died without output.
         terminate_function (Callable, optional): Function used to terminate running processes.
         pysqa_config_directory (str, optional): Path to the pysqa config directory.
         backend (str, optional): Name of the backend ("slurm", "flux", or None for subprocess).
@@ -478,6 +583,7 @@ def _shutdown_executor(
                 cache_dir_dict=cache_dir_dict,
                 process_dict=process_dict,
                 duplicate_dict=duplicate_dict,
+                status_check_dict=status_check_dict,
                 terminate_function=terminate_function,
                 pysqa_config_directory=pysqa_config_directory,
                 backend=backend,
@@ -493,6 +599,7 @@ def _shutdown_executor(
                 cache_dir_dict=cache_dir_dict,
                 process_dict=process_dict,
                 duplicate_dict=duplicate_dict,
+                status_check_dict=status_check_dict,
                 terminate_function=terminate_function,
                 pysqa_config_directory=pysqa_config_directory,
                 backend=backend,
@@ -512,6 +619,7 @@ def _shutdown_executor(
             cache_dir_dict=cache_dir_dict,
             process_dict=process_dict,
             duplicate_dict=duplicate_dict,
+            status_check_dict=status_check_dict,
             terminate_function=terminate_function,
             pysqa_config_directory=pysqa_config_directory,
             backend=backend,

tests/unit/task_scheduler/file/test_backend.py

@@ -1,7 +1,9 @@
 from concurrent.futures import Future
 import os
 import shutil
+import sys
 import unittest
+from unittest.mock import patch
 
 from executorlib.standalone.select import FutureSelector
 
@@ -16,6 +18,13 @@
 except ImportError:
     skip_h5io_test = True
 
+try:
+    import pysqa  # noqa: F401
+
+    skip_pysqa_test = False
+except ImportError:
+    skip_pysqa_test = True
+
 
 def my_funct(a, b):
     return a + b
@@ -219,5 +228,91 @@ def test_execute_function_error(self):
         with self.assertRaises(ValueError):
             future_file_obj.result()
 
+    @unittest.skipIf(sys.platform == "win32" or skip_pysqa_test, "pysqa module patching not supported on Windows or when pysqa is not installed")
+    def test_check_task_output_dead_job_without_output(self):
+        # Reproduces https://github.com/pyiron/executorlib/issues/1037 : a queuing system job
+        # which dies without ever writing its output file (e.g. walltime TIMEOUT, OOM, NODE_FAIL
+        # or an external scancel) must fail the future instead of leaving it pending forever.
+        cache_directory = os.path.abspath("executorlib_cache")
+        os.makedirs(cache_directory, exist_ok=True)
+        task_key, data_dict = serialize_funct(fn=my_funct, fn_args=[1], fn_kwargs={"b": 2})
+        future_obj = Future()
+        with patch(
+            "executorlib.standalone.command_pysqa.pysqa_get_status_of_job",
+            return_value=None,
+        ) as status_mock:
+            _check_task_output(
+                task_key=task_key,
+                future_obj=future_obj,
+                cache_directory=cache_directory,
+                queue_id=123,
+                backend="slurm",
+            )
+        status_mock.assert_called_once()
+        self.assertTrue(future_obj.done())
+        with self.assertRaises(RuntimeError):
+            future_obj.result()
+
+    @unittest.skipIf(sys.platform == "win32" or skip_pysqa_test, "pysqa module patching not supported on Windows or when pysqa is not installed")
+    def test_check_task_output_job_still_running(self):
+        cache_directory = os.path.abspath("executorlib_cache")
+        os.makedirs(cache_directory, exist_ok=True)
+        task_key, data_dict = serialize_funct(fn=my_funct, fn_args=[1], fn_kwargs={"b": 2})
+        future_obj = Future()
+        with patch(
+            "executorlib.standalone.command_pysqa.pysqa_get_status_of_job",
+            return_value="running",
+        ) as status_mock:
+            _check_task_output(
+                task_key=task_key,
+                future_obj=future_obj,
+                cache_directory=cache_directory,
+                queue_id=123,
+                backend="slurm",
+            )
+        status_mock.assert_called_once()
+        self.assertFalse(future_obj.done())
+
+    @unittest.skipIf(sys.platform == "win32" or skip_pysqa_test, "pysqa module patching not supported on Windows or when pysqa is not installed")
+    def test_check_task_output_status_check_is_throttled(self):
+        cache_directory = os.path.abspath("executorlib_cache")
+        os.makedirs(cache_directory, exist_ok=True)
+        task_key, data_dict = serialize_funct(fn=my_funct, fn_args=[1], fn_kwargs={"b": 2})
+        status_check_dict = {}
+        with patch(
+            "executorlib.standalone.command_pysqa.pysqa_get_status_of_job",
+            return_value="running",
+        ) as status_mock:
+            for _ in range(3):
+                _check_task_output(
+                    task_key=task_key,
+                    future_obj=Future(),
+                    cache_directory=cache_directory,
+                    queue_id=123,
+                    backend="slurm",
+                    status_check_dict=status_check_dict,
+                )
+        status_mock.assert_called_once()
+
+    @unittest.skipIf(sys.platform == "win32" or skip_pysqa_test, "pysqa module patching not supported on Windows or when pysqa is not installed")
+    def test_check_task_output_no_backend_never_queries_status(self):
+        # subprocess-backed tasks (backend=None) must never trigger a queuing system status check.
+        cache_directory = os.path.abspath("executorlib_cache")
+        os.makedirs(cache_directory, exist_ok=True)
+        task_key, data_dict = serialize_funct(fn=my_funct, fn_args=[1], fn_kwargs={"b": 2})
+        future_obj = Future()
+        with patch(
+            "executorlib.standalone.command_pysqa.pysqa_get_status_of_job",
+        ) as status_mock:
+            _check_task_output(
+                task_key=task_key,
+                future_obj=future_obj,
+                cache_directory=cache_directory,
+                queue_id=123,
+                backend=None,
+            )
+        status_mock.assert_not_called()
+        self.assertFalse(future_obj.done())
+
     def tearDown(self):
         shutil.rmtree("executorlib_cache", ignore_errors=True)