python – 在多处理池中处理工作者死亡

我有一个简单的服务器：

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

然后我在服务器运行时杀了47292.启动了新的工作进程,但服务器的输出是：

47292
47292
worker timed out
worker timed out
worker timed out

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

我已经完成了在服务器和工作程序中捕获信号的一些工作,我可以获得稍微好一点的行为,但服务器似乎仍在等待关闭死亡的孩子(即.pool.join()永远不会结束) .

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

如果没有工人死亡,那么从服务器进程中正常关闭工作人员似乎才有效.

(在Python 3.4.4上,但如果有帮助的话,很高兴升级.)

更新：
有趣的是,如果使用processes = 2创建池并且您杀死一个工作进程,等待几秒钟并终止另一个进程,则不会发生此工作者超时问题.但是,如果您快速连续杀死两个工作进程,则“工作超时”问题会再次出现.

也许相关的是,当问题发生时,终止服务器进程将使工作进程保持运行.

最佳答案

此行为来自multiprocessing.Pool的设计.当你杀死一个工人时,你可能会杀死一个持有call_queue.rlock的工人.当这个进程在持有锁时被终止时,其他进程将不再能够读入call_queue,因为它无法再与其工作者通信而破坏了Pool.
所以实际上没有办法杀死一个工人,并确保你的游泳池之后仍然没问题,因为你可能会陷入僵局.

multiprocessing.Pool不处理工人死亡.您可以尝试使用concurrent.futures.ProcesspoolExecutor(使用稍微不同的API)来处理默认情况下进程的失败.当进程在ProcesspoolExecutor中死亡时,整个执行程序将关闭并返回brokenProcesspool错误.

请注意,此实现中还有其他死锁,应在loky中修复.(免责声明：我是此库的维护者).此外,loky允许您使用ReusablePoolExecutor和方法_resize调整现有执行程序的大小.如果您有兴趣,请告诉我,从这个软件包开始,我可以为您提供一些帮助. (我意识到我们仍然需要对文档进行一些工作…… 0_0)

multiprocessing

python – 在多处理池中处理工作者死亡

相关文章