Demystifying Python func_timeout Module

In this article, we will be discussing the func_timeout module in Python. Let’s say a program needs to make a large function call, but it has a limited time period for the function to terminate. A practical way to limit the execution of a function call is to use func_timeout.func_timeout() from the Python func_timeout module.

About the Module

Python func_timeout module allows us to specify timeouts during existing function calls and stoppable threads. As of Python 3.10, it is NOT part of the Python standard library.

You can install it manually using PIP.

pip install func_timeout 

Use Case Demonstration of Python func_timeout

Let’s refer to the following sample program that implements the module.

import func_timeout


def largeFunction():
    while True: pass
    return 'PythonPool.com'


def runFunction(f, max_wait, default_value):
    try:
        return func_timeout.func_timeout(max_wait, largeFunction)
    except func_timeout.FunctionTimedOut:
        pass
    return default_value

x = runFunction(largeFunction, 5, 'Python')

print(x)

Output

Python

Explanation

The function runFunction makes the call func_timeout.func_timeout(max_wait,largeFunction) in order to all largeFunction without arguments nor wait time (max_wait).

If largeFunction() fails from stopping within the time interval, an error is thrown and caught by the except block.

As largeFunction() failed to return a value within the time interval, the default value “Python” is returned.

Python timeout Function After “n” Seconds Using func_timeout

func_timeout allows us to run the given function for up to “timeout” seconds.

func_timeout(timeout, func, args=(), kwargs=None)

Any exception raised during the call will return func returns.

timeout parameter indicates the maximum number of seconds to run func before exiting.

args receives a tuple of any ordered arguments to pass to the function.

FunctionTimedOut exception is thrown upon failure to run within the given time frame.

Finally, the function will return the return value of func.

Python Function Timeout Multiprocessing

Using the multiprocessing module in Python, we can terminate a function if it breaches a time hold threshold. Refer to the following program.

from multiprocessing import Pool, TimeoutError
import time

def f(x):
    return x*x

if __name__ == '__main__':

    with Pool(processes=4) as pool:
        res = pool.apply_async(f, (20,))      
        print(res.get(timeout=1))  
        ...
        ...
        ...
        res = pool.apply_async(time.sleep, (10,))

Creating a Python Loop Delay without time.sleep()

timelimit = 21
while timelimit > 0 :
  print "%s seconds remaining" % (timelimit)
  timelimit = timelimit -1

We need to iterate the above loop within 20 seconds and once per each second. In this situation, we cannot use time.sleep() as it will cause the entire application to be unresponsive for a particular amount of time.

import time
max = 31
start = time.time()

while True:
    time.sleep(0.1) 
    print( "looping...")

    #Remaining time after each iteration
    remaining = max + start - time.time()
    print( "%s seconds remaining" % int(remaining))

    #After final interation, break the loop
    if remaining <= 0:
        break

The above implementation uses time.sleep() in order to prevent numerous print statements in the output console. It has no direct effect on the iterations.

Stopping a Python Function after a Timeout

With the help of the multiprocessing module, we can do exactly that.

import multiprocessing
import time

# Sample Function
def myFunction():
    for i in range(100):
        print("iteration")
        time.sleep(1)

if __name__ == '__main__':
    # "Starting" myFunction as a process
    p = multiprocessing.Process(target=myFunction)
    p.start()

    # Waits for 10 seconds or until the function execution is over
    p.join(10)

    # If the process "p" is still alive
    if p.is_alive():
        print("end thread")

        # The process is terminated
        p.terminate()
        p.join()

The above program allows the function to run within 10 seconds. If 10 seconds have passed, the process is automatically terminated.

How to Implement a StoppableThread Class in Python

In programming, it is precarious to terminate a thread unexpectedly. A thread may contain a vital resource to the program; closing it without proper care may cause unexpected behavior. Threads may also consist of other sub-threads which need the same attention.

To ensure this, we can implement an exit request flag that checks at regular intervals for the exit. The Python threading module allows us to implement this. Let’s refer to the following program

import threading

class StoppableThread(threading.Thread):
    def __init__(self,  *args, **kwargs):
        super(StoppableThread, self).__init__(*args, **kwargs)
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def stopped(self):
        return self._stop_event.is_set()

StoppableThread has a method stop(). As it’s called, it will exit the thread and waits to exit properly using the join() method. The thread regularly checks the exit flag.

FAQs on Func timeout Python

Conclusion

In this article, we have reviewed the func_timeout module and how its classes allow us to control the runtime of a program. Other packages such as multiprocessing allow us to perform similar operations.