Executors

Executors run the WorkItem objects of a Submission, produced by a ParallelEvaluator, on a concrete execution mechanism (threads, processes, local jobs, or an HPC cluster).

See Parallel Evaluation for usage.

Executor is the interface a compute step sees; ExecutorBase implements the submission bookkeeping shared by the built-in executors and is the class to subclass when adding a new execution mechanism.

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 has left, so running its work items only occupies a worker.

Subclasses must implement the following abstract methods:

start: Starts the executor.
cancel: Stops the executor.
submit: Hands over a submission.
is_running: Reports whether the executor accepts work.

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.

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`	The result of the function, only meaningful once the work item has been delivered.
`name`	`str \| None`	Optional unique name of the work item.

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 has left.

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 was delivered.

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.

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 work items (`None` = 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.

Parameters:

Name	Type	Description	Default
`workdir`	`Path \| str \| None`	Directory for each work item's serialized I/O files and captured output. The default is a private temporary directory that this executor creates and removes when it closes — unless there is something in it to read, that is: if a work item failed, or `cleanup` is off, the directory is kept and its path logged.	`None`
`workers`	`int`	Maximum number of jobs running at once.	`1`
`interval`	`float`	Polling interval in seconds. Small by default: a local process is finished the moment it exits, so this is dead time rather than politeness towards a scheduler.	`0.1`
`retries`	`int`	Number of extra polls to wait for a work item's result. The default of `0` is enough, because a local job writes and renames its result before it exits, so the result is there the moment the process is gone.	`0`
`cleanup`	`bool`	Whether to remove a work item's files once its result is retrieved or its job is cancelled. A work item that failed keeps its captured output, which is the only record of why.	`True`

Raises:

Type	Description
`ValueError`	If `workdir` is not an existing directory, or if `workers`, `interval` or `retries` is out of range.
`ExecutionError`	If the system is not POSIX.

workdir `property`

workdir: Path

The directory the jobs read and write.

Worth asking for when you did not pass one: that directory is temporary, and this is the only way to find 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.

See Parallel Evaluation for configuration details.

Parameters:

Name	Type	Description	Default
`workdir`	`Path \| str`	Shared-filesystem directory for each work item's serialized I/O files; also passed as the job working directory (template-dependent). Must be an existing absolute path: there is no default, because only the caller knows which directory the cluster shares. Work item files are never overwritten, so concurrent executors need distinct workdirs.	required
`workers`	`int`	Maximum concurrent HPC jobs.	`1`
`interval`	`float`	Polling interval in seconds.	`1`
`config_path`	`Path \| str \| None`	The `pysqa` configuration directory, the one holding `queue.yaml` or `clusters.yaml`. Defaults to the site-wide configuration installed alongside ropt.	`None`
`cluster`	`str \| None`	Optional cluster name, when the configuration defines more than one. Defaults to the configured primary.	`None`
`queue`	`str \| None`	Optional name of a queue defined in the configuration. This is not necessarily the scheduler's partition name: it selects a queue entry, whose submission script names the partition. Defaults to the configured primary.	`None`
`template`	`str \| None`	A submission script template, submitted instead of any configuration. Everything the scheduler needs, the partition included, must be in it.	`None`
`scheduler`	`str \| None`	The queueing system a `template` is written for, for example `"slurm"` (the default). Only meaningful with a `template`, since a configuration names its own. `pysqa` calls this `queue_type`.	`None`
`cores`	`int`	CPUs per work item.	`1`
`memory_max`	`int \| str \| None`	Memory per work item. Rendered by the submission script; with a configuration it is also clamped to the queue's limit.	`None`
`run_time_max`	`int \| None`	Run time per work item, typically in seconds. With a configuration, the queue's own limit applies when this is not given.	`None`
`submit_options`	`dict[str, Any] \| None`	Extra variables for the submission script, for whatever the script declares beyond the standard names — an account, a reservation, a GPU request. Entries that are `None` are dropped, so omitting a key and passing `None` mean the same thing.	`None`
`retries`	`int`	Number of extra polls to wait for a work item's result after the first attempt fails (`0` gives up at once). This is about the shared filesystem, not about the scheduler.	`30`
`query_retries`	`int`	Number of extra attempts to query the scheduler after one fails (`0` gives up at once). A run this long fails every job that is out, because nothing can be said about a job that cannot be asked after.	`30`
`cleanup`	`bool`	Whether to remove work item files once their result is retrieved or their job is cancelled. A work item that failed keeps its captured output, which is the only record of why.	`True`

Raises:

Type	Description
`ValueError`	If `workdir` is not an existing absolute path, if `workers`, `interval`, `retries` or `query_retries` is out of range, if arguments from both submission modes are combined, or if `submit_options` names something the executor already passes.
`ExecutionError`	If no configuration can be found, if the requested cluster is unknown, if the queue is not available on the requested cluster, or if the queue cannot be resolved to exactly one cluster.

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.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`

init

work_items `property`

is_finished `property`

init

start `async`

init

start `async`

init

workdir `property`

start `async`

init