python.trfExe.athenaExecutor Class Reference

Inheritance diagram for python.trfExe.athenaExecutor:

Collaboration diagram for python.trfExe.athenaExecutor:

Public Member Functions
	__init__ (self, name='athena', trf=None, conf=None, skeletonFile=None, skeletonCA=None, inData=set(), outData=set(), inputDataTypeCountCheck=None, exe='athena.py', exeArgs=['athenaopts'], substep=None, inputEventTest=True, perfMonFile=None, tryDropAndReload=True, extraRunargs={}, runtimeRunargs={}, literalRunargs=[], dataArgs=[], checkEventCount=False, errorMaskFiles=None, manualDataDictionary=None, memMonitor=True, disableMT=False, disableMP=False, onlyMP=False, onlyMT=False, onlyMPWithRunargs=None)
	Initialise athena executor.
	inputDataTypeCountCheck (self)
	inputDataTypeCountCheck (self, value)
	substep (self)
	disableMP (self)
	disableMP (self, value)
	disableMT (self)
	disableMT (self, value)
	onlyMP (self)
	onlyMP (self, value)
	onlyMT (self)
	onlyMT (self, value)
	skeletonCA (self)
	preExecute (self, input=set(), output=set())
	postExecute (self)
	validate (self)
	exeArgs (self)
	exeArgs (self, value)
	exe (self)
	exe (self, value)
	execute (self)
	inData (self)
	inData (self, value)
	outData (self)
	outData (self, value)
	myMerger (self)
	Now define properties for these data members.
	name (self)
	name (self, value)
	trf (self)
	trf (self, value)
	inDataUpdate (self, value)
	outDataUpdate (self, value)
	input (self)
	output (self)
	extraMetadata (self)
	hasExecuted (self)
	rc (self)
	errMsg (self)
	validation (self)
	validation (self, value)
	hasValidated (self)
	isValidated (self)
	first (self)
	preExeStartTimes (self)
	exeStartTimes (self)
	exeStopTimes (self)
	valStartTimes (self)
	valStopTimes (self)
	preExeCpuTime (self)
	preExeWallTime (self)
	cpuTime (self)
	usrTime (self)
	sysTime (self)
	wallTime (self)
	memStats (self)
	memAnalysis (self)
	postExeCpuTime (self)
	postExeWallTime (self)
	validationCpuTime (self)
	validationWallTime (self)
	cpuTimeTotal (self)
	wallTimeTotal (self)
	eventCount (self)
	reSimEvent (self)
	athenaMP (self)
	dbMonitor (self)
	setPreExeStart (self)
	setValStart (self)
	doAll (self, input=set(), output=set())
	Convenience function.

Public Attributes
	exeArgs
	conf = conf
	Executor configuration:
	inData = value
	outData = value

Protected Member Functions
	_isCAEnabled (self)
	Check if running with CA.
	_prepAthenaCommandLine (self)
	Prepare the correct command line to be used to invoke athena.
	_writeAthenaWrapper (self, asetup=None, dbsetup=None, ossetup=None)
	Write a wrapper script which runs asetup and then Athena.
	_smartMerge (self, fileArg)
	Manage smart merging of output files.
	_targzipJiveXML (self)
	_buildStandardCommand (self)

Protected Attributes
	_substep = forceToAlphaNum(substep)
	_inputEventTest = inputEventTest
	_tryDropAndReload = tryDropAndReload
	Add –drop-and-reload if possible (and allowed!).
	_extraRunargs = extraRunargs
	_runtimeRunargs = runtimeRunargs
	_literalRunargs = literalRunargs
	_dataArgs = dataArgs
	_errorMaskFiles = errorMaskFiles
	_inputDataTypeCountCheck = inputDataTypeCountCheck
	_disableMT = disableMT
bool	_disableMP = disableMP
bool	_onlyMP = onlyMP
	_onlyMT = onlyMT
	_onlyMPWithRunargs = onlyMPWithRunargs
	_skeletonCA = skeletonCA
	Handle MPI setup.
	_perfMonFile = None
list	_skeleton = [skeletonFile]
	_jobOptionsTemplate = JobOptionsTemplate(exe = self, version = '$Id: trfExe.py 792052 2017-01-13 13:36:51Z mavogel $')
str	_athenaMPWorkerTopDir = 'athenaMP-workers-{0}-{1}'.format(self._name, self._substep)
str	_athenaMPFileReport = 'athenaMP-outputs-{0}-{1}'.format(self._name, self._substep)
str	_athenaMPEventOrdersFile = 'athenamp_eventorders.txt.{0}'.format(self._name)
bool	_athenaMPReadEventOrders = True
str	_athenaMPStrategy = self.conf.argdict['athenaMPStrategy'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
	_topOptionsFiles
	_envUpdate = trfEnv.environmentUpdate()
	Look for environment updates and perpare the athena command line.
	_logScan
list	_originalCmd = self._cmd
	_asetup = asetup
	_dbsetup = dbsetup
str	_wrapperFile = 'runwrapper.{name}.sh'.format(name = self._name)
str	_setupFile = 'setup.{name}.sh'.format(name = self._name)
	_exe = exe
	_exeArgs = exeArgs
bool	_echoOutput = False
list	_cmd = None
bool	_memMonitor = memMonitor
	_input = input
	_output = output
str	_logFileName = "log.{0}".format(self._name)
	_echologger = logging.getLogger(self._name)
	_exeLogFile = logging.FileHandler(self._logFileName, mode='w', **encargs)
	_echostream = logging.StreamHandler(_encoding_stream(sys.stdout))
	_workdir
str	_memSummaryFile = 'prmon.summary.' + self._name + '.json'
	_name = forceToAlphaNum(name)
	_inData = set(inData)
	_outData = set(outData)
bool	_hasExecuted = False
int	_rc = -1
str	_errMsg = None
bool	_hasValidated = False
bool	_isValidated = False
dict	_extraMetadata = {}
	_preExeStart = None
	_exeStart = None
	_valStart = None
dict	_memStats = {}
dict	_memLeakResult = {}
	_memFullFile = None
	_eventCount = None
int	_athenaMP = 0
int	_athenaMT = 0
int	_athenaConcurrentEvents = 0
	_dbMonitor = None
	_resimevents = None
	_alreadyInContainer = None
	_containerSetup = None
list	_myMerger = []
	_trf = value
	_validation = value
	_exeStop
	_valStop

Static Protected Attributes
int	_exitMessageLimit = 200
list	_defaultIgnorePatternFile = ['atlas_error_mask.db']

Detailed Description

Definition at line 844 of file trfExe.py.

Constructor & Destructor Documentation

__init__()

python.trfExe.athenaExecutor.__init__	(		self,
			name = 'athena',
			trf = None,
			conf = None,
			skeletonFile = None,
			skeletonCA = None,
			inData = set(),
			outData = set(),
			inputDataTypeCountCheck = None,
			exe = 'athena.py',
			exeArgs = ['athenaopts'],
			substep = None,
			inputEventTest = True,
			perfMonFile = None,
			tryDropAndReload = True,
			extraRunargs = {},
			runtimeRunargs = {},
			literalRunargs = [],
			dataArgs = [],
			checkEventCount = False,
			errorMaskFiles = None,
			manualDataDictionary = None,
			memMonitor = True,
			disableMT = False,
			disableMP = False,
			onlyMP = False,
			onlyMT = False,
			onlyMPWithRunargs = None )

Initialise athena executor.

Parameters

name	Executor name
trf	Parent transform
skeletonFile	athena skeleton job options file (optionally this can be a list of skeletons that will be given to athena.py in order); can be set to None to disable writing job options files at all
skeletonCA	ComponentAccumulator-compliant skeleton file (used with the –CA option)
inputDataTypeCountCheck	List of input datatypes to apply preExecute event count checks to; default is None, which means check all inputs
exe	Athena execution script
exeArgs	Transform argument names whose value is passed to athena
substep	The athena substep this executor represents (alias for the name)
inputEventTest	Boolean switching the skipEvents < inputEvents test
perfMonFile	Name of perfmon file for this substep (used to retrieve vmem/rss information) DEPRECATED
tryDropAndReload	Boolean switch for the attempt to add '–drop-and-reload' to athena args
extraRunargs	Dictionary of extra runargs to write into the job options file, using repr
runtimeRunargs	Dictionary of extra runargs to write into the job options file, using str
literalRunargs	List of extra lines to write into the runargs file
dataArgs	List of datatypes that will always be given as part of this transform's runargs even if not actually processed by this substep (used, e.g., to set random seeds for some generators)
checkEventCount	Compare the correct number of events in the output file (either input file size or maxEvents)
errorMaskFiles	List of files to use for error masks in logfile scanning (None means not set for this executor, so use the transform or the standard setting)
manualDataDictionary	Instead of using the inData/outData parameters that binds the data types for this executor to the workflow graph, run the executor manually with these data parameters (useful for post-facto executors, e.g., for AthenaMP merging)
memMonitor	Enable subprocess memory monitoring
disableMT	Ensure that AthenaMT is not used
disableMP	Ensure that AthenaMP is not used
onlyMP	Ensure that MP is always used, even if MT is requested
onlyMT	Ensure that MT is used even if MP is requested
onlyMPWithRunargs	Ensure that MP is always used, even if MT is requested when one of the listed runargs is provided

Note

The difference between extraRunargs, runtimeRunargs and literalRunargs is that: extraRunargs uses repr(), so the RHS is the same as the python object in the transform; runtimeRunargs uses str() so that a string can be interpreted at runtime; literalRunargs allows the direct insertion of arbitary python snippets into the runArgs file.

Definition at line 885 of file trfExe.py.

                 manualDataDictionary = None, memMonitor = True, disableMT = False, disableMP = False, onlyMP = False, onlyMT = False, onlyMPWithRunargs = None):
        
        self._substep = forceToAlphaNum(substep)
        self._inputEventTest = inputEventTest
        self._tryDropAndReload = tryDropAndReload
        self._extraRunargs = extraRunargs
        self._runtimeRunargs = runtimeRunargs
        self._literalRunargs = literalRunargs
        self._dataArgs = dataArgs
        self._errorMaskFiles = errorMaskFiles
        self._inputDataTypeCountCheck = inputDataTypeCountCheck
        self._disableMT = disableMT
        self._disableMP = disableMP
        self._onlyMP = onlyMP
        self._onlyMT = onlyMT
        self._onlyMPWithRunargs = onlyMPWithRunargs
        self._skeletonCA=skeletonCA
 
        if perfMonFile:
            self._perfMonFile = None
            msg.debug("Resource monitoring from PerfMon is now deprecated")
        
        # SkeletonFile can be None (disable) or a string or a list of strings - normalise it here
        if isinstance(skeletonFile, str):
            self._skeleton = [skeletonFile]
        else:
            self._skeleton = skeletonFile
            
        super(athenaExecutor, self).__init__(name=name, trf=trf, conf=conf, inData=inData, outData=outData, exe=exe, 
                                             exeArgs=exeArgs, memMonitor=memMonitor)
        
        # Add athena specific metadata
        self._extraMetadata.update({'substep': substep})
 
        # Setup JO templates
        if self._skeleton or self._skeletonCA:
            self._jobOptionsTemplate = JobOptionsTemplate(exe = self, version = '$Id: trfExe.py 792052 2017-01-13 13:36:51Z mavogel $')
        else:
            self._jobOptionsTemplate = None
 

