code: def test_evict_secondary_copies_before_spill(
      ray_start_cluster, object_spilling_config
  ):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1, object_store_memory=10**8)
      ray.init(address=cluster.address)
      for _ in range(3):
          cluster.add_node(num_cpus=1, object_store_memory=10**8)
      wait_for_condition(lambda: ray.cluster_resources()["CPU"] >= 4)
  
      # Spread data onto all nodes.
      @ray.remote
      def gen():
          time.sleep(0.5)
          return np.ones(10 * 1024 * 1024, dtype=np.uint8)
  
      refs = [
          gen.options(scheduling_strategy="SPREAD").remote()
          for _ in range(16)
      ]  # 160MiB
  
      # Iterate over the data on the worker nodes from the head node.
      for i in range(10):
          for j, r in enumerate(refs):
              print("Iteration", i, j)
              ray.get(r)
  
      _check_spilled()
  
imports: import json
  import platform
  import random
  import re
  import shutil
  import sys
  import time
  import zlib
  from collections import defaultdict
  import numpy
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.tests.test_object_spilling import assert_no_thrashing, is_dir_empty
label_desc: Use image node_id 13 for calling a global function with example usage: test_evict_secondary_copies_before_spill(ray_start_cluster, object_spilling_config) and returns: np