在多进程池进程中运行代码

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中的代码(以及其他一些代码)处理的，让它做正确的事情似乎是明智的。