超时一个子进程

6 投票
1 回答
1994 浏览
提问于 2025-04-16 05:06

我意识到这可能和使用'subprocess'模块设置超时的问题重复。如果是这样,我很抱歉,只是想澄清一些事情。

我正在创建一个子进程,我希望它能在一定时间内运行。如果在这个时间内没有完成,我希望它能抛出一个错误。像下面这样的代码可以工作吗?还是说我们必须像其他问题中提到的那样使用信号?提前谢谢!:

def run(self):
    self.runTestCmd()
    self.waitTestComplete(self.timeout)

def runTestCmd(self):
    self.proc = subprocess.Popen("./configure", shell=True)

def waitTestComplete(self, timeout):
    st = time.time() 
    while (time.time()-st) < timeout:
        if self.proc.poll() == 0:
            return True
        else:
            time.sleep(2)
    raise TestError("timed out waiting for test to complete")

1 个回答

6

这样做是可以的,但有个问题。即使你放弃了这个过程,它仍然会继续执行你之前要求它做的事情。如果你真的想让它停止,你需要给这个过程发送一个信号来结束它。

因为你正在启动一个新过程(./configure,这可能是一个配置脚本),而这个新过程又会创建很多子进程,所以事情会变得有点复杂。

import os

def runTestCmd(self):
    self.proc = subprocess.Popen(["./configure"], shell=False,
                                 preexec_fn=os.setsid)

然后,使用 os.kill(-process.pid, signal.SIGKILL) 可以结束所有的子进程。简单来说,你是在使用 preexec_fn 让你的新子进程 获取自己的会话组。接着,你就可以给这个会话组里的所有进程发送信号。

很多会生成子进程的程序都知道在自己结束之前需要清理这些子进程。所以,如果可以的话,尽量对它们好一点。你可以先试试 os.signal(-process.pid, signal.SIGTERM),等一两秒看这个进程是否会退出,如果不行再用 SIGKILL。大概是这样的:

import time, os, errno, signal

def waitTestComplete(self, timeout):
    st = time.time() 
    while (time.time()-st) < timeout:
        if self.proc.poll() is not None:  # 0 just means successful exit
            # Only return True if process exited successfully,
            # otherwise return False.
            return self.proc.returncode == 0
        else:
            time.sleep(2)
    # The process may exit between the time we check and the
    # time we send the signal.
    try:
        os.kill(-self.proc.pid, signal.SIGTERM)
    except OSError, e:
        if e.errno != errno.ESRCH:
            # If it's not because the process no longer exists,
            # something weird is wrong.
            raise
    time.sleep(1)
    if self.proc.poll() is None: # Still hasn't exited.
        try:
            os.kill(-self.proc.pid, signal.SIGKILL)
        except OSError, e:
            if e.errno != errno.ESRCH:
                raise
    raise TestError("timed out waiting for test to complete")

顺便提一句,除非你绝对确定需要,不要使用 shell=True。真的,这个选项非常危险,可能会引发很多安全问题和奇怪的行为。

撰写回答