Parallel population processing

golem/core/constants.py

@@ -1,5 +1,6 @@
 import numpy as np
 
+MAX_GRAPH_GEN_ATTEMPTS_PER_IND = 5
 MAX_GRAPH_GEN_ATTEMPTS = 1000
 MAX_TUNING_METRIC_VALUE = np.inf
 MIN_TIME_FOR_TUNING_IN_SEC = 3

golem/core/optimisers/genetic/evaluation.py

@@ -241,15 +241,18 @@ def dispatch(self, objective: ObjectiveFunction, timer: Optional[Timer] = None)
 
     def evaluate_population(self, individuals: PopulationT) -> PopulationT:
         individuals_to_evaluate, individuals_to_skip = self.split_individuals_to_evaluate(individuals)
-        # Evaluate individuals without valid fitness in parallel.
-        n_jobs = determine_n_jobs(self._n_jobs, self.logger)
 
-        parallel = Parallel(n_jobs=n_jobs, verbose=0, pre_dispatch="2*n_jobs")
+        # Evaluate individuals without valid fitness
         eval_func = partial(self.evaluate_single, logs_initializer=Log().get_parameters())
-        evaluation_results = parallel(delayed(eval_func)(ind.graph, ind.uid) for ind in individuals_to_evaluate)
+
+        if len(individuals_to_evaluate) == 1 or self._n_jobs == 1:
+            evaluation_results = [eval_func(ind.graph, ind.uid) for ind in individuals_to_evaluate]
+        else:
+            n_jobs = determine_n_jobs(self._n_jobs, self.logger)
+            parallel = Parallel(n_jobs=n_jobs)
+            evaluation_results = parallel(delayed(eval_func)(ind.graph, ind.uid) for ind in individuals_to_evaluate)
+
         individuals_evaluated = self.apply_evaluation_results(individuals_to_evaluate, evaluation_results)
-        # If there were no successful evals then try once again getting at least one,
-        # even if time limit was reached
         successful_evals = individuals_evaluated + individuals_to_skip
         self.population_evaluation_info(evaluated_pop_size=len(successful_evals),
                                         pop_size=len(individuals))

golem/core/optimisers/genetic/gp_optimizer.py

@@ -1,8 +1,5 @@
-from copy import deepcopy
-from random import choice
 from typing import Sequence, Union, Any
 
-from golem.core.constants import MAX_GRAPH_GEN_ATTEMPTS
 from golem.core.dag.graph import Graph
 from golem.core.optimisers.genetic.gp_params import GPAlgorithmParameters
 from golem.core.optimisers.genetic.operators.crossover import Crossover
