Merge branch 'main' into fix-env-type

Author: andre-merzky

docs/user_guide.rst

@@ -311,7 +311,7 @@ following example shows:
 .. literalinclude:: ../tests/user_guide/test_prelaunch.py
     :language: python
     :dedent: 4
-    :lines: 10-16
+    :lines: 12-18
 
 where the contents of ``pre_launch.sh`` is
 

requirements-docs.txt

@@ -12,6 +12,15 @@ sphinx_rtd_theme
 sphinx-tabs==3.2.0
 sphinx-multiversion
 
+# These are not requirements for PSI/J, but we must pin the version
+# since Sphinx 4.5.0 does not properly do so and later versions of
+# these plugins require Sphinx >= 5.0.0
+sphinxcontrib-applehelp==1.0.4
+sphinxcontrib-devhelp==1.0.2
+sphinxcontrib-htmlhelp==2.0.1
+sphinxcontrib-serializinghtml==1.1.5
+sphinxcontrib-qthelp==1.0.3
+
 # For the web version of the docs
 cloud_sptheme == 1.10.1.post20200504175005
 # sphinx-autodoc-typehints

src/psij-descriptors/pbs_descriptor.py

@@ -0,0 +1,11 @@
+from distutils.version import StrictVersion
+
+from psij.descriptor import Descriptor
+
+
+__PSI_J_EXECUTORS__ = [Descriptor(name='pbs', nice_name='PBS Pro', aliases=['pbspro'],
+                                  version=StrictVersion('0.0.2'),
+                                  cls='psij.executors.batch.pbs.PBSJobExecutor'),
+                       Descriptor(name='pbs_classic', nice_name='PBS Classic', aliases=['torque'],
+                                  version=StrictVersion('0.0.2'),
+                                  cls='psij.executors.batch.pbs_classic.PBSClassicJobExecutor')]

src/psij-descriptors/pbspro_descriptor.py

@@ -228,6 +228,11 @@ def submit(self, job: Job) -> None:
             self._set_job_status(job, JobStatus(JobState.QUEUED,
                                                 metadata={'native_id': job.native_id}))
         except subprocess.CalledProcessError as ex:
+            if logger.isEnabledFor(logging.DEBUG):
+                with submit_file_path.open('r') as submit_file:
+                    script = submit_file.read()
+                logger.debug('Job %s: submit script is: %s' % (job.id, script))
+
             raise SubmitException(ex.output) from None
 
         self._queue_poll_thread.register_job(job)
@@ -497,7 +502,7 @@ def _create_script_context(self, job: Job) -> Dict[str, object]:
     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)
+        return '%s:%s:%s' % (int(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)
@@ -562,7 +567,11 @@ def _read_aux_files(self, job: Job, status: JobStatus) -> None:
                 if status.message is None:
                     # only read output from submit script if another error message is not
                     # already present
-                    status.message = self._read_aux_file(job, '.out')
+                    out = self._read_aux_file(job, '.out')
+                    if out:
+                        launcher = self._get_launcher_from_job(job)
+                        if launcher.is_launcher_failure(out):
+                            status.message = launcher.get_launcher_failure_message(out)
                     logger.debug('Output from launcher: %s', status.message)
                 else:
                     self._delete_aux_file(job, '.out')

src/psij/executors/batch/batch_scheduler_executor.py

@@ -141,4 +141,4 @@ def get_list_command(self) -> List[str]:
     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)
