[Feature] Use seperate log files for different workers (#897)

* [Feature] Use seperate log files for different workers

* fixes

* Update test_flux_job.py

Author: jan-janssen

src/executorlib/standalone/interactive/spawner.py

@@ -11,6 +11,7 @@ def __init__(
         self,
         cwd: Optional[str] = None,
         cores: int = 1,
+        worker_id: int = 0,
         openmpi_oversubscribe: bool = False,
     ):
         """
@@ -20,9 +21,11 @@ def __init__(
             cwd (str): The current working directory.
             cores (int, optional): The number of cores to use. Defaults to 1.
             openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
+            worker_id (int): The worker ID. Defaults to 0.
         """
         self._cwd = cwd
         self._cores = cores
+        self._worker_id = worker_id
         self._openmpi_oversubscribe = openmpi_oversubscribe
 
     @abstractmethod
@@ -69,6 +72,7 @@ def __init__(
         self,
         cwd: Optional[str] = None,
         cores: int = 1,
+        worker_id: int = 0,
         openmpi_oversubscribe: bool = False,
         threads_per_core: int = 1,
     ):
@@ -79,11 +83,13 @@ def __init__(
             cwd (str, optional): The current working directory. Defaults to None.
             cores (int, optional): The number of cores to use. Defaults to 1.
             threads_per_core (int, optional): The number of threads per core. Defaults to 1.
+            worker_id (int): The worker ID. Defaults to 0.
             openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
         """
         super().__init__(
             cwd=cwd,
             cores=cores,
+            worker_id=worker_id,
             openmpi_oversubscribe=openmpi_oversubscribe,
         )
         self._process: Optional[subprocess.Popen] = None

src/executorlib/task_scheduler/interactive/blockallocation.py

@@ -208,7 +208,7 @@ def _execute_multiple_tasks(
     queue_join_on_shutdown: bool = True,
     log_obj_size: bool = False,
     error_log_file: Optional[str] = None,
-    worker_id: Optional[int] = None,
+    worker_id: int = 0,
     stop_function: Optional[Callable] = None,
     restart_limit: int = 0,
     **kwargs,
@@ -244,7 +244,7 @@ def _execute_multiple_tasks(
         command_lst=get_interactive_execute_command(
             cores=cores,
         ),
-        connections=spawner(cores=cores, **kwargs),
+        connections=spawner(cores=cores, worker_id=worker_id, **kwargs),
         hostname_localhost=hostname_localhost,
         log_obj_size=log_obj_size,
         worker_id=worker_id,

src/executorlib/task_scheduler/interactive/spawner_flux.py

@@ -31,6 +31,7 @@ class FluxPythonSpawner(BaseSpawner):
         threads_per_core (int, optional): The number of threads per base. Defaults to 1.
         gpus_per_core (int, optional): The number of GPUs per base. Defaults to 0.
         num_nodes (int, optional): The number of compute nodes to use for executing the task. Defaults to None.
+        worker_id (int): The worker ID. Defaults to 0.
         exclusive (bool): Whether to exclusively reserve the compute nodes, or allow sharing compute notes. Defaults to
                           False.
         openmpi_oversubscribe (bool, optional): Whether to oversubscribe. Defaults to False.
@@ -49,6 +50,7 @@ def __init__(
         threads_per_core: int = 1,
         gpus_per_core: int = 0,
         num_nodes: Optional[int] = None,
+        worker_id: int = 0,
         exclusive: bool = False,
         priority: Optional[int] = None,
         openmpi_oversubscribe: bool = False,
@@ -60,6 +62,7 @@ def __init__(
         super().__init__(
             cwd=cwd,
             cores=cores,
+            worker_id=worker_id,
             openmpi_oversubscribe=openmpi_oversubscribe,
         )
         self._threads_per_core = threads_per_core
@@ -121,12 +124,13 @@ def bootup(
         if self._cwd is not None:
             jobspec.cwd = self._cwd
             os.makedirs(self._cwd, exist_ok=True)
+        file_prefix = "flux_" + str(self._worker_id)
         if self._flux_log_files and self._cwd is not None:
-            jobspec.stderr = os.path.join(self._cwd, "flux.err")
-            jobspec.stdout = os.path.join(self._cwd, "flux.out")
+            jobspec.stderr = os.path.join(self._cwd, file_prefix + ".err")
+            jobspec.stdout = os.path.join(self._cwd, file_prefix + ".out")
         elif self._flux_log_files:
-            jobspec.stderr = os.path.abspath("flux.err")
-            jobspec.stdout = os.path.abspath("flux.out")
+            jobspec.stderr = os.path.abspath(file_prefix + ".err")
+            jobspec.stdout = os.path.abspath(file_prefix + ".out")
         if self._priority is not None:
             self._future = self._flux_executor.submit(
                 jobspec=jobspec, urgency=self._priority

src/executorlib/task_scheduler/interactive/spawner_pysqa.py

@@ -21,6 +21,7 @@ def __init__(
         threads_per_core: int = 1,
         gpus_per_core: int = 0,
         num_nodes: Optional[int] = None,
+        worker_id: int = 0,
         exclusive: bool = False,
         openmpi_oversubscribe: bool = False,
         slurm_cmd_args: Optional[list[str]] = None,
@@ -38,6 +39,7 @@ def __init__(
             threads_per_core (int): The number of threads per core. Defaults to 1.
             gpus_per_core (int): number of GPUs per worker - defaults to 0
             num_nodes (int, optional): The number of compute nodes to use for executing the task.  Defaults to None.
+            worker_id (int): The worker ID. Defaults to 0.
             exclusive (bool): Whether to exclusively reserve the compute nodes, or allow sharing compute notes. Defaults
                               to False.
             openmpi_oversubscribe (bool): Whether to oversubscribe the cores. Defaults to False.
@@ -49,6 +51,7 @@ def __init__(
         super().__init__(
             cwd=cwd,
             cores=cores,
+            worker_id=worker_id,
             openmpi_oversubscribe=openmpi_oversubscribe,
         )
         self._threads_per_core = threads_per_core

src/executorlib/task_scheduler/interactive/spawner_slurm.py

@@ -27,6 +27,7 @@ def __init__(
         threads_per_core: int = 1,
         gpus_per_core: int = 0,
         num_nodes: Optional[int] = None,
+        worker_id: int = 0,
         exclusive: bool = False,
         openmpi_oversubscribe: bool = False,
         slurm_cmd_args: Optional[list[str]] = None,
@@ -41,6 +42,7 @@ def __init__(
             threads_per_core (int, optional): The number of threads per core. Defaults to 1.
             gpus_per_core (int, optional): The number of GPUs per core. Defaults to 0.
             num_nodes (int, optional): The number of compute nodes to use for executing the task. Defaults to None.
+            worker_id (int): The worker ID. Defaults to 0.
             exclusive (bool): Whether to exclusively reserve the compute nodes, or allow sharing compute notes. Defaults to False.
             openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
             slurm_cmd_args (list[str], optional): Additional command line arguments. Defaults to [].
@@ -49,6 +51,7 @@ def __init__(
         super().__init__(
             cwd=cwd,
             cores=cores,
+            worker_id=worker_id,
             openmpi_oversubscribe=openmpi_oversubscribe,
             threads_per_core=threads_per_core,
         )

tests/unit/executor/test_flux_job.py

@@ -113,8 +113,8 @@ def test_single_task(self):
     def test_output_files_cwd(self):
         dirname = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
         os.makedirs(dirname, exist_ok=True)
-        file_stdout = os.path.join(dirname, "flux.out")
-        file_stderr = os.path.join(dirname, "flux.err")
+        file_stdout = os.path.join(dirname, "flux_0.out")
+        file_stderr = os.path.join(dirname, "flux_0.err")
         with FluxJobExecutor(
             max_cores=1,
             resource_dict={"cores": 1, "cwd": dirname},
@@ -133,8 +133,8 @@ def test_output_files_cwd(self):
         os.remove(file_stderr)
 
     def test_output_files_abs(self):
-        file_stdout = os.path.abspath("flux.out")
-        file_stderr = os.path.abspath("flux.err")
+        file_stdout = os.path.abspath("flux_0.out")
+        file_stderr = os.path.abspath("flux_0.err")
         with FluxJobExecutor(
             max_cores=1,
             resource_dict={"cores": 1},