Skip to content

aea.helpers.exec_timeout

Python code execution time limit tools.

TimeoutResult Objects

class TimeoutResult()

Result of ExecTimeout context manager.

__init__

def __init__() -> None

Init.

set_cancelled_by_timeout

def set_cancelled_by_timeout() -> None

Set code was terminated cause timeout.

is_cancelled_by_timeout

def is_cancelled_by_timeout() -> bool

Return True if code was terminated by ExecTimeout cause timeout.

Returns:

bool

TimeoutException Objects

class TimeoutException(BaseException)

TimeoutException raised by ExecTimeout context managers in thread with limited execution time.

Used internally, does not propagated outside of context manager

BaseExecTimeout Objects

class BaseExecTimeout(ABC)

Base class for implementing context managers to limit python code execution time.

exception_class - is exception type to raise in code controlled in case of timeout.

__init__

def __init__(timeout: float = 0.0) -> None

Init.

Arguments:

  • timeout: number of seconds to execute code before interruption

__enter__

def __enter__() -> TimeoutResult

Enter context manager.

Returns:

TimeoutResult

__exit__

def __exit__(exc_type: Type[Exception], exc_val: Exception,
             exc_tb: TracebackType) -> None

Exit context manager.

Arguments:

  • exc_type: the exception type
  • exc_val: the exception
  • exc_tb: the traceback

ExecTimeoutSigAlarm Objects

class ExecTimeoutSigAlarm(BaseExecTimeout)

ExecTimeout context manager implementation using signals and SIGALARM.

Does not support threads, have to be used only in main thread.

ExecTimeoutThreadGuard Objects

class ExecTimeoutThreadGuard(BaseExecTimeout)

ExecTimeout context manager implementation using threads and PyThreadState_SetAsyncExc.

Support threads. Requires supervisor thread start/stop to control execution time control. Possible will be not accurate in case of long c functions used inside code controlled.

__init__

def __init__(timeout: float = 0.0) -> None

Init ExecTimeoutThreadGuard variables.

Arguments:

  • timeout: number of seconds to execute code before interruption

start

@classmethod
def start(cls) -> None

Start supervisor thread to check timeouts.

Supervisor starts once but number of start counted.

stop

@classmethod
def stop(cls, force: bool = False) -> None

Stop supervisor thread.

Actual stop performed on force == True or if number of stops == number of starts

Arguments:

  • force: force stop regardless number of start.