GaudiMP.GMPBase.Subworker Class Reference

Inheritance diagram for GaudiMP.GMPBase.Subworker:

Collaboration diagram for GaudiMP.GMPBase.Subworker:

Public Member Functions
	__init__ (self, workerID, queues, events, params, subworkers)
	Engine (self)
	SetServices (self, a, evt, hvt, fsr, inc, pers, ts, cntr)
	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)
	processConfiguration (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 825 of file GMPBase.py.

Constructor & Destructor Documentation

__init__()

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

Definition at line 826 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.info(f"Subworker-{self.nodeID} Created OK")
        self.eventOutput = True
 

Member Function Documentation

checkExecutedPassed()

GaudiMP.GMPBase.Subworker.checkExecutedPassed	(		self,
			algName )

Definition at line 948 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.Subworker.Engine

(

self

)

Reimplemented from GaudiMP.GMPBase.GMPComponent.

Definition at line 837 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)
 
        self.initEvent.set()
        startEngine = time.time()
        msg = self.a.service("MessageSvc")
        msg.Format = "{:<13s} ".format(f"[{self.log.name}]") + self.msgFormat
 
        self.log.name = f"Worker-{self.nodeID}"
        self.log.info(f"Subworker {self.nodeID} starting Engine")
        self.filerecordsAgent = FileRecordsAgent(self)
 
        # populate the TESSerializer itemlist
        self.log.info(f"EVT WRITERS ON WORKER : {len(self.writerDict['events'])}")
 
        self.a.addAlgorithm(CollectHistograms(self))
 
        # 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)
 
            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.name = f"Worker-{self.nodeID}"
                self.log.warning("Did not Execute Event")
                self.evt.clearStore()
                continue
            if self.isEventPassed():
                pass
            else:
                self.log.name = f"Worker-{self.nodeID}"
                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("Subworker", "EndEvent"))
            self.eventLoopSyncer.set()
            self.evt.clearStore()
        self.log.name = f"Worker-{self.nodeID}"
        self.log.info(f"Setting <Last> Event {self.nodeID}")
        self.lastEvent.set()
 
        self.evcom.finalize()
        # Now send the FileRecords and stop/finalize the appMgr
        self.filerecordsAgent.SendFileRecords()
        self.tTime = time.time() - startEngine
        self.Finalize()
        self.Report()
        # self.finalEvent.set()
 

getCheckAlgs()

GaudiMP.GMPBase.Subworker.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 930 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.Subworker.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 957 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
 
 
# =============================================================================
 
 

SetServices()

GaudiMP.GMPBase.Subworker.SetServices	(		self,
			a,
			evt,
			hvt,
			fsr,
			inc,
			pers,
			ts,
			cntr )

Definition at line 918 of file GMPBase.py.

    def SetServices(self, a, evt, hvt, fsr, inc, pers, ts, cntr):
        self.a = a
        self.evt = evt
        self.hvt = hvt
        self.fsr = fsr
        # self.inc = inc
        self.inc = self.a.service("IncidentSvc", "IIncidentSvc")
        self.pers = pers
        self.ts = ts
        self.cntr = cntr
        self.TS = TESSerializer(self.ts, self.evt, self.nodeType, self.nodeID, self.log)
 

Member Data Documentation

acceptAlgs

GaudiMP.GMPBase.Subworker.acceptAlgs

Definition at line 833 of file GMPBase.py.

eventOutput

bool GaudiMP.GMPBase.Subworker.eventOutput = True

Definition at line 835 of file GMPBase.py.

filerecordsAgent

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

Definition at line 852 of file GMPBase.py.

requireAlgs

GaudiMP.GMPBase.Subworker.requireAlgs

Definition at line 833 of file GMPBase.py.

vetoAlgs

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

Definition at line 833 of file GMPBase.py.

writerDict

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

Definition at line 831 of file GMPBase.py.

---

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

• GaudiMP/python/GaudiMP/GMPBase.py