docstr: Run distributed training using placement groups.
  
  Each trial requests 4 CPUs and starts 4 remote training workers.
docstr_tok: ["Run","distributed","training","using","placement","groups",".","Each","trial","requests","4","CPUs","and","starts","4","remote","training","workers","."]
code: def testPlacementGroupDistributedTraining(self, reuse_actors=False):
      """Run distributed training using placement groups.
  
      Each trial requests 4 CPUs and starts 4 remote training workers.
      """
  
      head_bundle = {"CPU": 1, "GPU": 0, "custom": 0}
      child_bundle = {"CPU": 1}
  
      placement_group_factory = PlacementGroupFactory(
          [head_bundle, child_bundle, child_bundle, child_bundle]
      )
  
      @ray.remote
      class TrainingActor:
          def train(self, val):
              time.sleep(1)
              return val
  
      def train_fn(config):
          base = config["base"]
          actors = [TrainingActor.remote() for _ in range(4)]
          futures = [
              actor.train.remote(base + 2 * i)
              for i, actor in enumerate(actors)
          ]
          results = ray.get(futures)
  
          end = time.time() - config["start_time"]
          tune.report(avg=np.mean(results), end=end)
  
      trial_executor = RayTrialExecutor(reuse_actors=reuse_actors)
  
      start = time.time()
      out = tune.run(
          train_fn,
          config={
              "start_time": start,
              "base": tune.grid_search(list(range(0, 100, 10))),
          },
          resources_per_trial=placement_group_factory,
          num_samples=1,
          trial_executor=trial_executor,
          reuse_actors=reuse_actors,
          verbose=2,
      )
  
      avgs = sorted(t.last_result["avg"] for t in out.trials)
      self.assertSequenceEqual(avgs, list(range(3, 103, 10)))
  
      trial_end_times = sorted(t.last_result["end"] for t in out.trials)
      print("Trial end times:", trial_end_times)
      max_diff = trial_end_times[-1] - trial_end_times[0]
  
      # Not all trials have been run in parallel
      self.assertGreater(max_diff, 3)
  
      # Some trials should have run in parallel
      # Todo: Re-enable when using buildkite
      # self.assertLess(max_diff, 10)
  
      self._assertCleanup(trial_executor)
  
imports: import sys
  import os
  import time
  import numpy
  import unittest
  import ray
  from ray import tune
  from ray.tune.execution.ray_trial_executor import RayTrialExecutor
  from ray.tune.experiment import Trial
  from ray.tune import Callback
  from ray.tune.execution.trial_runner import TrialRunner
  from ray.tune.execution.placement_groups import PlacementGroupFactory
  from ray.util import placement_group_table
  from ray.cluster_utils import Cluster
  from ray.rllib import _register_all
label_desc: Use image node_id 8 for calling the TrialRunnerPlacementGroupTest obj's underlying member method code with example usage: obj.testPlacementGroupDistributedTraining(reuse_actors) and returns: val