rlightning.utils.placement

Placement module for global resource management and component scheduling.

This module provides: - GlobalResourceManager: Singleton manager for resource-aware component scheduling - ResourcePoolPlanner: Plans resource pools based on placement strategy (auto/manual) - ComponentScheduling: Defines resource requirements for each component type - PlacementStrategy: Strategies for creating Ray placement groups

class rlightning.utils.placement.ColocatedPlacementStrategy(scheduling: ComponentScheduling)

Bases: ResourcePoolPlacementStrategy

Colocated placement strategy.

All components share a global resource pool. Workers are distributed across nodes with consideration for resource utilization.

create_placement_groups(resource_pools: Dict[str, Dict[str, Any]] | None = None, max_colocate_count: int = 10, **kwargs) → Dict[str, MockModule('ray.util.placement_group.PlacementGroup')]

Create placement groups for colocated components.

Parameters:

• resource_pools – Dictionary with “global_pool” configuration.

• max_colocate_count – Maximum number of components per node.

• **kwargs – Additional strategy options kept for interface compatibility.

class rlightning.utils.placement.ComponentAllocation(component_type: str, node_to_gpu_indices: Dict[str, List[str]], total_gpus: int, worker_count: int)

Bases: object

Represents resource allocation for a specific component type.

component_type

Type of component (train, eval, env, buffer)

Type:

str

gpu_indices

List of GPU index ranges, e.g., [“0-7”, “8-15”]

total_gpus

Total number of GPUs allocated

Type:

int

worker_count

Number of workers for this component

Type:

int

component_type: str

get_node_index_string(node_id: str) → str

Get index string for a specific node_id.

node_to_gpu_indices: Dict[str, List[str]]

to_dict() → Dict[str, Any]

Convert to dictionary representation.

to_index_string() → str

Convert GPU indices to a compact string format like ‘0-7, 8-15’.

total_gpus: int

worker_count: int

class rlightning.utils.placement.ComponentScheduling(env_worker: List[Scheduling] | None = None, train_worker: Scheduling | None = None, eval_worker: Scheduling | None = None, buffer_worker: Scheduling | None = None)

Bases: object

Scheduling configuration for all components.

This class holds the resource requirements for each component type: - env_worker: Environment workers (can have multiple groups) - train_worker: Training policy workers - eval_worker: Evaluation policy workers (rollout) - buffer_worker: Data buffer storage workers

adjust_buffer_worker_num(train_node_count: int) → None

Adjust the buffer worker number to match the train worker number.

buffer_worker: Scheduling | None = None

env_worker: List[Scheduling] | None = None

eval_worker: Scheduling | None = None

get_component_requirements(component_type: str) → Tuple[float, int]

Get resource requirements for all components.

Returns:

Tuple of (total_gpus, total_cpus) for the specified component type.

Raises:

ValueError – If component type is invalid or the worker is not configured.

infer_auto_buffer_worker_num(cluster_info: Dict[str, Any]) → None

Infer the auto buffer worker number based on the train worker number.

rollout_pool_requirements() → Tuple[float, int]

Calculate resource requirements for rollout pool (eval + env).

Returns:

Tuple of (total_gpus, total_cpus).

summary() → str

Generate a human-readable summary.

to_dict() → Dict[str, Any]

Convert to dictionary representation.

train_pool_requirements() → Tuple[float, int]

Calculate resource requirements for train pool (train + buffer).

Returns:

Tuple of (total_gpus, total_cpus).

train_worker: Scheduling | None = None

class rlightning.utils.placement.DefaultPlacementStrategy(scheduling: ComponentScheduling)

Bases: PlacementStrategy

Default placement strategy with no specific placement groups.

Uses node affinity for buffer workers when multiple are needed, but otherwise relies on Ray’s default scheduling.

buffer_strategys

List of scheduling strategies for buffer workers.

_node_info

Node information for resource-aware choices.

create_placement_groups() → Dict[str, MockModule('ray.util.placement_group.PlacementGroup')]

Create placement groups based on the strategy.

For multiple buffer storages, creates node affinity strategies to place each buffer on a different node.

Returns:

Empty dictionary (no placement groups created).

get_scheduling_strategy(component_type: str, worker_index: int = 0) → Any

Get scheduling strategy for a component.

Parameters:

• component_type – Type of component.

• worker_index – Index of the worker.

Returns:

Node affinity strategy for buffers, ‘DEFAULT’ otherwise.

class rlightning.utils.placement.DisaggregatePlacementStrategy(scheduling: ComponentScheduling)

