Merge pull request #47080 from rmankel/rmankel-updateMPS

Update of Millepede production system (MPS): large campaigns

Author: cmsbuild

Alignment/MillePedeAlignmentAlgorithm/python/mpslib/Mpslibclass.py

@@ -42,6 +42,9 @@
 import time
 import os
 import sys
+import re
+import math
+import fileinput
 
 #-------------------------------------------------------------------------------
 class jobdatabase:
@@ -144,33 +147,49 @@ def print_memdb(self):
         print('pedeMem:\t',             self.pedeMem, '\n')
 
         #print interesting Job-level lists ---- to add: t/evt, fix remarks
-        print('###     dir      jobid    stat  try  rtime      nevt  remark   weight  name')
+        headFmt = '###     dir      jobid    stat  try  rtime      nevt  remark   weight  name'
+        jobFmt = '%03d  %6s  %10s%6s  %3d  %5d  %6d  %12s  %5s  %s'
+        mrgFmt = '%s  %6s  %10s%6s  %3d  %5d  %6d  %12s  %5s  %s'
+        if self.nJobs>999:
+            headFmt = '####     dir       jobid  stat  try  rtime      nevt    remark   weight  name'
+            jobFmt = '%04d  %7s  %10s%6s  %3d  %5d  %6d  %12s  %5s  %s'
+            mrgFmt = '%s   %7s  %10s%6s  %3d  %5d  %6d  %12s  %5s  %s'
+        if self.nJobs>9999:
+            jobFmt = '%d  %s  %10s%6s  %3d  %5d  %6d  %12s  %5s  %s'
+            mrgFmt = '%s    %8s  %10s%6s  %3d  %5d  %6d  %12s  %5s  %s'
+        print(headFmt)
         print("------------------------------------------------------------------------------")
         for i in range(self.nJobs):
