code: def test_worker_proc_child_no_leak(shutdown_only):
      """Verify a worker process is not leaked when placement group is removed"""
      ray.init(num_cpus=1)
  
      @ray.remote(num_cpus=1)
      class Actor:
          def __init__(self) -> None:
              self.p = None
  
          def sleep(self):
              mp_context = multiprocessing.get_context("spawn")
              self.p = mp_context.Process(
                  target=time.sleep, args=(1000,)
              )
              self.p.daemon = True
              self.p.start()
  
              print(
                  f"[pid={os.getpid()}ppid={os.getppid()}]sleeping for 1"
              )
              time.sleep(1)
              return (
                  ray.get_runtime_context().get_worker_id(),
                  self.p.pid,
              )
  
      # Create a placement group
      pg = ray.util.placement_group([{"CPU": 1}])
      ray.get(pg.ready())
  
      # Create an actor to a placement group.
      actor = Actor.options(
          scheduling_strategy=PlacementGroupSchedulingStrategy(
              placement_group=pg,
          )
      ).remote()
  
      ray.get(actor.__ray_ready__.remote())
      worker_id, child_pid = ray.get(actor.sleep.remote())
  
      # Remove the placement group
      ray.util.remove_placement_group(pg)
  
      def verify():
          try:
              psutil.Process(pid=child_pid).status()
          except psutil.NoSuchProcess:
              return True
  
      wait_for_condition(verify)
  
      def verify():
          w = get_worker(id=worker_id)
          assert "placement group was removed" in w.exit_detail
          assert w.exit_type == "INTENDED_SYSTEM_EXIT"
          return True
  
      wait_for_condition(verify)
  
imports: import sys
  import time
  import platform
  import os
  import signal
  import multiprocessing
  import pytest
  import ray
  import psutil
  from ray._private.test_utils import wait_for_condition, run_string_as_driver_nonblocking
  from ray.util.state import get_worker, list_tasks
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 14 for calling a global function with example usage: test_worker_proc_child_no_leak(shutdown_only) and returns: True, ray,  self, True