code: def test_reconstruction_suppression(ray_start_cluster_head):
      cluster = ray_start_cluster_head
      num_nodes = 5
      worker_nodes = [cluster.add_node() for _ in range(num_nodes)]
  
      @ray.remote(max_restarts=1)
      class Counter:
          def __init__(self):
              self.x = 0
  
          def inc(self):
              self.x += 1
              return self.x
  
      @ray.remote
      def inc(actor_handle):
          return ray.get(actor_handle.inc.remote())
  
      # Make sure all of the actors have started.
      actors = [Counter.remote() for _ in range(10)]
      ray.get([actor.inc.remote() for actor in actors])
  
      # Kill a node.
      cluster.remove_node(worker_nodes[0])
  
      # Submit several tasks per actor. These should be randomly scheduled to the
      # nodes, so that multiple nodes will detect and try to reconstruct the
      # actor that died, but only one should succeed.
      results = []
      for _ in range(10):
          results += [inc.remote(actor) for actor in actors]
      # Make sure that we can get the results from the restarted actor.
      results = ray.get(results)
  
imports: import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  from ray.util.state import list_actors
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, convert_actor_state, kill_actor_and_wait_for_failure, make_global_state_accessor, run_string_as_driver, wait_for_condition, wait_for_pid_to_exit
  from ray._private.ray_constants import gcs_actor_scheduling_enabled
  from ray.experimental.internal_kv import _internal_kv_get, _internal_kv_put
label_desc: Use image node_id 8 for calling a global function with example usage: test_reconstruction_suppression(ray_start_cluster_head) and returns: ray, self