code: def test_actor_broadcast(ray_start_cluster_with_resource):
      cluster, num_nodes = ray_start_cluster_with_resource
  
      @ray.remote
      class Actor:
          def ready(self):
              pass
  
          def set_weights(self, x):
              pass
  
      actors = [
          Actor._remote(
              args=[],
              kwargs={},
              num_cpus=0.01,
              resources={str(i % num_nodes): 1},
          )
          for i in range(30)
      ]
  
      # Wait for the actors to start up.
      ray.get([a.ready.remote() for a in actors])
  
      object_refs = []
  
      # Broadcast a large object to all actors.
      for _ in range(5):
          x_id = ray.put(np.zeros(1024 * 1024, dtype=np.uint8))
          object_refs.append(x_id)
          # Pass the object into a method for every actor.
          ray.get([a.set_weights.remote(x_id) for a in actors])
  
      # Wait for profiling information to be pushed to the profile table.
      time.sleep(1)
  
imports: import multiprocessing
  import time
  import warnings
  from collections import defaultdict
  import numpy
  import pytest
  import ray
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 5 for calling a global function with example usage: test_actor_broadcast(ray_start_cluster_with_resource) without return types