ExecStep.py

Go to the documentation of this file.

#
# Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
#
 
'''
Definitions of exec steps (main job) in Trigger ART tests
'''
 
import os
import re
 
from TrigValTools.TrigValSteering.Step import Step
from TrigValTools.TrigValSteering.Input import is_input_defined, get_input
 
 
class ExecStep(Step):
    '''
    Step executing the main job of a Trigger ART test. This can be either
    athena or athenaHLT or a transform. There can be several ExecSteps in
    one Test
    '''
 
    def __init__(self, name=None):
        super(ExecStep, self).__init__(name)
        self.input = None
        self.input_object = None
        self.job_options = None
        self.flags = []
        self.threads = None
        self.concurrent_events = None
        self.forks = None
        self.max_events = None
        self.skip_events = None
        self.use_pickle = False
        self.imf = True
        self.perfmon = True
        self.fpe_auditor = True
        self.costmon = False
        self.malloc = False
        self.prmon = True
        self.config_only = False
        self.auto_report_result = True
        self.required = True
        self.depends_on_previous = True
 
    def construct_name(self):
        if self.name and self.type == 'other':
            return self.name
        name = self.type if self.type else self.executable
        if self.name:
            name += '.'+self.name
        return name
 
    def get_log_file_name(self):
        if self.log_file_name is not None:
            return self.log_file_name
        else:
            return self.construct_name()+'.log'
 
    def configure(self, test):
        # Construct the name
        self.name = self.construct_name()
        self.log_file_name = self.name+'.log'
        self.log.debug('Configuring step %s', self.name)
 
        # Parse configuration
        self.configure_type()
        self.configure_input()
        self.configure_job_options()
        self.configure_args(test)
 
        super(ExecStep, self).configure(test)
 
    def configure_type(self):
        self.log.debug('Configuring type for step %s', self.name)
        # Check if type or executable is specified
        if self.type is None and self.executable is None:
            self.misconfig_abort('Cannot configure a step without specified type or executable')
 
        # Configure executable from type
        known_types = ['athena', 'athenaHLT', 'athenaEF', 'Reco_tf', 'Trig_reco_tf', 'Derivation_tf']
        if self.type in known_types:
            if self.executable is not None:
                self.log.warning('type=%s was specified, so executable=%s '
                                 'will be ignored and deduced automatically '
                                 'from type', self.type, self.executable)
            self.executable = '{}.py'.format(self.type)
        elif self.type == 'other' or self.type is None:
            self.type = 'other'
        else:
            self.misconfig_abort('Cannot determine type of this step')
 
        if self.executable.endswith('_tf.py'):
            def del_env(varname):
                if varname in os.environ:
                    del os.environ[varname]
 
            # Ensure no log duplication for transforms
            os.environ['TRF_NOECHO'] = '1'
            del_env('TRF_ECHO')
 
            # We don't use the Reco_tf auto-configuration for MP/MT transforms,
            # instead we pass our parameters of choice to athenaopts
            del_env('ATHENA_NPROC_NUM')
            del_env('ATHENA_CORE_NUMBER')
 
    def _is_good_file(self, path):
        if os.path.isfile(path):
            return True
        from ROOT import TFile
        with TFile.Open(path, "READ") as f:
            if not f.IsOpen() or f.IsZombie():
                return False
        return True
    def configure_input(self):
        self.log.debug('Configuring input for step %s', self.name)
        if self.input is None:
            self.misconfig_abort(
                'Input not provided for this step. To configure'
                'a step without input, use an empty string')
 
        # Step with no input
        if len(self.input) == 0:
            return
 
        # Try to interpret input as keyword
        if is_input_defined(self.input):
            self.input_object = get_input(self.input)
            if self.input_object is None:
                self.misconfig_abort('Failed to load input with keyword %s', self.input)
            return
 
        # Try to interpret explicit paths
        input_paths = self.input.split(',')
        for path in input_paths:
            if not self._is_good_file(path):
                self.misconfig_abort(f"The provided input does not exist: {path}")
        self.log.debug('Using explicit input: %s', self.input)
 
    def configure_job_options(self):
        '''Check job options configuration'''
        self.log.debug('Configuring job options for step %s', self.name)
        if self.type == 'other':
            self.log.debug('Skipping job options check for step.type=other')
            return
        if self.use_pickle:
            self.log.debug('Skipping job options check for step running from a pickle file')
            return
 
        if self.type.endswith('_tf'):
            if self.job_options is None:
                return
            else:
                self.misconfig_abort('Transform %s does not accept job options', self.type)
        elif self.job_options is None or len(self.job_options) == 0:
            self.misconfig_abort('Job options not provided for this step')
 
    def add_trf_precommand(self, precommand):
        '''Add preExec to transform command'''
 
        if self.type in ['Reco_tf', 'Trig_reco_tf', 'Derivation_tf']:
            if 'inputBS_RDOFile' in self.args:
                precommand = 'BSRDOtoRAW:' + precommand
            elif 'outputRDO_TRIGFile' in self.args or 'doRDO_TRIG' in self.args:
                precommand = 'RDOtoRDOTrigger:' + precommand
            else:
                self.log.debug('Skip adding precommand %s to step %s because it is a transform which '
                               'does not run Trigger', precommand, self.name)
                return
 
        # match --preExec, --preExec= ignoring spaces
        m = re.search(r'--preExec\s*=?\s*', self.args)
        if m is None:
            self.args += f' --preExec "{precommand}" '
        else:
            # Insert new preExec. It is important to not use the '=' sign so we
            # can chain multiple preExecs.
            self.args = self.args[:m.span()[0]] + f' --preExec "{precommand}" ' + self.args[m.span()[1]:]
 
    def configure_args(self, test):
        self.log.debug('Configuring args for step %s', self.name)
        if self.args is None:
            self.args = ''
        athenaopts = ''
 
        # Disable prmon for Reco_tf because it is already started inside the transform
        if self.type == 'Reco_tf' or self.type == 'Derivation_tf':
            self.prmon = False
 
        # Disable perfmon for multi-fork jobs as it cannot deal well with them
        if self.forks and self.forks > 1 and self.perfmon:
            self.log.debug('Disabling perfmon because forks=%d > 1', self.forks)
            self.perfmon = False
        # Disable perfmon for transforms (Reco_tf enables it itself, Trig_reco_tf would need special handling
        # depending on whether it runs athena or athenaHLT)
        if self.type.endswith('_tf') and self.perfmon:
            self.log.debug('Disabling perfmon for the transform step type %s', self.type)
            self.perfmon = False
 
        # Append imf/perfmon
        if self.type != 'other':
            if self.imf:
                athenaopts += ' --imf'
            if self.perfmon:
                if self.type == 'athenaHLT' or self.type == 'athenaEF':
                    athenaopts += ' --perfmon'
                elif self.type == 'athena':
                    athenaopts += ' --perfmon=fastmonmt'
            if self.malloc:
                athenaopts += " --stdcmalloc "
 
        # Enable CostMonitoring/FPEAuditor
        if self.type != 'other':
            if self.costmon:
                self.flags.append('Trigger.CostMonitoring.monitorAllEvents=True')
            if self.fpe_auditor:
                self.flags.append('Exec.FPE=1')
 
        # Run config-only if requested
        if self.config_only :
 
            if self.type == 'athenaHLT' or self.type == 'athenaEF' or (self.type == "other" and self.executable == "athenaHLT.py") or (self.type == "other" and self.executable == "athenaEF.py") :
                athenaopts += ' --dump-config-exit'
 
            elif self.type == 'athena' or self.type == 'Reco_tf' or self.type == 'Derivation_tf' or (self.type == "other" and self.executable == "athena.py") :
                athenaopts += ' --config-only=' + self.name + '.pkl'
 
            # No current support if it isn't clear exactly what's being run
            # This includes Trig_reco_tf and 'other' where the executable is not known
            else :
                self.misconfig_abort('Cannot determine what config-only option is needed. Consider adding the appropriate flag to "args" instead.')
 
        # Default threads/concurrent_events/forks
        if test.package_name == 'TrigP1Test' and self.type == 'athenaHLT':
            if self.threads is None:
                self.threads = 1
            if self.concurrent_events is None:
                self.concurrent_events = 1
            if self.forks is None:
                self.forks = 1
        if test.package_name == 'TrigP1Test' and self.type == 'athenaEF':
            if self.threads is None:
                self.threads = 1
            if self.concurrent_events is None:
                self.concurrent_events = 1
 
        # Append threads/concurrent_events/forks
        if self.threads is not None:
            athenaopts += ' --threads={}'.format(self.threads)
        if self.concurrent_events is not None:
            athenaopts += ' --concurrent-events={}'.format(
                self.concurrent_events)
        if self.forks is not None and self.type != 'athenaEF':
            athenaopts += ' --nprocs={}'.format(self.forks)
 
        # Append athenaopts
        athenaopts = athenaopts.strip()
        if self.type.endswith('_tf'):
            self.args += ' --athenaopts="{}"'.format(athenaopts)
        else:
            self.args += ' '+athenaopts
 
        # Default max events
        if self.max_events is None:
            if test.art_type == 'build':
                if test.package_name == 'TrigP1Test':
                    self.max_events = 80
                else:
                    self.max_events = 20
            else:
                self.max_events = 1000
 
        # Set prmon interval based on max events
        if self.prmon:
            if self.max_events <= 100:
                self.prmon_interval = 5
            else:
                self.prmon_interval = 10
 
        # Append max/skip events
        if self.type == 'athena':
            self.args += ' --evtMax={}'.format(self.max_events)
        elif self.type == 'athenaHLT' or self.type == 'athenaEF':
            self.args += ' --number-of-events={}'.format(self.max_events)
        elif self.type.endswith('_tf'):
            self.args += ' --maxEvents={}'.format(self.max_events)
        if self.skip_events is not None:
            if self.type == 'athena':
                self.args += ' --skipEvents={}'.format(self.skip_events)
            elif self.type == 'athenaHLT' or self.type == 'athenaEF':
                self.args += ' --skip-events={}'.format(self.skip_events)
            elif self.type.endswith('_tf'):
                self.args += ' --skipEvents={}'.format(self.skip_events)
 
        # Append input
        if len(self.input) > 0:
            if self.input_object is not None:
                if self.type == 'athenaHLT' or self.type == 'athenaEF':
                    input_str = ' --file='.join(self.input_object.paths)
                else:
                    input_str = ','.join(self.input_object.paths)
            else:
                input_str = self.input
            if self.type == 'athena':
                self.args += ' --filesInput={}'.format(input_str)
            elif self.type == 'athenaHLT' or self.type == 'athenaEF':
                self.args += ''.join([f" --file={inputFile}" for inputFile in input_str.split(',')])
            elif self.type.endswith('_tf'):
                if self.input_object is None:
                    self.misconfig_abort(
                        'Cannot build inputXYZFile string for transform '
                        ' from explicit input path. Use input=\'\' and '
                        'specify the input explicitly in args')
                if self.type == 'Trig_reco_tf' and '--prodSysBSRDO True' in self.args:
                    self.args += ' --inputBS_RDOFile={}'.format(input_str)
                else:
                    self.args += ' --input{}File={}'.format(
                        self.input_object.format, input_str)
 
 
        # Append job options
        if self.job_options is not None:
            self.args += ' '+self.job_options
 
        # Append flags
        if self.flags:
            if not isinstance(self.flags, (list, tuple)):
                self.misconfig_abort('Wrong type for flags. Expected list or tuple.')
 
            if self.type.endswith('_tf'):  # for transform, set flags as pre-exec
                if self.type == 'Trig_reco_tf':
                    # No 'flags.' prefix for the trigger transform
                    self.add_trf_precommand(' '.join(f'{flag}' for flag in self.flags))
                else:
                    self.add_trf_precommand(';'.join(f'flags.{flag}' for flag in self.flags))
            else:  # athena(HLT)
                self.args += ' ' + ' '.join(self.flags)
 
        # Strip extra whitespace
        self.args = self.args.strip()