code: def test_legacy_spillback_distribution(ray_start_cluster):
      cluster = ray_start_cluster
      # Create a head node and wait until it is up.
      cluster.add_node(
          num_cpus=0,
          _system_config={
              "scheduler_spread_threshold": 0,
          },
      )
      ray.init(address=cluster.address)
      cluster.wait_for_nodes()
  
      num_nodes = 2
      # create 2 worker nodes.
      for _ in range(num_nodes):
          cluster.add_node(num_cpus=8)
      cluster.wait_for_nodes()
  
      assert ray.cluster_resources()["CPU"] == 16
  
      @ray.remote
      def task():
          time.sleep(1)
          return ray._private.worker.global_worker.current_node_id
  
      # Make sure tasks are spilled back non-deterministically.
      locations = ray.get([task.remote() for _ in range(8)])
      counter = collections.Counter(locations)
      spread = max(counter.values()) - min(counter.values())
      # Ideally we'd want 4 tasks to go to each node, but we'll settle for
      # anything better than a 1-7 split since randomness is noisy.
      assert spread < 7
      assert len(counter) > 1
  
      @ray.remote(num_cpus=1)
      class Actor1:
          def __init__(self):
              pass
  
          def get_location(self):
              return ray._private.worker.global_worker.current_node_id
  
      actors = [Actor1.remote() for _ in range(10)]
      locations = ray.get(
          [actor.get_location.remote() for actor in actors]
      )
      counter = collections.Counter(locations)
      spread = max(counter.values()) - min(counter.values())
      assert spread < 7
      assert len(counter) > 1
  
imports: import collections
  import logging
  import platform
  import subprocess
  import sys
  import time
  import unittest
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  import ray.util.accelerators
  from ray._private.internal_api import memory_summary
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray._private.test_utils import Semaphore, SignalActor, object_memory_usage, get_metric_check_condition, wait_for_condition, MetricSamplePattern
label_desc: Use image node_id 4 for calling a global function with example usage: test_legacy_spillback_distribution(ray_start_cluster) and returns: ray, ray