From 87ec65c84b7368c13caf0484ca81faf66a8a6294 Mon Sep 17 00:00:00 2001 From: Jürgen Knödlseder Date: Mon, 30 Jul 2018 15:38:46 +0200 Subject: [PATCH] Reduce code duplication for multiprocessing in cscripts --- cscripts/cslightcrv.py | 33 +++------------------------------ cscripts/csphagen.py | 32 +++----------------------------- cscripts/csphasecrv.py | 31 ++----------------------------- cscripts/cspull.py | 32 ++------------------------------ cscripts/cssens.py | 35 ++++------------------------------- cscripts/csspec.py | 33 +++------------------------------ cscripts/cstsdist.py | 32 ++------------------------------ cscripts/mputils.py | 48 +++++++++++++++++++++++++++++++++++++++++++++--- 8 files changed, 64 insertions(+), 212 deletions(-) diff --git a/cscripts/cslightcrv.py b/cscripts/cslightcrv.py index 7539500..cd4ade8 100755 --- a/cscripts/cslightcrv.py +++ b/cscripts/cslightcrv.py @@ -26,33 +26,6 @@ from cscripts import ioutils from cscripts import mputils -# ============================================ # -# Global functions for multiprocessing support # -# ============================================ # -def _multiprocessing_func_wrapper(args): - return _multiprocessing_func(*args) -def _multiprocessing_func(cls, i): - - # Initialise thread logger - cls._log.clear() - cls._log.buffer_size(100000) - - # Compute light curve bin - cstart = cls.celapse() - result = cls._timebin(i) - celapse = cls.celapse() - cstart - log = cls._log.buffer() - - # Close logger - cls._log.close() - - # Collect thread information - info = {'celapse': celapse, 'log': log} - - # Return light curve bin result and thread information - return result, info - - # ================ # # cslightcrv class # # ================ # @@ -651,9 +624,9 @@ class cslightcrv(ctools.csobservation): if self._nthreads > 1: # Compute time bins - args = [(self, i) for i in range(self._tbins.size())] - poolresults = mputils.process(self._nthreads, - _multiprocessing_func_wrapper, args) + args = [(self, '_timebin', i) + for i in range(self._tbins.size())] + poolresults = mputils.process(self._nthreads, mputils.mpfunc, args) # Construct results results = [] diff --git a/cscripts/csphagen.py b/cscripts/csphagen.py index 1403fe2..fac4e41 100755 --- a/cscripts/csphagen.py +++ b/cscripts/csphagen.py @@ -25,32 +25,6 @@ import math import sys from cscripts import mputils -# ============================================ # -# Global functions for multiprocessing support # -# ============================================ # -def _multiprocessing_func_wrapper(args): - return _multiprocessing_func(*args) -def _multiprocessing_func(cls, i): - - # Initialise thread logger - cls._log.clear() - cls._log.buffer_size(100000) - - # Compute light curve bin - cstart = cls.celapse() - result = cls._process_observation(i) - celapse = cls.celapse() - cstart - log = cls._log.buffer() - - # Close logger - cls._log.close() - - # Collect thread information - info = {'celapse': celapse, 'log': log} - - # Return light curve bin result and thread information - return result, info - # =============== # # csfindobs class # @@ -579,9 +553,9 @@ class csphagen(ctools.csobservation): if self._nthreads > 1 and self.obs().size() > 1: # Compute observations - args = [(self, i) for i in range(self.obs().size())] - poolresults = mputils.process(self._nthreads, - _multiprocessing_func_wrapper, args) + args = [(self, '_process_observation', i) + for i in range(self.obs().size())] + poolresults = mputils.process(self._nthreads, mputils.mpfunc, args) # Construct results for i in range(self.obs().size()): diff --git a/cscripts/csphasecrv.py b/cscripts/csphasecrv.py index 9ac08ba..f28280f 100755 --- a/cscripts/csphasecrv.py +++ b/cscripts/csphasecrv.py @@ -25,32 +25,6 @@ from cscripts import obsutils from cscripts import ioutils from cscripts import mputils -# ============================================ # -# Global functions for multiprocessing support # -# ============================================ # -def _multiprocessing_func_wrapper(args): - return _multiprocessing_func(*args) -def _multiprocessing_func(cls, phbin): - - # Initialise thread logger - cls._log.clear() - cls._log.buffer_size(100000) - - # Compute light curve bin - cstart = cls.celapse() - result = cls._phase_bin(phbin) - celapse = cls.celapse() - cstart - log = cls._log.buffer() - - # Close logger - cls._log.close() - - # Collect thread information - info = {'celapse': celapse, 'log': log} - - # Return light curve bin result and thread information - return result, info - # ================ # # csphasecrv class # @@ -497,9 +471,8 @@ class csphasecrv(ctools.csobservation): if self._nthreads > 1: # Compute phase bins - args = [(self, phbin) for phbin in self._phbins] - poolresults = mputils.process(self._nthreads, - _multiprocessing_func_wrapper, args) + args = [(self, '_phase_bin', phbin) for phbin in self._phbins] + poolresults = mputils.process(self._nthreads, mputils.mpfunc, args) # Construct results for i in range(len(self._phbins)): diff --git a/cscripts/cspull.py b/cscripts/cspull.py index 9b1e0fd..b7e92c6 100755 --- a/cscripts/cspull.py +++ b/cscripts/cspull.py @@ -26,33 +26,6 @@ from cscripts import ioutils from cscripts import mputils -# ============================================ # -# Global functions for multiprocessing support # -# ============================================ # -def _multiprocessing_func_wrapper(args): - return _multiprocessing_func(*args) -def _multiprocessing_func(cls, i): - - # Initialise thread logger - cls._log.clear() - cls._log.buffer_size(100000) - - # Compute light curve bin - cstart = cls.celapse() - result = cls._trial(i) - celapse = cls.celapse() - cstart - log = cls._log.buffer() - - # Close logger - cls._log.close() - - # Collect thread information - info = {'celapse': celapse, 'log': log} - - # Return light curve bin result and thread information - return result, info - - # ============ # # cspull class # # ============ # @@ -385,9 +358,8 @@ class cspull(ctools.csobservation): # If more than a single thread is requested then use multiprocessing if self._nthreads > 1: - args = [(self, i + seed) for i in range(ntrials)] - poolresults = mputils.process(self._nthreads, - _multiprocessing_func_wrapper, args) + args = [(self, '_trial', i + seed) for i in range(ntrials)] + poolresults = mputils.process(self._nthreads, mputils.mpfunc, args) # Continue with regular processing for i in range(ntrials): diff --git a/cscripts/cssens.py b/cscripts/cssens.py index fb0cf84..6af4347 100755 --- a/cscripts/cssens.py +++ b/cscripts/cssens.py @@ -28,33 +28,6 @@ from cscripts import ioutils from cscripts import mputils -# ============================================ # -# Global functions for multiprocessing support # -# ============================================ # -def _multiprocessing_func_wrapper(args): - return _multiprocessing_func(*args) -def _multiprocessing_func(cls, i): - - # Initialise thread logger - cls._log.clear() - cls._log.buffer_size(100000) - - # Compute light curve bin - cstart = cls.celapse() - result = cls._e_bin(i) - celapse = cls.celapse() - cstart - log = cls._log.buffer() - - # Close logger - cls._log.close() - - # Collect thread information - info = {'celapse': celapse, 'log': log} - - # Return light curve bin result and thread information - return result, info - - # ============ # # cssens class # # ============ # @@ -715,10 +688,10 @@ class cssens(ctools.csobservation): # If using multiprocessing if self._nthreads > 1 and self._ebounds.size() > 1: - # Compute phase bins - args = [(self, ieng) for ieng in range(self._ebounds.size())] - poolresults = mputils.process(self._nthreads, - _multiprocessing_func_wrapper, args) + # Compute energy bins + args = [(self, '_e_bin', i) + for i in range(self._ebounds.size())] + poolresults = mputils.process(self._nthreads, mputils.mpfunc, args) # Construct results for ieng in range(self._ebounds.size()): diff --git a/cscripts/csspec.py b/cscripts/csspec.py index f79fa40..cd6a8b1 100755 --- a/cscripts/csspec.py +++ b/cscripts/csspec.py @@ -24,33 +24,6 @@ import ctools from cscripts import mputils -# ============================================ # -# Global functions for multiprocessing support # -# ============================================ # -def _multiprocessing_func_wrapper(args): - return _multiprocessing_func(*args) -def _multiprocessing_func(cls, i): - - # Initialise thread logger - cls._log.clear() - cls._log.buffer_size(100000) - - # Compute light curve bin - cstart = cls.celapse() - result = cls._fit_energy_bin(i) - celapse = cls.celapse() - cstart - log = cls._log.buffer() - - # Close logger - cls._log.close() - - # Collect thread information - info = {'celapse': celapse, 'log': log} - - # Return light curve bin result and thread information - return result, info - - # ============ # # csspec class # # ============ # @@ -775,9 +748,9 @@ class csspec(ctools.csobservation): if self._nthreads > 1: # Compute energy bins - args = [(self, i) for i in range(self._ebounds.size())] - poolresults = mputils.process(self._nthreads, - _multiprocessing_func_wrapper, args) + args = [(self, '_fit_energy_bin', i) + for i in range(self._ebounds.size())] + poolresults = mputils.process(self._nthreads, mputils.mpfunc, args) # Construct results for i in range(self._ebounds.size()): diff --git a/cscripts/cstsdist.py b/cscripts/cstsdist.py index 977ec2a..ccbaf99 100755 --- a/cscripts/cstsdist.py +++ b/cscripts/cstsdist.py @@ -27,33 +27,6 @@ from cscripts import ioutils from cscripts import mputils -# ============================================ # -# Global functions for multiprocessing support # -# ============================================ # -def _multiprocessing_func_wrapper(args): - return _multiprocessing_func(*args) -def _multiprocessing_func(cls, i): - - # Initialise thread logger - cls._log.clear() - cls._log.buffer_size(100000) - - # Compute light curve bin - cstart = cls.celapse() - result = cls._trial(i) - celapse = cls.celapse() - cstart - log = cls._log.buffer() - - # Close logger - cls._log.close() - - # Collect thread information - info = {'celapse': celapse, 'log': log} - - # Return light curve bin result and thread information - return result, info - - # ============== # # cstsdist class # # ============== # @@ -334,9 +307,8 @@ class cstsdist(ctools.csobservation): # If more than a single thread is requested then use multiprocessing if self._nthreads > 1: - args = [(self, i) for i in range(ntrials)] - poolresults = mputils.process(self._nthreads, - _multiprocessing_func_wrapper, args) + args = [(self, '_trial', i) for i in range(ntrials)] + poolresults = mputils.process(self._nthreads, mputils.mpfunc, args) # Continue with regular processing for i in range(ntrials): diff --git a/cscripts/mputils.py b/cscripts/mputils.py index eac46cc..722c011 100644 --- a/cscripts/mputils.py +++ b/cscripts/mputils.py @@ -20,9 +20,9 @@ import gammalib -# ====================== # -# Get number of threads # -# ====================== # +# ===================== # +# Get number of threads # +# ===================== # def nthreads(cls): """ Determines the number of parallel processes to use @@ -107,3 +107,45 @@ def process(nthreads, function, args): # Return results return results + + +# ======== # +# Function # +# ======== # +def mpfunc(args): + """ + Multiprocessing function + + Parameters + ---------- + args : tuple + Tuple comprised of class, function name and function argument + + Returns + ------- + result, info : tuple + Tuple of function result and information + """ + # Extract + cls = args[0] # Class + fct = args[1] # Class method name + i = args[2] # Function argument + + # Initialise thread logger + cls._log.clear() + cls._log.buffer_size(100000) + + # Compute function for argument + cstart = cls.celapse() + result = getattr(cls, fct)(i) + celapse = cls.celapse() - cstart + log = cls._log.buffer() + + # Close logger + cls._log.close() + + # Collect thread information + info = {'celapse': celapse, 'log': log} + + # Return light curve bin result and thread information + return result, info -- libgit2 0.21.4