code: def test_nondeterministic_task(
      ray_start_reconstruction, error_pubsub
  ):
      p = error_pubsub
      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 = 1000
      size = plasma_store_memory * 2 // (num_objects * 8)
  
      # Define a nondeterministic remote task with no dependencies, which
      # returns a random numpy array of the given size. This task should
      # produce an error on the driver if it is ever reexecuted.
      @ray.remote
      def foo(i, size):
          array = np.random.rand(size)
          array[0] = i
          return array
  
      # Define a deterministic remote task with no dependencies, which
      # returns a numpy array of zeros of the given size.
      @ray.remote
      def bar(i, size):
          array = np.zeros(size)
          array[0] = i
          return array
  
      # Launch num_objects instances, half deterministic and half
      # nondeterministic.
      args = []
      for i in range(num_objects):
          if i % 2 == 0:
              args.append(foo.remote(i, size))
          else:
              args.append(bar.remote(i, size))
  
      # 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
  
      def error_check(errors):
          if num_nodes == 1:
              # In a single-node setting, each object is evicted and
              # restarted exactly once, so exactly half the objects will
              # produce an error during reconstruction.
              min_errors = num_objects // 2
          else:
              # In a multinode setting, each object is evicted zero or one
              # times, so some of the nondeterministic tasks may not be
              # reexecuted.
              min_errors = 1
          return len(errors) >= min_errors
  
      errors = wait_for_errors(p, error_check)
      # Make sure all the errors have the correct type.
      assert all(
          error.type == ray_constants.HASH_MISMATCH_PUSH_ERROR
          for error in errors
      )
  
      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 7 for calling a global function with example usage: test_nondeterministic_task(ray_start_reconstruction, error_pubsub) and returns: array, array, unknown