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)

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

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

def	__init__ (self, nodeType, nodeID, queues, events, params, subworkers)

def	Start (self)

def	Engine (self)

def	processConfiguration (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

	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 940 of file GMPBase.py.

Constructor & Destructor Documentation

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

Definition at line 941 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
 

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

Definition at line 941 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

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

Definition at line 1123 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
 

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

Definition at line 1123 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
 

def GaudiMP.GMPBase.Worker.Engine ( self )

Definition at line 991 of file GMPBase.py.

     def Engine( self ) :
 
         # rename process
         import os
         import ctypes
         libc = ctypes.CDLL('libc.so.6')
         name = str(self.nodeType) + str(self.nodeID) + '\0'
         libc.prctl(15,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 != None:
                 self.cntr.setEventCounter( evtNumber )
 
             # subworkers are forked before the first event is processed
             # reader-thread for ConDB must be closed and reopened in each subworker
             # this is done by disconnect()
             if self.nIn == 0:
 
                 self.log.info("Fork new subworkers and disconnect from CondDB")
                 condDB = self.a.service('CondDBCnvSvc', gbl.ICondDBReader)
                 condDB.disconnect()
 
                 # 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()
 

def GaudiMP.GMPBase.Worker.Engine ( self )

Definition at line 991 of file GMPBase.py.

     def Engine( self ) :
 
         # rename process
         import os
         import ctypes
         libc = ctypes.CDLL('libc.so.6')
         name = str(self.nodeType) + str(self.nodeID) + '\0'
         libc.prctl(15,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 != None:
                 self.cntr.setEventCounter( evtNumber )
 
             # subworkers are forked before the first event is processed
             # reader-thread for ConDB must be closed and reopened in each subworker
             # this is done by disconnect()
             if self.nIn == 0:
 
                 self.log.info("Fork new subworkers and disconnect from CondDB")
                 condDB = self.a.service('CondDBCnvSvc', gbl.ICondDBReader)
                 condDB.disconnect()
 
                 # 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()
 

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

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

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 1130 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
 
 # =============================================================================
 

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 1130 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
 
 # =============================================================================
 

def GaudiMP.GMPBase.Worker.processConfiguration ( self )

Definition at line 951 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 = formatHead+'% F%18W%S%7W%R%T %0W%M'
 
         for key, lst in self.writerDict.iteritems() :
             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:
             # ignore errors when trying to change the configuration of the EvtCounter
             self.log.warning('Cannot configure EvtCounter')
 

def GaudiMP.GMPBase.Worker.processConfiguration ( self )

Definition at line 951 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 = formatHead+'% F%18W%S%7W%R%T %0W%M'
 
         for key, lst in self.writerDict.iteritems() :
             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:
             # ignore errors when trying to change the configuration of the EvtCounter
             self.log.warning('Cannot configure EvtCounter')
 

Member Data Documentation

GaudiMP.GMPBase.Worker.app

Definition at line 979 of file GMPBase.py.

GaudiMP.GMPBase.Worker.currentEvent

Definition at line 1085 of file GMPBase.py.

GaudiMP.GMPBase.Worker.eventOutput

Definition at line 949 of file GMPBase.py.

GaudiMP.GMPBase.Worker.filerecordsAgent

Definition at line 1003 of file GMPBase.py.

GaudiMP.GMPBase.Worker.firstEvTime

Definition at line 1067 of file GMPBase.py.

GaudiMP.GMPBase.Worker.nIn

Definition at line 1050 of file GMPBase.py.

GaudiMP.GMPBase.Worker.tTime

Definition at line 1100 of file GMPBase.py.

GaudiMP.GMPBase.Worker.vetoAlgs

Definition at line 946 of file GMPBase.py.

GaudiMP.GMPBase.Worker.writerDict

Definition at line 944 of file GMPBase.py.

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

InstallArea/x86_64-slc6-gcc48-opt/python/GaudiMP/GMPBase.py

Public Member Functions

Public Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation