GaudiMP.Parallel.WorkManager Class Reference

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

Public Attributes
	ncpus = multiprocessing.cpu_count()
	ppservers = ppservers
list	sessions = [SshSession(srv) for srv in ppservers]
	server = pp.Server(ncpus=self.ncpus, ppservers=self.ppservers)
str	mode = "cluster"
	pool = multiprocessing.Pool(self.ncpus)
dict	stats = {}

Protected Member Functions
	_printStatistics (self)
	_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 140 of file Parallel.py.

Constructor & Destructor Documentation

__init__()

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

Definition at line 145 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__()

GaudiMP.Parallel.WorkManager.__del__

(

self

)

Definition at line 162 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()

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

protected

Definition at line 213 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()

GaudiMP.Parallel.WorkManager._printStatistics ( self )

protected

Definition at line 202 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(
                f"       {stat.njob} |        {100.0 * stat.njob / njobs:6.2f} |     {stat.time:8.3f} |    {stat.time / stat.njob:8.3f} | {name}"
            )
 

process()

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

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

str GaudiMP.Parallel.WorkManager.mode = "cluster"

Definition at line 156 of file Parallel.py.

ncpus

GaudiMP.Parallel.WorkManager.ncpus = multiprocessing.cpu_count()

Definition at line 147 of file Parallel.py.

pool

GaudiMP.Parallel.WorkManager.pool = multiprocessing.Pool(self.ncpus)

Definition at line 158 of file Parallel.py.

ppservers

GaudiMP.Parallel.WorkManager.ppservers = ppservers

Definition at line 153 of file Parallel.py.

server

GaudiMP.Parallel.WorkManager.server = pp.Server(ncpus=self.ncpus, ppservers=self.ppservers)

Definition at line 155 of file Parallel.py.

sessions

list GaudiMP.Parallel.WorkManager.sessions = [SshSession(srv) for srv in ppservers]

Definition at line 154 of file Parallel.py.

stats

dict GaudiMP.Parallel.WorkManager.stats = {}

Definition at line 160 of file Parallel.py.

---

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

• GaudiMP/python/GaudiMP/Parallel.py