我有一个简单的服务器:
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)