Combine several jobs into one to save computation resources

alea/submitter.py

@@ -61,6 +61,7 @@ class Submitter:
 
     config_file_path: str
     template_path: str
+    combine_n_job: int = 1
     allowed_special_args: List[str] = []
     logging = logging.getLogger("submitter_logger")
 
@@ -387,6 +388,31 @@ def already_done(self, i_args: dict) -> bool:
             is_done = False
         return is_done
 
+    def combined_tickets_generator(self):
+        """Get the combined submission script for the current configuration. ``self.vcombine_n_job``
+        jobs will be combined into one submission script.
+
+        Yields:
+            (str, str): the combined submission script and name output_filename
+
+        """
+
+        _script = ""
+        n_combined = 0
+        for script, last_output_filename in self.computation_tickets_generator():
+            if n_combined == 0:
+                _script += script
+            else:
+                _script += " && " + script
+            n_combined += 1
+            if n_combined == self.combine_n_job:
+                yield _script, last_output_filename
+                n_combined = 0
+                _script = ""
+        else:
+            if n_combined > 0:
+                yield _script, last_output_filename
+
     @staticmethod
     def update_n_batch(runner_args):
         """Update n_mc if n_batch is provided.

alea/submitters/local.py

@@ -32,6 +32,7 @@ def __init__(self, *args, **kwargs):
         """Initialize the SubmitterLocal class."""
         self.local_configurations = kwargs.get("local_configurations", {})
         self.template_path = self.local_configurations.pop("template_path", None)
+        self.combine_n_job = self.local_configurations.pop("combine_n_job", 1)
         super().__init__(*args, **kwargs)
 
     @staticmethod
@@ -57,7 +58,7 @@ def submit(self):
         If debug is True, only return the first instance of Runner.
 
         """
-        for _, (script, _) in enumerate(self.computation_tickets_generator()):
+        for _, (script, _) in enumerate(self.combined_tickets_generator()):
             if self.debug:
                 print(script)
                 return self.initialized_runner(script)

alea/submitters/slurm.py

@@ -38,6 +38,7 @@ def __init__(self, *args, **kwargs):
         self.name = self.__class__.__name__
         self.slurm_configurations = kwargs.get("slurm_configurations", {})
         self.template_path = self.slurm_configurations.pop("template_path", None)
+        self.combine_n_job = self.slurm_configurations.pop("combine_n_job", 1)
         self.batchq_arguments = {**BATCHQ_DEFAULT_ARGUMENTS, **self.slurm_configurations}
         self._check_batchq_arguments()
         super().__init__(*args, **kwargs)
@@ -92,7 +93,7 @@ def submit(self, **kwargs):
         """
         _jobname = kwargs.pop("jobname", self.name.lower())
         batchq_kwargs = {}
-        for job, (script, output_filename) in enumerate(self.computation_tickets_generator()):
+        for job, (script, last_output_filename) in enumerate(self.combined_tickets_generator()):
             if self.debug:
                 print(script)
                 if job > 0:
@@ -101,7 +102,7 @@ def submit(self, **kwargs):
                 self.logging.info("Too many jobs. Sleeping for 30s.")
                 time.sleep(30)
             batchq_kwargs["jobname"] = f"{_jobname}_{job:03d}"
-            if output_filename is not None:
-                batchq_kwargs["log"] = os.path.join(self.log_dir, f"{output_filename}.log")
+            if last_output_filename is not None:
+                batchq_kwargs["log"] = os.path.join(self.log_dir, f"{last_output_filename}.log")
             self.logging.debug(f"Call '_submit' with job: {job} and kwargs: {batchq_kwargs}.")
             self._submit(script, **batchq_kwargs)