Darius Knowledge Hub

Python Timeout (for potentially dangerous long execution code)

3 min read

Python Timeout (for potentially dangerous long execution code)

Created: 16 Jan 2023, 02:55 PM | Modified: =dateformat(this.file.mtime,"dd MMM yyyy, hh:mm a") Tags: knowledge, tools

https://www.debuggingbook.org/html/Timeout.html

import time

import signal

import sys

from types import FrameType, TracebackType

from typing import Any, Callable, Optional, Union, Type

class SignalTimeout:

"""Variant 1: Unix (using signals, efficient)

Execute a code block raising a timeout.

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

Args:

timeout: timeout duration in seconds

"""

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()

class GenericTimeout:

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

Execute a code block raising a timeout.

Example:

*>>>* def some_long_running_function() -> None:

*>>>*   i = 10000000

*>>>*   while i > 0:

*>>>*     i -= 1

*>>>* try:

*>>>*   with GenericTimeout(0.2):

*>>>*     some_long_running_function()

*>>>*     print("Complete!")

*>>>* except TimeoutError:

*>>>*   print("Timeout!")

Args:

timeout: timeout duration in seconds

"""

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)

'''

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

example:

*>>>* import time

*>>>* from timeout import Timeout

*>>>* try:

*>>>*   with Timeout(0.2):

*>>>*     time.sleep(0.1)

*>>>*     print("complete!")

*>>>* except TimeoutError:

*>>>*   print("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.

Source: https://www.debuggingbook.org/html/Timeout.html

'''

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

Graph View