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 564 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 565 of file pTools.py.

00567                                                            :
00568         # Class to help synchronise the sub-processes
00569         self.limit  = limit
00570         self.step   = step
00571         self.d = {}
00572         self.manyEvents = manyEvents
00573         for i in xrange(-2, nWorkers) :
00574             self.d[ i ] = SyncMini( Event(), lastEvent=Event() )
00575             if self.manyEvents :
00576                 self.limitFirst = firstEvent
00577         self.keys       = self.d.keys()
00578         self.nWorkers   = nWorkers
00579         self.log        = log

Member Function Documentation

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

Definition at line 669 of file pTools.py.

00670                          :
00671         # Check the status of each Sync object
00672         # return True or False
00673         currentStatus = [ mini.check() for mini in self.d.values() ]
00674         return all( currentStatus )

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

Definition at line 675 of file pTools.py.

00676                                 :
00677        # check if all of the lastEvents are set to true in self.d[k][1]
00678        stat    = [ sMini.checkLast() for sMini in self.d.values() ]
00679        return all(stat)
00680 
00681 # =========================== Methods =========================================

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

Definition at line 663 of file pTools.py.

00664                             :
00665         self.log.critical('Some proc is hanging during Event processing!')
00666         for k in self.keys :
00667             self.log.critical( "Proc/Stat : %i / %s"%(k,self.d[k].check()) )
00668         return

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

Definition at line 580 of file pTools.py.

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

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

Definition at line 607 of file pTools.py.

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