Public Member Functions
def	__init__
def	syncAll
def	syncAllRolling
def	processHang
def	checkAll
def	checkLastEvents
Public Attributes
	limit
	step
	d
	manyEvents
	limitFirst
	keys
	nWorkers
	log

Detailed Description

Definition at line 555 of file pTools.py.

Constructor & Destructor Documentation

def GaudiMP::pTools::Syncer::__init__	(	self,
		nWorkers,
		log,
		manyEvents = `False`,
		limit = `None`,
		step = `None`,
		firstEvent = `None`
	)

Definition at line 556 of file pTools.py.

00558                                                            :
00559         # Class to help synchronise the sub-processes
00560         self.limit  = limit
00561         self.step   = step
00562         self.d = {}
00563         self.manyEvents = manyEvents
00564         for i in xrange(-2, nWorkers) :
00565             self.d[ i ] = SyncMini( Event(), lastEvent=Event() )
00566             if self.manyEvents :
00567                 self.limitFirst = firstEvent
00568         self.keys       = self.d.keys()
00569         self.nWorkers   = nWorkers
00570         self.log        = log

Member Function Documentation

def GaudiMP::pTools::Syncer::checkAll ( self )

Definition at line 660 of file pTools.py.

00661                          :
00662         # Check the status of each Sync object
00663         # return True or False
00664         currentStatus = [ mini.check() for mini in self.d.values() ]
00665         return all( currentStatus )

def GaudiMP::pTools::Syncer::checkLastEvents ( self )

Definition at line 666 of file pTools.py.

00667                                 :
00668        # check if all of the lastEvents are set to true in self.d[k][1]
00669        stat    = [ sMini.checkLast() for sMini in self.d.values() ]
00670        return all(stat)
00671 
00672 # =========================== Methods =========================================

def GaudiMP::pTools::Syncer::processHang ( self )

Definition at line 654 of file pTools.py.

00655                             :
00656         self.log.critical('Some proc is hanging during Event processing!')
00657         for k in self.keys :
00658             self.log.critical( "Proc/Stat : %i / %s"%(k,self.d[k].check()) )
00659         return

def GaudiMP::pTools::Syncer::syncAll	(	self,
		step = `"Not specified"`
	)

Definition at line 571 of file pTools.py.

00572                                               :
00573         # is it this method, or is it the rolling version needed?
00574         # if so, drop through...
00575 
00576         if self.manyEvents :
00577             sc = self.syncAllRolling( )
00578             return sc
00579 
00580         # Regular version ----------------------------
00581         for i in xrange( 0, self.limit, self.step ) :
00582             if self.checkAll( ) :
00583                 self.log.info('%s : All procs done @ %i s'%(step,i))
00584                 break
00585             else :
00586                 time.sleep(self.step)
00587 
00588         # Now the time limit is up... check the status one final time
00589         if self.checkAll() :
00590             self.log.info("All processes : %s ok."%(step))
00591             return SUCCESS
00592         else :
00593             self.log.critical('Some process is hanging on : %s'%(step))
00594             for k in self.keys :
00595                 hangString= "%s : Proc/Stat : %i/%s"%(step,k,self.d[k].check())
00596                 self.log.critical( hangString )
00597             return FAILURE

def GaudiMP::pTools::Syncer::syncAllRolling ( self )

Definition at line 598 of file pTools.py.

00599                                :
00600         # Keep track of the progress of Event processing
00601         # Each process syncs after each event, so keep clearing
00602         #  the sync Event, and re-checking
00603         # Note the time between True checks too, if the time
00604         #  between events exceeds singleEvent, this is considered a hang
00605 
00606         # set the initial time
00607         begin   = time.time()
00608         firstEv = {}
00609         timers  = {}
00610         for k in self.keys :
00611             self.d[k].reset()
00612             firstEv[k] = False
00613             timers[k]  = 0.0
00614 
00615         active = self.keys
00616         while True :
00617             # check the status of each sync object
00618             for k in active :
00619                 sMini = self.d[k]
00620                 # print sMini
00621                 if sMini.check() :
00622                     if sMini.checkLast() :
00623                         # if last Event set,then event loop finished
00624                         active.remove( k )
00625                         alive = time.time()-begin
00626                         self.log.info( "Audit : Node %i alive for %5.2f"\
00627                                        %(k,alive) )
00628                     else :
00629                         sMini.reset()
00630                 else :
00631                     # the event still has not been checked, how long is that?
00632                     # is it the first Event?
00633                     wait = time.time()-sMini.getTime()
00634                     cond = wait > self.limit
00635                     if not firstEv[k] :
00636                         cond       = wait > self.limitFirst
00637                         firstEv[k] = True
00638                     if cond :
00639                         # It is hanging!
00640                         self.log.critical('Single event wait : %5.2f'%(wait))
00641                         self.processHang()
00642                         return FAILURE
00643 
00644             # Termination Criteria : if all procs have been removed, we're done
00645             if self.checkLastEvents() :
00646                 self.log.info('TC met for event loop')
00647                 break
00648             else :
00649                 # sleep, loop again
00650                 time.sleep(self.step)
00651 
00652         self.log.info("All processes Completed all Events ok")
00653         return SUCCESS