First of all, trying to use eventlet
for this is the wrong tool. eventlet
is primarily for non-blocking I/O using coroutines in the form of greenlets. It's got little to do with CPU-bound problems (in fact the docs for eventlet.Timeout
read):
If the code block in the try/finally or with-block never cooperatively yields, the timeout cannot be raised. In Eventlet, this should rarely be a problem, but be aware that you cannot time out CPU-only operations with this class.
What this means is that if you have some block of code that's performing a CPU-bound operation, you still won't break out of it because the code never yields to another thread. This is more for cases where you might have some code that, say, checks if a socket has data to read on it, and if not yields.
If you want to set an interrupt after some time period to interrupt a long-running calculation, you can use signal.alarm
and set a SIGALRM
handler. You can wrap this all up in a context manager like:
>>> import signal
>>> from contextlib import contextmanager
>>> class TimeoutError(RuntimeError): pass
...
>>> @contextmanager
... def timeout(seconds):
... def handler(*args):
... raise TimeoutError("timed out after {} seconds".format(seconds))
... orig_handler = signal.signal(signal.SIGALRM, handler)
... signal.alarm(seconds)
... try:
... yield
... finally:
... signal.alarm(0)
... signal.signal(signal.SIGALRM, orig_handler)
...
>>> with timeout(10):
... while True: pass
...
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
File "<stdin>", line 4, in handler
__main__.TimeoutError: timed out after 10 seconds
>>> with timeout(10):
... print(1 + 1)
...
2
>>> with timeout(10):
... 4 ** 5 ** 20
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
File "<stdin>", line 4, in handler
__main__.TimeoutError: timed out after 10 seconds
Note: This will only work on *NIX systems, though there are ways to do this on Windows as well.