Member Function Documentation

_buildStandardCommand()

python.trfExe.scriptExecutor._buildStandardCommand ( self )

protectedinherited

Definition at line 706 of file trfExe.py.

    def _buildStandardCommand(self):
        if self._exe:
            self._cmd = [self.exe, ]
        else:
            raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_EXEC_SETUP_FAIL'), 
                                                            'No executor set in {0}'.format(self.__class__.__name__))
        for arg in self.exeArgs:
            if arg in self.conf.argdict:
                # If we have a list then add each element to our list, else just str() the argument value
                # Note if there are arguments which need more complex transformations then
                # consider introducing a special toExeArg() method.
                if isinstance(self.conf.argdict[arg].value, list):
                    self._cmd.extend([ str(v) for v in self.conf.argdict[arg].value])
                else:
                    self._cmd.append(str(self.conf.argdict[arg].value))
 
 

_isCAEnabled()

python.trfExe.athenaExecutor._isCAEnabled ( self )

protected

Check if running with CA.

Definition at line 1437 of file trfExe.py.

    def _isCAEnabled(self):
        # CA not present
        if 'CA' not in self.conf.argdict:
            # If there is no legacy skeleton, then we are running with CA
            if not self._skeleton:
                return True
            else:
                return False
 
        # CA present but None, all substeps running with CA
        if self.conf.argdict['CA'] is None:
            return True
 
        # CA enabled for a substep, running with CA
        if self.conf.argdict['CA'].returnMyValue(name=self.name, substep=self.substep) is True:
            return True
 
        return False
 

_prepAthenaCommandLine()

python.trfExe.athenaExecutor._prepAthenaCommandLine ( self )

protected

Prepare the correct command line to be used to invoke athena.

Definition at line 1457 of file trfExe.py.

    def _prepAthenaCommandLine(self):
        
        if 'athena' in self.conf.argdict:
            self._exe = self.conf.argdict['athena'].value
        self._cmd = [self._exe]
        
        # Find options for the current substep. Name is prioritised (e.g. RAWtoALL) over alias (e.g. r2a). Last look for 'all'
        currentSubstep = None
        if 'athenaopts' in self.conf.argdict:
            currentName = commonExecutorStepName(self.name)
            if currentName in self.conf.argdict['athenaopts'].value:
                currentSubstep = currentName
                if self.substep in self.conf.argdict['athenaopts'].value:
                    msg.info('Athenaopts found for {0} and {1}, joining options. '
                             'Consider changing your configuration to use just the name or the alias of the substep.'
                             .format(currentSubstep, self.substep))
                    self.conf.argdict['athenaopts'].value[currentSubstep].extend(self.conf.argdict['athenaopts'].value[self.substep])
                    del self.conf.argdict['athenaopts'].value[self.substep]
                    msg.debug('Athenaopts: {0}'.format(self.conf.argdict['athenaopts'].value))
            elif self.substep in self.conf.argdict['athenaopts'].value:
                currentSubstep = self.substep
            elif 'all' in self.conf.argdict['athenaopts'].value:
                currentSubstep = 'all'
 
        # See if there's a preloadlibs and a request to update LD_PRELOAD for athena
        preLoadUpdated = dict()
        if 'LD_PRELOAD' in self._envUpdate._envdict:
            preLoadUpdated[currentSubstep] = False
            if 'athenaopts' in self.conf.argdict:
                if currentSubstep is not None:
                    for athArg in self.conf.argdict['athenaopts'].value[currentSubstep]:
                        # This code is pretty ugly as the athenaopts argument contains
                        # strings which are really key/value pairs
                        if athArg.startswith('--preloadlib'):
                            try:
                                i = self.conf.argdict['athenaopts'].value[currentSubstep].index(athArg)
                                v = athArg.split('=', 1)[1]
                                msg.info('Updating athena --preloadlib option for substep {1} with: {0}'.format(self._envUpdate.value('LD_PRELOAD'), self.name))
                                newPreloads = ":".join(set(v.split(":")) | set(self._envUpdate.value('LD_PRELOAD').split(":")))
                                self.conf.argdict['athenaopts']._value[currentSubstep][i] = '--preloadlib={0}'.format(newPreloads)
                            except Exception as e:
                                msg.warning('Failed to interpret athena option: {0} ({1})'.format(athArg, e))
                            preLoadUpdated[currentSubstep] = True
                        break
            if not preLoadUpdated[currentSubstep]:
                msg.info('Setting athena preloadlibs for substep {1} to: {0}'.format(self._envUpdate.value('LD_PRELOAD'), self.name))
                if 'athenaopts' in self.conf.argdict:
                    if currentSubstep is not None:
                        self.conf.argdict['athenaopts'].value[currentSubstep].append("--preloadlib={0}".format(self._envUpdate.value('LD_PRELOAD')))
                    else:
                        self.conf.argdict['athenaopts'].value['all'] = ["--preloadlib={0}".format(self._envUpdate.value('LD_PRELOAD'))]
                else:
                    self.conf.argdict['athenaopts'] = trfArgClasses.argSubstepList(["--preloadlib={0}".format(self._envUpdate.value('LD_PRELOAD'))])
 
        # Now update command line with the options we have (including any changes to preload)
        if 'athenaopts' in self.conf.argdict:
            if currentSubstep is None and "all" in self.conf.argdict['athenaopts'].value:
                self._cmd.extend(self.conf.argdict['athenaopts'].value['all'])
            elif currentSubstep in self.conf.argdict['athenaopts'].value:
                self._cmd.extend(self.conf.argdict['athenaopts'].value[currentSubstep])
        
        if currentSubstep is None:
            currentSubstep = 'all'
        
        if self._tryDropAndReload:
            if self._isCAEnabled():
                msg.info('ignoring "--drop-and-reload" for CA-based transforms, config cleaned up anyway')
            elif 'valgrind' in self.conf._argdict and self.conf._argdict['valgrind'].value is True:
                msg.info('Disabling "--drop-and-reload" because the job is configured to use Valgrind')
            elif 'athenaopts' in self.conf.argdict:
                athenaConfigRelatedOpts = ['--config-only','--drop-and-reload']
                # Note for athena options we split on '=' so that we properly get the option and not the whole "--option=value" string
                if currentSubstep in self.conf.argdict['athenaopts'].value:
                    conflictOpts = set(athenaConfigRelatedOpts).intersection(set([opt.split('=')[0] for opt in self.conf.argdict['athenaopts'].value[currentSubstep]]))
                    if len(conflictOpts) > 0:
                        msg.info('Not appending "--drop-and-reload" to athena command line because these options conflict: {0}'.format(list(conflictOpts)))
                    else:
                        msg.info('Appending "--drop-and-reload" to athena options')
                        self._cmd.append('--drop-and-reload')
                else:
                    msg.info('No Athenaopts for substep {0}, appending "--drop-and-reload" to athena options'.format(self.name))
                    self._cmd.append('--drop-and-reload')
            else:
                # This is the 'standard' case - so drop and reload should be ok
                msg.info('Appending "--drop-and-reload" to athena options')
                self._cmd.append('--drop-and-reload')
        else:
            msg.info('Skipping test for "--drop-and-reload" in this executor')
            
        if not self._isCAEnabled():  #For CA-jobs, threads and nproc set in runargs file
            # For AthenaMT apply --threads=N if threads have been configured via ATHENA_CORE_NUMBER + multithreaded
            if self._athenaMT > 0 and not self._disableMT:
                if not ('athenaopts' in self.conf.argdict and
                        any('--threads' in opt for opt in self.conf.argdict['athenaopts'].value[currentSubstep])):
                        self._cmd.append('--threads=%s' % str(self._athenaMT))
 
            # For AthenaMP apply --nprocs=N if threads have been configured via ATHENA_CORE_NUMBER + multiprocess
            if self._athenaMP > 0 and not self._disableMP:
                if not ('athenaopts' in self.conf.argdict and
                        any('--nprocs' in opt for opt in self.conf.argdict['athenaopts'].value[currentSubstep])):
                        self._cmd.append('--nprocs=%s' % str(self._athenaMP))
 
        # Add topoptions
        # Note that _writeAthenaWrapper removes this from the end of _cmd when preparing the options for VTuneCommand, so assumes it comes last.
        if self._skeleton or self._skeletonCA:
            self._cmd += self._topOptionsFiles
            msg.info('Updated script arguments with topoptions: %s', self._cmd)
 
 

_smartMerge()

python.trfExe.athenaExecutor._smartMerge ( self, fileArg )

protected

Manage smart merging of output files.

Parameters

fileArg

File argument to merge

Definition at line 1751 of file trfExe.py.

    def _smartMerge(self, fileArg):
        
        if 'selfMerge' not in dir(fileArg):
            msg.info('Files in {0} cannot merged (no selfMerge() method is implemented)'.format(fileArg.name))
            return
        
        if fileArg.mergeTargetSize == 0:
            msg.info('Files in {0} will not be merged as target size is set to 0'.format(fileArg.name))
            return
 
        
        mergeCandidates = [list()]
        currentMergeSize = 0
        for fname in fileArg.value:
            size = fileArg.getSingleMetadata(fname, 'file_size')
            if not isinstance(size, int):
                msg.warning('File size metadata for {0} was not correct, found type {1}. Aborting merge attempts.'.format(fileArg, type(size)))
                return
            # if there is no file in the job, then we must add it
            if len(mergeCandidates[-1]) == 0:
                msg.debug('Adding file {0} to current empty merge list'.format(fname))
                mergeCandidates[-1].append(fname)
                currentMergeSize += size
                continue
            # see if adding this file gets us closer to the target size (but always add if target size is negative)
            if fileArg.mergeTargetSize < 0 or math.fabs(currentMergeSize + size - fileArg.mergeTargetSize) < math.fabs(currentMergeSize - fileArg.mergeTargetSize):
                msg.debug('Adding file {0} to merge list {1} as it gets closer to the target size'.format(fname, mergeCandidates[-1]))
                mergeCandidates[-1].append(fname)
                currentMergeSize += size
                continue
            # close this merge list and start a new one
            msg.debug('Starting a new merge list with file {0}'.format(fname))
            mergeCandidates.append([fname])
            currentMergeSize = size
            
        msg.debug('First pass splitting will merge files in this way: {0}'.format(mergeCandidates))
        
        if len(mergeCandidates) == 1:
            # Merging to a single file, so use the original filename that the transform
            # was started with
            mergeNames = [fileArg.originalName]
        else:
            # Multiple merge targets, so we need a set of unique names
            counter = 0
            mergeNames = []
            for mergeGroup in mergeCandidates:
                # Note that the individual worker files get numbered with 3 digit padding,
                # so these non-padded merges should be fine
                mergeName = fileArg.originalName + '_{0}'.format(counter)
                while path.exists(mergeName):
                    counter += 1
                    mergeName = fileArg.originalName + '_{0}'.format(counter)
                mergeNames.append(mergeName)
                counter += 1
        # Now actually do the merges
        for targetName, mergeGroup, counter in zip(mergeNames, mergeCandidates, list(range(len(mergeNames)))):
            msg.info('Want to merge files {0} to {1}'.format(mergeGroup, targetName))
            if len(mergeGroup) <= 1:
                msg.info('Skip merging for single file')
            else:
                
                self._myMerger.append(fileArg.selfMerge(output=targetName, inputs=mergeGroup, counter=counter, argdict=self.conf.argdict))
 
 

