GaudiMP::pTools::Syncer Class Reference

List of all members.

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

Member Function Documentation

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

Definition at line 555 of file pTools.py.

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

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

Definition at line 659 of file pTools.py.

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

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

Definition at line 665 of file pTools.py.

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

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

Definition at line 653 of file pTools.py.

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

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

Definition at line 570 of file pTools.py.

00572 00573 00574 00575 00576 00577 00578 00579 00580 00581 00582 00583 00584 00585 00586 00587 00588 00589 00590 00591 00592 00593 00594 00595 00596

class="fragment">00571 : # is it this method, or is it the rolling version needed? # if so, drop through... if self.manyEvents : sc = self.syncAllRolling( ) return sc # Regular version ---------------------------- for i in xrange( 0, self.limit, self.step ) : if self.checkAll( ) : self.log.info('%s : All procs done @ %i s'%(step,i)) break else : time.sleep(self.step) # Now the time limit is up... check the status one final time if self.checkAll() : self.log.info("All processes : %s ok."%(step)) return SUCCESS else : self.log.critical('Some process is hanging on : %s'%(step)) for k in self.keys : hangString= "%s : Proc/Stat : %i/%s"%(step,k,self.d[k].check()) self.log.critical( hangString ) return FAILURE

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

Definition at line 597 of file pTools.py.

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