Skip to content

Instantly share code, notes, and snippets.

@gawen
Created April 11, 2012 12:58
Show Gist options
  • Save gawen/2359156 to your computer and use it in GitHub Desktop.
Save gawen/2359156 to your computer and use it in GitHub Desktop.
Python Watchdog with gevent
""" Use watchdogs to make your code safer. (http://en.wikipedia.org/wiki/Watchdog_timer)
Usage:
with watchdog.Watchdog(duration = 1.0):
# Some code which takes less than 1s
# If it takes more, the exception Watchdog.Reset will be raised
# To notify the watchdog the section is still alive, call the method
# 'tick' to reset its internal timer.
Example:
with watchdog.Watchdog(duration = 1.0):
print "Hello, World! :-)"
will print "Hello, World! :-)".
with watchdog.Watchdog(duration = 1.0):
time.sleep(2.0)
will raises the exception Watchdog.Reset
with watchdog.Watchdog(duration = 1.0) as w:
for i in range(10):
time.sleep(0.9)
w.tick()
won't raise any exception.
"""
import gevent
class Watchdog(object):
class Reset(Exception):
pass
def __init__(self, duration):
super(Watchdog, self).__init__()
self.duration = duration
self.timeout = None
def start(self):
assert not self.timeout, "The watchdog is already started."
self.timeout = gevent.Timeout(self.duration, Watchdog.Reset)
self.timeout.start()
def tick(self):
self.stop()
self.start()
def stop(self):
if self.timeout:
self.timeout.cancel()
self.timeout = None
def __enter__(self):
self.start()
return self
def __exit__(self, *exc):
self.stop()
""" nosetests for gevent_watchdog.py """
import nose
import gevent_watchdog as watchdog
import contextlib
import time
import gevent
@contextlib.contextmanager
def assert_exception(exc = None):
try:
yield
except BaseException, e:
if exc and not isinstance(e, exc):
raise
else:
raise AssertionError("No exception had been raised.")
def test_init():
with watchdog.Watchdog(0.1):
pass
def test_fail():
with assert_exception(watchdog.Watchdog.Reset):
with watchdog.Watchdog(0.1):
gevent.sleep(0.2)
def test_tick():
with watchdog.Watchdog(0.1) as w:
for i in range(4):
gevent.sleep(0.09)
w.tick()
@gawen
Copy link
Author

gawen commented Apr 11, 2012

There's no direct implementation of the 'tick' method in the gevent.Timeout class.

@denik
Copy link

denik commented Apr 11, 2012

I see, did not notice tick method. You can probably emulate it though, by cancelling/starting the timeout.

@gawen
Copy link
Author

gawen commented Apr 11, 2012

I fixed this. The code is indeed lighter.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment