GaudiMP.Parallel.WorkManager Class Reference

Inheritance diagram for GaudiMP.Parallel.WorkManager:

Collaboration diagram for GaudiMP.Parallel.WorkManager:

Public Member Functions
def	__init__ (self, ncpus="autodetect", ppservers=None)
def	__del__ (self)
def	process (self, task, items, timeout=90000)

Public Attributes
	ncpus
	ppservers
	sessions
	server
	mode
	pool
	stats

Private Member Functions
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 145 of file Parallel.py.

Constructor & Destructor Documentation

__init__()

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

Definition at line 150 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 = {}
 

__del__()

def GaudiMP.Parallel.WorkManager.__del__

(

self

)

Definition at line 167 of file Parallel.py.

    def __del__(self):
        if hasattr(self, "server"):
            self.server.destroy()
        if hasattr(self, "pool"):
            self.pool.close()
 

Member Function Documentation

_mergeStatistics()

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

private

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

_printStatistics()

def GaudiMP.Parallel.WorkManager._printStatistics (   self )

private

Definition at line 207 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.0 * stat.njob / njobs,
                    stat.time,
                    stat.time / stat.njob,
                    name,
                )
            )
 

process()

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

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

mode

GaudiMP.Parallel.WorkManager.mode

Definition at line 161 of file Parallel.py.

ncpus

GaudiMP.Parallel.WorkManager.ncpus

Definition at line 152 of file Parallel.py.

pool

GaudiMP.Parallel.WorkManager.pool

Definition at line 163 of file Parallel.py.

ppservers

GaudiMP.Parallel.WorkManager.ppservers

Definition at line 158 of file Parallel.py.

server

GaudiMP.Parallel.WorkManager.server

Definition at line 160 of file Parallel.py.

sessions

GaudiMP.Parallel.WorkManager.sessions

Definition at line 159 of file Parallel.py.

stats

GaudiMP.Parallel.WorkManager.stats

Definition at line 165 of file Parallel.py.

---

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

• GaudiMP/python/GaudiMP/Parallel.py