Inheritance diagram for GaudiMP.GMPBase.Coord:

Collaboration diagram for GaudiMP.GMPBase.Coord:

Public Member Functions
def	__init__ (self, nWorkers, config, log)

def	getSyncEvents (self, nodeID)

def	getQueues (self, nodeID)

def	Go (self)

def	Terminate (self)

def	Stop (self)

def	SetupQueues (self)

def	__init__ (self, nWorkers, config, log)

def	getSyncEvents (self, nodeID)

def	getQueues (self, nodeID)

def	Go (self)

def	Terminate (self)

def	Stop (self)

def	SetupQueues (self)

Public Attributes
	log

	config

	nWorkers

	qs

	hq

	fq

	sInit

	sRun

	sFin

	histSyncEvent

	subworkers

	reader

	workers

	writer

	system

Detailed Description

Definition at line 1286 of file GMPBase.py.

Constructor & Destructor Documentation

def GaudiMP.GMPBase.Coord.__init__	(	self,
		nWorkers,
		config,
		log
	)

Definition at line 1287 of file GMPBase.py.

     def __init__( self, nWorkers, config, log ) :
 
         self.log = log
         self.config = config
         # set up Logging
         self.log.name = 'GaudiPython-Parallel-Logger'
         self.log.info( 'GaudiPython Parallel Process Co-ordinator beginning' )
 
         if nWorkers == -1 :
             # The user has requested all available cpus in the machine
             self.nWorkers = cpu_count()
         else :
             self.nWorkers = nWorkers
 
 
         self.qs = self.SetupQueues( )    # a dictionary of queues (for Events)
         self.hq = JoinableQueue( )       # for Histogram data
         self.fq = JoinableQueue( )       # for FileRecords data
 
         # Make a Syncer for Initalise, Run, and Finalise
         self.sInit = Syncer( self.nWorkers, self.log,
                              limit=WAIT_INITIALISE,
                              step=STEP_INITIALISE    )
         self.sRun  = Syncer( self.nWorkers, self.log,
                              manyEvents=True,
                              limit=WAIT_SINGLE_EVENT,
                              step=STEP_EVENT,
                              firstEvent=WAIT_FIRST_EVENT )
         self.sFin  = Syncer( self.nWorkers, self.log,
                              limit=WAIT_FINALISE,
                              step=STEP_FINALISE )
         # and one final one for Histogram Transfer
         self.histSyncEvent = Event()
 
         # params are common to al subprocesses
         params = (self.nWorkers, self.histSyncEvent, self.config, self.log)
 
         self.subworkers = []
         # Declare SubProcesses!
         for i in range(1, self.nWorkers ) :
             sub = Subworker( i, self.getQueues(i), self.getSyncEvents(i), params, self.subworkers )
             self.subworkers.append( sub )
         self.reader= Reader(self.getQueues(-1), self.getSyncEvents(-1), params, self.subworkers)
         self.workers = []
         wk = Worker( 0, self.getQueues(0), self.getSyncEvents(0), params, self.subworkers )
         self.writer= Writer(self.getQueues(-2), self.getSyncEvents(-2), params, self.subworkers)
 
         self.system = []
         self.system.append(self.writer)
         self.system.append(wk)
         self.system.append(self.reader)
 

def GaudiMP.GMPBase.Coord.__init__	(	self,
		nWorkers,
		config,
		log
	)

Definition at line 1287 of file GMPBase.py.

     def __init__( self, nWorkers, config, log ) :
 
         self.log = log
         self.config = config
         # set up Logging
         self.log.name = 'GaudiPython-Parallel-Logger'
         self.log.info( 'GaudiPython Parallel Process Co-ordinator beginning' )
 
         if nWorkers == -1 :
             # The user has requested all available cpus in the machine
             self.nWorkers = cpu_count()
         else :
             self.nWorkers = nWorkers
 
 
         self.qs = self.SetupQueues( )    # a dictionary of queues (for Events)
         self.hq = JoinableQueue( )       # for Histogram data
         self.fq = JoinableQueue( )       # for FileRecords data
 
         # Make a Syncer for Initalise, Run, and Finalise
         self.sInit = Syncer( self.nWorkers, self.log,
                              limit=WAIT_INITIALISE,
                              step=STEP_INITIALISE    )
         self.sRun  = Syncer( self.nWorkers, self.log,
                              manyEvents=True,
                              limit=WAIT_SINGLE_EVENT,
                              step=STEP_EVENT,
                              firstEvent=WAIT_FIRST_EVENT )
         self.sFin  = Syncer( self.nWorkers, self.log,
                              limit=WAIT_FINALISE,
                              step=STEP_FINALISE )
         # and one final one for Histogram Transfer
         self.histSyncEvent = Event()
 
         # params are common to al subprocesses
         params = (self.nWorkers, self.histSyncEvent, self.config, self.log)
 
         self.subworkers = []
         # Declare SubProcesses!
         for i in range(1, self.nWorkers ) :
             sub = Subworker( i, self.getQueues(i), self.getSyncEvents(i), params, self.subworkers )
             self.subworkers.append( sub )
         self.reader= Reader(self.getQueues(-1), self.getSyncEvents(-1), params, self.subworkers)
         self.workers = []
         wk = Worker( 0, self.getQueues(0), self.getSyncEvents(0), params, self.subworkers )
         self.writer= Writer(self.getQueues(-2), self.getSyncEvents(-2), params, self.subworkers)
 
         self.system = []
         self.system.append(self.writer)
         self.system.append(wk)
         self.system.append(self.reader)
 

