Function with_timeout

benchmarks/src/utils/timing.py:30–42

(fn: Callable[..., T], timeout_s: float)

def with_timeout(fn: Callable[..., T], timeout_s: float) -> T:
    global TIMEOUT_POOL

    if TIMEOUT_POOL is None:
        # it needs to be more than 1 to avoid deadlocks when with_timeout is nested
        TIMEOUT_POOL = ThreadPool(8)
        atexit.register(TIMEOUT_POOL.close)

    future = TIMEOUT_POOL.apply_async(fn)
    try:
        return future.get(timeout=timeout_s)
    except multiprocessing.context.TimeoutError:
        raise TimeoutException()


class Timings: