Issue with parallelization via multiprocessing.Pool

[WeilinDeng]

I was trying to implement parallelization via multiprocessing.Pool and ThreadPool. The ThreadPool works for my problem, but multiprocessing.Pool did not. Can you pls help?

Below is the code:

import numpy as np
import matplotlib.pyplot as plt
import multiprocessing
from multiprocessing.pool import ThreadPool
from pymoo.core.problem import StarmapParallelization
from pymoo.core.problem import ElementwiseProblem
from pymoo.core.variable import Choice
from pymoo.visualization.scatter import Scatter
from pymoo.algorithms.moo.nsga2 import NSGA2, RankAndCrowdingSurvival
from pymoo.core.mixed import MixedVariableMating, MixedVariableSampling, MixedVariableDuplicateElimination
from pymoo.optimize import minimize
from pymoo.termination import get_termination
from pymoo.operators.crossover.sbx import SBX
from pymoo.operators.mutation.pm import PM


mat_prop = {"mat1": [0.08, 0.2, 0.47, 0.6, 0.78, 1.1, 1.3, 1.59, 1.87],
            "mat2": [0.15, 0.23, 0.4, 0.6, 0.82, 1.2, 1.6, 1.76, 2.01]
            }

class MyProblem(ElementwiseProblem):

    def __init__(self, **kwargs):
        vars = {
            "x": Choice(options=mat_prop["mat1"]),
            "y": Choice(options=mat_prop["mat2"])
        }
        super().__init__(vars=vars,
                         n_obj=2,
                         n_ieq_constr=1,
                         **kwargs)

    def _evaluate(self, X, out, *args, **kwargs):
        x, y = X["x"], X["y"]
        f1 = x**4 + y**4 - np.exp(np.cos(x-y)*np.sin(x+y))
        f2 = (x-2*y+1)**3 - (2*x+y-1)**3 + np.exp(np.sin(x-y)*np.cos(x+y))
        g1 = x + y - 2.8
        
        out["F"] = [f1, f2]
        out["G"] = [g1]


if __name__ == "__main__":
    # initialize the thread pool and create the runner
    # n_threads = 4
    # pool = ThreadPool(n_threads)

    n_proccess = 4
    pool = multiprocessing.Pool(n_proccess)

    runner = StarmapParallelization(pool.starmap)
    problem = MyProblem(elementwise_runner=runner)

    algorithm = NSGA2(pop_size=20,
                      n_offsprings=20,
                      sampling=MixedVariableSampling(),
                      crossover=SBX(prob=0.9, eta=10),
                      mutation=PM(prob=0.05, eta=15),
                      mating=MixedVariableMating(
                          eliminate_duplicates=MixedVariableDuplicateElimination()),
                      eliminate_duplicates=MixedVariableDuplicateElimination(),
                      survival=RankAndCrowdingSurvival()
                      )

    termination = get_termination("n_gen", 200)

    res = minimize(problem,
                   algorithm,
                   termination,
                   seed=1,
                   save_history=True,
                   verbose=True)

    print('Threads:', res.exec_time)
    pool.close()

No issue with

n_threads = 4
pool = ThreadPool(n_threads)

Bur when using

    n_proccess= 4
    pool = multiprocessing.Pool(n_proccess)

I get the error:

