Inheritance diagram for GaudiMP::GMPBase::Worker:

Collaboration diagram for GaudiMP::GMPBase::Worker:

Public Member Functions
def	__init__
def	processConfiguration
def	Engine
def	getCheckAlgs
def	checkExecutedPassed
def	isEventPassed
Public Attributes
	writerDict
	vetoAlgs
	eventOutput
	nodeID
	filerecordsAgent
	nIn
	firstEvTime
	currentEvent
	tTime

Detailed Description

Definition at line 732 of file GMPBase.py.

Constructor & Destructor Documentation

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

Definition at line 733 of file GMPBase.py.

00734                                                            :
00735         GMPComponent.__init__(self,'Worker', workerID, queues, events, params)
00736         # Identify the writer streams
00737         self.writerDict = self.IdentifyWriters( )
00738         # Identify the accept/veto checks for each event
00739         self.acceptAlgs, self.requireAlgs, self.vetoAlgs = self.getCheckAlgs()
00740         self.log.debug("Worker-%i Created OK"%(self.nodeID))
00741         self.eventOutput = True

Member Function Documentation

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

Definition at line 863 of file GMPBase.py.

00864                                              :
00865         if  self.a.algorithm( algName )._ialg.isExecuted()\
00866         and self.a.algorithm( algName )._ialg.filterPassed() :
00867             return True
00868         else :
00869             return False

def GaudiMP::GMPBase::Worker::Engine ( self )

Reimplemented from GaudiMP::GMPBase::GMPComponent.

Definition at line 769 of file GMPBase.py.

00770                        :
00771         startEngine = time.time()
00772         self.log.name = "Worker-%i"%(self.nodeID)
00773         self.log.info("Worker %i starting Engine"%(self.nodeID))
00774         self.Initialize()
00775         self.filerecordsAgent = FileRecordsAgent(self)
00776 
00777         # populate the TESSerializer itemlist
00778         self.log.info('EVT WRITERS ON WORKER : %i'\
00779                        %( len(self.writerDict['events'])))
00780 
00781         nEventWriters = len( self.writerDict[ "events" ] )
00782         if nEventWriters :
00783             for m in self.writerDict[ "events" ] :
00784                 for item in m.ItemList :
00785                     self.log.debug(' adding ItemList Item to ts : %s'%(item))
00786                     self.ts.addItem( item )
00787                 for item in m.OptItemList :
00788                     self.log.debug(' adding Optional Item to ts : %s'%(item))
00789                     self.ts.addOptItem( item )
00790         else :
00791             self.log.info( 'There is no Event Output for this app' )
00792             self.eventOutput = False
00793 
00794         # add the Histogram Collection Algorithm
00795         self.a.addAlgorithm( CollectHistograms(self) )
00796 
00797         # Begin processing
00798         self.log.name = "Worker-%i"%(self.nodeID)
00799         Go = True
00800         while Go :
00801             packet = self.evcom.receive( )
00802             if packet : pass
00803             else      : continue
00804             if packet == 'FINISHED' : break
00805             evtNumber, tbin = packet    # unpack
00806             self.nIn += 1
00807             self.TS.Load( tbin )
00808             # print 'Worker-%i : Event %i'%(self.nodeID, evtNumber)
00809             t = time.time()
00810             sc = self.a.executeEvent()
00811             if self.nIn == 1 :
00812                 self.firstEvTime = time.time()-t
00813             else :
00814                 self.rTime += (time.time()-t)
00815             if sc.isSuccess() :
00816                 pass
00817             else :
00818                 self.log.warning('Did not Execute Event')
00819                 self.evt.clearStore()
00820                 continue
00821             if self.isEventPassed() :
00822                 pass
00823             else :
00824                 self.log.warning( 'Event did not pass : %i'%(evtNumber) )
00825                 self.evt.clearStore()
00826                 continue
00827             if self.eventOutput :
00828                 # It may be the case of generating Event Tags; hence
00829                 #   no event output
00830                 self.currentEvent = self.getEventNumber( )
00831                 tb = self.TS.Dump( )
00832                 self.evcom.send( (self.currentEvent, tb) )
00833                 self.nOut += 1
00834             self.inc.fireIncident(gbl.Incident('Worker','EndEvent'))
00835             self.eventLoopSyncer.set()
00836             self.evt.clearStore( )
00837         self.log.info('Setting <Last> Event')
00838         self.lastEvent.set()
00839 
00840         self.evcom.finalize()
00841         self.log.info( 'Worker-%i Finished Processing Events'%(self.nodeID) )
00842         # Now send the FileRecords and stop/finalize the appMgr
00843         self.filerecordsAgent.SendFileRecords()
00844         self.Finalize()
00845         self.tTime = time.time()-startEngine
00846         self.Report()

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

