code: def create_colocated_actors(
      actor_specs: Sequence[Tuple[Type, Any, Any, int]],
      node: Optional[str] = "localhost",
      max_attempts: int = 10,
  ) -> Dict[Type, List[ActorHandle]]:
      """Create co-located actors of any type(s) on any node.
  
      Args:
          actor_specs: Tuple/list with tuples consisting of: 1) The
              (already @ray.remote) class(es) to construct, 2) c'tor args,
              3) c'tor kwargs, and 4) the number of actors of that class with
              given args/kwargs to construct.
          node: The node to co-locate the actors on. By default ("localhost"),
              place the actors on the node the caller of this function is
              located on. Use None for indicating that any (resource fulfilling)
              node in the cluster may be used.
          max_attempts: The maximum number of co-location attempts to
              perform before throwing an error.
  
      Returns:
          A dict mapping the created types to the list of n ActorHandles
          created (and co-located) for that type.
      """
      if node == "localhost":
          node = platform.node()
  
      # Maps each entry in `actor_specs` to lists of already co-located actors.
      ok = [[] for _ in range(len(actor_specs))]
  
      # Try n times to co-locate all given actor types (`actor_specs`).
      # With each (failed) attempt, increase the number of actors we try to
      # create (on the same node), then kill the ones that have been created in
      # excess.
      for attempt in range(max_attempts):
          # If any attempt to co-locate fails, set this to False and we'll do
          # another attempt.
          all_good = True
          # Process all `actor_specs` in sequence.
          for i, (typ, args, kwargs, count) in enumerate(actor_specs):
              args = args or []  # Allow None.
              kwargs = kwargs or {}  # Allow None.
              # We don't have enough actors yet of this spec co-located on
              # the desired node.
              if len(ok[i]) < count:
                  co_located = try_create_colocated(
                      cls=typ,
                      args=args,
                      kwargs=kwargs,
                      count=count * (attempt + 1),
                      node=node,
                  )
                  # If node did not matter (None), from here on, use the host
                  # that the first actor(s) are already co-located on.
                  if node is None:
                      node = ray.get(co_located[0].get_host.remote())
                  # Add the newly co-located actors to the `ok` list.
                  ok[i].extend(co_located)
                  # If we still don't have enough -> We'll have to do another
                  # attempt.
                  if len(ok[i]) < count:
                      all_good = False
              # We created too many actors for this spec -> Kill/truncate
              # the excess ones.
              if len(ok[i]) > count:
                  for a in ok[i][count:]:
                      a.__ray_terminate__.remote()
                  ok[i] = ok[i][:count]
  
          # All `actor_specs` have been fulfilled, return lists of
          # co-located actors.
          if all_good:
              return ok
  
      raise Exception(
          "Unable to create enough colocated actors -> aborting."
      )
  
file_path: ray/python/ray/rllib/utils/actors.py
imports: from collections import defaultdict, deque
  import logging
  import platform
  from typing import Any, Dict, List, Optional, Sequence, Tuple, Type
  import ray
  from ray.actor import ActorClass, ActorHandle
label_desc: Use image node_id 1 for calling a global function with example usage: create_colocated_actors(actor_specs, node, max_attempts) and returns: ok