Time out a process with context manager in python

Question

I am trying to write a context manager to timeout a process, and I have following:

import threading
import _thread
import time
from contextlib import contextmanager

@contextmanager
def raise_timeout(timeout):
    timer = threading.Timer(timeout, _thread.interrupt_main)
    timer.start()

    try:
        yield
    except:
        print(f"timeout after {timeout} seconds")
        raise
    finally:
        timer.cancel()

with raise_timeout(1):
    time.sleep(5)

the problem is when the time out is reached, it still waits for the time.sleep(5) to finish before the exception is raised. when I run the script I waited for 5 seconds before getting the following output in the terminal:

> timeout after 1 seconds Traceback (most recent call last):   File
> "scratch1.py",
> line xx, in <module>
>     time.sleep(5) KeyboardInterrupt

but if I run the timer directly as below I got the exception right after the 1 second timeout.

timer = threading.Timer(1, _thread.interrupt_main)
timer.start()

I just wonder how the process in the with statement can block the exception till it finishes?

Answer 1

I think that your Timer solution is not functioning as time.sleep doesn't get killed when calling _thread.interrupt_main().

The _thread docs make a suggestion about how it functions:

Threads interact strangely with interrupts: the KeyboardInterrupt exception will be received by an arbitrary thread. (When the signal module is available, interrupts always go to the main thread.)

Though an issue on Python's bug tracker suggests the documentation for _thread.interrupt_main() is imprecise.

It should be noted, that substituting in a long running process instead of time.sleep your context manager works as expected.

Try:

with raise_timeout(1):
    for _ in range(1_000_000_000):
        pass

and it will raise a KeyboardInterrupt

---

If portability is not an issue, here's a solution using signal that is only available on Unix as it uses SIGALARM.

import signal
import time
from contextlib import contextmanager

class TimeOutException(Exception):
    pass

@contextmanager
def raise_timeout(timeout):
    def _handler(signum, frame):
        raise TimeOutException()
    signal.signal(signal.SIGALRM, _handler)
    signal.alarm(timeout)

    try:
        yield
    except TimeOutException:
        print(f"Timeout after {timeout} seconds")
        raise
    finally:
        signal.alarm(0)

with raise_timeout(1):
    time.sleep(2)

Which will raise a TimeOutException