code: def test_many_objects(num_objects, num_actors):
      if num_objects == 0:
          logger.info("Skipping test with no objects")
          return
  
      pg = placement_group([{"CPU": 1}] * num_actors, strategy="SPREAD")
      ray.get(pg.ready())
  
      # We will try to put actors on multiple nodes.
      @ray.remote
      class ObjectActor:
          def __init__(self):
              self.objs = []
  
          def create_objs(self, num_objects):
              import os
  
              for i in range(num_objects):
                  # Object size shouldn't matter here.
                  self.objs.append(ray.put(bytearray(os.urandom(1024))))
                  if i + 1 % 100 == 0:
                      logger.info(f"Created object {i+1}...")
  
              return self.objs
  
          def ready(self):
              pass
  
      actors = [
          ObjectActor.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg,
              )
          ).remote()
          for _ in tqdm.trange(num_actors, desc="Creating actors...")
      ]
  
      waiting_actors = [actor.ready.remote() for actor in actors]
      for _ in tqdm.trange(
          len(actors), desc="Waiting actors to be ready..."
      ):
          _ready, waiting_actors = ray.wait(waiting_actors)
  
      # Splitting objects to multiple actors for creation,
      # credit: https://stackoverflow.com/a/2135920
      def _split(a, n):
          k, m = divmod(len(a), n)
          return (
              a[i * k + min(i, m) : (i + 1) * k + min(i + 1, m)]
              for i in range(n)
          )
  
      num_objs_per_actor = [
          len(objs) for objs in _split(range(num_objects), num_actors)
      ]
  
      waiting_actors = [
          actor.create_objs.remote(num_objs)
          for actor, num_objs in zip(actors, num_objs_per_actor)
      ]
  
      total_objs_created = 0
      for _ in tqdm.trange(
          num_actors, desc="Waiting actors to create objects..."
      ):
          objs, waiting_actors = ray.wait(waiting_actors)
          total_objs_created += len(ray.get(*objs))
  
      assert (
          total_objs_created == num_objects
      ), "Expect correct number of objects created."
  
      invoke_state_api_n(
          lambda res: len(res) == num_objects,
          list_objects,
          filters=[
              ("reference_type", "=", "LOCAL_REFERENCE"),
              ("type", "=", "WORKER"),
          ],
          key_suffix=f"{num_objects}",
          limit=STATE_LIST_LIMIT,
      )
  
      del actors
      remove_placement_group(pg)
  
imports: import click
  import json
  import ray
  from ray._private.ray_constants import LOG_PREFIX_ACTOR_NAME, LOG_PREFIX_JOB_ID
  from ray._private.state_api_test_utils import STATE_LIST_LIMIT, StateAPIMetric, aggregate_perf_results, invoke_state_api, GLOBAL_STATE_STATS
  import ray._private.test_utils
  import tqdm
  import time
  import os
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray.util.state import get_log, list_actors, list_objects, list_tasks
  import logging
label_desc: Use image node_id 4 for calling a global function with example usage: test_many_objects(num_objects, num_actors) and returns: unknown, self