2

マルチスレッドプログラム（学生の宿題-バグの可能性が高い）をテストするためのコードを書いていますが、デッドロックを検出できるようにしたいと考えています。正しく実行されると、プログラムは定期的に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."""

自分で書くこともできますが、正しく理解するのは少し難しいので、可能であれば既存のコードを使用したいと思います。誰かが似たようなものを書いた？

わかりました。以下の解決策をご覧ください。同様のことをしている場合は、サブプロセスモジュールも関連している可能性があります。

4

3 に答える 3

5

これを行うには、expect（tcl）またはpexpect（python）を使用できます。

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 に答える