docstr: In this test we try to start 10 trials but only have resources
  for 2. Placement groups should still be created and PENDING.
  
  Eventually they should be scheduled sequentially (i.e. in pairs
  of two).
docstr_tok: ["In","this","test","we","try","to","start","10","trials","but","only","have","resources","for","2",".","Placement","groups","should","still","be","created","and","PENDING",".","Eventually","they","should","be","scheduled","sequentially","(","i.e",".","in","pairs","of","two",")","."]
code: def testPlacementGroupRequests(
      self, reuse_actors=False, scheduled=10
  ):
      """In this test we try to start 10 trials but only have resources
      for 2. Placement groups should still be created and PENDING.
  
      Eventually they should be scheduled sequentially (i.e. in pairs
      of two)."""
      # Since we check per-step placement groups, set the reconcilation
      # interval to 0
      os.environ["TUNE_PLACEMENT_GROUP_RECON_INTERVAL"] = "0"
  
      def train_fn(config):
          time.sleep(1)
          now = time.time()
          tune.report(end=now - config["start_time"])
  
      head_bundle = {"CPU": 4, "GPU": 0, "custom": 0}
      child_bundle = {"custom": 1}
      # Manually calculated number of parallel trials
      max_num_parallel = 2
  
      placement_group_factory = PlacementGroupFactory(
          [head_bundle, child_bundle, child_bundle]
      )
  
      trial_executor = RayTrialExecutor(reuse_actors=reuse_actors)
      trial_executor.setup(max_pending_trials=max_num_parallel)
  
      this = self
  
      class _TestCallback(Callback):
          def on_step_end(self, iteration, trials, **info):
              num_finished = len(
                  [
                      t
                      for t in trials
                      if t.status == Trial.TERMINATED
                      or t.status == Trial.ERROR
                  ]
              )
  
              resource_manager = trial_executor._resource_manager
  
              num_staging = sum(
                  len(s)
                  for s in resource_manager._request_to_staged_pgs.values()
              )
              num_ready = sum(
                  len(s)
                  for s in resource_manager._request_to_ready_pgs.values()
              )
              num_in_use = len(resource_manager._acquired_pgs)
              num_cached = (
                  trial_executor._actor_cache.num_cached_objects
              )
  
              total_num_tracked = (
                  num_staging + num_ready + num_in_use + num_cached
              )
  
              # All trials should be scheduled
              this.assertEqual(
                  scheduled,
                  min(scheduled, len(trials)),
                  msg=f"Num trials iter {iteration}",
              )
  
              # The following two tests were relaxed for reuse_actors=True
              # so that up to `max_num_parallel` more placement groups can
              # exist than we would expect. This is because caching
              # relies on reconciliation for cleanup to avoid overscheduling
              # of new placement groups.
              num_parallel_reuse = int(reuse_actors) * max_num_parallel
  
              # The number of PGs should decrease when trials finish
              # We allow a constant excess of 1 here because the trial will
              # be TERMINATED and the resources only returned after the trainable
              # cleanup future succeeded. Because num_finished will increase,
              # this still asserts that the number of PGs goes down over time.
              this.assertGreaterEqual(
                  max(scheduled, len(trials))
                  - num_finished
                  + 1
                  + num_parallel_reuse,
                  total_num_tracked,
                  msg=f"Num tracked iter {iteration}, {len(trials)}, "
                  f"{scheduled}, {num_finished}, {num_parallel_reuse}",
              )
  
      start = time.time()
      out = tune.run(
          train_fn,
          config={"start_time": start},
          resources_per_trial=placement_group_factory,
          num_samples=10,
          trial_executor=trial_executor,
          callbacks=[_TestCallback()],
          reuse_actors=reuse_actors,
          verbose=2,
      )
  
      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 4 for calling the TrialRunnerPlacementGroupTest obj's underlying member method code with example usage: obj.testPlacementGroupRequests(reuse_actors, scheduled) without return types