~/python/how-implement-timed-functions-max-timeout.md

Python - How to Implement Timed-Function which gets Timeout After Specified Max Timeout Value

2 min readMay 09, 2022by Gorav Singal#python
// TL;DR

Implement a Python decorator using the signal module's SIGALRM to enforce a maximum execution timeout on any function, raising a custom exception when the limit is exceeded.

Python - How to Implement Timed-Function which gets Timeout After Specified Max Timeout Value

Introduction

We often require to execute in timed manner, i.e. to specify a max timeout after which function should get terminated.

Code - Decorator Function - Annotation

We will write a decorator function in python, which we can use as annotation in any code where want this behavior.

import signal
import time

class TimedFunction(Exception):
    """
    Its an exception thrown when timeout happens.
    
    # NOTE: this will break in multithreading
    # it should work fine in multiprocessing
    """

    prev = list()

    def __init__(self, timeout=10):
        """
        :param int timeout: timeout in seconds, default 10s
        """
        self._timeout = timeout
        self._started = 0
        self._prev = []

        super(TimedFunction, self).__init__(repr(self))

    def __repr__(self):
        return f'Timeout: {self._timeout} seconds'

    def restore(self, ended=False):
        """ 
        this method restores the original alarm signal handler, or sets up
        the next timer on the pushdown stack.
        """
        if not ended and signal.getitimer(signal.ITIMER_REAL)[0] > 0:
            return
        while self._prev and self in self._prev:
            self._prev.pop()
        if self._prev:
            prev = self._prev[-1]
            time_diff = time.time() - prev.started
            time_remaining = prev.timeout - time_diff
            if time_remaining > 0:
                signal.signal(signal.SIGALRM, prev.fire_timer)
                signal.setitimer(signal.ITIMER_REAL, time_remaining)
            else:
                prev.restore()
        else:
            signal.setitimer(signal.ITIMER_REAL, 0)
            signal.signal(signal.SIGALRM, signal.SIG_DFL)

    def fire_timer(self, *_sig_param):
        """
        when an itimer fires, execution enters this method
        which either clears timers, sets up the next timer in a nest
        as specified by the options.

        After the timers are handled, this method raises the TimedFunction
        exception.
        """
        self.restore()
        raise self

    def __enter__(self):
        """ 
        the logic that starts the timers is normally fired by the with
        keyword though, with just calls this __enter__ function. The timers
        are started here.
        """
        self._prev.append(self)
        signal.signal(signal.SIGALRM, self.fire_timer)
        self._started = time.time()
        signal.setitimer(signal.ITIMER_REAL, self._timeout)
        return self

    def __exit__(self, e_type, e_obj, e_tb):
        """ 
        when the code leaves the a TimedFunction with block, execution enters this __exit__
        method. It attempts to clean up any remaining timers.
        """
        self.restore(ended=True)


def timedfunction_wrapper(**t_kw):
    """ 
    wrap decroated function in a with TimedFunction block and guard against exceptions
    The options are roughly the same as for TimedFunction with a minor exception.
    options:
    """
    def _decorator(actual):
        def _wrapper(*a, **kw):
            with TimedFunction(**t_kw):
                return actual(*a, **kw)

        return _wrapper

    return _decorator

Note: I placed this code in a folder: python_utils/timed_function.py

Code to Test

import time
from python_utils.timed_function import timedfunction_wrapper

@timedfunction_wrapper(timeout=2)
def test():
    print('In Test')
    for i in range(1, 100):
        print(f'Sleeping - {i}')
        time.sleep(1)

test()

The github link for this code is: https://github.com/goravsingal/python_utils

Thanks for reading.

$ subscribe --weekly

One deep dive a week, in your inbox

System design, AI engineering, trading systems and interview prep. Tested code, real numbers. Free, and you can unsubscribe any time.

or read it on gyanbyte.substack.com

Share

Related Posts

Python SMTP Email Code - How to Send HTML Email from Python Code with Authentication at SMTP Server

Python SMTP Email Code - How to Send HTML Email from Python Code with Authentication at SMTP Server

Introduction This post has the complete code to send email through smtp server…

Python Build Quality Gates with Pylint and Unittest Coverage Thresholds

Python Build Quality Gates with Pylint and Unittest Coverage Thresholds

Introduction It is very important to introduce few process so that your code and…

How to Solve Circular Import Error in Python

How to Solve Circular Import Error in Python

Introduction To give some context, I have two python files. (Both in same folder…

Python Code - How To Read CSV with Headers into an Array of Dictionary

Python Code - How To Read CSV with Headers into an Array of Dictionary

Introduction Lets assume we have a csv something similar to following: Python…

Python Code - How To Read CSV into an Array of Arrays

Python Code - How To Read CSV into an Array of Arrays

Introduction In last post, we saw How to read CSV with Headers into Dictionary…

Python - Some useful Pytest Commands

Python - Some useful Pytest Commands

Introduction In this post, we will explore some useful command line options for…

Latest Posts

How to Stay Valuable as a Software Engineer in the AI Era: A Practical Playbook

How to Stay Valuable as a Software Engineer in the AI Era: A Practical Playbook

In the previous article I argued that AI is not killing software engineering…

How to Review AI-Generated Code: A Checklist for Catching What the Model Missed

How to Review AI-Generated Code: A Checklist for Catching What the Model Missed

A few months back I approved a PR that was almost entirely AI-generated. It…

How to Debug AI-Generated Code: Understand First, Fix Last

How to Debug AI-Generated Code: Understand First, Fix Last

In the last article I wrote about reviewing AI code before it goes in. This one…

The Future of Software Engineering: What AI Changes, What Stays, and Where the Jobs Go

The Future of Software Engineering: What AI Changes, What Stays, and Where the Jobs Go

Every few weeks somebody sends me a scary headline — “AI will replace all…

The Transactional Outbox Pattern: Never Lose an Event Again

The Transactional Outbox Pattern: Never Lose an Event Again

This is one of those patterns that looks like nothing until you get burnt once…

Event Sourcing, Explained: Store What Happened, Not Just the Final State

Event Sourcing, Explained: Store What Happened, Not Just the Final State

Here is a question I like to ask in interviews: your bank account shows a…