[Fix] Rename resource_dict to executor_kwargs (#936)

* [Fix] Rename resource_dict to executor_kwargs

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fixes

* fix test

* fix execute_tasks_h5() function

* fix tests

* more fixes

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: jan-janssen

src/executorlib/executor/flux.py

@@ -194,7 +194,7 @@ def __init__(
                         max_workers=max_workers,
                         cache_directory=cache_directory,
                         max_cores=max_cores,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         pmi_mode=pmi_mode,
                         flux_executor=flux_executor,
                         flux_executor_nesting=flux_executor_nesting,
@@ -221,7 +221,7 @@ def __init__(
                     max_workers=max_workers,
                     cache_directory=cache_directory,
                     max_cores=max_cores,
-                    resource_dict=resource_dict,
+                    executor_kwargs=resource_dict,
                     pmi_mode=pmi_mode,
                     flux_executor=flux_executor,
                     flux_executor_nesting=flux_executor_nesting,
@@ -404,7 +404,7 @@ def __init__(
                         pmi_mode=pmi_mode,
                         init_function=init_function,
                         max_workers=max_workers,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         pysqa_config_directory=pysqa_config_directory,
                         backend="flux",
                     )
@@ -420,7 +420,7 @@ def __init__(
                         backend="flux",
                         max_cores=max_cores,
                         cache_directory=cache_directory,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         flux_executor=None,
                         pmi_mode=pmi_mode,
                         flux_executor_nesting=False,
@@ -441,7 +441,7 @@ def __init__(
                         max_workers=max_workers,
                         cache_directory=cache_directory,
                         max_cores=max_cores,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         pmi_mode=None,
                         flux_executor=None,
                         flux_executor_nesting=False,
@@ -463,7 +463,7 @@ def create_flux_executor(
     max_workers: Optional[int] = None,
     max_cores: Optional[int] = None,
     cache_directory: Optional[str] = None,
-    resource_dict: Optional[dict] = None,
+    executor_kwargs: Optional[dict] = None,
     pmi_mode: Optional[str] = None,
     flux_executor=None,
     flux_executor_nesting: bool = False,
@@ -484,7 +484,7 @@ def create_flux_executor(
                            max_cores is recommended, as computers have a limited number of compute cores.
         max_cores (int): defines the number cores which can be used in parallel
         cache_directory (str, optional): The directory to store cache files. Defaults to "executorlib_cache".
-        resource_dict (dict): A dictionary of resources required by the task. With the following keys:
+        executor_kwargs (dict): A dictionary of arguments required by the executor. With the following keys:
                               - cores (int): number of MPI cores to be used for each function call
                               - threads_per_core (int): number of OpenMP threads to be used for each function call
                               - gpus_per_core (int): number of GPUs per worker - defaults to 0
@@ -524,29 +524,31 @@ def create_flux_executor(
         validate_max_workers,
     )
 
-    if resource_dict is None:
-        resource_dict = {}
-    cores_per_worker = resource_dict.get("cores", 1)
-    resource_dict["cache_directory"] = cache_directory
-    resource_dict["hostname_localhost"] = hostname_localhost
-    resource_dict["log_obj_size"] = log_obj_size
+    if executor_kwargs is None:
+        executor_kwargs = {}
+    cores_per_worker = executor_kwargs.get("cores", 1)
+    executor_kwargs["cache_directory"] = cache_directory
+    executor_kwargs["hostname_localhost"] = hostname_localhost
+    executor_kwargs["log_obj_size"] = log_obj_size
     check_init_function(block_allocation=block_allocation, init_function=init_function)
     check_pmi(backend="flux_allocation", pmi=pmi_mode)
-    check_oversubscribe(oversubscribe=resource_dict.get("openmpi_oversubscribe", False))
+    check_oversubscribe(
+        oversubscribe=executor_kwargs.get("openmpi_oversubscribe", False)
+    )
     check_command_line_argument_lst(
-        command_line_argument_lst=resource_dict.get("slurm_cmd_args", [])
+        command_line_argument_lst=executor_kwargs.get("slurm_cmd_args", [])
     )
     check_wait_on_shutdown(wait_on_shutdown=wait)
-    if "openmpi_oversubscribe" in resource_dict:
-        del resource_dict["openmpi_oversubscribe"]
-    if "slurm_cmd_args" in resource_dict:
-        del resource_dict["slurm_cmd_args"]
-    resource_dict["pmi_mode"] = pmi_mode
-    resource_dict["flux_executor"] = flux_executor
-    resource_dict["flux_executor_nesting"] = flux_executor_nesting
-    resource_dict["flux_log_files"] = flux_log_files
+    if "openmpi_oversubscribe" in executor_kwargs:
+        del executor_kwargs["openmpi_oversubscribe"]
+    if "slurm_cmd_args" in executor_kwargs:
+        del executor_kwargs["slurm_cmd_args"]
+    executor_kwargs["pmi_mode"] = pmi_mode
+    executor_kwargs["flux_executor"] = flux_executor
+    executor_kwargs["flux_executor_nesting"] = flux_executor_nesting
+    executor_kwargs["flux_log_files"] = flux_log_files
     if block_allocation:
-        resource_dict["init_function"] = init_function
+        executor_kwargs["init_function"] = init_function
         max_workers = validate_number_of_cores(
             max_cores=max_cores,
             max_workers=max_workers,
@@ -556,18 +558,18 @@ def create_flux_executor(
         validate_max_workers(
             max_workers=max_workers,
             cores=cores_per_worker,
-            threads_per_core=resource_dict.get("threads_per_core", 1),
+            threads_per_core=executor_kwargs.get("threads_per_core", 1),
         )
         return BlockAllocationTaskScheduler(
             max_workers=max_workers,
-            executor_kwargs=resource_dict,
+            executor_kwargs=executor_kwargs,
             spawner=FluxPythonSpawner,
             restart_limit=restart_limit,
         )
     else:
         return OneProcessTaskScheduler(
             max_cores=max_cores,
             max_workers=max_workers,
-            executor_kwargs=resource_dict,
+            executor_kwargs=executor_kwargs,
             spawner=FluxPythonSpawner,
         )

src/executorlib/executor/single.py

@@ -176,7 +176,7 @@ def __init__(
                         max_workers=max_workers,
                         cache_directory=cache_directory,
                         max_cores=max_cores,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         hostname_localhost=hostname_localhost,
                         block_allocation=block_allocation,
                         init_function=init_function,
@@ -199,7 +199,7 @@ def __init__(
                     max_workers=max_workers,
                     cache_directory=cache_directory,
                     max_cores=max_cores,
-                    resource_dict=resource_dict,
+                    executor_kwargs=resource_dict,
                     hostname_localhost=hostname_localhost,
                     block_allocation=block_allocation,
                     init_function=init_function,
@@ -359,7 +359,7 @@ def __init__(
                     backend=None,
                     max_cores=max_cores,
                     cache_directory=cache_directory,
-                    resource_dict=resource_dict,
+                    executor_kwargs=resource_dict,
                     flux_executor=None,
                     pmi_mode=None,
                     flux_executor_nesting=False,
@@ -381,7 +381,7 @@ def __init__(
                         max_workers=max_workers,
                         cache_directory=cache_directory,
                         max_cores=max_cores,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         hostname_localhost=hostname_localhost,
                         block_allocation=block_allocation,
                         init_function=init_function,
@@ -400,7 +400,7 @@ def create_single_node_executor(
     max_workers: Optional[int] = None,
     max_cores: Optional[int] = None,
     cache_directory: Optional[str] = None,
-    resource_dict: Optional[dict] = None,
+    executor_kwargs: Optional[dict] = None,
     hostname_localhost: Optional[bool] = None,
     block_allocation: bool = False,
     init_function: Optional[Callable] = None,
@@ -417,7 +417,7 @@ def create_single_node_executor(
                            max_cores is recommended, as computers have a limited number of compute cores.
         max_cores (int): defines the number cores which can be used in parallel
         cache_directory (str, optional): The directory to store cache files. Defaults to "executorlib_cache".
-        resource_dict (dict): A dictionary of resources required by the task. With the following keys:
+        executor_kwargs (dict): A dictionary of arguments required by the executor. With the following keys:
                               - cores (int): number of MPI cores to be used for each function call
                               - threads_per_core (int): number of OpenMP threads to be used for each function call
                               - gpus_per_core (int): number of GPUs per worker - defaults to 0
@@ -447,42 +447,42 @@ def create_single_node_executor(
     Returns:
         InteractiveStepExecutor/ InteractiveExecutor
     """
-    if resource_dict is None:
-        resource_dict = {}
-    cores_per_worker = resource_dict.get("cores", 1)
-    resource_dict["cache_directory"] = cache_directory
-    resource_dict["hostname_localhost"] = hostname_localhost
-    resource_dict["log_obj_size"] = log_obj_size
+    if executor_kwargs is None:
+        executor_kwargs = {}
+    cores_per_worker = executor_kwargs.get("cores", 1)
+    executor_kwargs["cache_directory"] = cache_directory
+    executor_kwargs["hostname_localhost"] = hostname_localhost
+    executor_kwargs["log_obj_size"] = log_obj_size
 
     check_init_function(block_allocation=block_allocation, init_function=init_function)
-    check_gpus_per_worker(gpus_per_worker=resource_dict.get("gpus_per_core", 0))
+    check_gpus_per_worker(gpus_per_worker=executor_kwargs.get("gpus_per_core", 0))
     check_command_line_argument_lst(
-        command_line_argument_lst=resource_dict.get("slurm_cmd_args", [])
+        command_line_argument_lst=executor_kwargs.get("slurm_cmd_args", [])
     )
     check_wait_on_shutdown(wait_on_shutdown=wait)
-    if "threads_per_core" in resource_dict:
-        del resource_dict["threads_per_core"]
-    if "gpus_per_core" in resource_dict:
-        del resource_dict["gpus_per_core"]
-    if "slurm_cmd_args" in resource_dict:
-        del resource_dict["slurm_cmd_args"]
+    if "threads_per_core" in executor_kwargs:
+        del executor_kwargs["threads_per_core"]
+    if "gpus_per_core" in executor_kwargs:
+        del executor_kwargs["gpus_per_core"]
+    if "slurm_cmd_args" in executor_kwargs:
+        del executor_kwargs["slurm_cmd_args"]
     if block_allocation:
-        resource_dict["init_function"] = init_function
+        executor_kwargs["init_function"] = init_function
         return BlockAllocationTaskScheduler(
             max_workers=validate_number_of_cores(
                 max_cores=max_cores,
                 max_workers=max_workers,
                 cores_per_worker=cores_per_worker,
                 set_local_cores=True,
             ),
-            executor_kwargs=resource_dict,
+            executor_kwargs=executor_kwargs,
             spawner=MpiExecSpawner,
             restart_limit=restart_limit,
         )
     else:
         return OneProcessTaskScheduler(
             max_cores=max_cores,
             max_workers=max_workers,
-            executor_kwargs=resource_dict,
+            executor_kwargs=executor_kwargs,
             spawner=MpiExecSpawner,
         )

src/executorlib/executor/slurm.py

@@ -189,7 +189,7 @@ def __init__(
                         pmi_mode=pmi_mode,
                         init_function=init_function,
                         max_workers=max_workers,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         pysqa_config_directory=pysqa_config_directory,
                         backend="slurm",
                     ),
@@ -206,7 +206,7 @@ def __init__(
                         backend="slurm",
                         max_cores=max_cores,
                         cache_directory=cache_directory,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         pmi_mode=pmi_mode,
                         flux_executor=None,
                         flux_executor_nesting=False,
@@ -227,7 +227,7 @@ def __init__(
                         max_workers=max_workers,
                         cache_directory=cache_directory,
                         max_cores=max_cores,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         hostname_localhost=hostname_localhost,
                         block_allocation=block_allocation,
                         init_function=init_function,
@@ -408,7 +408,7 @@ def __init__(
                         max_workers=max_workers,
                         cache_directory=cache_directory,
                         max_cores=max_cores,
-                        resource_dict=resource_dict,
+                        executor_kwargs=resource_dict,
                         pmi_mode=pmi_mode,
                         hostname_localhost=hostname_localhost,
                         block_allocation=block_allocation,
@@ -432,7 +432,7 @@ def __init__(
                     max_workers=max_workers,
                     cache_directory=cache_directory,
                     max_cores=max_cores,
-                    resource_dict=resource_dict,
+                    executor_kwargs=resource_dict,
                     pmi_mode=pmi_mode,
                     hostname_localhost=hostname_localhost,
                     block_allocation=block_allocation,
@@ -448,7 +448,7 @@ def create_slurm_executor(
     max_workers: Optional[int] = None,
     max_cores: Optional[int] = None,
     cache_directory: Optional[str] = None,
-    resource_dict: Optional[dict] = None,
+    executor_kwargs: Optional[dict] = None,
     pmi_mode: Optional[str] = None,
     hostname_localhost: Optional[bool] = None,
     block_allocation: bool = False,
@@ -466,7 +466,7 @@ def create_slurm_executor(
                            max_cores is recommended, as computers have a limited number of compute cores.
         max_cores (int): defines the number cores which can be used in parallel
         cache_directory (str, optional): The directory to store cache files. Defaults to "executorlib_cache".
-        resource_dict (dict): A dictionary of resources required by the task. With the following keys:
+        executor_kwargs (dict): A dictionary of arguments required by the executor. With the following keys:
                               - cores (int): number of MPI cores to be used for each function call
                               - threads_per_core (int): number of OpenMP threads to be used for each function call
                               - gpus_per_core (int): number of GPUs per worker - defaults to 0
@@ -502,17 +502,17 @@ def create_slurm_executor(
     Returns:
         InteractiveStepExecutor/ InteractiveExecutor
     """
-    if resource_dict is None:
-        resource_dict = {}
-    cores_per_worker = resource_dict.get("cores", 1)
-    resource_dict["cache_directory"] = cache_directory
-    resource_dict["hostname_localhost"] = hostname_localhost
-    resource_dict["log_obj_size"] = log_obj_size
-    resource_dict["pmi_mode"] = pmi_mode
+    if executor_kwargs is None:
+        executor_kwargs = {}
+    cores_per_worker = executor_kwargs.get("cores", 1)
+    executor_kwargs["cache_directory"] = cache_directory
+    executor_kwargs["hostname_localhost"] = hostname_localhost
+    executor_kwargs["log_obj_size"] = log_obj_size
+    executor_kwargs["pmi_mode"] = pmi_mode
     check_init_function(block_allocation=block_allocation, init_function=init_function)
     check_wait_on_shutdown(wait_on_shutdown=wait)
     if block_allocation:
-        resource_dict["init_function"] = init_function
+        executor_kwargs["init_function"] = init_function
         max_workers = validate_number_of_cores(
             max_cores=max_cores,
             max_workers=max_workers,
@@ -522,18 +522,18 @@ def create_slurm_executor(
         validate_max_workers(
             max_workers=max_workers,
             cores=cores_per_worker,
-            threads_per_core=resource_dict.get("threads_per_core", 1),
+            threads_per_core=executor_kwargs.get("threads_per_core", 1),
         )
         return BlockAllocationTaskScheduler(
             max_workers=max_workers,
-            executor_kwargs=resource_dict,
+            executor_kwargs=executor_kwargs,
             spawner=SrunSpawner,
             restart_limit=restart_limit,
         )
     else:
         return OneProcessTaskScheduler(
             max_cores=max_cores,
             max_workers=max_workers,
-            executor_kwargs=resource_dict,
+            executor_kwargs=executor_kwargs,
             spawner=SrunSpawner,
         )

src/executorlib/task_scheduler/file/shared.py

@@ -57,7 +57,7 @@ def done(self) -> bool:
 def execute_tasks_h5(
     future_queue: queue.Queue,
     execute_function: Callable,
-    resource_dict: dict,
+    executor_kwargs: dict,
     terminate_function: Optional[Callable] = None,
     pysqa_config_directory: Optional[str] = None,
     backend: Optional[str] = None,
@@ -71,7 +71,7 @@ def execute_tasks_h5(
 
     Args:
         future_queue (queue.Queue): The queue containing the tasks.
-        resource_dict (dict): A dictionary of resources required by the task. With the following keys:
+        executor_kwargs (dict): A dictionary of executor arguments required by the task. With the following keys:
                               - cores (int): number of MPI cores to be used for each function call
                               - cwd (str/None): current working directory where the parallel python task is executed
         execute_function (Callable): The function to execute the tasks.
@@ -119,7 +119,7 @@ def execute_tasks_h5(
             task_resource_dict, cache_key, cache_directory, error_log_file = (
                 _get_task_input(
                     task_resource_dict=task_dict["resource_dict"].copy(),
-                    resource_dict=resource_dict,
+                    executor_kwargs=executor_kwargs,
                 )
             )
             task_key, data_dict = serialize_funct(
@@ -352,10 +352,10 @@ def _cancel_processes(
 
 
 def _get_task_input(
-    task_resource_dict: dict, resource_dict: dict
+    task_resource_dict: dict, executor_kwargs: dict
 ) -> tuple[dict, Optional[str], str, Optional[str]]:
     task_resource_dict.update(
-        {k: v for k, v in resource_dict.items() if k not in task_resource_dict}
+        {k: v for k, v in executor_kwargs.items() if k not in task_resource_dict}
     )
     cache_key = task_resource_dict.pop("cache_key", None)
     cache_directory = os.path.abspath(task_resource_dict.pop("cache_directory"))