rhadd.haddStep Class Reference

Inheritance diagram for rhadd.haddStep:

Collaboration diagram for rhadd.haddStep:

Public Member Functions
def	__init__ (self, inputFiles, bunchNumber, finalOutputFile, level=None)
def	executeAll (self, parallel=1)
def	outputFiles (self)
def	numHadds (self)
def	__str__ (self)

Private Member Functions
def	_defineMergeJobs (self)

Private Attributes
	_inputFiles
	_bunchNumber
	_finalOutputFile
	_haddJobArray
	_level

Detailed Description

An hadd iteration - takes a bunch of input files and produces a bunch of output files

Definition at line 86 of file rhadd.py.

Constructor & Destructor Documentation

__init__()

def rhadd.haddStep.__init__	(		self,
			inputFiles,
			bunchNumber,
			finalOutputFile,
			level = None
	)

Definition at line 88 of file rhadd.py.

    def __init__(self, inputFiles, bunchNumber, finalOutputFile, level = None):
        self._inputFiles = inputFiles
        self._bunchNumber = bunchNumber
        self._finalOutputFile = finalOutputFile
        self._haddJobArray = []
        self._level = level
        self._defineMergeJobs()
    
    

Member Function Documentation

__str__()

def rhadd.haddStep.__str__

(

self

)

Definition at line 160 of file rhadd.py.

    def __str__(self):
        return 'Merging level %s: %s' % (self._level, str([ str(job) for job in self._haddJobArray ]))
 
 

_defineMergeJobs()

def rhadd.haddStep._defineMergeJobs (   self )

private

Definition at line 97 of file rhadd.py.

    def _defineMergeJobs(self):
        # How many merges to do in this step?
        nMerges = (len(self._inputFiles)-1) // self._bunchNumber + 1
        logging.debug('Need %d merges for level %d' % (nMerges, self._level))
        if nMerges == 1:
            logging.debug('Final merge job: %s -> %s' % (self._inputFiles, self._inputFiles))
            self._haddJobArray.append(haddJob(self._inputFiles, self._finalOutputFile))
            return
        
        # With >1 merge need temporary files
        nextFile = 0
        for job in range(nMerges):
            # Try to ensure we have ~equal numbers of files in each merge
            fileCounter = len(self._inputFiles) * float(job+1) / nMerges
            # Add 0.5 to ensure that rounding errors don't lose a file off the back... (very unlikely!)
            lastFile = int(fileCounter + 0.5)
            tempOutput = mkstemp(dir='.', prefix='tmp.')
            os.close(tempOutput[0])
            logging.debug('Intermediate merge job %d: %s -> %s' % (job, self._inputFiles[nextFile:lastFile], tempOutput[1]))
            self._haddJobArray.append(haddJob(self._inputFiles[nextFile:lastFile], tempOutput[1]))
            nextFile = lastFile
        
        

executeAll()

def rhadd.haddStep.executeAll	(		self,
			parallel = 1
	)

Definition at line 120 of file rhadd.py.

    def executeAll(self, parallel = 1):
        if parallel > 1:
            # Funky parallel processing
            logging.info('Starting merge using up to %d hadd processes in parallel' % parallel)
            logging.warning('Parallel merging is experimental')
            pool = Pool(processes = parallel)
            parallelResultsArray = []
            for job in self._haddJobArray:
                parallelResultsArray.append(pool.apply_async(job, ()))
            pool.close()
            # The next two lines will stick until all the worker processes are finished
            # Really one needs a progress loop monitor with a timeout... 
            pool.join()
            
            # Update our hadd exit codes to the parallel processed return code, because the copy of the 
            # instance held by the worker was the one where the exe method was actually called
            for i, job in enumerate(self._haddJobArray):
                job.exitCode = parallelResultsArray[i].get(timeout=0)
            
            for job in self._haddJobArray:
                if job.exitCode != 0:
                    logging.error('Merging job %s failed, exit code %s' % (job, job.exitCode))
                    sys.exit(1)
        else:
            # Safe and slow serial processing
            for job in self._haddJobArray:
                job.exe()
                if job.exitCode != 0:
                    logging.error('Merging job %s failed, exit code %s' % (job, job.exitCode))
                    sys.exit(1)
 
 

numHadds()

def rhadd.haddStep.numHadds

(

self

)

Definition at line 157 of file rhadd.py.

    def numHadds(self):
        return len(self._haddJobArray)
    

outputFiles()

def rhadd.haddStep.outputFiles

(

self

)

Definition at line 153 of file rhadd.py.

    def outputFiles(self):
        return [ job.outputFile for job in self._haddJobArray ]
    

Member Data Documentation

_bunchNumber

rhadd.haddStep._bunchNumber

private

Definition at line 90 of file rhadd.py.

_finalOutputFile

rhadd.haddStep._finalOutputFile

private

Definition at line 91 of file rhadd.py.

_haddJobArray

rhadd.haddStep._haddJobArray

private

Definition at line 92 of file rhadd.py.

_inputFiles

rhadd.haddStep._inputFiles

private

Definition at line 89 of file rhadd.py.

_level

rhadd.haddStep._level

private

Definition at line 93 of file rhadd.py.

---

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

• rhadd.py