GaudiMP.Parallel.WorkManager Class Reference

Inheritance diagram for GaudiMP.Parallel.WorkManager:

Collaboration diagram for GaudiMP.Parallel.WorkManager:

Public Member Functions
def	__init__

def	__del__

def	process

Public Attributes
	ncpus

	ppservers

	sessions

	server

	mode

	pool

	stats

Private Member Functions
def	_printStatistics

def	_mergeStatistics

Detailed Description

Class to in charge of managing the tasks and distributing them to
    the workers. They can be local (using other cores) or remote
    using other nodes in the local cluster

Definition at line 101 of file Parallel.py.

Constructor & Destructor Documentation

def GaudiMP.Parallel.WorkManager.__init__	(	self,
		ncpus = `'autodetect'`,
		ppservers = `None`
	)

Definition at line 106 of file Parallel.py.

 
     def __init__( self, ncpus='autodetect', ppservers=None) :
         if ncpus == 'autodetect' : self.ncpus = multiprocessing.cpu_count()
         else :                     self.ncpus = ncpus
         if ppservers :
             import pp
             self.ppservers = ppservers
             self.sessions = [ SshSession(srv) for srv in ppservers ]
             self.server = pp.Server(ncpus=self.ncpus, ppservers=self.ppservers)
             self.mode = 'cluster'
         else :
             self.pool = multiprocessing.Pool(self.ncpus)
             self.mode = 'multicore'
         self.stats = {}

Member Function Documentation

def GaudiMP.Parallel.WorkManager._mergeStatistics	(	self,
		stat
	)

private

Definition at line 157 of file Parallel.py.

 
     def _mergeStatistics(self, stat):
         if stat.name not in self.stats : self.stats[stat.name] = Statistics()
         s = self.stats[stat.name]
         s.time += stat.time
         s.njob += 1
 

def GaudiMP.Parallel.WorkManager._printStatistics ( self )

private

Definition at line 148 of file Parallel.py.

 
     def _printStatistics(self):
         njobs = 0
         for stat in self.stats.values():
             njobs += stat.njob
         print 'Job execution statistics:'
         print 'job count | % of all jobs | job time sum | time per job | job server'
         for name, stat  in self.stats.items():
             print '       %d |        %6.2f |     %8.3f |    %8.3f | %s' % (stat.njob, 100.*stat.njob/njobs, stat.time, stat.time/stat.njob, name)

def GaudiMP.Parallel.WorkManager.process	(	self,
		task,
		items,
		timeout = `90000`
	)

Definition at line 123 of file Parallel.py.

 
     def process(self, task, items, timeout=90000):
         if not isinstance(task,Task) :
             raise TypeError("task argument needs to be an 'Task' instance")
         # --- Call the Local initialialization
         task.initializeLocal()
         # --- Schedule all the jobs ....
         if self.mode == 'cluster' :
             jobs = [self.server.submit(_prefunction, (_ppfunction, task, item), (), ('GaudiMP.Parallel','time')) for item in items]
             for job in jobs :
                 result, stat = job()
                 task._mergeResults(result)
                 self._mergeStatistics(stat)
             self._printStatistics()
             self.server.print_stats()
         elif self.mode == 'multicore' :
             start = time.time()
             jobs = self.pool.map_async(_ppfunction, zip([task for i in items] , items ))
             for result, stat in  jobs.get(timeout) :
                 task._mergeResults(result)
                 self._mergeStatistics(stat)
             end = time.time()
             self._printStatistics()
             print 'Time elapsed since server creation %f' %(end-start)
         # --- Call the Local Finalize

        task.finalize()

Gaudi Framework, version v25r1

Public Member Functions

Public Attributes

Private Member Functions

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation