pynenc.runner.multi_thread_runner

Module Contents

Classes

ProcessState
ProcessStatus
MultiThreadRunner	MultiThreadRunner spawns separate processes, each running a ThreadRunner. It scales processes based on pending invocations and terminates those that remain idle.

Functions

thread_runner_process_main

Runs a ThreadRunner in a separate process.

API

class pynenc.runner.multi_thread_runner.ProcessState(*args, **kwds)

Bases: enum.Enum

ACTIVE

‘active’

IDLE

‘idle’

class pynenc.runner.multi_thread_runner.ProcessStatus

Bases: typing.NamedTuple

last_update: float

None

active_count: int

None

state: pynenc.runner.multi_thread_runner.ProcessState

None

is_idle(now: float, idle_timeout: float) → bool

Return True if the process is idle and has been idle longer than idle_timeout.

pynenc.runner.multi_thread_runner.thread_runner_process_main(app: pynenc.app.Pynenc, *, runner_cache: dict, shared_status: dict[str, pynenc.runner.multi_thread_runner.ProcessStatus], process_key: str) → None

Runs a ThreadRunner in a separate process.

class pynenc.runner.multi_thread_runner.MultiThreadRunner(app: pynenc.app.Pynenc, runner_cache: Optional[dict] = None)

Bases: pynenc.runner.base_runner.BaseRunner

MultiThreadRunner spawns separate processes, each running a ThreadRunner. It scales processes based on pending invocations and terminates those that remain idle.

Initialization

processes: dict[str, multiprocessing.Process]

None

manager: multiprocessing.Manager

None

shared_status: dict[str, pynenc.runner.multi_thread_runner.ProcessStatus]

None

max_processes: int

None

runner_cache: dict

None

conf() → pynenc.conf.config_runner.ConfigMultiThreadRunner

property cache: dict

Returns the cache for the ProcessRunner instance.

static mem_compatible() → bool

Indicates if the runner is compatible with in-memory components.

Returns:

False, as each Thread runs in a separate process with independent memory.

property max_parallel_slots: int

The maximum number of parallel tasks that the runner can handle.

Returns:

An integer representing the maximum number of parallel tasks, based on config or CPU count.

_on_start() → None

_spawn_thread_runner_process() → None

_on_stop() → None

_safe_remove_shared_state(key: str) → None

Safely remove a process’s shared state, handling manager shutdown cases.

Parameters:

key (str) – The process key to remove from shared state

_on_stop_runner_loop() → None

Internal method called after receiving a signal to stop the runner loop. Clears the wait_invocation dictionary.

_cleanup_dead_processes() → None

Remove processes that are no longer alive from our tracking dictionaries.

_scale_up_processes() → None

Spawns new processes based on enforce_max_processes setting and pending tasks.

_terminate_idle_processes() → None

Terminates processes that are idle longer than the configured timeout. A process is considered idle if its shared status indicates IDLE and the time since its last update exceeds idle_timeout_process_sec. Respects enforce_max_processes if enabled.

runner_loop_iteration() → None

waiting_for_results(running_invocation: Optional[pynenc.invocation.dist_invocation.DistributedInvocation], result_invocations: list[pynenc.invocation.dist_invocation.DistributedInvocation], runner_args: Optional[dict[str, Any]] = None) → None

Handle waiting for results when called from outside any process. This happens when checking results from the main thread or test environment.