code: def test_recursive(ray_start_reconstruction):
      plasma_store_memory, num_nodes, cluster = ray_start_reconstruction
      # Define the size of one task's return argument so that the combined
      # sum of all objects' sizes is at least twice the plasma stores'
      # combined allotted memory.
      num_objects = 100
      size = int(plasma_store_memory * 1.5 / (num_objects * 8))
  
      # Define a root task with no dependencies, which returns a numpy array
      # of the given size.
      @ray.remote
      def no_dependency_task(size):
          array = np.zeros(size)
          return array
  
      # Define a task with a single dependency, which returns its one
      # argument.
      @ray.remote
      def single_dependency(i, arg):
          arg = np.copy(arg)
          arg[0] = i
          return arg
  
      # Launch num_objects instances of the remote task, each dependent on
      # the one before it.
      arg = no_dependency_task.remote(size)
      args = []
      for i in range(num_objects):
          arg = single_dependency.remote(i, arg)
          args.append(arg)
  
      # Get each value to force each task to finish. After some number of
      # gets, old values should be evicted.
      for i in range(num_objects):
          value = ray.get(args[i])
          assert value[0] == i
      # Get each value again to force reconstruction.
      for i in range(num_objects):
          value = ray.get(args[i])
          assert value[0] == i
      # Get 10 values randomly.
      random_indexes = sorted_random_indexes(num_objects, 10)
      for i in random_indexes:
          value = ray.get(args[i])
          assert value[0] == i
      # Get values sequentially, in chunks.
      num_chunks = 4 * num_nodes
      chunk = num_objects // num_chunks
      for i in range(num_chunks):
          values = ray.get(args[i * chunk : (i + 1) * chunk])
          del values
  
      assert cluster.remaining_processes_alive()
  
file_path: ray/python/ray/tests/test_stress_failure.py
imports: import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import get_error_message
  from ray.cluster_utils import Cluster, cluster_not_supported
label_desc: Use image node_id 4 for calling a global function with example usage: test_recursive(ray_start_reconstruction) and returns: array, arg