code: def test_many_placement_groups():
      # @ray.remote(num_cpus=1, resources={"node": 0.02})
      @ray.remote
      class C1:
          def ping(self):
              return "pong"
  
      # @ray.remote(num_cpus=1)
      @ray.remote
      class C2:
          def ping(self):
              return "pong"
  
      # @ray.remote(resources={"node": 0.02})
      @ray.remote
      class C3:
          def ping(self):
              return "pong"
  
      bundle1 = {"node": 0.02, "CPU": 1}
      bundle2 = {"CPU": 1}
      bundle3 = {"node": 0.02}
  
      pgs = []
      for _ in tqdm.trange(MAX_PLACEMENT_GROUPS, desc="Creating pgs"):
          pg = placement_group(bundles=[bundle1, bundle2, bundle3])
          pgs.append(pg)
  
      for pg in tqdm.tqdm(pgs, desc="Waiting for pgs to be ready"):
          ray.get(pg.ready())
  
      actors = []
      for pg in tqdm.tqdm(pgs, desc="Scheduling tasks"):
          actors.append(
              C1.options(
                  scheduling_strategy=PlacementGroupSchedulingStrategy(
                      placement_group=pg
                  )
              ).remote()
          )
          actors.append(
              C2.options(
                  scheduling_strategy=PlacementGroupSchedulingStrategy(
                      placement_group=pg
                  )
              ).remote()
          )
          actors.append(
              C3.options(
                  scheduling_strategy=PlacementGroupSchedulingStrategy(
                      placement_group=pg
                  )
              ).remote()
          )
  
      not_ready = [actor.ping.remote() for actor in actors]
      for _ in tqdm.trange(len(actors)):
          ready, not_ready = ray.wait(not_ready)
          assert ray.get(*ready) == "pong"
  
      for pg in tqdm.tqdm(pgs, desc="Cleaning up pgs"):
          remove_placement_group(pg)
  
file_path: ray/release/benchmarks/distributed/test_many_pgs.py
imports: import os
  import ray
  import ray._private.test_utils
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from dashboard_test import DashboardTestAtScale
  import time
  import tqdm
label_desc: Use image node_id 1 for calling a global function with example usage: test_many_placement_groups() and returns: str, str, str