code: def test_automatic_cleanup_detached_actors(ray_start_cluster):
      # Make sure the placement groups created by a
      # detached actors are cleaned properly.
      cluster = ray_start_cluster
      num_nodes = 3
      num_cpu_per_node = 2
      # Create 3 nodes cluster.
      for _ in range(num_nodes):
          cluster.add_node(num_cpus=num_cpu_per_node)
      cluster.wait_for_nodes()
  
      info = ray.init(
          address=cluster.address, namespace="default_test_namespace"
      )
      available_cpus = ray.available_resources()["CPU"]
      assert available_cpus == num_nodes * num_cpu_per_node
  
      driver_code = f"""
  import ray
  
  ray.init(address="{info["address"]}", namespace="default_test_namespace")
  
  def create_pg():
      pg = ray.util.placement_group(
              [{{"CPU": 1}} for _ in range(3)],
              strategy="STRICT_SPREAD")
      ray.get(pg.ready())
      return pg
  
  # TODO(sang): Placement groups created by tasks launched by detached actor
  # is not cleaned with the current protocol.
  # @ray.remote(num_cpus=0)
  # def f():
  #     create_pg()
  
  @ray.remote(num_cpus=0, max_restarts=1, max_task_retries=-1)
  class A:
      def create_pg(self):
          create_pg()
      def create_child_pg(self):
          self.a = A.options(name="B").remote()
          ray.get(self.a.create_pg.remote())
      def kill_child_actor(self):
          ray.kill(self.a)
          try:
              ray.get(self.a.create_pg.remote())
          except Exception:
              pass
  
  a = A.options(lifetime="detached", name="A").remote()
  ray.get(a.create_pg.remote())
  # TODO(sang): Currently, child tasks are cleaned when a detached actor
  # is dead. We cannot test this scenario until it is fixed.
  # ray.get(a.create_child_pg.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_num_cpus(expected_num_cpus):
          if expected_num_cpus == 0:
              return "CPU" not in ray.available_resources()
          return ray.available_resources()["CPU"] == expected_num_cpus
  
      wait_for_condition(is_job_done)
      wait_for_condition(lambda: assert_num_cpus(num_nodes))
  
      # Make sure when a child actor spawned by a detached actor
      # is killed, the placement group is removed.
      a = ray.get_actor("A")
      # TODO(sang): child of detached actors
      # seem to be killed when jobs are done. We should fix this before
      # testing this scenario.
      # ray.get(a.kill_child_actor.remote())
      # assert assert_num_cpus(num_nodes)
  
      # Make sure placement groups are cleaned when detached actors are killed.
      ray.kill(a, no_restart=False)
      wait_for_condition(
          lambda: assert_num_cpus(num_nodes * num_cpu_per_node)
      )
      # The detached actor a should've been restarted.
      # Recreate a placement group.
      ray.get(a.create_pg.remote())
      wait_for_condition(lambda: assert_num_cpus(num_nodes))
      # Kill it again and make sure the placement group
      # that is created is deleted again.
      ray.kill(a, no_restart=False)
      wait_for_condition(
          lambda: assert_num_cpus(num_nodes * num_cpu_per_node)
      )
  
imports: import sys
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.cluster_utils
  from ray._private.test_utils import convert_actor_state, generate_system_config_map, get_error_message, get_other_nodes, kill_actor_and_wait_for_failure, placement_group_assert_no_leak, run_string_as_driver, wait_for_condition
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import get_current_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 11 for calling a global function with example usage: test_automatic_cleanup_detached_actors(ray_start_cluster) and returns: False, unknown, unknown, True