I would like to implement a simple watchdog timer in Python with two use cases:
x secondsy secondsHow do I do that?
Just publishing my own solution to this:
from threading import Timer
class Watchdog(Exception):
def __init__(self, timeout, userHandler=None): # timeout in seconds
self.timeout = timeout
self.handler = userHandler if userHandler is not None else self.defaultHandler
self.timer = Timer(self.timeout, self.handler)
self.timer.start()
def reset(self):
self.timer.cancel()
self.timer = Timer(self.timeout, self.handler)
self.timer.start()
def stop(self):
self.timer.cancel()
def defaultHandler(self):
raise self
Usage if you want to make sure function finishes in less than x seconds:
watchdog = Watchdog(x)
try:
# do something that might take too long
except Watchdog:
# handle watchdog error
watchdog.stop()
Usage if you regularly execute something and want to make sure it is executed at least every y seconds:
import sys
def myHandler():
print "Whoa! Watchdog expired. Holy heavens!"
sys.exit()
watchdog = Watchdog(y, myHandler)
def doSomethingRegularly():
# make sure you do not return in here or call watchdog.reset() before returning
watchdog.reset()
Timer() here?
reset method. In your __init__ your Timer switches based on if userHandler (which should be is not None BTW) is supplied, in which case it's used. In reset - it's always self.handler that will then be used...
Watchdog raises the Exception (itself) in a separate thread (the Timer thread), so the try/except in the usage example is useless. The Watchdog also does not interrupt the long operation, which would be the point of a watchdog.signal.alarm() sets a timeout for your program, and you can call it in your main loop, and set it to the greater of the two times you are prepared to tolerate:
import signal
while True:
signal.alarm(10)
infloop()
Here is a wdt I use in my app without class. It has no way to stop it:
from threading import Event, Thread
def wdt(time, callback):
# a reset flag
reset_e = Event()
# a function to reset the wdt
def reset(): reset_e.set()
# the function to run in a differen thread
def checker():
# check if reset flag is set.
# wait for specified time to give chance to reset.
while reset_e.wait(time):
# it was set in time. clear and wait again
reset_e.clear()
# time run out.
callback()
# the event is not set by default. Set it
reset()
# create and start the wdt
t = Thread(target=checker)
t.start()
# return the resetter
return reset
# The callback to run if wdt is not reset
def bark():
print('woof')
# Test
import time
reset = wdt(1.0, bark)
time.sleep(0.9)
reset()
time.sleep(2.0)
WatchdogTimerimplementation that creates only one thread