GaudiMP.GMPBase.Worker Class Reference

Inheritance diagram for GaudiMP.GMPBase.Worker:

Collaboration diagram for GaudiMP.GMPBase.Worker:

Public Member Functions
def	__init__ (self, workerID, queues, events, params, subworkers)
def	processConfiguration (self)
def	Engine (self)
def	getCheckAlgs (self)
def	checkExecutedPassed (self, algName)
def	isEventPassed (self)
Public Member Functions inherited from GaudiMP.GMPBase.GMPComponent
def	__init__ (self, nodeType, nodeID, queues, events, params, subworkers)
def	Start (self)
def	SetupGaudiPython (self)
def	StartGaudiPython (self)
def	LoadTES (self, tbufferfile)
def	getEventNumber (self)
def	IdentifyWriters (self)
def	dumpHistograms (self)
def	Initialize (self)
def	Finalize (self)
def	Report (self)

Public Attributes
	writerDict
	vetoAlgs
	eventOutput
	app
	filerecordsAgent
	nIn
	firstEvTime
	currentEvent
	tTime
Public Attributes inherited from GaudiMP.GMPBase.GMPComponent
	nodeType
	finalEvent
	lastEvent
	log
	subworkers
	nodeID
	msgFormat
	currentEvent
	queues
	num
	app
	evcom
	evcoms
	hq
	fq
	nIn
	nOut
	stat
	iTime
	rTime
	fTime
	firstEvTime
	tTime
	proc
	a
	evt
	hvt
	fsr
	inc
	pers
	ts
	TS
	cntr

Detailed Description

Definition at line 996 of file GMPBase.py.

Constructor & Destructor Documentation

__init__()

def GaudiMP.GMPBase.Worker.__init__	(		self,
			workerID,
			queues,
			events,
			params,
			subworkers
	)

Definition at line 997 of file GMPBase.py.

    def __init__(self, workerID, queues, events, params, subworkers):
        GMPComponent.__init__(
            self, "Worker", workerID, queues, events, params, subworkers
        )
        # Identify the writer streams
        self.writerDict = self.IdentifyWriters()
        # Identify the accept/veto checks for each event
        self.acceptAlgs, self.requireAlgs, self.vetoAlgs = self.getCheckAlgs()
        self.log.name = "Worker-%i" % (self.nodeID)
        self.log.info("Worker-%i Created OK" % (self.nodeID))
        self.eventOutput = True
 

Member Function Documentation

checkExecutedPassed()

def GaudiMP.GMPBase.Worker.checkExecutedPassed	(		self,
			algName
	)

Definition at line 1188 of file GMPBase.py.

    def checkExecutedPassed(self, algName):
        if (
            self.a.algorithm(algName)._ialg.isExecuted()
            and self.a.algorithm(algName)._ialg.filterPassed()
        ):
            return True
        else:
            return False
 

Engine()

def GaudiMP.GMPBase.Worker.Engine

(

self

)

Reimplemented from GaudiMP.GMPBase.GMPComponent.

