code: def test_driver_put_errors(
      ray_start_object_store_memory, error_pubsub
  ):
      p = error_pubsub
      plasma_store_memory = ray_start_object_store_memory
      # 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 = plasma_store_memory * 2 // (num_objects * 8)
  
      # Define a task with a single dependency, a numpy array, that returns
      # another array.
      @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. The first instance of the task takes a numpy array
      # as an argument, which is put into the object store.
      args = []
      arg = single_dependency.remote(0, np.zeros(size))
      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 starting from the beginning to force reconstruction.
      # Currently, since we're not able to reconstruct `ray.put` objects that
      # were evicted and whose originating tasks are still running, this
      # for-loop should hang on its first iteration and push an error to the
      # driver.
      ray.wait([args[0]], timeout=30)
  
      def error_check(errors):
          return len(errors) > 1
  
      errors = wait_for_errors(p, error_check)
      assert all(
          error.type == ray_constants.PUT_RECONSTRUCTION_PUSH_ERROR
          or "ray.exceptions.ObjectLostError" in error.error_messages
          for error in errors
      )
  
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 8 for calling a global function with example usage: test_driver_put_errors(ray_start_object_store_memory, error_pubsub) and returns: arg, unknown