python.PDSFJobRunner.PDSFJobRunner Class Reference

Inheritance diagram for python.PDSFJobRunner.PDSFJobRunner:

Collaboration diagram for python.PDSFJobRunner.PDSFJobRunner:

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

Public Attributes
	params
	paramOrder
	paramToBeCopied
	jobs
	jobStatus

Detailed Description

PDSFJobRunner - run jobs using the SGE batch system on PDSF

Definition at line 36 of file PDSFJobRunner.py.

Constructor & Destructor Documentation

__init__()

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

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

Reimplemented from python.JobRunner.JobRunner.

Definition at line 39 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()

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

inherited

Add a list of input files to be processed.

Definition at line 248 of file JobRunner.py.

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

addFilesToPoolFileCatalog()

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

inherited

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

Definition at line 253 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()

def 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 178 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()

def 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 210 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()

def python.JobRunner.JobRunner.configure (   self )

inherited

Configure all jobs.

Definition at line 375 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()

def 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 258 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()

def 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 237 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()

def python.JobRunner.JobRunner.getNJobs (   self )

inherited

Definition at line 477 of file JobRunner.py.

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

getOutputFiles()

def python.JobRunner.JobRunner.getOutputFiles (   self )

inherited

Get a list of all output files.

Definition at line 504 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()

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

inherited

Get value of a parameter by name.

Definition at line 193 of file JobRunner.py.

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

getRunningJobs()

def python.JobRunner.JobRunner.getRunningJobs (   self )

inherited

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

Definition at line 484 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()

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

inherited

Check if parameter requires special treatment.

Definition at line 198 of file JobRunner.py.

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

log()

def 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 513 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()

def 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 203 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()

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

inherited

Run all configured job(s)

Definition at line 466 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()

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

inherited

Run a single configured job.

Definition at line 456 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()

def 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 158 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()

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

inherited

Show current job parameters.

Definition at line 228 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()

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

Submit a JobRunner job as a SGE batch job.

Reimplemented from python.JobRunner.JobRunner.

Definition at line 54 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()

def python.JobRunner.JobRunner.wait (   self )

inherited

Wait until all jobs have completed

Definition at line 493 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

jobs

python.JobRunner.JobRunner.jobs

inherited

Definition at line 107 of file JobRunner.py.

jobStatus

python.JobRunner.JobRunner.jobStatus

inherited

Definition at line 108 of file JobRunner.py.

paramOrder

python.JobRunner.JobRunner.paramOrder

inherited

Definition at line 105 of file JobRunner.py.

params

python.JobRunner.JobRunner.params

inherited

Definition at line 104 of file JobRunner.py.

paramToBeCopied

python.JobRunner.JobRunner.paramToBeCopied

inherited

Definition at line 106 of file JobRunner.py.

---

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

• PDSFJobRunner.py