`nemo_rl.distributed.virtual_cluster`#

Module Contents#

Classes#

`ClusterConfig`
`PY_EXECUTABLES`	Command each Ray actor launches under, one entry per uv extra combination.
`RayVirtualCluster`	Creates a virtual distributed cluster using Ray placement groups.

Functions#

`uv_py_executable`	py_executable of a uv-managed venv with the given extras (same shape as PY_EXECUTABLES.*).
`_get_node_ip_and_free_port`
`_get_node_ip_local`
`_bind_socket_in_range`	Try to bind sock to a random port in [port_range_low, port_range_high).
`_get_free_port_local`	Find a free port, holding it only long enough to learn its number.
`_get_free_consecutive_ports_local`	Find `consecutive` contiguous bindable ports and return the base.
`_reserve_data_plane_ports`	Reserve count distinct ports for the data plane’s driver-side servers.
`init_ray`	Initialise Ray, under a `rl.setup.ray_init` span.
`_init_ray`	Initialise Ray.
`_get_gpu_id_info`	Return (gpu_id, nvlink_domain, topo_rank) for the current worker’s bundle.
`get_reordered_bundle`	Return bundle indices and GPU IDs ordered by physical topology.
`get_ray_cluster_topology`	Query all alive Ray nodes for their NVLink domain and topo_rank.
`select_segment_nodes`	Partition Ray node IDs into segment-aligned selected nodes and remainder.
`prepare_segment_topology`	Compute node resource constraints for topology-aware cluster placement.
`_sort_bundle_indices_by_topology`	Compute topology-aware sort order for bundle indices.

Data#

`logger`
`RAY_CLUSTER_SOURCE_ATTR`
`RAY_CLUSTER_EXTERNAL`
`RAY_CLUSTER_REUSED_LOCAL`
`RAY_CLUSTER_STARTED_LOCAL`
`_WORKER_PROCESS_SETUP_HOOK`
`dir_path`
`git_root`
`DEFAULT_GENERATION_ROUTER_PORT_RANGE_LOW`
`DEFAULT_GENERATION_ROUTER_PORT_RANGE_HIGH`
`DEFAULT_GENERATION_PORT_RANGE_LOW`
`DEFAULT_GENERATION_PORT_RANGE_HIGH`
`DEFAULT_DYNAMO_CONTROL_PORT_RANGE_LOW`
`DEFAULT_DYNAMO_CONTROL_PORT_RANGE_HIGH`
`DEFAULT_DYNAMO_HTTP_PORT_RANGE_LOW`
`DEFAULT_DYNAMO_HTTP_PORT_RANGE_HIGH`
`DEFAULT_DYNAMO_SYSTEM_PORT_RANGE_LOW`
`DEFAULT_DYNAMO_SYSTEM_PORT_RANGE_HIGH`
`DEFAULT_GYM_PORT_RANGE_LOW`
`DEFAULT_GYM_PORT_RANGE_HIGH`
`DEFAULT_VLLM_PORT_RANGE_LOW`
`DEFAULT_VLLM_PORTS_PER_ENGINE`
`DEFAULT_DISAGG_FRONTEND_PORT_RANGE_LOW`
`DEFAULT_DISAGG_FRONTEND_PORT_RANGE_HIGH`
`DEFAULT_SGLANG_ROUTER_PORT_RANGE_LOW`
`DEFAULT_SGLANG_ROUTER_PORT_RANGE_HIGH`
`DEFAULT_SGLANG_PROMETHEUS_PORT_RANGE_LOW`
`DEFAULT_SGLANG_PROMETHEUS_PORT_RANGE_HIGH`
`DEFAULT_MASTER_PORT_RANGE_LOW`
`DEFAULT_MASTER_PORT_RANGE_HIGH`
`DEFAULT_DATA_PLANE_PORT_RANGE_LOW`
`DEFAULT_DATA_PLANE_PORT_RANGE_HIGH`
`NVLINK_DOMAIN_PREFIX`	Ray resource key prefix for the NVLink domain. Each node registers one resource `nvlink_domain_<ClusterUUID>: 1` where ClusterUUID is parsed directly from `nvidia-smi -q` output by ray.sub. Nodes sharing the same key belong to the same NVLink switch fabric (e.g. one GB200 NVL72 rack).
`TOPO_RANK_KEY`	Ray resource key for the SLURM topological rank. Derived from `SLURM_TOPOLOGY_ADDR` (when `SLURM_TOPOLOGY_ADDR_PATTERN=block.node`), falling back to `SLURM_PROCID + 2` on worker nodes (head node is pinned to `1`), then to hostname digits when SLURM is unavailable. Values are always `>= 1` so that Ray does not drop the custom resource (Ray drops value-0 custom resources). Used to sort nodes within and across NVLink domains so rank assignment follows physical topology.
`NVLINK_DOMAIN_UNKNOWN`	Sentinel returned when no NVLink domain info is available for a node.
`TOPO_RANK_UNKNOWN`	Sentinel returned when no topological rank is available for a node.