Bases: ResourcePoolPlacementStrategy

Disaggregate placement strategy.

Allocates separate resource pools for: - train_pool: Train workers + Buffer workers (colocated) - rollout_pool: Eval workers + Env workers (colocated)

This provides resource isolation between training and evaluation.

create_placement_groups(resource_pools: Dict[str, Dict[str, Any]] | None = None, **kwargs) → Dict[str, MockModule('ray.util.placement_group.PlacementGroup')]

Create placement groups based on resource pools.

Parameters:

• resource_pools – Dictionary with “train_pool” and “rollout_pool” configurations.

• **kwargs – Additional strategy options kept for interface compatibility.

class rlightning.utils.placement.GlobalResourceManager

Bases: object

Singleton resource manager for global resource management.

This manager orchestrates the following workflow: 1. Initialize with configuration and scheduling requirements 2. Discover cluster resources via ResourcePoolPlanner 3. Plan resource pools based on placement strategy 4. Create placement groups via the selected strategy 5. Track component distribution for monitoring

Usage:

manager = GlobalResourceManager.get_instance() manager.initialize(placement_config, scheduling) strategy = manager.get_scheduling_strategy(“train”, worker_index=0)

cleanup() → None

Clean up placement groups.

get_cluster_summary() → Dict[str, Any]

Get a summary of cluster resources and allocation.

get_component_distribution() → Dict[str, Dict[str, Dict[str, Any]]]

Return mapping of node_id -> {component_type: {“count”: N, “ids”: […]}}.

Format is aligned with ResourcePoolPlacementStrategy._node_component_distribution.

classmethod get_instance() → GlobalResourceManager

Get the singleton instance..

Returns:

The singleton GlobalResourceManager instance.

get_placement_config() → Config | None

Get the placement configuration.

Returns:

The placement configuration object.

get_placement_strategy() → str

Get the placement strategy.

get_pool_for_component(component_type: str) → ResourcePool | None

Get the resource pool containing a specific component type.

get_resource_planner() → ResourcePoolPlanner | None

Get the resource planner instance.

get_resource_pools(pool_name: str = None) → Dict[str, ResourcePool]

Get the planned resource pool by pool_name.

get_scheduling() → ComponentScheduling | None

Get the component scheduling requirements.

Returns:

The component scheduling object.

get_scheduling_strategy(component_type: str, worker_index: int = 0) → Any

Get Ray scheduling strategy for a component.

Parameters:

• component_type – Type of component (“train”, “eval”, “buffer”, “env”).

• worker_index – Index of the worker within its type.

Returns:

Ray scheduling strategy (PlacementGroupSchedulingStrategy or “DEFAULT”).

get_storage_to_train_workers() → Dict[int, List[int]] | None

Get storage -> train worker mapping from the active strategy.

Returns:

Dictionary mapping storage indices to train worker indices.

Raises:

RuntimeError – If PlacementManager is not initialized.

initialize(cluster_cfg: ClusterConfig, scheduling: ComponentScheduling, config_path: str | None = None) → None

Initialize the placement manager with configuration.

This method performs the following steps: 1. Store configuration and scheduling requirements 2. Discover cluster resources 3. Validate scheduling requirements against available resources 4. Plan resource pools based on placement strategy 5. Create placement groups via the selected strategy 6. Track initial component distribution

Parameters:

• cluster_cfg – Cluster configuration, including placement settings.

• scheduling – Component scheduling requirements.

• config_path – Optional path to the configuration file.

property is_initialized: bool

Check if the placement manager is initialized with a strategy..

Returns:

True if a strategy has been set.

refresh_component_distribution() → Dict[str, Dict[str, int]]

Refresh and return the current component distribution across nodes.

This queries Ray’s actor registry to get real-time distribution.

classmethod reset() → None

Reset singleton instance (mainly for testing).

save_yaml_config(cluster_config_path: str, filename: str = 'resource_pool_auto.yaml', subdir: str = 'resource_pool') → str

Save current resource_pool yaml to disk.

property strategy: PlacementStrategy | None

Get current placement strategy.

Returns:

The current PlacementStrategy instance or None.

class rlightning.utils.placement.NodeResource(node_id: str, ip: str, total_cpus: int, total_gpus: int, allocations: Dict[str, ~typing.List[~typing.Tuple[int, int]]]=<factory>, gpu_cursor: int = 0, max_allocated_gpus: int = 0)

Bases: object

Represents a node’s resources.

