Servers

Servers dispatch Task objects produced by an AsyncEvaluator to a concrete execution backend (asyncio, multiprocessing, or HPC). EventServer dispatches events from worker threads to handlers running in the asyncio event loop.

See Parallel Evaluation for usage.

ropt.workflow.servers.Server

Bases: ABC

Abstract base class for server components within an optimization workflow.

Subclasses must implement the following abstract methods and properties:

• start: Starts the server.
• cancel: Stops the server.
• task_queue: Retrieves the servers task queue.
• loop: Retrieves the currently running asyncio loop.
• task_group: The asyncio.Taskgroup used by this server.
• is_running: Checks if the server 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 server.

task_group abstractmethod property

task_group: TaskGroup | None

The task group used by this server.

start abstractmethod async

start(task_group: TaskGroup) -> None

Start the evaluation server.

Parameters:

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

Raises:

Type	Description
RuntimeError	If the evaluator is already running or using an external queue.

cancel abstractmethod

cancel() -> None

Stop the evaluation server.

is_running abstractmethod

is_running() -> bool

Check if the server is not running.

Returns:

Type	Description
bool	True if the server is not running.

ropt.workflow.servers.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.servers.ThreadingServer

Bases: ServerBase

An evaluator server that dispatches tasks to worker threads.

__init__

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

Initialize the server.

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

Parameters:

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

cleanup

cleanup() -> None

Cleanup the server.

ropt.workflow.servers.MultiprocessingServer

Bases: ServerBase

An evaluator server 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 server.

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

Parameters:

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

cleanup

cleanup() -> None

Cleanup the server.

ropt.workflow.servers.HPCServer

Bases: ServerBase

A server 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 server.

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

Parameters:

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

cleanup

cleanup() -> None

Clean up the server resources.

ropt.workflow.servers.EventServer

An event server that dispatches events to handlers from the asyncio event loop.

See Parallel Evaluation for usage.

add_event_handler

add_event_handler(
    handler: EventHandler, *, run_in_thread: bool = False
) -> None

Add an event handler.

By default the handler is called directly in the event loop's thread, which is efficient for handlers that only do in-memory work. Pass run_in_thread=True for handlers that perform blocking operations such as file I/O, database writes, or network calls. Multiple handlers with run_in_thread=True that match the same event are dispatched in parallel via asyncio.gather.

Parameters:

Name	Type	Description	Default
handler	EventHandler	The handler to add.	required
run_in_thread	bool	If True, dispatch via the thread pool instead of the event loop.	False

put_event

put_event(event: EnOptEvent) -> None

Submit an event from any thread.

Parameters:

Name	Type	Description	Default
event	EnOptEvent	The event to submit.	required

is_running

is_running() -> bool

Check if the server is running.

Returns:

Type	Description
bool	True if the server is running.

start async

start(task_group: TaskGroup) -> None

Start the event server.

Parameters:

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

Raises:

Type	Description
RuntimeError	If the server is already running.

cancel

cancel() -> None

Stop the event server.