From 9beb2dccd5f536aab24940afd40ad70ca6600a5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Jan=C3=9Fen?= Date: Wed, 22 Jul 2026 07:35:50 +0200 Subject: [PATCH 1/3] Fix future hang when a queuing-system job dies without output (#1037) FileTaskScheduler only resolved a task's future once its _o.h5 output file appeared, so a job killed by the scheduler (walltime TIMEOUT, OOM, NODE_FAIL, scancel) never wrote that file and future.result() blocked forever. _check_task_output now falls back to querying the job status via pysqa when the output file is missing and fails the future if the job is no longer known to the queuing system. The status query is throttled per task (_JOB_STATUS_CHECK_INTERVAL) and imported lazily so subprocess-only task submissions never pay the pysqa import cost. Co-Authored-By: Claude Sonnet 5 --- src/executorlib/standalone/command_pysqa.py | 25 ++++ src/executorlib/task_scheduler/file/shared.py | 114 +++++++++++++++++- .../unit/task_scheduler/file/test_backend.py | 83 +++++++++++++ 3 files changed, 219 insertions(+), 3 deletions(-) diff --git a/src/executorlib/standalone/command_pysqa.py b/src/executorlib/standalone/command_pysqa.py index 77525a54..c910d040 100644 --- a/src/executorlib/standalone/command_pysqa.py +++ b/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, diff --git a/src/executorlib/task_scheduler/file/shared.py b/src/executorlib/task_scheduler/file/shared.py index 8d346b71..eb4a3e5c 100644 --- a/src/executorlib/task_scheduler/file/shared.py +++ b/src/executorlib/task_scheduler/file/shared.py @@ -2,7 +2,7 @@ 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 @@ -10,6 +10,12 @@ 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,15 +208,29 @@ 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. @@ -215,8 +238,27 @@ def _check_task_output( """ 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, diff --git a/tests/unit/task_scheduler/file/test_backend.py b/tests/unit/task_scheduler/file/test_backend.py index bf189f7e..736226dc 100644 --- a/tests/unit/task_scheduler/file/test_backend.py +++ b/tests/unit/task_scheduler/file/test_backend.py @@ -2,6 +2,7 @@ import os import shutil import unittest +from unittest.mock import patch from executorlib.standalone.select import FutureSelector @@ -219,5 +220,87 @@ def test_execute_function_error(self): with self.assertRaises(ValueError): future_file_obj.result() + 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() + + 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()) + + 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() + + 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) From b9db96d966eb51cb4efefd857192c4ed85553a8d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 22 Jul 2026 07:06:47 +0000 Subject: [PATCH 2/3] Skip pysqa status tests on Windows --- tests/unit/task_scheduler/file/test_backend.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/unit/task_scheduler/file/test_backend.py b/tests/unit/task_scheduler/file/test_backend.py index 736226dc..ce67b8ec 100644 --- a/tests/unit/task_scheduler/file/test_backend.py +++ b/tests/unit/task_scheduler/file/test_backend.py @@ -1,6 +1,7 @@ from concurrent.futures import Future import os import shutil +import sys import unittest from unittest.mock import patch @@ -220,6 +221,7 @@ def test_execute_function_error(self): with self.assertRaises(ValueError): future_file_obj.result() + @unittest.skipIf(sys.platform == "win32", "pysqa module patching not supported on Windows") 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 @@ -244,6 +246,7 @@ def test_check_task_output_dead_job_without_output(self): with self.assertRaises(RuntimeError): future_obj.result() + @unittest.skipIf(sys.platform == "win32", "pysqa module patching not supported on Windows") def test_check_task_output_job_still_running(self): cache_directory = os.path.abspath("executorlib_cache") os.makedirs(cache_directory, exist_ok=True) @@ -263,6 +266,7 @@ def test_check_task_output_job_still_running(self): status_mock.assert_called_once() self.assertFalse(future_obj.done()) + @unittest.skipIf(sys.platform == "win32", "pysqa module patching not supported on Windows") def test_check_task_output_status_check_is_throttled(self): cache_directory = os.path.abspath("executorlib_cache") os.makedirs(cache_directory, exist_ok=True) @@ -283,6 +287,7 @@ def test_check_task_output_status_check_is_throttled(self): ) status_mock.assert_called_once() + @unittest.skipIf(sys.platform == "win32", "pysqa module patching not supported on Windows") 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") From bca61640ddc26adbcdd2742fb0ee60d080bb2839 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:11:11 +0000 Subject: [PATCH 3/3] Skip pysqa tests when pysqa is not installed --- tests/unit/task_scheduler/file/test_backend.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/tests/unit/task_scheduler/file/test_backend.py b/tests/unit/task_scheduler/file/test_backend.py index ce67b8ec..ff3ba090 100644 --- a/tests/unit/task_scheduler/file/test_backend.py +++ b/tests/unit/task_scheduler/file/test_backend.py @@ -18,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 @@ -221,7 +228,7 @@ def test_execute_function_error(self): with self.assertRaises(ValueError): future_file_obj.result() - @unittest.skipIf(sys.platform == "win32", "pysqa module patching not supported on Windows") + @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 @@ -246,7 +253,7 @@ def test_check_task_output_dead_job_without_output(self): with self.assertRaises(RuntimeError): future_obj.result() - @unittest.skipIf(sys.platform == "win32", "pysqa module patching not supported on Windows") + @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) @@ -266,7 +273,7 @@ def test_check_task_output_job_still_running(self): status_mock.assert_called_once() self.assertFalse(future_obj.done()) - @unittest.skipIf(sys.platform == "win32", "pysqa module patching not supported on Windows") + @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) @@ -287,7 +294,7 @@ def test_check_task_output_status_check_is_throttled(self): ) status_mock.assert_called_once() - @unittest.skipIf(sys.platform == "win32", "pysqa module patching not supported on Windows") + @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")