API#

nemo_rl.distributed.virtual_cluster.logger#: ‘getLogger(…)’

nemo_rl.distributed.virtual_cluster.RAY_CLUSTER_SOURCE_ATTR#: ‘rl.ray.cluster_source’

nemo_rl.distributed.virtual_cluster.RAY_CLUSTER_EXTERNAL#: ‘attached_external’

nemo_rl.distributed.virtual_cluster.RAY_CLUSTER_REUSED_LOCAL#: ‘reused_local’

nemo_rl.distributed.virtual_cluster.RAY_CLUSTER_STARTED_LOCAL#: ‘started_local’

nemo_rl.distributed.virtual_cluster._WORKER_PROCESS_SETUP_HOOK#: ‘nemo_rl.distributed.ray_worker_setup.ray_worker_process_setup’

class nemo_rl.distributed.virtual_cluster.ClusterConfig#

Bases: pydantic.BaseModel

gpus_per_node: int#: None

num_nodes: int#: None

master_port_range_low: Optional[int]#: None

master_port_range_high: Optional[int]#: None

segment_size: Optional[int]#: None

nemo_rl.distributed.virtual_cluster.dir_path#: ‘dirname(…)’

nemo_rl.distributed.virtual_cluster.git_root#: ‘abspath(…)’

class nemo_rl.distributed.virtual_cluster.PY_EXECUTABLES#

Command each Ray actor launches under, one entry per uv extra combination.

Every uv command below is rewritten to SYSTEM when NEMO_RL_PY_EXECUTABLES_SYSTEM is set to 1, so callers never apply that check themselves.

SYSTEM#: None

BASE#: None

VLLM#: None

AUTOMODEL#: None

MCORE#: None

NEMO_GYM#: None

VLLM_GYM#: None

SGLANG#: None

TRTLLM#: None

MODELOPT_VLLM#: None

MODELOPT_AUTOMODEL#: None

MODELOPT_MCORE#: None

classmethod _resolve_system_overrides() → None#: Rewrite every uv command constant to the system executable when the flag is set.

nemo_rl.distributed.virtual_cluster.uv_py_executable(extras: Sequence[str]) → str#

py_executable of a uv-managed venv with the given extras (same shape as PY_EXECUTABLES.*).

Honors NEMO_RL_PY_EXECUTABLES_SYSTEM the same way the PY_EXECUTABLES constants do. The check has to live here as well: _resolve_system_overrides rewrites the class attributes, but this builds a fresh string, so the rewrite cannot reach it.

nemo_rl.distributed.virtual_cluster.DEFAULT_GENERATION_ROUTER_PORT_RANGE_LOW#: 1202

nemo_rl.distributed.virtual_cluster.DEFAULT_GENERATION_ROUTER_PORT_RANGE_HIGH#: 1300

