Python 多处理 : how to limit the number of waiting processes?

Question

当使用Pool.apply_async运行大量任务(大参数)时，进程被分配并进入等待状态，等待进程数没有限制。这可能会吃掉所有内存，如下例所示:

import multiprocessing
import numpy as np

def f(a,b):
    return np.linalg.solve(a,b)

def test():

    p = multiprocessing.Pool()
    for _ in range(1000):
        p.apply_async(f, (np.random.rand(1000,1000),np.random.rand(1000)))
    p.close()
    p.join()

if __name__ == '__main__':
    test()

我正在寻找一种方法来限制等待队列，这样只有有限数量的等待进程，并且在等待队列已满时阻塞 Pool.apply_async。

Answer 1

multiprocessing.Pool 有一个类型为 multiprocessing.Queue 的 _taskqueue 成员，它有一个可选的 maxsize 参数;不幸的是，它在没有设置 maxsize 参数集的情况下构造它。

我建议使用 multiprocessing.Pool.__init__ 的复制粘贴将 multiprocessing.Pool 子类化，将 maxsize 传递给 _taskqueue 构造函数。

猴子修补对象(池或队列)也可以，但您必须猴子修补 pool._taskqueue._maxsize 和 pool._taskqueue._sem 所以它会很脆弱:

pool._taskqueue._maxsize = maxsize
pool._taskqueue._sem = BoundedSemaphore(maxsize)