class_docstr: Group of Ray Actors that can execute arbitrary functions.
  
  ``WorkerGroup`` launches Ray actors according to the given
  specification. It can then execute arbitrary Python functions in each of
  these workers.
  
  If not enough resources are available to launch the actors, the Ray
  cluster will automatically scale up if autoscaling is enabled.
  
  Args:
      num_workers: The number of workers (Ray actors) to launch.
          Defaults to 1.
      num_cpus_per_worker: The number of CPUs to reserve for each
          worker. Fractional values are allowed. Defaults to 1.
      num_gpus_per_worker: The number of GPUs to reserve for each
          worker. Fractional values are allowed. Defaults to 0.
      additional_resources_per_worker (Optional[Dict[str, float]]):
          Dictionary specifying the extra resources that will be
          requested for each worker in addition to ``num_cpus_per_worker``
          and ``num_gpus_per_worker``.
      actor_cls (Optional[Type]): If specified use this class as the
          remote actors.
      remote_cls_args, remote_cls_kwargs: If ``remote_cls`` is provided,
          these args will be used for the worker initialization.
      placement_group (PlacementGroup|str): The placement group that workers
          should be created in. Defaults to "default" which will inherit the
          parent placement group (if child tasks should be captured).
  
  
  Example:
  
  .. code_block:: python
  
      worker_group = WorkerGroup(num_workers=2)
      output = worker_group.execute(lambda: 1)
      assert len(output) == 2
      assert all(o == 1 for o in output)
docstr: Reorder the workers by their node ip and the lowest GPU id.
  
  This is useful for collocating workers on the same node.
  
  Example:
      Given workers with the following attributes:
          worker_0: ip=1, gpu_ids=[1]
          worker_1: ip=0, gpu_ids=[0]
          worker_2: ip=1, gpu_ids=[0]
          worker_3: ip=0, gpu_ids=[1]
  
      The function will perform the following steps:
          1. Group by node IP:
              ip=0: worker_1, worker_3
              ip=1: worker_0, worker_2
  
          2. Sort each group by GPU ID:
              ip=0: worker_1 (gpu_id=0), worker_3 (gpu_id=1)
              ip=1: worker_2 (gpu_id=0), worker_0 (gpu_id=1)
  
      Resulting in the order: [worker_1, worker_3, worker_2, worker_0]
  
  Args:
      _first_ip: The first IP to group by.
          Hack to avoid OOMs.
          This is just a temporary solution for Train loading entire checkpoints
          into memory by ensuring that the rank 0 worker is on the same node as
          trainable, thus allowing for lazy checkpoint transfer to be used.
          See https://github.com/ray-project/ray/issues/33073
          for more context.
          TODO remove this argument.
docstr_tok: ["Reorder","the","workers","by","their","node","ip","and","the","lowest","GPU","id",".","This","is","useful","for","collocating","workers","on","the","same","node",".","Example",":","Given","workers","with","the","following","attributes",":","worker_0",":","ip=1",",","gpu_ids=","[","1","]","worker_1",":","ip=0",",","gpu_ids=","[","0","]","worker_2",":","ip=1",",","gpu_ids=","[","0","]","worker_3",":","ip=0",",","gpu_ids=","[","1","]","The","function","will","perform","the","following","steps",":","1",".","Group","by","node","IP",":","ip=0",":","worker_1",",","worker_3","ip=1",":","worker_0",",","worker_2","2",".","Sort","each","group","by","GPU","ID",":","ip=0",":","worker_1","(","gpu_id=0",")",",","worker_3","(","gpu_id=1",")","ip=1",":","worker_2","(","gpu_id=0",")",",","worker_0","(","gpu_id=1",")","Resulting","in","the","order",":","[","worker_1",",","worker_3",",","worker_2",",","worker_0","]","Args",":","_first_ip",":","The","first","IP","to","group","by",".","Hack","to","avoid","OOMs",".","This","is","just","a","temporary","solution","for","Train","loading","entire","checkpoints","into","memory","by","ensuring","that","the","rank","0","worker","is","on","the","same","node","as","trainable",",","thus","allowing","for","lazy","checkpoint","transfer","to","be","used",".","See","https",":","\/\/github.com\/ray-project\/ray\/issues\/33073","for","more","context",".","TODO","remove","this","argument","."]
code: def sort_workers_by_ip_and_gpu_id(
      self, _first_ip: Optional[str] = None
  ):
      """Reorder the workers by their node ip and the lowest GPU id.
  
      This is useful for collocating workers on the same node.
  
      Example:
          Given workers with the following attributes:
              worker_0: ip=1, gpu_ids=[1]
              worker_1: ip=0, gpu_ids=[0]
              worker_2: ip=1, gpu_ids=[0]
              worker_3: ip=0, gpu_ids=[1]
  
          The function will perform the following steps:
              1. Group by node IP:
                  ip=0: worker_1, worker_3
                  ip=1: worker_0, worker_2
  
              2. Sort each group by GPU ID:
                  ip=0: worker_1 (gpu_id=0), worker_3 (gpu_id=1)
                  ip=1: worker_2 (gpu_id=0), worker_0 (gpu_id=1)
  
          Resulting in the order: [worker_1, worker_3, worker_2, worker_0]
  
      Args:
          _first_ip: The first IP to group by.
              Hack to avoid OOMs.
              This is just a temporary solution for Train loading entire checkpoints
              into memory by ensuring that the rank 0 worker is on the same node as
              trainable, thus allowing for lazy checkpoint transfer to be used.
              See https://github.com/ray-project/ray/issues/33073
              for more context.
              TODO remove this argument.
      """
      ip_to_workers = defaultdict(list)
  
      if _first_ip is not None:
          ip_to_workers[_first_ip] = []
  
      for worker in self.workers:
          ip_to_workers[worker.metadata.node_ip].append(worker)
  
      # Sort workers on the same node by the lowest GPU id
      # More details: https://github.com/ray-project/ray/issues/40803
      def get_lowest_gpu_id(worker) -> int:
          gpu_ids = worker.metadata.resource_ids.get("GPU", [])
          # If there are no GPU IDs, return 0 as a default
          if not gpu_ids:
              return 0
  
          # Attempt to convert GPU IDs to integers and find the minimum ID.
          # Fallback to return the minimum string-based ID
          try:
              return min(int(gpu_id) for gpu_id in gpu_ids)
          except ValueError:
              return min(gpu_ids)
  
      for node_ip in ip_to_workers:
          ip_to_workers[node_ip].sort(key=get_lowest_gpu_id)
  
      sorted_workers = []
      for workers in ip_to_workers.values():
          sorted_workers.extend(workers)
  
      self.workers = sorted_workers
  
imports: import logging
  import os
  import socket
  from collections import defaultdict
  from dataclasses import dataclass
  from typing import Callable, Dict, List, Optional, Tuple, Type, TypeVar, Union
  import ray
  from ray.actor import ActorHandle
  from ray.air._internal.util import exception_cause, skip_exceptions
  from ray.types import ObjectRef
  from ray.util.placement_group import PlacementGroup
label_desc: Use image node_id 10 for calling the WorkerGroup obj's underlying member method code with example usage: obj.sort_workers_by_ip_and_gpu_id(_first_ip) and returns: int, min, min