Merge pull request #409 from ExaWorks/fix_walltimes

Fixed walltimes

Author: hategan

src/psij/executors/batch/batch_scheduler_executor.py

@@ -4,6 +4,7 @@
 import time
 import traceback
 from abc import abstractmethod
+from datetime import timedelta
 from pathlib import Path
 from threading import Thread, RLock
 from typing import Optional, List, Dict, Collection, cast, TextIO, Union
@@ -15,7 +16,6 @@
     JobStatus, JobState
 from psij.executors.batch.template_function_library import ALL as FUNCTION_LIBRARY
 
-
 UNKNOWN_ERROR = 'PSIJ: Unknown error'
 
 logger = logging.getLogger(__name__)
@@ -471,8 +471,18 @@ def _create_script_context(self, job: Job) -> Dict[str, object]:
                 'script_dir': str(self.work_directory)
             }
         }
+        assert job.spec is not None
+        if job.spec.attributes:
+            duration = job.spec.attributes.duration
+            if duration is not None:
+                ctx['formatted_job_duration'] = self._format_duration(duration)
         return ctx
 
+    def _format_duration(self, d: timedelta) -> str:
+        # the default is hh:mm:ss, with hh not limited to 24; this is the least ambiguous
+        # choice
+        return '%s:%s:%s' % (d.total_seconds() // 3600, (d.seconds // 60) % 60, d.seconds % 60)
+
     def _run_command(self, cmd: List[str]) -> str:
         res = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
         if res.returncode != 0:

src/psij/executors/batch/cobalt.py

@@ -1,5 +1,5 @@
 """Defines a JobExecutor for the Cobalt resource manager."""
-
+from datetime import timedelta
 from pathlib import Path
 from typing import Optional, Collection, List, Dict, TextIO
 import re
@@ -121,3 +121,10 @@ def job_id_from_submit_output(self, out: str) -> str:
         if match is None:
             raise SubmitException(out)
         return match.group(0)
+
+    def _format_duration(self, d: timedelta) -> str:
+        # https://trac.mcs.anl.gov/projects/cobalt/wiki/qsub.1.html:
+        #   The time may be specified as eitehr an integer number of minutes or a colon-delimited
+        #   value of the format: HH:MM:SS. Enter 0 to get the max allowed walltime.
+        # base class _format_duration is HH:MM:SS
+        return super()._format_duration(d)

src/psij/executors/batch/cobalt/cobalt.mustache

@@ -13,10 +13,10 @@
 #COBALT --proccount={{.}}
     {{/process_count}}
 {{/job.spec.resources}}
-{{#job.spec.attributes}}
-    {{#duration}}
+{{#formatted_job_duration}}
 #COBALT --time={{duration}}
-    {{/duration}}
+{{/formatted_job_duration}}
+{{#job.spec.attributes}}
     {{#queue_name}}
 #COBALT --queue={{.}}
     {{/queue_name}}

src/psij/executors/batch/lsf.py

@@ -1,5 +1,5 @@
 """Defines the LsfJobExecutor class and its config class."""
-
+from datetime import timedelta
 from pathlib import Path
 import re
 import json
@@ -131,3 +131,8 @@ def job_id_from_submit_output(self, out: str) -> str:
     def get_list_command(self) -> List[str]:
         """See :meth:`~.BatchSchedulerExecutor.get_list_command`."""
         return [_BJOBS_COMMAND, '-a', '-noheader', '-o', 'jobid', '-u', self._current_user()]
+
+    def _format_duration(self, d: timedelta) -> str:
+        # https://www.ibm.com/docs/en/spectrum-lsf/10.1.0?topic=o-w-1:
+        #   bsub -W [hour:]minute[/host_name | /host_model]
+        return "%s:%s" % (d.total_seconds() // 3600, (d.seconds // 60) % 60)

src/psij/executors/batch/lsf/lsf.mustache

@@ -35,9 +35,9 @@
 {{/job.spec.resources}}
 
 
-{{#job_duration}}
+{{#formatted_job_duration}}
 #BSUB -W {{.}}
-{{/job_duration}}
+{{/formatted_job_duration}}
 
 {{#job.spec.attributes}}
     {{#queue_name}}

src/psij/executors/batch/pbspro.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from pathlib import Path
 from typing import Optional, Collection, List, Dict, TextIO
 
@@ -145,3 +146,9 @@ def parse_list_output(self, out: str) -> List[str]:
     def _get_state(self, state: str) -> JobState:
         assert state in _STATE_MAP, f"PBS state {state} is not known to PSI/J"
         return _STATE_MAP[state]
+
+    def _format_duration(self, d: timedelta) -> str:
+        # There isn't a clear specification for the walltime in
+        # https://help.altair.com/2022.1.0/PBS%20Professional/PBSReferenceGuide2022.1.pdf,
+        # but all examples use hh:mm:ss, which is the default in the base class
+        return super()._format_duration(d)

src/psij/executors/batch/pbspro/pbspro.mustache

@@ -11,11 +11,11 @@
 #PBS -V
 {{/job.spec.inherit_environment}}
 
-{{#job.spec.attributes}}
-    {{#duration}}
+{{#formatted_job_duration}}
 #PBS -l walltime={{.}}
-# TODO: unclear what the syntax is here for times... its always a mess
-    {{/duration}}
+{{/formatted_job_duration}}
+
+{{#job.spec.attributes}}
     {{#projectName}}
 #PBS -P {{.}}
     {{/projectName}}

src/psij/executors/batch/slurm.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from pathlib import Path
 from typing import Optional, Collection, List, Dict, TextIO
 
@@ -175,3 +176,12 @@ def _get_message(self, reason: str) -> str:
     def job_id_from_submit_output(self, out: str) -> str:
         """See :meth:`~.BatchSchedulerExecutor.job_id_from_submit_output`."""
         return out.strip().split()[-1]
+
+    def _format_duration(self, d: timedelta) -> str:
+        # https://slurm.schedmd.com/sbatch.html#OPT_time:
+        #   Acceptable time formats include "minutes", "minutes:seconds", "hours:minutes:seconds",
+        #   "days-hours", "days-hours:minutes" and "days-hours:minutes:seconds".
+        days = ''
+        if d.days > 0:
+            days = str(d.days) + '-'
+        return days + "%s:%s:%s" % (d.seconds // 3600, (d.seconds // 60) % 60, d.seconds % 60)

src/psij/executors/batch/slurm/slurm.mustache

@@ -45,11 +45,11 @@
     {{/cpu_cores_per_process}}
 {{/job.spec.resources}}
 
-{{#job.spec.attributes}}
-    {{#duration}}
+{{#formatted_job_duration}}
 #SBATCH --time={{.}}
-    {{/duration}}
+{{/formatted_job_duration}}
 
+{{#job.spec.attributes}}
     {{#queue_name}}
 #SBATCH --partition="{{.}}"
     {{/queue_name}}