Welcome, guest | Sign In | My Account | Store | Cart

This recipe provides 2 ways to invoke any callable with a time limit. Usable for Python 2.2 thru 3.0, see the Note in the module __doc__.

Python, 245 lines
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
#!/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
   <http://docs.python.org/dev/3.1/library/2to3.html>

   The core of the function  timelimited is copied from
   <http://code.activestate.com/recipes/473878/>.
'''
__all__ = ('timelimited', 'TimeLimited', 'TimeLimitExpired')
__version__ = '4  2009-06-08'

from threading import Thread

 # The #PYCHOK marks are intended for postprocessing
 # by <http://code.activestate.com/recipes/546532/>

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
             # <http://code.activestate.com/recipes/496960/>
             # <http://sebulba.wikispaces.com/recipe+thread2>
            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')

Added reference to the Python 2to3 translator in version 4.

9 comments

Michael Shepanski 12 years, 6 months ago  # | flag

Turn this into a decorator, and then we're in business! :D

Michael Shepanski 12 years, 6 months ago  # | flag

Actually, I take that back.. I can see how this is useful in a non-decorator form It allows you to limit the time for built-ins as well.. or change the allowed time for various calls to the same function at runtime.

ionel maries cristian 12 years, 6 months ago  # | flag

You should mention that this doesn't try to stop the thread so the callable still runs after the timeout is raised. There are some thread killing techniques in python but they don't work for all the cases.

Jean Brouwers (author) 12 years, 6 months ago  # | flag

It does not look like the threads are still running. Add the following lines at the end of the recipe.

from threading import activeCount
print('thread count: %s' % activeCount())

Then run again and only 2 threads remain (one is probably the main thread). But 5 of the 7 tests created a thread in the timelimited function.

Jean Brouwers (author) 12 years, 6 months ago  # | flag

Instead of the 2 lines mentioned before, add the following 4 and run again.

import threading
time.sleep(2)
for t in threading.enumerate():
    print('%s thread %r, alive=%r' % (sys.argv[0], t.name, t.is_alive()))

Only the main thread remains, as expected. All other threads terminated and have been gc'd, eventually.

Jean Brouwers (author) 12 years, 6 months ago  # | flag

It turns out that expired threads continue to run, as pointed out by the OP. The recipe (version 1.1) has been updated with an attempt to handle that.

Also, a test has been added to check that all _Timelimited threads are no longer running. The sleep time of some tests has been increased to 20 second.

Lastly, the recipe can be used with Python 2.6 and below, but needs to be edited for use with Python 3.0 as indicated in the module __doc__.

Jean Brouwers (author) 12 years, 6 months ago  # | flag

Version 3 fixes several typos and includes an important warning in the module __doc__. Other changes are mostly cosmetic.

eliben 10 years, 4 months ago  # | flag

This doesn't really stop the thread, of course. Try this code:

import time
def func():
    while True:
        print 'in func'
        time.sleep(0.3)

try:
    timelimited(2, func)
    print 'done normally'
except TimeLimitExpired:
    print 'time limit expired'
    time.sleep(10)

You'll see "in func" will keep being printed even after the time limit has expired.

The __stop() method of Thread doesn't kill it, it just marks that the thread has stopped so Thread.join will return immediately. There's no real way to kill/stop a thread in Python without resorting to C-level trickery.

Stephen Chappell 9 years, 6 months ago  # | flag

If you do not mind accomplishing the same things with processes, you might try recipe 577028. It has the advantage of being able to kill the code being executed. The module was an important advance for being able to run and kill a differentiation algorithm when necessary.