daniel-s
7/26/2016 - 12:18 PM

Add timeout to function. http://stackoverflow.com/a/35139284/3149679

from concurrent.futures import ThreadPoolExecutor, TimeoutError
import functools

def timeout(max_timeout):
    """Timeout decorator, parameter in seconds."""
    def timeout_decorator(item):
        """Wrap the original function."""
        @functools.wraps(item)
        def func_wrapper(*args, **kwargs):
            """Closure for function."""
            executor = ThreadPoolExecutor(max_workers=1)
            future = executor.submit(item, *args, **kwargs)
            return future.result(timeout=max_timeout)
        return func_wrapper
    return timeout_decorator
import multiprocessing.pool
import functools

def timeout(max_timeout):
    """Timeout decorator, parameter in seconds."""
    def timeout_decorator(item):
        """Wrap the original function."""
        @functools.wraps(item)
        def func_wrapper(*args, **kwargs):
            """Closure for function."""
            pool = multiprocessing.pool.ThreadPool(processes=1)
            async_result = pool.apply_async(item, args, kwargs)
            # raises a TimeoutError if execution exceeds max_timeout
            return async_result.get(max_timeout)
        return func_wrapper
    return timeout_decorator
import timeout

@timeout(5.0)  # if execution takes longer than 5 seconds, raise a TimeoutError
def test_base_regression(self):
    ...