class_docstr: A base class for creating distributed data loading routines.
  
  Args:
      current_ctx (DistContext): Distributed context info of the current
          process.
      master_addr (str, optional): RPC address for distributed loader
          communication.
          Refers to the IP address of the master node. (default: :obj:`None`)
      master_port (int or str, optional): The open port for RPC communication
          with the master node. (default: :obj:`None`)
      channel (mp.Queue, optional): A communication channel for messages.
          (default: :obj:`None`)
      num_rpc_threads (int, optional): The number of threads in the
          thread-pool used by
          :class:`~torch.distributed.rpc.TensorPipeAgent` to execute
          requests. (default: :obj:`16`)
      rpc_timeout (int, optional): The default timeout in seconds for RPC
          requests.
          If the RPC has not completed in this timeframe, an exception will
          be raised.
          Callers can override this timeout for
          individual RPCs in :meth:`~torch.distributed.rpc.rpc_sync` and
          :meth:`~torch.distributed.rpc.rpc_async` if necessary.
          (default: :obj:`180`)
class_docstr_tok: ["A","base","class","for","creating","distributed","data","loading","routines",".","Args",":","current_ctx","(","DistContext",")",":","Distributed","context","info","of","the","current","process",".","master_addr","(","str",",","optional",")",":","RPC","address","for","distributed","loader","communication",".","Refers","to","the","IP","address","of","the","master","node",".","(","default",":",":","obj",":","`","None","`",")","master_port","(","int","or","str",",","optional",")",":","The","open","port","for","RPC","communication","with","the","master","node",".","(","default",":",":","obj",":","`","None","`",")","channel","(","mp.Queue",",","optional",")",":","A","communication","channel","for","messages",".","(","default",":",":","obj",":","`","None","`",")","num_rpc_threads","(","int",",","optional",")",":","The","number","of","threads","in","the","thread-pool","used","by",":","class",":","`","~torch.distributed.rpc.TensorPipeAgent","`","to","execute","requests",".","(","default",":",":","obj",":","`","16","`",")","rpc_timeout","(","int",",","optional",")",":","The","default","timeout","in","seconds","for","RPC","requests",".","If","the","RPC","has","not","completed","in","this","timeframe",",","an","exception","will","be","raised",".","Callers","can","override","this","timeout","for","individual","RPCs","in",":","meth",":","`","~torch.distributed.rpc.rpc_sync","`","and",":","meth",":","`","~torch.distributed.rpc.rpc_async","`","if","necessary",".","(","default",":",":","obj",":","`","180","`",")"]
code: def __init__(
      self,
      current_ctx: DistContext,
      master_addr: Optional[str] = None,
      master_port: Optional[Union[int, str]] = None,
      channel: Optional[mp.Queue] = None,
      num_rpc_threads: int = 16,
      rpc_timeout: int = 180,
      dist_sampler: DistNeighborSampler = None,
      **kwargs,
  ):
      if (
          master_addr is None
          and os.environ.get("MASTER_ADDR") is not None
      ):
          master_addr = os.environ["MASTER_ADDR"]
      if master_addr is None:
          raise ValueError(
              f"Missing master address for RPC communication "
              f"in '{self.__class__.__name__}'. Try to provide "
              f"it or set it via the 'MASTER_ADDR' environment "
              f"variable."
          )
  
      if (
          master_port is None
          and os.environ.get("MASTER_PORT") is not None
      ):
          # Select next port to MASTER_PORT used for DDP.
          # If multiple loaders are launched in the same script, please
          # provide distinct ports for each.
          master_port = int(os.environ["MASTER_PORT"]) + 1
      if master_port is None:
          raise ValueError(
              f"Missing master port for RPC communication in "
              f"'{self.__class__.__name__}'. Try to provide it "
              f"or set it via the 'MASTER_ADDR' environment "
              f"variable."
          )
  
      assert num_rpc_threads > 0
      assert rpc_timeout > 0
  
      self.dist_sampler = dist_sampler
      self.current_ctx = current_ctx
      self.master_addr = master_addr
      self.master_port = master_port
      self.channel = channel
      self.pid = mp.current_process().pid
      self.num_rpc_threads = num_rpc_threads
      self.rpc_timeout = rpc_timeout
      self.num_workers = kwargs.get("num_workers", 0)
  
      logging.info(
          f"[{self}] MASTER_ADDR={master_addr}, "
          f"MASTER_PORT={master_port}"
      )
  
      if self.num_workers == 0:  # Initialize RPC in main process:
          self.worker_init_fn(0)
  
file_path: pytorch_geometric/torch_geometric/distributed/dist_loader.py
imports: import atexit
  import logging
  import os
  from typing import Any, Optional, Union
  import torch.multiprocessing
  from torch_geometric.distributed import DistNeighborSampler
  from torch_geometric.distributed.dist_context import DistContext
  from torch_geometric.distributed.rpc import global_barrier, init_rpc, shutdown_rpc
  from torch_geometric.loader.base import DataLoaderIterator
label_desc: Use image node_id 1 to create a new DistLoader object  with example: obj = DistLoader(current_ctx, master_addr, master_port, channel, num_rpc_threads, rpc_timeout, dist_sampler)