code: def test_memory_util(config, ray_start_cluster):
      cluster = ray_start_cluster
      # Head node with no resources.
      cluster.add_node(
          num_cpus=0,
          resources={"head": 1},
          _system_config=config,
          enable_object_reconstruction=True,
      )
      ray.init(address=cluster.address)
      # Node to place the initial object.
      node_to_kill = cluster.add_node(
          num_cpus=1,
          resources={"node1": 1},
          object_store_memory=10**8,
      )
      cluster.wait_for_nodes()
  
      @ray.remote
      def large_object(sema=None):
          if sema is not None:
              ray.get(sema.acquire.remote())
          return np.zeros(10**7, dtype=np.uint8)
  
      @ray.remote
      def dependent_task(x, sema):
          ray.get(sema.acquire.remote())
          return x
  
      def stats():
          info = memory_summary(cluster.address, line_wrap=False)
          print(info)
          info = info.split("\n")
          reconstructing_waiting = [
              line
              for line in info
              if "Attempt #2" in line
              and WAITING_FOR_DEPENDENCIES in line
          ]
          reconstructing_scheduled = [
              line
              for line in info
              if "Attempt #2" in line and WAITING_FOR_EXECUTION in line
          ]
          reconstructing_finished = [
              line
              for line in info
              if "Attempt #2" in line and FINISHED in line
          ]
          return (
              len(reconstructing_waiting),
              len(reconstructing_scheduled),
              len(reconstructing_finished),
          )
  
      sema = Semaphore.options(resources={"head": 1}).remote(value=0)
      obj = large_object.options(resources={"node1": 1}).remote(sema)
      x = dependent_task.options(resources={"node1": 1}).remote(
          obj, sema
      )
      ref = dependent_task.options(resources={"node1": 1}).remote(
          x, sema
      )
      ray.get(sema.release.remote())
      ray.get(sema.release.remote())
      ray.get(sema.release.remote())
      ray.get(ref)
      wait_for_condition(lambda: stats() == (0, 0, 0))
      del ref
  
      cluster.remove_node(node_to_kill, allow_graceful=False)
      node_to_kill = cluster.add_node(
          num_cpus=1,
          resources={"node1": 1},
          object_store_memory=10**8,
      )
  
      ref = dependent_task.remote(x, sema)
      wait_for_condition(lambda: stats() == (1, 1, 0))
      ray.get(sema.release.remote())
      wait_for_condition(lambda: stats() == (0, 1, 1))
      ray.get(sema.release.remote())
      ray.get(sema.release.remote())
      ray.get(ref)
      wait_for_condition(lambda: stats() == (0, 0, 2))
  
file_path: ray/python/ray/tests/test_reconstruction_2.py
imports: import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import Semaphore, SignalActor, wait_for_condition
label_desc: Use image node_id 8 for calling a global function with example usage: test_memory_util(config, ray_start_cluster) and returns: np, x, len,  len,  len