When we use the Third party library for network requests, we can see that it has a parameter called timeout, which means that when the network request is sent, if no response is received after timeout, a timeout exception will be thrown. (Of course, there are special cases where a timeout will expire, see Timeouts and Cancellation for humans* as the author’s example in this article, we will not consider such special cases).

But have you ever thought about how to set a timeout for a normal function? In particular, when running some data processing, AI-related code, a function may take a long time to run, and we want to implement an automatic error when the function runs beyond a certain period of time.

For example, in one scenario, I wrote a function calc_statistic(datas) that calculates a value based on the data passed in by the user. However, if the data passed in by the user is very large, this function may run for a long time. I want to set this function to run for a maximum of 10 seconds. If the operation is not completed within 10 seconds, an error is reported. What should I do?

If you’re running Linux or macOS, you can use Signal.

A few days ago, we introduced the use of signal to take over the interrupt signal from the keyboard, using signal.sigint. Today we’re going to use signal.sigalrm.

First let’s look at how this signal is used:

import time
import signal


def handler(signum, _):
    print('Timed! ')
    raise Exception('It's time! ')

def clac_statistic(datas):
    time.sleep(100)
    

signal.signal(signal.SIGALRM, handler)
signal.alarm(5)
clac_statistic('xxx')
Copy the code

The operating effect is shown in the figure below:

The signal.sigalrm event is first bound to the handler function, and then a signal is sent with signal.alarm(10) on a delay of 10 seconds. After 10 seconds, the function handler is run. An exception was thrown in the function, causing the program to end. The clac_statistic function was supposed to run for 100 seconds, but stopped after 10 seconds, thus implementing the timeout function.

Based on the above principles, we implemented a decorator to simplify setting timeout for different functions:

import time
import signal


class FuncTimeoutException(Exception):
    pass

def handler(signum, _):
    raise FuncTimeoutException('Function timed! ')

def func_timeout(times=0):
    def decorator(func):
        if not times:
            return func
        def wraps(*args, **kwargs):
            signal.alarm(times)
            result = func(*args, **kwargs)
            signal.alarm(0)  The function completes ahead of time, cancelling the signal
            return result
        return wraps
    return decorator

signal.signal(signal.SIGALRM, handler)
Copy the code

Let’s test the function timeout decorator. First test function run time is less than the timeout time, the program runs normally no problem:

Let’s test if the function runs longer than the timeout:

The FuncTimeoutException exception is normally thrown.

In practice, try… Except FuncTimeoutException catches this exception and then implements a custom processing flow, such as:

try:
    clac_statistic(100)
except FuncTimeException:
    print('This function runs timeout, running custom processing flow')
Copy the code

Of course, you can skip this exception if you want:

import contextlib:
with contextlib.supress(FuncTimeException):
    clac_statistic(100)
Copy the code