3 ways to set timeout on function call in Python

Introduction

Have you ever encountered a situation where you needed to set a timeout on a function call? Whether you’re working with JavaScript, Python, or any other programming language, there are several ways to achieve this. In this blog post, we will explore three different methods to set a timeout on a function call in Python.

Using Signal

import signal
import functools
import os
import errno
import time

class TimeoutError(Exception):
    pass

def timeout(seconds=10, error_message=os.strerror(errno.ETIME)):
    def decorator(func):
        def _handle_timeout(signum, frame):
            raise TimeoutError(error_message)

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, _handle_timeout)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result

        return wrapper

    return decorator

@timeout
def long_running_function1():
    time.sleep(6)

@timeout(5)
def long_running_function2():
    time.sleep(6)

@timeout(3, os.strerror(errno.ETIMEDOUT))
def long_running_function3():
    time.sleep(6)

if __name__ == '__main__':
    try:
        long_running_function3()
    except Exception as e:
        print(e)

or wrap in one class

import time
import signal

class Timeout:
    def __init__(self, seconds=1, error_message='Timeout'):
        self.seconds = seconds
        self.error_message = error_message

    def handle_timeout(self, signum, frame):
        raise TimeoutError(self.error_message)

    def __enter__(self):
        signal.signal(signal.SIGALRM, self.handle_timeout)
        signal.alarm(self.seconds)

    def __exit__(self, type, value, traceback):
        signal.alarm(0)

if __name__ == '__main__':
    try:
        with Timeout(seconds=3):
            time.sleep(5)
    except TimeoutError as e:
        print(e)

Using ProcessPoolExecutor

from concurrent.futures import TimeoutError
from time import sleep
import concurrent

max_numbers = [1000]

def run_loop(max_number):
    sleep(max_number)

def main():
    with concurrent.futures.ProcessPoolExecutor(max_workers=len(max_numbers)) as executor:
        try:
            for future in concurrent.futures.as_completed(executor.map(run_loop, max_numbers, timeout=1), timeout=1):
                print(future.result(timeout=1))
        except concurrent.futures._base.TimeoutError:
            print("This took to long...")
            stop_process_pool(executor)

def stop_process_pool(executor):
    for pid, process in executor._processes.items():
        process.terminate()
    executor.shutdown()

if __name__ == '__main__':
    main()

Using multiprocessing

import multiprocessing
import time

def long_running_function():
    time.sleep(10)

if __name__ == '__main__':
    p = multiprocessing.Process(target=long_running_function)
    p.start()
    p.join(3)
    if p.is_alive():
        print("timeout, terminated!")
        p.terminate()
        p.join()

Reference

• https://docs.python.org/3/library/concurrent.futures.html
• https://medium.com/@chaoren/how-to-timeout-in-python-726002bf2291
• https://stackoverflow.com/questions/492519/timeout-on-a-function-call
• https://alexandra-zaharia.github.io/posts/function-timeout-in-python-signal/

Leave a message

Name

Email



Send