python - 在多处理池进程中运行代码

Question

如何处理 multiprocessing.Pool() 初始化错误？

总之，我用
pool = multiprocessing.Pool(os.cpu_count(), initializer=init)

def init(): 
   # stop everything if that fails with exception
   # init..

相关问题 How to handle initializer error in multiprocessing.Pool?

但它并没有展示如何干净地退出主进程。

是否可以从池中获取每个 Process 实例，并在每个实例中运行一个函数？请注意，这与 pool.map 不同，因为我不知道我的函数将在哪个进程中运行。

Answer 1

使用 Queue 怎么样？监听初始化失败？

例如:

from multiprocessing import Pool, Manager
from random import random

# an initialisation function that fails 10% of the time
def init(initdone):
    try:
        # fail sometimes!
        assert random() < 0.9
    except Exception as err:
        # record the failure
        initdone.put(err)
    finally:
        # record that initialisation was successful
        initdone.put(None)

# we need a manager to maintain the queue
with Manager() as manager:
    # somewhere to store initialisation state
    initresult = manager.Queue()

    # number of workers we want to wait for
    nprocs = 5

    with Pool(nprocs, initializer=init, initargs=(initresult,)) as pool:
        # wait for initializers to run
        for i in range(nprocs):
            res = initresult.get()
            # reraise if it failed (or whatever logic is appropriate)
            if res is not None:
                raise res

        # do something now we've got this pool set up
        print(sum(pool.map(int, range(20))))

我应该注意 multiprocessing 可以重新启动失败的工作进程，因此您可能会得到超过 nprocs 个条目进入队列。在 OP 的情况下，这可能没问题，因为他们说他们想在这种情况下中止，但在其他情况下可能并非如此。

我想不出有什么方法可以使队列中的条目少于 nprocs，如果您看到任何内容，请发表评论!请注意，我故意不在 init 中捕获 BaseException，因为这是由 multiprocessing 中的代码(以及其他一些代码)处理的，这似乎是明智的让它做正确的事。