nemo_rl.distributed.virtual_cluster.DEFAULT_GENERATION_PORT_RANGE_LOW#: 3000

nemo_rl.distributed.virtual_cluster.DEFAULT_GENERATION_PORT_RANGE_HIGH#: 4999

nemo_rl.distributed.virtual_cluster.DEFAULT_DYNAMO_CONTROL_PORT_RANGE_LOW#: 1313

nemo_rl.distributed.virtual_cluster.DEFAULT_DYNAMO_CONTROL_PORT_RANGE_HIGH#: 1400

nemo_rl.distributed.virtual_cluster.DEFAULT_DYNAMO_HTTP_PORT_RANGE_LOW#: 3000

nemo_rl.distributed.virtual_cluster.DEFAULT_DYNAMO_HTTP_PORT_RANGE_HIGH#: 4000

nemo_rl.distributed.virtual_cluster.DEFAULT_DYNAMO_SYSTEM_PORT_RANGE_LOW#: 4000

nemo_rl.distributed.virtual_cluster.DEFAULT_DYNAMO_SYSTEM_PORT_RANGE_HIGH#: 4100

nemo_rl.distributed.virtual_cluster.DEFAULT_GYM_PORT_RANGE_LOW#: 5000

nemo_rl.distributed.virtual_cluster.DEFAULT_GYM_PORT_RANGE_HIGH#: 5999

nemo_rl.distributed.virtual_cluster.DEFAULT_VLLM_PORT_RANGE_LOW#: 7000

nemo_rl.distributed.virtual_cluster.DEFAULT_VLLM_PORTS_PER_ENGINE#: 100

nemo_rl.distributed.virtual_cluster.DEFAULT_DISAGG_FRONTEND_PORT_RANGE_LOW#: 8300

nemo_rl.distributed.virtual_cluster.DEFAULT_DISAGG_FRONTEND_PORT_RANGE_HIGH#: 8556

nemo_rl.distributed.virtual_cluster.DEFAULT_SGLANG_ROUTER_PORT_RANGE_LOW#: 8600

nemo_rl.distributed.virtual_cluster.DEFAULT_SGLANG_ROUTER_PORT_RANGE_HIGH#: 8799

nemo_rl.distributed.virtual_cluster.DEFAULT_SGLANG_PROMETHEUS_PORT_RANGE_LOW#: 8800

nemo_rl.distributed.virtual_cluster.DEFAULT_SGLANG_PROMETHEUS_PORT_RANGE_HIGH#: 8999

nemo_rl.distributed.virtual_cluster.DEFAULT_MASTER_PORT_RANGE_LOW#: 1400

nemo_rl.distributed.virtual_cluster.DEFAULT_MASTER_PORT_RANGE_HIGH#: 1999

nemo_rl.distributed.virtual_cluster.DEFAULT_DATA_PLANE_PORT_RANGE_LOW#: 1150

nemo_rl.distributed.virtual_cluster.DEFAULT_DATA_PLANE_PORT_RANGE_HIGH#: 1200

nemo_rl.distributed.virtual_cluster.NVLINK_DOMAIN_PREFIX#

‘nvlink_domain_’

Ray resource key prefix for the NVLink domain. Each node registers one resource nvlink_domain_<ClusterUUID>: 1 where ClusterUUID is parsed directly from nvidia-smi -q output by ray.sub. Nodes sharing the same key belong to the same NVLink switch fabric (e.g. one GB200 NVL72 rack).

nemo_rl.distributed.virtual_cluster.TOPO_RANK_KEY#

‘topo_rank’

Ray resource key for the SLURM topological rank. Derived from SLURM_TOPOLOGY_ADDR (when SLURM_TOPOLOGY_ADDR_PATTERN=block.node), falling back to SLURM_PROCID + 2 on worker nodes (head node is pinned to 1), then to hostname digits when SLURM is unavailable. Values are always >= 1 so that Ray does not drop the custom resource (Ray drops value-0 custom resources). Used to sort nodes within and across NVLink domains so rank assignment follows physical topology.

