具有管理器和异步方法的 multiprocessing.pool

Question

我正在尝试使用 Manager(( 在进程之间共享字典并尝试了以下代码：

from multiprocessing import Manager, Pool
def f(d):
d['x'] += 2
if __name__ == '__main__':
manager = Manager()
d = manager.dict()
d['x'] = 2
p= Pool(4)
for _ in range(2000):
p.map_async(f, (d,))  #apply_async, map
p.close()
p.join()
print (d)  # expects this result --> {'x': 4002}

使用 map_async 和 apply_async，打印的结果总是不同的(例如 {'x'： 3838}， {'x'： 3770}(。 但是，使用 map 将给出预期的结果。 另外，我尝试使用进程而不是池，结果也不同。

有什么见解吗？ 非阻塞部分和竞争条件上的内容不是由经理处理的？

Answer 1

当您调用map(而不是map_async(时，它将阻塞，直到处理器完成您传递的所有请求，在您的情况下，这只是对函数f的一次调用。因此，即使您的池大小为 4，您实际上也是一次执行一个 2000 个进程。要实际并行执行，您应该执行单个p.map(f, [d]*2000)而不是循环。

但是当你调用map_async时，你不会阻塞，而是返回一个结果对象。对结果对象的get调用将阻塞，直到进程完成，并将返回函数调用的结果。因此，现在您一次最多运行 4 个进程。但是，对字典的更新不会跨处理器序列化。我已经修改了代码以使用多处理锁强制序列化d[x] += 2。您将看到结果现在是 4002。

from multiprocessing import Manager, Pool, Lock

def f(d):
lock.acquire()
d['x'] += 2
lock.release()
def init(l):
global lock
lock = l
if __name__ == '__main__':
with Manager() as manager:
d = manager.dict()
d['x'] = 2
lock = Lock()
p = Pool(4, initializer=init, initargs=(lock,)) # Create the multiprocessing lock that is sharable by all the processes
results = [] # if the function returnd a result we wanted
for _ in range(2000):
results.append(p.map_async(f, (d,)))  #apply_async, map
"""
for i in range(2000): # if the function returned a result we wanted
results[i].get() # wait for everything to finish
"""
p.close()
p.join()
print(d)