00848                              :
00849         '''
00850         For some output writers, a check is performed to see if the event has
00851         executed certain algorithms.
00852         These reside in the AcceptAlgs property for those writers
00853         '''
00854         acc = []
00855         req = []
00856         vet = []
00857         for m in self.writerDict[ "events" ] :
00858             if hasattr(m.w, 'AcceptAlgs')  : acc += m.w.AcceptAlgs
00859             if hasattr(m.w, 'RequireAlgs') : req += m.w.RequireAlgs
00860             if hasattr(m.w, 'VetoAlgs')    : vet += m.w.VetoAlgs
00861         return (acc, req, vet)
00862

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

00871                               :
00872         '''
00873         Check the algorithm status for an event.
00874         Depending on output writer settings, the event
00875           may be declined based on various criteria.
00876         This is a transcript of the check that occurs in GaudiSvc::OutputStream
00877         '''
00878         passed = False
00879 
00880         self.log.debug('self.acceptAlgs is %s'%(str(self.acceptAlgs)))
00881         if self.acceptAlgs :
00882             for name in self.acceptAlgs :
00883                 if self.checkExecutedPassed( name ) :
00884                     passed = True
00885                     break
00886         else :
00887             passed = True
00888 
00889         self.log.debug('self.requireAlgs is %s'%(str(self.requireAlgs)))
00890         for name in self.requireAlgs :
00891             if self.checkExecutedPassed( name ) :
00892                 pass
00893             else :
00894                 self.log.info('Evt declined (requireAlgs) : %s'%(name) )
00895                 passed = False
00896 
00897         self.log.debug('self.vetoAlgs is %s'%(str(self.vetoAlgs)))
00898         for name in self.vetoAlgs :
00899             if self.checkExecutedPassed( name ) :
00900                 pass
00901             else :
00902                 self.log.info( 'Evt declined : (vetoAlgs) : %s'%(name) )
00903                 passed = False
00904         return passed
00905 
00906 # =============================================================================

def GaudiMP::GMPBase::Worker::processConfiguration ( self )

Reimplemented from GaudiMP::GMPBase::GMPComponent.

Definition at line 742 of file GMPBase.py.

00743                                      :
00744         # Worker :
00745         #   No input
00746         #   No output
00747         #   No Histos
00748         self.config[ 'EventSelector'  ].Input     = []
00749         self.config[ 'ApplicationMgr' ].OutStream = []
00750         if "HistogramPersistencySvc" in self.config.keys() :
00751             self.config[ 'HistogramPersistencySvc' ].OutputFile = ''
00752         formatHead = '[Worker-%i] '%(self.nodeID)
00753         self.config['MessageSvc'].Format = formatHead+'% F%18W%S%7W%R%T %0W%M'
00754 
00755         for key, lst in self.writerDict.iteritems() :
00756             self.log.info( 'Writer Type : %s\t : %i'%(key, len(lst)) )
00757 
00758         for m in self.writerDict[ "tuples" ] :
00759             # rename Tuple output file with an appendix
00760             # based on worker id, for merging later
00761             newName = m.getNewName( '.', '.w%i.'%(self.nodeID) )
00762             self.config[ m.key ].Output = newName
00763 
00764         # Suppress INFO Output for all but Worker-0
00765         if self.nodeID == 0 :
00766             pass
00767         else                :
00768             self.config[ 'MessageSvc' ].OutputLevel = ERROR