nemo_rl.distributed.virtual_cluster.NVLINK_DOMAIN_UNKNOWN#

‘unknown’

Sentinel returned when no NVLink domain info is available for a node.

nemo_rl.distributed.virtual_cluster.TOPO_RANK_UNKNOWN: int#

None

Sentinel returned when no topological rank is available for a node.

nemo_rl.distributed.virtual_cluster._get_node_ip_and_free_port( port_range_low: int = DEFAULT_MASTER_PORT_RANGE_LOW, port_range_high: int = DEFAULT_MASTER_PORT_RANGE_HIGH, ) → tuple[str, int]#

nemo_rl.distributed.virtual_cluster._get_node_ip_local() → str#

nemo_rl.distributed.virtual_cluster._bind_socket_in_range( sock: socket.socket, port_range_low: int, port_range_high: int, max_retries: int | None = 50, excluded_ports: set[int] | None = None, rng: Optional[random.Random] = None, ) → int#

Try to bind sock to a random port in [port_range_low, port_range_high).

When max_retries is None, try every non-excluded port once. Otherwise, preserve the existing bounded random-retry behavior.

Parameters:: rng – Source of candidate ports. Defaults to the random module. That module’s state is process-wide and seeded per run, so every rank of a job draws the same sequence; ranks sharing a node then contend for one port, and whichever binds after the others have closed silently reuses it. Pass a rank-seeded random.Random to decorrelate them.

Raises RuntimeError after max_retries failed attempts.

nemo_rl.distributed.virtual_cluster._get_free_port_local( port_range_low: int = DEFAULT_MASTER_PORT_RANGE_LOW, port_range_high: int = DEFAULT_MASTER_PORT_RANGE_HIGH, *, max_retries: int | None = 50, excluded_ports: set[int] | None = None, rng: Optional[random.Random] = None, ) → int#

Find a free port, holding it only long enough to learn its number.

Parameters:: rng – Source of candidate ports; see :func:_bind_socket_in_range. Callers that run on several ranks of one node should pass a rank-seeded generator, or they will all be handed the same port.

nemo_rl.distributed.virtual_cluster._get_free_consecutive_ports_local( port_range_low: int, port_range_high: int, consecutive: int = 1, start_port: Optional[int] = None, ) → int#

Find consecutive contiguous bindable ports and return the base.

Scans upward from start_port within [port_range_low, port_range_high). start_port lets a caller thread a per-node cursor so successive blocks do not overlap. Raises RuntimeError if no such block exists in the range.

nemo_rl.distributed.virtual_cluster._reserve_data_plane_ports(count: int) → list[int]#

Reserve count distinct ports for the data plane’s driver-side servers.

The defaults those servers land on – 50050 and 50051 from TransferQueue’s config.yaml, 9003 from mooncake_master’s metrics_port gflag – sit inside the 9000-65000 ephemeral range these nodes use. The kernel can therefore hand any of them to an outgoing connection during the minutes between job start and the master’s bind, after which the master dies with EADDRINUSE. That is why ray.sub’s port map keeps every service below 9000, and this allocates from the band that map leaves free.

Every one of these ports is passed to the servers explicitly and reaches clients through the TQ controller actor rather than being assumed, so relocating them costs nothing.

excluded_ports keeps the returned ports distinct, and _get_free_port_local binds without SO_REUSEADDR, matching mooncake_master: a port obtainable only by reusing a lingering slot is not one the master could bind either.

Returns:: count ports from the data-plane band, in allocation order.

nemo_rl.distributed.virtual_cluster.init_ray(log_dir: Optional[str] = None) → None#

Initialise Ray, under a rl.setup.ray_init span.

Split from :func:_init_ray only so the span can record which of the three outcomes below happened. Attaching to a cluster someone else is running and booting a fresh one differ by tens of seconds, and that difference is the usual reason two otherwise identical runs disagree on time-to-first-step.

