5

I am trying to create a watchdog class, that will throw an exception after specified time:

from threading import Timer
from time import sleep

class watchdog():
  def _timeout(self):
    #raise self
    raise TypeError

  def __init__(self):
    self.t = Timer(1, self._timeout)

  def start(self):
    self.t.start()

try:
  w = watchdog()
  w.start()
  sleep(2)
except TypeError, e:
  print "Exception caught"
else:
  print "Of course I didn't catch the exception"

This exception is not caught, as the exception is thrown from completely different context, hence we will see the last message.

My question is, how can I modify the code, so the exception will be caught?

Melon
  • 604
  • 1
  • 7
  • 30
  • 1
    I am pretty sure this is not the remmended way to do whatever you want to do. Unless you are dealing with existing code. I think you can use [condition objects](https://docs.python.org/2/library/threading.html#condition-objects) instead – Elazar May 22 '15 at 08:06
  • Suppose a test runs for one hour, but after a minute I can already tell that it has failed. The test checks if any exceptions were risen. If exception is risen the test is terminated. This way I don't have to wait 59 minutes for the results since the test failed at first minute. – Melon May 22 '15 at 08:40
  • As per [the documentation](https://docs.python.org/2/library/threading.html#timer-objects), Timer creates a separate thread. So what you're trying to do is probably a specific case of [catching a thread's Exception in the caller thread](http://stackoverflow.com/questions/2829329/catch-a-threads-exception-in-the-caller-thread-in-python), and the answer to that question was to use Message Passing using the `Queue` object. – zehnpaard May 22 '15 at 08:59
  • This won't solve my problem, because tests need to run and in your solution I would have to hang on the queue, while I need to feed the watchdog. – Melon May 22 '15 at 09:06
  • Why do not simply use external process as a watchdog and send `SIGTERM` to a python script after timeout? – myaut May 22 '15 at 09:12
  • Because there are other tests in the test suite that can be run. We do not need to terminate the whole suite just because one test has failed – Melon May 22 '15 at 09:14

1 Answers1

3

This is not possible, as you suggested, and there is no api for abruptly stopping a thread, either, which rules out other potential solutions.

I believe your best solution is to let the watchdog set a flag, and let the test read it at certain points. Similarly, your test can simply check the duration from time to time.

Note that if the "flag" would set in a way that will cause the main thread to raise an exception (for example, deleting attributes from objects), it'll be just as effective.

The other possibility is to use multiprocessing instead of multythreading, if it is possible for your application.

Elazar
  • 20,415
  • 4
  • 46
  • 67