+        return "%s:%s" % (int(d.total_seconds()) // 3600, (d.seconds // 60) % 60)

src/psij/executors/batch/lsf.py

@@ -0,0 +1,27 @@
+from pathlib import Path
+from typing import Optional
+
+from psij.executors.batch.pbs_base import PBSExecutorConfig, GenericPBSJobExecutor
+from psij.executors.batch.script_generator import TemplatedScriptGenerator
+
+
+class PBSJobExecutor(GenericPBSJobExecutor):
+    """A :class:`~psij.JobExecutor` for PBS Pro and friends.
+
+    This executor uses resource specifications specific to PBS Pro
+    """
+
+    def __init__(self, url: Optional[str] = None, config: Optional[PBSExecutorConfig] = None):
+        """
+        Parameters
+        ----------
+        url
+            Not used, but required by the spec for automatic initialization.
+        config
+            An optional configuration for this executor.
+        """
+        if not config:
+            config = PBSExecutorConfig()
+        generator = TemplatedScriptGenerator(config, Path(__file__).parent / 'pbspro'
+                                             / 'pbspro.mustache')
+        super().__init__(generator, url=url, config=config)

src/psij/executors/batch/pbs.py

@@ -0,0 +1,71 @@
+#!/bin/bash
+
+
+{{#job.name}}
+#PBS -N="{{.}}"
+{{/job.name}}
+
+{{#job.spec.inherit_environment}}
+#PBS -V
+{{/job.spec.inherit_environment}}
+
+{{#job.spec.resources}}
+    {{#process_count}}
+#PBS -l nodes={{job.spec.resources.computed_node_count}}:ppn={{.}}{{#job.spec.resources.gpu_cores_per_process}}:gpus={{.}}{{/job.spec.resources.gpu_cores_per_process}}
+    {{/process_count}}
+    {{#exclusive_node_use}}
+#PBS -n
+    {{/exclusive_node_use}}
+{{/job.spec.resources}}
+
+{{#formatted_job_duration}}
+#PBS -l walltime={{.}}
+{{/formatted_job_duration}}
+
+{{#job.spec.attributes}}
+    {{#project_name}}
+#PBS -P {{.}}
+    {{/project_name}}
+    {{#queue_name}}
+#PBS -q {{.}}
+    {{/queue_name}}
+    {{!PBS uses specially named queues for reservations, so we send the job to the respective
+    queue when a reservation ID is specified.}}
+    {{#reservation_id}}
+#PBS -q {{.}}
+    {{/reservation_id}}
+{{/job.spec.attributes}}
+
+{{#custom_attributes}}
+    {{#pbs}}
+#PBS -{{key}} "{{value}}"
+    {{/pbs}}
+{{/custom_attributes}}
+
+
+{{!since we redirect the output manually, below, tell pbs not to do its own thing, since it
+only results in empty files that are not cleaned up}}
+#PBS -e /dev/null
+#PBS -o /dev/null
+
+{{#job.spec.inherit_environment}}
+#PBS -V
+{{/job.spec.inherit_environment}}
+{{#env}}
+#PBS -v {{name}}={{value}}
+{{/env}}
+
+PSIJ_NODEFILE="$PBS_NODEFILE"
+export PSIJ_NODEFILE
+
+
+{{#job.spec.directory}}
+cd "{{.}}"
+{{/job.spec.directory}}
+
+exec &>> "{{psij.script_dir}}/$PBS_JOBID.out"
+
+{{#psij.launch_command}}{{.}} {{/psij.launch_command}}
+
+{{!we redirect to a file tied to the native ID so that we can reach the file with attach().}}
+echo "$?" > "{{psij.script_dir}}/$PBS_JOBID.ec"

src/psij/executors/batch/pbs/pbs_classic.mustache

@@ -1,7 +1,5 @@
 #!/bin/bash
 
-# TODO: job.spec.resources
-# TODO: job.spec.attributes
 
 {{#job.name}}
 #PBS -N="{{.}}"
@@ -11,6 +9,18 @@
 #PBS -V
 {{/job.spec.inherit_environment}}
 
+{{#job.spec.resources}}
+    {{#process_count}}
+#PBS -l select={{job.spec.resources.computed_node_count}}:ncpus={{.}}
+    {{/process_count}}
+    {{#exclusive_node_use}}
+#PBS -l place=scatter:exclhost
+    {{/exclusive_node_use}}
+    {{^exclusive_node_use}}
+#PBS -l place=scatter:shared
+    {{/exclusive_node_use}}
+{{/job.spec.resources}}
+
 {{#formatted_job_duration}}
 #PBS -l walltime={{.}}
 {{/formatted_job_duration}}

…j/executors/batch/pbspro/pbspro.mustache …psij/executors/batch/pbs/pbspro.mustachesrc/psij/executors/batch/pbspro/pbspro.mustache renamed to src/psij/executors/batch/pbs/pbspro.mustache src/psij/executors/batch/pbspro/pbspro.mustache renamed to src/psij/executors/batch/pbs/pbspro.mustache

@@ -11,7 +11,7 @@
 
 _QSTAT_COMMAND = 'qstat'
 
-# This table maps PBS Pro state codes to the corresponding PSI/J
+# This table maps PBS state codes to the corresponding PSI/J
 # JobState.
 # See https://www.altair.com/pdfs/pbsworks/PBSReferenceGuide2021.1.pdf
 # page 361, section 8.1 "Job States"
@@ -39,14 +39,14 @@
 }
 
 
-class PBSProExecutorConfig(BatchSchedulerExecutorConfig):
-    """A configuration class for the PBS executor."""
+class PBSExecutorConfig(BatchSchedulerExecutorConfig):
+    """A generic configuration class for PBS-type executors."""
 
     pass
 
 
-class PBSProJobExecutor(BatchSchedulerExecutor):
-    """A :class:`~psij.JobExecutor` for PBS.
+class GenericPBSJobExecutor(BatchSchedulerExecutor):
+    """A generic :class:`~psij.JobExecutor` for PBS-type schedulers.
 
     PBS, originally developed by NASA, is one of the oldest resource managers still in use.
     A number of variations are available: `PBS Pro <https://www.altair.com/pbs-professional/>`_,
@@ -60,7 +60,8 @@ class PBSProJobExecutor(BatchSchedulerExecutor):
     Creates a batch script with #PBS directives when submitting a job.
     """
 
-    def __init__(self, url: Optional[str] = None, config: Optional[PBSProExecutorConfig] = None):
+    def __init__(self, generator: TemplatedScriptGenerator, url: Optional[str] = None,
+                 config: Optional[PBSExecutorConfig] = None) -> None:
         """
         Parameters
         ----------
@@ -69,11 +70,8 @@ def __init__(self, url: Optional[str] = None, config: Optional[PBSProExecutorCon
         config
             An optional configuration for this executor.
         """
-        if not config:
-            config = PBSProExecutorConfig()
         super().__init__(url=url, config=config)
-        self.generator = TemplatedScriptGenerator(config, Path(__file__).parent / 'pbspro'
-                                                  / 'pbspro.mustache')
+        self.generator = generator
 
     # Submit methods