Member Function Documentation

def GaudiMP.GMPBase.Coord.getQueues	(	self,
		nodeID
	)

Definition at line 1345 of file GMPBase.py.

     def getQueues( self, nodeID ) :
         eventQ = self.qs[ nodeID ]
         histQ  = self.hq
         fsrQ   = self.fq
         return ( eventQ, histQ, fsrQ )
 

def GaudiMP.GMPBase.Coord.getQueues	(	self,
		nodeID
	)

Definition at line 1345 of file GMPBase.py.

     def getQueues( self, nodeID ) :
         eventQ = self.qs[ nodeID ]
         histQ  = self.hq
         fsrQ   = self.fq
         return ( eventQ, histQ, fsrQ )
 

def GaudiMP.GMPBase.Coord.getSyncEvents	(	self,
		nodeID
	)

Definition at line 1339 of file GMPBase.py.

     def getSyncEvents( self, nodeID ) :
         init = self.sInit.d[nodeID].event
         run  = ( self.sRun.d[nodeID].event, self.sRun.d[nodeID].lastEvent )
         fin  = self.sFin.d[nodeID].event
         return ( init, run, fin )
 

def GaudiMP.GMPBase.Coord.getSyncEvents	(	self,
		nodeID
	)

Definition at line 1339 of file GMPBase.py.

     def getSyncEvents( self, nodeID ) :
         init = self.sInit.d[nodeID].event
         run  = ( self.sRun.d[nodeID].event, self.sRun.d[nodeID].lastEvent )
         fin  = self.sFin.d[nodeID].event
         return ( init, run, fin )
 

def GaudiMP.GMPBase.Coord.Go ( self )

Definition at line 1351 of file GMPBase.py.

     def Go( self ) :
 
         # Initialise
         self.log.name = 'GaudiPython-Parallel-Logger'
         self.log.info( 'INITIALISING SYSTEM' )
 
         # Start reader, writer and main worker
         for p in self.system :
             p.Start()
 
         sc = self.sInit.syncAll(step="Initialise")
         if sc == SUCCESS: pass
         else  : self.Terminate() ; return FAILURE
 
         # Run
         self.log.name = 'GaudiPython-Parallel-Logger'
         self.log.info( 'RUNNING SYSTEM' )
         sc = self.sRun.syncAll(step="Run")
         if sc == SUCCESS: pass
         else  : self.Terminate() ; return FAILURE
 
         # Finalise
         self.log.name = 'GaudiPython-Parallel-Logger'
         self.log.info( 'FINALISING SYSTEM' )
         sc = self.sFin.syncAll(step="Finalise")
         if sc == SUCCESS: pass
         else  : self.Terminate() ; return FAILURE
 
         # if we've got this far, finally report SUCCESS
         self.log.info( "Cleanly join all Processes" )
         self.Stop()
         self.log.info( "Report Total Success to Main.py" )
         return SUCCESS
 

def GaudiMP.GMPBase.Coord.Go ( self )

Definition at line 1351 of file GMPBase.py.

     def Go( self ) :
 
         # Initialise
         self.log.name = 'GaudiPython-Parallel-Logger'
         self.log.info( 'INITIALISING SYSTEM' )
 
         # Start reader, writer and main worker
         for p in self.system :
             p.Start()
 
         sc = self.sInit.syncAll(step="Initialise")
         if sc == SUCCESS: pass
         else  : self.Terminate() ; return FAILURE
 
         # Run
         self.log.name = 'GaudiPython-Parallel-Logger'
         self.log.info( 'RUNNING SYSTEM' )
         sc = self.sRun.syncAll(step="Run")
         if sc == SUCCESS: pass
         else  : self.Terminate() ; return FAILURE
 
         # Finalise
         self.log.name = 'GaudiPython-Parallel-Logger'
         self.log.info( 'FINALISING SYSTEM' )
         sc = self.sFin.syncAll(step="Finalise")
         if sc == SUCCESS: pass
         else  : self.Terminate() ; return FAILURE
 
         # if we've got this far, finally report SUCCESS
         self.log.info( "Cleanly join all Processes" )
         self.Stop()
         self.log.info( "Report Total Success to Main.py" )
         return SUCCESS
 

