[Maintenance] Refactor pysqa interface

Author: jan-janssen

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,

src/executorlib/standalone/scheduler.py …/executorlib/standalone/command_pysqa.pysrc/executorlib/standalone/scheduler.py renamed to src/executorlib/standalone/command_pysqa.py src/executorlib/standalone/scheduler.py renamed to 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,

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:

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,

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.
 

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,

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,

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(

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())

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,23 +70,23 @@ 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()
 
     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))
 
     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,27 +218,27 @@ 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"))
         os.makedirs(os.path.dirname(file_name), exist_ok=True)
         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=[],