mlox.scheduler

@dataclass

class QueueEntry:

QueueEntry( state: str, process: Callable, callback: Callable, params_process: dict, params_callback: dict)

state: str

process: Callable

callback: Callable

params_process: dict

params_callback: dict

class ProcessSchedulerError:

ProcessSchedulerError(e, tb)

class ProcessScheduler:

ProcessScheduler( max_processes: int = 2, watchdog_wakeup_sec: int = 1, watchdog_timeout_sec: int = 1500, disable_garbage_collection: bool = False)

STATE_IDLE = 'Idle'

STATE_RUNNING = 'Running'

STATE_FINISHED = 'Finished'

STATE_TIMEOUT = 'Failure (timeout)'

STATE_ERROR = 'Failure (unknown)'

max_processes: int

watchdog_wakeup_sec: int

watchdog_timeout_sec: int

watchdog_cleanup_iter: int

watchdog_cleanup_cntr: int

gc: bool

queue_lock

queue: Dict[int, QueueEntry]

queue_key_counter

processes_results: 'DictProxy[int, object]'

processes: list[tuple[datetime.datetime, multiprocessing.context.Process, int]]

parent_pid

watchdog_name

watchdog_name_shutdown_postfix

watchdog_shutdown

watchdog_timer: threading.Timer | None

def shutdown(self) -> None:

def parent_process_exists(self) -> bool:

Return True if process with pid exists, False otherwise.

def get_next(self) -> int:

Generator that yields indices of idle processes asynchronously. Usage: for idx in scheduler.get_next(): ...

def remove_entries_by_state(self, state: str | None = None) -> None:

Remove all entries in the queues with the given state (default: FINISHED). Blocks add and get_next while running.

def add( self, process: Callable, callback: Callable, params_process: dict, params_callback: dict) -> None: