In your real application, can you switch to multiprocessing? It looks like what you are asking for can be done using multiprocessing + threading.Timer + try/except .
Take a look at this:
class SafeProcess(Process): def __init__(self, queue, *args, **kwargs): self.queue = queue super().__init__(*args, **kwargs) def run(self): print('Running') try: result = self._target(*self._args, **self._kwargs) self.queue.put_nowait(result) except: print('Exception') result = None while result != 'it worked!!': q = Queue() p = SafeProcess(q, target=unreliable_code) p.start() t = Timer(1, p.terminate)
This in one (happy) case gave me:
Running Empty None Running it worked!!
In your code samples, you have 4 out of 5 chances of getting an error message, so you can also create a pool or something that will improve your chances of getting the right result.
Rik poggi
source share