Rename Interface to Spawner

executorlib/interactive/__init__.py

@@ -18,14 +18,14 @@
     validate_backend,
     validate_number_of_cores,
 )
-from executorlib.shared.interface import (
+from executorlib.shared.spawner import (
     SLURM_COMMAND,
-    MpiExecInterface,
-    SrunInterface,
+    MpiExecSpawner,
+    SrunSpawner,
 )
 
 try:  # The PyFluxExecutor requires flux-core to be installed.
-    from executorlib.interactive.flux import FluxPythonInterface
+    from executorlib.interactive.flux import FluxPythonSpawner
 
     flux_installed = "FLUX_URI" in os.environ
 except ImportError:
@@ -122,13 +122,13 @@ def create_executor(
             return InteractiveExecutor(
                 max_workers=int(max_cores / cores_per_worker),
                 executor_kwargs=executor_kwargs,
-                interface_class=FluxPythonInterface,
+                interface_class=FluxPythonSpawner,
             )
         else:
             return InteractiveStepExecutor(
                 max_cores=max_cores,
                 executor_kwargs=executor_kwargs,
-                interface_class=FluxPythonInterface,
+                interface_class=FluxPythonSpawner,
             )
     elif backend == "slurm":
         check_executor(executor=executor)
@@ -142,13 +142,13 @@ def create_executor(
             return InteractiveExecutor(
                 max_workers=int(max_cores / cores_per_worker),
                 executor_kwargs=executor_kwargs,
-                interface_class=SrunInterface,
+                interface_class=SrunSpawner,
             )
         else:
             return InteractiveStepExecutor(
                 max_cores=max_cores,
                 executor_kwargs=executor_kwargs,
-                interface_class=SrunInterface,
+                interface_class=SrunSpawner,
             )
     else:  # backend="local"
         check_threads_per_core(threads_per_core=threads_per_core)
@@ -164,11 +164,11 @@ def create_executor(
             return InteractiveExecutor(
                 max_workers=int(max_cores / cores_per_worker),
                 executor_kwargs=executor_kwargs,
-                interface_class=MpiExecInterface,
+                interface_class=MpiExecSpawner,
             )
         else:
             return InteractiveStepExecutor(
                 max_cores=max_cores,
                 executor_kwargs=executor_kwargs,
-                interface_class=MpiExecInterface,
+                interface_class=MpiExecSpawner,
             )

executorlib/interactive/executor.py

@@ -4,7 +4,7 @@
     execute_parallel_tasks,
     execute_separate_tasks,
 )
-from executorlib.shared.interface import BaseInterface, MpiExecInterface
+from executorlib.shared.spawner import BaseSpawner, MpiExecSpawner
 from executorlib.shared.thread import RaisingThread
 
 
@@ -19,7 +19,7 @@ class InteractiveExecutor(ExecutorBroker):
     Args:
         max_workers (int): defines the number workers which can execute functions in parallel
         executor_kwargs (dict): keyword arguments for the executor
-        interface_class (BaseInterface): interface class to initiate python processes
+        interface_class (BaseSpawner): interface class to initiate python processes
 
     Examples:
 
@@ -46,7 +46,7 @@ def __init__(
         self,
         max_workers: int = 1,
         executor_kwargs: dict = {},
-        interface_class: BaseInterface = MpiExecInterface,
+        interface_class: BaseSpawner = MpiExecSpawner,
     ):
         super().__init__()
         executor_kwargs["future_queue"] = self._future_queue
@@ -73,7 +73,7 @@ class InteractiveStepExecutor(ExecutorSteps):
     Args:
         max_cores (int): defines the number workers which can execute functions in parallel
         executor_kwargs (dict): keyword arguments for the executor
-        interface_class (BaseInterface): interface class to initiate python processes
+        interface_class (BaseSpawner): interface class to initiate python processes
 
     Examples:
 
@@ -98,7 +98,7 @@ def __init__(
         self,
         max_cores: int = 1,
         executor_kwargs: dict = {},
-        interface_class: BaseInterface = MpiExecInterface,
+        interface_class: BaseSpawner = MpiExecSpawner,
     ):
         super().__init__()
         executor_kwargs["future_queue"] = self._future_queue

executorlib/interactive/flux.py

@@ -3,10 +3,10 @@
 
 import flux.job
 
-from executorlib.shared.interface import BaseInterface
+from executorlib.shared.spawner import BaseSpawner
 
 
-class FluxPythonInterface(BaseInterface):
+class FluxPythonSpawner(BaseSpawner):
     """
     A class representing the FluxPythonInterface.
 

executorlib/shared/__init__.py

@@ -7,7 +7,7 @@
     interface_shutdown,
 )
 from executorlib.shared.executor import cancel_items_in_queue
-from executorlib.shared.interface import MpiExecInterface, SrunInterface
+from executorlib.shared.spawner import MpiExecSpawner, SrunSpawner
 from executorlib.shared.thread import RaisingThread
 
 __all__ = [
@@ -19,6 +19,6 @@
     interface_receive,
     cancel_items_in_queue,
     RaisingThread,
-    MpiExecInterface,
-    SrunInterface,
+    MpiExecSpawner,
+    SrunSpawner,
 ]

executorlib/shared/communication.py

@@ -10,15 +10,15 @@ class SocketInterface:
     The SocketInterface is an abstraction layer on top of the zero message queue.
 
     Args:
-        interface (executorlib.shared.interface.BaseInterface): Interface for starting the parallel process
+        interface (executorlib.shared.spawner.BaseSpawner): Interface for starting the parallel process
     """
 
     def __init__(self, interface=None):
         """
         Initialize the SocketInterface.
 
         Args:
-            interface (executorlib.shared.interface.BaseInterface): Interface for starting the parallel process
+            interface (executorlib.shared.spawner.BaseSpawner): Interface for starting the parallel process
         """
         self._context = zmq.Context()
         self._socket = self._context.socket(zmq.PAIR)
@@ -133,7 +133,7 @@ def interface_bootup(
 
     Args:
         command_lst (list): List of commands as strings
-        connections (executorlib.shared.interface.BaseInterface): Interface to start parallel process, like MPI, SLURM
+        connections (executorlib.shared.spawner.BaseSpawner): Interface to start parallel process, like MPI, SLURM
                                                                   or Flux
         hostname_localhost (boolean): use localhost instead of the hostname to establish the zmq connection. In the
                                       context of an HPC cluster this essential to be able to communicate to an

executorlib/shared/executor.py

@@ -19,7 +19,7 @@
     check_resource_dict,
     check_resource_dict_is_empty,
 )
-from executorlib.shared.interface import BaseInterface, MpiExecInterface
+from executorlib.shared.spawner import BaseSpawner, MpiExecSpawner
 from executorlib.shared.thread import RaisingThread
 
 
@@ -301,7 +301,7 @@ def cloudpickle_register(ind: int = 2):
 def execute_parallel_tasks(
     future_queue: queue.Queue,
     cores: int = 1,
-    interface_class: BaseInterface = MpiExecInterface,
+    interface_class: BaseSpawner = MpiExecSpawner,
     hostname_localhost: bool = False,
     init_function: Optional[Callable] = None,
     prefix_name: Optional[str] = None,
@@ -314,7 +314,7 @@ def execute_parallel_tasks(
     Args:
        future_queue (queue.Queue): task queue of dictionary objects which are submitted to the parallel process
        cores (int): defines the total number of MPI ranks to use
-       interface_class (BaseInterface): Interface to start process on selected compute resources
+       interface_class (BaseSpawner): Interface to start process on selected compute resources
        hostname_localhost (boolean): use localhost instead of the hostname to establish the zmq connection. In the
                                      context of an HPC cluster this essential to be able to communicate to an
                                      Executor running on a different compute node within the same allocation. And
@@ -362,7 +362,7 @@ def execute_parallel_tasks(
 
 def execute_separate_tasks(
     future_queue: queue.Queue,
-    interface_class: BaseInterface = MpiExecInterface,
+    interface_class: BaseSpawner = MpiExecSpawner,
     max_cores: int = 1,
     hostname_localhost: bool = False,
     **kwargs,
@@ -372,7 +372,7 @@ def execute_separate_tasks(
 
     Args:
        future_queue (queue.Queue): task queue of dictionary objects which are submitted to the parallel process
-       interface_class (BaseInterface): Interface to start process on selected compute resources
+       interface_class (BaseSpawner): Interface to start process on selected compute resources
        max_cores (int): defines the number cores which can be used in parallel
        hostname_localhost (boolean): use localhost instead of the hostname to establish the zmq connection. In the
                                      context of an HPC cluster this essential to be able to communicate to an
@@ -621,7 +621,7 @@ def _submit_function_to_separate_process(
     task_dict: dict,
     active_task_dict: dict,
     qtask: queue.Queue,
-    interface_class: BaseInterface,
+    interface_class: BaseSpawner,
     executor_kwargs: dict,
     max_cores: int = 1,
     hostname_localhost: bool = False,
@@ -633,7 +633,7 @@ def _submit_function_to_separate_process(
                           {"fn": callable, "args": (), "kwargs": {}, "resource_dict": {}}
         active_task_dict (dict): Dictionary containing the future objects and the number of cores they require
         qtask (queue.Queue): Queue to communicate with the thread linked to the process executing the python function
-        interface_class (BaseInterface): Interface to start process on selected compute resources
+        interface_class (BaseSpawner): Interface to start process on selected compute resources
         executor_kwargs (dict): keyword parameters used to initialize the Executor
         max_cores (int): defines the number cores which can be used in parallel
         hostname_localhost (boolean): use localhost instead of the hostname to establish the zmq connection. In the

executorlib/shared/interface.py → executorlib/shared/spawner.py

@@ -6,7 +6,7 @@
 SLURM_COMMAND = "srun"
 
 
-class BaseInterface(ABC):
+class BaseSpawner(ABC):
     def __init__(self, cwd: str, cores: int = 1, oversubscribe: bool = False):
         """
         Base class for interface implementations.
@@ -55,7 +55,7 @@ def poll(self):
         raise NotImplementedError
 
 
-class SubprocessInterface(BaseInterface):
+class SubprocessSpawner(BaseSpawner):
     def __init__(
         self,
         cwd: Optional[str] = None,
@@ -143,7 +143,7 @@ def poll(self) -> bool:
         return self._process is not None and self._process.poll() is None
 
 
-class MpiExecInterface(SubprocessInterface):
+class MpiExecSpawner(SubprocessSpawner):
     def generate_command(self, command_lst: list[str]) -> list[str]:
         """
         Generate the command list for the MPIExec interface.
@@ -163,7 +163,7 @@ def generate_command(self, command_lst: list[str]) -> list[str]:
         )
 
 
-class SrunInterface(SubprocessInterface):
+class SrunSpawner(SubprocessSpawner):
     def __init__(
         self,
         cwd: Optional[str] = None,

tests/test_executor_backend_flux.py

@@ -8,7 +8,7 @@
 
 try:
     import flux.job
-    from executorlib.interactive.flux import FluxPythonInterface
+    from executorlib.interactive.flux import FluxPythonSpawner
 
     skip_flux_test = "FLUX_URI" not in os.environ
     pmi = os.environ.get("PYMPIPOOL_PMIX", None)

tests/test_flux_executor.py

@@ -11,7 +11,7 @@
 
 try:
     import flux.job
-    from executorlib.interactive.flux import FluxPythonInterface
+    from executorlib.interactive.flux import FluxPythonSpawner
 
     skip_flux_test = "FLUX_URI" not in os.environ
     pmi = os.environ.get("PYMPIPOOL_PMIX", None)
@@ -50,7 +50,7 @@ def test_flux_executor_serial(self):
         with InteractiveExecutor(
             max_workers=2,
             executor_kwargs={"executor": self.executor},
-            interface_class=FluxPythonInterface,
+            interface_class=FluxPythonSpawner,
         ) as exe:
             fs_1 = exe.submit(calc, 1)
             fs_2 = exe.submit(calc, 2)
@@ -63,7 +63,7 @@ def test_flux_executor_threads(self):
         with InteractiveExecutor(
             max_workers=1,
             executor_kwargs={"executor": self.executor, "threads_per_core": 2},
-            interface_class=FluxPythonInterface,
+            interface_class=FluxPythonSpawner,
         ) as exe:
             fs_1 = exe.submit(calc, 1)
             fs_2 = exe.submit(calc, 2)
@@ -76,7 +76,7 @@ def test_flux_executor_parallel(self):
         with InteractiveExecutor(
             max_workers=1,
             executor_kwargs={"executor": self.executor, "cores": 2, "pmi": pmi},
-            interface_class=FluxPythonInterface,
+            interface_class=FluxPythonSpawner,
         ) as exe:
             fs_1 = exe.submit(mpi_funct, 1)
             self.assertEqual(fs_1.result(), [(1, 2, 0), (1, 2, 1)])
@@ -86,7 +86,7 @@ def test_single_task(self):
         with InteractiveExecutor(
             max_workers=1,
             executor_kwargs={"executor": self.executor, "cores": 2, "pmi": pmi},
-            interface_class=FluxPythonInterface,
+            interface_class=FluxPythonSpawner,
         ) as p:
             output = p.map(mpi_funct, [1, 2, 3])
         self.assertEqual(
@@ -104,7 +104,7 @@ def test_execute_task(self):
             future_queue=q,
             cores=1,
             executor=self.executor,
-            interface_class=FluxPythonInterface,
+            interface_class=FluxPythonSpawner,
         )
         self.assertEqual(f.result(), 2)
         q.join()
@@ -120,7 +120,7 @@ def test_execute_task_threads(self):
             cores=1,
             threads_per_core=1,
             executor=self.executor,
-            interface_class=FluxPythonInterface,
+            interface_class=FluxPythonSpawner,
         )
         self.assertEqual(f.result(), 2)
         q.join()
@@ -133,7 +133,7 @@ def test_internal_memory(self):
                 "cores": 1,
                 "init_function": set_global,
             },
-            interface_class=FluxPythonInterface,
+            interface_class=FluxPythonSpawner,
         ) as p:
             f = p.submit(get_global)
             self.assertFalse(f.done())
@@ -142,13 +142,13 @@ def test_internal_memory(self):
 
     def test_interface_exception(self):
         with self.assertRaises(ValueError):
-            flux_interface = FluxPythonInterface(
+            flux_interface = FluxPythonSpawner(
                 executor=self.executor, oversubscribe=True
             )
             flux_interface.bootup(command_lst=[])
         with self.assertRaises(ValueError):
-            flux_interface = FluxPythonInterface(executor=self.executor)
+            flux_interface = FluxPythonSpawner(executor=self.executor)
             flux_interface.bootup(command_lst=[], prefix_path="/path/to/conda/env")
         with self.assertRaises(ValueError):
-            flux_interface = FluxPythonInterface(executor=self.executor)
+            flux_interface = FluxPythonSpawner(executor=self.executor)
             flux_interface.bootup(command_lst=[], prefix_name="conda_env_name")