def GaudiMP.GMPBase.Coord.SetupQueues ( self )

Definition at line 1402 of file GMPBase.py.

     def SetupQueues( self ) :
         # This method will set up the network of Queues needed
         # N Queues = nWorkers + 1
         # Each Worker has a Queue in, and a Queue out
         # Reader has Queue out only
         # Writer has nWorkers Queues in
 
         # one queue from Reader-Workers
         rwk = JoinableQueue()
         # one queue from each worker to writer
         workersWriter = [ JoinableQueue() for i in xrange(self.nWorkers) ]
         d = {}
         d[-1] = (None, rwk)              # Reader
         d[-2] = (workersWriter, None)    # Writer
         for i in xrange(self.nWorkers) : d[i] =  (rwk, workersWriter[i])
         return d
 
 # ============================= EOF ===========================================

def GaudiMP.GMPBase.Coord.SetupQueues ( self )

Definition at line 1402 of file GMPBase.py.

     def SetupQueues( self ) :
         # This method will set up the network of Queues needed
         # N Queues = nWorkers + 1
         # Each Worker has a Queue in, and a Queue out
         # Reader has Queue out only
         # Writer has nWorkers Queues in
 
         # one queue from Reader-Workers
         rwk = JoinableQueue()
         # one queue from each worker to writer
         workersWriter = [ JoinableQueue() for i in xrange(self.nWorkers) ]
         d = {}
         d[-1] = (None, rwk)              # Reader
         d[-2] = (workersWriter, None)    # Writer
         for i in xrange(self.nWorkers) : d[i] =  (rwk, workersWriter[i])
         return d
 
 # ============================= EOF ===========================================

def GaudiMP.GMPBase.Coord.Stop ( self )

Definition at line 1395 of file GMPBase.py.

     def Stop( self ) :
         # procs should be joined in reverse order to launch
         self.system.reverse()
         for s in self.system :
             s.proc.join()
         return SUCCESS
 

def GaudiMP.GMPBase.Coord.Stop ( self )

Definition at line 1395 of file GMPBase.py.

     def Stop( self ) :
         # procs should be joined in reverse order to launch
         self.system.reverse()
         for s in self.system :
             s.proc.join()
         return SUCCESS
 

def GaudiMP.GMPBase.Coord.Terminate ( self )

Definition at line 1385 of file GMPBase.py.

     def Terminate( self ) :
         # Brutally kill sub-processes
         children = multiprocessing.active_children()
         for i in children:
             i.terminate()
 
         #self.writer.proc.terminate()
         #[ w.proc.terminate() for w in self.workers]
         #self.reader.proc.terminate()
 

def GaudiMP.GMPBase.Coord.Terminate ( self )

Definition at line 1385 of file GMPBase.py.

     def Terminate( self ) :
         # Brutally kill sub-processes
         children = multiprocessing.active_children()
         for i in children:
             i.terminate()
 
         #self.writer.proc.terminate()
         #[ w.proc.terminate() for w in self.workers]
         #self.reader.proc.terminate()
 

Member Data Documentation

GaudiMP.GMPBase.Coord.config

Definition at line 1290 of file GMPBase.py.

GaudiMP.GMPBase.Coord.fq

Definition at line 1304 of file GMPBase.py.

GaudiMP.GMPBase.Coord.histSyncEvent

Definition at line 1319 of file GMPBase.py.

GaudiMP.GMPBase.Coord.hq

Definition at line 1303 of file GMPBase.py.

GaudiMP.GMPBase.Coord.log

Definition at line 1289 of file GMPBase.py.

GaudiMP.GMPBase.Coord.nWorkers

Definition at line 1297 of file GMPBase.py.

GaudiMP.GMPBase.Coord.qs

Definition at line 1302 of file GMPBase.py.

GaudiMP.GMPBase.Coord.reader

Definition at line 1329 of file GMPBase.py.

GaudiMP.GMPBase.Coord.sFin

Definition at line 1315 of file GMPBase.py.

GaudiMP.GMPBase.Coord.sInit

Definition at line 1307 of file GMPBase.py.

GaudiMP.GMPBase.Coord.sRun

Definition at line 1310 of file GMPBase.py.

GaudiMP.GMPBase.Coord.subworkers

Definition at line 1324 of file GMPBase.py.

GaudiMP.GMPBase.Coord.system

Definition at line 1334 of file GMPBase.py.

GaudiMP.GMPBase.Coord.workers

Definition at line 1330 of file GMPBase.py.

GaudiMP.GMPBase.Coord.writer

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

GaudiMP
GMPBase
Coord
Generated on Tue Jul 21 2015 12:24:33 for The Gaudi Framework by 1.8.9.1

Public Member Functions

Public Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation