在Python中实现看门狗定时器的工具

Question

我正在写一些代码来测试多线程程序（这是学生的作业，可能会有bug），我想要能够检测到它们是否出现了死锁。正常运行时，这些程序会定期向标准输出（stdout）发送信息，所以这就比较简单了：如果在X秒内没有输出，就杀掉它并报告死锁。下面是这个函数的原型：

def run_with_watchdog(command, timeout):
    """Run shell command, watching for output.  If the program doesn't
     produce any output for <timeout> seconds, kill it and return 1.  
     If the program ends successfully, return 0."""

我可以自己写这个功能，但要做到正确有点棘手，所以如果能使用现有的代码我会更喜欢。有没有人写过类似的东西？

---

好的，下面是解决方案。 如果你在做类似的事情，子进程模块可能也会有用。

Answer 1

另一种解决方案：

class Watchdog:
  def __init__(self, timeout, userHandler=None): # timeout in seconds
    self.timeout = timeout
    if userHandler != None:
      self.timer = Timer(self.timeout, userHandler)
    else:
      self.timer = Timer(self.timeout, self.handler)

  def reset(self):
    self.timer.cancel()
    self.timer = Timer(self.timeout, self.handler)

  def stop(self):
    self.timer.cancel()

  def handler(self):
    raise self;

如果你想确保某个函数在少于 x 秒内完成，可以这样使用：

watchdog = Watchdog(x)
try
  ... do something that might hang ...
except Watchdog:
  ... handle watchdog error ...
watchdog.stop()

如果你经常执行某个操作，并且想确保它至少每 y 秒执行一次，可以这样使用：

def myHandler():
  print "Watchdog expired"

watchdog = Watchdog(y, myHandler)

def doSomethingRegularly():
  ...
  watchdog.reset()

Answer 2

这里有一个经过很少测试，但看起来可以用的解决方案：

import sys
import time
import pexpect
# From http://pypi.python.org/pypi/pexpect/

DEADLOCK = 1

def run_with_watchdog(shell_command, timeout):
    """Run <shell_command>, watching for output, and echoing it to stdout.
    If the program doesn't produce any output for <timeout> seconds,
    kill it and return 1.  If the program ends successfully, return 0.
    Note: Assumes timeout is >> 1 second. """

    child = pexpect.spawn('/bin/bash', ["-c", shell_command])
    child.logfile_read = sys.stdout
    while True:
        try:
            child.read_nonblocking(1000, timeout)
        except pexpect.TIMEOUT:
            # Child seems deadlocked.  Kill it, return 1.
            child.close(True)
            return DEADLOCK
        except pexpect.EOF:
            # Reached EOF, means child finished properly.
            return 0
        # Don't spin continuously.
        time.sleep(1)

if __name__ == "__main__":
    print "Running with timer..."
    ret = run_with_watchdog("./test-program < trace3.txt", 10) 
    if ret == DEADLOCK:
        print "DEADLOCK!"
    else:
        print "Finished normally"

Answer 3

你可以使用 expect（tcl语言）或者 pexpect（python语言）来实现这个功能。

import pexpect
c=pexpect.spawn('your_command')
c.expect("expected_output_regular_expression", timeout=10)