implemented EvalParallel class in fitness_transformations

	modified:   cma/fitness_transformations.py

cma/fitness_transformations.py

@@ -2,7 +2,8 @@
 """
 from __future__ import absolute_import, division, print_function  #, unicode_literals, with_statement
 import numpy as np
-from numpy import array
+from multiprocessing import Pool as ProcessingPool
+# from pathos.multiprocessing import ProcessingPool
 import time
 from .utilities import utils
 from .transformations import ConstRandnShift, Rotation
@@ -12,6 +13,98 @@
 
 rotate = Rotation()
 
+class EvalParallel(object):
+    """A class and context manager for parallel evaluations.
+
+    To be used with the `with` statement (otherwise `terminate` needs to
+    be called to free resources)::
+
+        with EvalParallel() as eval_all:
+            fvals = eval_all(fitness, solutions)
+
+    assigns a callable `EvalParallel` class instance to ``eval_all``.
+    The instance can be called with a `list` (or `tuple` or any
+    sequence) of solutions and returns their fitness values. That is::
+
+        eval_all(fitness, solutions) == [fitness(x) for x in solutions]
+
+    `EvalParallel.__call__` may take two additional optional arguments,
+    namely `args` passed to ``fitness`` and `timeout` passed to the
+    `multiprocessing.pool.ApplyResult.get` method which raises
+    `multiprocessing.TimeoutError` in case.
+
+    Examples::
+
+        >>> import cma
+        >>> from cma.fitness_transformations import EvalParallel
+        >>> # class usage, don't forget to call terminate
+        >>> ep = EvalParallel()
+        >>> ep(cma.fitness_functions.elli, [[1,2], [3,4], [4, 5]])  # doctest:+ELLIPSIS
+        [4000000.944...
+        >>> ep.terminate()
+        ...
+        >>> # use with `with` statement (context manager)
+        >>> es = cma.CMAEvolutionStrategy(3 * [1], 1, dict(verbose=-9))
+        >>> with EvalParallel(12) as eval_all:
+        ...     while not es.stop():
+        ...         X = es.ask()
+        ...         es.tell(X, eval_all(cma.fitness_functions.elli, X))
+        >>> assert es.result[1] < 1e-13 and es.result[2] < 1500
+
+    Parameters: the `EvalParallel` constructor takes the number of
+    processes as optional input argument, which is by default
+    ``multiprocessing.cpu_count()``.
+    
+    Limitations: The `multiprocessing` module (on which this class is
+    based upon) does not work with class instance method calls.
+    
+    In some cases the execution may be considerably slowed down,
+    as for example with test suites from coco/bbob.
+
+    """
+    def __init__(self, number_of_processes=None):
+        self.processes = number_of_processes
+        self.pool = ProcessingPool(self.processes)
+
+    def __call__(self, fitness_function, solutions, args=(), timeout=None):
+        """evaluate a list/sequence of solution-"vectors", return a list
+        of corresponding f-values.
+
+        Raises `multiprocessing.TimeoutError` if `timeout` is given and
+        exceeded.
+        """
+        warning_str = ("WARNING: `fitness_function` must be a function,"
+                       " not an instancemethod, to work with"
+                       " `multiprocessing`")
+        if isinstance(fitness_function, type(self.__init__)):
+            print(warning_str)
+        jobs = [self.pool.apply_async(fitness_function,
+                                      (x,) + args) for x in solutions]
+        try:
+            return [job.get(timeout) for job in jobs]
+        except:
+            print(warning_str)
+            raise
+
+    def terminate(self):
+        """free allocated processing pool"""
+        # self.pool.close()  # would wait for job termination
+        self.pool.terminate()  # terminate jobs regardless
+        self.pool.join()  # end spawning
+
+    def __enter__(self):
+        # we could assign self.pool here, but then `EvalParallel` would
+        # *only* work when using the `with` statement
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.terminate()
+
+    def __del__(self):
+        """though generally not recommended `__del__` should be OK here"""
+        self.terminate()
+
+
 class Function(object):
     """a declarative base class, indicating that a derived class instance
     "is" a (fitness/objective) function.