Менеджер контекста времени ожидания Python с потоками
я имею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()