2

我正在编写一些代码来测试多线程程序（学生作业——可能是错误的），并且希望能够检测到它们何时死锁。正常运行时，程序会定期将输出输出到标准输出，因此非常简单：如果 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."""

我可以自己编写，但要正确使用有点棘手，所以如果可能的话，我更愿意使用现有代码。有人写过类似的吗？

好的，请参阅下面的解决方案。如果您正在做类似的事情，子流程模块也可能是相关的。

4

3 回答 3

5

You can use expect (tcl) or pexpect (python) to do this.

import pexpect
c=pexpect.spawn('your_command')
c.expect("expected_output_regular_expression", timeout=10)
于 2009-12-14T05:41:00.720 回答
1

这是一个经过轻微测试但看似有效的解决方案：

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"
于 2009-12-14T06:57:45.590 回答
0

另一种解决方案：

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()
于 2013-04-22T13:42:33.920 回答