Python多处理池类目标函数的相互作用与神经进化

2024-03-28 23:29:13 发布

您现在位置:Python中文网/ 问答频道 /正文

警告,我要说得尽可能具体还需要很长时间。你知道吗


精确问题:这是一个多处理问题。我已经确保我的类在以前的实验中都表现得像构建的/预期的那样。你知道吗

编辑:事先说。你知道吗


当我在线程环境中运行我的问题的玩具示例时,一切都会正常运行;然而,当我转换到真正的问题时,代码会中断。具体来说,我得到一个TypeError: can't pickle _thread.lock objects错误。满堆在底部。你知道吗

我在这里的线程需求与我从--https://github.com/CMA-ES/pycma/issues/31改编代码的示例略有不同。在这个例子中,我们有一个适应度函数,它可以由每个求值独立调用,并且没有一个函数调用可以相互作用。然而,在我的实际问题中,我们尝试使用遗传算法优化神经网络权重。遗传算法将建议潜在的权重,我们需要在我们的环境中评估这些神经网络控制器的权重。在单线程的情况下,我们可以有一个环境,在这个环境中,我们使用一个简单的for循环来评估权重:[nn.evaluate(weights) for weights in potential_candidates],找到性能最好的个体,并在下一轮变异中使用这些权重。但是,我们不能简单地在线程环境中进行一个模拟。你知道吗

因此,我没有传入一个函数来进行评估,而是传入了一个函数列表(每个函数对应一个函数,其中环境是相同的,但是我们已经分叉了进程,这样通信流就不会在个体之间进行交互)

还有一件需要注意的事: 我正在使用一个构建用于并行评估的数据结构,它来自于neat

从整齐、平行导入ParallelEvaluator#使用多处理池你知道吗

玩具示例代码:

NPARAMS = nn.flat_init_weights.shape[0]    # make this a 1000-dimensional problem.
NPOPULATION = 5                            # use population size of 5.
MAX_ITERATION = 100                        # run each solver for 100 function calls.

import time
from neat.parallel import ParallelEvaluator  # uses multiprocessing.Pool
import cma

def fitness(x):
    time.sleep(0.1)
    return sum(x**2)

# # serial evaluation of all solutions
# def serial_evals(X, f=fitness, args=()):
#     return [f(x, *args) for x in X]

# parallel evaluation of all solutions
def _evaluate2(self, weights, *args):
    """redefine evaluate without the dependencies on neat-internal data structures
    """
    jobs = []
    for i, w in enumerate(weights):
        jobs.append(self.pool.apply_async(self.eval_function[i], (w, ) + args))

    return [job.get() for job in jobs]

ParallelEvaluator.evaluate2 = _evaluate2
parallel_eval = ParallelEvaluator(12, [fitness]*NPOPULATION)

# time both
for eval_all in [parallel_eval.evaluate2]:
    es = cma.CMAEvolutionStrategy(NPARAMS * [1], 1, {'maxiter': MAX_ITERATION, 
                                                     'popsize': NPOPULATION})
    es.disp_annotation()
    while not es.stop():
        X = es.ask()
        es.tell(X, eval_all(X))
    es.disp()

必要背景:

当我从玩具示例切换到我的真实代码时,上面的操作失败了。你知道吗

我的课程是:

LevelGenerator (simple GA class that implements mutate, etc)
GridGame (OpenAI wrapper; launches a Java server in which to run the simulation; 
          handles all communication between the Agent and the environment)
Agent    (neural-network class, has an evaluate fn which uses the NN to play a single rollout)
Objective (handles serializing/de-serializing weights: numpy <--> torch; launching the evaluate function)

# The classes get composed to get the necessary behavior:
env   = GridGame(Generator)
agent = NNAgent(env)                # NNAgent is a subclass of (Random) Agent)
obj   = PyTorchObjective(agent)

# My code normally all interacts like this in the single-threaded case:

def test_solver(solver): # Solver: CMA-ES, Differential Evolution, EvolutionStrategy, etc
    history = []
    for j in range(MAX_ITERATION):
        solutions = solver.ask() #2d-numpy array. (POPSIZE x NPARAMS)
        fitness_list = np.zeros(solver.popsize)
        for i in range(solver.popsize):
            fitness_list[i] = obj.function(solutions[i], len(solutions[i]))
        solver.tell(fitness_list)
        result = solver.result() # first element is the best solution, second element is the best fitness
        history.append(result[1])

        scores[j] = fitness_list

    return history, result

所以,当我尝试跑步时:

NPARAMS = nn.flat_init_weights.shape[0]        
NPOPULATION = 5                                
MAX_ITERATION = 100                            

