Python function timeout in windows

Question

I am trying to implement a timeout for a specific function. I checked many questions in SE and could not find a solution that fits my problem because

• I am running python on windows
• The timeout is applied to the python function, which I do not control, i.e. it is defined in an already developed module.
• Python function is not a subprocess

I have an already developed user module (say, MyModule) developed for specific tasks, and there are functions defined in it. One of the functions (e.g. MyFunc) tends to work forever due to external factors, and I just don't want the python script to freeze.

I plan to add a timeout function as below, pseudo code

import MyModule set_timeout(T) MyResult=MyModule.MyFunc() #Come to this part of script after execution of MyFunc() or after T seconds (the latter on priority) if Timeout occurred: print 'MyFunc did not execute completely' else: print 'MyFunc completed' 

But I'm not sure which module can be used to achieve this in python. Please note that I am new and all the scripts I have written are based directly on SE Answers or Python Documentation.

Answer 1

I think a good way to get closer to this is to create a decorator and use the Thread.join (timeout) method. keep in mind that there is no good way to kill a thread, so it will continue to run in the background, more or less if your program is running.

first create a decorator as follows:

 from threading import Thread import functools def timeout(timeout): def deco(func): @functools.wraps(func) def wrapper(*args, **kwargs): res = [Exception('function [%s] timeout [%s seconds] exceeded!' % (func.__name__, timeout))] def newFunc(): try: res[0] = func(*args, **kwargs) except Exception, e: res[0] = e t = Thread(target=newFunc) t.daemon = True try: t.start() t.join(timeout) except Exception, je: print 'error starting thread' raise je ret = res[0] if isinstance(ret, BaseException): raise ret return ret return wrapper return deco 

then do something like this:

 func = timeout(timeout=16)(MyModule.MyFunc) try: func() except: pass #handle errors here 

You can use this decorator anywhere you need:

 @timeout(60) def f(): ... 

Answer 2

@acushner answer adapted for python 3.5:

 from threading import Thread import functools def timeout(seconds_before_timeout): def deco(func): @functools.wraps(func) def wrapper(*args, **kwargs): res = [Exception('function [%s] timeout [%s seconds] exceeded!' % (func.__name__, seconds_before_timeout))] def newFunc(): try: res[0] = func(*args, **kwargs) except Exception as e: res[0] = e t = Thread(target=newFunc) t.daemon = True try: t.start() t.join(seconds_before_timeout) except Exception as e: print('error starting thread') raise e ret = res[0] if isinstance(ret, BaseException): raise ret return ret return wrapper return deco