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

前端之家收集整理的这篇文章主要介绍了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)

原文链接:https://www.f2er.com/python/438779.html

猜你在找的Python相关文章