nemo_rl.distributed.virtual_cluster._init_ray(log_dir: Optional[str] = None) → str#

Initialise Ray.

Try to attach to an existing local cluster. If that cluster uses the same CUDA_VISIBLE_DEVICES or Slurm managed tag we will reuse it. Otherwise, we will detach and start a fresh local cluster.

Any process env var that must reach every worker (e.g. a backend engine knob such as data_plane’s) has to be set before this call, in the caller — this function snapshots dict(os.environ) below into runtime_env["env_vars"], which is the only point such a setting becomes cluster-wide. See

Func:: ~nemo_rl.data_plane.factory.maybe_configure_data_plane_env, which a data-plane-enabled launcher calls immediately before this one.
Parameters:: log_dir – Optional directory to store Ray logs and temp files.
Returns:: one of the RAY_CLUSTER_* constants.
Return type:: How the cluster was obtained

nemo_rl.distributed.virtual_cluster._get_gpu_id_info() → tuple[int, str, int]#

Return (gpu_id, nvlink_domain, topo_rank) for the current worker’s bundle.

Reads custom resources set by ray.sub (see NVLINK_DOMAIN_PREFIX / TOPO_RANK_KEY).

nemo_rl.distributed.virtual_cluster.get_reordered_bundle( pg: ray.util.placement_group.PlacementGroup, segment_size: int | None = None, gpus_per_node: int | None = None, ) → tuple[list[int], list[int], tuple[str, ...]]#

Return bundle indices and GPU IDs ordered by physical topology.

Spins up one short-lived task per bundle (_get_gpu_id_info) to discover each bundle’s physical GPU ID, NVLink domain, and topo_rank, then orders bundles via _sort_bundle_indices_by_topology:

No topology info available -> sort by (node_id, gpu_id).
Topology info available -> sort by (domain_min_topo_rank, topo_rank, gpu_id).
segment_size set -> additionally drop bundles from NVLink domains that cannot form a complete segment_size-node segment.

Parameters:

pg – Placement group whose bundles to reorder.
segment_size – Nodes per NVLink domain segment for topology-aware alignment; None disables segment filtering.
gpus_per_node – Required when segment_size is set.

Returns:

(reordered_bundle_indices, reordered_gpu_ids, nvlink_domain_per_bundle_index) where nvlink_domain_per_bundle_index is indexed by original bundle index.

exception nemo_rl.distributed.virtual_cluster.ResourceInsufficientError#

Bases: Exception

Exception raised when the cluster does not have enough resources to satisfy the requested configuration.

Initialization

Initialize self. See help(type(self)) for accurate signature.

nemo_rl.distributed.virtual_cluster.get_ray_cluster_topology() → dict[str, tuple[str, int]]#

Query all alive Ray nodes for their NVLink domain and topo_rank.

Returns:: Dict mapping node_id -> (nvlink_domain, topo_rank). nvlink_domain is NVLINK_DOMAIN_UNKNOWN and topo_rank is TOPO_RANK_UNKNOWN if topology info is unavailable.

nemo_rl.distributed.virtual_cluster.select_segment_nodes( topology: dict[str, tuple[str, int]], segment_size: int, num_nodes: int, ) → tuple[list[str], list[str]]#

Partition Ray node IDs into segment-aligned selected nodes and remainder.

Greedily selects complete segments (segment_size nodes) from each NVLink domain, sorted by topological order, until num_nodes is reached.

Parameters:

topology – Dict mapping node_id -> (nvlink_domain, topo_rank) from get_ray_cluster_topology().
segment_size – Number of nodes per NVLink domain segment.
num_nodes – Total number of nodes to select.

Returns:

Selected nodes are in topological order.

Return type:

(selected_node_ids, remaining_node_ids)

Raises:

ValueError – If segment_size does not evenly divide num_nodes.
ResourceInsufficientError – If not enough complete segments can be formed.

