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 fuzzingbook.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 SIGALRM signals (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 interrupting some code execution after a given time interval.

import bookutils.setup

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!")

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!")

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.