code: def test_detached_placement_group(ray_start_cluster):
      cluster = ray_start_cluster
      for _ in range(2):
          cluster.add_node(num_cpus=3)
      cluster.wait_for_nodes()
      info = ray.init(address=cluster.address)
  
      # Make sure detached placement group will alive when job dead.
      driver_code = f"""
  import ray
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  
  ray.init(address="{info["address"]}")
  
  pg = ray.util.placement_group(
          [{{"CPU": 1}} for _ in range(2)],
          strategy="STRICT_SPREAD", lifetime="detached")
  ray.get(pg.ready())
  
  @ray.remote(num_cpus=1)
  class Actor:
      def ready(self):
          return True
  
  for bundle_index in range(2):
      actor = Actor.options(lifetime="detached",
          scheduling_strategy=PlacementGroupSchedulingStrategy(placement_group=pg,
                  placement_group_bundle_index=bundle_index)).remote()
      ray.get(actor.ready.remote())
  
  ray.shutdown()
      """
  
      run_string_as_driver(driver_code)
  
      # Wait until the driver is reported as dead by GCS.
      def is_job_done():
          jobs = ray._private.state.jobs()
          for job in jobs:
              if job["IsDead"]:
                  return True
          return False
  
      def assert_alive_num_pg(expected_num_pg):
          alive_num_pg = 0
          for (
              _,
              placement_group_info,
          ) in ray.util.placement_group_table().items():
              if placement_group_info["state"] == "CREATED":
                  alive_num_pg += 1
          return alive_num_pg == expected_num_pg
  
      def assert_alive_num_actor(expected_num_actor):
          alive_num_actor = 0
          for actor_info in ray._private.state.actors().values():
              if actor_info["State"] == convert_actor_state(
                  gcs_utils.ActorTableData.ALIVE
              ):
                  alive_num_actor += 1
          return alive_num_actor == expected_num_actor
  
      wait_for_condition(is_job_done)
  
      assert assert_alive_num_pg(1)
      assert assert_alive_num_actor(2)
  
      # Make sure detached placement group will alive when its creator which
      # is detached actor dead.
      # Test actors first.
      @ray.remote(num_cpus=1)
      class NestedActor:
          def ready(self):
              return True
  
      @ray.remote(num_cpus=1)
      class Actor:
          def __init__(self):
              self.actors = []
  
          def ready(self):
              return True
  
          def schedule_nested_actor_with_detached_pg(self):
              # Create placement group which is detached.
              pg = ray.util.placement_group(
                  [{"CPU": 1} for _ in range(2)],
                  strategy="STRICT_SPREAD",
                  lifetime="detached",
                  name="detached_pg",
              )
              ray.get(pg.ready())
              # Schedule nested actor with the placement group.
              for bundle_index in range(2):
                  actor = NestedActor.options(
                      scheduling_strategy=PlacementGroupSchedulingStrategy(
                          placement_group=pg,
                          placement_group_bundle_index=bundle_index,
                      ),
                      lifetime="detached",
                  ).remote()
                  ray.get(actor.ready.remote())
                  self.actors.append(actor)
  
      a = Actor.options(lifetime="detached").remote()
      ray.get(a.ready.remote())
      # 1 parent actor and 2 children actor.
      ray.get(a.schedule_nested_actor_with_detached_pg.remote())
  
      # Kill an actor and wait until it is killed.
      kill_actor_and_wait_for_failure(a)
      with pytest.raises(ray.exceptions.RayActorError):
          ray.get(a.ready.remote())
  
      # We should have 2 alive pgs and 4 alive actors.
      assert assert_alive_num_pg(2)
      assert assert_alive_num_actor(4)
  
imports: import sys
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.cluster_utils
  import ray.experimental.internal_kv
  from ray._private.ray_constants import DEBUG_AUTOSCALING_ERROR, DEBUG_AUTOSCALING_STATUS
  from ray.autoscaler._private.constants import AUTOSCALER_UPDATE_INTERVAL_S
  from ray._private.test_utils import convert_actor_state, generate_system_config_map, is_placement_group_removed, kill_actor_and_wait_for_failure, reset_autoscaler_v2_enabled_cache, run_string_as_driver, wait_for_condition
  from ray.autoscaler._private.commands import debug_status
  from ray.exceptions import RaySystemError
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 6 for calling a global function with example usage: test_detached_placement_group(ray_start_cluster) and returns: False, unknown, unknown, True, True, True