Reference Manual

Inheritance diagram for production_tasks.MonitorJobs:

Public Member Functions
def	__init__

def	getjobid

def	monitor

def	run

Public Member Functions inherited from production_tasks.Task
def	__init__

def	addOption

def	getname

def	run

Additional Inherited Members
Public Attributes inherited from production_tasks.Task
	dataset

	instance

	name

	options

	user

Detailed Description

Monitor LSF jobs created with cmsBatch.py. Blocks until all jobs are finished.

Definition at line 575 of file production_tasks.py.

Constructor & Destructor Documentation

def production_tasks.MonitorJobs.__init__	(	self,
		dataset,
		user,
		options
	)

Definition at line 577 of file production_tasks.py.

 
     def __init__(self, dataset, user, options):
         Task.__init__(self,'MonitorJobs', dataset, user, options)

production_tasks.MonitorJobs.__init__

def __init__

Definition: production_tasks.py:577

Member Function Documentation

def production_tasks.MonitorJobs.getjobid	(	self,
		job_dir
	)

Parse the LSF output to find the job id

Definition at line 580 of file production_tasks.py.

References mergeVDriftHistosByStation.file, SiPixelLorentzAngle_cfi.read, and split.

Referenced by production_tasks.MonitorJobs.run().

 
     def getjobid(self, job_dir):
         """Parse the LSF output to find the job id"""
         input = os.path.join(job_dir,'job_id.txt')
         result = None
         if os.path.exists(input):
             contents = file(input).read()
             for c in contents.split('\n'):
                 if c and re.match('^Job <\\d*> is submitted to queue <.*>',c) is not None:
                     try:
                         result = c.split('<')[1].split('>')[0]
                     except Exception, e:
                         print >> sys.stderr, 'Job ID parsing error',str(e),c
         return result

mergeVDriftHistosByStation.file

tuple file

Definition: mergeVDriftHistosByStation.py:54

SiPixelLorentzAngle_cfi.read

tuple read

Definition: SiPixelLorentzAngle_cfi.py:7

production_tasks.MonitorJobs.getjobid

def getjobid

Definition: production_tasks.py:580

split

double split

Definition: MVATrainer.cc:139

def production_tasks.MonitorJobs.monitor	(	self,
		jobs,
		previous
	)

Definition at line 594 of file production_tasks.py.

References gen.parseHeader(), and split.

Referenced by production_tasks.MonitorJobs.run().

 
     def monitor(self, jobs, previous):
 
         #executes bjobs with a list of job IDs
         cmd = ['bjobs','-u',self.options.batch_user]
         cmd.extend([v for v in jobs.values() if v is not None])#filter out unknown IDs
         child = subprocess.Popen(cmd, stdout=subprocess.PIPE,stderr=subprocess.PIPE)
         stdout, stderr = child.communicate()
 
         def parseHeader(header):
             """Parse the header from bjobs"""
             tokens = [t for t in header.split(' ') if t]
             result = {}
             for i in xrange(len(tokens)):
                 result[tokens[i]] = i          
 
             return result
 
         result = {}
         if stdout:
             lines = stdout.split('\n')
             if lines:
                 header = parseHeader(lines[0])                                 
                 if not 'STAT' in header or not 'JOBID' in header:
                     print >> sys.stderr, 'Problem parsing bjobs header\n',lines
                     return result
                 for line in lines[1:]:
                     #TODO: Unreliable for some fields, e.g. dates
                     tokens = [t for t in line.split(' ') if t]
                     if len(tokens) < len(header): continue
                     id = tokens[header['JOBID']]
                     user = tokens[header['USER']]
                     status = tokens[header['STAT']]
 
                     result[id] = status               
                     
         if stderr:
             lines = stderr.split('\n')
             if lines:
                 for line in lines:
                     if line and re.match('^Job <\\d*> is not found',line) is not None:
                         try:
                             id = line.split('<')[1].split('>')[0]
                             if not result.has_key(id) and not previous.has_key(id):
                                 result[id] = 'FORGOTTEN'
                         except Exception, e:
                             print >> sys.stderr, 'Job ID parsing error in STDERR',str(e),line
         
         #after one hour the status is no longer available     
         if result:
             for id in jobs.values():
                 if not result.has_key(id) and previous.has_key(id):
                     result[id] = previous[id]
         return result

