Skip to content

Instantly share code, notes, and snippets.

@mgedmin
Last active January 24, 2016 01:13
Show Gist options
  • Save mgedmin/2839290 to your computer and use it in GitHub Desktop.
Save mgedmin/2839290 to your computer and use it in GitHub Desktop.
@timeout(10) decorator for unittest.TestCase methods that I never ended up using
import signal
import sys
from functools import contextmanager
@contextmanager
def timeout(p, seconds):
"""Kill subprocess ``p`` if the with block times out.
Usage example ::
p = subprocess.Popen([...], stdout=subprocess.PIPE)
with timeout(p, 30):
data = p.communicate()[0]
"""
finished = threading.Event()
def killIt():
finished.wait(seconds)
if not finished.is_set():
# print once to stderr, so user sees it, and once to stdout,
# so the right doctest fails
print("\nTimed out after {} seconds, killing process {}".format(
seconds, p.pid), file=sys.stderr)
print("Timed out after {} seconds, killing process {}".format(
seconds, p.pid))
p.kill()
t = threading.Thread(name='timeout-killer', target=killIt)
try:
t.start()
yield
finally:
finished.set()
import signal
import sys
from functools import wraps
class TimedOut(BaseException):
pass
def timeout(seconds):
"""Decorator that makes a function time out.
Because test suites that hang are no fun. Especially on buildbots.
Currently only implemented for Unix.
"""
def decorator(fn):
if hasattr(signal, 'alarm'):
# yay, unix!
@wraps(fn)
def wrapper(*args, **kw):
this_frame = sys._getframe()
def raiseTimeOut(signal, frame):
# the if statement here is meant to prevent an exception in the
# finally: clause before clean up can take place
if frame is not this_frame:
raise TimedOut('timed out after %s seconds' % seconds)
prev_handler = signal.signal(signal.SIGALRM, raiseTimeOut)
try:
signal.alarm(seconds)
return fn(*args, **kw)
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, prev_handler)
return wrapper
else:
# XXX um, could someone please implement this for Windows and other
# strange platforms?
return fn
return decorator
import sys
import signal
from functools import
@contextmanager
def timeout(seconds):
"""Kill subprocess ``p`` if the with block times out.
Usage example ::
p = subprocess.Popen([...], stdout=subprocess.PIPE)
with timeout(p, 30):
data = p.communicate()[0]
"""
this_frame = sys._getframe()
def raiseTimeout(signal, frame):
# the if statement here is meant to prevent an exception in the
# finally: clause before clean up can take place
if frame is not this_frame:
raise Timeout('timed out after %s seconds' % seconds)
prev_handler = signal.signal(signal.SIGALRM, raiseTimeout)
try:
signal.alarm(seconds)
yield
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, prev_handler)
@mgedmin
Copy link
Author

mgedmin commented May 30, 2012

The problem with this: it's unable to interrupt a Queue.Queue().get(). To be fair, Ctrl+C is also unable to interrupt that nasty function, I end up having to hit Ctrl-\ to dump core.

@mgedmin
Copy link
Author

mgedmin commented Feb 14, 2013

Added a context manager version as well.

I can't use this one too -- turns out the function I'm trying to time out is being called in a thread. You can't use signals from a thread. :(

@mgedmin
Copy link
Author

mgedmin commented Feb 14, 2013

Added a context manager that kills a given subprocess, to avoid the thread issue.

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