Executors

Executors dispatch Task objects produced by a ParallelEvaluator to a concrete execution mechanism (threads, processes, or an HPC cluster).

See Parallel Evaluation for usage.

ropt.workflow.executors.Executor

Bases: ABC

Abstract base class for executor components within an optimization workflow.

Subclasses must implement the following abstract methods and properties:

• start: Starts the executor.
• cancel: Stops the executor.
• task_queue: Retrieves the executor's task queue.
• loop: Retrieves the currently running asyncio loop.
• task_group: The asyncio.Taskgroup used by this executor.
• is_running: Checks if the executor is running.

task_queue abstractmethod property

task_queue: Queue[Any]

The task queue.

loop abstractmethod property

loop: AbstractEventLoop | None

The asyncio loop used by this executor.

task_group abstractmethod property

task_group: TaskGroup | None

The task group used by this executor.

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
RuntimeError	If the executor is already running or using an external queue.

cancel abstractmethod

cancel() -> None

Stop the executor.

is_running abstractmethod

is_running() -> bool

Check if the executor is running.

Returns:

Type	Description
bool	True if the executor is running.

ropt.workflow.executors.Task dataclass

Bases: ABC

A task to be executed by a worker.

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.
results_queue	ResultsQueue	The queue to put the result in.
result	Any | None	The result of the function, or None if no result is available.
name	str | None	Optional unique name of the task.

put_result

put_result(result: Any) -> None

Put the result in the result queue.

cancel_all

cancel_all() -> None

Stop putting results in the result queue.

ropt.workflow.executors.ThreadingExecutor

Bases: ExecutorBase

An executor that dispatches tasks to worker threads.

__init__

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

Initialize the executor.

Parameters:

Name	Type	Description	Default
workers	int	The number of workers to use.	1
queue_size	int	Maximum size of the tasks queue.	0

start async

start(task_group: TaskGroup) -> None

Start the executor.

Parameters:

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

cleanup

cleanup() -> None

Clean up the executor.

ropt.workflow.executors.MultiprocessingExecutor

Bases: ExecutorBase

An executor that employs a pool of multiprocessing workers.

__init__

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

Initialize the executor.

Parameters:

Name	Type	Description	Default
workers	int	Number of worker processes.	1
queue_size	int	Maximum task queue size (0 = unlimited).	0
max_tasks_per_child	int | None	Restart workers after this many tasks (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

cleanup

cleanup() -> None

Clean up the executor.

ropt.workflow.executors.HPCExecutor

Bases: ExecutorBase

An executor for submitting tasks to an HPC cluster.

Interfaces with an HPC queueing system (e.g. 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,
    queue_size: int = 0,
    interval: float = 1,
    queue_type: str = "slurm",
    template: str | None = None,
    config_path: Path | str | None = None,
    cluster: str | None = None,
    queue: str | None = None,
    cores: int = 1,
    retries: int = 30,
    cleanup: bool = True,
) -> None

Initialize the HPC executor.

See Parallel Evaluation for configuration details.

Parameters:

Name	Type	Description	Default
workdir	Path | str	Shared filesystem directory for temporary I/O files.	'./'
workers	int	Maximum concurrent HPC jobs.	1
queue_size	int	Maximum task queue size (0 = unlimited).	0
interval	float	Polling interval in seconds.	1
queue_type	str	Queueing system type (e.g. "slurm").	'slurm'
template	str | None	Optional submission script template string.	None
config_path	Path | str | None	Optional path to pysqa configuration directory.	None
cluster	str | None	Optional cluster name.	None
queue	str | None	Optional queue/partition name.	None
cores	int	CPUs per task.	1
retries	int	Number of polling retries before declaring a task failed.	30
cleanup	bool	Whether to remove task files after result retrieval.	True

Raises:

Type	Description
RuntimeError	If neither a template is provided nor a valid config_path can be found.

start async

start(task_group: TaskGroup) -> None

Start the executor.

Parameters:

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

cleanup

cleanup() -> None

Clean up the executor resources.