Timeout

The code in this notebook helps in interrupting execution after a given time.

Prerequisites

  • This notebook needs some understanding on advanced concepts in Python, notably
    • classes
    • the Python with statement
    • the Python signal functions
    • measuring time

Synopsis

To use the code provided in this chapter, write

>>> from debuggingbook.Timeout import <identifier>

and then make use of the following features.

The Timeout class throws a TimeoutError exception after a given timeout has expired. Its typical usage is in conjunction with a with clause:

>>> try:
>>>     with Timeout(0.2):
>>>         some_long_running_function()
>>>     print("complete!")
>>> except TimeoutError:
>>>     print("Timeout!")
Timeout!

Note: On Unix/Linux systems, the Timeout class uses https://docs.python.org/3.10/library/signal.html (interrupts) to implement timeouts; this has no effect on performance of the tracked code. On other systems (notably Windows), Timeout uses the sys.settrace() function to check the timer after each line of code, which affects performance of the tracked code.

Measuring Time

The class Timeout allows to interrupt some code execution after a given time interval.

import bookutils
import time
from types import FrameType, TracebackType

Variant 1: Unix (using signals, efficient)

import signal
class SignalTimeout:
    """Execute a code block raising a timeout."""

    def __init__(self, timeout: Union[int, float]) -> None:
        """
        Constructor. Interrupt execution after `timeout` seconds.
        """
        self.timeout = timeout
        self.old_handler: Any = signal.SIG_DFL
        self.old_timeout = 0.0

    def __enter__(self) -> Any:
        """Begin of `with` block"""
        # Register timeout() as handler for signal 'SIGALRM'"
        self.old_handler = signal.signal(signal.SIGALRM, self.timeout_handler)
        self.old_timeout, _ = signal.setitimer(signal.ITIMER_REAL, self.timeout)
        return self

    def __exit__(self, exc_type: Type, exc_value: BaseException,
                 tb: TracebackType) -> None:
        """End of `with` block"""
        self.cancel()
        return  # re-raise exception, if any

    def cancel(self) -> None:
        """Cancel timeout"""
        signal.signal(signal.SIGALRM, self.old_handler)
        signal.setitimer(signal.ITIMER_REAL, self.old_timeout)

    def timeout_handler(self, signum: int, frame: Optional[FrameType]) -> None:
        """Handle timeout (SIGALRM) signal"""
        raise TimeoutError()

Here's an example:

def some_long_running_function() -> None:
    i = 10000000
    while i > 0:
        i -= 1
try:
    with SignalTimeout(0.2):
        some_long_running_function()
        print("Complete!")
except TimeoutError:
    print("Timeout!")
Timeout!

Variant 2: Generic / Windows (using trace, not very efficient)

import sys
class GenericTimeout:
    """Execute a code block raising a timeout."""

    def __init__(self, timeout: Union[int, float]) -> None:
        """
        Constructor. Interrupt execution after `timeout` seconds.
        """

        self.seconds_before_timeout = timeout
        self.original_trace_function: Optional[Callable] = None
        self.end_time: Optional[float] = None

    def check_time(self, frame: FrameType, event: str, arg: Any) -> Callable:
        """Tracing function"""
        if self.original_trace_function is not None:
            self.original_trace_function(frame, event, arg)

        current_time = time.time()
        if self.end_time and current_time >= self.end_time:
            raise TimeoutError

        return self.check_time

    def __enter__(self) -> Any:
        """Begin of `with` block"""
        start_time = time.time()
        self.end_time = start_time + self.seconds_before_timeout

        self.original_trace_function = sys.gettrace()
        sys.settrace(self.check_time)
        return self

    def __exit__(self, exc_type: type, 
                 exc_value: BaseException, tb: TracebackType) -> Optional[bool]:
        """End of `with` block"""
        self.cancel()
        return None  # re-raise exception, if any

    def cancel(self) -> None:
        """Cancel timeout"""
        sys.settrace(self.original_trace_function)

Again, our example:

try:
    with GenericTimeout(0.2):
        some_long_running_function()
        print("Complete!")
except TimeoutError:
    print("Timeout!")
Timeout!

Choosing the right variant

Timeout: Type[SignalTimeout] = SignalTimeout if hasattr(signal, 'SIGALRM') else GenericTimeout

Exercises

Create a Timeout variant that works efficiently on Windows. Note that how to do this a long debated issue in programming forums.

Creative Commons License The content of this project is licensed under the Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License. The source code that is part of the content, as well as the source code used to format and display that content is licensed under the MIT License. Last change: 2022-01-03 13:02:36+01:00CiteImprint