nemo_rl.distributed.virtual_cluster.prepare_segment_topology( segment_size: int | None, num_nodes: int, *, topology: dict[str, tuple[str, int]] | None = None, role: str = 'training', ) → tuple[list[dict[str, float]] | None, list[str], dict[str, tuple[str, int]]]#

Compute node resource constraints for topology-aware cluster placement.

Fetches cluster topology if not provided, selects segment-aligned nodes, and returns per-node domain constraints ready for RayVirtualCluster.

Parameters:

segment_size – Nodes per NVLink domain segment. None disables topology logic.
num_nodes – Number of nodes to select.
topology – Pre-fetched topology dict; fetched automatically when None.
role – Label used in progress messages (e.g. "training", "inference").

Returns:

(node_resource_constraints, remaining_node_ids, topology)

node_resource_constraints: per-node domain-pinning dicts for RayVirtualCluster, or None when segment_size is None or no NVLink domain info is available.
remaining_node_ids: node IDs not selected; pass the corresponding sub-topology to a follow-up call to allocate an inference cluster.
topology: the topology dict used (empty dict when segment_size is None, for safe sub-topology slicing by callers).

nemo_rl.distributed.virtual_cluster._sort_bundle_indices_by_topology( bundle_data: list[tuple[int, str, int, str]], segment_size: int | None = None, gpus_per_node: int | None = None, ) → list[int]#

Compute topology-aware sort order for bundle indices.

When topology information is available: sort by (domain_min_topo_rank, topo_rank, gpu_id). When segment_size is set: additionally validate that each NVLink domain contributes complete segments (segment_size nodes), discarding bundles from incomplete domains. Else: sort by (node_id, gpu_id).

Parameters:

bundle_data – For each bundle i, (gpu_id, nvlink_domain, topo_rank, node_id).
segment_size – If set, number of nodes per NVLink domain segment. Bundles from domains with fewer than segment_size nodes are excluded.
gpus_per_node – Required when segment_size is set. Number of GPUs per node.

Returns:

List of bundle indices in sorted order.

Raises:

ValueError – If segment_size is set but gpus_per_node is not.

class nemo_rl.distributed.virtual_cluster.RayVirtualCluster( bundle_ct_per_node_list: list[int], use_gpus: bool = True, max_colocated_worker_groups: int = 1, num_gpus_per_node: int = 8, name: str = '', placement_group_strategy: str = 'SPREAD', port_range_low: Optional[int] = None, port_range_high: Optional[int] = None, segment_size: int | None = None, node_resource_constraints: list[dict[str, float]] | None = None, )#

Creates a virtual distributed cluster using Ray placement groups.

This class simplifies distributed training setup by:

Creating placement groups that represent logical compute nodes
Allocating GPU and CPU resources for distributed workers
Managing communication between distributed processes
Bundle: A resource allocation unit (ex: 4 GPUs on a single node)
Worker: A process that performs computation (model training/inference)
Node: A physical or virtual machine containing multiple bundles

Initialization

Initialize a virtual cluster using Ray placement groups.

Parameters:

bundle_ct_per_node_list – List specifying GPU bundles per node (e.g., [2,2] creates 2 nodes with 2 GPU bundles each)
use_gpus – Whether to allocate GPU resources
max_colocated_worker_groups – Maximum number of worker groups that can be colocated
num_gpus_per_node – Number of GPUs per node
name – Name prefix for placement groups
placement_group_strategy – Ray placement group strategy (“STRICT_PACK”, “PACK”, or “SPREAD”)
port_range_low – Lower bound (inclusive) of the port range for master address allocation. Falls back to DEFAULT_MASTER_PORT_RANGE_LOW if None.
port_range_high – Upper bound (exclusive) of the port range for master address allocation. Falls back to DEFAULT_MASTER_PORT_RANGE_HIGH if None.
segment_size – Nodes per NVLink domain segment for topology-aware alignment. When set, _sort_bundle_indices_by_topology trims incomplete domain segments.
node_resource_constraints – Per-logical-node extra Ray resource requirements. Length must match bundle_ct_per_node_list. Each dict is merged into every bundle spec for that node, pinning it to a physical domain. Built from NVLink domain resources injected by ray.sub. Example: [{“nvlink_domain_”: 0.001}] * 16 pins 16 nodes to a single NVLink domain.

