code: def test_actor_load_balancing(ray_start_cluster):
      cluster = ray_start_cluster
      num_nodes = 3
      for i in range(num_nodes):
          cluster.add_node(num_cpus=1)
      ray.init(address=cluster.address)
  
      @ray.remote
      class Actor1:
          def __init__(self):
              pass
  
          def get_location(self):
              return ray._private.worker.global_worker.node.unique_id
  
      # Create a bunch of actors.
      num_actors = 30
      num_attempts = 20
      minimum_count = 5
  
      # Make sure that actors are spread between the raylets.
      attempts = 0
      while attempts < num_attempts:
          actors = [Actor1.remote() for _ in range(num_actors)]
          locations = ray.get(
              [actor.get_location.remote() for actor in actors]
          )
          names = set(locations)
          counts = [locations.count(name) for name in names]
          print("Counts are {}.".format(counts))
          if len(names) == num_nodes and all(
              count >= minimum_count for count in counts
          ):
              break
          attempts += 1
      assert attempts < num_attempts
  
      # Make sure we can get the results of a bunch of tasks.
      results = []
      for _ in range(1000):
          index = np.random.randint(num_actors)
          results.append(actors[index].get_location.remote())
      ray.get(results)
  
imports: import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  from ray.util.state import list_actors
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, convert_actor_state, kill_actor_and_wait_for_failure, make_global_state_accessor, run_string_as_driver, wait_for_condition, wait_for_pid_to_exit
  from ray._private.ray_constants import gcs_actor_scheduling_enabled
  from ray.experimental.internal_kv import _internal_kv_get, _internal_kv_put
label_desc: Use image node_id 3 for calling a global function with example usage: test_actor_load_balancing(ray_start_cluster) and returns: ray