-            print('%03d  %6s  %9s  %6s  %3d  %5d  %8d  %8s  %5s  %s' % (
+            remarkField = self.JOBHOST[i]
+            if self.JOBSTATUS[i] == "FAIL":
+                remarkField = self.JOBREMARK[i]
+            print(jobFmt % (
                 self.JOBNUMBER[i],
                 self.JOBDIR[i],
                 self.JOBID[i],
-                self.JOBSTATUS[i],
+                self.JOBSTATUS[i][:6],
                 self.JOBNTRY[i],
                 self.JOBRUNTIME[i],
                 self.JOBNEVT[i],
-                self.JOBHOST[i],
+                remarkField[:12],
                 self.JOBSP2[i],
                 self.JOBSP3[i]))
 
         #print merge Jobs if merge mode
         if self.driver == 'merge':
             for i in range(self.nJobs,len(self.JOBDIR)):
-                print('%s  %6s  %9s  %6s  %3d  %5d  %8d  %8s  %5s  %s' % (
+                remarkField = self.JOBHOST[i]
+                if (self.JOBSTATUS[i] == "FAIL") or (self.JOBSTATUS[i] == "WARN"):
+                    remarkField = self.JOBREMARK[i]
+                print(mrgFmt % (
                     'MMM',
                     self.JOBDIR[i],
                     self.JOBID[i],
-                    self.JOBSTATUS[i],
+                    self.JOBSTATUS[i][:6],
                     self.JOBNTRY[i],
                     self.JOBRUNTIME[i],
                     self.JOBNEVT[i],
-                    self.JOBHOST[i],
+                    remarkField[:12],
                     self.JOBSP2[i],
                     self.JOBSP3[i]))
 
@@ -192,7 +211,7 @@ def print_memdb(self):
     #-------------------------------------------------------------------------------
     # writes a new mps.db file from the members. Replaces the old mps.db
     def write_db(self):
-        self.header = "mps database schema 3.2"
+        self.header = "mps database schema 4.0"
         self.currentTime = int(time.time())
         self.elapsedTime = 0;
         if self.updateTime != 0:
@@ -218,7 +237,7 @@ def write_db(self):
 
         #write mps.db jobinfo
         for i in range(len(self.JOBID)):
-            DBFILE.write('%03d:%s:%s:%s:%s:%s:%s:%s:%s:%s:%s:%s:%s\n' %
+            DBFILE.write('%d:%s:%s:%s:%s:%s:%s:%s:%s:%s:%s:%s:%s\n' %
                          (i+1,
                           self.JOBDIR[i],
                           self.JOBID[i],
@@ -252,3 +271,61 @@ def get_class(self, argument=''):
         else:
             print('\nget_class():\n  Know class only for \'mille\' or \'pede\', not %s!\n\n' %argument)
             sys.exit(1)
+
+    #-------------------------------------------------------------------------------
+    #  Take card file, blank all INFI directives and insert the INFI directives 
+    #  from the modifier file instead
+    def mps_splice(self,inCfg='',modCfg='',outCfg='the.py',isn=0,skip_events=0,max_events=0):
+
+        with open(inCfg, 'r') as INFILE:
+            body = INFILE.read()
+
+        # read modifier file
+        with open(modCfg, 'r') as MODFILE:
+            mods = MODFILE.read()
+        mods = mods.strip()
+
+        # prepare the new fileNames directive. Delete first line if necessary.
+        fileNames = mods.split('\n')
+        if 'CastorPool=' in fileNames[0]:
+            del fileNames[0]
+
+        # replace ISN number (input is a string of three digit number with leading zeros though)
+        body = re.sub(re.compile('ISN',re.M), isn, body)
+
+        # print to outCfg
+        with open(outCfg, 'w') as OUTFILE:
+            OUTFILE.write(body)
+
+        # Number of total files and number of extends for cms.vstring are needed
+        numberOfFiles   = len(fileNames)
+        numberOfExtends = int(math.ceil(numberOfFiles/255.))
+
+        # Create and insert the readFile.extend lines
+        for j in range(numberOfExtends):
+            insertBlock = "readFiles.extend([\n    "
+            i=0
+            currentStart = j*255
+            while (i<255) and ((currentStart+i)<numberOfFiles):
+                entry = fileNames[currentStart+i].strip()
+                if (i==254) or ((currentStart+i+1)==numberOfFiles):
+                    insertBlock += "\'"+entry+"\'])\n"
+                else:
+                    insertBlock += "\'"+entry+"\',\n    "
+                i+=1
+
+            for line in fileinput.input(outCfg, inplace=1):
+                print(line,end='')
+                if re.match('readFiles\s*=\s*cms.untracked.vstring()',line):
+                    print(insertBlock,end='')
+
+        if skip_events != 0:
+            with open(outCfg, "a") as f:
+                f.write("process.source.skipEvents = cms.untracked.uint32({0:d})\n"
+                        .format(skip_events))
+
+        if max_events != 0:
+            with open(outCfg, "a") as f:
+                f.write("process.maxEvents = cms.untracked.PSet(input = "
+                        "cms.untracked.int32({0:d}))\n".format(max_events))
+

Alignment/MillePedeAlignmentAlgorithm/scripts/mps_check.py

@@ -34,6 +34,7 @@
     batchExited = 0
     finished = 0
     endofjob = 0
+    badInputFile = 0
     eofile = 1  # do not deal with timel yet
     timel = 0
     killed = 0
@@ -58,6 +59,11 @@
     quotaspace = 0
     copyerr=0
     ispede=0
+    mismatchedInputFiles = 0
+    missingBinaryfile = 0
+    missingTreefile = 0
+    missingMonitor = 0
+    zeroEvents = 0
 
     kill_reason = None
     pedeLogErrStr = ""
@@ -116,9 +122,15 @@
                     if match:
                         cpuFactor = 2.125
                         cputime = int(round(int(match.group(1))/cpuFactor)) # match.group(1) is the matched digit
+                    if re.search(re.compile('Missing milleBinary',re.M), line):
+                         missingBinaryfile = 1
+                    if re.search(re.compile('Missing treeFile',re.M), line):
+                         missingTreefile = 1
+                    if re.search(re.compile('Missing Monitor',re.M), line):
+                         missingMonitor = 1
 
             # gzip it afterwards:
-            print('gzip -f '+stdOut)
+            print('Checked '+stdOut)
             os.system('gzip -f '+stdOut)
         except IOError as e:
             if e.args == (2, "No such file or directory"):
@@ -132,7 +144,7 @@
             condor_log = subprocess.check_output(["condor_q", lib.JOBID[i],
                                                   "-userlog", log_file,
                                                   "-af",
-                                                  "RemoteSysCpu",
+                                                  "RemoteUserCpu",
                                                   "JobStatus",
                                                   "RemoveReason"],
                                                  stderr = subprocess.STDOUT).decode()
@@ -186,6 +198,10 @@
             with open(eazeLog,'r') as INFILE:
                 # scan records in input file
                 for line in INFILE:
+                    # check whether any file could not be opened
+                    if re.search(re.compile('Failed to open the file',re.M), line):
+                        badInputFile = 1
+                        break
                     # check if end of file has been reached
                     if re.search(re.compile('\<StorageStatistics\>',re.M), line):
                         eofile = 1
@@ -211,6 +227,10 @@
                     # AP 07.09.2009 - Check that the job got to a normal end
                     if re.search(re.compile('AlignmentProducerAsAnalyzer::endJob\(\)',re.M), line):
                         endofjob = 1
+                    if re.search(re.compile('MismatchedInputFiles',re.M), line):
+                        mismatchedInputFiles = 1
+                    if re.search(re.compile('Did not process any events',re.M), line):
+                        zeroEvents = 1
                     if re.search(re.compile('FwkReport            -i main_input:sourc',re.M), line):
                         array = line.split()
                         nEvent = int(array[5])
@@ -221,6 +241,11 @@
                     if nEvent==0 and re.search(re.compile('FwkReport            -i AfterSource',re.M), line):
                         array = line.split()
                         nEvent = int(array[5])
+                    # RM 03.02.2023
+                    if nEvent==0:
+                        x = re.search(r"FwkReport            -f AfterSource\s+(\d+)",line)
+                        if x:
+                            nEvent = int(x.group(1))
 
             if logZipped == 'true':
                 os.system('gzip -f '+eazeLog)
@@ -238,7 +263,6 @@
             #$mOutSize = `nsls -l $mssDir | grep $milleOut | head -1 | awk '{print \$5}'`;
             #$mOutSize = `cmsLs -l $mssDir | grep $milleOut | head -1 | awk '{print \$2}'`;
             mOutSize = 0
-            #print(">>>eoslsoutput:", eoslsoutput, " \ttype(eoslsoutput):", type(eoslsoutput))
             for line in eoslsoutput:
                 if milleOut in line:
                     columns = line.split()
@@ -369,6 +393,10 @@
         farmhost = ' '
 
         okStatus = 'OK'
+        if badInputFile == 1:
+            print(lib.JOBDIR[i],lib.JOBID[i],'had FileOpenError')
+            okStatus = 'FAIL'
+            remark = 'FileOpenError'
         if not eofile == 1:
             print(lib.JOBDIR[i],lib.JOBID[i],'did not reach end of file')
             okStatus = 'ABEND'
@@ -455,7 +483,26 @@
             remark = 'copy to eos failed'
             okStatus = 'FAIL'
 
-
+        if missingBinaryfile == 1:
+            print(lib.JOBDIR[i],lib.JOBID[i],'Missing binary file')
+            remark = 'Missing binary file'
+            okStatus = 'FAIL'
+        if missingTreefile == 1:
+            print(lib.JOBDIR[i],lib.JOBID[i],'Missing Treefile')
+            remark = 'Missing Treefile'
+            okStatus = 'FAIL'
+        if missingMonitor == 1:
+            print(lib.JOBDIR[i],lib.JOBID[i],'Missing Monitor')
+            remark = 'Missing Monitor file'
+            okStatus = 'FAIL'
+        if zeroEvents == 1:
+            print(lib.JOBDIR[i],lib.JOBID[i],'Did not process any events')
+            remark = 'Did not process any events'
+            okStatus = 'FAIL'
+        if mismatchedInputFiles == 1:
+            print(lib.JOBDIR[i],lib.JOBID[i],'MismatchedInputFiles')
+            remark = 'MismatchedInputFiles'
+            okStatus = 'FAIL'
         # print warning line to stdout
         if okStatus != "OK":
             print(lib.JOBDIR[i],lib.JOBID[i],' -------- ',okStatus)

Alignment/MillePedeAlignmentAlgorithm/scripts/mps_kill.pl

@@ -30,7 +30,7 @@ BEGIN
 
 # parse the arguments
 while (@ARGV) {
-  $arg = shift(ARGV);
+  $arg = shift(@ARGV);
   if ($arg =~ /\A-/) {  # check for option 
     if ($arg =~ "h") {
       $helpwanted = 1;

Alignment/MillePedeAlignmentAlgorithm/scripts/mps_retry.pl

@@ -26,7 +26,7 @@ BEGIN
 
 # parse the arguments
 while (@ARGV) {
-  $arg = shift(ARGV);
+  $arg = shift(@ARGV);
   if ($arg =~ /\A-/) {  # check for option 
     if ($arg =~ "h") {
       $helpwanted = 1;

Alignment/MillePedeAlignmentAlgorithm/scripts/mps_setup.py

@@ -11,7 +11,6 @@
 import Alignment.MillePedeAlignmentAlgorithm.mpslib.tools as mps_tools
 import Alignment.MillePedeAlignmentAlgorithm.mpslib.Mpslibclass as mpslib
 
-
 parser = argparse.ArgumentParser(description = "Setup local mps database")
 parser.add_argument("-m", "--setup-merge", dest = "setup_merge",
                     action = "store_true", default = False,
@@ -157,28 +156,25 @@
 # Create the job directories
 nJobExist = 0
 if args.append and os.path.isdir("jobData"):
-    # Append mode, and "jobData" exists
+    # Append mode, and "jobData" exists. Find the highest existing job number
     jobs = os.listdir("jobData")
-    job_regex = re.compile(r"job([0-9]{3})") # should we really restrict it to 3 digits?
-    existing_jobs = [job_regex.search(item) for item in jobs]
-    existing_jobs = [int(job.group(1)) for job in existing_jobs if job is not None]
-    nJobExist = sorted(existing_jobs)[-1]
-
-if nJobExist == 0 or nJobExist <=0 or nJobExist > 999: # quite rude method... -> enforce job number limit earlier?
-    # Delete all
-    mps_tools.remove_existing_object("jobData")
-    os.makedirs("jobData")
-    nJobExist = 0;
+    job_regex = re.compile(r"job(\d+)") # can have any number of digits
+    existing_jobs_set = set()
+    for item in jobs:
+        job_regex = re.compile(r"job(\d+)")
+        x = job_regex.search(item)
+        if x:
+            #print(x.group(1))
+            existing_jobs_set.add(int(x.group(1)))
+    nJobExist = max(existing_jobs_set)
 
 for j in range(1, args.n_jobs + 1):
     i = j+nJobExist
     jobdir = "job{0:03d}".format(i)
-    print("jobdir", jobdir)
     os.makedirs(os.path.join("jobData", jobdir))
 
 # build the absolute job directory path (needed by mps_script)
 theJobData = os.path.abspath("jobData")
-print("theJobData =", theJobData)
 
 if args.append:
     # save current values
@@ -243,7 +239,7 @@
     cmd = ["mps_split.pl", args.input_file_list,
            str(j if args.max_events is None else 1),
            str(args.n_jobs if args.max_events is None else 1)]
-    print(" ".join(cmd)+" > jobData/{}/theSplit".format(jobdir))
+    #print(" ".join(cmd)+" > jobData/{}/theSplit".format(jobdir))
     with open("jobData/{}/theSplit".format(jobdir), "w") as f:
         try:
             subprocess.check_call(cmd, stdout = f)
@@ -253,19 +249,20 @@
     theIsn = "{0:03d}".format(i)
 
     # create the cfg file
-    cmd = ["mps_splice.py", args.config_template,
-           "jobData/{}/theSplit".format(jobdir),
-           "jobData/{}/the.py".format(jobdir), theIsn]
+    skip_events = 0
+    max_events = 0
     if args.max_events is not None:
         chunk_size = int(args.max_events/args.n_jobs)
-        event_options = ["--skip-events", str(chunk_size*(j-1))]
+        skip_events = chunk_size*(j-1)
         max_events = (args.max_events - (args.n_jobs-1)*chunk_size
                       if j == args.n_jobs    # last job gets the remaining events
                       else chunk_size)
-        event_options.extend(["--max-events", str(max_events)])
-        cmd.extend(event_options)
-    print(" ".join(cmd))
-    mps_tools.run_checked(cmd)
+    
+    lib.mps_splice(args.config_template,
+                   "jobData/{}/theSplit".format(jobdir),
+                   "jobData/{}/the.py".format(jobdir),
+                   theIsn)
+
 
     # create the run script
     print("mps_script.pl {}  jobData/{}/theScript.sh {}/{} the.py jobData/{}/theSplit {} {} {}".format(args.batch_script, jobdir, theJobData, jobdir, jobdir, theIsn, args.mss_dir, lib.mssDirPool))