_init_placement_groups( strategy: str | None = None, use_unified_pg: bool = False, ) → list[ray.util.placement_group.PlacementGroup]#

Creates placement groups based on whether cross-node model parallelism is needed.

Parameters:

strategy – Ray placement group strategy (defaults to self.placement_group_strategy)
use_unified_pg – If True, create a single unified placement group. If False, create per-node placement groups.

Returns:

List of placement groups

_create_placement_groups_internal( strategy: str, use_unified_pg: bool = False, ) → list[ray.util.placement_group.PlacementGroup]#: Internal method to create placement groups without retry logic.

get_placement_groups() → list[ray.util.placement_group.PlacementGroup]#

world_size() → int#

node_count() → int#

get_available_address_and_port( pg_idx: int, bundle_idx: int, ) → tuple[str, int]#

Gets an available address and port for the given placement group index and bundle index.

Returns:: Tuple of (address, port)

get_available_addresses_and_ports_batch( pg_bundle_pairs: list[tuple[int, int]], batch_size: int = 256, ) → list[tuple[str, int]]#

Discovers available addresses and ports for multiple bundles in parallel.

Fires all remote tasks up front, then collects results in batches via ray.wait() to avoid putting too many objects into the Ray object store at once. See https://docs.ray.io/en/latest/ray-core/patterns/ray-get-too-many-objects.html

Parameters:

pg_bundle_pairs – List of (pg_idx, bundle_idx) pairs.
batch_size – Maximum number of ready futures to fetch at once.

Returns:

List of (address, port) tuples in the same order as pg_bundle_pairs.

get_master_address_and_port() → tuple[str, int]#

Gets the master address and port for the distributed training setup.

Each call returns a unique port that has not been returned by previous calls on this cluster instance. This prevents NCCL process-group collisions when multiple worker groups (e.g. policy and value) share the same cluster and node.

Returns:: Tuple of (address, port)

_get_sorted_bundle_indices() → Optional[list[int]]#

Gets the sorted bundle indices for the placement groups.

Returns:: List of bundle indices in sorted order.

get_topology_sorted_pg_indices() → list[int]#

Placement group indices ordered by physical topology.

The unified-PG path sorts bundles (see _get_sorted_bundle_indices), but with per-node placement groups the caller consumes whole PGs, and Ray decides which physical node backs each one. Two PGs that are adjacent in creation order can therefore sit in different NVLink domains, which for a consumer whose unit spans several nodes (a TRT-LLM disaggregated replica, say) silently demotes its inter-node traffic from NVLink to InfiniBand.

Sorting by (nvlink_domain's smallest topo_rank, topo_rank) puts nodes of one domain next to each other, so consecutive PGs belong to the same domain wherever the allocation allows it. Ordering within a domain is not a bandwidth concern (an NVL72 fabric is uniform) but keeps a given config reproducible across runs.

Returns:: Indices into get_placement_groups(). Falls back to the existing order when the cluster is CPU-only, is a single unified PG, or has no topology information – ordering is an optimization, never a requirement.

shutdown() → bool#

Cleans up and releases all resources associated with this virtual cluster.

This includes removing all placement groups and resetting the internal state.

This method is idempotent and can be safely called multiple times.

__del__() → None#

Shutsdown the virtual cluster when the object is deleted or is garbage collected.

This is an extra safety net in case the user forgets to call shutdown and the pointer to the cluster is lost due to leaving a function scope. It’s always recommended that the user calls shutdown().

nemo_rl.distributed.virtual_cluster#

Module Contents#

Classes#

Functions#

Data#

API#

`nemo_rl.distributed.virtual_cluster`#