_x = NNAgent(GridGame(Generator))

gyms = [_x.mutate(0.0) for _ in range(NPOPULATION)]
objs = [PyTorchObjective(a) for a in gyms]

def evaluate(objective, weights):
    return objective.fun(weights, len(weights))

import time
from neat.parallel import ParallelEvaluator  # uses multiprocessing.Pool
import cma

def fitness(agent):
    return agent.evalute()

# # serial evaluation of all solutions
# def serial_evals(X, f=fitness, args=()):
#     return [f(x, *args) for x in X]

# parallel evaluation of all solutions
def _evaluate2(self, X, *args):
    """redefine evaluate without the dependencies on neat-internal data structures
    """
    jobs = []
    for i, x in enumerate(X):
        jobs.append(self.pool.apply_async(self.eval_function[i], (x, ) + args))

    return [job.get() for job in jobs]

ParallelEvaluator.evaluate2 = _evaluate2
parallel_eval = ParallelEvaluator(12, [obj.fun for obj in objs])
# obj.fun takes in the candidate weights, loads them into the NN, and then evaluates the NN in the environment.

# time both
for eval_all in [parallel_eval.evaluate2]:
    es = cma.CMAEvolutionStrategy(NPARAMS * [1], 1, {'maxiter': MAX_ITERATION, 
                                                     'popsize': NPOPULATION})
    es.disp_annotation()
    while not es.stop():
        X = es.ask()
        es.tell(X, eval_all(X, NPARAMS))
    es.disp()

我得到以下错误:

TypeError                            Traceback (most recent call last)
<ipython-input-57-3e6b7bf6f83a> in <module>
      6     while not es.stop():
      7         X = es.ask()
----> 8         es.tell(X, eval_all(X, NPARAMS))
      9     es.disp()

<ipython-input-55-2182743d6306> in _evaluate2(self, X, *args)
     14         jobs.append(self.pool.apply_async(self.eval_function[i], (x, ) + args))
     15 
---> 16     return [job.get() for job in jobs]

<ipython-input-55-2182743d6306> in <listcomp>(.0)
     14         jobs.append(self.pool.apply_async(self.eval_function[i], (x, ) + args))
     15 
---> 16     return [job.get() for job in jobs]

~/miniconda3/envs/thesis/lib/python3.7/multiprocessing/pool.py in get(self, timeout)
    655             return self._value
    656         else:
--> 657             raise self._value
    658 
    659     def _set(self, i, obj):

~/miniconda3/envs/thesis/lib/python3.7/multiprocessing/pool.py in _handle_tasks(taskqueue, put, outqueue, pool, cache)
    429                         break
    430                     try:
--> 431                         put(task)
    432                     except Exception as e:
    433                         job, idx = task[:2]

~/miniconda3/envs/thesis/lib/python3.7/multiprocessing/connection.py in send(self, obj)
    204         self._check_closed()
    205         self._check_writable()
--> 206         self._send_bytes(_ForkingPickler.dumps(obj))
    207 
    208     def recv_bytes(self, maxlength=None):

~/miniconda3/envs/thesis/lib/python3.7/multiprocessing/reduction.py in dumps(cls, obj, protocol)
     49     def dumps(cls, obj, protocol=None):
     50         buf = io.BytesIO()
---> 51         cls(buf, protocol).dump(obj)
     52         return buf.getbuffer()
     53 

TypeError: can't pickle _thread.lock objects

我在这里还读到,这可能是因为这是一个类函数--TypeError: can't pickle _thread.lock objects--所以我创建了全局范围的适应度函数def fitness(agent): return agent.evalute(),但这也不起作用。你知道吗

我认为这个错误可能来自这样一个事实:最初,我在PyTorchObjective类中将evaluate函数作为lambda函数,但当我更改时,它仍然中断。你知道吗

如果您有任何见解,我们将不胜感激,感谢您阅读这堵巨大的文字墙。你知道吗


Tags: the函数inselfobjforreturnes
1条回答
网友
1楼 · 发布于 2024-03-28 23:29:13

您没有使用多线程。You are using multiple processes.

传递给apply_async的所有参数(包括函数本身)都在后台序列化(pickle)并通过IPC通道传递给工作进程(请阅读^{} documentation了解详细信息)。因此,您不能传递任何与本质上是本地进程的事物相关联的实体。这包括大多数同步原语,因为它们必须使用锁来执行原子操作。你知道吗

每当发生这种情况(as many other questions on this error message show)时,您可能会试图变得太聪明,并向并行化框架传递一个已经内置了并行化逻辑的对象。


如果您想用这种“并行化对象”创建“多级并行化”,您最好:

相关问题 更多 >