Definition at line 1049 of file GMPBase.py.

    def Engine(self):
        # rename process
        import ctypes
 
        libc = ctypes.CDLL("libc.so.6")
        name = str(self.nodeType) + str(self.nodeID) + "\0"
        libc.prctl(15, ctypes.create_unicode_buffer(name), 0, 0, 0)
 
        startEngine = time.time()
        self.log.info("Worker %i starting Engine" % (self.nodeID))
        self.Initialize()
        self.filerecordsAgent = FileRecordsAgent(self)
 
        # populate the TESSerializer itemlist
        self.log.info("EVT WRITERS ON WORKER : %i" % (len(self.writerDict["events"])))
 
        nEventWriters = len(self.writerDict["events"])
        if nEventWriters:
            itemList = set()
            optItemList = set()
            for m in self.writerDict["events"]:
                for item in m.ItemList:
                    hsh = item.find("#")
                    if hsh != -1:
                        item = item[:hsh]
                    itemList.add(item)
                for item in m.OptItemList:
                    hsh = item.find("#")
                    if hsh != -1:
                        item = item[:hsh]
                    optItemList.add(item)
            # If an item is mandatory and optional, keep it only in the optional list
            itemList -= optItemList
            for item in sorted(itemList):
                self.log.info(" adding ItemList Item to ts : %s" % (item))
                self.ts.addItem(item)
            for item in sorted(optItemList):
                self.log.info(" adding Optional Item to ts : %s" % (item))
                self.ts.addOptItem(item)
        else:
            self.log.info("There is no Event Output for this app")
            self.eventOutput = False
 
        # Begin processing
        Go = True
        while Go:
            packet = self.evcom.receive()
            if packet:
                pass
            else:
                continue
            if packet == "FINISHED":
                break
            evtNumber, tbin = packet  # unpack
            if self.cntr != cppyy.nullptr:
                self.cntr.setEventCounter(evtNumber)
 
            # subworkers are forked before the first event is processed
            if self.nIn == 0:
                self.log.info("Fork new subworkers")
 
                # Fork subworkers and share services
                for k in self.subworkers:
                    k.SetServices(
                        self.a,
                        self.evt,
                        self.hvt,
                        self.fsr,
                        self.inc,
                        self.pers,
                        self.ts,
                        self.cntr,
                    )
                    k.Start()
                    self.a.addAlgorithm(CollectHistograms(self))
            self.nIn += 1
            self.TS.Load(tbin)
 
            t = time.time()
            sc = self.a.executeEvent()
            if self.nIn == 1:
                self.firstEvTime = time.time() - t
            else:
                self.rTime += time.time() - t
            if sc.isSuccess():
                pass
            else:
                self.log.warning("Did not Execute Event")
                self.evt.clearStore()
                continue
            if self.isEventPassed():
                pass
            else:
                self.log.warning("Event did not pass : %i" % (evtNumber))
                self.evt.clearStore()
                continue
            if self.eventOutput:
                # It may be the case of generating Event Tags; hence
                #   no event output
                self.currentEvent = self.getEventNumber()
                tb = self.TS.Dump()
                self.evcom.send((self.currentEvent, tb))
                self.nOut += 1
            self.inc.fireIncident(gbl.Incident("Worker", "EndEvent"))
            self.eventLoopSyncer.set()
            self.evt.clearStore()
        self.log.info("Setting <Last> Event")
        self.lastEvent.set()
 
        self.evcom.finalize()
        self.log.info("Worker-%i Finished Processing Events" % (self.nodeID))
        # Now send the FileRecords and stop/finalize the appMgr
        self.filerecordsAgent.SendFileRecords()
        self.Finalize()
        self.tTime = time.time() - startEngine
        self.Report()
 
        for k in self.subworkers:
            self.log.info("Join subworkers")
            k.proc.join()
 

getCheckAlgs()

def GaudiMP.GMPBase.Worker.getCheckAlgs

(

self

)

For some output writers, a check is performed to see if the event has
executed certain algorithms.
These reside in the AcceptAlgs property for those writers

Definition at line 1170 of file GMPBase.py.

    def getCheckAlgs(self):
        """
        For some output writers, a check is performed to see if the event has
        executed certain algorithms.
        These reside in the AcceptAlgs property for those writers
        """
        acc = []
        req = []
        vet = []
        for m in self.writerDict["events"]:
            if hasattr(m.w, "AcceptAlgs"):
                acc += m.w.AcceptAlgs
            if hasattr(m.w, "RequireAlgs"):
                req += m.w.RequireAlgs
            if hasattr(m.w, "VetoAlgs"):
                vet += m.w.VetoAlgs
        return (acc, req, vet)
 

isEventPassed()

def GaudiMP.GMPBase.Worker.isEventPassed

(

self

)

Check the algorithm status for an event.
Depending on output writer settings, the event
  may be declined based on various criteria.
This is a transcript of the check that occurs in GaudiSvc::OutputStream

Definition at line 1197 of file GMPBase.py.

    def isEventPassed(self):
        """
        Check the algorithm status for an event.
        Depending on output writer settings, the event
          may be declined based on various criteria.
        This is a transcript of the check that occurs in GaudiSvc::OutputStream
        """
        passed = False
 
        self.log.debug("self.acceptAlgs is %s" % (str(self.acceptAlgs)))
        if self.acceptAlgs:
            for name in self.acceptAlgs:
                if self.checkExecutedPassed(name):
                    passed = True
                    break
        else:
            passed = True
 
        self.log.debug("self.requireAlgs is %s" % (str(self.requireAlgs)))
        for name in self.requireAlgs:
            if self.checkExecutedPassed(name):
                pass
            else:
                self.log.info("Evt declined (requireAlgs) : %s" % (name))
                passed = False
 
        self.log.debug("self.vetoAlgs is %s" % (str(self.vetoAlgs)))
        for name in self.vetoAlgs:
            if self.checkExecutedPassed(name):
                pass
            else:
                self.log.info("Evt declined : (vetoAlgs) : %s" % (name))
                passed = False
        return passed
 
 
# =============================================================================
 
 

processConfiguration()

def GaudiMP.GMPBase.Worker.processConfiguration

(

self

)

Reimplemented from GaudiMP.GMPBase.GMPComponent.

Definition at line 1009 of file GMPBase.py.

    def processConfiguration(self):
        # Worker :
        #   No input
        #   No output
        #   No Histos
        self.config["EventSelector"].Input = []
        self.config["ApplicationMgr"].OutStream = []
        if "HistogramPersistencySvc" in self.config.keys():
            self.config["HistogramPersistencySvc"].OutputFile = ""
        formatHead = "[Worker-%i]" % (self.nodeID)
        self.config["MessageSvc"].Format = "%-13s " % formatHead + self.msgFormat
 
        for key, lst in self.writerDict.items():
            self.log.info("Writer Type : %s\t : %i" % (key, len(lst)))
 
        for m in self.writerDict["tuples"]:
            # rename Tuple output file with an appendix
            # based on worker id, for merging later
            newName = m.getNewName(".", ".w%i." % (self.nodeID))
            self.config[m.key].Output = newName
 
        # Suppress INFO Output for all but Worker-0
        # if self.nodeID == 0 :
        #    pass
        # else                :
        #    self.config[ 'MessageSvc' ].OutputLevel = ERROR
 
        if self.app == "Gauss":
            try:
                if "ToolSvc.EvtCounter" not in self.config:
                    from Configurables import EvtCounter
 
                    counter = EvtCounter()
                else:
                    counter = self.config["ToolSvc.EvtCounter"]
                counter.UseIncident = False
            except Exception:
                # ignore errors when trying to change the configuration of the EvtCounter
                self.log.warning("Cannot configure EvtCounter")
 

Member Data Documentation

app

GaudiMP.GMPBase.Worker.app

Definition at line 1036 of file GMPBase.py.

currentEvent

GaudiMP.GMPBase.Worker.currentEvent

Definition at line 1148 of file GMPBase.py.

eventOutput

GaudiMP.GMPBase.Worker.eventOutput

Definition at line 1007 of file GMPBase.py.

filerecordsAgent

GaudiMP.GMPBase.Worker.filerecordsAgent

Definition at line 1060 of file GMPBase.py.

firstEvTime

GaudiMP.GMPBase.Worker.firstEvTime

Definition at line 1130 of file GMPBase.py.

nIn

GaudiMP.GMPBase.Worker.nIn

Definition at line 1107 of file GMPBase.py.

tTime

GaudiMP.GMPBase.Worker.tTime

Definition at line 1163 of file GMPBase.py.

vetoAlgs

GaudiMP.GMPBase.Worker.vetoAlgs

Definition at line 1004 of file GMPBase.py.

writerDict

GaudiMP.GMPBase.Worker.writerDict

Definition at line 1002 of file GMPBase.py.

---

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

• GaudiMP/python/GaudiMP/GMPBase.py