Configuration/PyReleaseValidation/python/WorkFlowRunner.py - master - githubmirror/cmssw

Configuration/PyReleaseValidation/python/WorkFlowRunner.py
332 строки14 KB
Kevin Pedro
fix syntax errors
22 май 2026, 18:54
22 май 2026, 18:549c51c7e
from threading import Thread
from Configuration.PyReleaseValidation import WorkFlow
import os,time
import shutil
import re
from subprocess import Popen 
from os.path import exists, basename, join
from datetime import datetime

class WorkFlowRunner(Thread):
    def __init__(self, wf, opt, noRun=False, dryRun=False, cafVeto=True, jobNumber=None, gpu = None):
        Thread.__init__(self)
        self.wf = wf

        self.status = -1
        self.report  =''
        self.nfail = 0
        self.npass = 0
        self.noRun = noRun
        self.dryRun = dryRun
        self.cafVeto = cafVeto
        self.gpu = gpu

        self.dasOptions = opt.dasOptions
        self.jobReport = opt.jobReports
        self.nThreads = opt.nThreads
        self.nStreams = opt.nStreams
        self.maxSteps = opt.maxSteps
        self.nEvents = opt.nEvents
        self.recoOutput = ''
        self.startFrom = opt.startFrom
        self.recycle = opt.recycle
        self.useRNTuple = opt.useRNTuple
        
        self.wfDir=str(self.wf.numId)+'_'+self.wf.nameId
        if jobNumber is not None:
            self.wfDir = self.wfDir + '_job' + str(jobNumber)

        return

    def doCmd(self, cmd):

        msg = "\n# in: " +os.getcwd()
        if self.dryRun: msg += " dryRun for '"
        else:      msg += " going to execute "
        msg += cmd.replace(';','\n')
        print(msg)

        cmdLog = open(self.wfDir+'/cmdLog','a')
        cmdLog.write(msg+'\n')
        cmdLog.close()
        
        ret = 0
        if not self.dryRun:
            p = Popen(cmd, shell=True)
            ret = os.waitpid(p.pid, 0)[1]
            if ret != 0:
                print("ERROR executing ",cmd,'ret=', ret)

        return ret
    
    @staticmethod
    def replace_filein_extensions(command_line, outputExtensionForStep, defaultExtension, fileOption='--filein'):
        # Pattern to match --filein followed by file:file.ext entries (comma-separated)
        filein_pattern = re.compile(
            r'('+fileOption+r'\s+)((?:file:[a-zA-Z0-9_]+\.[a-z]+(?:,\s*)?)*)'
        )

        # Inner patterns to match individual file entries
        # For stepN naming need to know the N
        file_pattern_step = re.compile(r'file:step([1-9]+)(_[a-zA-Z]+)?\.[a-z]+')
        # Some ALCA steps use special file names without stepN, those
        # are assumed to use the default extension
        file_pattern_gen = re.compile(r'file:([a-zA-Z0-9_]+)\.[a-z]+')

        def replace_filein_match(filein_match):
            filein_prefix = filein_match.group(1)
            file_list_str = filein_match.group(2)

            # Replace extensions in the file list
            m = file_pattern_step.search(file_list_str)
            if m:
                new_file_list = file_pattern_step.sub(
                    lambda m: 'file:step{0}{1}{2}'.format(m.group(1), m.group(2) or "", outputExtensionForStep[int(m.group(1))]),
                    file_list_str
                )
            else:
                new_file_list = file_pattern_gen.sub(
                    lambda m: 'file:{0}{1}'.format(m.group(1), defaultExtension),
                    file_list_str
                )

            return filein_prefix + new_file_list

        # Replace the whole --filein section with updated extensions
        new_command_line = filein_pattern.sub(replace_filein_match, command_line)

        return new_command_line 

    def run(self):

        startDir = os.getcwd()

        if not os.path.exists(self.wfDir):
            os.makedirs(self.wfDir)
        elif not self.dryRun: # clean up to allow re-running in the same overall devel area, then recreate the dir to make sure it exists
            print("cleaning up ", self.wfDir, ' in ', os.getcwd())
            shutil.rmtree(self.wfDir) 
            os.makedirs(self.wfDir)

        preamble = 'cd '+self.wfDir+'; '
       
        realstarttime = datetime.now()
        startime='date %s' %time.asctime()

        # check where we are running:
        onCAF = False
        if 'cms/caf/cms' in os.environ['CMS_PATH']:
            onCAF = True

        ##needs to set
        #self.report
        self.npass  = []
        self.nfail = []
        self.stat = []
        self.retStep = []

        def closeCmd(i,ID):
            return ' > %s 2>&1; ' % ('step%d_'%(i,)+ID+'.log ',)

        # For --secondfilein the primary and secondary files must have
        # the same format (TTree or RNTuple). For now find the last
        # step that uses --secondfilein, and use TTree for all steps
        # up to that step. Theoretically we could identify the exact
        # steps that need TTree output in this case, but given the way
        # --secondfilein is being used now, and the deployment plan
        # for RNTuple for HL-LHC, that complexity does not seem worth it.
        lastStepWithSecondFileIn = None
        if self.useRNTuple:
            for (istepmone,com) in enumerate(self.wf.cmds):
                # I don't know what to do in case com is something else
                if isinstance(com, str):
                    if "--secondfilein" in com:
                        lastStepWithSecondFileIn = istepmone+1

        inFile=None
        lumiRangeFile=None
        aborted=False
        outputExtensionForStep = {}
        for (istepmone,com) in enumerate(self.wf.cmds):
            # isInputOk is used to keep track of the das result. In case this
            # is False we use a different error message to indicate the failed
            # das query.
            isInputOk=True
            istep=istepmone+1
            cmd = preamble
            outputExtensionForStep[istep]=''
            if aborted:
                self.npass.append(0)
                self.nfail.append(0)
                self.retStep.append(0)
                self.stat.append('NOTRUN')
                continue
            if not isinstance(com,str):
                if self.recycle:
                    inFile = self.recycle
                    continue
                if self.cafVeto and (com.location == 'CAF' and not onCAF):
                    print("You need to be no CAF to run",self.wf.numId)
                    self.npass.append(0)
                    self.nfail.append(0)
                    self.retStep.append(0)
                    self.stat.append('NOTRUN')
                    aborted=True
                    continue
                #create lumiRange file first so if das fails we get its error code
                cmd2 = com.lumiRanges()
                if cmd2:
                    cmd2 =cmd+cmd2+closeCmd(istep,'lumiRanges')
                    lumiRangeFile='step%d_lumiRanges.log'%(istep,)
                    retStep = self.doCmd(cmd2)
                if (com.dataSetParent):
                    cmd3=cmd+com.das(self.dasOptions,com.dataSetParent)+closeCmd(istep,'dasparentquery')
                    retStep = self.doCmd(cmd3)
                cmd+=com.das(self.dasOptions,com.dataSet)
                cmd+=closeCmd(istep,'dasquery')
                retStep = self.doCmd(cmd)
                #don't use the file list executed, but use the das command of cmsDriver for next step
                # If the das output is not there or it's empty, consider it an
                # issue of this step, not of the next one.
                dasOutputPath = join(self.wfDir, 'step%d_dasquery.log'%(istep,))
                # Check created das output in no-dryRun mode only
                if not self.dryRun:
                    if not exists(dasOutputPath):
                        retStep = 1
                        dasOutput = None
                    else:
                        # We consider only the files which have at least one logical or physical filename
                        # in it. This is because sometimes das fails and still prints out junk.
                        dasOutput = [l for l in open(dasOutputPath).read().split("\n") if l.startswith("/") or l.startswith("root://eoscms.cern.ch")]
                    if not dasOutput:
                        retStep = 1
                        isInputOk = False
                 
                inFile = 'filelist:' + basename(dasOutputPath)

                if com.skimEvents:
                    lumiRangeFile='step%d_lumiRanges.log'%(istep,)
                    cmd2 = preamble + "mv lumi_ranges.txt " + lumiRangeFile
                    retStep = self.doCmd(cmd2)

                print("---")

            else:
                #chaining IO , which should be done in WF object already and not using stepX.root but <stepName>.root
                if self.gpu is not None:
                    cmd = cmd + self.gpu

                cmd += com

                if self.useRNTuple and not \
                   (lastStepWithSecondFileIn is not None and istep < lastStepWithSecondFileIn):
                    cmd+=' --rntuple_out'
                if self.startFrom:
                    steps = cmd.split("-s ")[1].split(" ")[0]
                    if self.startFrom not in steps:
                        continue
                    else:
                        self.startFrom = False
                        inFile = self.recycle
                
                if self.noRun:
                    cmd +=' --no_exec'
                # in case previous step used DAS query (either filelist of das:)
                # not to be applied for premixing stage1 to allow combined stage1+stage2 workflow
                # & similar for HybridPU combined workflow
                if inFile and not 'premix_stage1' in cmd and not 'FASTSIM' in cmd:
                    cmd += ' --filein '+inFile
                    inFile=None
                if lumiRangeFile: #DAS query can also restrict lumi range
                    cmd += ' --lumiToProcess '+lumiRangeFile
                    lumiRangeFile=None
                # 134 is an existing workflow where harvesting has to operate on AlcaReco and NOT on DQM; hard-coded..    
                if 'HARVESTING' in cmd and not 134==self.wf.numId and not '--filein' in cmd:
                    cmd+=' --filein file:step%d_inDQM.root --fileout file:step%d.root '%(istep-1,istep)
                    outputExtensionForStep[istep] = '.root'
                else:
                    # Disable input for premix stage1 to allow combined stage1+stage2 workflow
                    # Disable input for premix stage2 in FastSim to allow combined stage1+stage2 workflow (in FS, stage2 does also GEN)
                    # & similar for HybridPU combined workflow
                    # Ugly hack but works
                    extension = '.root'
                    if '--rntuple_out' in cmd:
                        extension = '.rntpl'
                    outputExtensionForStep[istep] = extension
                    if istep!=1 and not '--filein' in cmd and not 'premix_stage1' in cmd and not ("--fast" in cmd and "premix_stage2" in cmd) and not 'FASTSIM' in cmd:
                        steps = cmd.split("-s ")[1].split(" ")[0] ## relying on the syntax: cmsDriver -s STEPS --otherFlags
                        if "ALCA" not in steps:
                            cmd+=' --filein  file:step%s%s '%(istep-1,outputExtensionForStep[istep-1])
                        elif "ALCA" in steps and "RECO" in steps:
                            cmd+=' --filein  file:step%s%s '%(istep-1,outputExtensionForStep[istep-1])
                        elif self.recoOutput:
                            cmd+=' --filein %s'%(self.recoOutput)
                        else:
                            cmd+=' --filein  file:step%s%s '%(istep-1,outputExtensionForStep[istep-1])
                    elif istep!=1 and '--filein' in cmd and '--filetype' not in cmd:
                        # make sure correct extension is being used
                        cmd = self.replace_filein_extensions(cmd, outputExtensionForStep, extension)
                    if '--pileup_input' in cmd and '--filetype' not in cmd:
                        # make sure correct extension is being used
                        cmd = self.replace_filein_extensions(cmd, outputExtensionForStep, extension, fileOption='--pileup_input')
                    if not '--fileout' in com:
                        cmd+=' --fileout file:step%s%s '%(istep,extension)
                        if "RECO" in cmd:
                            self.recoOutput = "file:step%d%s"%(istep,extension)
                if self.jobReport:
                  cmd += ' --suffix "-j JobReport%s.xml " ' % istep
                if (self.nThreads > 1) and ('HARVESTING' not in cmd) and ('ALCAHARVEST' not in cmd):
                  cmd += ' --nThreads %s' % self.nThreads
                if (self.nStreams > 0) and ('HARVESTING' not in cmd) and ('ALCAHARVEST' not in cmd):
                  cmd += ' --nStreams %s' % self.nStreams
                if (self.nEvents > 0):
                  event_token = " -n "
                  split = cmd.split(event_token)
                  pos_cmd = " ".join(split[1].split(" ")[1:])
                  cmd = split[0] + event_token + '%s ' % self.nEvents + pos_cmd
                cmd+=closeCmd(istep,self.wf.nameId)            
                retStep = 0

                if istep>self.maxSteps:
                   wf_stats = open("%s/wf_steps.txt" % self.wfDir,"a")
                   wf_stats.write('step%s:%s\n' % (istep, cmd))
                   wf_stats.close()
                else: retStep = self.doCmd(cmd)
            
            self.retStep.append(retStep)
            if retStep == 32000:
                # A timeout occurred
                self.npass.append(0)
                self.nfail.append(1)
                self.stat.append('TIMEOUT')
                aborted = True
            elif (retStep!=0):
                #error occured
                self.npass.append(0)
                self.nfail.append(1)
                if not isInputOk:
                  self.stat.append("DAS_ERROR")
                else:
                  self.stat.append('FAILED')
                #to skip processing
                aborted=True
            else:
                #things went fine
                self.npass.append(1)
                self.nfail.append(0)
                self.stat.append('PASSED')

        os.chdir(startDir)
        endtime='date %s' %time.asctime()
        tottime='%s-%s'%(endtime,startime)
        

        #### wrap up ####

        logStat=''
        for i,s in enumerate(self.stat):
            logStat+='Step%d-%s '%(i,s)
        #self.report='%s_%s+%s %s - time %s; exit: '%(self.wf.numId,self.wf.nameId,'+'.join(self.wf.stepList),logStat,tottime)+' '.join(map(str,self.retStep))+'\n'
        self.report='%s_%s %s - time %s; exit: '%(self.wf.numId,self.wf.nameId,logStat,tottime)+' '.join(map(str,self.retStep))+'\n'

        return