python asyncronous thread exception handling

Question

I'm trying to implement a timeout functionality in Python.

It works by wrapping functions with a function decorator that calls the function as a thread but also calls a 'watchdog' thread that will raise an exception in the function thread after a specified period has elapsed.

It currently works for threads that don't sleep. During the do_rand call, I suspect the 'asynchronous' exception is actually being called after the time.sleep call and after the execution has moved beyond the try/except block, as this would explain the Unhandled exception in thread started by error. Additionally, the error from the do_rand call is generated 7 seconds after the call (the duration of time.sleep).

How would I go about 'waking' a thread up (using ctypes?) to get it to respond to an asynchronous exception ?

Or possibly a different approach altogether ?

Code:

# Import System libraries
import ctypes
import random
import sys
import threading
import time

class TimeoutException(Exception):
    pass

def terminate_thread(thread, exc_type = SystemExit):
    """Terminates a python thread from another thread.

    :param thread: a threading.Thread instance
    """
    if not thread.isAlive():
        return

    exc = ctypes.py_object(exc_type)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), exc)

    if res == 0:
        raise ValueError("nonexistent thread id")
    elif res > 1:
        # """if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"""
        ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None)
        raise SystemError("PyThreadState_SetAsyncExc failed")

class timeout_thread(threading.Thread):
    def __init__(self, interval, target_thread):
        super(timeout_thread, self).__init__()
        self.interval     = interval
        self.target_thread = target_thread
        self.done_event = threading.Event()
        self.done_event.clear()

    def run(self):
        timeout = not self.done_event.wait(self.interval)
        if timeout:
            terminate_thread(self.target_thread, TimeoutException)

class timeout_wrapper(object):
    def __init__(self, interval = 300):
        self.interval = interval

    def __call__(self, f):
        def wrap_func(*args, **kwargs):
            thread = threading.Thread(target = f, args = args, kwargs = kwargs)
            thread.setDaemon(True)
            timeout_ticker = timeout_thread(self.interval, thread)
            timeout_ticker.setDaemon(True)
            timeout_ticker.start()
            thread.start()
            thread.join()
            timeout_ticker.done_event.set()
        return wrap_func

@timeout_wrapper(2)
def print_guvnah():
    try:
        while True:
            print "guvnah"

    except TimeoutException:
        print "blimey"

def print_hello():
    try:
        while True:
            print "hello"

    except TimeoutException:
        print "Whoops, looks like I timed out"

def do_rand(*args):
    try:
        rand_num   = 7 #random.randint(0, 10)
        rand_pause = 7 #random.randint(0,  5)
        print "Got rand: %d" % rand_num
        print "Waiting for %d seconds" % rand_pause
        time.sleep(rand_pause)
    except TimeoutException:
        print "Waited too long"

print_guvnah()
timeout_wrapper(3)(print_hello)()
timeout_wrapper(2)(do_rand)()

Answer 1

The problem is that time.sleep blocks. And it blocks really hard, so the only thing that can actually interrupt it is process signals. But the code with it gets really messy and in some cases even signals don't work ( when for example you are doing blocking socket.recv(), see this: recv() is not interrupted by a signal in multithreaded environment ).

So generally interrupting a thread (without killing entire process) cannot be done (not to mention that someone can simply override your signal handling from a thread).

But in this particular case instead of using time.sleep you can use Event class from threading module:

Thread 1

from threading import Event

ev = Event()
ev.clear()

state = ev.wait(rand_pause) # this blocks until timeout or .set() call

Thread 2 (make sure it has access to the same ev instance)

ev.set() # this will unlock .wait above

Note that state will be the internal state of the event. Thus state == True will mean that it was unlocked with .set() while state == False will mean that timeout occured.

Read more about events here:

http://docs.python.org/2/library/threading.html#event-objects

Answer 2

You'd need to use something other than sleep, or you'd need to send a signal to the other thread in order to make it wake up.

One option I've used is to set up a pair of file descriptors and use select or poll instead of sleep, this lets you write something to the file descriptor to wake up the other thread. Alternatively you just wear waiting until the sleep finishes if all you need is for the operation to error out because it took too long and nothing else is depending on it.