python.PDSFJobRunner.PDSFJobRunner Class Reference

Inheritance diagram for python.PDSFJobRunner.PDSFJobRunner:

Collaboration diagram for python.PDSFJobRunner.PDSFJobRunner:

Public Member Functions
	__init__ (self, **params)
	submitJob (self, jobConfig)
	setParam (self, name, value=None, description=None, isSpecial=None, insertAtFront=False)
	appendParam (self, name, value=None, description=None, isSpecial=None, insertAtFront=False, endOfLine='\n')
	getParam (self, name)
	isSpecialParam (self, name)
	registerToBeCopied (self, paramName)
	checkParams (self)
	showParams (self, maxLineLength=80)
	dumpParams (self, format='%(name) -20s=%(value) s', maxLineLength=0)
	addFiles (self, fileList)
	addFilesToPoolFileCatalog (self, fileList)
	configureJob (self, jobnr, inputList=[])
	configure (self)
	runJob (self, jobnr)
	run (self, jobnr=None)
	getNJobs (self)
	getRunningJobs (self)
	wait (self)
	getOutputFiles (self)
	log (self, subject, body, **data)

Public Attributes
dict	params = { }
list	paramOrder = [ ]
list	paramToBeCopied = [ ]
dict	jobs = { }
dict	jobStatus = { }

Private Attributes
	__class__

Detailed Description

PDSFJobRunner - run jobs using the SGE batch system on PDSF

Definition at line 35 of file PDSFJobRunner.py.

Constructor & Destructor Documentation

__init__()

python.PDSFJobRunner.PDSFJobRunner.__init__	(		self,
		**	params )

Constructor (takes any number of parameters as an argument).

Definition at line 38 of file PDSFJobRunner.py.

    def __init__(self,**params):
        """Constructor (takes any number of parameters as an argument)."""
        JobRunner.__init__(self)
 
        # Set user-specified parameters only after PDSFJobRunner defaults
        self.setParam('batchqueue','all.64bit.q','Batch queue')
        script = self.getParam('script')
        self.setParam('script',preScript+script)
        for k in params.keys():
            self.setParam(k,params[k])
        # SGE runs jobs from where they were submitted, so change rundir explicitly to jobdir
        self.setParam('rundir','%(jobdir)s')
        self.setParam('maketmpdir','')
        self.checkParams()
 

Member Function Documentation

addFiles()

python.JobRunner.JobRunner.addFiles ( self, fileList )

inherited

Add a list of input files to be processed.

Definition at line 247 of file JobRunner.py.

    def addFiles(self,fileList):
        """ Add a list of input files to be processed."""
        self.getParam('inputfiles').extend(fileList)
 
 

addFilesToPoolFileCatalog()

python.JobRunner.JobRunner.addFilesToPoolFileCatalog ( self, fileList )

inherited

Add a list of files to be inserted to the POOL file catalog.

Definition at line 252 of file JobRunner.py.

    def addFilesToPoolFileCatalog(self,fileList):
        """Add a list of files to be inserted to the POOL file catalog."""
        self.getParam('filesforpoolcatalog').extend(fileList)
 
 

appendParam()

python.JobRunner.JobRunner.appendParam ( self, name, value = None, description = None, isSpecial = None, insertAtFront = False, endOfLine = '\n' )

inherited

Append to the value of a job parameter. If the parameter doesn't exist yet,
   setParam is called to create it. If it does exist, only the value is updated
   and the description and flag arguments are ignored.

Definition at line 177 of file JobRunner.py.

    def appendParam(self,name,value=None,description=None,isSpecial=None,insertAtFront=False,endOfLine='\n'):
        """Append to the value of a job parameter. If the parameter doesn't exist yet,
           setParam is called to create it. If it does exist, only the value is updated
           and the description and flag arguments are ignored."""
        if name in self.params:
            if value is not None:
                p = self.params[name]
                if isinstance(p.value,str) and p.value:
                    p.value = p.value + endOfLine + value
                else:
                    p.value = p.value + value
        else:
            self.setParam(name,value,description,isSpecial,insertAtFront)
 
 

checkParams()

python.JobRunner.JobRunner.checkParams ( self )

inherited

A generic routine for checking job parameters. Check that all parameters
   can be evaluated. May be overrridden by a subclass if desired.

Definition at line 209 of file JobRunner.py.

    def checkParams(self):
        """A generic routine for checking job parameters. Check that all parameters
           can be evaluated. May be overrridden by a subclass if desired."""
        if len(self.paramOrder)!=len(self.params):
            raise JobRunnerError ('Inconsistent parameter definition')
        try:
            tmp = { }
            for p in self.paramOrder:
                value = ''   # this is necessary to prevent an exception from picking up the previous value
                value = self.getParam(p)
                if isinstance(value,str):
                    tmp[p] = value % tmp
                else:
                    tmp[p] = value
        except Exception:
            raise JobRunnerError ('Unable to evaluate parameter: '+p+' = '+value+' (check parameter order)')
 
 

configure()

python.JobRunner.JobRunner.configure ( self )

inherited

Configure all jobs.

Definition at line 374 of file JobRunner.py.

    def configure(self):
        """Configure all jobs."""
        if self.getParam('filesperjob')==0 or not self.getParam('inputfiles'):
            raise JobRunnerError ("No input files or illegal parameter 'filesperjob'")
        lbperjob = self.getParam('lbperjob')
        if lbperjob:
            # Bunched job submission with all files from LB range in single job
            if lbperjob<=0:
                raise JobRunnerError ('Negative number of luminosity blocks per job not allowed')
            inputfiles = self.getParam('inputfiles')
            jobInputDict = {}
            jobLBDict = {}
            lbpattern = re.compile(r'lb(\d+)')
            for f in inputfiles:
                lbnrs = lbpattern.findall(f)
                
                if(len(lbnrs) < 1) :
                    raise JobRunnerError ('Unable to determine luminosity block number of file %s' % f)
 
                if(len(lbnrs) > 2) :
                    raise JobRunnerError ('Too many luminosity block numbers in filename %s' % f)
 
                lbnr = int(lbnrs[0])
                if(len(lbnrs) > 1) :
                    lbnrmax = int(lbnrs[1])
                else :
                    lbnrmax = lbnr    
 
                while (lbnr <=  lbnrmax) :                    
 
                    jobId = int((lbnr-1)/lbperjob)
                    #print ('LB = %4i  jobid = %i' % (lbnr,jobId))
                    if jobId not in jobInputDict:
                        jobInputDict[jobId] = [f]
                        jobLBDict[jobId] = [lbnr]
                    else:
                        if f not in jobInputDict[jobId] :
                            jobInputDict[jobId].append(f)
                        if lbnr not in jobLBDict[jobId] :
                            jobLBDict[jobId].append(lbnr)
 
                    lbnr = lbnr+1
 
            maxJobs = self.getParam('maxjobs')
            if not maxJobs:
                maxJobs = len(jobInputDict)
            for i in  sorted(jobInputDict.keys())[:maxJobs]:
                if i<0:
                    jobnr = 0
                else:
                    jobnr = i*lbperjob+1    # use first LB number as job number
                self.setParam('lbList', jobLBDict[i])
                self.configureJob(jobnr,jobInputDict[i])
 
        else:
            # Normal (unbunched job submission)
            njobs = int(math.ceil(float(len(self.getParam('inputfiles')))/self.getParam('filesperjob')))
            if self.getParam('maxjobs'):
                njobs = min(self.getParam('maxjobs'),njobs)
            print ('\nConfiguring %i job(s) ...' % (njobs))
            for i in range(njobs):
                self.configureJob(i)
 
 

configureJob()

python.JobRunner.JobRunner.configureJob ( self, jobnr, inputList = [] )

inherited

Configure parameters for a single job and write job configuration files. If inputLIst
   is given, this is the list of input files used for the job.

Definition at line 257 of file JobRunner.py.

    def configureJob(self,jobnr,inputList=[]):
        """Configure parameters for a single job and write job configuration files. If inputLIst
           is given, this is the list of input files used for the job."""
        if jobnr in self.jobs:
            raise JobRunnerError ('Job number %s already configured' % jobnr)
        jobConfig = { }
 
        jobConfig['jobnr'] = jobnr
        self.setParam('jobnr',jobnr)
 
        # Check if any input files are available for this job
        if inputList:
            jobConfig['inputfiles'] = inputList
        else:
            inputfiles = self.getParam('inputfiles')
            iFirst = self.getParam('filesperjob')*jobnr
            if iFirst >= len(inputfiles):
                raise JobRunnerError ('Jobnr = %i too high for available number of files' % jobnr)
            iLast = iFirst + self.getParam('filesperjob')
            if iLast > len(inputfiles):
                iLast=len(inputfiles)
            jobConfig['inputfiles'] = inputfiles[iFirst:iLast]
 
        # Update regular parameters
        for p in self.paramOrder:
            if self.isSpecialParam(p): continue
            value = self.getParam(p)
            if isinstance(value,str):
                jobConfig[p] = value % jobConfig
            else:
                jobConfig[p] = value
 
        # Define statement to set up the release
        if jobConfig['setuprelease']:
            # Use explicit command if given
            jobConfig['cmdsetup'] = self.getParam('cmdsetup') % jobConfig
            # otherwise take from CMDSETUP env var if set or assume normal asetup 
            if not jobConfig['cmdsetup']:
                jobConfig['cmdsetup'] = os.getenv('CMDSETUP','source /afs/cern.ch/atlas/software/dist/AtlasSetup/scripts/asetup.sh %(release)s --noautocdtest') % jobConfig
        else:
            jobConfig['cmdsetup'] = ''
 
        # Collect files to be copied locally, and replace corresponding jobConfig
        # parameters by base file name (to point to the local copy)
        jobConfig['cmdcopyfiles'] = self.getParam('cmdcopyfiles') % jobConfig
        if not jobConfig['cmdcopyfiles']:
            filestobecopied = ''
            for p in self.paramToBeCopied:
                if not jobConfig[p]: continue
                filestobecopied = ' '.join([filestobecopied,jobConfig[p]])
                jobConfig[p] = os.path.basename(jobConfig[p])
            if filestobecopied:
                jobConfig['cmdcopyfiles'] = 'cp -p '+filestobecopied+' .'
 
        # Unless the user sets a specific value for parameter cmddefinepoolcatalog, determine
        # what should be added and create appropriate catalog
        jobConfig['cmddefinepoolcatalog'] = self.getParam('cmddefinepoolcatalog') % jobConfig
        if not jobConfig['cmddefinepoolcatalog']:
            if jobConfig['addinputtopoolcatalog']:
                argstring = ' '.join(jobConfig['filesforpoolcatalog']+jobConfig['inputfiles']) % jobConfig
            else:
                argstring = ' '.join(jobConfig['filesforpoolcatalog']) % jobConfig
            if argstring:
                jobConfig['cmddefinepoolcatalog'] = 'pool_insertFileToCatalog '+argstring
 
        # Define statement used to exit the shell script
        if jobConfig['returnstatuscode']:
            jobConfig['cmdexit'] = self.getParam('cmdexit') % jobConfig
            if not jobConfig['cmdexit']:
                jobConfig['cmdexit'] = 'exit $status' % jobConfig
        else:
            jobConfig['cmdexit'] = ''
  
        # Generate string with comma-separated list of output files with full name
        jobConfig['outputfilestring'] = ','.join([jobConfig['outputfileprefix']+f for f in jobConfig['outputfilelist']])
 
        # Update remaining special parameters
        if jobConfig['taskpostprocsteps']:
            jobConfig['doneflag'] = jobConfig['postprocflag']
        else:
            jobConfig['doneflag'] = jobConfig['completedflag']
        jobConfig['script'] = self.getParam('script') % jobConfig
 
        # Check if job's output files exist already in order to prevent overwriting of data
        # NOTE: cannot just check for existence of jobdir, since this might be set to the
        #       current directory
        for f in jobConfig['outputfilelist']:
            if os.access(jobConfig['outputfileprefix']+f,os.F_OK):
                raise JobRunnerError ('Job output file %s exists already' % jobConfig[f])
        for f in ('configfile', 'scriptfile', 'logfile'):
            if os.access(jobConfig[f],os.F_OK):
                raise JobRunnerError ('Job configuration or log file %s exists already' % jobConfig[f])
 
        # Make sure start directory where script and config files will be written to exists
        os.makedirs('%(jobdir)s' % jobConfig) 
 
        # Write job configuration file
        config = open(jobConfig['configfile'],'w')
        config.write('# Job configuration data for job %(jobname)s\n' % jobConfig)
        config.write('# Generated by JobRunner.py\n\n')
        if pprint.isreadable(jobConfig):
            config.write('jobConfig = '+pprint.pformat(jobConfig))
        else:
            config.write('jobConfig = '+repr(jobConfig))
        config.write('\n')
        config.close()
 
        # Write script file
        script = open(jobConfig['scriptfile'],'w')
        script.write(jobConfig['script'])
        script.close()
        os.chmod(jobConfig['scriptfile'],0o755)
 
        # Record job configuration
        self.jobs[jobnr] = jobConfig
 
 

dumpParams()