_targzipJiveXML()

python.trfExe.athenaExecutor._targzipJiveXML ( self )

protected

Definition at line 1815 of file trfExe.py.

    def _targzipJiveXML(self):
        #tgzipping JiveXML files
        targetTGZName = self.conf.dataDictionary['TXT_JIVEXMLTGZ'].value[0]
        if os.path.exists(targetTGZName):
            os.remove(targetTGZName)
 
        import tarfile
        fNameRE = re.compile(r"JiveXML\_\d+\_\d+.xml")
 
        # force gz compression
        tar = tarfile.open(targetTGZName, "w:gz")
        for fName in os.listdir('.'):
            matches = fNameRE.findall(fName)
            if len(matches) > 0:
                if fNameRE.findall(fName)[0] == fName:
                    msg.info('adding %s to %s', fName, targetTGZName)
                    tar.add(fName)
 
        tar.close()
        msg.info('JiveXML compression: %s has been written and closed.', targetTGZName)
 
 

_writeAthenaWrapper()

python.trfExe.athenaExecutor._writeAthenaWrapper ( self, asetup = None, dbsetup = None, ossetup = None )

protected

Write a wrapper script which runs asetup and then Athena.

Definition at line 1569 of file trfExe.py.

        ):
        self._originalCmd        = self._cmd
        self._asetup             = asetup
        self._dbsetup            = dbsetup
        self._containerSetup     = ossetup
        self._workdir            = os.getcwd()
        self._alreadyInContainer = self._workdir.startswith("/srv")
        self._wrapperFile        = 'runwrapper.{name}.sh'.format(name = self._name)
        self._setupFile          = 'setup.{name}.sh'.format(name = self._name)
 
        # Create a setupATLAS script
        setupATLAS = 'my_setupATLAS.sh'
        with open(setupATLAS, 'w') as f:
            print("#!/bin/bash", file=f)
            print("""
if [ -z $ATLAS_LOCAL_ROOT_BASE ]; then
  export ATLAS_LOCAL_ROOT_BASE=/cvmfs/atlas.cern.ch/repo/ATLASLocalRootBase
fi
source ${ATLAS_LOCAL_ROOT_BASE}/user/atlasLocalSetup.sh"""
                  , file=f)
        os.chmod(setupATLAS, 0o755)
 
        msg.debug(
            'Preparing wrapper file {wrapperFileName} with '
            'asetup={asetupStatus} and dbsetup={dbsetupStatus}'.format(
                wrapperFileName = self._wrapperFile,
                asetupStatus    = self._asetup,
                dbsetupStatus   = self._dbsetup
            )
        )
 
        container_cmd = None
        try:
            with open(self._wrapperFile, 'w') as wrapper:
                print('#!/bin/sh', file=wrapper)
                if self._containerSetup is not None:
                    container_cmd = [ os.path.abspath(setupATLAS),
                                     "-c",
                                     self._containerSetup,
                                     "--pwd",
                                     self._workdir,
                                     "-s",
                                     os.path.join('.', self._setupFile),
                                     "-r"]
                    print('echo "This wrapper is executed within a container! For a local re-run, do:"', file=wrapper)
                    print('echo " '+ " ".join(['setupATLAS'] + container_cmd[1:] + [path.join('.', self._wrapperFile)]) + '"', file=wrapper)
                    print('echo "N.B.: if launching a nested container, navigate to /srv before running the above command"',
                          file = wrapper)
                    print('echo "      and use --pwd workdir, where workdir is the transform running directory within /srv"',
                          file=wrapper)
                    print('echo', file=wrapper)
 
                if asetup:
                    wfile = wrapper
                    asetupFile = None
                    # if the substep is executed within a container, setup athena with a separate script
                    # e.g. setupATLAS -c el9 -s setup.RDOtoRDOTrigger.sh -r runwrapper.RDOtoRDOTrigger.sh ...
                    if self._containerSetup is not None:
                        asetupFile = open(self._setupFile, 'w')
                        wfile = asetupFile
                    print(f'source ./{setupATLAS} -q', file=wfile)
                    print(f'asetup {asetup}', file=wfile)
                    print('if [ ${?} != "0" ]; then exit 255; fi', file=wfile)
                if dbsetup:
                    dbroot = path.dirname(dbsetup)
                    dbversion = path.basename(dbroot)
                    print("# DBRelease setup", file=wrapper)
                    print('echo Setting up DBRelease {dbroot} environment'.format(dbroot = dbroot), file=wrapper)
                    print('export DBRELEASE={dbversion}'.format(dbversion = dbversion), file=wrapper)
                    print('export CORAL_AUTH_PATH={directory}'.format(directory = path.join(dbroot, 'XMLConfig')), file=wrapper)
                    print('export CORAL_DBLOOKUP_PATH={directory}'.format(directory = path.join(dbroot, 'XMLConfig')), file=wrapper)
                    print('export TNS_ADMIN={directory}'.format(directory = path.join(dbroot, 'oracle-admin')), file=wrapper)
                    print('DATAPATH={dbroot}:$DATAPATH'.format(dbroot = dbroot), file=wrapper)
                if self._disableMT:
                    print("# AthenaMT explicitly disabled for this executor", file=wrapper)
                if self._disableMP:
                    print("# AthenaMP explicitly disabled for this executor", file=wrapper)
                if self._envUpdate.len > 0:
                    for envSetting in  self._envUpdate.values:
                        if not envSetting.startswith('LD_PRELOAD'):
                            print("export", envSetting, file=wrapper)
                # If Valgrind is engaged, a serialised Athena configuration file
                # is generated for use with a subsequent run of Athena with
                # Valgrind.
                if 'valgrind' in self.conf._argdict and self.conf._argdict['valgrind'].value is True:
                    msg.info('Valgrind engaged')
                    # Define the file name of the serialised Athena
                    # configuration.
                    AthenaSerialisedConfigurationFile = "{name}Conf.pkl".format(
                        name = self._name
                    )
                    # Run Athena for generation of its serialised configuration.
                    print(' '.join(self._cmd), "--config-only={0}".format(AthenaSerialisedConfigurationFile), file=wrapper)
                    print('if [ $? != "0" ]; then exit 255; fi', file=wrapper)
                    # Generate a Valgrind command, suppressing or ussing default
                    # options as requested and extra options as requested.
                    if 'valgrindDefaultOpts' in self.conf._argdict:
                        defaultOptions = self.conf._argdict['valgrindDefaultOpts'].value
                    else:
                        defaultOptions = True
                    if 'valgrindExtraOpts' in self.conf._argdict:
                        extraOptionsList = self.conf._argdict['valgrindExtraOpts'].value
                    else:
                        extraOptionsList = None
                    msg.debug("requested Valgrind command basic options: {options}".format(options = defaultOptions))
                    msg.debug("requested Valgrind command extra options: {options}".format(options = extraOptionsList))
                    command = ValgrindCommand(
                        defaultOptions = defaultOptions,
                        extraOptionsList = extraOptionsList,
                        AthenaSerialisedConfigurationFile = \
                            AthenaSerialisedConfigurationFile
                    )
                    msg.debug("Valgrind command: {command}".format(command = command))
                    print(command, file=wrapper)
                # If VTune is engaged, a serialised Athena configuration file
                # is generated for use with a subsequent run of Athena with
                # VTune.
                elif 'vtune' in self.conf._argdict and self.conf._argdict['vtune'].value is True:
                    msg.info('VTune engaged')
                    # Define the file name of the serialised Athena
                    # configuration.
                    AthenaSerialisedConfigurationFile = "{name}Conf.pkl".format(
                        name = self._name
                    )
                    # Run Athena for generation of its serialised configuration.
                    print(' '.join(self._cmd), "--config-only={0}".format(AthenaSerialisedConfigurationFile), file=wrapper)
                    print('if [ $? != "0" ]; then exit 255; fi', file=wrapper)
                    # Generate a VTune command, suppressing or ussing default
                    # options as requested and extra options as requested.
                    if 'vtuneDefaultOpts' in self.conf._argdict:
                        defaultOptions = self.conf._argdict['vtuneDefaultOpts'].value
                    else:
                        defaultOptions = True
                    if 'vtuneExtraOpts' in self.conf._argdict:
                        extraOptionsList = self.conf._argdict['vtuneExtraOpts'].value
                    else:
                        extraOptionsList = None
 
                    # replace the _topOptionsFiles from the Athena command with the AthenaSerialisedConfigurationFile.
                    if (self._skeleton or self._skeletonCA) and len(self._topOptionsFiles) > 0:
                        AthenaCommand = self._cmd[:-len(self._topOptionsFiles)]
                    else:
                        AthenaCommand = self._cmd
                    AthenaCommand.append(AthenaSerialisedConfigurationFile)
 
                    msg.debug("requested VTune command basic options: {options}".format(options = defaultOptions))
                    msg.debug("requested VTune command extra options: {options}".format(options = extraOptionsList))
                    command = VTuneCommand(
                        defaultOptions = defaultOptions,
                        extraOptionsList = extraOptionsList,
                        AthenaCommand = AthenaCommand
                    )
                    msg.debug("VTune command: {command}".format(command = command))
                    print(command, file=wrapper)
                else:
                    msg.info('Valgrind/VTune not engaged')
                    # run Athena command
                    print(' '.join(self._cmd), file=wrapper)
            os.chmod(self._wrapperFile, 0o755)
        except OSError as e:
            errMsg = 'error writing athena wrapper {fileName}: {error}'.format(
                fileName = self._wrapperFile,
                error = e
            )
            msg.error(errMsg)
            raise trfExceptions.TransformExecutionException(
                trfExit.nameToCode('TRF_EXEC_SETUP_WRAPPER'),
                errMsg
            )
        self._cmd = [ path.join('.', self._wrapperFile) ]
        if self._containerSetup is not None:
            asetupFile.close()
            self._cmd = container_cmd + self._cmd
 
 

athenaMP()

python.trfExe.transformExecutor.athenaMP ( self )

inherited

Definition at line 452 of file trfExe.py.

    def athenaMP(self):
        return self._athenaMP
 

cpuTime()

python.trfExe.transformExecutor.cpuTime ( self )

inherited

Definition at line 366 of file trfExe.py.

    def cpuTime(self):
        if self._exeStart and self._exeStop:
            return calcCpuTime(self._exeStart, self._exeStop)
        else:
            return None
 

cpuTimeTotal()

python.trfExe.transformExecutor.cpuTimeTotal ( self )

inherited

Definition at line 430 of file trfExe.py.

    def cpuTimeTotal(self):
        if self._preExeStart and self._valStop:
            return calcCpuTime(self._preExeStart, self._valStop)
        else:
            return None
 

dbMonitor()

python.trfExe.transformExecutor.dbMonitor ( self )

inherited

Definition at line 456 of file trfExe.py.

    def dbMonitor(self):
        return self._dbMonitor
 
 

disableMP() [1/2]

python.trfExe.athenaExecutor.disableMP

(

self

)

Definition at line 942 of file trfExe.py.

    def disableMP(self):
        return self._disableMP
    

disableMP() [2/2]

python.trfExe.athenaExecutor.disableMP	(		self,
			value )

Definition at line 946 of file trfExe.py.

    def disableMP(self, value):
        self._disableMP = value
    

disableMT() [1/2]

python.trfExe.athenaExecutor.disableMT

(

self

)

Definition at line 950 of file trfExe.py.

    def disableMT(self):
        return self._disableMT
    

disableMT() [2/2]

python.trfExe.athenaExecutor.disableMT	(		self,
			value )

Definition at line 954 of file trfExe.py.

    def disableMT(self, value):
        self._disableMT = value
 

doAll()

python.trfExe.transformExecutor.doAll ( self, input = set(), output = set() )

inherited

Convenience function.

Definition at line 499 of file trfExe.py.

    def doAll(self, input=set(), output=set()):
        self.preExecute(input, output)
        self.execute()
        self.postExecute()
        self.validate()
 

errMsg()

python.trfExe.transformExecutor.errMsg ( self )

inherited

Definition at line 304 of file trfExe.py.

    def errMsg(self):
        return self._errMsg
    

eventCount()

python.trfExe.transformExecutor.eventCount ( self )

inherited

Definition at line 444 of file trfExe.py.

    def eventCount(self):
        return self._eventCount
 

exe() [1/2]

python.trfExe.scriptExecutor.exe ( self )

inherited

Definition at line 641 of file trfExe.py.

    def exe(self):
        return self._exe
    

exe() [2/2]

python.trfExe.scriptExecutor.exe ( self, value )

inherited

Definition at line 645 of file trfExe.py.

    def exe(self, value):
        self._exe = value
        self._extraMetadata['script'] = value
 

exeArgs() [1/2]

python.trfExe.scriptExecutor.exeArgs ( self )

inherited

Definition at line 650 of file trfExe.py.

    def exeArgs(self):
        return self._exeArgs
    

exeArgs() [2/2]

python.trfExe.scriptExecutor.exeArgs ( self, value )

inherited

Definition at line 654 of file trfExe.py.

    def exeArgs(self, value):
        self._exeArgs = value
#        self._extraMetadata['scriptArgs'] = value
 

execute()

python.trfExe.scriptExecutor.execute ( self )

inherited

Reimplemented from python.trfExe.transformExecutor.

Reimplemented in python.trfExe.bsMergeExecutor, and python.trfExe.POOLMergeExecutor.

Definition at line 723 of file trfExe.py.

    def execute(self):
        self._hasExecuted = True
        msg.info('Starting execution of {0} ({1})'.format(self._name, self._cmd))
        
        self._exeStart = os.times()
        msg.debug('exeStart time is {0}'.format(self._exeStart))
        if ('execOnly' in self.conf.argdict and self.conf.argdict['execOnly'] is True):
            msg.info('execOnly flag is set - execution will now switch, replacing the transform')
            os.execvp(self._cmd[0], self._cmd)
 
        encargs = {'encoding' : 'utf8'}
        try:
            # if we are already inside a container, then
            # must map /srv of the nested container to /srv of the parent container:
            # https://twiki.atlas-canada.ca/bin/view/AtlasCanada/Containers#Mount_point_summary
            if self._alreadyInContainer and self._containerSetup is not None:
                msg.info("chdir /srv to launch a nested container for the substep")
                os.chdir("/srv")
            p = subprocess.Popen(self._cmd, shell = False, stdout = subprocess.PIPE, stderr = subprocess.STDOUT, bufsize = 1, **encargs)
            # go back to the original working directory immediately after to store prmon in it
            if self._alreadyInContainer and self._containerSetup is not None:
                msg.info("chdir {} after launching the nested container".format(self._workdir))
                os.chdir(self._workdir)
 
            if self._memMonitor:
                try:
                    self._memSummaryFile = 'prmon.summary.' + self._name + '.json'
                    self._memFullFile = 'prmon.full.' + self._name
                    memMonitorCommand = ['prmon', '--pid', str(p.pid), '--filename', 'prmon.full.' + self._name, 
                                         '--json-summary', self._memSummaryFile, '--log-filename', 'prmon.' + self._name + '.log',
                                         '--interval', '30']
                    mem_proc = subprocess.Popen(memMonitorCommand, shell = False, close_fds=True, **encargs)
                    # TODO - link mem.full.current to mem.full.SUBSTEP
                except Exception as e:
                    msg.warning('Failed to spawn memory monitor for {0}: {1}'.format(self._name, e))
                    self._memMonitor = False
            
            while p.poll() is None:
                line = p.stdout.readline()
                if line:
                    self._echologger.info(line.rstrip())
            # Hoover up remaining buffered output lines
            for line in p.stdout:
                self._echologger.info(line.rstrip())
    
            self._rc = p.returncode
            msg.info('%s executor returns %d', self._name, self._rc)
            self._exeStop = os.times()
            msg.debug('exeStop time is {0}'.format(self._exeStop))
        except OSError as e:
            errMsg = 'Execution of {0} failed and raised OSError: {1}'.format(self._cmd[0], e)
            msg.error(errMsg)
            raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_EXEC'), errMsg)
        finally:
            if self._memMonitor:
                try:
                    mem_proc.send_signal(signal.SIGUSR1)
                    countWait = 0
                    while (not mem_proc.poll()) and countWait < 10:
                        time.sleep(0.1)
                        countWait += 1
                except OSError:
                    pass
        
        

exeStartTimes()

python.trfExe.transformExecutor.exeStartTimes ( self )

inherited

Definition at line 336 of file trfExe.py.

    def exeStartTimes(self):
        return self._exeStart
 

exeStopTimes()

python.trfExe.transformExecutor.exeStopTimes ( self )

inherited

Definition at line 340 of file trfExe.py.

    def exeStopTimes(self):
        return self._exeStop
 

extraMetadata()

python.trfExe.transformExecutor.extraMetadata ( self )

inherited

Definition at line 292 of file trfExe.py.

    def extraMetadata(self):
        return self._extraMetadata
    

first()

python.trfExe.transformExecutor.first ( self )

inherited

Note

At the moment only athenaExecutor sets this property, but that might be changed...

Definition at line 325 of file trfExe.py.

    def first(self):
        if hasattr(self, '_first'):
            return self._first
        else:
            return None
 

hasExecuted()

python.trfExe.transformExecutor.hasExecuted ( self )

inherited

Definition at line 296 of file trfExe.py.

    def hasExecuted(self):
        return self._hasExecuted
    

hasValidated()

python.trfExe.transformExecutor.hasValidated ( self )

inherited

Definition at line 316 of file trfExe.py.

    def hasValidated(self):
        return self._hasValidated
    

inData() [1/2]

python.trfExe.transformExecutor.inData ( self )

inherited

Definition at line 235 of file trfExe.py.

    def inData(self):
        
        if '_inData' in dir(self):
            return self._inData
        return None
    

inData() [2/2]

python.trfExe.transformExecutor.inData ( self, value )

inherited

Definition at line 242 of file trfExe.py.

    def inData(self, value):
        self._inData = set(value)
        

inDataUpdate()

python.trfExe.transformExecutor.inDataUpdate ( self, value )

inherited

Definition at line 245 of file trfExe.py.

    def inDataUpdate(self, value):
        
        if '_inData' in dir(self):
            self._inData.update(value)
        else:
            
            self.inData = value
 
 

input()

python.trfExe.transformExecutor.input ( self )

inherited

Note

This returns the actual input data with which this executor ran (c.f. inData which returns all the possible data types this executor could run with)

Definition at line 276 of file trfExe.py.

    def input(self):
        
        if '_input' in dir(self):
            return self._input
        return None
    

inputDataTypeCountCheck() [1/2]

python.trfExe.athenaExecutor.inputDataTypeCountCheck

(

self

)

Definition at line 930 of file trfExe.py.

    def inputDataTypeCountCheck(self):
        return self._inputDataTypeCountCheck
    

inputDataTypeCountCheck() [2/2]

python.trfExe.athenaExecutor.inputDataTypeCountCheck	(		self,
			value )

Definition at line 934 of file trfExe.py.

    def inputDataTypeCountCheck(self, value):
        self._inputDataTypeCountCheck = value
        

isValidated()

python.trfExe.transformExecutor.isValidated ( self )

inherited

Definition at line 320 of file trfExe.py.

    def isValidated(self):
        return self._isValidated
    

memAnalysis()

python.trfExe.transformExecutor.memAnalysis ( self )

inherited

Definition at line 398 of file trfExe.py.

    def memAnalysis(self):
        return self._memLeakResult
 

memStats()

python.trfExe.transformExecutor.memStats ( self )

inherited

Definition at line 394 of file trfExe.py.

    def memStats(self):
        return self._memStats
 

myMerger()

python.trfExe.transformExecutor.myMerger ( self )

inherited

Now define properties for these data members.

Definition at line 207 of file trfExe.py.

    def myMerger(self):
        return self._myMerger
 

name() [1/2]

python.trfExe.transformExecutor.name ( self )

inherited

Definition at line 211 of file trfExe.py.

    def name(self):
        return self._name
 

name() [2/2]

python.trfExe.transformExecutor.name ( self, value )

inherited

Definition at line 215 of file trfExe.py.

    def name(self, value):
        self._name = value
 

onlyMP() [1/2]

python.trfExe.athenaExecutor.onlyMP

(

self

)

Definition at line 958 of file trfExe.py.

    def onlyMP(self):
        return self._onlyMP
    

onlyMP() [2/2]

python.trfExe.athenaExecutor.onlyMP	(		self,
			value )

Definition at line 962 of file trfExe.py.

    def onlyMP(self, value):
        self._onlyMP = value
        

onlyMT() [1/2]

python.trfExe.athenaExecutor.onlyMT

(

self

)

Definition at line 966 of file trfExe.py.

    def onlyMT(self):
        return self._onlyMT
 

onlyMT() [2/2]

python.trfExe.athenaExecutor.onlyMT	(		self,
			value )

Definition at line 970 of file trfExe.py.

    def onlyMT(self, value):
        self._onlyMT = value
 

outData() [1/2]

python.trfExe.transformExecutor.outData ( self )

inherited

