code: def _init_torch_distributed(
      init_method: str,
      backend: str,
      rank: int,
      world_size: int,
      local_rank: int,
      local_world_size: int,
      master_addr: str,
      master_port: str,
      gpu_ids: List[int],
      **init_process_group_kwargs,
  ):
      """Initialize torch distributed backend"""
      if init_method == "env":
          os.environ["MASTER_ADDR"] = str(master_addr)
          os.environ["MASTER_PORT"] = str(master_port)
          url = "env://"
      elif init_method == "tcp":
          url = f"tcp://{master_addr}:{master_port}"
      else:
          raise ValueError(
              f"The provided init_method ("
              f"{init_method}) is not supported. Must "
              f"be either 'env' or 'tcp'."
          )
  
      if backend == "nccl":
          # Same as in Ray Train
          os.environ["NCCL_ASYNC_ERROR_HANDLING"] = "1"
          # All workers on a same node should share the same set of
          # visible GPUs. Otherwise they can't talk among themselves.
          os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(
              str(gid) for gid in gpu_ids
          )
          if "NCCL_SOCKET_IFNAME" not in os.environ:
              os.environ[
                  "NCCL_SOCKET_IFNAME"
              ] = DEFAULT_NCCL_SOCKET_IFNAME
  
      init_process_group_kwargs.update(
          dict(
              backend=backend,
              init_method=url,
              rank=rank,
              world_size=world_size,
          )
      )
      init_process_group_kwargs.setdefault(
          "timeout", timedelta(seconds=1800)
      )
  
      dist.init_process_group(**init_process_group_kwargs)
  
      os.environ["RANK"] = str(rank)
      os.environ["LOCAL_RANK"] = str(local_rank)
      os.environ["WORLD_SIZE"] = str(world_size)
      os.environ["LOCAL_WORLD_SIZE"] = str(local_world_size)
  
file_path: ray/python/ray/air/util/torch_dist.py
imports: from abc import ABC
  from collections import defaultdict
  from datetime import timedelta
  import os
  import torch
  import torch.distributed
  from typing import Callable, List, T
  import ray
  from ray.actor import ActorHandle
  from ray.train._internal.utils import get_address_and_port
  from ray.train.constants import DEFAULT_NCCL_SOCKET_IFNAME
  from ray.air._internal.torch_utils import get_device
label_desc: Use image node_id 1 for calling a global function with example usage: _init_torch_distributed(init_method, backend, rank, world_size, local_rank, local_world_size, master_addr, master_port, gpu_ids) without return types