Executors

ropt.components.executors

Executors: where an evaluator sends work to be run.

An executor runs the WorkItem objects of a Submission on a concrete mechanism: threads, processes, local jobs, or an HPC cluster. Executor is the interface a compute step sees; ExecutorBase adds the submission bookkeeping the built-in executors share. See Parallel Evaluation for usage, and Implementing a Component for writing one.

ropt.components.executors.Executor

Bases: ABC

Abstract base class for executor components within an optimization workflow.

Handing a submission to an executor transfers responsibility for it: the executor either runs its work items or ends the submission, so a caller waiting in collect is always released. That holds for a submission still queued, one whose work is in flight, and one accepted before the executor was stopped.

A submission that is already finished when it reaches a worker needs no work done: its caller is no longer waiting, so running its work items only occupies a worker.

See Error handling for the distinction an implementation must make between an infrastructure failure, delivered as an ExecutorFailure result, and an exception from the work item's own function, which ends the submission.

start `abstractmethod` `async`

start(task_group: TaskGroup) -> None

Start the executor.

Parameters:

Name	Type	Description	Default
`task_group`	`TaskGroup`	The task group to use.	required

Raises:

Type	Description
`WorkflowError`	If the executor is already running.

cancel `abstractmethod`

cancel() -> None

Stop the executor.

Every submission that was accepted and has not finished must be ended, so that its caller is released rather than left waiting for results that can no longer arrive.

May be called from any thread.

on_worker_loop

on_worker_loop() -> bool

Report whether the caller is on the event loop that runs the work.

Blocking that loop starves the work being waited for, so callers use this to refuse rather than deadlock. Implementations that run their work on an event loop must override this; the default False is for executors that do not have one.

Returns:

Type	Description
`bool`	`True` if the calling thread is running the executor's loop.

on_worker_thread

on_worker_thread() -> bool

Report whether the caller is running as one of this executor's workers.

Such a caller occupies a worker for as long as it waits, so work it submits here can only start once it stops waiting. Implementations whose workers run in this process must override this; the default False is for executors whose workers cannot submit back in the first place, and leaves the refusal in submit inactive.

Returns:

Type	Description
`bool`	`True` if the calling thread is running this executor's work.

is_running `abstractmethod`

is_running() -> bool

Report whether the executor accepts work.

May be called from any thread. A False result means a submission would be aborted rather than run, so a caller that is able to do the work itself may fall back to doing so.

Returns:

Type	Description
`bool`	`True` if the executor accepts work, `False` otherwise.

submit `abstractmethod`

submit(submission: Submission) -> None

Hand a submission to the executor.

May be called from any thread, except one of the executor's own workers: that caller waits for workers it is itself occupying, so it is refused rather than left to deadlock. A submission handed to an executor that is no longer running is aborted rather than queued, so its caller is never left waiting for results that cannot arrive.

Parameters:

Name	Type	Description	Default
`submission`	`Submission`	The submission to run.	required

Raises:

Type	Description
`WorkflowError`	If called from one of this executor's workers.

ropt.components.executors.ExecutorBase

Bases: Executor

A base class for asynchronous executors.

Owns every submission it accepts, so stopping the executor releases all waiting callers from a single place.

Implementations must call _begin_start before creating any resources, and _finish_start once they are in place.

init

__init__() -> None

Initialize the executor.

submit

submit(submission: Submission) -> None

Hand a submission to the executor.

Parameters:

Name	Type	Description	Default
`submission`	`Submission`	The submission to run.	required

Raises:

Type	Description
`WorkflowError`	If called from one of this executor's workers.

is_running

is_running() -> bool

Report whether the executor accepts work.

Returns:

Type	Description
`bool`	`True` if the executor accepts work, `False` otherwise.

on_worker_loop

on_worker_loop() -> bool

Report whether the caller is on the event loop that runs the work.

Returns:

Type	Description
`bool`	`True` if the calling thread is running the executor's loop.

cancel

cancel() -> None

Stop the executor.

May be called from any thread.

ropt.components.executors.WorkItem `dataclass`

A single unit of work to run on a worker.

A work item is a plain description of a call. It carries no delivery channel, so it can be handed to a worker process without dragging the submission that owns it along. result is only meaningful once the work item has been delivered, and holds an ExecutorFailure if the call could not be run at all.

Attributes:

Name	Type	Description
`function`	`Callable[..., Any]`	The function to execute.
`args`	`tuple[Any, ...]`	The arguments to pass to the function.
`kwargs`	`dict[str, Any]`	The keyword arguments to pass to the function.
`result`	`Any`	What the function returned, or why it could not be run.
`name`	`str \| None`	Optional unique name of the work item.

ropt.components.executors.ExecutorFailure `dataclass`

The executor could not run a work item.

Delivered as a work item's result rather than raised, so the executor stays available for further work.

Attributes:

Name	Type	Description
`message`	`str`	What went wrong.

ropt.components.executors.Submission

A group of work items and the channel back to the caller awaiting them.

