工具實現在python

我在寫測試多線程程序（學生家庭作業 - 可能越野車）的一些代碼，一個看門狗定時器，並希望能夠當它們發生死鎖檢測。當正常運行，程序經常產生輸出到標準輸出，這樣就使得它相當簡單：如果沒有輸出X秒，殺了它，並報告死鎖。這裏的函數原型：工具實現在python

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."""

我可以寫我自己，但它是一個有點棘手得到正確的，所以我寧願儘可能使用現有的代碼。任何人都寫類似的東西？

好的，請參閱下面的解決方案。該子模塊也可能是相關的，如果你正在做類似的事情。

來源

2009-12-14 Victor Shnayder

可以使用期望（TCL）或Pexpect的（蟒蛇）來做到這一點。

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

來源

2009-12-14 05:41:00

這裏有一個非常輕微的測試，但似乎工作，解決方法：

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-14 06:57:45

另一種解決方案：

如果你想確保功能

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-22 13:42:33

工具實現在python

回答

相關問題