Timeout a function (windows)?

Question

I am trying to implement timeout for a particular function. I have checked many of the questions in SE and couldn't find any solution which fits my problem, because:

1. I am running python in Windows
2. Timeout is applied on a python function which I don't have control on, i.e. it is defined in an already designed module.
3. The python function is not a subprocess

I am having a already designed custom module (say MyModule) developed for particular tasks, and there are functions defined in it. One of the function (say MyFunc) has a tendency to run forever because of external factors, and I just don't want the python script to hang.

I am planning to add a timeout feature, as said below pseudocode:

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 am not sure which module can be used to achieve this on python. Note that I am a newbie, and all the scripts I have written are directly based on SE Answers or Python Documentation.

Solution 1

I think a good way to approach this is to create a decorator and use the Thread.join(timeout=seconds) method. Bear in mind that there's no good way to kill the thread, so it will continue to run in the background, more or less, as long as your program is running.

First, create a decorator like this:

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 as e:
                    res[0] = e
            t = Thread(target=newFunc)
            t.daemon = True
            try:
                t.start()
                t.join(timeout)
            except Exception as 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 with something like:

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

Solution 2

@acushner's 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

Comments

[punarvak]

Thanks for the answer. I tried the above decorator function, and it works as expected. The only change I had to do is "MyResult = timeout(timeout=16)(MyModule.MyFunc)(MyArgs)", instead of putting into 'func' and call it.

[acushner]

no prob. either way should work, though, as timeout(timeout=16)(MyModule.MyFunc) returns a function that you can then call with your args (i just didn't include them in the answer).

[propjk007]

Yay! Finally found a good answer to this that didn't include signal (the alarm feature doesn't work on Windows). Thank you @acushner.

[Ami Hollander]

Thank you, best one

[LeanMan]

when I use this code, pytest produces double output in its error report. Why does it do this?

[ailoher]

Thanks a lot for your answer @acushner -works as charm. I was wondering how to deal if you have some specific values (say lists) under constant update in your function and you want to return them even if timeout expires. Do you -or anyone- have a clue about how to proceed in such case? I am guessing that you should make some changes in your original function's code, but which? Thanks in advance!

[Leonardo Rick]

"Bear in mind that there's no good way to kill the thread, so it will continue to run in the background". I don't think I fully understood this frase. Is it exiting or not?