rhadd.haddStep Class Reference

Inheritance diagram for rhadd.haddStep:

Collaboration diagram for rhadd.haddStep:

Public Types
typedef HLT::TypeInformation::for_each_type_c< typenameEDMLIST::map, my_functor, my_result<>, my_arg< HLT::TypeInformation::get_cont, CONTAINER > >::type	result

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

Protected Member Functions
	_defineMergeJobs (self)

Protected Attributes
	_inputFiles = inputFiles
	_bunchNumber = bunchNumber
	_finalOutputFile = finalOutputFile
list	_haddJobArray = []
	_level = level

Detailed Description

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

Definition at line 84 of file rhadd.py.

Member Typedef Documentation

result

typedef HLT::TypeInformation::for_each_type_c<typenameEDMLIST::map,my_functor,my_result<>,my_arg<HLT::TypeInformation::get_cont,CONTAINER>>::type master_search< typenameEDMLIST::map, HLT::TypeInformation::get_cont, CONTAINER >::result

inherited

Definition at line 90 of file EDM_MasterSearch.h.

Constructor & Destructor Documentation

__init__()

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

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

rhadd.haddStep.__str__

(

self

)

Definition at line 158 of file rhadd.py.

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

_defineMergeJobs()

rhadd.haddStep._defineMergeJobs ( self )

protected

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

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

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

rhadd.haddStep.numHadds

(

self

)

Definition at line 155 of file rhadd.py.

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

outputFiles()

rhadd.haddStep.outputFiles

(

self

)

Definition at line 151 of file rhadd.py.

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

Member Data Documentation

_bunchNumber

rhadd.haddStep._bunchNumber = bunchNumber

protected

Definition at line 88 of file rhadd.py.

_finalOutputFile

rhadd.haddStep._finalOutputFile = finalOutputFile

protected

Definition at line 89 of file rhadd.py.

_haddJobArray

rhadd.haddStep._haddJobArray = []

protected

Definition at line 90 of file rhadd.py.

_inputFiles

rhadd.haddStep._inputFiles = inputFiles

protected

Definition at line 87 of file rhadd.py.

_level

rhadd.haddStep._level = level

protected

Definition at line 91 of file rhadd.py.

---

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

• rhadd.py