I have this decorator:
def timed_out(timeout):
def decorate(f):
if not hasattr(signal, "SIGALRM"):
return f
def handler(signum, frame):
raise TimedOutExc()
@functools.wraps(f)
def new_f(*args, **kwargs):
old = signal.signal(signal.SIGALRM, handler)
signal.alarm(timeout)
try:
result = f(*args, **kwargs)
finally:
signal.signal(signal.SIGALRM, old)
signal.alarm(0)
return result
new_f.func_name = f.func_name
return new_f
return decorate
The code only does anything on linux, though, as on windows, there is no SIGALRM. What would be the simplest way to have this code work in Windows as well?
It's not very pretty, but I had to do something similar in a cross-platform way, and I came up with using a separate thread. Signal based systems did not work on all platforms reliably.
Use of this class could be wrapped up in a decorator, or made into a with context handler.
YMMV.
#!/usr/bin/env python2.7
import time, threading
class Ticker(threading.Thread):
"""A very simple thread that merely blocks for :attr:`interval` and sets a
:class:`threading.Event` when the :attr:`interval` has elapsed. It then waits
for the caller to unset this event before looping again.
Example use::
t = Ticker(1.0) # make a ticker
t.start() # start the ticker in a new thread
try:
while t.evt.wait(): # hang out til the time has elapsed
t.evt.clear() # tell the ticker to loop again
print time.time(), "FIRING!"
except:
t.stop() # tell the thread to stop
t.join() # wait til the thread actually dies
"""
# SIGALRM based timing proved to be unreliable on various python installs,
# so we use a simple thread that blocks on sleep and sets a threading.Event
# when the timer expires, it does this forever.
def __init__(self, interval):
super(Ticker, self).__init__()
self.interval = interval
self.evt = threading.Event()
self.evt.clear()
self.should_run = threading.Event()
self.should_run.set()
def stop(self):
"""Stop the this thread. You probably want to call :meth:`join` immediately
afterwards
"""
self.should_run.clear()
def consume(self):
was_set = self.evt.is_set()
if was_set:
self.evt.clear()
return was_set
def run(self):
"""The internal main method of this thread. Block for :attr:`interval`
seconds before setting :attr:`Ticker.evt`
.. warning::
Do not call this directly! Instead call :meth:`start`.
"""
while self.should_run.is_set():
time.sleep(self.interval)
self.evt.set()