带线程的 Python 超时上下文管理器

Question

我有 timeout 上下文管理器，它可以完美地处理信号，但它会在多线程模式下引发错误，因为信号仅在主线程中工作。

def timeout_handler(signum, frame):
    raise TimeoutException()

@contextmanager
def timeout(seconds):
    old_handler = signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

我见过 timeout 的装饰器实现，但我不知道如何在派生自 threading.Thread 的类中传递 yield。我的变体不起作用。

@contextmanager
def timelimit(seconds):
    class FuncThread(threading.Thread):
        def run(self):
            yield

    it = FuncThread()        
    it.start()
    it.join(seconds)

    if it.isAlive():
        raise TimeoutException()

Answer 1

如果上下文管理器保护的代码是基于循环的，请考虑以人们处理线程终止的方式来处理它。杀死另一个线程通常是不安全的，因此标准方法是让控制线程设置一个对工作线程可见的标志。工作线程定期检查该标志并彻底关闭自己。以下是您可以执行与超时类似的操作的方法:

class timeout(object):
    def __init__(self, seconds):
        self.seconds = seconds
    def __enter__(self):
        self.die_after = time.time() + self.seconds
        return self
    def __exit__(self, type, value, traceback):
        pass
    @property
    def timed_out(self):
        return time.time() > self.die_after

这是一个单线程使用示例:

with timeout(1) as t:
    while True: # this will take a long time without a timeout
        # periodically check for timeouts
        if t.timed_out:
            break # or raise an exception
        # do some "useful" work
        print "."
        time.sleep(0.2)

和一个多线程的:

import thread
def print_for_n_secs(string, seconds):
    with timeout(seconds) as t:
        while True:
            if t.timed_out:
                break # or raise an exception
            print string,
            time.sleep(0.5)

for i in xrange(5):
    thread.start_new_thread(print_for_n_secs,
                            ('thread%d' % (i,), 2))
    time.sleep(0.25)

这种方法比使用信号更具侵入性，但它适用于任意线程。