rename from cancel_futures_on_shutdown to wait

Author: jan-janssen

src/executorlib/executor/flux.py

@@ -2,7 +2,7 @@
 
 from executorlib.executor.base import BaseExecutor
 from executorlib.standalone.inputcheck import (
-    check_cancel_futures_on_shutdown,
+    check_wait_on_shutdown,
     check_command_line_argument_lst,
     check_init_function,
     check_log_obj_size,
@@ -310,7 +310,7 @@ def __init__(
         plot_dependency_graph_filename: Optional[str] = None,
         export_workflow_filename: Optional[str] = None,
         log_obj_size: bool = False,
-        cancel_futures_on_shutdown: bool = False,
+        wait: bool = True,
     ):
         """
         The executorlib.FluxClusterExecutor leverages either the message passing interface (MPI), the SLURM workload
@@ -357,8 +357,7 @@ def __init__(
             plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
             export_workflow_filename (str): Name of the file to store the exported workflow graph in.
             log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-            cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes
-                                               on shutdown.
+            wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
         """
         default_resource_dict: dict = {
@@ -418,7 +417,7 @@ def __init__(
                         block_allocation=block_allocation,
                         init_function=init_function,
                         disable_dependencies=disable_dependencies,
-                        cancel_futures_on_shutdown=cancel_futures_on_shutdown,
+                        wait=wait,
                     )
                 )
         else:
@@ -459,7 +458,7 @@ def create_flux_executor(
     block_allocation: bool = False,
     init_function: Optional[Callable] = None,
     log_obj_size: bool = False,
-    cancel_futures_on_shutdown: bool = False,
+    wait: bool = True,
 ) -> Union[OneProcessTaskScheduler, BlockAllocationTaskScheduler]:
     """
     Create a flux executor
@@ -498,8 +497,7 @@ def create_flux_executor(
                                     of the individual function.
         init_function (None): optional function to preset arguments for functions which are submitted later
         log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-        cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes on
-                                           shutdown.
+        wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
     Returns:
         InteractiveStepExecutor/ InteractiveExecutor
@@ -521,9 +519,7 @@ def create_flux_executor(
     check_command_line_argument_lst(
         command_line_argument_lst=resource_dict.get("slurm_cmd_args", [])
     )
-    check_cancel_futures_on_shutdown(
-        cancel_futures_on_shutdown=cancel_futures_on_shutdown
-    )
+    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:

src/executorlib/executor/single.py

@@ -2,7 +2,7 @@
 
 from executorlib.executor.base import BaseExecutor
 from executorlib.standalone.inputcheck import (
-    check_cancel_futures_on_shutdown,
+    check_wait_on_shutdown,
     check_command_line_argument_lst,
     check_gpus_per_worker,
     check_init_function,
@@ -61,8 +61,7 @@ class SingleNodeExecutor(BaseExecutor):
         plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
         export_workflow_filename (str): Name of the file to store the exported workflow graph in.
         log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-        cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes on
-                                           shutdown.
+        wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
     Examples:
         ```
@@ -100,7 +99,7 @@ def __init__(
         plot_dependency_graph_filename: Optional[str] = None,
         export_workflow_filename: Optional[str] = None,
         log_obj_size: bool = False,
-        cancel_futures_on_shutdown: bool = False,
+        wait: bool = True,
     ):
         """
         The executorlib.SingleNodeExecutor leverages either the message passing interface (MPI), the SLURM workload
@@ -146,8 +145,7 @@ def __init__(
             plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
             export_workflow_filename (str): Name of the file to store the exported workflow graph in.
             log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-            cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes
-                                               on shutdown.
+            wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
         """
         default_resource_dict: dict = {
@@ -175,7 +173,7 @@ def __init__(
                         block_allocation=block_allocation,
                         init_function=init_function,
                         log_obj_size=log_obj_size,
-                        cancel_futures_on_shutdown=cancel_futures_on_shutdown,
+                        wait=wait,
                     ),
                     max_cores=max_cores,
                     refresh_rate=refresh_rate,
@@ -197,7 +195,7 @@ def __init__(
                     block_allocation=block_allocation,
                     init_function=init_function,
                     log_obj_size=log_obj_size,
-                    cancel_futures_on_shutdown=cancel_futures_on_shutdown,
+                    wait=wait,
                 )
             )
 
@@ -240,8 +238,7 @@ class TestClusterExecutor(BaseExecutor):
         plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
         export_workflow_filename (str): Name of the file to store the exported workflow graph in.
         log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-        cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes on
-                                           shutdown.
+        wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
     Examples:
         ```
@@ -279,7 +276,7 @@ def __init__(
         plot_dependency_graph_filename: Optional[str] = None,
         export_workflow_filename: Optional[str] = None,
         log_obj_size: bool = False,
-        cancel_futures_on_shutdown: bool = False,
+        wait: bool = True,
     ):
         """
         The executorlib.api.TestClusterExecutor is designed to test the file based communication used in the
@@ -318,8 +315,7 @@ def __init__(
             plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
             export_workflow_filename (str): Name of the file to store the exported workflow graph in.
             log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-            cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes
-                                               on shutdown.
+            wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
         """
         default_resource_dict: dict = {
@@ -359,7 +355,7 @@ def __init__(
                     init_function=init_function,
                     disable_dependencies=disable_dependencies,
                     execute_function=execute_in_subprocess,
-                    cancel_futures_on_shutdown=cancel_futures_on_shutdown,
+                    wait=wait,
                 )
             )
         else:
@@ -393,7 +389,7 @@ def create_single_node_executor(
     block_allocation: bool = False,
     init_function: Optional[Callable] = None,
     log_obj_size: bool = False,
-    cancel_futures_on_shutdown: bool = False,
+    wait: bool = True,
 ) -> Union[OneProcessTaskScheduler, BlockAllocationTaskScheduler]:
     """
     Create a single node executor
@@ -428,8 +424,7 @@ def create_single_node_executor(
                                     of the individual function.
         init_function (None): optional function to preset arguments for functions which are submitted later
         log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-        cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes on
-                                           shutdown.
+        wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
     Returns:
         InteractiveStepExecutor/ InteractiveExecutor
@@ -446,9 +441,7 @@ def create_single_node_executor(
     check_command_line_argument_lst(
         command_line_argument_lst=resource_dict.get("slurm_cmd_args", [])
     )
-    check_cancel_futures_on_shutdown(
-        cancel_futures_on_shutdown=cancel_futures_on_shutdown
-    )
+    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:

src/executorlib/executor/slurm.py

@@ -2,7 +2,7 @@
 
 from executorlib.executor.base import BaseExecutor
 from executorlib.standalone.inputcheck import (
-    check_cancel_futures_on_shutdown,
+    check_wait_on_shutdown,
     check_init_function,
     check_log_obj_size,
     check_plot_dependency_graph,
@@ -66,8 +66,7 @@ class SlurmClusterExecutor(BaseExecutor):
         plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
         export_workflow_filename (str): Name of the file to store the exported workflow graph in.
         log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-        cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes on
-                                           shutdown.
+        wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
     Examples:
         ```
@@ -107,7 +106,7 @@ def __init__(
         plot_dependency_graph_filename: Optional[str] = None,
         export_workflow_filename: Optional[str] = None,
         log_obj_size: bool = False,
-        cancel_futures_on_shutdown: bool = False,
+        wait: bool = True,
     ):
         """
         The executorlib.SlurmClusterExecutor leverages either the message passing interface (MPI), the SLURM workload
@@ -154,8 +153,7 @@ def __init__(
             plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
             export_workflow_filename (str): Name of the file to store the exported workflow graph in.
             log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-            cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes
-                                               on shutdown.
+            wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
         """
         default_resource_dict: dict = {
@@ -216,7 +214,7 @@ def __init__(
                         block_allocation=block_allocation,
                         init_function=init_function,
                         disable_dependencies=disable_dependencies,
-                        cancel_futures_on_shutdown=cancel_futures_on_shutdown,
+                        wait=wait,
                     )
                 )
         else:
@@ -288,8 +286,7 @@ class SlurmJobExecutor(BaseExecutor):
         plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
         export_workflow_filename (str): Name of the file to store the exported workflow graph in.
         log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-        cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes on
-                                           shutdown.
+        wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
     Examples:
         ```
@@ -328,7 +325,7 @@ def __init__(
         plot_dependency_graph_filename: Optional[str] = None,
         export_workflow_filename: Optional[str] = None,
         log_obj_size: bool = False,
-        cancel_futures_on_shutdown: bool = False,
+        wait: bool = True,
     ):
         """
         The executorlib.SlurmJobExecutor leverages either the message passing interface (MPI), the SLURM workload
@@ -378,8 +375,7 @@ def __init__(
             plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
             export_workflow_filename (str): Name of the file to store the exported workflow graph in.
             log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-            cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes
-                                               on shutdown.
+            wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
         """
         default_resource_dict: dict = {
@@ -408,7 +404,7 @@ def __init__(
                         block_allocation=block_allocation,
                         init_function=init_function,
                         log_obj_size=log_obj_size,
-                        cancel_futures_on_shutdown=cancel_futures_on_shutdown,
+                        wait=wait,
                     ),
                     max_cores=max_cores,
                     refresh_rate=refresh_rate,
@@ -431,7 +427,7 @@ def __init__(
                     block_allocation=block_allocation,
                     init_function=init_function,
                     log_obj_size=log_obj_size,
-                    cancel_futures_on_shutdown=cancel_futures_on_shutdown,
+                    wait=wait,
                 )
             )
 
@@ -446,7 +442,7 @@ def create_slurm_executor(
     block_allocation: bool = False,
     init_function: Optional[Callable] = None,
     log_obj_size: bool = False,
-    cancel_futures_on_shutdown: bool = False,
+    wait: bool = True,
 ) -> Union[OneProcessTaskScheduler, BlockAllocationTaskScheduler]:
     """
     Create a SLURM executor
@@ -486,8 +482,7 @@ def create_slurm_executor(
                                     of the individual function.
         init_function (None): optional function to preset arguments for functions which are submitted later
         log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.
-        cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes on
-                                           shutdown.
+        wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
     Returns:
         InteractiveStepExecutor/ InteractiveExecutor
@@ -500,9 +495,7 @@ def create_slurm_executor(
     resource_dict["log_obj_size"] = log_obj_size
     resource_dict["pmi_mode"] = pmi_mode
     check_init_function(block_allocation=block_allocation, init_function=init_function)
-    check_cancel_futures_on_shutdown(
-        cancel_futures_on_shutdown=cancel_futures_on_shutdown
-    )
+    check_wait_on_shutdown(wait_on_shutdown=wait)
     if block_allocation:
         resource_dict["init_function"] = init_function
         max_workers = validate_number_of_cores(

src/executorlib/standalone/inputcheck.py

@@ -17,15 +17,15 @@ def check_oversubscribe(oversubscribe: bool) -> None:
         )
 
 
-def check_cancel_futures_on_shutdown(
-    cancel_futures_on_shutdown: bool,
+def check_wait_on_shutdown(
+    wait_on_shutdown: bool,
 ) -> None:
     """
-    Check if cancel_futures_on_shutdown is True and raise a ValueError if it is.
+    Check if wait_on_shutdown is False and raise a ValueError if it is.
     """
-    if cancel_futures_on_shutdown:
+    if not wait_on_shutdown:
         raise ValueError(
-            "The cancel_futures_on_shutdown parameter is only supported for the executorlib.FluxClusterExecutor and executorlib.SlurmClusterExecutor."
+            "The wait_on_shutdown parameter is only supported for the executorlib.FluxClusterExecutor and executorlib.SlurmClusterExecutor."
         )
 
 

src/executorlib/task_scheduler/file/shared.py

@@ -57,7 +57,7 @@ def execute_tasks_h5(
     backend: Optional[str] = None,
     disable_dependencies: bool = False,
     pmi_mode: Optional[str] = None,
-    cancel_futures_on_shutdown: bool = False,
+    wait: bool = True,
 ) -> None:
     """
     Execute tasks stored in a queue using HDF5 files.
@@ -73,8 +73,7 @@ def execute_tasks_h5(
         backend (str, optional): name of the backend used to spawn tasks.
         disable_dependencies (boolean): Disable resolving future objects during the submission.
         pmi_mode (str): PMI interface to use (OpenMPI v5 requires pmix) default is None (Flux only)
-        cancel_futures_on_shutdown (bool): Whether to cancel pending futures and the corresponding Python processes on
-                                           shutdown.
+        wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
 
     Returns:
         None
@@ -89,7 +88,7 @@ def execute_tasks_h5(
         with contextlib.suppress(queue.Empty):
             task_dict = future_queue.get_nowait()
         if task_dict is not None and "shutdown" in task_dict and task_dict["shutdown"]:
-            if task_dict["wait"] and not cancel_futures_on_shutdown:
+            if task_dict["wait"] and wait:
                 while len(memory_dict) > 0:
                     memory_dict = {
                         key: _check_task_output(
@@ -100,7 +99,7 @@ def execute_tasks_h5(
                         for key, value in memory_dict.items()
                         if not value.done()
                     }
-            if not task_dict["cancel_futures"] and not cancel_futures_on_shutdown:
+            if not task_dict["cancel_futures"] and wait:
                 if (
                     terminate_function is not None
                     and terminate_function == terminate_subprocess