code: def test_spread_scheduling_strategy(
      ray_start_cluster, connect_to_client
  ):
      cluster = ray_start_cluster
      # Create a head node
      cluster.add_node(
          num_cpus=0,
          _system_config={
              "scheduler_spread_threshold": 1,
          },
      )
      ray.init(address=cluster.address)
      for i in range(2):
          cluster.add_node(num_cpus=8, resources={f"foo:{i}": 1})
      cluster.wait_for_nodes()
  
      with connect_to_client_or_not(connect_to_client):
  
          @ray.remote
          def get_node_id():
              return ray.get_runtime_context().get_node_id()
  
          worker_node_ids = {
              ray.get(
                  get_node_id.options(
                      resources={f"foo:{i}": 1}
                  ).remote()
              )
              for i in range(2)
          }
          # Wait for updating driver raylet's resource view.
          time.sleep(5)
  
          @ray.remote(scheduling_strategy="SPREAD")
          def task1():
              internal_kv._internal_kv_put("test_task1", "task1")
              while internal_kv._internal_kv_exists("test_task1"):
                  time.sleep(0.1)
              return ray.get_runtime_context().get_node_id()
  
          @ray.remote
          def task2():
              internal_kv._internal_kv_put("test_task2", "task2")
              return ray.get_runtime_context().get_node_id()
  
          locations = []
          locations.append(task1.remote())
          while not internal_kv._internal_kv_exists("test_task1"):
              time.sleep(0.1)
          # Wait for updating driver raylet's resource view.
          time.sleep(5)
          locations.append(
              task2.options(scheduling_strategy="SPREAD").remote()
          )
          while not internal_kv._internal_kv_exists("test_task2"):
              time.sleep(0.1)
          internal_kv._internal_kv_del("test_task1")
          internal_kv._internal_kv_del("test_task2")
          assert set(ray.get(locations)) == worker_node_ids
  
          # Wait for updating driver raylet's resource view.
          time.sleep(5)
  
          # Make sure actors can be spreaded as well.
          @ray.remote(num_cpus=1)
          class Actor:
              def ping(self):
                  return ray.get_runtime_context().get_node_id()
  
          actors = []
          locations = []
          for i in range(8):
              actors.append(
                  Actor.options(scheduling_strategy="SPREAD").remote()
              )
              locations.append(ray.get(actors[-1].ping.remote()))
          locations.sort()
          expected_locations = list(worker_node_ids) * 4
          expected_locations.sort()
          assert locations == expected_locations
  
imports: import os
  import platform
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.experimental.internal_kv
  from ray._private.test_utils import make_global_state_accessor, wait_for_condition, get_metric_check_condition, MetricSamplePattern
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, PlacementGroupSchedulingStrategy
  from ray._private.test_utils import SignalActor
label_desc: Use image node_id 8 for calling a global function with example usage: test_spread_scheduling_strategy(ray_start_cluster, connect_to_client) and returns: ray, ray, ray, ray