GaudiMP.Parallel.WorkManager Class Reference

Inheritance diagram for GaudiMP.Parallel.WorkManager:

Collaboration diagram for GaudiMP.Parallel.WorkManager:

Public Member Functions
def	__init__
def	__del__ (self)
def	process
def	__init__
def	__del__ (self)
def	process

Public Attributes
	ncpus
	ppservers
	sessions
	server
	mode
	pool
	stats

Private Member Functions
def	_printStatistics (self)
def	_mergeStatistics (self, stat)
def	_printStatistics (self)
def	_mergeStatistics (self, stat)

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 = {}

def GaudiMP.Parallel.WorkManager.__del__

(

self

)

Definition at line 120 of file Parallel.py.

    def __del__(self):
        if hasattr(self,'server') : self.server.destroy()

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 = {}

def GaudiMP.Parallel.WorkManager.__del__

(

self

)

Definition at line 120 of file Parallel.py.

    def __del__(self):
        if hasattr(self,'server') : self.server.destroy()

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._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._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()

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()

Member Data Documentation

GaudiMP.Parallel.WorkManager.mode

Definition at line 114 of file Parallel.py.

GaudiMP.Parallel.WorkManager.ncpus

Definition at line 107 of file Parallel.py.

GaudiMP.Parallel.WorkManager.pool

Definition at line 116 of file Parallel.py.

GaudiMP.Parallel.WorkManager.ppservers

Definition at line 111 of file Parallel.py.

GaudiMP.Parallel.WorkManager.server

Definition at line 113 of file Parallel.py.

GaudiMP.Parallel.WorkManager.sessions

Definition at line 112 of file Parallel.py.

GaudiMP.Parallel.WorkManager.stats

Definition at line 118 of file Parallel.py.

---

The documentation for this class was generated from the following file:

• InstallArea/x86_64-slc6-gcc48-opt/python/GaudiMP/Parallel.py