#!/usr/bin/env python '''This module exposes function timelimited and two classes TimeLimited and TimeLimitExpired. Function timelimited can be used to invoke any callable object with a time limit. Class TimeLimited wraps any callable object into a time limited callable with an equivalent signature. Beware, any critical resources like locks, memory or files, etc. acquired or opened by the callable may not be released respectively closed. Therefore, time limiting such callables may cause deadlock or leaks or both. No signals or timers are affected and any errors are propagated as usual. Decorators and with statements are avoided for backward compatibility. Tested with Python 2.2.3, 2.3.7, 2.4.5, 2.5.2, 2.6.2 or 3.0.1 on CentOS 4.7, MacOS X 10.4.11 Tiger (Intel) and 10.3.9 Panther (PPC), Solaris 10 and Windows XP. Note, for Python 3.0 and beyond, replace ', e:' with ' as e:' in the 3 except lines marked #XXX below or run the Python 2to3 translator on this file, see The core of the function timelimited is copied from . ''' __all__ = ('timelimited', 'TimeLimited', 'TimeLimitExpired') __version__ = '4 2009-06-08' from threading import Thread # The #PYCHOK marks are intended for postprocessing # by try: # UGLY! private method __stop _Thread_stop = Thread._Thread__stop #PYCHOK false except AttributeError: # _stop in Python 3.0 _Thread_stop = Thread._stop #PYCHOK expected class TimeLimitExpired(Exception): '''Exception raised when time limit expires. ''' pass def timelimited(timeout, function, *args, **kwds): '''Invoke the given function with the positional and keyword arguments under a time constraint. The function result is returned if the function finishes within the given time limit, otherwise a TimeLimitExpired error is raised. The timeout value is in seconds and has the same resolution as the standard time.time function. A timeout value of None invokes the given function without imposing any time limit. A TypeError is raised if function is not callable, a ValueError is raised for negative timeout values and any errors occurring inside the function are passed along as-is. ''' class _Timelimited(Thread): _error_ = TimeLimitExpired # assume timeout _result_ = None def run(self): try: self._result_ = function(*args, **kwds) self._error_ = None except Exception, e: #XXX as for Python 3.0 self._error_ = e def _stop(self): # UGLY! force the thread to stop by (ab)using # the private __stop or _stop method, but that # seems to work better than these recipes # # if self.isAlive(): _Thread_stop(self) if not hasattr(function, '__call__'): raise TypeError('function not callable: %s' % repr(function)) if timeout is None: # shortcut return function(*args, **kwds) if timeout < 0: raise ValueError('timeout invalid: %s' % repr(timeout)) t = _Timelimited() t.start() t.join(timeout) if t._error_ is None: return t._result_ if t._error_ is TimeLimitExpired: t._stop() raise TimeLimitExpired('timeout %r for %s' % (timeout, repr(function))) else: raise t._error_ class TimeLimited(object): '''Create a time limited version of any callable. For example, to limit function f to t seconds, first create a time limited version of f. from timelimited import * f_t = TimeLimited(f, t) Then, instead of invoking f(...), use f_t like try: r = f_t(...) except TimeLimitExpired: r = ... # timed out ''' def __init__(self, function, timeout=None): '''See function timelimited for a description of the arguments. ''' self._function = function self._timeout = timeout def __call__(self, *args, **kwds): '''See function timelimited for a description of the behavior. ''' return timelimited(self._timeout, self._function, *args, **kwds) def __str__(self): return '<%s of %r, timeout=%s>' % (repr(self)[1:-1], self._function, self._timeout) def _timeout_get(self): return self._timeout def _timeout_set(self, timeout): self._timeout = timeout timeout = property(_timeout_get, _timeout_set, None, 'Property to get and set the timeout value') if __name__ == '__main__': import sys, time, threading #PYCHOK expected _format = '%s test %%d/8 %%s in Python %s: %%s' % ( sys.argv[0], sys.version.split()[0]) _tests = 0 def passed(arg='OK'): global _tests _tests += 1 print(_format % (_tests, 'passed', arg)) def failed(fmt, *args): global _tests _tests += 1 if args: t = fmt % args else: t = fmt print(_format % (_tests, 'failed', t)) def check(timeout, sleep, result, arg='OK'): if timeout > sleep: x = None # time.sleep(0) result elif isinstance(result, TimeLimitExpired): x = result else: x = TimeLimitExpired if result is x: passed(arg) else: failed('expected %r, but got %r', x, result) # check timelimited function for t, s in ((2.0, 1), (1.0, 20)): # note, 20! try: r = timelimited(t, time.sleep, s) except Exception, e: #XXX as for Python 3.0 r = e check(t, s, r, timelimited) # check TimeLimited class and property f = TimeLimited(time.sleep) for t, s in ((2.0, 1), (1.0, 20)): # note, 20! f.timeout = t try: r = f(s) except Exception, e: #XXX as for Python 3.0 r = e check(t, s, r, f) # check TypeError try: t = timelimited(0, None) failed('no %r', TypeError) except TypeError: passed(TypeError) except: failed('expected %r', TypeError) # check ValueError try: t = timelimited(-10, time.time) failed('no %r', ValueError) except ValueError: passed(ValueError) except: failed('expected %r', ValueError) # check error passing from thread try: r = timelimited(1, lambda x: 1/x, 0) failed('no %r', ZeroDivisionError) except ZeroDivisionError: passed(ZeroDivisionError) except: failed('expected %r', ZeroDivisionError) # check that all created threads stopped for t in threading.enumerate(): if t.isAlive() and repr(t).startswith('<_Timelimited('): failed('thread %r still alive', t) break else: passed('all _Timelimited threads stopped')