[train] Add a placement group cleaner for Ray Train

[liulehui]

Description

1. Previously, the placement group lifetime is tied to the Ray job driver, which means if we use Tune + Train V2 or Train V2 with Async validation where validation task creates its own placement group, those placement group owned by non-main job driver will sticks around for the rest of the main job driver.

2. Why did Train v1 + Tune not run into this issue?
   Tune’s driver process kept track of the placement groups spawned for children, including Train. So the Tune driver process was able to remove the placement group after stopping the trial.
   If the Tune driver was launched in a remote task and was killed, you’d run into the same issue as long as the job driver was still alive.

3. To resolve this, we proposed to add a placement group cleaner runs as a detached actor together with Ray Train controller through ControllerCallback and WorkerGroupCallback. This cleaner will monitor the liveness of the controller, and if controller dies without exit gracefully, cleans up the PG this controller spawns.

4. Now the flow will look like below:
   a. after controller start, pg cleaner registered with controller id
   b. after worker group start and pg created, pg cleaner registered with pg
   c. pg cleaner runs the monitor loop, if controller is not alive, try to clean up the pg

Related issues

#54305 #53921

Additional information

1. Repro: https://gist.github.com/liulehui/4b8ddb074f8db338cb5b331bcee0fd09, see logs in the gist comments

2. local test for pg cleaner shutdown:

3. vanilla train run log after this change see below:

2025-11-23 22:20:35,483	INFO worker.py:2014 -- Started a local Ray instance. View the dashboard at 127.0.0.1:8265 
(TrainController pid=94660) A run snapshot was found in storage folder at: '/Users/lehui/ray_results/train_with_failure'
(TrainController pid=94660) This snapshot contains a list of checkpoints reported via `ray.train.report` and will be loaded. This allows the latest checkpoint found in the snapshot to be accessible within your training function via `ray.train.get_checkpoint`.
(TrainController pid=94660) If you meant to start a brand new training job without any information about previous checkpoints found in this directory, please configure a new, unique `RunConfig(name)` or delete the existing folder at '/Users/lehui/ray_results/train_with_failure'.
(TrainController pid=94660) Attempting to start training worker group of size 2 with the following resources: [{'CPU': 1}] * 2
(TrainController pid=94660) Started training worker group of size 2: 
(TrainController pid=94660) - (ip=127.0.0.1, pid=94666) world_rank=0, local_rank=0, node_rank=0
(TrainController pid=94660) - (ip=127.0.0.1, pid=94667) world_rank=1, local_rank=1, node_rank=0
(RayTrainWorker pid=94666) Checkpoint successfully created at: Checkpoint(filesystem=local, path=/Users/lehui/ray_results/train_with_failure/checkpoint_2025-11-23_22-20-38.349831)
(RayTrainWorker pid=94666) Reporting training result 1: TrainingReport(checkpoint=Checkpoint(filesystem=local, path=/Users/lehui/ray_results/train_with_failure/checkpoint_2025-11-23_22-20-38.349831), metrics={'m': 0}, validation_spec=None)
(RayTrainWorker pid=94666) Checkpoint successfully created at: Checkpoint(filesystem=local, path=/Users/lehui/ray_results/train_with_failure/checkpoint_2025-11-23_22-20-38.353886)
(RayTrainWorker pid=94666) Reporting training result 2: TrainingReport(checkpoint=Checkpoint(filesystem=local, path=/Users/lehui/ray_results/train_with_failure/checkpoint_2025-11-23_22-20-38.353886), metrics={'m': 1}, validation_spec=None)

[justinvyu]

Looking good!

[TimothySeah]

Nice, this looks great! It might also be worth testing this by running a bunch of train runs in a workspace and verifying that we don't end up with a bunch of detached actors as pointed out in https://github.com/ray-project/ray/pull/58515/files#r2543599519

[justinvyu]

Can you also copy paste the a sample train run's output in the PR description? Want to make sure our new logs do not spam the driver logs.

[justinvyu]

🚢