GaudiMP.GMPBase.Worker Class Reference

Inheritance diagram for GaudiMP.GMPBase.Worker:

Collaboration diagram for GaudiMP.GMPBase.Worker:

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

Public Attributes
	writerDict = self.IdentifyWriters()
	acceptAlgs
	requireAlgs
	vetoAlgs = self.getCheckAlgs()
bool	eventOutput = True
	filerecordsAgent = FileRecordsAgent(self)
Public Attributes inherited from GaudiMP.GMPBase.GMPComponent
str	nodeType = nodeType
	initEvent
	finalEvent = events
	eventLoopSyncer
	lastEvent = eventLoopSyncer
	nWorkers
	sEvent
	config
	log = params
	subworkers = subworkers
int	nodeID = nodeID
	msgFormat = self.config["MessageSvc"].Format
	currentEvent = None
	queues = queues
int	num = 0
str	app = None
	evcom = EventCommunicator(self, qin, qout)
list	evcoms = []
	hq = histq
	fq = fq
int	nIn = 0
int	nOut = 0
	stat = SUCCESS
float	iTime = 0.0
float	rTime = 0.0
float	fTime = 0.0
float	firstEvTime = 0.0
float	tTime = 0.0
	proc = Process(target=self.Engine)
	a = AppMgr()
	evt = self.a.evtsvc()
	hvt = self.a.histsvc()
	fsr = self.a.filerecordsvc()
	inc = self.a.service("IncidentSvc", "IIncidentSvc")
	pers = self.a.service("EventPersistencySvc", "IAddressCreator")
	ts = gbl.GaudiMP.TESSerializer(self.evt._idp, self.pers)
	TS = TESSerializer(self.ts, self.evt, self.nodeType, self.nodeID, self.log)
	cntr = InterfaceCast(gbl.IEventCounter)(tool.getInterface())

Detailed Description

Definition at line 996 of file GMPBase.py.

Constructor & Destructor Documentation

__init__()

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 = f"Worker-{self.nodeID}"
        self.log.info(f"Worker-{self.nodeID} Created OK")
        self.eventOutput = True
 

Member Function Documentation

checkExecutedPassed()

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()

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(f"Worker {self.nodeID} starting Engine")
        self.Initialize()
        self.filerecordsAgent = FileRecordsAgent(self)
 
        # populate the TESSerializer itemlist
        self.log.info(f"EVT WRITERS ON WORKER : {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(f" adding ItemList Item to ts : {item}")
                self.ts.addItem(item)
            for item in sorted(optItemList):
                self.log.info(f" adding Optional Item to ts : {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(f"Event did not pass : {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(f"Worker-{self.nodeID} Finished Processing Events")
        # 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()

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()

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(f"self.acceptAlgs is {self.acceptAlgs}")
        if self.acceptAlgs:
            for name in self.acceptAlgs:
                if self.checkExecutedPassed(name):
                    passed = True
                    break
        else:
            passed = True
 
        self.log.debug(f"self.requireAlgs is {self.requireAlgs}")
        for name in self.requireAlgs:
            if self.checkExecutedPassed(name):
                pass
            else:
                self.log.info(f"Evt declined (requireAlgs) : {name}")
                passed = False
 
        self.log.debug(f"self.vetoAlgs is {self.vetoAlgs}")
        for name in self.vetoAlgs:
            if self.checkExecutedPassed(name):
                pass
            else:
                self.log.info(f"Evt declined : (vetoAlgs) : {name}")
                passed = False
        return passed
 
 
# =============================================================================
 
 

processConfiguration()

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 = f"[Worker-{self.nodeID}]"
        self.config["MessageSvc"].Format = f"{formatHead:<13s} " + self.msgFormat
 
        for key, lst in self.writerDict.items():
            self.log.info(f"Writer Type : {key}\t : {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(".", f".w{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

acceptAlgs

GaudiMP.GMPBase.Worker.acceptAlgs

Definition at line 1004 of file GMPBase.py.

eventOutput

bool GaudiMP.GMPBase.Worker.eventOutput = True

Definition at line 1007 of file GMPBase.py.

filerecordsAgent

GaudiMP.GMPBase.Worker.filerecordsAgent = FileRecordsAgent(self)

Definition at line 1060 of file GMPBase.py.

requireAlgs

GaudiMP.GMPBase.Worker.requireAlgs

Definition at line 1004 of file GMPBase.py.

vetoAlgs

GaudiMP.GMPBase.Worker.vetoAlgs = self.getCheckAlgs()

Definition at line 1004 of file GMPBase.py.

writerDict

GaudiMP.GMPBase.Worker.writerDict = self.IdentifyWriters()

Definition at line 1002 of file GMPBase.py.

---

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

• GaudiMP/python/GaudiMP/GMPBase.py