Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import signal
- def signal_handler(signum, frame):
- raise Exception("Timed out!")
- signal.signal(signal.SIGALRM, signal_handler)
- signal.alarm(10) # Ten seconds
- try:
- long_function_call()
- catch Exception, msg:
- print "Timed out!"
- from __future__ import with_statement # Required in 2.5
- import signal
- from contextlib import contextmanager
- class TimeoutException(Exception): pass
- @contextmanager
- def time_limit(seconds):
- def signal_handler(signum, frame):
- raise TimeoutException, "Timed out!"
- signal.signal(signal.SIGALRM, signal_handler)
- signal.alarm(seconds)
- try:
- yield
- finally:
- signal.alarm(0)
- try:
- with time_limit(10):
- long_function_call()
- except TimeoutException, msg:
- print "Timed out!"
- def timeout(func, args=(), kwargs={}, timeout_duration=1, default=None):
- '''This function will spwan a thread and run the given function using the args, kwargs and
- return the given default value if the timeout_duration is exceeded
- '''
- import threading
- class InterruptableThread(threading.Thread):
- def __init__(self):
- threading.Thread.__init__(self)
- self.result = default
- def run(self):
- try:
- self.result = func(*args, **kwargs)
- except:
- self.result = default
- it = InterruptableThread()
- it.start()
- it.join(timeout_duration)
- if it.isAlive():
- return it.result
- else:
- return it.result
- def function_with_enforced_timeout():
- f = open_temporary_file()
- try:
- ...
- finally:
- here()
- unlink(f.filename)
Add Comment
Please, Sign In to add comment