This class is used in two contexts: - Cluster discovery: total_* reflects node total capacity; available_* is computed. - ResourcePool membership: total_* represents the node’s capacity,

and allocations records per-component GPU index ranges within the node’s index space.

Resource tracking uses gpu_cursor to track the next available GPU index. available_gpus is computed as total_gpus - gpu_cursor.

allocate(gpus: int | List[int], component_types: List[str] | None = None, consume: bool = True) → None

Allocate resources directly on this node.

Modifies self.allocations and advances gpu_cursor.

Parameters:

• gpus – GPU units to allocate. Can be int (same for all) or list (per-component).

• component_types – Components to record allocations for. If None, just advances cursor.

• consume – If True, consume the GPUs from the node.

allocations: Dict[str, List[Tuple[int, int]]]

property available_gpus: int

Remaining GPUs available for allocation.

property component_types: List[str]

Get list of component types that have allocations on this node.

copy() → NodeResource

Create a deep copy of this node resource.

gpu_cursor: int = 0

has_resources(cpus: int = 0, gpus: int = 0) → bool

Check if node has sufficient available resources.

ip: str

property is_empty: bool

Check if node has no more allocatable resources.

max_allocated_gpus: int = 0

node_id: str

total_cpus: int

total_gpus: int

class rlightning.utils.placement.PlacementStrategy(scheduling: ComponentScheduling)

Bases: ABC

Abstract base class for placement strategies.

Defines the interface for creating Ray placement groups and determining scheduling strategies for different component types.

scheduling

Cluster scheduling configuration.

placement_groups

Created placement groups by name.

storage_to_train_workers

Mapping from storage index to train worker indices.

cleanup() → None

Clean up placement groups.

abstractmethod create_placement_groups() → Dict[str, MockModule('ray.util.placement_group.PlacementGroup')]

Create placement groups based on the strategy.

Returns:

Dictionary mapping group names to PlacementGroup instances.

abstractmethod get_scheduling_strategy(component_type: str, worker_index: int = 0) → Any

Get scheduling strategy for a specific component.

Parameters:

• component_type – Type of component (‘env’, ‘train’, ‘eval’, ‘buffer’).

• worker_index – Index of the worker within its type.

Returns:

Scheduling strategy (PlacementGroupSchedulingStrategy or ‘DEFAULT’).

get_storage_to_train_workers() → Dict[int, List[int]]

Return storage -> train worker mapping (may be empty if not used).

Returns:

Dictionary mapping storage indices to lists of train worker indices.

class rlightning.utils.placement.ResourcePool(name: str, nodes: ~typing.List[~rlightning.utils.placement.resource_pool.NodeResource], _component_types: ~typing.List[str] | None = None, component_allocations: ~typing.Dict[str, ~rlightning.utils.placement.resource_pool.ComponentAllocation] = <factory>)

Bases: object

A resource pool containing allocated nodes for specific components.

This class tracks: - Which nodes belong to this pool - Which component types use this pool - Fine-grained GPU index allocations per component per node

component_types is auto-inferred from node allocations if not provided. If ‘train’ exists in allocations, ‘buffer’ is automatically added.

component_allocations: Dict[str, ComponentAllocation]

property component_types: List[str]

Get component types for this pool.

Auto-infers from node allocations if not explicitly set. Auto-adds ‘buffer’ if ‘train’ exists.

classmethod from_yaml_dict(pool_cfg: Dict[str, Any], cluster_nodes: Dict[str, NodeResource], *, used_node_ids: set[str] | None = None) → ResourcePool

Build a ResourcePool from a YAML pool dict.

Supported input fields:

• name: str

• num_node: int

• num_gpus: int (per-node) OR List[int] (per-node gpus list)

• optional node_ids: List[str] to pin pools to specific nodes

• component keys: train/eval/env/buffer/… with values like “0-7, 8-15”

component_types is auto-inferred from allocations if not explicitly listed. ‘buffer’ is auto-added if ‘train’ exists.

get_component_indices(component_type: str) → str

Get the GPU index string for a component type across all nodes.

Returns:

Index string like “0-7” or “0-7, 8-15” for multi-node.

get_component_node_count(component_type: str) → int

Get the number of nodes that have the component type.

name: str

property node_ids: List[str]

List of node IDs in this pool.

nodes: List[NodeResource]

property num_nodes: int

Number of nodes in this pool.

to_dict() → Dict[str, Any]

Convert to dictionary for serialization.

to_yaml_dict() → Dict[str, Any]

Convert to YAML-friendly dictionary format.

Output format:

