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()