Timout on a function

Question

Let us say we have a python function magical_attack(energy) which may or may not last more than a second. It could even be an infinite loop? How would I run, but if it goes over a second, terminate it, and tell the rest of the program. I am looking for a sleek module to do this. Example:

import timeout
try: timout.run(magical_attack(5), 1)
except timeout.timeouterror:
    blow_up_in_face(wizard)

Note: It is impossible to modify the function. It comes from the outside during runtime.

Which version of python are you using for this? 3.2+ you can use the futures module.... otherwise more work will be involved — Chris Pak, Aug 29 '13 at 20:56
@ChrisPak: Or, for 2.5-3.1, you can just install the [`futures` backport](https://pypi.python.org/pypi/futures), and then there's no more work than in 3.2+… — abarnert, Aug 29 '13 at 20:57
Also note that in either case, you are probably going to get a performance hit, because python is ultimately single threaded. — Chris Pak, Aug 29 '13 at 20:58
@ChrisPak: That's incorrect—one thread that's just waiting on another thread with a timeout doesn't have to do any work, so it won't cause a performance hit. And it's also misleading to say "python is ultimately single threaded". Python threads really are separate native-OS threads; it's just that only one of them can be running the interpreter at any given time. — abarnert, Aug 29 '13 at 20:59
Isn't there a way for a python script to interpret another, line by line, and stop when it goes over time? — PyRulez, Aug 29 '13 at 21:20

score 1 · Answer 1 · answered Aug 29 '13 at 21:04

The simplest way to do this is to run the background code in a thread:

t = threading.Thread(target=magical_attack, args=(5,))
t.start()
t.join(1)
if not t.isAlive():
    blow_up_in_face(wizard)

However, note that this will not cancel the magical_attack function; it could still keep spinning along in the background for as long as it wants even though you no longer care about the results.

Canceling threads safely is inherently hard to do, and different on each platform, so Python doesn't attempt to provide a way to do it. If you need that, there are three alternatives:

If you can edit the code of magical_attack to check a flag every so often, you can cancel it cooperatively by just setting that flag.
You can use a child process instead of a thread, which you can then kill safely.
You can use ctypes, pywin32, PyObjC, etc. to access platform-specific routines to kill the thread. But you have to really know what you're doing to make sure you do it safely, and don't confuse Python in doing it.

As Chris Pak pointed out, the futures module in Python 3.2+ makes this even easier. For example, you can throw off thousands of jobs without having thousands of threads; you can apply timeouts to a whole group of jobs as if they were a single job; etc. Plus, you can switch from threads to processes with a trivial one-liner change. Unfortunately, Python 2.7 does not have this module—but there is a quasi-official backport that you can install and use just as easily.

score 0 · Answer 2 · edited May 23 '17 at 10:32

Abamert beat me there on the answer I was preparing, except for this detail:

If, and only if, the outside function is executed through the Python interpreter, even though you can't change it (for example, from a compiled module), you might be able to use the technique described in this other question to kill the thread that calls that function using an exception.

Is there any way to kill a Thread in Python?

Of course, if you did have control over the function you were calling, the StoppableThread class from that answer works well for this:

import threading

class StoppableThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""

    def __init__(self):
        super(StoppableThread, self).__init__()
        self._stop = threading.Event()

    def stop(self):
        self._stop.set()

    def stopped(self):
        return self._stop.isSet()

class Magical_Attack(StoppableThread):
    def __init__(self, enval):
        self._energy = enval
        super(Magical_Attack, self).__init__()

    def run(self):
        while True and not self.stopped():
            print self._energy

if __name__ == "__main__":
    a = Magical_Attack(5)
    a.start()
    a.join(5.0)
    a.stop()

Timout on a function

2 Answers2