Merge branch 'feat/batch_prioritization_darks' into 'master'

Feat/batch prioritization darks

See merge request detectors/pycalibration!228

webservice/webservice.py

@@ -355,9 +355,10 @@ async def copy_untouched_files(file_list, out_folder, run):
         logging.info("Copying {} to {}".format(f, of))
 
 
-async def run_correction(conn, cmd, mode, proposal, run, rid):
-    """ Run a correction command
+async def run_action(job_db, cmd, mode, proposal, run, rid):
+    """ Run action command (CORRECT OR DARK)
 
+    :param job_db: jobs database
     :param cmd: to run, should be a in list for as expected by subprocess.run
     :param mode: "prod" or "sim", in the latter case nothing will be executed
                  but the command will be logged
@@ -377,8 +378,9 @@ async def run_correction(conn, cmd, mode, proposal, run, rid):
             else:
                 logging.info(Success.START_CORRECTION.format(proposal, run))
             # enter jobs in job db
-            c = conn.cursor()
+            c = job_db.cursor()
             rstr = ret.stdout.decode()
+
             query = "INSERT INTO jobs VALUES ('{rid}', '{jobid}', '{proposal}', '{run}', 'PD', '{now}', '{det}', '{act}')"  # noqa
             for r in rstr.split("\n"):
                 if "Submitted job:" in r:
@@ -387,7 +389,7 @@ async def run_correction(conn, cmd, mode, proposal, run, rid):
                                            proposal=proposal, run=run,
                                            now=datetime.now().isoformat(),
                                            det=cmd[3], act=cmd[4]))
-            conn.commit()
+            job_db.commit()
             logging.debug(" ".join(cmd))
             if "DARK" in cmd:
                 return Success.START_CHAR.format(proposal, run)
@@ -687,7 +689,8 @@ async def server_runner(config, mode):
                                             "EPIX", "EPIX10K"]:
                         priority = '0'
                     cmd = ["python", "-m", "xfel_calibrate.calibrate",
-                           detector, "DARK", '--priority', priority]
+                           detector, "DARK", '--priority', priority,
+                           '--slurm-scheduling', str(config[action]['sched-prio'])]  #noqa
 
                     # Avoid giving a reservation parameter after the
                     # ITDM changes for giving xcal high priority by default
@@ -705,7 +708,7 @@ async def server_runner(config, mode):
 
                     cmd = await parse_config(cmd, dconfig)
 
-                    ret = await run_correction(job_db, cmd, mode, proposal,
+                    ret = await run_action(job_db, cmd, mode, proposal,
                                                wait_runs[0], 0)
                     status.append(ret)
 
@@ -755,13 +758,14 @@ async def server_runner(config, mode):
                     if "-" in detector:
                         detector, _ = detector.split("-")
                     cmd = ["python", "-m", "xfel_calibrate.calibrate",
-                           detector, "CORRECT"]
+                           detector, "CORRECT", '--slurm-scheduling',
+                           str(config[action]['sched-prio'])]
 
                     cmd = await parse_config(cmd, dconfig)
 
                     if priority:
                         cmd += ["--priority", str(priority)]
-                    ret = await run_correction(job_db, cmd, mode, proposal,
+                    ret = await run_action(job_db, cmd, mode, proposal,
                                                runnr, rid)
                     status.append(ret)
 

webservice/webservice.yaml

@@ -25,8 +25,10 @@ metadata-client:
 correct:
     in-folder: /gpfs/exfel/exp/{instrument}/{cycle}/p{proposal}/raw
     out-folder: /gpfs/exfel/d/proc/{instrument}/{cycle}/p{proposal}/{run}
+    sched-prio: 100
 
 dark:
     in-folder: /gpfs/exfel/exp/{instrument}/{cycle}/p{proposal}/raw
     out-folder: /gpfs/exfel/u/usr/{instrument}/{cycle}/p{proposal}/dark/runs_{runs}
+    sched-prio: 10
 

xfel_calibrate/calibrate.py

@@ -53,10 +53,12 @@ def make_initial_parser():
                         help="Do not run as a cluster job")
 
     parser.add_argument('--report-to', type=str,
-                        help='Filename (and optionally path) for output report')
+                        help='Filename (and optionally path) for output'
+                             ' report')
 
     parser.add_argument('--priority', type=int, default=2,
-                        help="Priority of batch jobs. If priority<=1, reserved nodes become available.")
+                        help="Priority of batch jobs. If priority<=1, reserved"
+                             " nodes become available.")
 
     parser.add_argument('--vector-figs', action="store_true", default=False,
                         help="Use vector graphics for figures in the report.")
@@ -67,8 +69,10 @@ def make_initial_parser():
     parser.add_argument('--slurm-name', type=str, default='xfel_calibrate',
                         help='Name of slurm job')
 
-    parser.add_argument('--slurm-priority', type=int, default=0,
-                        help='Change priority of srurm job +- 2147483645 (negative value increases priority)')
+    parser.add_argument('--slurm-scheduling', type=int, default=0,
+                        help='Change scheduling priority for a slurm job '
+                             '+- 2147483645 (negative value increases '
+                             'priority)')
 
     parser.add_argument_group('required arguments')