gpt4 book ai didi

Python (3.7+) 多处理 : replace Pipe connection between master and workers with asyncio for IO concurrency

转载 作者:行者123 更新时间:2023-12-01 09:00:48 25 4
gpt4 key购买 nike

假设我们有一个以下玩具版本的主从管道来并行数据收集

# pip install gym
import gym
import numpy as np
from multiprocessing import Process, Pipe

def worker(master_conn, worker_conn):
master_conn.close()

env = gym.make('Pendulum-v0')
env.reset()

while True:
cmd, data = worker_conn.recv()

if cmd == 'close':
worker_conn.close()
break
elif cmd == 'step':
results = env.step(data)
worker_conn.send(results)

class Master(object):
def __init__(self):
self.master_conns, self.worker_conns = zip(*[Pipe() for _ in range(10)])
self.list_process = [Process(target=worker, args=[master_conn, worker_conn], daemon=True)
for master_conn, worker_conn in zip(self.master_conns, self.worker_conns)]
[p.start() for p in self.list_process]
[worker_conn.close() for worker_conn in self.worker_conns]

def go(self, actions):
[master_conn.send(['step', action]) for master_conn, action in zip(self.master_conns, actions)]
results = [master_conn.recv() for master_conn in self.master_conns]

return results

def close(self):
[master_conn.send(['close', None]) for master_conn in self.master_conns]
[p.join() for p in self.list_process]

master = Master()
l = []
T = 1000
for t in range(T):
actions = np.random.rand(10, 1)
results = master.go(actions)
l.append(len(results))

sum(l)

由于master和worker之间有Pipe连接,所以对于每个时间步,我们都必须通过Pipe向worker发送命令,然后worker发回结果。我们需要从长远来看这样做。由于频繁的通信,有时会有点慢。

因此,我想知道,如果我正确理解其功能,使用最新的 Python 功能 asyncio 结合 Process 来替换 Pipe,是否可能由于 IO 并发而带来加速。

最佳答案

Multiprocessing模块已经有并行任务处理的解决方案:multiprocessing.Pool

from multiprocessing import Pool

def f(x):
return x*x

if __name__ == '__main__':
with Pool(processes=4) as pool: # start 4 worker processes
print(pool.map(f, range(10))) # prints "[0, 1, 4,..., 81]"

您可以使用 multiprocessing.Queue 实现相同的目的。我相信这就是 pool.map() 的内部实现方式。

那么,multiprocessing.Queue 之间有什么区别?和 multiprocessing.PipeQueue 只是一个Pipe 加上一些锁定机制。因此,多个工作进程只能共享一个 Queue(或者更确切地说,两个 - 一个用于命令,一个用于结果),但使用 Pipe 每个进程都需要它自己的 管道(或一对,或双工),正是您现在所做的。

Queue 的唯一缺点是性能 - 因为所有进程共享一个队列互斥体,所以它对于许多进程来说不能很好地扩展。为了确保它可以处理数万个项目/秒,我会选择 Pipe,但对于经典的并行任务处理用例,我认为 Queue 或只是 Pool.map () 可能没问题,因为它们更容易使用。 (管理进程可能很棘手,asyncio 也没有让它变得更容易。)

希望有所帮助,我知道我回答的问题与您提出的问题有些不同:)

关于Python (3.7+) 多处理 : replace Pipe connection between master and workers with asyncio for IO concurrency,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/52467049/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com