Traceback (most recent call last):
File "C:\Users\W\Documents\ma-optimization-template-help-snipt.py", line 67, in
res = minimize(problem,
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\optimize.py", line 67, in minimize
res = algorithm.run()
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\algorithm.py", line 141, in run
self.next()
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\algorithm.py", line 161, in next
self.evaluator.eval(self.problem, infills, algorithm=self)
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\evaluator.py", line 69, in eval
self._eval(problem, pop[I], evaluate_values_of, **kwargs)
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\evaluator.py", line 90, in _eval
out = problem.evaluate(X, return_values_of=evaluate_values_of, return_as_dictionary=True, **kwargs)
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\problem.py", line 187, in evaluate
_out = self.do(X, return_values_of, *args, **kwargs)
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\problem.py", line 227, in do
self._evaluate_elementwise(X, out, *args, **kwargs)
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\problem.py", line 245, in _evaluate_elementwise
elems = self.elementwise_runner(f, X)
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\problem.py", line 37, in call
return list(self.starmap(f, [[x] for x in X]))
File "C:\Users\W\Anaconda3\envs\ma\lib\multiprocessing\pool.py", line 375, in starmap
return self._map_async(func, iterable, starmapstar, chunksize).get()
File "C:\Users\W\Anaconda3\envs\ma\lib\multiprocessing\pool.py", line 774, in get
raise self._value
File "C:\Users\W\Anaconda3\envs\ma\lib\multiprocessing\pool.py", line 540, in _handle_tasks
put(task)
File "C:\Users\W\Anaconda3\envs\ma\lib\multiprocessing\connection.py", line 206, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "C:\Users\W\Anaconda3\envs\ma\lib\multiprocessing\reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
File "C:\Users\W\Anaconda3\envs\ma\lib\site-packages\pymoo\core\problem.py", line 41, in getstate
state.pop("starmap")
KeyError: 'starmap'

[blankjul]

I copied this from the tutorial here: https://pymoo.org/problems/parallelization.html#Processes

from pymoo.algorithms.soo.nonconvex.ga import GA
from pymoo.optimize import minimize


# initialize the thread pool and create the runner
n_proccess = 8
pool = multiprocessing.Pool(n_proccess)
runner = StarmapParallelization(pool.starmap)

# define the problem by passing the starmap interface of the thread pool
problem = MyProblem(elementwise_runner=runner)

res = minimize(problem, GA(), termination=("n_gen", 200), seed=1)
print('Threads:', res.exec_time)

pool.close()

Does this resolve the issue?

[WeilinDeng]

Thanks for your reply, Julian.

I was aware of the tutorial example and it worked perfectly. But when I copied the same "multiprocessing" definition for my mixed variable problem, it gave an error of "KeyError: 'starmap'" as shown in my thread. Can you please check my snippet code to see why by any chance? Thanks again!

[blankjul]

For MixedVariableProblem please use the code on the very bottom which is the custom parallelization (directly in the problem itself). For mixed variables we had to write a new problem class which does not support all the default features such as parallelization.

[blankjul]

from pymoo.core.problem import Problem

pool = ThreadPool(8)

class MyProblem(Problem):

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

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

        # define the function
        def my_eval(x):
            return (x ** 2).sum()

        # prepare the parameters for the pool
        params = [[X[k]] for k in range(len(X))]

        # calculate the function values in a parallelized manner and wait until done
        F = pool.starmap(my_eval, params)

        # store the function values and return them.
        out["F"] = np.array(F)

problem = MyProblem()

[shoaib42]

@blankjul
Looks like something is iffy, I tried multiprocess myself and then the code from the tutorial. it just fails to work. Seems like a get attribute is missing. I want to move away from DEAP and back to pymoo, so that I can keep screwdrivers/wrenches in the toolbox and not functions :)

from pymoo.algorithms.soo.nonconvex.ga import GA
from pymoo.core.problem import StarmapParallelization
from pymoo.optimize import minimize
from pymoo.core.problem import ElementwiseProblem

class MyProblem(ElementwiseProblem):

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

    def _evaluate(self, x, out, *args, **kwargs):
         out["F"] = (x ** 2).sum()
# initialize the thread pool and create the runner
n_proccess = 8
pool = multiprocessing.Pool(n_proccess)
runner = StarmapParallelization(pool.starmap)

# define the problem by passing the starmap interface of the thread pool
problem = MyProblem(elementwise_runner=runner)

res = minimize(problem, GA(), termination=("n_gen", 200), seed=1)
print('Threads:', res.exec_time)

pool.close()

The error

AttributeError: Can't get attribute 'MyProblem' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>
Traceback (most recent call last):
  File "/usr/local/Cellar/[email protected]/3.12.4/Frameworks/Python.framework/Versions/3.12/lib/python3.12/multiprocessing/process.py", line 314, in _bootstrap
    self.run()
  File "/usr/local/Cellar/[email protected]/3.12.4/Frameworks/Python.framework/Versions/3.12/lib/python3.12/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/Cellar/[email protected]/3.12.4/Frameworks/Python.framework/Versions/3.12/lib/python3.12/multiprocessing/pool.py", line 114, in worker
    task = get()
           ^^^^^
  File "/usr/local/Cellar/[email protected]/3.12.4/Frameworks/Python.framework/Versions/3.12/lib/python3.12/multiprocessing/queues.py", line 389, in get
    return _ForkingPickler.loads(res)

[blankjul]

That's not really a pymoo problem but a serializiation problem. You need tell Python where the main method starts in order to serialize the object correctly:

import multiprocessing

from pymoo.algorithms.soo.nonconvex.ga import GA
from pymoo.core.problem import ElementwiseProblem
from pymoo.core.problem import StarmapParallelization
from pymoo.optimize import minimize


class MyProblem(ElementwiseProblem):

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

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


if __name__ == "__main__":
    # initialize the thread pool and create the runner
    n_proccess = 8
    pool = multiprocessing.Pool(n_proccess)
    runner = StarmapParallelization(pool.starmap)

    # define the problem by passing the starmap interface of the thread pool
    problem = MyProblem(elementwise_runner=runner)

    res = minimize(problem, GA(), termination=("n_gen", 200), seed=1)
    print('Threads:', res.exec_time)

    pool.close()

does that work for you?

[shoaib42]

Haven't tried it but I can definitely call myself an idiot for trying the examples on a notebook and not a py script. Thanks!

PS - I opened a PR, looks like some tests are failing on your/github end, my change is simple, fixing the doc and example for visualization #633