python.JobRunner.JobRunner.dumpParams ( self, format = '%(name)-20s = %(value)s', maxLineLength = 0 )

inherited

Dump all parameters into a string in a user-specified format.

Definition at line 236 of file JobRunner.py.

    def dumpParams(self,format='%(name)-20s = %(value)s',maxLineLength=0):
        """Dump all parameters into a string in a user-specified format."""
        dump = ''
        for p in self.paramOrder:
            s = format % self.params[p].__dict__
            if maxLineLength:
                s = s[0:maxLineLength-3]+'...'
            dump += s+'\n'
        return dump
 
 

getNJobs()

python.JobRunner.JobRunner.getNJobs ( self )

inherited

Definition at line 476 of file JobRunner.py.

    def getNJobs(self):
        if not self.jobs:
            return 0
        else:
            return len(self.jobs)
 
 

getOutputFiles()

python.JobRunner.JobRunner.getOutputFiles ( self )

inherited

Get a list of all output files.

Definition at line 503 of file JobRunner.py.

    def getOutputFiles(self):
        """Get a list of all output files."""
        outputFiles = [ ]
        for j in self.jobs:
            for f in sorted(glob.glob(self.jobs[j]['outputfileprefix']+'*')):
                outputFiles.append(f)
        return outputFiles
 
 

getParam()

python.JobRunner.JobRunner.getParam ( self, name )

inherited

Get value of a parameter by name.

Definition at line 192 of file JobRunner.py.

    def getParam(self,name):
        """Get value of a parameter by name."""
        return self.params[name].value
 
 

getRunningJobs()

python.JobRunner.JobRunner.getRunningJobs ( self )

inherited

Get list of jobs submitted by this JobRunner that are still running

Definition at line 483 of file JobRunner.py.

    def getRunningJobs(self):
        """Get list of jobs submitted by this JobRunner that are still running"""
        runningJobs = [ ]
        for j in self.jobs:
            if os.path.exists(self.jobs[j]['subflag']) or os.path.exists(self.jobs[j]['runflag']):
                runningJobs.append(self.jobs[j]['jobname'])
        return runningJobs
 
 

isSpecialParam()

python.JobRunner.JobRunner.isSpecialParam ( self, name )

inherited

Check if parameter requires special treatment.

Definition at line 197 of file JobRunner.py.

    def isSpecialParam(self,name):
        """Check if parameter requires special treatment."""
        return self.params[name].isSpecial
 
 

log()

python.JobRunner.JobRunner.log ( self, subject, body, ** data )

inherited

Send a log message to a set of users specified by parameter 'logmail'.

Definition at line 512 of file JobRunner.py.

    def log(self,subject,body,**data):
        """Send a log message to a set of users specified by parameter 'logmail'."""
        if self.getParam('logmail'):
            msg = 'Generated by JobRunner on host '+socket.gethostname()+' by '+pwd.getpwuid(os.getuid())[0]+' at '+time.asctime()
            msg += '\n\nCurrent working directory = '+os.getcwd()
            if body: msg += '\n\n'+body
            for k in data.keys():
                if isinstance(data[k],dict):
                    msg += "\n\n%s:" % k
                    items = data[k].keys()
                    items.sort()
                    for i in items:
                        msg += "\n    %-20s = %s" % (i,data[k][i])
                else:
                    msg += "\n\n%-20s = %s" % (k,data[k])
            subprocess.call("echo '%s' | mail -s '%s' '%s'" %
                (msg,subject,self.getParam('logmail')), shell=True)
 
 
#
# Test code
#

registerToBeCopied()

python.JobRunner.JobRunner.registerToBeCopied ( self, paramName )

inherited

Register a parameter holding the name of an input file to be
   copied before a job starts.

Definition at line 202 of file JobRunner.py.

    def registerToBeCopied(self,paramName):
        """Register a parameter holding the name of an input file to be
           copied before a job starts."""
        if paramName not in self.paramToBeCopied:
            self.paramToBeCopied.append(paramName)
 
 

run()

python.JobRunner.JobRunner.run ( self, jobnr = None )

inherited

Run all configured job(s)

Definition at line 465 of file JobRunner.py.

    def run(self,jobnr=None):
        """Run all configured job(s)"""
        if not self.jobs:
            raise JobRunnerError ('No configured jobs')
        njobs = len(self.jobs)
        self.log('JobRunner: starting '+str(njobs)+' jobs',self.dumpParams())
        for j in sorted(self.jobs.keys()):
            print ('\nStarting job %i (using %s) ...\n' % (j,self.__class__))
            self.runJob(j)
 
 

