class_docstr: Multiprocessing DataLoader.
  
  Iterates over the data pipeline with everything before feature fetching
  (i.e. :class:`dgl.graphbolt.FeatureFetcher`) in subprocesses, and
  everything after feature fetching in the main process. The datapipe
  is modified in-place as a result.
  
  Only works on single GPU.
  
  Parameters
  ----------
  datapipe : DataPipe
      The data pipeline.
  num_workers : int, optional
      Number of worker processes. Default is 0.
  persistent_workers : bool, optional
      If True, the data loader will not shut down the worker processes after a
      dataset has been consumed once. This allows to maintain the workers
      instances alive.
class_docstr_tok: ["Multiprocessing","DataLoader",".","Iterates","over","the","data","pipeline","with","everything","before","feature","fetching","(","i.e",".",":","class",":","`","dgl.graphbolt.FeatureFetcher","`",")","in","subprocesses",",","and","everything","after","feature","fetching","in","the","main","process",".","The","datapipe","is","modified","in-place","as","a","result",".","Only","works","on","single","GPU",".","Parameters","--","--","--","--","--","datapipe",":","DataPipe","The","data","pipeline",".","num_workers",":","int",",","optional","Number","of","worker","processes",".","Default","is","0.","persistent_workers",":","bool",",","optional","If","True",",","the","data","loader","will","not","shut","down","the","worker","processes","after","a","dataset","has","been","consumed","once",".","This","allows","to","maintain","the","workers","instances","alive","."]
code: def __init__(self, datapipe, num_workers=0, persistent_workers=True):
      # Multiprocessing requires two modifications to the datapipe:
      #
      # 1. Insert a stage after ItemSampler to distribute the
      #    minibatches evenly across processes.
      # 2. Cut the datapipe at FeatureFetcher, and wrap the inner datapipe
      #    of the FeatureFetcher with a multiprocessing PyTorch DataLoader.
  
      datapipe_graph = dp_utils.traverse_dps(datapipe)
      datapipe_adjlist = datapipe_graph_to_adjlist(datapipe_graph)
  
      # (1) Insert minibatch distribution.
      # TODO(BarclayII): Currently I'm using sharding_filter() as a
      # concept demonstration. Later on minibatch distribution should be
      # merged into ItemSampler to maximize efficiency.
      item_samplers = dp_utils.find_dps(
          datapipe_graph,
          ItemSampler,
      )
      for item_sampler in item_samplers:
          datapipe_graph = dp_utils.replace_dp(
              datapipe_graph,
              item_sampler,
              item_sampler.sharding_filter(),
          )
  
      # (2) Cut datapipe at FeatureFetcher and wrap.
      _find_and_wrap_parent(
          datapipe_graph,
          datapipe_adjlist,
          FeatureFetcher,
          MultiprocessingWrapper,
          num_workers=num_workers,
          persistent_workers=persistent_workers,
      )
  
      # (3) Cut datapipe at CopyTo and wrap with prefetcher. This enables the
      # data pipeline up to the CopyTo operation to run in a separate thread.
      _find_and_wrap_parent(
          datapipe_graph,
          datapipe_adjlist,
          CopyTo,
          dp.iter.Prefetcher,
          buffer_size=2,
      )
  
      # The stages after feature fetching is still done in the main process.
      # So we set num_workers to 0 here.
      super().__init__(datapipe, batch_size=None, num_workers=0)
  
file_path: dgl/python/dgl/graphbolt/dataloader.py
imports: import torch.utils.data
  import torchdata.dataloader2.graph
  import torchdata.datapipes
  from .base import CopyTo
  from .feature_fetcher import FeatureFetcher
  from .internal import datapipe_graph_to_adjlist
  from .item_sampler import ItemSampler
label_desc: Use image node_id 1 to create a new DataLoader object from inherited base classes: torch.utils.data with example: obj = DataLoader(datapipe, num_workers, persistent_workers)