code: def test_actor_distribution_balance(ray_start_cluster_enabled, args):
      cluster = ray_start_cluster_enabled
  
      node_count = args[0]
      actor_count = args[1]
  
      for i in range(node_count):
          cluster.add_node(
              memory=1024**3,
              _system_config={"gcs_actor_scheduling_enabled": True}
              if i == 0
              else {},
          )
      ray.init(address=cluster.address)
      cluster.wait_for_nodes()
  
      @ray.remote(
          memory=100 * 1024**2,
          num_cpus=0.01,
          scheduling_strategy="SPREAD",
      )
      class Foo:
          def method(self):
              return ray._private.worker.global_worker.node.unique_id
  
      actor_distribution = {}
      actor_list = [Foo.remote() for _ in range(actor_count)]
      for actor in actor_list:
          node_id = ray.get(actor.method.remote())
          if node_id not in actor_distribution.keys():
              actor_distribution[node_id] = []
          actor_distribution[node_id].append(actor)
  
      if node_count >= actor_count:
          assert len(actor_distribution) == actor_count
          for node_id, actors in actor_distribution.items():
              assert len(actors) == 1
      else:
          assert len(actor_distribution) == node_count
          for node_id, actors in actor_distribution.items():
              assert len(actors) <= int(actor_count / node_count)
  
file_path: ray/python/ray/tests/test_advanced_5.py
imports: import logging
  import sys
  import time
  import numpy
  import pytest
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, client_test_enabled
label_desc: Use image node_id 4 for calling a global function with example usage: test_actor_distribution_balance(ray_start_cluster_enabled, args) and returns: ray