Added terminator to gracefully exit optimisation.

Author: Jaquier Aurélien Tristan

bluepyopt/deapext/algorithms.py

@@ -93,7 +93,8 @@ def eaAlphaMuPlusLambdaCheckpoint(
         halloffame=None,
         cp_frequency=1,
         cp_filename=None,
-        continue_cp=False):
+        continue_cp=False,
+        terminator=None):
     r"""This is the :math:`(~\alpha,\mu~,~\lambda)` evolutionary algorithm
 
     Args:
@@ -108,6 +109,8 @@ def eaAlphaMuPlusLambdaCheckpoint(
         cp_frequency(int): generations between checkpoints
         cp_filename(string): path to checkpoint filename
         continue_cp(bool): whether to continue
+        terminator (multiprocessing.Event): exit loop when is set.
+            Not taken into account if None.
     """
 <<<<<<< HEAD
 
@@ -157,7 +160,9 @@ def eaAlphaMuPlusLambdaCheckpoint(
     # Begin the generational process
     gen = start_gen + 1
     stopping_params = {"gen": gen}
-    while not(_check_stopping_criteria(stopping_criteria, stopping_params)):
+    while utils.run_next_gen(
+            not(_check_stopping_criteria(stopping_criteria, stopping_params)),
+            terminator):
         offspring = _get_offspring(parents, toolbox, cxpb, mutpb)
 
         population = parents + offspring

bluepyopt/deapext/optimisations.py

@@ -252,7 +252,8 @@ def run(self,
             continue_cp=False,
             cp_filename=None,
             cp_frequency=1,
-            parent_population=None):
+            parent_population=None,
+            terminator=None):
         """Run optimisation"""
         # Allow run function to override offspring_size
         # TODO probably in the future this should not be an object field
@@ -307,7 +308,8 @@ def run(self,
             halloffame=self.hof,
             cp_frequency=cp_frequency,
             continue_cp=continue_cp,
-            cp_filename=cp_filename)
+            cp_filename=cp_filename,
+            terminator=terminator)
 
         # Update hall of fame
         self.hof = hof

bluepyopt/deapext/optimisationsCMA.py

@@ -233,7 +233,12 @@ def setup_deap(self):
             self.toolbox.register("map", self.map_function)
 
     def run(
-        self, max_ngen=0, cp_frequency=1, continue_cp=False, cp_filename=None
+        self,
+        max_ngen=0,
+        cp_frequency=1,
+        continue_cp=False,
+        cp_filename=None,
+        terminator=None,
     ):
         """ Run the optimizer until a stopping criteria is met.
 
@@ -242,6 +247,8 @@ def run(
             cp_frequency(int): generations between checkpoints
             continue_cp(bool): whether to continue
             cp_filename(string): path to checkpoint filename
+            terminator (multiprocessing.Event): exit loop when is set.
+                Not taken into account if None.
         """
 
         stats = self.get_stats()
@@ -288,7 +295,7 @@ def run(
         pop = CMA_es.get_population(self.to_space)
 
         # Run until a termination criteria is met
-        while CMA_es.active:
+        while utils.run_next_gen(CMA_es.active, terminator):
             logger.info("Generation {}".format(gen))
 
             # Generate the new populations

bluepyopt/deapext/utils.py

@@ -137,3 +137,10 @@ def uniform(lower_list, upper_list, dimensions):
 def reduce_method(meth):
     """Overwrite reduce"""
     return (getattr, (meth.__self__, meth.__func__.__name__))
+
+
+def run_next_gen(criteria, terminator):
+    """Condition to stay inside the loop."""
+    if terminator is None:
+        return criteria
+    return criteria and not terminator.is_set()

bluepyopt/tests/test_deapext/test_utils.py

@@ -0,0 +1,36 @@
+"""bluepyopt.utils tests"""
+
+import multiprocessing
+import time
+
+import bluepyopt.deapext.utils as utils
+import nose.tools as nt
+
+def flag(event):
+    """Send a multiprocessing event."""
+    time.sleep(1)
+    event.set()
+
+def catch_event(event):
+    """Verify that run_next_gen changes when event is caught."""
+    # None case
+    nt.assert_true(utils.run_next_gen(True, None))
+
+    # event is not set case
+    nt.assert_true(utils.run_next_gen(True, event))
+
+    # event is set by another process case
+    time.sleep(2)
+    nt.assert_equal(utils.run_next_gen(True, event), False)
+
+def test_run_next_gen_condition():
+    """deapext.utils: Testing run_next_gen."""
+    event = multiprocessing.Event()
+    p1 = multiprocessing.Process(target=catch_event, args=(event,))
+    p2 = multiprocessing.Process(target=flag, args=(event,))
+
+    p1.start()
+    p2.start()
+
+    p1.join()
+    p2.join()