code: def test_fate_sharing(ray_start_cluster, use_actors, node_failure):
      config = {
          "health_check_initial_delay_ms": 0,
          "health_check_period_ms": 100,
          "health_check_failure_threshold": 10,
      }
      cluster = ray_start_cluster
      # Head node with no resources.
      cluster.add_node(num_cpus=0, _system_config=config)
      ray.init(address=cluster.address)
      # Node to place the parent actor.
      node_to_kill = cluster.add_node(
          num_cpus=1, resources={"parent": 1}
      )
      # Node to place the child actor.
      cluster.add_node(num_cpus=1, resources={"child": 1})
      cluster.wait_for_nodes()
  
      @ray.remote
      def sleep():
          time.sleep(1000)
  
      @ray.remote(resources={"child": 1})
      def probe():
          return
  
      # TODO(swang): This test does not pass if max_restarts > 0 for the
      # raylet codepath. Add this parameter once the GCS actor service is enabled
      # by default.
      @ray.remote
      class Actor(object):
          def __init__(self):
              return
  
          def start_child(self, use_actors):
              if use_actors:
                  child = Actor.options(resources={"child": 1}).remote()
                  ray.get(child.sleep.remote())
              else:
                  ray.get(
                      sleep.options(resources={"child": 1}).remote()
                  )
  
          def sleep(self):
              time.sleep(1000)
  
          def get_pid(self):
              return os.getpid()
  
      # Returns whether the "child" resource is available.
      def child_resource_available():
          p = probe.remote()
          ready, _ = ray.wait([p], timeout=1)
          return len(ready) > 0
  
      # Test fate sharing if the parent process dies.
      def test_process_failure(use_actors):
          a = Actor.options(resources={"parent": 1}).remote()
          pid = ray.get(a.get_pid.remote())
          a.start_child.remote(use_actors=use_actors)
          # Wait for the child to be scheduled.
          wait_for_condition(lambda: not child_resource_available())
          # Kill the parent process.
          os.kill(pid, 9)
          wait_for_condition(child_resource_available)
  
      # Test fate sharing if the parent node dies.
      def test_node_failure(node_to_kill, use_actors):
          a = Actor.options(resources={"parent": 1}).remote()
          a.start_child.remote(use_actors=use_actors)
          # Wait for the child to be scheduled.
          wait_for_condition(lambda: not child_resource_available())
          # Kill the parent process.
          cluster.remove_node(node_to_kill, allow_graceful=False)
          node_to_kill = cluster.add_node(
              num_cpus=1, resources={"parent": 1}
          )
          wait_for_condition(child_resource_available)
          return node_to_kill
  
      if node_failure:
          test_node_failure(node_to_kill, use_actors)
      else:
          test_process_failure(use_actors)
  
imports: import os
  import signal
  import sys
  import threading
  import time
  import numpy
  import pytest
  import ray
  import ray._private.ray_constants
  import ray._private.utils
  from ray._private.ray_constants import DEBUG_AUTOSCALING_ERROR
  from ray._private.test_utils import Semaphore, get_error_message, get_log_batch, init_error_pubsub, run_string_as_driver_nonblocking, wait_for_condition
  from ray.cluster_utils import cluster_not_supported
  from ray.experimental.internal_kv import _internal_kv_get
label_desc: Use image node_id 8 for calling a global function with example usage: test_fate_sharing(ray_start_cluster, use_actors, node_failure) and returns: unknown, node_to_kill, os