runJob()

python.JobRunner.JobRunner.runJob ( self, jobnr )

inherited

Run a single configured job.

Definition at line 455 of file JobRunner.py.

    def runJob(self,jobnr):
        """Run a single configured job."""
        if jobnr not in self.jobs:
            raise JobRunnerError ('Job number %s is not configured' % jobnr)
        jobConfig = self.jobs[jobnr]
        subprocess.call('touch '+jobConfig['subflag'], shell=True)
        status = self.submitJob(jobConfig)
        self.jobStatus[jobnr] = status
 
 

setParam()

python.JobRunner.JobRunner.setParam ( self, name, value = None, description = None, isSpecial = None, insertAtFront = False )

inherited

Set the value of a job parameter. Job parameters may be templates that name
   previously defined job parameters and will be evaluated when a given job
   is configured. Therefore the order in which setParam is called for the different
   parameters is relevant. insertAtFront can be set to True to force early
   evaluation of a given parameter.

Definition at line 157 of file JobRunner.py.

    def setParam(self,name,value=None,description=None,isSpecial=None,insertAtFront=False):
        """Set the value of a job parameter. Job parameters may be templates that name
           previously defined job parameters and will be evaluated when a given job
           is configured. Therefore the order in which setParam is called for the different
           parameters is relevant. insertAtFront can be set to True to force early
           evaluation of a given parameter."""
        p = self.params[name] = self.params.get(name,JobRunnerParameter(name))
        if value is not None:
            p.value = value
        if description is not None:
            p.description = description
        if isSpecial is not None:
            p.isSpecial=isSpecial
        if name not in self.paramOrder:
            if insertAtFront:
                self.paramOrder.insert(0,name)
            else:
                self.paramOrder.append(name)
 
 

showParams()

python.JobRunner.JobRunner.showParams ( self, maxLineLength = 80 )

inherited

Show current job parameters.

Definition at line 227 of file JobRunner.py.

    def showParams(self,maxLineLength=80):
        """Show current job parameters."""
        for p in self.paramOrder:
            s = str(self.params[p])
            if maxLineLength > 0 and len(s) > maxLineLength:
                s = s[0:maxLineLength-3] + '...'
            print (s)
 
 

submitJob()

python.PDSFJobRunner.PDSFJobRunner.submitJob	(		self,
			jobConfig )

Submit a JobRunner job as a SGE batch job.

Reimplemented from python.JobRunner.JobRunner.

Definition at line 53 of file PDSFJobRunner.py.

    def submitJob(self,jobConfig):
        """Submit a JobRunner job as a SGE batch job."""
        #batchCmd = 'qsub -q %(batchqueue)s -N %(jobname)s -j y -o %(logfile)s %(scriptfile)s' % jobConfig
        batchCmd = 'qsub -N %(jobname)s -j y -o %(logfile)s %(scriptfile)s' % jobConfig
        print (batchCmd)
        time.sleep(5)
        os.system(batchCmd)
        return None

wait()

python.JobRunner.JobRunner.wait ( self )

inherited

Wait until all jobs have completed

Definition at line 492 of file JobRunner.py.

    def wait(self):
        """Wait until all jobs have completed"""
        while 1:
            time.sleep(30)
            runningJobs = self.getRunningJobs()
            if not runningJobs: break
            print()
            print (time.asctime(),' Waiting for %2s job(s) (%s)' % (len(runningJobs),runningJobs))
        self.log('JobRunner: finished',self.dumpParams(),outputFiles=self.getOutputFiles())
 
 

Member Data Documentation

__class__

python.JobRunner.JobRunner.__class__

privateinherited

Definition at line 472 of file JobRunner.py.

jobs

python.JobRunner.JobRunner.jobs = { }

inherited

Definition at line 106 of file JobRunner.py.

jobStatus

dict python.JobRunner.JobRunner.jobStatus = { }

inherited

Definition at line 107 of file JobRunner.py.

paramOrder

python.JobRunner.JobRunner.paramOrder = [ ]

inherited

Definition at line 104 of file JobRunner.py.

params

dict python.JobRunner.JobRunner.params = { }

inherited

Definition at line 103 of file JobRunner.py.

paramToBeCopied

list python.JobRunner.JobRunner.paramToBeCopied = [ ]

inherited

Definition at line 105 of file JobRunner.py.

---

The documentation for this class was generated from the following file:

• PDSFJobRunner.py