Source code for pvp.common.utils

import signal
import time
from contextlib import contextmanager
from pvp.common.loggers import init_logger
from pvp import prefs

[docs]class TimeoutException(Exception): pass
_TIMEOUT = prefs.get_pref('TIMEOUT')
[docs]@contextmanager def time_limit(seconds): def signal_handler(signum, frame): raise TimeoutException("Timed out!") signal.signal(signal.SIGALRM, signal_handler) signal.setitimer(signal.ITIMER_REAL, seconds) try: yield finally: signal.alarm(0)
[docs]def timeout(func): """ Defines a decorator for a 50ms timeout. Usage/Test: @timeout def foo(sleeptime): time.sleep(sleeptime) print("hello") """ def wrapper(*args, **kwargs): try: with time_limit(globals()['_TIMEOUT']): ret = func(*args, **kwargs) return ret except TimeoutException as e: log_str = f'Method call timed out - Method: {func}' logger = init_logger('timeouts') logger.exception(log_str) raise e return wrapper