A submission owns its results channel, so ending it is one operation on one object. Handing a submission to an executor transfers responsibility: the executor either runs the work items or aborts the submission, so a caller blocked in collect is always released.

See Error handling for how infrastructure failures (delivered via deliver) and user-code exceptions (ended via fail) are distinguished.

init

__init__(work_items: Sequence[WorkItem]) -> None

Initialize the submission.

Parameters:

Name	Type	Description	Default
`work_items`	`Sequence[WorkItem]`	The work items to run.	required

work_items `property`

work_items: list[WorkItem]

The work items to run.

Returns:

Type	Description
`list[WorkItem]`	The work items.

is_finished `property`

is_finished: bool

Whether anything more will be delivered.

Returns:

Type	Description
`bool`	`True` if every work item was delivered, or the submission ended.

deliver

deliver(work_item: WorkItem, result: Any) -> None

Deliver the result of a single work item.

Parameters:

Name	Type	Description	Default
`work_item`	`WorkItem`	The work item that ran.	required
`result`	`Any`	The result it produced.	required

fail

fail(exc: BaseException) -> None

End the submission, re-raising an exception in the caller.

Parameters:

Name	Type	Description	Default
`exc`	`BaseException`	The exception raised by the work item's function.	required

abort

abort() -> None

End the submission, releasing the caller with ExecutorStopped.

collect

collect(on_result: Callable[[WorkItem], None]) -> None

Wait for every work item and pass each finished one to on_result.

The submission is ended if this returns early, including when on_result itself raises, so the executor never keeps delivering to a caller that is no longer waiting.

Parameters:

Name	Type	Description	Default
`on_result`	`Callable[[WorkItem], None]`	Callback invoked with each finished work item.	required

Raises:

Type	Description
`ExecutorStopped`	If the submission ended before every result arrived.

ropt.components.executors.ThreadExecutor

Bases: ExecutorBase

An executor that dispatches work items to worker threads.

init

__init__(*, workers: int = 1) -> None

Initialize the executor.

Parameters:

Name	Type	Description	Default
`workers`	`int`	The number of workers to use.	`1`

start `async`

start(task_group: TaskGroup) -> None

Start the executor.

Parameters:

Name	Type	Description	Default
`task_group`	`TaskGroup`	The task group to use.	required

on_worker_thread

on_worker_thread() -> bool

Report whether the caller is running as one of this executor's workers.

A thread started by a work item is not a worker: it holds none of this executor's workers, and is not recognized here.

Returns:

Type	Description
`bool`	`True` if the calling thread is one of this executor's workers.

ropt.components.executors.ProcessExecutor

Bases: ExecutorBase

An executor that employs a pool of multiprocessing workers.

See Parallel Evaluation for details, including the if __name__ == "__main__": guard that the entry point must use.

Warning

Stopping terminates the worker processes and nothing else. A program a work item started itself keeps running, without an error being raised. Use LocalJobExecutor where an evaluation launches external programs.

init

__init__(
    *,
    workers: int = 1,
    max_tasks_per_child: int | None = None,
) -> None

Initialize the executor.

Parameters:

Name	Type	Description	Default
`workers`	`int`	Number of worker processes.	`1`
`max_tasks_per_child`	`int \| None`	Restart workers after this many items, or never.	`None`

start `async`

start(task_group: TaskGroup) -> None

Start the executor.

Parameters:

Name	Type	Description	Default
`task_group`	`TaskGroup`	The task group to use.	required

ropt.components.executors.LocalJobExecutor

Bases: JobExecutorBase

An executor that runs each work item as a separate local process.

Needs no extras and no configuration. See Parallel Evaluation for details.

POSIX only: cancelling kills a job's whole process group, so that whatever the job started itself goes with it, and Windows has no equivalent.

init

__init__(
    *,
    workdir: Path | str | None = None,
    workers: int = 1,
    interval: float = 0.1,
    retries: int = 0,
    cleanup: bool = True,
) -> None

Initialize the local job executor.

The default workdir is a private temporary directory that this executor creates and removes when it closes, unless there is something in it to read: if a work item failed, or cleanup is off, the directory is kept and its path logged. A work item that failed keeps its captured output, which is the only record of why.

A local process is finished the moment it exits, and writes and renames its result before it does, so the small interval is dead time rather than restraint towards a scheduler, and retries=0 is enough.

Parameters:

Name	Type	Description	Default
`workdir`	`Path \| str \| None`	Directory for each work item's files and captured output.	`None`
`workers`	`int`	Maximum number of jobs running at once.	`1`
`interval`	`float`	Polling interval in seconds.	`0.1`
`retries`	`int`	Number of extra polls to wait for a work item's result.	`0`
`cleanup`	`bool`	Whether to remove a work item's files once it settles.	`True`

Raises:

Type	Description
`ValueError`	If `workdir` is missing or an argument is out of range.
`ExecutionError`	If the system is not POSIX.

workdir `property`

workdir: Path

The directory the jobs read and write.

