refactor

Author: jan-janssen

executorlib/task_scheduler/interactive/onetoone.py

@@ -204,7 +204,6 @@ def _wrap_execute_task_in_separate_process(
             "task_dict": task_dict,
             "spawner": spawner,
             "hostname_localhost": hostname_localhost,
-            "init_function": None,
         }
     )
     process = Thread(

executorlib/task_scheduler/interactive/shared.py

@@ -74,17 +74,13 @@ def execute_multiple_tasks(
                 future_queue.join()
             break
         elif "fn" in task_dict and "future" in task_dict:
-            if error_log_file is not None:
-                task_dict["error_log_file"] = error_log_file
-            if cache_directory is None:
-                _execute_task_without_cache(interface=interface, task_dict=task_dict)
-            else:
-                _execute_task_with_cache(
-                    interface=interface,
-                    task_dict=task_dict,
-                    cache_directory=cache_directory,
-                    cache_key=cache_key,
-                )
+            _execute_task_dict(
+                task_dict=task_dict,
+                interface=interface,
+                cache_directory=cache_directory,
+                cache_key=cache_key,
+                error_log_file=error_log_file,
+            )
             _task_done(future_queue=future_queue)
 
 
@@ -93,7 +89,6 @@ def execute_single_task(
     cores: int = 1,
     spawner: type[BaseSpawner] = MpiExecSpawner,
     hostname_localhost: Optional[bool] = None,
-    init_function: Optional[Callable] = None,
     cache_directory: Optional[str] = None,
     cache_key: Optional[str] = None,
     log_obj_size: bool = False,
@@ -105,40 +100,64 @@ def execute_single_task(
     Execute a single tasks in parallel using the message passing interface (MPI).
 
     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
-       spawner (BaseSpawner): Spawner 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
-                                     in principle any computer should be able to resolve that their own hostname
-                                     points to the same address as localhost. Still MacOS >= 12 seems to disable
-                                     this look up for security reasons. So on MacOS it is required to set this
-                                     option to true
-       init_function (Callable): optional function to preset arguments for functions which are submitted later
-       cache_directory (str, optional): The directory to store cache files. Defaults to "executorlib_cache".
-       cache_key (str, optional): By default the cache_key is generated based on the function hash, this can be
-                                  overwritten by setting the cache_key.
-       queue_join_on_shutdown (bool): Join communication queue when thread is closed. Defaults to True.
-       log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-       error_log_file (str): Name of the error log file to use for storing exceptions raised by the Python functions
-                             submitted to the Executor.
-       worker_id (int): Communicate the worker which ID was assigned to it for future reference and resource
-                        distribution.
+        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
+        spawner (BaseSpawner): Spawner 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
+                                      in principle any computer should be able to resolve that their own hostname
+                                      points to the same address as localhost. Still MacOS >= 12 seems to disable
+                                      this look up for security reasons. So on MacOS it is required to set this
+                                      option to true
+        init_function (Callable): optional function to preset arguments for functions which are submitted later
+        cache_directory (str, optional): The directory to store cache files. Defaults to "executorlib_cache".
+        cache_key (str, optional): By default the cache_key is generated based on the function hash, this can be
+                                   overwritten by setting the cache_key.
+        queue_join_on_shutdown (bool): Join communication queue when thread is closed. Defaults to True.
+        log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
+        error_log_file (str): Name of the error log file to use for storing exceptions raised by the Python functions
+                              submitted to the Executor.
+        worker_id (int): Communicate the worker which ID was assigned to it for future reference and resource
+                         distribution.
     """
-    interface = interface_bootup(
-        command_lst=get_interactive_execute_command(
-            cores=cores,
+    _execute_task_dict(
+        task_dict=task_dict,
+        interface=interface_bootup(
+            command_lst=get_interactive_execute_command(
+                cores=cores,
+            ),
+            connections=spawner(cores=cores, **kwargs),
+            hostname_localhost=hostname_localhost,
+            log_obj_size=log_obj_size,
+            worker_id=worker_id,
         ),
-        connections=spawner(cores=cores, **kwargs),
-        hostname_localhost=hostname_localhost,
-        log_obj_size=log_obj_size,
-        worker_id=worker_id,
+        cache_directory=cache_directory,
+        cache_key=cache_key,
+        error_log_file=error_log_file,
     )
-    if init_function is not None:
-        interface.send_dict(
-            input_dict={"init": True, "fn": init_function, "args": (), "kwargs": {}}
-        )
+
+
+def _execute_task_dict(
+    task_dict: dict,
+    interface: SocketInterface,
+    cache_directory: Optional[str] = None,
+    cache_key: Optional[str] = None,
+    error_log_file: Optional[str] = None,
+):
+    """
+    Execute the task in the task_dict by communicating it via the interface.
+
+    Args:
+        task_dict (dict): task submitted to the executor as dictionary. This dictionary has the following keys
+                          {"fn": Callable, "args": (), "kwargs": {}, "resource_dict": {}}
+        interface (SocketInterface): socket interface for zmq communication
+        cache_directory (str, optional): The directory to store cache files. Defaults to "executorlib_cache".
+        cache_key (str, optional): By default the cache_key is generated based on the function hash, this can be
+                                  overwritten by setting the cache_key.
+        error_log_file (str): Name of the error log file to use for storing exceptions raised by the Python functions
+                              submitted to the Executor.
+    """
     if error_log_file is not None:
         task_dict["error_log_file"] = error_log_file
     if cache_directory is None: