BrunelScenarioForwardScheduler.py

Go to the documentation of this file.

from Gaudi.Configuration import *
from Configurables import GaudiExamplesCommonConf, CPUCruncher, HiveSlimEventLoopMgr, HiveWhiteBoard, ForwardSchedulerSvc

#-------------------------------------------------------------------------------
# Metaconfig

NUMBEROFEVENTS = 23
NUMBEROFEVENTSINFLIGHT = 7
NUMBEROFALGOSINFLIGHT = 11
NUMBEROFTHREADS = 2
CARDINALITY = 11
SCALE = .1
VERBOSITY = 5

NumberOfEvents = NUMBEROFEVENTS
NumberOfEventsInFlight = NUMBEROFEVENTSINFLIGHT
NumberOfAlgosInFlight = NUMBEROFALGOSINFLIGHT
NumberOfThreads = NUMBEROFTHREADS
Cardinality = CARDINALITY
Scale = SCALE
Verbosity = VERBOSITY


#-------------------------------------------------------------------------------


def load_brunel_scenario(filename):
    algs = {}
    timing = {}
    objs = []
    curr = None
    order = 0
    nodes = ('/Event', '/Event/Rec', '/Event/DAQ')
    for l in open(filename).readlines():
        if l.find('StoreTracer') == 0:
            if l.find('Executing Algorithm') != -1:
                alg = l.split()[-1]
                if alg not in algs.keys():
                    algs[alg] = (order, set(), set())
                curr = alg
                order += 1
            elif l.find('Done with Algorithm') != -1:
                curr = None
            elif l.find('[EventDataSvc]') != -1 and curr:
                obj = l.split()[-1]
                if obj in nodes:
                    continue
                if obj.find('/Event/') == 0:
                    obj = obj[7:]
                obj = obj.replace('/', '_')
                if obj not in objs:
                    objs.append(obj)
                talg = algs[curr]
                if l.find('RETRIEVE') != -1:
                    if obj not in talg[1]:
                        talg[1].add(obj)
                elif l.find('REGOBJ') != -1:
                    if obj not in talg[2]:
                        talg[2].add(obj)
        if l.find("TimingAuditor") != -1:
            algo = l.split()[2]  # .rstrip("|")
            index = 13
            if algo.endswith("|"):
                index = 12
                algo = algo.rstrip("|")
            if algo in algs.keys():
                timing[algo] = l.split()[index]
            else:
                for name in algs.keys():
                    if name.startswith(algo):
                        timing[name] = l.split()[index]

    all_inputs = set()
    all_outputs = set()
    all_algos = []
    all_algos_inputs = []

    # Scale all algo timings if needed
    if Scale != -1:
        for alg in timing.keys():
            old_timing = float(timing[alg])
            new_timing = old_timing * Scale
            # print "Algorithm %s: %f --> %f" %(alg, old_timing, new_timing)
            timing[alg] = new_timing

    for i, (alg, deps) in enumerate(algs.items()):
        if alg in ["PatPVOffline", "PrsADCs"]:
            continue
        if deps[1] or deps[2]:
            inputs = []
            inputs = [item for item in deps[1] if item not in (
                "DAQ_ODIN", "DAQ_RawEvent") and item not in deps[2]]
            outputs = [item for item in deps[2]]
            new_algo = CPUCruncher(alg,
                                   avgRuntime=float(timing[alg]),
                                   inpKeys=map(
                                       lambda s: "/Event/" + s, inputs),
                                   outKeys=map(
                                       lambda s: "/Event/" + s, outputs),
                                   # DataInputs=map( lambda s: "/Event/"+s, inputs),
                                   # DataOutputs=map( lambda s: "/Event/"+s, outputs),
                                   OutputLevel=6,
                                   shortCalib=True
                                   )

            for item in deps[1]:
                all_inputs.add(item)
            for item in deps[2]:
                all_outputs.add(item)
            all_algos.append(new_algo)
            all_algos_inputs.append(inputs)
    # look for the objects that haven't been provided within the job. Assume this needs to come via input
    new_algo = CPUCruncher("input",
                           avgRuntime=1,
                           inpKeys=[],
                           outKeys=map(
                               lambda s: "/Event/" + s, [item for item in all_inputs.difference(all_outputs)]),
                           OutputLevel=WARNING
                           )
    all_algos.append(new_algo)
    all_algos_inputs.append([])
    for algo in all_algos:
        algo.Cardinality = Cardinality
        OutputLevel = WARNING

    return all_algos, all_algos_inputs


# Set output level threshold 2=DEBUG, 3=INFO, 4=WARNING, 5=ERROR, 6=FATAL )
ms = MessageSvc()
ms.OutputLevel = Verbosity

crunchers, inputs = load_brunel_scenario("Brunel.TES.trace.log")

whiteboard = HiveWhiteBoard("EventDataSvc",
                            EventSlots=NumberOfEventsInFlight)

slimeventloopmgr = HiveSlimEventLoopMgr(OutputLevel=INFO)

scheduler = ForwardSchedulerSvc(MaxEventsInFlight=NumberOfEventsInFlight,
                                MaxAlgosInFlight=NumberOfAlgosInFlight,
                                ThreadPoolSize=NumberOfThreads,
                                OutputLevel=INFO
                                )

# And the Application Manager
app = ApplicationMgr()
app.TopAlg = crunchers
app.EvtSel = "NONE"  # do not use any event input
app.EvtMax = NumberOfEvents
app.EventLoop = slimeventloopmgr
app.ExtSvc = [whiteboard]
app.MessageSvcType = "InertMessageSvc"
app.OutputLevel = INFO