[train] Share PlacementGroupCleaner across Train runs

[OneSizeFitsQuorum]

Description

Ray Train v2 currently creates one detached PlacementGroupCleaner actor per
Train controller. Since cleaners are pinned to the head node, their memory use
grows linearly with concurrent Train runs. In the workload described in #65443,
24 cleaners use approximately 14.4 GB on the head node.

This PR replaces the per-controller actors with one cluster-wide shared
cleaner. The cleaner tracks controllers independently, so a failed controller
only triggers cleanup for its own placement groups.

The shared cleaner:

• uses a fixed, versioned name and internal namespace with get_if_exists=True;
• remains detached, pinned to the head node, and scheduled outside captured
  placement groups;
• persists controller and placement-group registrations in GCS internal KV;
• restores active and incomplete cleanup records after actor restart;
• rejects late placement-group registration after controller death;
• isolates State API and placement-group cleanup failures between controllers;
• scopes State API queries to its own cluster when multiple clusters are active.

Related issues

Fixes #65443.

Testing

PYTHONPATH="$PWD/python" python -m pytest -q \
  python/ray/train/v2/tests/test_placement_group_cleaner.py
# 16 passed

PYTHONPATH="$PWD/python" python -m pytest -q \
  python/ray/train/v2/tests/test_controller_callback_behaviour.py
# 10 passed

pre-commit run --files \
  python/ray/train/v2/_internal/callbacks/placement_group_callback.py \
  python/ray/train/v2/_internal/execution/controller/placement_group_cleaner.py \
  python/ray/train/v2/_internal/state/util.py \
  python/ray/train/v2/tests/test_placement_group_cleaner.py
# Passed

git diff --check origin/master...HEAD
# Passed

A local end-to-end multi-job scenario used two independent drivers, each
running a real Train v2 DataParallelTrainer with a controller, worker, and
placement group. Both runs shared one cleaner. After one driver was killed,
only its placement group was removed; the other run remained active and
completed normally. The cleaner stayed alive throughout.

Contribution notes

No other open PR referencing #65443 or matching this PlacementGroupCleaner
change was found before requesting review.

AI assistance was used for implementation, testing, and self-review of this
change.

[gemini-code-assist]

Code Review

This pull request refactors the PlacementGroupCleaner into a shared detached actor that manages placement group cleanups for multiple Train controllers, persisting state in Ray's internal KV store to handle ungraceful controller exits and cleaner restarts. The review comments correctly identify Python 3.8 compatibility issues due to the use of removeprefix on bytes and str objects in both the implementation and the test suite, suggesting slicing as a backward-compatible alternative.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 0b212e0. Configure here.}

[cursor]

Live PGs cleaned after false death

High Severity

A missing or cleaning controller record is treated as terminal, and a later register_placement_group is kept for deletion. is_actor_alive also treats a missing State API result as dead. Every run has an empty-PG window after after_controller_start (and again between worker-group shutdown and restart). One transient miss can drop the live record; the next PG register is then cleaned by the shared actor and tears down an active run. The old per-controller actor exited instead, so a failed later register left the PG alone.

Additional Locations (2)

• python/ray/train/v2/_internal/execution/controller/placement_group_cleaner.py#L171-L198
• python/ray/train/v2/_internal/state/util.py#L53-L60

 

^{Reviewed by Cursor Bugbot for commit 0b212e0. Configure here.}