When no directory was passed, this is a temporary one, and this property is the only way to locate it while the executor is running.

Returns:

Type	Description
`Path`	The working directory.

start `async`

start(task_group: TaskGroup) -> None

Start the executor.

Parameters:

Name	Type	Description	Default
`task_group`	`TaskGroup`	The task group to use.	required

ropt.components.executors.HPCExecutor

Bases: JobExecutorBase

An executor for submitting tasks to an HPC cluster.

Interfaces with an HPC queueing system (for example Slurm) via pysqa. Requires ropt[hpc] to be installed.

See Parallel Evaluation for full details on configuration and lifecycle.

init

__init__(
    *,
    workdir: Path | str,
    workers: int = 1,
    interval: float = 1,
    config_path: Path | str | None = None,
    cluster: str | None = None,
    queue: str | None = None,
    template: str | None = None,
    scheduler: str | None = None,
    cores: int = 1,
    memory_max: int | str | None = None,
    run_time_max: int | None = None,
    submit_options: dict[str, Any] | None = None,
    retries: int = 30,
    query_retries: int = 30,
    cleanup: bool = True,
) -> None

Initialize the HPC executor.

There are two ways to say how jobs are submitted, and they are mutually exclusive. Either a pysqa configuration supplies the clusters, the queues and their submission scripts — the normal case on an installed cluster, where only queue need be given — or a template supplies the submission script directly, in which case nothing is configured and scheduler names the queueing system.

workdir must be an existing absolute path, and concurrently running executors need distinct ones: work item files are never overwritten. A queue is not a scheduler partition; it selects a configured entry whose submission script names the partition. retries covers a result the shared filesystem does not yet show, query_retries a scheduler that cannot be reached; the grace period each allows is its count times interval.

config_path is the directory holding queue.yaml or clusters.yaml, and defaults to the configuration installed alongside ropt; cluster and queue default to the configured primary. scheduler is what pysqa calls queue_type, defaults to "slurm", and is meaningful only with a template, which must itself carry everything the scheduler needs, the partition included.

With a configuration, cores is clamped to the queue's minimum and maximum, run_time_max to its maximum and defaults to it when not given, and a numeric memory_max to its maximum — a string such as "4G" is passed through unchanged. Entries of submit_options that are None are dropped, so omitting a key and passing None mean the same thing; a name the executor sets itself is rejected. A work item that failed keeps its captured output even under cleanup, which is the only record of why.

See Parallel Evaluation for configuration details.

Parameters:

Name	Type	Description	Default
`workdir`	`Path \| str`	Shared-filesystem directory for each work item's files.	required
`workers`	`int`	Maximum concurrent HPC jobs.	`1`
`interval`	`float`	Polling interval in seconds.	`1`
`config_path`	`Path \| str \| None`	The `pysqa` configuration directory.	`None`
`cluster`	`str \| None`	Optional cluster name, for a multi-cluster configuration.	`None`
`queue`	`str \| None`	Optional name of a queue defined in the configuration.	`None`
`template`	`str \| None`	A submission script template, used instead of a configuration.	`None`
`scheduler`	`str \| None`	The queueing system a `template` is written for.	`None`
`cores`	`int`	CPUs per work item.	`1`
`memory_max`	`int \| str \| None`	Memory per work item.	`None`
`run_time_max`	`int \| None`	Run time per work item, typically in seconds.	`None`
`submit_options`	`dict[str, Any] \| None`	Extra variables for the submission script.	`None`
`retries`	`int`	Extra polls to wait for a work item's result.	`30`
`query_retries`	`int`	Extra attempts to query the scheduler after one fails.	`30`
`cleanup`	`bool`	Whether to remove a work item's files once it settles.	`True`

Raises:

Type	Description
`ValueError`	If an argument is out of range or the modes are mixed.
`ExecutionError`	If the configuration, cluster or queue cannot be resolved.

Executors

ropt.components.executors

ropt.components.executors.Executor

start abstractmethod async

cancel abstractmethod

on_worker_loop

on_worker_thread

is_running abstractmethod

submit abstractmethod

ropt.components.executors.ExecutorBase

__init__

submit

is_running

on_worker_loop

cancel

ropt.components.executors.WorkItem dataclass

ropt.components.executors.ExecutorFailure dataclass

ropt.components.executors.Submission

__init__

work_items property

is_finished property

deliver

fail

abort

collect

ropt.components.executors.ThreadExecutor

__init__

start async

on_worker_thread

ropt.components.executors.ProcessExecutor

__init__

start async

ropt.components.executors.LocalJobExecutor

__init__

workdir property

start async

ropt.components.executors.HPCExecutor

__init__

start `abstractmethod` `async`

cancel `abstractmethod`

is_running `abstractmethod`

submit `abstractmethod`

init

ropt.components.executors.WorkItem `dataclass`

ropt.components.executors.ExecutorFailure `dataclass`

init

work_items `property`

is_finished `property`

init

start `async`

init

start `async`

init

workdir `property`

start `async`

init