4

I have a function that should not take too long to run. I would like to set a timeout limit on it. I can find one proposed solutions on the internet. See the following SO post. Timeout on a function call

The solution uses signals, which is not available on Windows. There is a similar use of signals for making a ticker, which has a windows port as explained on this SO post: python: windows equivalent of SIGALRM this is not an answer to the timeout directly, but could be adapted to work for timeouts. It is written for python 2.7 though.

Since the answers are roughly 10 years old, my question is: Is there any more modern python (e.g. python 3.7) way to create a context manager/decorator/similar wrapper do make a "normal function" into a timeout-limited function on a windows system?

martineau
  • 119,623
  • 25
  • 170
  • 301
LudvigH
  • 3,662
  • 5
  • 31
  • 49
  • 2
    You can try this https://stackoverflow.com/questions/56305195/is-it-possible-to-specify-the-max-amount-of-time-to-wait-for-code-to-run-with-py/56305465?noredirect=1#comment99226611_56305465. – Praveenkumar May 29 '19 at 08:31
  • Works all okay. The only thing on my wishlist was a contextwrapper or a decorator for easy use. I have no experience of coding that whatsoever though... But since I only have one point of use right now, I guess it solves my problem at hand... I'll rewrite my question slightly for clarity. – LudvigH May 29 '19 at 08:50

1 Answers1

4

Here's a way to convert @Praveenkumar's answer into an easy-to-use decorator, which it seems you now would like to know:

import time
import concurrent.futures as futures


def timeout(timelimit):
    def decorator(func):
        def decorated(*args, **kwargs):
            with futures.ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(func, *args, **kwargs)
                try:
                    result = future.result(timelimit)
                except futures.TimeoutError:
                    print('Timedout!')
                    raise TimeoutError from None
                else:
                    print(result)
                executor._threads.clear()
                futures.thread._threads_queues.clear()
                return result
        return decorated
    return decorator


@timeout(3)
def test(n):
    print(f'Sleeping for {n} seconds')
    time.sleep(n)
    # Real code here.
    return 'Done'

test(2)  # OK
test(5)  # -> Causes timeout.
martineau
  • 119,623
  • 25
  • 170
  • 301
  • This still tries to execute the function after time out. For eg: Try this function. @timeout(1) def clean(txto): time.sleep(20) print('HI') Ideally, the function should end in 1 sec. But the function timeout after 20 secs. – Raj Mehta Aug 12 '21 at 20:31
  • 1
    @RajMehta: It's not trying to execute the function in another thread, it's doing so, which means the function has already been started and is running, so it would need to terminate it in the middle of whatever it's doing should a timeout occur. However there's no official way to kill a thread. and furthermore it would generally not be a good idea to do something like that abruptly. However there may be ways to do it in some cases if you're managing your own threads. See [Is there any way to kill a Thread?](https://stackoverflow.com/questions/323972/is-there-any-way-to-kill-a-thread) – martineau Aug 12 '21 at 21:22
  • 1
    @RajMehta: I recently heard about a third-party module named [`timeout-decorator`](https://pypi.org/project/timeout-decorator/) that might do what you want because it uses signals — mind the caveats though. – martineau Aug 13 '21 at 08:36