simplebench.timeout.timeout module

Run a callable with a timeout, ensuring safe termination.

class simplebench.timeout.timeout.Timeout(timeout_interval: float | int)

Bases: Generic[_T]

Executes a callable in a separate daemon thread and enforces a timeout.

This implementation avoids the risks of raising asynchronous exceptions by running the target function in a worker thread and using thread.join() with a timeout. If the worker thread is still alive after the timeout, it is considered to have timed out. Because the worker is a daemon thread, the main program can exit without needing to forcefully terminate it.

Warning

After a timeout, it is recommended to exit the program cleanly rather than continuing execution.

The worker thread may still be running in the background and this can lead to undefined behavior. This design ensures that the main thread remains responsive and avoids the complexities of forcibly terminating threads.

Example of using the new Timeout class.

  def my_long_running_task():
      time.sleep(10)
      return "done"

  timeout = Timeout(5.0)
  try:
      result = timeout.run(my_long_running_task)
      print(f"Task finished with result: {result}")
  except SimpleBenchTimeoutError as e:
      print(f"Task {e.func_name} timed out. Final state: {timeout.state}")
      sys.exit(1)

Creates a Timeout instance.

A Timeout instance can be used to run a callable with via the run method in a separate daemon thread with the specified timeout interval.

If the callable does not complete within the timeout, a SimpleBenchTimeoutError is raised. On Windows, the worker thread will continue running in the background as a daemon thread. On Linux or Unix based platforms, the worker thread may be forcefully terminated after a grace period.

Warning

You should avoid continuing execution after a timeout.

The worker thread may still be running in the background and this can lead to undefined behavior. As may the forceful termination of the worker thread if it is not still running.

This design ensures that the main thread remains responsive and prevents ‘hanging’ due to unresponsive or long-running operations.

So you should catch the SimpleBenchTimeoutError exception, perform any necessary cleanup, and then exit the program cleanly.

A Timeout instance can be reused to run multiple callables with the same timeout interval.

Parameters:

timeout_interval – float or int duration to wait for the callable to complete.

Raises:

• SimpleBenchTypeError – If timeout_interval is not a float or int.

• SimpleBenchValueError – If timeout_interval is not greater than zero.

run(_calling_func: ~typing.Callable[[~_P], ~simplebench.timeout.timeout._RT], *args: ~typing.~_P, **kwargs: ~typing.~_P) → _RT

Runs the given callable in a worker thread with a timeout.

Parameters:

• _calling_func – The callable to execute.

• args – Positional arguments to pass to the callable.

• kwargs – Keyword arguments to pass to the callable.

Raises:

• SimpleBenchTimeoutError – If the callable does not complete within the specified timeout.

• BaseException – Any exception raised by the callable will be re-raised in the main thread.

Returns:

The return value of the callable if it completes successfully.

property state: TimeoutState

Get the final state of the timeout execution.

property timeout_interval: float

Get the timeout interval in seconds.