Skip invalid individuals in evaluation streams

geneticengine/evaluation/api.py

@@ -30,6 +30,10 @@ def register_evaluation(self, individual: IndT, problem: Problem):
         if problem.is_solved(individual.get_fitness(problem)):
             raise IndividualFoundException(individual)
 
+    def register_invalid_evaluation(self):
+        """Count an attempted evaluation whose individual must be skipped."""
+        self.count += 1
+
     def number_of_evaluations(self):
         return self.count
 

geneticengine/evaluation/parallel.py

@@ -1,38 +1,42 @@
-from abc import ABCMeta
-from pickle import _Pickler as StockPickler
-from typing import Any, Generator, Iterable, Optional  # attr-defined: ignore
-from dill import register
+from concurrent.futures import ThreadPoolExecutor
+from itertools import islice
+from os import cpu_count
+from typing import Any, Generator, Iterable
+
 from geneticengine.problems import Fitness, InvalidFitnessException, Problem
 from geneticengine.evaluation.api import Evaluator, IndT
 
 
-@register(ABCMeta)
-def save_abc(pickler, obj):
-    StockPickler.save_type(pickler, obj)  # pyright: ignore
-
-
 class ParallelEvaluator(Evaluator):
-    """Evaluates individuals in parallel, each time they are needed."""
+    """Evaluates individuals lazily in bounded batches of worker threads."""
 
     def evaluate_async(
         self,
         problem: Problem,
         individuals: Iterable[IndT],
     ) -> Generator[IndT, Any, Any]:
-        indivs = list(individuals)
-
-        def mapper(ind: IndT) -> Optional[Fitness]:
+        def mapper(ind: IndT) -> tuple[IndT, Fitness | None, bool]:
+            if ind.has_fitness(problem):
+                return ind, ind.get_fitness(problem), False
             try:
-                return self.eval_single(problem, ind)
+                return ind, self.eval_single(problem, ind), True
             except InvalidFitnessException:
-                return problem.get_invalid_fitness()
-
-
-        from pathos.multiprocessing import ProcessingPool as Pool  # pyright: ignore
-
-        with Pool(len(indivs)) as pool:
-            fitnesses = pool.map(mapper, indivs)
-            for i, f in zip(indivs, fitnesses):
-                i.set_fitness(problem, f)
-                self.register_evaluation(i, problem)
-                yield i
+                return ind, None, True
+
+        with ThreadPoolExecutor(max_workers=self.workers) as executor:
+            while batch := list(islice(individuals, self.workers)):
+                fitnesses = executor.map(mapper, batch)
+                fitnesses = list(fitnesses)
+
+                for i, f, evaluated in fitnesses:
+                    if f is None:
+                        self.register_invalid_evaluation()
+                        continue
+                    if evaluated:
+                        i.set_fitness(problem, f)
+                        self.register_evaluation(i, problem)
+                    yield i
+
+    def __init__(self, workers: int | None = None):
+        super().__init__()
+        self.workers = workers or min(cpu_count() or 1, 8)

geneticengine/evaluation/sequential.py

@@ -17,7 +17,8 @@ def evaluate_async(
                 try:
                     f = self.eval_single(problem, individual)
                 except InvalidFitnessException:
-                    f = problem.get_invalid_fitness()
+                    self.register_invalid_evaluation()
+                    continue
                 individual.set_fitness(problem, f)
                 self.register_evaluation(individual, problem)
                 yield individual

tests/core/fitness_helpers_test.py

@@ -5,6 +5,8 @@
 from geneticengine.random.sources import NativeRandomSource
 from geneticengine.representations.tree.initializations import MaxDepthDecider
 from geneticengine.solutions.individual import PhenotypicIndividual
+from geneticengine.solutions.individual import ConcreteIndividual
+from geneticengine.evaluation.parallel import ParallelEvaluator
 from geneticengine.evaluation.sequential import SequentialEvaluator
 from geneticengine.problems.helpers import is_better
 from geneticengine.problems import InvalidFitnessException
@@ -59,4 +61,37 @@ def custom_fit(l:Leaf):
 
         problem = SingleObjectiveProblem(fitness_function=custom_fit, minimize=True)
         evaluated = [ ind for ind in evaluator.evaluate(problem, [a, b])]
-        assert problem.is_better(evaluated[0].get_fitness(problem), evaluated[1].get_fitness(problem))
+        assert evaluated == [a]
+        assert evaluator.number_of_evaluations() == 2
+
+    def test_invalid_fitness_is_skipped_from_a_stream(self):
+        g = extract_grammar([Leaf], Root)
+        r = NativeRandomSource(0)
+        representation = TreeBasedRepresentation(g, MaxDepthDecider(r, g, 2))
+        evaluator = SequentialEvaluator()
+
+        def custom_fit(leaf: Leaf):
+            if leaf.a == 1:
+                raise InvalidFitnessException()
+            return leaf.a
+
+        problem = SingleObjectiveProblem(fitness_function=custom_fit, minimize=True)
+        stream = (PhenotypicIndividual(Leaf(value), representation) for value in [1, 2, 1, 3])
+        evaluated = list(evaluator.evaluate(problem, stream))
+        assert [ind.get_phenotype().a for ind in evaluated] == [2, 3]
+        assert evaluator.number_of_evaluations() == 4
+
+    def test_parallel_evaluator_skips_invalid_fitness_from_a_stream(self):
+        evaluator = ParallelEvaluator(workers=2)
+
+        def custom_fit(value: int):
+            if value == 1:
+                raise InvalidFitnessException()
+            return value
+
+        problem = SingleObjectiveProblem(fitness_function=custom_fit, minimize=True)
+        stream = (ConcreteIndividual(value) for value in [1, 2, 1, 3])
+        evaluated = list(evaluator.evaluate(problem, stream))
+
+        assert [ind.get_phenotype() for ind in evaluated] == [2, 3]
+        assert evaluator.number_of_evaluations() == 4