python - 处理多处理池中的 worker 死亡

Question

我有一个简单的服务器:

from multiprocessing import Pool, TimeoutError
import time
import os


if __name__ == '__main__':
    # start worker processes
    pool = Pool(processes=1)

    while True:
        # evaluate "os.getpid()" asynchronously
        res = pool.apply_async(os.getpid, ())  # runs in *only* one process
        try:
            print(res.get(timeout=1))             # prints the PID of that process
        except TimeoutError:
            print('worker timed out')

        time.sleep(5)

    pool.close()
    print("Now the pool is closed and no longer available")
    pool.join()
    print("Done")

如果我运行这个，我会得到类似的东西:

47292
47292

然后我 kill 47292在服务器运行时。启动了一个新的工作进程，但服务器的输出是:

47292
47292
worker timed out
worker timed out
worker timed out

池仍在尝试向旧工作进程发送请求。

我已经完成了一些在服务器和工作线程中捕获信号的工作，我可以得到稍微好一点的行为，但是在工作线程被杀死后，服务器似乎仍在等待关闭时死掉的 child (即 pool.join() 永远不会结束) .

处理 worker 死亡的正确方法是什么？

从服务器进程正常关闭工作人员似乎只有在没有工作人员死亡的情况下才有效。

(在 Python 3.4.4 上，但如果有帮助，很乐意升级。)

更新:
有趣的是，如果池是用 processes=2 创建的，并且您杀死一个工作进程，等待几秒钟并杀死另一个，则不会发生此工作超时问题。但是，如果您快速连续杀死两个工作进程，那么“工作人员超时”问题就会再次出现。

也许相关的是，当问题发生时，杀死服务器进程将使工作进程继续运行。

Answer 1

这种行为来自 multiprocessing.Pool 的设计.当你杀死一个 worker 时，你可能会杀死持有 call_queue.rlock 的那个人。 .当这个进程在持有锁的情况下被杀死时，其他进程将永远无法读取 call_queue不再，打破 Pool因为它无法再与其 worker 通信。
所以实际上没有办法杀死一个 worker 并确保你的 Pool之后仍然可以，因为您可能会陷入僵局。
multiprocessing.Pool不处理 worker 死亡。您可以尝试使用 concurrent.futures.ProcessPoolExecutor相反(使用稍微不同的 API)，它默认处理进程的失败。当进程在 ProcessPoolExecutor 中终止时，整个执行程序都关闭了，你会得到一个 BrokenProcessPool错误。

请注意，此实现中还有其他死锁，应在 loky 中修复。 . (免责声明:我是这个库的维护者)。另外，loky让您调整现有 executor 的大小使用 ReusablePoolExecutor和方法 _resize .如果您有兴趣，请告诉我，我可以从这个包开始为您提供一些帮助。 (我意识到我们仍然需要在文档上做一些工作...... 0_0)