code: def test_multiple_actor_restart(ray_start_cluster_head):
      cluster = ray_start_cluster_head
      # This test can be made more stressful by increasing the numbers below.
      # The total number of actors created will be
      # num_actors_at_a_time * num_nodes.
      num_nodes = 5
      num_actors_at_a_time = 3
      num_function_calls_at_a_time = 10
  
      worker_nodes = [
          cluster.add_node(num_cpus=3) for _ in range(num_nodes)
      ]
  
      @ray.remote(max_restarts=-1, max_task_retries=-1)
      class SlowCounter:
          def __init__(self):
              self.x = 0
  
          def inc(self, duration):
              time.sleep(duration)
              self.x += 1
              return self.x
  
      # Create some initial actors.
      actors = [
          SlowCounter.remote() for _ in range(num_actors_at_a_time)
      ]
  
      # Wait for the actors to start up.
      time.sleep(1)
  
      # This is a mapping from actor handles to object refs returned by
      # methods on that actor.
      result_ids = collections.defaultdict(lambda: [])
  
      # In a loop we are going to create some actors, run some methods, kill
      # a raylet, and run some more methods.
      for node in worker_nodes:
          # Create some actors.
          actors.extend(
              [
                  SlowCounter.remote()
                  for _ in range(num_actors_at_a_time)
              ]
          )
          # Run some methods.
          for j in range(len(actors)):
              actor = actors[j]
              for _ in range(num_function_calls_at_a_time):
                  result_ids[actor].append(
                      actor.inc.remote(j**2 * 0.000001)
                  )
          # Kill a node.
          cluster.remove_node(node)
  
          # Run some more methods.
          for j in range(len(actors)):
              actor = actors[j]
              for _ in range(num_function_calls_at_a_time):
                  result_ids[actor].append(
                      actor.inc.remote(j**2 * 0.000001)
                  )
  
      # Get the results and check that they have the correct values.
      for _, result_id_list in result_ids.items():
          results = ray.get(result_id_list)
          for i, result in enumerate(results):
              if i == 0:
                  assert result == 1
              else:
                  assert result == results[i - 1] + 1 or result == 1
  
imports: import atexit
  import asyncio
  import collections
  import numpy
  import os
  import pytest
  import signal
  import sys
  import time
  import ray
  from ray.actor import exit_actor
  import ray.cluster_utils
  from ray._private.test_utils import wait_for_condition, wait_for_pid_to_exit, generate_system_config_map, SignalActor
label_desc: Use image node_id 9 for calling a global function with example usage: test_multiple_actor_restart(ray_start_cluster_head) and returns: self