Catching exception thrown by threading.Timer in a context manager

Question

I have written a simple TimeManager: a context manager that fires of a threading.Timer when the context is entered and cancels it when it is exited. If the timer goes off before exiting the context, it raises an exception:

import threading

class TimeManager(object):
    def __init__(self):
        self._timeout = 1

    def _timeoutHandler(self):
        raise Exception("Timeout!")

    def __enter__(self):
        self.timer = threading.Timer(self._timeout, self._timeoutHandler)
        self.timer.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.timer.cancel()
        return False

Obviously I can't catch the exception in the main thread, as it belongs to a separate thread:

>>> with TimeManager() as t:
...   try:
...     time.sleep(5)
...   except Exception:
...     print "caught"
... 
Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib64/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/usr/lib64/python2.6/threading.py", line 736, in run
    self.function(*self.args, **self.kwargs)
  File "<stdin>", line 5, in _timeoutHandler
Exception: Timeout!

So, how can I catch the exception in the main thread? Should I abandon the idea of the context manager?

Notice the problem is different from the one described here, there are no multiple threads involved there. I think it is also different from this, where message passing would negate the purpose of the timeout.

There's a module called [`stopit`](https://github.com/glenfant/stopit) that provides interruptible context managers, using threads or signals. Each approach has its own limitations, though. For example using threads, you can't actually interrupt a blocking call (like `time.sleep`). Signals can, but are only available on Unix, and aren't safe for us in multi-threaded applications. — dano, Mar 25 '15 at 14:36
Thanks for pointing the `stopit` module out: it is indeed what I was looking for. If you turn your comment into an answer I'll be more than happy to mark it as the final one :) — Jir, Mar 25 '15 at 16:16

score 4 · Accepted Answer · answered Mar 25 '15 at 16:22

There's a module called stopit that provides interruptible context managers, using threads or signals. Each approach has its own limitations, though. For example using threads, you can't actually interrupt a blocking call (like time.sleep). Signals can, but are only available on Unix, and aren't safe for use in multi-threaded applications.

It looks like it leverages The C-API function PyThreadState_SetAsyncExc to asynchronously raise an exception in the desired thread.

Here's an example usage (taken from their docs):

>>> import time
>>> def variable_duration_func(duration):
...     t0 = time.time()
...     while True:
...         dummy = 0
...         if time.time() - t0 > duration:
...             break
>>>
>>> start_time = time.time()
>>> with Timeout(2.0) as timeout_ctx:
...     variable_duration_func(5.0)
>>> time.time() - start_time < 2.2
True
>>> timeout_ctx.state == timeout_ctx.TIMED_OUT
True

Catching exception thrown by threading.Timer in a context manager

1 Answers1