name: “train_pool” num_node: 1 num_gpus: 8 # per-node allocated gpus, NOT total train: “0-7”

property total_cpus: int

Total available CPUs in this pool.

property total_gpus: int

Total available GPUs in this pool.

class rlightning.utils.placement.ResourcePoolPlanner(scheduling: ComponentScheduling, cluster_info: Dict[str, Any] | None = None)

Bases: object

Plans and manages resource pools for component placement.

This planner: 1. Discovers cluster resources 2. Validates scheduling requirements against available resources 3. Allocates resources to pools based on the placement strategy 4. Tracks fine-grained component-to-GPU mappings

discover_cluster_resources() → Dict[str, NodeResource]

Discover and cache cluster node resources.

Returns:

Dictionary mapping node_id to NodeResource.

get_cluster_info() → Dict[str, Any]

Get raw cluster information.

get_component_node_count(component_type: str) → int

Get the number of nodes that have the component type.

get_node_resources() → Dict[str, NodeResource]

Get node resources dictionary.

get_pool_for_component(component_type: str) → ResourcePool | None

Get the resource pool that contains a specific component type.

get_resource_pools(pool_name: str = None) → Dict[str, ResourcePool]

Get the planned resource pools.

load_manual_resource_pools(resource_pool_cfg: List[Dict[str, Any]]) → Dict[str, ResourcePool]

Load resource pools from a manual YAML config list.

plan_resource_pools(strategy: str = 'disaggregate', env_strategy: str = 'default') → Dict[str, ResourcePool]

Plan resource pools based on placement strategy.

Parameters:

• strategy – The placement strategy to use.

• env_strategy – The environment placement strategy.

Returns:

Dictionary mapping pool name to ResourcePool.

resource_summary: Dict[str, Any]

summary() → Dict[str, Any]

Get a summary of the resource planning.

to_yaml_config() → Dict[str, Any]

Generate a YAML-compatible configuration representing the resource pools.

Output format:

resource_pool:

• name: “train_pool” num_node: 1 num_gpus: 8 # per-node GPUs train: “0-7”

• name: “rollout_pool” num_node: 2 num_gpus: 8 # per-node GPUs eval: “0-7, 8-15” env: “0-7, 8-15”

validate_scheduling(strategy: str = 'disaggregate', env_strategy: str = 'default') → Tuple[bool, str]

Validate that cluster resources can satisfy scheduling requirements.

Parameters:

• strategy – The placement strategy to use.

• env_strategy – The environment strategy to use.

Returns:

Tuple of (is_valid, error_message).

class rlightning.utils.placement.Scheduling(worker_num: int, num_cpus: int, num_gpus: float, node_list: str | None = None)

Bases: object

Scheduling configuration for a single worker type.

worker_num

Number of workers of this type.

Type:

int

num_cpus

Number of CPUs per worker.

Type:

int

num_gpus

Number of GPUs per worker.

Type:

float

node_list

Optional comma-separated list of node IDs for placement.

Type:

str | None

node_list: str | None = None

num_cpus: int

num_gpus: float

to_dict() → Dict[str, Any]

Convert to dictionary representation.

total_cpus() → int

Total CPU requirements for all workers of this type.

total_gpus() → float

Total GPU requirements for all workers of this type.

worker_num: int

rlightning.utils.placement.get_global_resource_manager() → GlobalResourceManager | None

Get the initialized global resource manager singleton instance.

Returns:

GlobalResourceManager if initialized, None otherwise.

rlightning.utils.placement.setup_cluster_scheduling(cfg: MainConfig) → ComponentScheduling

Set up the scheduling configuration for all components from MainConfig.

This function: 1. Reads cluster configuration 2. Validates buffer storage configuration 3. Creates Scheduling objects for each component type

Parameters:

cfg – The main configuration object.

Returns:

ComponentScheduling with all component requirements.

Raises:

ValueError – If configuration is invalid.

rlightning.utils.placement.setup_component_scheduling(cfg: MainConfig) → ComponentScheduling

Set up the scheduling configuration for all components from MainConfig.

This function: 1. Reads cluster configuration 2. Validates buffer storage configuration 3. Creates Scheduling objects for each component type

Parameters:

cfg – The main configuration object.

Returns:

ComponentScheduling with all component requirements.

Raises:

ValueError – If configuration is invalid.

Submodules

• rlightning.utils.placement.placement_manager
• rlightning.utils.placement.placement_strategies
• rlightning.utils.placement.resource_pool
• rlightning.utils.placement.scheduling