Definition at line 255 of file trfExe.py.

    def outData(self):
        
        if '_outData' in dir(self):
            return self._outData
        return None
 

outData() [2/2]

python.trfExe.transformExecutor.outData ( self, value )

inherited

Definition at line 262 of file trfExe.py.

    def outData(self, value):
        self._outData = set(value)
    

outDataUpdate()

python.trfExe.transformExecutor.outDataUpdate ( self, value )

inherited

Definition at line 265 of file trfExe.py.

    def outDataUpdate(self, value):
        
        if '_outData' in dir(self):
            self._outData.update(value)
        else:
            
            self.outData = value
    

output()

python.trfExe.transformExecutor.output ( self )

inherited

Note

This returns the actual output data with which this executor ran (c.f. outData which returns all the possible data types this executor could run with)

Definition at line 285 of file trfExe.py.

    def output(self):
        
        if '_output' in dir(self):
            return self._output
        return None
    

postExeCpuTime()

python.trfExe.transformExecutor.postExeCpuTime ( self )

inherited

Definition at line 402 of file trfExe.py.

    def postExeCpuTime(self):
        if self._exeStop and self._valStart:
            return calcCpuTime(self._exeStop, self._valStart)
        else:
            return None
 

postExecute()

python.trfExe.athenaExecutor.postExecute

(

self

)

Reimplemented from python.trfExe.scriptExecutor.

Definition at line 1244 of file trfExe.py.

    def postExecute(self):
        super(athenaExecutor, self).postExecute()
        # MPI merging
        if 'mpi' in self.conf.argdict:
            mpi.mergeOutputs()
 
        # Handle executor substeps
        if self.conf.totalExecutorSteps > 1:
            if self._athenaMP > 0:
                outputDataDictionary = dict([ (dataType, self.conf.dataDictionary[dataType]) for dataType in self._output ])
                athenaMPOutputHandler(self._athenaMPFileReport, self._athenaMPWorkerTopDir, outputDataDictionary, self._athenaMP, False, self.conf.argdict)
            if self.conf.executorStep == self.conf.totalExecutorSteps - 1:
                # first loop over datasets for the output
                for dataType in self._output:
                    newValue = []
                    if self._athenaMP > 0:
                        # assume the same number of workers all the time
                        for i in range(self.conf.totalExecutorSteps):
                            for v in self.conf.dataDictionary[dataType].value:
                                newValue.append(v.replace('_{0}{1}_'.format(executorStepSuffix, self.conf.executorStep),
                                                          '_{0}{1}_'.format(executorStepSuffix, i)))
                    else:
                        self.conf.dataDictionary[dataType].multipleOK = True
                        # just combine all executors
                        for i in range(self.conf.totalExecutorSteps):
                            newValue.append(self.conf.dataDictionary[dataType].originalName + '_{0}{1}'.format(executorStepSuffix, i))
                    self.conf.dataDictionary[dataType].value = newValue
 
                    # do the merging if needed
                    if self.conf.dataDictionary[dataType].io == "output" and len(self.conf.dataDictionary[dataType].value) > 1:
                        self._smartMerge(self.conf.dataDictionary[dataType])
 
        # If this was an athenaMP run then we need to update output files
        elif self._athenaMP > 0:
            outputDataDictionary = dict([ (dataType, self.conf.dataDictionary[dataType]) for dataType in self._output ])
            
            skipFileChecks=False
            if 'eventService' in self.conf.argdict and self.conf.argdict['eventService'].value:
                skipFileChecks=True
            athenaMPOutputHandler(self._athenaMPFileReport, self._athenaMPWorkerTopDir, outputDataDictionary, self._athenaMP, skipFileChecks, self.conf.argdict)
            for dataType in self._output:
                if self.conf.dataDictionary[dataType].io == "output" and len(self.conf.dataDictionary[dataType].value) > 1:
                    self._smartMerge(self.conf.dataDictionary[dataType])
        
        if 'TXT_JIVEXMLTGZ' in self.conf.dataDictionary:
            self._targzipJiveXML()
 
        # Summarise events passed the filter ISF_SimEventFilter from log.ReSim
        # This is a bit ugly to have such a specific feature here though
        # TODO
        # The best is to have a general approach so that user can extract useful info from log
        # Instead of hard coding a pattern, one idea could be that user provides a regExp pattern
        # in which the wanted variable is grouped by a name, then transforms could decode the pattern
        # and use it to extract required info and do the summation during log scan.
        if self._logFileName=='log.ReSim' and self.name=='ReSim':
            msg.info('scanning {0} for reporting events passed the filter ISF_SimEventFilter'.format(self._logFileName))
            self._resimevents = reportEventsPassedSimFilter(self._logFileName)
 
        # Remove intermediate input/output files of sub-steps
        # Delete only files with io="temporay" which are files with pattern "tmp*"
        # Some stubs like tmp.RDO_TRIG_000 created in AthenaMP mode or
        # tmp.HIST_ESD_INT, tmp.HIST_AOD_INT as input to DQHistogramMerge.py are not deleted
        # Enable if --deleteIntermediateOutputfiles is set
        if ('deleteIntermediateOutputfiles' in self.conf._argdict and self.conf._argdict['deleteIntermediateOutputfiles'].value):
          inputDataDictionary = dict([ (dataType, self.conf.dataDictionary[dataType]) for dataType in self._input ])
 
          for k, v in inputDataDictionary.items():
            if not v.io == 'temporary':
              continue
            for filename in v.value:
              if os.access(filename, os.R_OK) and not filename.startswith("/cvmfs"):
                msg.info("Removing intermediate {0} input file {1}".format(k, filename))
                # Check if symbolic link and delete also linked file
                if (os.path.realpath(filename) != filename):
                  targetpath = os.path.realpath(filename)
                os.unlink(filename)
                if (targetpath) and os.access(targetpath, os.R_OK):
                  os.unlink(targetpath)
 
 

postExeWallTime()

python.trfExe.transformExecutor.postExeWallTime ( self )

inherited

Definition at line 409 of file trfExe.py.

    def postExeWallTime(self):
        if self._exeStop and self._valStart:
            return calcWallTime(self._exeStop, self._valStart)
        else:
            return None
 

preExeCpuTime()

python.trfExe.transformExecutor.preExeCpuTime ( self )

inherited

Definition at line 352 of file trfExe.py.

    def preExeCpuTime(self):
        if self._preExeStart and self._exeStart:
            return calcCpuTime(self._preExeStart, self._exeStart)
        else:
            return None
 

preExecute()

python.trfExe.athenaExecutor.preExecute	(		self,
			input = set(),
			output = set() )

Reimplemented from python.trfExe.scriptExecutor.

Reimplemented in python.trfExe.POOLMergeExecutor, and python.trfExe.reductionFrameworkExecutor.

