precedence.py

Go to the documentation of this file.

import json
import os
import random
import sys
 
# FIXME: workaround for the old version of networkx in LCG 100
import warnings
 
warnings.filterwarnings("ignore", message='"is" with a literal', category=SyntaxWarning)
 
import networkx as nx
from Configurables import CPUCruncher, Gaudi__Sequencer
 
from Gaudi.Configuration import INFO
 
 
def _buildFilePath(filePath):
    if not os.path.exists(filePath):
        __fullFilePath__ = os.path.realpath(
            os.path.join(
                os.environ.get("ENV_PROJECT_SOURCE_DIR", ""),
                "GaudiHive",
                "data",
                filePath,
            )
        )
        if not os.path.exists(__fullFilePath__):
            __fullFilePath__ = os.path.realpath(
                os.path.join(
                    os.environ.get("ENV_PROJECT_SOURCE_DIR", ""),
                    "Gaudi",
                    "GaudiHive",
                    "data",
                    filePath,
                )
            )
            if not os.path.exists(__fullFilePath__):
                print(
                    f"\nERROR: invalid file path '{filePath}'. "
                    "It must be either absolute, or relative to "
                    "'$ENV_PROJECT_SOURCE_DIR/GaudiHive/data/' or to "
                    "'$ENV_PROJECT_SOURCE_DIR/Gaudi/GaudiHive/data/'."
                )
                sys.exit(1)
    else:
        __fullFilePath__ = filePath
 
    return __fullFilePath__
 
 
class UniformTimeValue:
    """A class to manage uniform algorithm timing"""
 
    def __init__(self, avgRuntime, varRuntime=0):
        self.avgRuntime = avgRuntime
        self.varRuntime = varRuntime
 
    def get(self, algoName=""):
        """Get time and its variance (in a tuple) for a given algorithm name"""
 
        return self.avgRuntime, self.varRuntime
 
 
class RealTimeValue:
    """A class to manage real algorithm timing"""
 
    def __init__(self, path, defaultTime, factor=1):
        """
        defaultTime -- run time, assigned to an algorithm if no time is found in provided timing library
                       (and it will also be scaled by the 'factor' argument)
        """
 
        self.path = os.path.realpath(_buildFilePath(path))
        self.factor = factor
        self.defaultTime = defaultTime  # typically 0.05s
        self.varRuntime = 0
 
        self.file = open(self.path)
        self.timings = json.load(self.file)
 
    def get(self, algoName=""):
        """Get time for a given algorithm name"""
 
        if algoName in self.timings:
            time = float(self.timings[algoName])
        else:
            capAlgoName = algoName[0].upper() + algoName[1 : len(algoName)]
 
            if capAlgoName in self.timings:
                time = float(self.timings[capAlgoName])
            else:
                time = self.defaultTime
                print(
                    f"WARNING: Timing for {algoName} (or {capAlgoName}) not found in the provided library, using default one: {time}"
                )
 
        time = time * self.factor
 
        return time, self.varRuntime
 
 
class UniformBooleanValue:
    def __init__(self, value):
        self.value = value
 
    def get(self):
        return self.value
 
 
class RndBiasedBooleanValue:
    """Provides randomly ordered set of boolean values with requested proportion of True and False."""
 
    def __init__(self, pattern, seed=None):
        """
        Keyword arguments:
        pattern -- either a dictionary describing proportion of True and False (e.g., {True:5,False:15}), or
                   a list/tuple containing a pattern to be used as-is (e.g., [False,True,True,False])
        seed -- an int, long or other hashable object to initialize random number generator (passed to random.shuffle as-is)
        """
 
        if isinstance(pattern, dict):
            proportion = pattern
 
            length = proportion[True] + proportion[False]
            if length <= 0:
                raise ValueError(f"Wrong set length requested: {length}")
 
            self.pattern = [False for i in range(proportion[False])] + [
                True for i in range(proportion[True])
            ]
 
            if seed is not None:
                random.seed(seed)
 
            random.shuffle(self.pattern)
 
        elif isinstance(pattern, (list, tuple)):
            self.pattern = pattern
        else:
            raise "ERROR: unknown pattern type"
 
        self.generator = self._create_generator(self.pattern)
 
    def _create_generator(self, pattern):
        yield from pattern
 
    def get(self):
        return next(self.generator)
 
    def get_pattern(self):
        return self.pattern
 
 
class CruncherSequence:
    """Constructs the sequence tree of CPUCrunchers with provided control flow and data flow precedence rules."""
 
    unique_sequencers = []
    dupl_seqs = {}
    OR_sequencers = []
    unique_algos = []
    dupl_algos = {}
 
    unique_data_objects = []
 
    def __init__(
        self,
        timeValue,
        BlockingBoolValue,
        sleepFraction,
        cfgPath,
        dfgPath,
        topSequencer,
        showStat=False,
        timeline=False,
        outputLevel=INFO,
        cardinality=1,
    ):
        """
        Keyword arguments:
        timeValue -- timeValue object to set algorithm execution time
        BlockingBoolValue -- *BooleanValue object to set whether an algorithm has to experience CPU-blocking execution
        cfgPath -- relative to $ENV_PROJECT_SOURCE_DIR/GaudiHive/data path to GRAPHML file with control flow dependencies
        dfgPath -- relative to $ENV_PROJECT_SOURCE_DIR/GaudiHive/data path to GRAPHML file with data flow dependencies
        showStat -- print out statistics on precedence graph
        """
 
        self.cardinality = cardinality
        self.timeValue = timeValue
        self.BlockingBoolValue = BlockingBoolValue
        self.sleepFraction = sleepFraction
 
        self.cfg = nx.read_graphml(_buildFilePath(cfgPath))
        self.dfg = nx.read_graphml(_buildFilePath(dfgPath))
 
        self.enableTimeline = timeline
 
        self.outputLevel = outputLevel
 
        # Generate control flow part
        self.sequencer = self._generate_sequence(topSequencer)
 
        if showStat:
            print("\n===== Statistics on Algorithms =====")
            print(
                "Total number of algorithm nodes: ",
                len(self.unique_algos)
                + sum([self.dupl_algos[i] - 1 for i in self.dupl_algos]),
            )
            print("Number of unique algorithms: ", len(self.unique_algos))
            print(
                "  -->",
                len(self.dupl_algos),
                "of them being re-used with the following distribution: ",
                [self.dupl_algos[i] for i in self.dupl_algos],
            )
            # pprint.pprint(dupl_algos)
 
            print("\n===== Statistics on Sequencers =====")
            print(
                "Total number of sequencers: ",
                len(self.unique_sequencers)
                + sum([self.dupl_seqs[i] - 1 for i in self.dupl_seqs]),
            )
            print("Number of unique sequencers: ", len(self.unique_sequencers))
            print(
                "  -->",
                len(self.dupl_seqs),
                "of them being re-used with the following distribution: ",
                [self.dupl_seqs[i] for i in self.dupl_seqs],
            )
            # pprint.pprint(dupl_seqs)
            print("Number of OR-sequencers: ", len(self.OR_sequencers))
 
            print("\n===== Statistics on DataObjects =====")
            print("Number of unique DataObjects: ", len(self.unique_data_objects))
            # pprint.pprint(self.unique_data_objects)
            print()
 
    def get(self):
        return self.sequencer
 
    def _declare_data_deps(self, algo_name, algo):
        """Declare data inputs and outputs for a given algorithm."""
 
        # Declare data inputs
        for inNode, outNode in self.dfg.in_edges(algo_name):
            dataName = inNode
            if dataName not in self.unique_data_objects:
                self.unique_data_objects.append(dataName)
 
            if dataName not in algo.inpKeys:
                algo.inpKeys.append(dataName)
 
        # Declare data outputs
        for inNode, outNode in self.dfg.out_edges(algo_name):
            dataName = outNode
            if dataName not in self.unique_data_objects:
                self.unique_data_objects.append(dataName)
 
            if dataName not in algo.outKeys:
                algo.outKeys.append(dataName)
 
    def _generate_sequence(self, name, seq=None):
        """Assemble the tree of sequencers."""
 
        if not seq:
            seq = Gaudi__Sequencer(name, ShortCircuit=False)
 
        for n in self.cfg[name]:
            # extract entity name and type
            algo_name = n.split("/")[1] if "/" in n else n
 
            if "type" in self.cfg.nodes[n]:
                # first rely on explicit type, if given
                algo_type = self.cfg.nodes[n].get("type")
            else:
                # if the type is not given explicitly, try to extract it from entity name,
                # and, if unsuccessful, assume it is an algorithm
                algo_type = n.split("/")[0] if "/" in n else "Algorithm"
 
            if algo_type in ["GaudiSequencer", "AthSequencer", "ProcessPhase"]:
                if algo_name in ["RecoITSeq", "RecoOTSeq", "RecoTTSeq"]:
                    continue
 
                if n not in self.unique_sequencers:
                    self.unique_sequencers.append(n)
                else:
                    if n not in self.dupl_seqs:
                        self.dupl_seqs[n] = 2
                    else:
                        self.dupl_seqs[n] += 1
 
                seq_daughter = Gaudi__Sequencer(algo_name, OutputLevel=INFO)
                if self.cfg.nodes[n].get("ModeOR") == "True":
                    self.OR_sequencers.append(n)
                    seq_daughter.ModeOR = True
                # if self.cfg.nodes[n].get('Lazy') == 'False':
                #    print "Non-Lazy - ", n
                seq_daughter.ShortCircuit = False
                if seq_daughter not in seq.Members:
                    seq.Members += [seq_daughter]
                    # iterate deeper
                    self._generate_sequence(n, seq_daughter)
            else:
                # rndname = ''.join(random.choice(string.lowercase) for i in range(5))
                # if algo_name in unique_algos: algo_name = algo_name + "-" + rndname
                if n not in self.unique_algos:
                    self.unique_algos.append(n)
                else:
                    if n not in self.dupl_algos:
                        self.dupl_algos[n] = 2
                    else:
                        self.dupl_algos[n] += 1
 
                avgRuntime, varRuntime = self.timeValue.get(algo_name)
 
                algo_daughter = CPUCruncher(
                    algo_name,
                    Cardinality=self.cardinality,
                    OutputLevel=self.outputLevel,
                    varRuntime=varRuntime,
                    avgRuntime=avgRuntime,
                    SleepFraction=self.sleepFraction
                    if self.BlockingBoolValue.get()
                    else 0.0,
                    Timeline=self.enableTimeline,
                )
 
                self._declare_data_deps(algo_name, algo_daughter)
 
                if algo_daughter not in seq.Members:
                    seq.Members += [algo_daughter]
 
        return seq