Update DaskEvaluator

examples/multiobjective/nsgaii/distributed_nsgaii_with_dask_evaluator.py

@@ -16,11 +16,11 @@
 
     algorithm = NSGAII(
         problem=problem,
-        population_size=10,
-        offspring_population_size=10,
+        population_size=100,
+        offspring_population_size=100,
         mutation=PolynomialMutation(probability=1.0 / problem.number_of_variables(), distribution_index=20),
         crossover=SBXCrossover(probability=1.0, distribution_index=20),
-        population_evaluator=DaskEvaluator(),
+        population_evaluator=DaskEvaluator(number_of_cores=8),
         termination_criterion=StoppingByEvaluations(max_evaluations=max_evaluations),
     )
 

src/jmetal/util/evaluator.py

@@ -2,6 +2,7 @@
 from abc import ABC, abstractmethod
 from multiprocessing.pool import Pool, ThreadPool
 from typing import Generic, List, TypeVar
+from concurrent.futures import ThreadPoolExecutor
 
 try:
     import dask
@@ -75,11 +76,12 @@ def evaluate_solution(solution, problem):
 
 
 class DaskEvaluator(Evaluator[S]):
-    def __init__(self, scheduler="processes"):
+    def __init__(self, scheduler="processes", number_of_cores=4):
         self.scheduler = scheduler
+        self.number_of_cores = number_of_cores
 
     def evaluate(self, solution_list: List[S], problem: Problem) -> List[S]:
-        with dask.config.set(scheduler=self.scheduler):
+        with dask.config.set(scheduler=self.scheduler, pool=ThreadPoolExecutor(self.number_of_cores)):
             return list(
                 dask.compute(
                     *[dask.delayed(evaluate_solution)(solution=solution, problem=problem) for solution in solution_list]