@@ -44,7 +41,12 @@ def __init__(self,
         self.elitism = Elitism(graph_optimizer_params)
         self.operators = [self.regularization, self.selection, self.crossover,
                           self.mutation, self.inheritance, self.elitism]
-        self.reproducer = ReproductionController(graph_optimizer_params, self.selection, self.mutation, self.crossover)
+
+        self.reproducer = ReproductionController(parameters=graph_optimizer_params,
+                                                 selection=self.selection,
+                                                 mutation=self.mutation,
+                                                 crossover=self.crossover,
+                                                 verifier=self.graph_generation_params.verifier)
 
         # Define adaptive parameters
         self._pop_size: PopulationSize = init_adaptive_pop_size(graph_optimizer_params, self.generations)
@@ -68,36 +70,10 @@ def _initial_population(self, evaluator: EvaluationOperator):
         pop_size = self.graph_optimizer_params.pop_size
 
         if len(self.initial_individuals) < pop_size:
-            self.initial_individuals = self._extend_population(self.initial_individuals, pop_size)
+            self.initial_individuals += self.reproducer._mutate_over_population(population=self.initial_individuals,
+                                                                                evaluator=evaluator)
             # Adding of extended population to history
-            self._update_population(evaluator(self.initial_individuals), 'extended_initial_assumptions')
-
-    def _extend_population(self, pop: PopulationT, target_pop_size: int) -> PopulationT:
-        verifier = self.graph_generation_params.verifier
-        extended_pop = list(pop)
-        pop_graphs = [ind.graph for ind in extended_pop]
-
-        # Set mutation probabilities to 1.0
-        initial_req = deepcopy(self.requirements)
-        initial_req.mutation_prob = 1.0
-        self.mutation.update_requirements(requirements=initial_req)
-
-        for iter_num in range(MAX_GRAPH_GEN_ATTEMPTS):
-            if len(extended_pop) == target_pop_size:
-                break
-            new_ind = self.mutation(choice(pop))
-            if new_ind:
-                new_graph = new_ind.graph
-                if new_graph not in pop_graphs and verifier(new_graph):
-                    extended_pop.append(new_ind)
-                    pop_graphs.append(new_graph)
-        else:
-            self.log.warning(f'Exceeded max number of attempts for extending initial graphs, stopping.'
-                             f'Current size {len(pop)}, required {target_pop_size} graphs.')
-
-        # Reset mutation probabilities to default
-        self.mutation.update_requirements(requirements=self.requirements)
-        return extended_pop
+            self._update_population(self.initial_individuals, 'extended_initial_assumptions')
 
     def _evolve_population(self, evaluator: EvaluationOperator) -> PopulationT:
         """ Method realizing full evolution cycle """
@@ -120,7 +96,6 @@ def _evolve_population(self, evaluator: EvaluationOperator) -> PopulationT:
         # Use some part of previous pop in the next pop
         new_population = self.inheritance(self.population, new_population)
         new_population = self.elitism(self.generations.best_individuals, new_population)
-
         return new_population
 
     def _update_requirements(self):

golem/core/optimisers/genetic/gp_params.py

@@ -76,6 +76,7 @@ class GPAlgorithmParameters(AlgorithmParameters):
     mutation_prob: float = 0.8
     variable_mutation_num: bool = True
     max_num_of_operator_attempts: int = 100
+    max_num_of_mutation_attempts: int = 3
     mutation_strength: MutationStrengthEnum = MutationStrengthEnum.mean
     min_pop_size_with_elitism: int = 5
     required_valid_ratio: float = 0.9

golem/core/optimisers/genetic/operators/crossover.py

@@ -4,6 +4,8 @@
 from random import choice, random, sample
 from typing import Callable, Union, Iterable, Tuple, TYPE_CHECKING
 
+from joblib import Parallel, delayed
+
 from golem.core.adapter import register_native
 from golem.core.dag.graph_utils import nodes_from_layer, node_depth
 from golem.core.optimisers.genetic.gp_operators import equivalent_subtree, replace_subtrees
@@ -40,12 +42,13 @@ def __init__(self,
         self.graph_generation_params = graph_generation_params
 
     def __call__(self, population: PopulationT) -> PopulationT:
-        if len(population) == 1:
-            new_population = population
+        if len(population) > 1:
+            with Parallel(n_jobs=self.requirements.n_jobs) as parallel:
+                new_population = parallel(delayed(self._crossover)(ind_1, ind_2)
+                                          for ind_1, ind_2 in Crossover.crossover_parents_selection(population))
+                new_population = list(chain(*new_population))
         else:
-            new_population = []
-            for ind_1, ind_2 in Crossover.crossover_parents_selection(population):
-                new_population += self._crossover(ind_1, ind_2)
+            new_population = population[:]
         return new_population
 
     @staticmethod

golem/core/optimisers/genetic/operators/mutation.py

@@ -22,6 +22,7 @@
 if TYPE_CHECKING:
     from golem.core.optimisers.genetic.gp_params import GPAlgorithmParameters
 
+MutationType = Union[MutationTypesEnum, Callable]
 MutationFunc = Callable[[Graph, GraphRequirements, GraphGenerationParams, AlgorithmParameters], Graph]
 MutationIdType = Hashable
 MutationRepo = Mapping[MutationIdType, MutationFunc]
@@ -81,11 +82,11 @@ def __call__(self, population: Union[Individual, PopulationT]) -> Union[Individu
         if isinstance(population, Individual):
             population = [population]
 
-        final_population, mutations_applied, application_attempts = tuple(zip(*map(self._mutation, population)))
-
-        # drop individuals to which mutations could not be applied
-        final_population = [ind for ind, init_ind, attempt in zip(final_population, population, application_attempts)
-                            if not attempt or ind.graph != init_ind.graph]
+        final_population = []
+        for individual in population:
+            new_ind, _, applied = self._mutation(individual)
+            if not applied or new_ind.graph != individual.graph:
+                final_population.append(new_ind)
 
         if len(population) == 1:
             return final_population[0] if final_population else final_population
@@ -160,3 +161,18 @@ def _get_mutation_func(self, mutation_type: Union[MutationTypesEnum, Callable])
             mutation_func = self._mutations_repo[mutation_type]
         adapted_mutation_func = self.graph_generation_params.adapter.adapt_func(mutation_func)
         return adapted_mutation_func
+
+
+class SinglePredefinedMutation(Mutation):
+    def __call__(self, individual: Individual, mutation_type: MutationType) -> Individual:
+        new_graph = deepcopy(individual.graph)
+        mutation_func = self._get_mutation_func(mutation_type)
+
+        new_graph = mutation_func(new_graph, requirements=self.requirements,
+                                  graph_gen_params=self.graph_generation_params,
+                                  parameters=self.parameters)
+
+        parent_operator = ParentOperator(type_='mutation', operators=mutation_type, parent_individuals=individual)
+        individual = Individual(new_graph, parent_operator,
+                                metadata=self.requirements.static_individual_metadata)
+        return individual, mutation_type

golem/core/optimisers/genetic/operators/regularization.py

@@ -34,6 +34,7 @@ def __call__(self, population: PopulationT, evaluator: EvaluationOperator) -> Po
             raise ValueError(f'Required regularization type not found: {regularization_type}')
 
     def _decremental_regularization(self, population: PopulationT, evaluator: EvaluationOperator) -> PopulationT:
+        # TODO: do it in parallel if it can be done
         size = self.parameters.pop_size
         additional_inds = []
         prev_nodes_ids = set()