From d2389675c0dc18d33db56904753c9919fc0b7d8b Mon Sep 17 00:00:00 2001 From: Jan Janssen Date: Fri, 3 Jul 2026 17:55:17 +0200 Subject: [PATCH 1/3] [Maintenance] Refactor pysqa interface --- src/executorlib/executor/single.py | 4 +- .../{scheduler.py => command_pysqa.py} | 2 +- src/executorlib/task_scheduler/file/shared.py | 4 +- .../task_scheduler/file/spawner_pysqa.py | 6 +-- .../task_scheduler/file/spawner_subprocess.py | 4 +- .../task_scheduler/file/task_scheduler.py | 16 ++++---- .../interactive/spawner_pysqa.py | 4 +- tests/unit/executor/test_flux_cluster.py | 4 +- tests/unit/task_scheduler/file/test_mpi.py | 4 +- tests/unit/task_scheduler/file/test_serial.py | 38 +++++++++---------- 10 files changed, 43 insertions(+), 43 deletions(-) rename src/executorlib/standalone/{scheduler.py => command_pysqa.py} (98%) diff --git a/src/executorlib/executor/single.py b/src/executorlib/executor/single.py index 1aa52605e..3b2b91142 100644 --- a/src/executorlib/executor/single.py +++ b/src/executorlib/executor/single.py @@ -393,7 +393,7 @@ def __init__( ) if not plot_dependency_graph: from executorlib.task_scheduler.file.spawner_subprocess import ( - execute_in_subprocess, + subprocess_execute, ) from executorlib.task_scheduler.file.task_scheduler import ( create_file_executor, @@ -415,7 +415,7 @@ def __init__( block_allocation=block_allocation, init_function=init_function, disable_dependencies=disable_dependencies, - execute_function=execute_in_subprocess, + execute_function=subprocess_execute, wait=wait, refresh_rate=refresh_rate, validator=validate_resource_dict_with_optional_keys, diff --git a/src/executorlib/standalone/scheduler.py b/src/executorlib/standalone/command_pysqa.py similarity index 98% rename from src/executorlib/standalone/scheduler.py rename to src/executorlib/standalone/command_pysqa.py index bc68187b0..77525a540 100644 --- a/src/executorlib/standalone/scheduler.py +++ b/src/executorlib/standalone/command_pysqa.py @@ -5,7 +5,7 @@ from pysqa import QueueAdapter -def terminate_with_pysqa( +def pysqa_terminate( queue_id: int, config_directory: Optional[str] = None, backend: Optional[str] = None, diff --git a/src/executorlib/task_scheduler/file/shared.py b/src/executorlib/task_scheduler/file/shared.py index ae7b43b71..8d346b712 100644 --- a/src/executorlib/task_scheduler/file/shared.py +++ b/src/executorlib/task_scheduler/file/shared.py @@ -8,7 +8,7 @@ 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 terminate_subprocess +from executorlib.task_scheduler.file.spawner_subprocess import subprocess_terminate class FutureItem: @@ -381,7 +381,7 @@ def _cancel_processes( 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. """ - if terminate_function is not None and terminate_function == terminate_subprocess: + if terminate_function is not None and terminate_function == subprocess_terminate: for task in process_dict.values(): terminate_function(task=task) elif terminate_function is not None and backend is not None: diff --git a/src/executorlib/task_scheduler/file/spawner_pysqa.py b/src/executorlib/task_scheduler/file/spawner_pysqa.py index 759c6ec53..ea749c97c 100644 --- a/src/executorlib/task_scheduler/file/spawner_pysqa.py +++ b/src/executorlib/task_scheduler/file/spawner_pysqa.py @@ -9,7 +9,7 @@ from executorlib.standalone.interactive.spawner import ( set_current_directory_in_environment, ) -from executorlib.standalone.scheduler import pysqa_execute_command, terminate_with_pysqa +from executorlib.standalone.command_pysqa import pysqa_execute_command, pysqa_terminate def execute_with_pysqa( @@ -122,7 +122,7 @@ def terminate_tasks_in_cache( for f in hdf5_file_lst: queue_id = get_queue_id(f) if queue_id is not None: - terminate_with_pysqa( + pysqa_terminate( queue_id=queue_id, config_directory=pysqa_config_directory, backend=backend, @@ -148,7 +148,7 @@ def terminate_task_in_cache( file_name = os.path.join(cache_directory, cache_key + "_i.h5") queue_id = get_queue_id(file_name=file_name) if queue_id is not None: - terminate_with_pysqa( + pysqa_terminate( queue_id=queue_id, config_directory=pysqa_config_directory, backend=backend, diff --git a/src/executorlib/task_scheduler/file/spawner_subprocess.py b/src/executorlib/task_scheduler/file/spawner_subprocess.py index b4582c756..febeedc7b 100644 --- a/src/executorlib/task_scheduler/file/spawner_subprocess.py +++ b/src/executorlib/task_scheduler/file/spawner_subprocess.py @@ -10,7 +10,7 @@ ) -def execute_in_subprocess( +def subprocess_execute( command: list, file_name: str, data_dict: dict, @@ -65,7 +65,7 @@ def execute_in_subprocess( return subprocess.Popen(command, universal_newlines=True, cwd=cwd) -def terminate_subprocess(task): +def subprocess_terminate(task): """ Terminate a subprocess and wait for it to complete. diff --git a/src/executorlib/task_scheduler/file/task_scheduler.py b/src/executorlib/task_scheduler/file/task_scheduler.py index 53684f2ef..ef989ba9d 100644 --- a/src/executorlib/task_scheduler/file/task_scheduler.py +++ b/src/executorlib/task_scheduler/file/task_scheduler.py @@ -12,17 +12,17 @@ from executorlib.task_scheduler.base import TaskSchedulerBase, validate_resource_dict from executorlib.task_scheduler.file.shared import execute_tasks_h5 from executorlib.task_scheduler.file.spawner_subprocess import ( - execute_in_subprocess, - terminate_subprocess, + subprocess_execute, + subprocess_terminate, ) try: - from executorlib.standalone.scheduler import terminate_with_pysqa + from executorlib.standalone.command_pysqa import pysqa_terminate from executorlib.task_scheduler.file.spawner_pysqa import execute_with_pysqa except ImportError: # If pysqa is not available fall back to executing tasks in a subprocess - execute_with_pysqa = execute_in_subprocess # type: ignore - terminate_with_pysqa = None # type: ignore + execute_with_pysqa = subprocess_execute # type: ignore + pysqa_terminate = None # type: ignore class FileTaskScheduler(TaskSchedulerBase): @@ -126,10 +126,10 @@ def create_file_executor( check_executor(executor=flux_executor) check_nested_flux_executor(nested_flux_executor=flux_executor_nesting) check_flux_log_files(flux_log_files=flux_log_files) - if execute_function != execute_in_subprocess: - terminate_function = terminate_with_pysqa # type: ignore + if execute_function != subprocess_execute: + terminate_function = pysqa_terminate # type: ignore else: - terminate_function = terminate_subprocess # type: ignore + terminate_function = subprocess_terminate # type: ignore return FileTaskScheduler( executor_kwargs=executor_kwargs, pysqa_config_directory=pysqa_config_directory, diff --git a/src/executorlib/task_scheduler/interactive/spawner_pysqa.py b/src/executorlib/task_scheduler/interactive/spawner_pysqa.py index ccefe1d0f..d17f1511d 100644 --- a/src/executorlib/task_scheduler/interactive/spawner_pysqa.py +++ b/src/executorlib/task_scheduler/interactive/spawner_pysqa.py @@ -10,7 +10,7 @@ BaseSpawner, set_current_directory_in_environment, ) -from executorlib.standalone.scheduler import pysqa_execute_command, terminate_with_pysqa +from executorlib.standalone.command_pysqa import pysqa_execute_command, pysqa_terminate from executorlib.task_scheduler.base import validate_resource_dict from executorlib.task_scheduler.interactive.blockallocation import ( BlockAllocationTaskScheduler, @@ -162,7 +162,7 @@ def shutdown(self, wait: bool = True): wait (bool, optional): Whether to wait for the interface to shutdown. Defaults to True. """ if self._process is not None: - terminate_with_pysqa( + pysqa_terminate( queue_id=self._process, config_directory=self._config_directory, backend=self._backend, diff --git a/tests/unit/executor/test_flux_cluster.py b/tests/unit/executor/test_flux_cluster.py index 8071a6584..21b23b5e8 100644 --- a/tests/unit/executor/test_flux_cluster.py +++ b/tests/unit/executor/test_flux_cluster.py @@ -13,7 +13,7 @@ from executorlib import terminate_tasks_in_cache, terminate_task_in_cache from executorlib.standalone.hdf import dump from executorlib.task_scheduler.file.spawner_pysqa import execute_with_pysqa - from executorlib.standalone.scheduler import terminate_with_pysqa + from executorlib.standalone.command_pysqa import pysqa_terminate from executorlib.task_scheduler.interactive.spawner_pysqa import PysqaSpawner skip_flux_test = "FLUX_URI" not in os.environ @@ -245,7 +245,7 @@ def test_pysqa_interface(self): cache_directory="executorlib_cache", backend="flux" ) - self.assertIsNone(terminate_with_pysqa(queue_id=queue_id, backend="flux")) + self.assertIsNone(pysqa_terminate(queue_id=queue_id, backend="flux")) def test_executor_existing_files(self): with FluxClusterExecutor( diff --git a/tests/unit/task_scheduler/file/test_mpi.py b/tests/unit/task_scheduler/file/test_mpi.py index 76c52f587..6c345d4c6 100644 --- a/tests/unit/task_scheduler/file/test_mpi.py +++ b/tests/unit/task_scheduler/file/test_mpi.py @@ -5,7 +5,7 @@ try: from executorlib.task_scheduler.file.task_scheduler import FileTaskScheduler - from executorlib.task_scheduler.file.spawner_subprocess import execute_in_subprocess + from executorlib.task_scheduler.file.spawner_subprocess import subprocess_execute skip_h5py_test = False except ImportError: @@ -30,7 +30,7 @@ def mpi_funct(i): class TestCacheExecutorMPI(unittest.TestCase): def test_executor(self): with FileTaskScheduler( - executor_kwargs={"cores": 2}, execute_function=execute_in_subprocess + executor_kwargs={"cores": 2}, execute_function=subprocess_execute ) as exe: fs1 = exe.submit(mpi_funct, 1) self.assertFalse(fs1.done()) diff --git a/tests/unit/task_scheduler/file/test_serial.py b/tests/unit/task_scheduler/file/test_serial.py index 030544754..bd89b6b66 100644 --- a/tests/unit/task_scheduler/file/test_serial.py +++ b/tests/unit/task_scheduler/file/test_serial.py @@ -8,8 +8,8 @@ try: from executorlib.task_scheduler.file.spawner_subprocess import ( - execute_in_subprocess, - terminate_subprocess, + subprocess_execute, + subprocess_terminate, ) from executorlib.task_scheduler.file.task_scheduler import FileTaskScheduler, create_file_executor from executorlib.task_scheduler.file.shared import execute_tasks_h5, _convert_args_and_kwargs @@ -36,21 +36,21 @@ def get_error(a): ) class TestCacheExecutorSerial(unittest.TestCase): def test_submit_with_positional_and_keyword_args(self): - with FileTaskScheduler(execute_function=execute_in_subprocess) as exe: + with FileTaskScheduler(execute_function=subprocess_execute) as exe: fs1 = exe.submit(my_funct, 1, b=2) self.assertFalse(fs1.done()) self.assertEqual(fs1.result(), 3) self.assertTrue(fs1.done()) def test_submit_with_custom_cache_key(self): - with FileTaskScheduler(execute_function=execute_in_subprocess) as exe: + with FileTaskScheduler(execute_function=subprocess_execute) as exe: fs1 = exe.submit(my_funct, 1, b=2, resource_dict={"cache_key": "a/b/c"}) self.assertFalse(fs1.done()) self.assertEqual(fs1.result(), 3) self.assertTrue(fs1.done()) def test_submit_dependency_with_keyword_arg_future(self): - with FileTaskScheduler(execute_function=execute_in_subprocess) as exe: + with FileTaskScheduler(execute_function=subprocess_execute) as exe: fs1 = exe.submit(my_funct, 1, b=2) fs2 = exe.submit(my_funct, 1, b=fs1) self.assertFalse(fs2.done()) @@ -70,7 +70,7 @@ def test_create_file_executor_error(self): def test_submit_dependency_raises_when_dependencies_disabled(self): with self.assertRaises(ValueError): with FileTaskScheduler( - execute_function=execute_in_subprocess, disable_dependencies=True + execute_function=subprocess_execute, disable_dependencies=True ) as exe: fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2)) fs.result() @@ -78,7 +78,7 @@ def test_submit_dependency_raises_when_dependencies_disabled(self): def test_executor_working_directory(self): cwd = os.path.join(os.path.dirname(__file__), "..", "..", "..", "static") with FileTaskScheduler( - executor_kwargs={"cwd": cwd}, execute_function=execute_in_subprocess + executor_kwargs={"cwd": cwd}, execute_function=subprocess_execute ) as exe: fs1 = exe.submit(list_files_in_working_directory) self.assertEqual(fs1.result(), os.listdir(cwd)) @@ -86,7 +86,7 @@ def test_executor_working_directory(self): def test_executor_error(self): cwd = os.path.join(os.path.dirname(__file__), "..", "..", "..", "static") with FileTaskScheduler( - executor_kwargs={"cwd": cwd}, execute_function=execute_in_subprocess + executor_kwargs={"cwd": cwd}, execute_function=subprocess_execute ) as exe: fs1 = exe.submit(get_error, a=1) with self.assertRaises(ValueError): @@ -97,7 +97,7 @@ def test_executor_error_file(self): cwd = os.path.join(os.path.dirname(__file__), "..", "..", "..", "static") with FileTaskScheduler( executor_kwargs={"cwd": cwd, "error_log_file": "error.out"}, - execute_function=execute_in_subprocess + execute_function=subprocess_execute ) as exe: fs1 = exe.submit(get_error, a=1) with self.assertRaises(ValueError): @@ -125,9 +125,9 @@ def test_executor_function(self): target=execute_tasks_h5, kwargs={ "future_queue": q, - "execute_function": execute_in_subprocess, + "execute_function": subprocess_execute, "executor_kwargs": {"cores": 1, "cwd": None, "cache_directory": cache_dir}, - "terminate_function": terminate_subprocess, + "terminate_function": subprocess_terminate, }, ) process.start() @@ -165,9 +165,9 @@ def test_executor_function_dependence_kwargs(self): target=execute_tasks_h5, kwargs={ "future_queue": q, - "execute_function": execute_in_subprocess, + "execute_function": subprocess_execute, "executor_kwargs": {"cores": 1, "cwd": None, "cache_directory": cache_dir}, - "terminate_function": terminate_subprocess, + "terminate_function": subprocess_terminate, }, ) process.start() @@ -205,9 +205,9 @@ def test_executor_function_dependence_args(self): target=execute_tasks_h5, kwargs={ "future_queue": q, - "execute_function": execute_in_subprocess, + "execute_function": subprocess_execute, "executor_kwargs": {"cores": 1, "cache_directory": cache_dir}, - "terminate_function": terminate_subprocess, + "terminate_function": subprocess_terminate, }, ) process.start() @@ -218,12 +218,12 @@ def test_executor_function_dependence_args(self): process.join() def test_execute_in_subprocess(self): - process = execute_in_subprocess( + process = subprocess_execute( command=["sleep", "5"], file_name="test.h5", data_dict={"fn": sleep, "args": (5,)}, ) - self.assertIsNone(terminate_subprocess(task=process)) + self.assertIsNone(subprocess_terminate(task=process)) def test_execute_in_subprocess_errors(self): file_name = os.path.abspath(os.path.join(os.path.dirname(__file__), "executorlib_cache", "test.h5")) @@ -231,14 +231,14 @@ def test_execute_in_subprocess_errors(self): with open(file_name, "w") as f: f.write("test") with self.assertRaises(ValueError): - execute_in_subprocess( + subprocess_execute( file_name=file_name, data_dict={}, command=[], config_directory="test", ) with self.assertRaises(ValueError): - execute_in_subprocess( + subprocess_execute( file_name=file_name, data_dict={}, command=[], From 82fd56fd535fb46bc0fea48460789435bac899a4 Mon Sep 17 00:00:00 2001 From: Jan Janssen Date: Fri, 3 Jul 2026 17:55:59 +0200 Subject: [PATCH 2/3] fixes --- tests/unit/standalone/test_slurm_command.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/standalone/test_slurm_command.py b/tests/unit/standalone/test_slurm_command.py index 140a45d06..d251d1da9 100644 --- a/tests/unit/standalone/test_slurm_command.py +++ b/tests/unit/standalone/test_slurm_command.py @@ -2,7 +2,7 @@ from executorlib.standalone.command import generate_slurm_command try: - from executorlib.standalone.scheduler import pysqa_execute_command + from executorlib.standalone.command_pysqa import pysqa_execute_command skip_pysqa_test = False except ImportError: From 6c4769fb7c01ccbb14d737cee846cf3dd704b504 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 3 Jul 2026 17:09:14 +0000 Subject: [PATCH 3/3] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/executorlib/task_scheduler/file/spawner_pysqa.py | 2 +- src/executorlib/task_scheduler/interactive/spawner_pysqa.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/executorlib/task_scheduler/file/spawner_pysqa.py b/src/executorlib/task_scheduler/file/spawner_pysqa.py index ea749c97c..e2c8f4082 100644 --- a/src/executorlib/task_scheduler/file/spawner_pysqa.py +++ b/src/executorlib/task_scheduler/file/spawner_pysqa.py @@ -4,12 +4,12 @@ from pysqa import QueueAdapter +from executorlib.standalone.command_pysqa import pysqa_execute_command, pysqa_terminate from executorlib.standalone.hdf import dump, get_queue_id from executorlib.standalone.inputcheck import check_file_exists from executorlib.standalone.interactive.spawner import ( set_current_directory_in_environment, ) -from executorlib.standalone.command_pysqa import pysqa_execute_command, pysqa_terminate def execute_with_pysqa( diff --git a/src/executorlib/task_scheduler/interactive/spawner_pysqa.py b/src/executorlib/task_scheduler/interactive/spawner_pysqa.py index d17f1511d..f58355437 100644 --- a/src/executorlib/task_scheduler/interactive/spawner_pysqa.py +++ b/src/executorlib/task_scheduler/interactive/spawner_pysqa.py @@ -5,12 +5,12 @@ from pysqa import QueueAdapter +from executorlib.standalone.command_pysqa import pysqa_execute_command, pysqa_terminate from executorlib.standalone.inputcheck import validate_number_of_cores from executorlib.standalone.interactive.spawner import ( BaseSpawner, set_current_directory_in_environment, ) -from executorlib.standalone.command_pysqa import pysqa_execute_command, pysqa_terminate from executorlib.task_scheduler.base import validate_resource_dict from executorlib.task_scheduler.interactive.blockallocation import ( BlockAllocationTaskScheduler,