python异步线程异常处理

Question

我正在尝试在 Python 中实现超时功能。

它通过使用函数装饰器包装函数来工作，该装饰器将函数作为线程调用，但也调用“看门狗”线程，该线程将在指定时间段后在函数线程中引发异常。

它目前适用于不休眠的线程。在do_rand 调用期间，我怀疑“异步”异常实际上是在time.sleep 调用之后和执行超出try/except 块之后调用的，因为这可以解释Unhandled exception in thread started by 错误。此外，来自do_rand 调用的错误是在调用后 7 秒（time.sleep 的持续时间）生成的。

我将如何“唤醒”一个线程（使用 ctypes？）以使其响应异步异常？

或者可能完全不同的方法？

代码：

# 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

问题在于time.sleep 阻塞。而且它真的很难阻塞，所以唯一能真正中断它的是进程信号。但是带有它的代码变得非常混乱，在某些情况下甚至信号都不起作用（例如，当您正在阻止 socket.recv() 时，请参见：recv() is not interrupted by a signal in multithreaded environment）。

因此通常无法中断线程（不杀死整个进程）（更不用说有人可以简单地覆盖线程中的信号处理）。

但在这种特殊情况下，您可以使用线程模块中的Event 类，而不是使用time.sleep：

线程 1

from threading import Event

ev = Event()
ev.clear()

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

线程 2（确保它可以访问相同的 ev 实例）

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

请注意，state 将是事件的内部状态。因此state == True 将意味着它被.set() 解锁，而state == False 将意味着发生超时。

在此处阅读有关活动的更多信息：

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

Answer 2

您需要使用睡眠以外的其他东西，或者您需要向其他线程发送信号以使其唤醒。

我使用的一个选项是设置一对文件描述符并使用 select 或 poll 代替睡眠，这使您可以向文件描述符写入一些内容以唤醒另一个线程。或者，如果您需要的只是操作出错，您只需等待睡眠结束，因为它花费了太长时间并且没有其他依赖它。