当我多处理.pool.apply_async比我有处理器多次时会发生什么

Question

我有以下设置:

results = [f(args) for _ in range(10**3)]

但是,f(args)需要很长时间才能计算出来.所以我想抛出多处理.我想这样做:

pool = mp.pool(mp.cpu_count() -1) # mp.cpu_count() -> 8
results = [pool.apply_async(f, args) for _ in range(10**3)]

显然,我的计算机上没有1000个处理器,所以我担心:
上述调用是否导致1000个进程同时竞争CPU时间或7个进程同时运行,迭代计算上一个f(args)调用结束时的下一个？

我想我可以做一些类似于pool.async_map(f, (args for _ in range(10**3)))获得相同结果的事情,但这篇文章的目的是了解其行为pool.apply_async

Answer 1

你的工作流程永远不会超过池中的工作人员(在你的情况下mp.cpu_count() - 1.如果你打电话apply_async,所有的工人都很忙,那么一旦工人释放,任务就会排队并执行.你可以看到这个一个简单的测试程序:

#!/usr/bin/python

import time
import multiprocessing as mp

def worker(chunk):
    print('working')
    time.sleep(10)
    return

def main():
    pool = mp.Pool(2)  # Only two workers
    for n in range(0, 8):
        pool.apply_async(worker, (n,))
        print("called it")
    pool.close()
    pool.join()

if __name__ == '__main__':
    main()

输出是这样的:

called it
called it
called it
called it
called it
called it
called it
called it
working
working
<delay>
working
working
<delay>
working 
working
<delay>
working
working

Answer 2

工作进程的数量完全由参数控制mp.pool().因此,如果mp.cpu_count()您的盒子上返回8,则将创建7个工作进程.

然后,所有pool方法(apply_async()其中)只使用那么多工作进程.在封面下,参数在主程序中被pickle并通过进程间管道发送到工作进程.这个隐藏的机器有效地创建了一个工作队列,固定数量的工作进程从中拉出工作描述(函数名+参数).

除此之外,它只是魔术;-)