PBS job manager

jobmanagers/BUILD.bazel

@@ -51,11 +51,17 @@ py_library(
     srcs = ["slurm_queue.py"],
 )
 
+py_library(
+    name = "pbs_queue",
+    srcs = ["pbs_queue.py"],
+)
+
 filegroup(
     name = "queue_checks",
     srcs = [
         ":sge_queue",
         ":slurm_queue",
+        ":pbs_queue",
     ],
     visibility = [
         "//:__pkg__",

jobmanagers/config.json

@@ -81,6 +81,8 @@
       },
       "pbspro": {
           "cmd": "qsub",
+          "queue_query": "pbs_queue.py",
+          "queue_query_grace_secs": 300,
           "mem_is_vmem": true,
           "envs": [ ]
       },

jobmanagers/pbs_queue.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python
+#
+# Copyright (c) 2017 10X Genomics, Inc. All rights reserved.
+#
+
+"""Queries qstat about a list of jobs and parses the output, returning the list
+of jobs which are queued, running, or on hold."""
+
+import subprocess
+import sys
+import json
+
+# PBS Pro "job states" to be regarded as "alive"
+ALIVE = "QHWSRE"
+
+
+def get_ids():
+    """Returns the set of jobids to query from standard input."""
+    ids = []
+    for jobid in sys.stdin:
+        ids.extend(jobid.split())
+    return ids
+
+
+def mkopts(ids):
+    """Gets the command line for qstat."""
+    if not ids:
+        sys.exit(0)
+    return ["qstat", "-x", "-F", "json", "-f"] + ids
+
+
+def execute(cmd):
+    """Executes qstat and captures its output."""
+    with subprocess.Popen(
+        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+    ) as proc:
+        out, err = proc.communicate()
+        if proc.returncode:
+            raise OSError(err)
+        if not isinstance(out, str):
+            out = out.decode()
+        if len(out) < 500:
+            sys.stderr.write(out)
+        else:
+            sys.stderr.write(out[:496] + "...")
+        return out
+
+
+def parse_output(out):
+    """Parses the JSON-format output of qstat and yields the ids of pending
+    jobs."""
+    data = json.loads(out)
+    for jid, info in data.get("Jobs", {}).items():
+        if info.get("job_state") in ALIVE:
+            yield jid
+
+
+def main():
+    """Reads a set of ids from standard input, queries qstat, and outputs the
+    jobids to standard output for jobs which are in the pending state."""
+    for jobid in parse_output(execute(mkopts(get_ids()))):
+        sys.stdout.write(f"{jobid}\n")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

jobmanagers/pbspro.template.example

@@ -6,9 +6,12 @@
 # Setup Instructions
 # =============================================================================
 #
-# 1. Add any other necessary PBSpro arguments such as queue (-q) or account
-#    (-A). If your system requires a walltime (-l walltime), 24 hours (24:00:00)
-#    is sufficient.  We recommend you do not remove any arguments below or
+# 1. Add any other necessary PBSpro arguments such as queue (-q), account
+#    (-A), project (-P) or volumes (-l storage=). 
+#    If your system requires a walltime (-l walltime), 24 hours (24:00:00)
+#    is sufficient, but this can often be reduced to 4 hours or less if 
+#    '--maxjobs' is at least 10. 
+#    We recommend you do not remove any arguments below or
 #    Martian may not run properly.
 #
 # 2. Change filename of pbspro.template.example to pbspro.template.
@@ -18,6 +21,7 @@
 # =============================================================================
 #
 #PBS -N __MRO_JOB_NAME__
+#PBS -S /bin/bash
 #PBS -V
 #PBS -l select=1:ncpus=__MRO_THREADS__
 #PBS -l mem=__MRO_MEM_GB__gb