具有線程的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裝飾實現，但我不知道怎麼打發yield內從threading.Thread派生類。我的變體不起作用。

@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

用信號完成系統調用的超時。發生信號時，大多數阻塞系統調用都會返回EINTR，因此您可以使用警報來實現超時。

下面是一個上下文管理器，它與大多數系統調用一起工作，如果時間過長，會導致阻塞系統調用引發IOError。

import signal, errno 
from contextlib import contextmanager 
import fcntl 

@contextmanager 
def timeout(seconds): 
    def timeout_handler(signum, frame): 
     pass 

    original_handler = signal.signal(signal.SIGALRM, timeout_handler) 

    try: 
     signal.alarm(seconds) 
     yield 
    finally: 
     signal.alarm(0) 
     signal.signal(signal.SIGALRM, original_handler) 

with timeout(1): 
    f = open("test.lck", "w") 
    try: 
     fcntl.flock(f.fileno(), fcntl.LOCK_EX) 
    except IOError, e: 
     if e.errno != errno.EINTR: 
      raise e 
     print "Lock timed out" 

Answer 2

我看不到這樣做的一個方式，你與上下文管理提出了什麼，你不能yield從一個線程到另一個流。 我會做的是用超時的可變內存線程來包裝你的函數。這是一個recipe。

你將有一個額外的線程和語法不會很好，但它會工作。

Answer 3

如果上下文管理器保護的代碼是基於循環的，請考慮處理這種人爲處理線程查殺的方式。殺死另一個線程通常是不安全的，因此標準方法是讓控制線程設置一個工作線程可見的標誌。工作線程會定期檢查該標誌並自動關閉。這裏是你如何做一些類似與超時：

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) 

這種方法比更具侵入性使用信號，但它適用於任意線程。

Answer 4

我知道它遲了，但我只是讀這個，但創建自己的signaller/context manager呢？我剛接觸python會喜歡這個實現的經驗豐富的開發人員的反饋。

這是基於開來「福茲先生」

class TimeoutSignaller(Thread): 
    def __init__(self, limit, handler): 
     Thread.__init__(self) 
     self.limit = limit 
     self.running = True 
     self.handler = handler 
     assert callable(handler), "Timeout Handler needs to be a method" 

    def run(self): 
     timeout_limit = datetime.datetime.now() + datetime.timedelta(seconds=self.limit) 
     while self.running: 
      if datetime.datetime.now() >= timeout_limit: 
       self.handler() 
       self.stop_run() 
       break 

    def stop_run(self): 
     self.running = False 

class ProcessContextManager: 
    def __init__(self, process, seconds=0, minutes=0, hours=0): 
     self.seconds = (hours * 3600) + (minutes * 60) + seconds 
     self.process = process 
     self.signal = TimeoutSignaller(self.seconds, self.signal_handler) 

    def __enter__(self): 
     self.signal.start() 
     return self.process 

    def __exit__(self, exc_type, exc_val, exc_tb): 
     self.signal.stop_run() 

    def signal_handler(self): 
     # Make process terminate however you like 
     # using self.process reference 
     raise TimeoutError("Process took too long to execute") 

使用情況下的答案：

with ProcessContextManager(my_proc) as p: 
    # do stuff e.g. 
    p.execute()