production_tasks.MonitorJobs.monitor

def monitor

Definition: production_tasks.py:594

gen::parseHeader

static std::map< std::string, std::string > parseHeader(const std::vector< std::string > &header)

Definition: JetMatchingMadgraph.cc:190

split

double split

Definition: MVATrainer.cc:139

def production_tasks.MonitorJobs.run	(	self,
		input
	)

Definition at line 648 of file production_tasks.py.

 
     def run(self, input):
 
         # return #COLIN
         jobsdir = input['RunCMSBatch']['LSFJobsTopDir']
         if not os.path.exists(jobsdir):
             raise Exception("LSF jobs dir does not exist: '%s'" % jobsdir)
         
         subjobs = [s for s in glob.glob("%s/Job_[0-9]*" % jobsdir) if os.path.isdir(s)]
         jobs = {}
         for s in subjobs:
             jobs[s] = self.getjobid(s)
 
         def checkStatus(stat):
             
             #gzip files on the fly
             actions = {'FilesToCompress':{'Files':[]}}
             
             result = {}
             for j, id in jobs.iteritems():
                 if id is None:
                     result[j] = 'UNKNOWN'
                 else:
                     if stat.has_key(id):
                         result[j] = stat[id]
                         if result[j] in ['DONE','EXIT','FORGOTTEN']:
                             stdout = os.path.join(j,'LSFJOB_%s' % id,'STDOUT')
                             if os.path.exists(stdout):
                                 #compress this file
                                 actions['FilesToCompress']['Files'].append(stdout)
                                 result[j] = '%s.gz' % stdout
                             elif os.path.exists('%s.gz' % stdout):
                                 result[j] = '%s.gz' % stdout
                             else:
                                 result[j] = 'NOSTDOUT'
                         
                             #also compress the stderr, although this is mostly empty
                             stderr = os.path.join(j,'LSFJOB_%s' % id,'STDERR')
                             if os.path.exists(stderr):
                                 #compress this file
                                 actions['FilesToCompress']['Files'].append(stderr)
                                 
             compress = GZipFiles(self.dataset,self.user,self.options)
             compress.run(actions)
             return result
         
         def countJobs(stat):
             """Count jobs that are monitorable - i.e. not in a final state"""
             result = []
             for j, id in jobs.iteritems():
                 if id is not None and stat.has_key(id):
                     st = stat[id]
                     if st in ['PEND','PSUSP','RUN','USUSP','SSUSP','WAIT']:
                         result.append(id)
             return result
         
         def writeKillScript(mon):
             """Write a shell script to kill the jobs we know about"""
             kill = os.path.join(jobsdir,'kill_jobs.sh')
             output = file(kill,'w')
             script = """
 #!/usr/bin/env bash
 echo "Killing jobs"
 bkill -u %s %s
             """ % (self.options.batch_user," ".join(mon))
             output.write(script)
             output.close()
             return mon
                     
         #continue monitoring while there are jobs to monitor
         status = self.monitor(jobs,{})
         monitorable = writeKillScript(countJobs(status))
         count = 0
         
         while monitorable:
             job_status = checkStatus(status)
             time.sleep(60)
             status = self.monitor(jobs,status)
             monitorable = writeKillScript(countJobs(status))
             if not (count % 3):
                 print '%s: Monitoring %i jobs (%s)' % (self.name,len(monitorable),self.dataset)
             count += 1
             
return {'LSFJobStatus':checkStatus(status),'LSFJobIDs':jobs}  