Skip to content

fix(rlix-scheduler): more consistent naming GENERATION_CLUSTER_NAME #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
taoluo merged 1 commit into from
Jul 5, 2026
Merged

fix(rlix-scheduler): more consistent naming GENERATION_CLUSTER_NAME #30

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions rlix/scheduler/scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -894,11 +894,11 @@ def _should_background_rebalance_locked(self) -> bool:
continue

pipeline_id, cluster_name = parse_cluster_id(cluster_id)
if cluster_name != "actor_infer":
if cluster_name != GENERATION_CLUSTER_NAME:
continue

infer_cfg = (
self._state.pipeline_registry.get(pipeline_id, {}).get("cluster_configs", {}).get("actor_infer")
self._state.pipeline_registry.get(pipeline_id, {}).get("cluster_configs", {}).get(GENERATION_CLUSTER_NAME)
)
if infer_cfg is None:
continue
Expand Down Expand Up @@ -990,7 +990,7 @@ async def scheduling_cycle(self) -> None:
bundle = alloc.dp_rank_to_gpus.get(dp_rank)
if bundle is None:
pipeline_id, _ = parse_cluster_id(cluster_id)
infer_cfg = self._state.pipeline_registry[pipeline_id]["cluster_configs"]["actor_infer"]
infer_cfg = self._state.pipeline_registry[pipeline_id]["cluster_configs"][GENERATION_CLUSTER_NAME]
tp_size = int(infer_cfg.get("tp_size", 1))
device_mapping = list(infer_cfg.get("device_mapping") or [])
start = dp_rank * tp_size
Expand Down Expand Up @@ -1054,7 +1054,7 @@ async def scheduling_cycle(self) -> None:
continue
tp_size = int(
self._state.pipeline_registry[parse_cluster_id(donor_cid)[0]]["cluster_configs"][
"actor_infer"
GENERATION_CLUSTER_NAME
]["tp_size"]
)
active_ranks = sorted(donor_alloc.active_dp_ranks)
Expand Down Expand Up @@ -1133,7 +1133,7 @@ async def scheduling_cycle(self) -> None:
for pending in pending_gen:
cluster_id = pending.request.cluster_id
pipeline_id, cluster_name = parse_cluster_id(cluster_id)
if cluster_name != "actor_infer":
if cluster_name != GENERATION_CLUSTER_NAME:
continue
# Signal when any dp worker is active (partial allocation is valid).
if not active_dp_workers.get(pipeline_id):
Expand Down Expand Up @@ -1244,9 +1244,9 @@ def _reconstruct_bundle_for_dp_rank(self, *, cluster_id: str, dp_rank: int) -> S
can be reused by _collect_shrink_trace_infos_locked without duplicating logic.
"""
pipeline_id, cluster_name = parse_cluster_id(cluster_id)
if cluster_name != "actor_infer":
if cluster_name != GENERATION_CLUSTER_NAME:
return set()
infer_cfg = self._state.pipeline_registry[pipeline_id]["cluster_configs"]["actor_infer"]
infer_cfg = self._state.pipeline_registry[pipeline_id]["cluster_configs"][GENERATION_CLUSTER_NAME]
tp_size = int(infer_cfg.get("tp_size", 1))
device_mapping = list(infer_cfg.get("device_mapping") or [])
start = dp_rank * tp_size
Expand Down Expand Up @@ -1328,7 +1328,7 @@ def _add_remove(cluster_id: str, dp_ranks: List[int]) -> None:
if not dp_ranks:
return
pipeline_id, cluster_name = parse_cluster_id(cluster_id)
if cluster_name != "actor_infer":
if cluster_name != GENERATION_CLUSTER_NAME:
return
s = pipeline_to_remove.setdefault(pipeline_id, set())
for r in dp_ranks:
Expand All @@ -1338,7 +1338,7 @@ def _add_add(cluster_id: str, dp_ranks: List[int]) -> None:
if not dp_ranks:
return
pipeline_id, cluster_name = parse_cluster_id(cluster_id)
if cluster_name != "actor_infer":
if cluster_name != GENERATION_CLUSTER_NAME:
return
s = pipeline_to_add.setdefault(pipeline_id, set())
for r in dp_ranks:
Expand Down Expand Up @@ -1725,7 +1725,7 @@ async def await_release_gpus(
req = existing
else:
pipeline_id, cluster_name = parse_cluster_id(cluster_id)
if cluster_name != "actor_infer":
if cluster_name != GENERATION_CLUSTER_NAME:
raise RuntimeError(
f"await_release_gpus only supports actor_infer generation clusters, got {cluster_id!r}"
)
Expand Down