code: def test_actor_creation_node_failure(ray_start_cluster):
      # TODO(swang): Refactor test_raylet_failed, etc to reuse the below code.
      cluster = ray_start_cluster
  
      @ray.remote
      class Child:
          def __init__(self, death_probability):
              self.death_probability = death_probability
  
          def get_probability(self):
              return self.death_probability
  
          def ping(self):
              # Exit process with some probability.
              exit_chance = np.random.rand()
              if exit_chance < self.death_probability:
                  sys.exit(-1)
  
      num_children = 25
      # Children actors will die about half the time.
      death_probability = 0.5
  
      children = [
          Child.remote(death_probability) for _ in range(num_children)
      ]
      while len(cluster.list_all_nodes()) > 1:
          for j in range(2):
              # Submit some tasks on the actors. About half of the actors will
              # fail.
              children_out = [child.ping.remote() for child in children]
              # Wait a while for all the tasks to complete. This should trigger
              # reconstruction for any actor creation tasks that were forwarded
              # to nodes that then failed.
              ready, _ = ray.wait(
                  children_out,
                  num_returns=len(children_out),
                  timeout=5 * 60.0,
              )
              assert len(ready) == len(children_out)
  
              # Replace any actors that died.
              for i, out in enumerate(children_out):
                  try:
                      ray.get(out)
                  except ray.exceptions.RayActorError:
                      children[i] = Child.remote(death_probability)
  
              children_out = [
                  child.get_probability.remote() for child in children
              ]
              # Wait for new created actors to finish creation before
              # removing a node. This is needed because right now we don't
              # support reconstructing actors that died in the process of
              # being created.
              ready, _ = ray.wait(
                  children_out,
                  num_returns=len(children_out),
                  timeout=5 * 60.0,
              )
              assert len(ready) == len(children_out)
  
          # Remove a node. Any actor creation tasks that were forwarded to this
          # node must be restarted.
          cluster.remove_node(get_other_nodes(cluster, True)[-1])
  
file_path: ray/python/ray/tests/test_component_failures_3.py
imports: import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import get_other_nodes
label_desc: Use image node_id 1 for calling a global function with example usage: test_actor_creation_node_failure(ray_start_cluster) and returns: self