Definition at line 976 of file trfExe.py.

    def preExecute(self, input = set(), output = set()):
        self.setPreExeStart()
        msg.debug('Preparing for execution of {0} with inputs {1} and outputs {2}'.format(self.name, input, output))
        
        # Check we actually have events to process!
        inputEvents = 0
        dt = ""
        if self._inputDataTypeCountCheck is None:
            self._inputDataTypeCountCheck = input
        for dataType in self._inputDataTypeCountCheck:
            if self.conf.dataDictionary[dataType].nentries == 'UNDEFINED':
                continue
 
            thisInputEvents = self.conf.dataDictionary[dataType].nentries
            if thisInputEvents > inputEvents:
                inputEvents = thisInputEvents
                dt = dataType
 
        # Now take into account skipEvents and maxEvents
        if ('skipEvents' in self.conf.argdict and 
            self.conf.argdict['skipEvents'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor) is not None):
            mySkipEvents = self.conf.argdict['skipEvents'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
        else:
            mySkipEvents = 0
 
        if ('maxEvents' in self.conf.argdict and 
            self.conf.argdict['maxEvents'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor) is not None):
            myMaxEvents = self.conf.argdict['maxEvents'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
        else:
            myMaxEvents = -1
        
        # Any events to process...?
        if (self._inputEventTest and mySkipEvents > 0 and mySkipEvents >= inputEvents):
            raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_NOEVENTS'),
                                                           'No events to process: {0} (skipEvents) >= {1} (inputEvents of {2}'.format(mySkipEvents, inputEvents, dt))
        
        try:
            # Expected events to process
            if (myMaxEvents != -1):
                if (self.inData and next(iter(self.inData)) == 'inNULL'):
                    expectedEvents = myMaxEvents
                else:
                    expectedEvents = min(inputEvents-mySkipEvents, myMaxEvents)
            else:
                expectedEvents = inputEvents-mySkipEvents
        except TypeError:
            # catching type error from UNDEFINED inputEvents count
            msg.info('input event count is UNDEFINED, setting expectedEvents to 0')
            expectedEvents = 0
        
        
        OSSetupString = None
 
        # Extract the asetup string
        asetupString = None
        legacyThreadingRelease = False
        if 'asetup' in self.conf.argdict:
            asetupString = self.conf.argdict['asetup'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
            legacyThreadingRelease = asetupReleaseIsOlderThan(asetupString, 22)
        else:
            msg.info('Asetup report: {0}'.format(asetupReport()))
 
        if asetupString is not None:
            legacyOSRelease = asetupReleaseIsOlderThan(asetupString, 24)
            currentOS = os.environ['ALRB_USER_PLATFORM']
            if legacyOSRelease and "centos7" not in currentOS:
                OSSetupString = "centos7"
                msg.info('Legacy release required for the substep {}, will setup a container running {}'.format(self._substep, OSSetupString))
 
 
        # allow overriding the container OS using a flag
        if 'runInContainer' in self.conf.argdict:
            OSSetupString = self.conf.argdict['runInContainer'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
            msg.info('The step {} will be performed in a container running {}, as explicitly requested'.format(self._substep, OSSetupString))
            if OSSetupString is not None and asetupString is None:
                raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_EXEC_SETUP_FAIL'),
                                                                '--asetup must be used for the substep which requires --runInContainer')
 
        # Conditional MP based on runtime arguments
        if self._onlyMPWithRunargs:
            for k in self._onlyMPWithRunargs:
                if k in self.conf._argdict:
                    self._onlyMP = True
 
        # Check the consistency of parallel configuration: CLI flags + evnironment.
        if ((('multithreaded' in self.conf._argdict and self.conf._argdict['multithreaded'].value) or ('multiprocess' in self.conf._argdict and self.conf._argdict['multiprocess'].value)) and
            ('ATHENA_CORE_NUMBER' not in os.environ)):
            # At least one of the parallel command-line flags has been provided but ATHENA_CORE_NUMBER environment has not been set
            msg.warning('either --multithreaded or --multiprocess argument used but ATHENA_CORE_NUMBER environment not set. Athena will continue in Serial mode')
        else:
            # Try to detect AthenaMT mode, number of threads and number of concurrent events
            if not self._disableMT:
                self._athenaMT, self._athenaConcurrentEvents = detectAthenaMTThreads(self.conf.argdict, self.name, legacyThreadingRelease)
 
            # Try to detect AthenaMP mode and number of workers
            if not self._disableMP:
                self._athenaMP = detectAthenaMPProcs(self.conf.argdict, self.name, legacyThreadingRelease)
 
            # Check that we actually support MT
            if self._onlyMP and self._athenaMT > 0:
                msg.info("This configuration does not support MT, falling back to MP")
                if self._athenaMP == 0:
                    self._athenaMP = self._athenaMT
                self._athenaMT = 0
                self._athenaConcurrentEvents = 0
 
            # Check that we actually support MP
            if self._onlyMT and self._athenaMP > 0:
                    msg.info("This configuration does not support MP, using MT")
                    if self._athenaMT == 0:
                        self._athenaMT = self._athenaMP
                        self._athenaConcurrentEvents = self._athenaMP
                    self._athenaMP = 0
 
        # Small hack to detect cases where there are so few events that it's not worthwhile running in MP mode
        # which also avoids issues with zero sized files. Distinguish from the no-input case (e.g. evgen)
        if not self._disableMP and expectedEvents < self._athenaMP and not self._inData=={'inNULL'}:
            msg.info("Disabling AthenaMP as number of input events to process is too low ({0} events for {1} workers)".format(expectedEvents, self._athenaMP))
            self._disableMP = True
            self._athenaMP = 0
 
        # Handle executor steps
        if self.conf.totalExecutorSteps > 1:
            for dataType in output:
                if self.conf._dataDictionary[dataType].originalName:
                    self.conf._dataDictionary[dataType].value[0] = self.conf._dataDictionary[dataType].originalName
                else:
                    self.conf._dataDictionary[dataType].originalName = self.conf._dataDictionary[dataType].value[0]
                self.conf._dataDictionary[dataType].value[0] += "_{0}{1}".format(executorStepSuffix, self.conf.executorStep)
                msg.info("Updated athena output filename for {0} to {1}".format(dataType, self.conf._dataDictionary[dataType].value[0]))
 
        # And if this is (still) athenaMP, then set some options for workers and output file report
        if self._athenaMP > 0:
            self._athenaMPWorkerTopDir = 'athenaMP-workers-{0}-{1}'.format(self._name, self._substep)
            self._athenaMPFileReport = 'athenaMP-outputs-{0}-{1}'.format(self._name, self._substep)
            self._athenaMPEventOrdersFile = 'athenamp_eventorders.txt.{0}'.format(self._name)
            if 'athenaMPUseEventOrders' in self.conf.argdict and self.conf._argdict['athenaMPUseEventOrders'].value is True:
                self._athenaMPReadEventOrders = True
            else:
                self._athenaMPReadEventOrders = False          
            # Decide on scheduling
            if ('athenaMPStrategy' in self.conf.argdict and 
                (self.conf.argdict['athenaMPStrategy'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor) is not None)):
                self._athenaMPStrategy = self.conf.argdict['athenaMPStrategy'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
            else:
                self._athenaMPStrategy = 'SharedQueue'
 
            # See if we have options for the target output file size
            if 'athenaMPMergeTargetSize' in self.conf.argdict:
                for dataType in output:
                    if dataType in self.conf.argdict['athenaMPMergeTargetSize'].value:
                        self.conf._dataDictionary[dataType].mergeTargetSize = self.conf.argdict['athenaMPMergeTargetSize'].value[dataType] * 1000000 # Convert from MB to B
                        msg.info('Set target merge size for {0} to {1}'.format(dataType, self.conf._dataDictionary[dataType].mergeTargetSize))
                    else:
                        # Use a globbing strategy
                        matchedViaGlob = False
                        for mtsType, mtsSize in self.conf.argdict['athenaMPMergeTargetSize'].value.items():
                            if fnmatch(dataType, mtsType):
                                self.conf._dataDictionary[dataType].mergeTargetSize = mtsSize * 1000000 # Convert from MB to B
                                msg.info('Set target merge size for {0} to {1} from "{2}" glob'.format(dataType, self.conf._dataDictionary[dataType].mergeTargetSize, mtsType))
                                matchedViaGlob = True
                                break
                        if not matchedViaGlob and "ALL" in self.conf.argdict['athenaMPMergeTargetSize'].value:
                            self.conf._dataDictionary[dataType].mergeTargetSize = self.conf.argdict['athenaMPMergeTargetSize'].value["ALL"] * 1000000 # Convert from MB to B
                            msg.info('Set target merge size for {0} to {1} from "ALL" value'.format(dataType, self.conf._dataDictionary[dataType].mergeTargetSize))
         
            # For AthenaMP jobs we ensure that the athena outputs get the suffix _000
            # so that the mother process output file (if it exists) can be used directly
            # as soft linking can lead to problems in the PoolFileCatalog (see ATLASJT-317)
            for dataType in output:
                if self.conf.totalExecutorSteps <= 1:
                    self.conf._dataDictionary[dataType].originalName = self.conf._dataDictionary[dataType].value[0]
                if 'eventService' not in self.conf.argdict or 'eventService' in self.conf.argdict and self.conf.argdict['eventService'].value is False:
                    if 'sharedWriter' in self.conf.argdict and self.conf.argdict['sharedWriter'].value:
                        msg.info("SharedWriter: not updating athena output filename for {0}".format(dataType))
                    else:
                        self.conf._dataDictionary[dataType].value[0] += "_000"
                        msg.info("Updated athena output filename for {0} to {1}".format(dataType, self.conf._dataDictionary[dataType].value[0]))
        else:
            self._athenaMPWorkerTopDir = self._athenaMPFileReport = None
 
        
        if 'mpi' in self.conf.argdict:
            msg.info("Running in MPI mode")
            mpi.setupMPIConfig(output, self.conf.dataDictionary)
 
        
        if self._skeleton or self._skeletonCA:
            inputFiles = dict()
            for dataType in input:
                inputFiles[dataType] = self.conf.dataDictionary[dataType]
            outputFiles = dict()
            for dataType in output:
                outputFiles[dataType] = self.conf.dataDictionary[dataType]
 
            # See if we have any 'extra' file arguments
            nameForFiles = commonExecutorStepName(self._name)
            for dataType, dataArg in self.conf.dataDictionary.items():
                if isinstance(dataArg, list) and dataArg:
                    if self.conf.totalExecutorSteps <= 1:
                        raise ValueError('Multiple input arguments provided but only running one substep')
                    if self.conf.totalExecutorSteps != len(dataArg):
                        raise ValueError(f'{len(dataArg)} input arguments provided but running {self.conf.totalExecutorSteps} substeps')
 
                    if dataArg[self.conf.executorStep].io == 'input' and nameForFiles in dataArg[self.conf.executorStep].executor:
                        inputFiles[dataArg[self.conf.executorStep].subtype] = dataArg
                else:
                    if dataArg.io == 'input' and nameForFiles in dataArg.executor:
                        inputFiles[dataArg.subtype] = dataArg
 
            msg.debug('Input Files: {0}; Output Files: {1}'.format(inputFiles, outputFiles))
            
            # Get the list of top options files that will be passed to athena (=runargs file + all skeletons)
            self._topOptionsFiles = self._jobOptionsTemplate.getTopOptions(input = inputFiles, 
                                                                           output = outputFiles)
 
        
        if len(input) > 0:
            self._extraMetadata['inputs'] = list(input)
        if len(output) > 0:
            self._extraMetadata['outputs'] = list(output)
 
        
        dbrelease = dbsetup = None
        if 'DBRelease' in self.conf.argdict:
            dbrelease = self.conf.argdict['DBRelease'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
            if path.islink(dbrelease):
                dbrelease = path.realpath(dbrelease)
            if dbrelease:
                # Classic tarball - filename format is DBRelease-X.Y.Z.tar.gz
                dbdMatch = re.match(r'DBRelease-([\d\.]+)\.tar\.gz', path.basename(dbrelease))
                if dbdMatch:
                    msg.debug('DBRelease setting {0} matches classic tarball file'.format(dbrelease))
                    if not os.access(dbrelease, os.R_OK):
                        msg.warning('Transform was given tarball DBRelease file {0}, but this is not there'.format(dbrelease))
                        msg.warning('I will now try to find DBRelease {0} in cvmfs'.format(dbdMatch.group(1)))
                        dbrelease = dbdMatch.group(1)
                        dbsetup = cvmfsDBReleaseCheck(dbrelease)
                    else:
                        # Check if the DBRelease is setup
                        msg.debug('Setting up {0} from {1}'.format(dbdMatch.group(1), dbrelease))
                        unpacked, dbsetup = unpackDBRelease(tarball=dbrelease, dbversion=dbdMatch.group(1))
                        if unpacked:
                            # Now run the setup.py script to customise the paths to the current location...
                            setupDBRelease(dbsetup)
                # For cvmfs we want just the X.Y.Z release string (and also support 'current')
                else:
                    dbsetup = cvmfsDBReleaseCheck(dbrelease)
        
        
        self._envUpdate = trfEnv.environmentUpdate()
        self._envUpdate.setStandardEnvironment(self.conf.argdict, name=self.name, substep=self.substep)
        self._prepAthenaCommandLine() 
        
                
        super(athenaExecutor, self).preExecute(input, output)
        
        # Now we always write a wrapper, because it's very convenient for re-running individual substeps
        # This will have asetup and/or DB release setups in it
        # Do this last in this preExecute as the _cmd needs to be finalised
        msg.info('Now writing wrapper for substep executor {0}'.format(self._name))
        self._writeAthenaWrapper(asetup=asetupString, dbsetup=dbsetup, ossetup=OSSetupString)
        msg.info('Athena will be executed in a subshell via {0}'.format(self._cmd))
        
                

preExeStartTimes()

python.trfExe.transformExecutor.preExeStartTimes ( self )

inherited

Definition at line 332 of file trfExe.py.

    def preExeStartTimes(self):
        return self._preExeStart
    

preExeWallTime()

python.trfExe.transformExecutor.preExeWallTime ( self )

inherited

Definition at line 359 of file trfExe.py.

    def preExeWallTime(self):
        if self._preExeStart and self._exeStart:
            return calcWallTime(self._preExeStart, self._exeStart)
        else:
            return None
 

rc()

python.trfExe.transformExecutor.rc ( self )

inherited

Definition at line 300 of file trfExe.py.

    def rc(self):
        return self._rc
    

reSimEvent()

python.trfExe.transformExecutor.reSimEvent ( self )

inherited

Definition at line 448 of file trfExe.py.

    def reSimEvent(self):
        return self._resimevents
 

setPreExeStart()

python.trfExe.transformExecutor.setPreExeStart ( self )

inherited

Definition at line 461 of file trfExe.py.

    def setPreExeStart(self):
        if self._preExeStart is None:
            self._preExeStart = os.times()
            msg.debug('preExeStart time is {0}'.format(self._preExeStart))
 

setValStart()

python.trfExe.transformExecutor.setValStart ( self )

inherited

Definition at line 466 of file trfExe.py.

    def setValStart(self):
        if self._valStart is None:
            self._valStart = os.times()
            msg.debug('valStart time is {0}'.format(self._valStart))
 

skeletonCA()

python.trfExe.athenaExecutor.skeletonCA

(

self

)

Definition at line 973 of file trfExe.py.

    def skeletonCA(self):
        return self._skeletonCA
 

substep()

python.trfExe.athenaExecutor.substep

(

self

)

Reimplemented from python.trfExe.transformExecutor.

Definition at line 938 of file trfExe.py.

    def substep(self):
        return self._substep
 

sysTime()

python.trfExe.transformExecutor.sysTime ( self )

inherited

Definition at line 380 of file trfExe.py.

    def sysTime(self):
        if self._exeStart and self._exeStop:
            return self._exeStop[3] - self._exeStart[3]
        else:
            return None
 

trf() [1/2]

python.trfExe.transformExecutor.trf ( self )

inherited

Definition at line 225 of file trfExe.py.

    def trf(self):
        if '_trf' in dir(self):
            return self._trf
        return None
    

trf() [2/2]

python.trfExe.transformExecutor.trf ( self, value )

inherited

Definition at line 231 of file trfExe.py.

    def trf(self, value):
        self._trf = value
        

usrTime()

python.trfExe.transformExecutor.usrTime ( self )

inherited

Definition at line 373 of file trfExe.py.

    def usrTime(self):
        if self._exeStart and self._exeStop:
            return self._exeStop[2] - self._exeStart[2]
        else:
            return None
        

validate()

python.trfExe.athenaExecutor.validate

(

self

)

Reimplemented from python.trfExe.scriptExecutor.

Reimplemented in python.trfExe.optionalAthenaExecutor.

Definition at line 1324 of file trfExe.py.

    def validate(self):
        self.setValStart()
        self._hasValidated = True
        deferredException = None
        memLeakThreshold = 5000
        _hasMemLeak = False
 
        
        try:
            super(athenaExecutor, self).validate()
        except trfExceptions.TransformValidationException as e:
            # In this case we hold this exception until the logfile has been scanned
            msg.error('Validation of return code failed: {0!s}'.format(e))
            deferredException = e
 
        
        if self._memFullFile:
            msg.info('Analysing memory monitor output file {0} for possible memory leak'.format(self._memFullFile))
            self._memLeakResult = analytic().getFittedData(self._memFullFile)
            if self._memLeakResult:
                if self._memLeakResult['slope'] > memLeakThreshold:
                    _hasMemLeak = True
                    msg.warning('Possible memory leak; abnormally high values in memory monitor parameters (ignore this message if the job has finished successfully)')
            else:
                msg.warning('Failed to analyse the memory monitor file {0}'.format(self._memFullFile))
        else:
            msg.info('No memory monitor file to be analysed')
 
        # Logfile scan setup
        # Always use ignorePatterns from the command line
        # For patterns in files, pefer the command line first, then any special settings for
        # this executor, then fallback to the standard default (atlas_error_mask.db)
        if 'ignorePatterns' in self.conf.argdict:
            igPat = self.conf.argdict['ignorePatterns'].value
        else:
            igPat = []
        if 'ignoreFiles' in self.conf.argdict:
            ignorePatterns = trfValidation.ignorePatterns(files = self.conf.argdict['ignoreFiles'].value, extraSearch=igPat)
        elif self._errorMaskFiles is not None:
            ignorePatterns = trfValidation.ignorePatterns(files = self._errorMaskFiles, extraSearch=igPat)
        else:
            ignorePatterns = trfValidation.ignorePatterns(files = athenaExecutor._defaultIgnorePatternFile, extraSearch=igPat)
        
        # Now actually scan my logfile
        msg.info('Scanning logfile {0} for errors in substep {1}'.format(self._logFileName, self._substep))
        self._logScan = trfValidation.athenaLogFileReport(logfile=self._logFileName, substepName=self._substep,
                                                          ignoreList=ignorePatterns)
        worstError = self._logScan.worstError()
        eventLoopWarnings = self._logScan.eventLoopWarnings()
        self._dbMonitor = self._logScan.dbMonitor()
        
 
        # In general we add the error message to the exit message, but if it's too long then don't do
        # that and just say look in the jobReport
        if worstError['firstError']:
            if len(worstError['firstError']['message']) > athenaExecutor._exitMessageLimit:
                if 'CoreDumpSvc' in worstError['firstError']['message']:
                    exitErrorMessage = "Core dump at line {0} (see jobReport for further details)".format(worstError['firstError']['firstLine'])
                elif 'G4Exception' in worstError['firstError']['message']:
                    exitErrorMessage = "G4 exception at line {0} (see jobReport for further details)".format(worstError['firstError']['firstLine'])
                else:
                    exitErrorMessage = "Long {0} message at line {1} (see jobReport for further details)".format(worstError['level'], worstError['firstError']['firstLine'])
            else:
                exitErrorMessage = "Logfile error in {0}: \"{1}\"".format(self._logFileName, worstError['firstError']['message'])
        else:
            exitErrorMessage = "Error level {0} found (see athena logfile for details)".format(worstError['level'])
 
        # If we failed on the rc, then abort now
        if deferredException is not None:
            # Add any logfile information we have
            if worstError['nLevel'] >= stdLogLevels['ERROR']:
                deferredException.errMsg = deferredException.errMsg + "; {0}".format(exitErrorMessage)
            # Add the result of memory analysis
            if _hasMemLeak:
                deferredException.errMsg = deferredException.errMsg + "; Possible memory leak: 'pss' slope: {0} KB/s".format(self._memLeakResult['slope'])
            raise deferredException
        
        
        # Very simple: if we get ERROR or worse, we're dead, except if ignoreErrors=True
        if worstError['nLevel'] == stdLogLevels['ERROR'] and ('ignoreErrors' in self.conf.argdict and self.conf.argdict['ignoreErrors'].value is True):
            msg.warning('Found ERRORs in the logfile, but ignoring this as ignoreErrors=True (see jobReport for details)')
        # Act as if ignoreErrors=True if running in MPI, because we want to be tolerant to the occasional event failure
        elif worstError['nLevel'] >= stdLogLevels['ERROR'] and (not mpi.mpiShouldValidate()):
            msg.warning(f'Found {worstError["level"]} in the logfile in MPI rank {mpi.getMPIRank()} but moving on to be failure-tolerant')
        elif worstError['nLevel'] >= stdLogLevels['ERROR']:
            self._isValidated = False
            msg.error('Fatal error in athena logfile (level {0})'.format(worstError['level']))
            # Add the result of memory analysis
            if _hasMemLeak:
                exitErrorMessage = exitErrorMessage + "; Possible memory leak: 'pss' slope: {0} KB/s".format(self._memLeakResult['slope'])
            raise trfExceptions.TransformLogfileErrorException(trfExit.nameToCode('TRF_EXEC_LOGERROR'), 
                                                                   'Fatal error in athena logfile: "{0}"'.format(exitErrorMessage))
 
        # Print event loop warnings
        if (len(eventLoopWarnings) > 0):
            msg.warning('Found WARNINGS in the event loop, as follows:')
            for element in eventLoopWarnings:
                msg.warning('{0} {1} ({2} instances)'.format(element['item']['service'],element['item']['message'],element['count']))
        
        # Must be ok if we got here!
        msg.info('Executor {0} has validated successfully'.format(self.name))
        self._isValidated = True
 
        self._valStop = os.times()
        msg.debug('valStop time is {0}'.format(self._valStop))
 

validation() [1/2]

python.trfExe.transformExecutor.validation ( self )

inherited

Definition at line 308 of file trfExe.py.

    def validation(self):
        return self._validation
    

validation() [2/2]

python.trfExe.transformExecutor.validation ( self, value )

inherited

Definition at line 312 of file trfExe.py.

    def validation(self, value):
        self._validation = value
        

validationCpuTime()

python.trfExe.transformExecutor.validationCpuTime ( self )

inherited

Definition at line 416 of file trfExe.py.

    def validationCpuTime(self):
        if self._valStart and self._valStop:
            return calcCpuTime(self._valStart, self._valStop)
        else:
            return None
    

validationWallTime()

python.trfExe.transformExecutor.validationWallTime ( self )

inherited

Definition at line 423 of file trfExe.py.

    def validationWallTime(self):
        if self._valStart and self._valStop:
            return calcWallTime(self._valStart, self._valStop)
        else:
            return None
 

valStartTimes()

python.trfExe.transformExecutor.valStartTimes ( self )

inherited

Definition at line 344 of file trfExe.py.

    def valStartTimes(self):
        return self._valStart
 

valStopTimes()

python.trfExe.transformExecutor.valStopTimes ( self )

inherited

Definition at line 348 of file trfExe.py.

    def valStopTimes(self):
        return self._valStop
    

wallTime()

python.trfExe.transformExecutor.wallTime ( self )

inherited

Definition at line 387 of file trfExe.py.

    def wallTime(self):
        if self._exeStart and self._exeStop:
            return calcWallTime(self._exeStart, self._exeStop)
        else:
            return None
        

wallTimeTotal()

python.trfExe.transformExecutor.wallTimeTotal ( self )

inherited

Definition at line 437 of file trfExe.py.

    def wallTimeTotal(self):
        if self._preExeStart and self._valStop:
            return calcWallTime(self._preExeStart, self._valStop)
        else:
            return None
        

Member Data Documentation

_alreadyInContainer

python.trfExe.transformExecutor._alreadyInContainer = None

protectedinherited

Definition at line 198 of file trfExe.py.

_asetup

python.trfExe.athenaExecutor._asetup = asetup

protected

Definition at line 1576 of file trfExe.py.

_athenaConcurrentEvents

int python.trfExe.transformExecutor._athenaConcurrentEvents = 0

protectedinherited

Definition at line 195 of file trfExe.py.

_athenaMP

int python.trfExe.transformExecutor._athenaMP = 0

protectedinherited

Definition at line 193 of file trfExe.py.

_athenaMPEventOrdersFile

str python.trfExe.athenaExecutor._athenaMPEventOrdersFile = 'athenamp_eventorders.txt.{0}'.format(self._name)

protected

Definition at line 1113 of file trfExe.py.

_athenaMPFileReport

python.trfExe.athenaExecutor._athenaMPFileReport = 'athenaMP-outputs-{0}-{1}'.format(self._name, self._substep)

protected

Definition at line 1112 of file trfExe.py.

_athenaMPReadEventOrders

bool python.trfExe.athenaExecutor._athenaMPReadEventOrders = True

protected

Definition at line 1115 of file trfExe.py.

_athenaMPStrategy

str python.trfExe.athenaExecutor._athenaMPStrategy = self.conf.argdict['athenaMPStrategy'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)

protected

Definition at line 1121 of file trfExe.py.

_athenaMPWorkerTopDir

python.trfExe.athenaExecutor._athenaMPWorkerTopDir = 'athenaMP-workers-{0}-{1}'.format(self._name, self._substep)

protected

Definition at line 1111 of file trfExe.py.

_athenaMT

int python.trfExe.transformExecutor._athenaMT = 0

protectedinherited

Definition at line 194 of file trfExe.py.

_cmd

python.trfExe.scriptExecutor._cmd = None

protectedinherited

Note

If an inherited class has set self._cmd leave it alone

Definition at line 636 of file trfExe.py.

_containerSetup

python.trfExe.transformExecutor._containerSetup = None

protectedinherited

Definition at line 199 of file trfExe.py.

_dataArgs

python.trfExe.athenaExecutor._dataArgs = dataArgs

protected

Definition at line 897 of file trfExe.py.

_dbMonitor

python.trfExe.transformExecutor._dbMonitor = None

protectedinherited

Definition at line 196 of file trfExe.py.

_dbsetup

python.trfExe.athenaExecutor._dbsetup = dbsetup

protected

Definition at line 1577 of file trfExe.py.

_defaultIgnorePatternFile

list python.trfExe.athenaExecutor._defaultIgnorePatternFile = ['atlas_error_mask.db']

staticprotected

Definition at line 846 of file trfExe.py.

_disableMP

bool python.trfExe.athenaExecutor._disableMP = disableMP

protected

Definition at line 901 of file trfExe.py.

_disableMT

python.trfExe.athenaExecutor._disableMT = disableMT

protected

Definition at line 900 of file trfExe.py.

_echologger

python.trfExe.scriptExecutor._echologger = logging.getLogger(self._name)

protectedinherited

Note

Query the environment for echo configuration Let the manual envars always win over auto-detected settings

Definition at line 692 of file trfExe.py.

_echoOutput

bool python.trfExe.scriptExecutor._echoOutput = False

protectedinherited

Note

Query the environment for echo configuration Let the manual envars always win over auto-detected settings

Definition at line 633 of file trfExe.py.

_echostream

python.trfExe.scriptExecutor._echostream = logging.StreamHandler(_encoding_stream(sys.stdout))

protectedinherited

Definition at line 702 of file trfExe.py.

_envUpdate

python.trfExe.athenaExecutor._envUpdate = trfEnv.environmentUpdate()

protected

Look for environment updates and perpare the athena command line.

Definition at line 1229 of file trfExe.py.

_errMsg

str python.trfExe.transformExecutor._errMsg = None

protectedinherited

Definition at line 172 of file trfExe.py.

_errorMaskFiles

python.trfExe.athenaExecutor._errorMaskFiles = errorMaskFiles

protected

Definition at line 898 of file trfExe.py.

_eventCount

python.trfExe.transformExecutor._eventCount = None

protectedinherited

Definition at line 192 of file trfExe.py.

_exe

python.trfExe.scriptExecutor._exe = exe

protectedinherited

Definition at line 623 of file trfExe.py.

_exeArgs

python.trfExe.scriptExecutor._exeArgs = exeArgs

protectedinherited

Definition at line 626 of file trfExe.py.

_exeLogFile

python.trfExe.scriptExecutor._exeLogFile = logging.FileHandler(self._logFileName, mode='w', **encargs)

protectedinherited

Definition at line 697 of file trfExe.py.

_exeStart

python.trfExe.transformExecutor._exeStart = None

protectedinherited

Definition at line 187 of file trfExe.py.

_exeStop

python.trfExe.transformExecutor._exeStop

protectedinherited

Definition at line 367 of file trfExe.py.

_exitMessageLimit

int python.trfExe.athenaExecutor._exitMessageLimit = 200

staticprotected

Definition at line 845 of file trfExe.py.

_extraMetadata

dict python.trfExe.transformExecutor._extraMetadata = {}

protectedinherited

Definition at line 181 of file trfExe.py.

_extraRunargs

python.trfExe.athenaExecutor._extraRunargs = extraRunargs

protected

Definition at line 894 of file trfExe.py.

_hasExecuted

bool python.trfExe.transformExecutor._hasExecuted = False

protectedinherited

Definition at line 170 of file trfExe.py.

_hasValidated

bool python.trfExe.transformExecutor._hasValidated = False

protectedinherited

Definition at line 175 of file trfExe.py.

_inData

python.trfExe.transformExecutor._inData = set(inData)

protectedinherited

Definition at line 144 of file trfExe.py.

_input

python.trfExe.scriptExecutor._input = input

protectedinherited

Definition at line 662 of file trfExe.py.

_inputDataTypeCountCheck

python.trfExe.athenaExecutor._inputDataTypeCountCheck = inputDataTypeCountCheck

protected

Definition at line 899 of file trfExe.py.

_inputEventTest

python.trfExe.athenaExecutor._inputEventTest = inputEventTest

protected

Definition at line 892 of file trfExe.py.

_isValidated

bool python.trfExe.transformExecutor._isValidated = False

protectedinherited

Definition at line 176 of file trfExe.py.

_jobOptionsTemplate

python.trfExe.athenaExecutor._jobOptionsTemplate = JobOptionsTemplate(exe = self, version = '$Id: trfExe.py 792052 2017-01-13 13:36:51Z mavogel $')

protected

Definition at line 925 of file trfExe.py.

_literalRunargs

python.trfExe.athenaExecutor._literalRunargs = literalRunargs

protected

Definition at line 896 of file trfExe.py.

_logFileName

python.trfExe.scriptExecutor._logFileName = "log.{0}".format(self._name)

protectedinherited

Note

If an inherited class has set self._cmd leave it alone

Query the environment for echo configuration Let the manual envars always win over auto-detected settings

Definition at line 671 of file trfExe.py.

_logScan

python.trfExe.athenaExecutor._logScan

protected

Initial value:

=  trfValidation.athenaLogFileReport(logfile=self._logFileName, substepName=self._substep,
                                                          ignoreList=ignorePatterns)

Definition at line 1375 of file trfExe.py.

_memFullFile

python.trfExe.transformExecutor._memFullFile = None

protectedinherited

Definition at line 191 of file trfExe.py.

_memLeakResult

dict python.trfExe.transformExecutor._memLeakResult = {}

protectedinherited

Definition at line 190 of file trfExe.py.

_memMonitor

bool python.trfExe.scriptExecutor._memMonitor = memMonitor

protectedinherited

Definition at line 638 of file trfExe.py.

_memStats

dict python.trfExe.transformExecutor._memStats = {}

protectedinherited

Definition at line 189 of file trfExe.py.

_memSummaryFile

python.trfExe.scriptExecutor._memSummaryFile = 'prmon.summary.' + self._name + '.json'

protectedinherited

Definition at line 749 of file trfExe.py.

_myMerger

list python.trfExe.transformExecutor._myMerger = []

protectedinherited

Definition at line 202 of file trfExe.py.

_name

python.trfExe.transformExecutor._name = forceToAlphaNum(name)

protectedinherited

Definition at line 141 of file trfExe.py.

_onlyMP

bool python.trfExe.athenaExecutor._onlyMP = onlyMP

protected

Definition at line 902 of file trfExe.py.

_onlyMPWithRunargs

python.trfExe.athenaExecutor._onlyMPWithRunargs = onlyMPWithRunargs

protected

Definition at line 904 of file trfExe.py.

_onlyMT

python.trfExe.athenaExecutor._onlyMT = onlyMT

protected

Definition at line 903 of file trfExe.py.

_originalCmd

list python.trfExe.athenaExecutor._originalCmd = self._cmd

protected

Definition at line 1575 of file trfExe.py.

_outData

python.trfExe.transformExecutor._outData = set(outData)

protectedinherited

Definition at line 145 of file trfExe.py.

_output

python.trfExe.scriptExecutor._output = output

protectedinherited

Definition at line 663 of file trfExe.py.

_perfMonFile

python.trfExe.athenaExecutor._perfMonFile = None

protected

Definition at line 908 of file trfExe.py.

_preExeStart

python.trfExe.transformExecutor._preExeStart = None

protectedinherited

Note

Place holders for resource consumption. CPU and walltime are available for all executors but currently only athena is instrumented to fill in memory stats (and then only if PerfMonSD is enabled).

Definition at line 186 of file trfExe.py.

_rc

python.trfExe.transformExecutor._rc = -1

protectedinherited

Definition at line 171 of file trfExe.py.

_resimevents

python.trfExe.transformExecutor._resimevents = None

protectedinherited

Definition at line 197 of file trfExe.py.

_runtimeRunargs

python.trfExe.athenaExecutor._runtimeRunargs = runtimeRunargs

protected

Definition at line 895 of file trfExe.py.

_setupFile

str python.trfExe.athenaExecutor._setupFile = 'setup.{name}.sh'.format(name = self._name)

protected

Definition at line 1582 of file trfExe.py.

_skeleton

list python.trfExe.athenaExecutor._skeleton = [skeletonFile]

protected

Definition at line 913 of file trfExe.py.

_skeletonCA

python.trfExe.athenaExecutor._skeletonCA = skeletonCA

protected

Handle MPI setup.

Write the skeleton file and prep athena

Definition at line 905 of file trfExe.py.

_substep

python.trfExe.athenaExecutor._substep = forceToAlphaNum(substep)

protected

Definition at line 891 of file trfExe.py.

_topOptionsFiles

python.trfExe.athenaExecutor._topOptionsFiles

protected

Initial value:

=  self._jobOptionsTemplate.getTopOptions(input = inputFiles, 
                                                                           output = outputFiles)

Definition at line 1191 of file trfExe.py.

_trf

python.trfExe.transformExecutor._trf = value

protectedinherited

Definition at line 232 of file trfExe.py.

_tryDropAndReload

python.trfExe.athenaExecutor._tryDropAndReload = tryDropAndReload

protected

Add –drop-and-reload if possible (and allowed!).

Definition at line 893 of file trfExe.py.

_validation

python.trfExe.transformExecutor._validation = value

protectedinherited

Definition at line 313 of file trfExe.py.

_valStart

python.trfExe.transformExecutor._valStart = None

protectedinherited

Definition at line 188 of file trfExe.py.

_valStop

python.trfExe.transformExecutor._valStop

protectedinherited

Definition at line 417 of file trfExe.py.

_workdir

python.trfExe.scriptExecutor._workdir

protectedinherited

Definition at line 744 of file trfExe.py.

_wrapperFile

python.trfExe.athenaExecutor._wrapperFile = 'runwrapper.{name}.sh'.format(name = self._name)

protected

Definition at line 1581 of file trfExe.py.

conf

python.trfExe.transformExecutor.conf = conf

inherited

Executor configuration:

Note

that if conf and trf are None then we'll probably set the conf up later (this is allowed and expected to be done once the master transform has figured out what it's doing for this job)

Definition at line 163 of file trfExe.py.

exeArgs

python.trfExe.scriptExecutor.exeArgs

inherited

Definition at line 712 of file trfExe.py.

inData

python.trfExe.transformExecutor.inData = value

inherited

Note

Protect against _inData not yet being defined

Use normal setter

Definition at line 251 of file trfExe.py.

outData

python.trfExe.transformExecutor.outData = value

inherited

Note

Protect against _outData not yet being defined

Use normal setter

Definition at line 271 of file trfExe.py.

---

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

• trfExe.py