Pymoo learning (7): Parallelization

1 introduce

   In practice, , Parallelization It can significantly improve the efficiency of optimization . Based on Population The algorithm of , You can evaluate itself by parallelization , Realize the evaluation of a set of solutions .

2 Vectorization matrix operation

   One way is to use Numpy Matrix operations , It has been used for almost all in Pymoo Testing problems implemented in . By default ,elementwise_evaluation Set to False, It means _evaluate Retrieve a set of solutions . therefore , Input matrix $x$ Each line of is an individual , Each column is a variable ：

import numpy as np
from pymoo.core.problem import Problem
from pymoo.algorithms.soo.nonconvex.ga import GA
from pymoo.optimize import minimize


class MyProblem(Problem):

    def __init__(self, **kwargs):
        super().__init__(n_var=10, n_obj=1, n_constr=0, xl=-5, xu=5, **kwargs)

    def _evaluate(self, x, out, *args, **kwargs):
        out["F"] = np.sum(x ** 2, axis=1)


res = minimize(MyProblem(), GA())
print('Threads:', res.exec_time)

   Output is as follows ：

Threads: 1.416006326675415

3 Starmap Interface

  Starmap from Python Standard library multiprocessing.Pool.starmap Provide , It is convenient to parallelize . You need to set elementwise_evaluation=True, Means every call _evaluate Evaluate only one solution .

3.1 Threads

import numpy as np
from pymoo.core.problem import Problem
from pymoo.core.problem import starmap_parallelized_eval
from pymoo.algorithms.soo.nonconvex.pso import PSO
from pymoo.optimize import minimize
from multiprocessing.pool import ThreadPool


class MyProblem(Problem):

    def __init__(self, **kwargs):
        super().__init__(n_var=10, n_obj=1, n_constr=0, xl=-5, xu=5, **kwargs)

    def _evaluate(self, x, out, *args, **kwargs):
        out["F"] = np.sum(x ** 2, axis=1)


n_threads = 8
pool = ThreadPool(n_threads)
problem = MyProblem(runner=pool.starmap, func_eval=starmap_parallelized_eval)
res = minimize(problem, PSO(), seed=1, n_gen=100)
print('Threads:', res.exec_time)

   Output is as follows ：

Threads: 0.5501224994659424

3.2 process

import multiprocessing


n_proccess = 8
pool = multiprocessing.Pool(n_proccess)
problem = MyProblem(runner=pool.starmap, func_eval=starmap_parallelized_eval)
res = minimize(problem, PSO(), seed=1, n_gen=100)
print('Processes:', res.exec_time)

   Output is as follows ：

Processes: 1.1640357971191406

3.3 Dask

   A more advanced method is to assign the evaluation function to several worker. stay Pymoo Recommended in Dask.
   notes ： You may need to install the following libraries ：

pip install dask distributed

   The code is as follows ：

import numpy as np
from dask.distributed import Client
from pymoo.core.problem import dask_parallelized_eval
from pymoo.core.problem import Problem
from pymoo.algorithms.soo.nonconvex.pso import PSO
from pymoo.optimize import minimize


class MyProblem(Problem):

    def __init__(self, **kwargs):
        super().__init__(n_var=10, n_obj=1, n_constr=0, xl=-5, xu=5, **kwargs)

    def _evaluate(self, x, out, *args, **kwargs):
        out["F"] = np.sum(x ** 2, axis=1)


if __name__ == '__main__':
    client = Client()
    client.restart()
    print("STARTED")

    client = Client()
    problem = MyProblem(runner=client, func_eval=dask_parallelized_eval)

    res = minimize(problem, PSO(), seed=1, n_gen=100)
    print('Dask:', res.exec_time)

   Output is as follows ：

STARTED
Dask: 1.30446195602417

4 Personalized parallel

4.1 Threads

import numpy as np
from multiprocessing.pool import ThreadPool
from pymoo.core.problem import Problem
from pymoo.algorithms.soo.nonconvex.pso import PSO
from pymoo.optimize import minimize


class MyProblem(Problem):

    def __init__(self, **kwargs):
        super().__init__(n_var=10, n_obj=1, n_constr=0, xl=-5, xu=5, **kwargs)

    def _evaluate(self, X, out, *args, **kwargs):

        def my_eval(x):
            return (x ** 2).sum()

        params = [[X[k]] for k in range(len(X))]
        F = pool.starmap(my_eval, params)
        out["F"] = np.array(F)


if __name__ == '__main__':
    pool = ThreadPool(8)
    problem = MyProblem()
    res = minimize(problem, PSO(), seed=1, n_gen=100)
    print('Threads:', res.exec_time)

   Output is as follows ：

Threads: 1.0212376117706299

4.2 Dask

import numpy as np
from dask.distributed import Client
from pymoo.algorithms.soo.nonconvex.pso import PSO
from pymoo.core.problem import Problem
from pymoo.optimize import minimize


class MyProblem(Problem):

    def __init__(self, *args, **kwargs):
        super().__init__(n_var=10, n_obj=1, n_constr=0, xl=-5, xu=5,
                         elementwise_evaluation=False, *args, **kwargs)

    def _evaluate(self, X, out, *args, **kwargs):
        def fun(x):
            return np.sum(x ** 2)

        jobs = [client.submit(fun, x) for x in X]
        out["F"] = np.row_stack([job.result() for job in jobs])


if __name__ == '__main__':
    client = Client(processes=False)
    problem = MyProblem()
    res = minimize(problem, PSO(), seed=1, n_gen=100)
    print('Dask:', res.exec_time)
    client.close()

   Output is as follows ：

Dask: 19.102460861206055

reference

【1】https://pymoo.org/problems/parallelization.html
【2】https://blog.csdn.net/u013066730/article/details/105821888