code: def __init__(
      self,
      dataset,
      batch_size=1,
      drop_last=False,
      shuffle=False,
      num_workers=1,
      collate_fn=None,
      buf_size=1000,
      stream_shuffle_size=0,
  ):
      self.dataset = dataset
      self.batch_size = batch_size
      self.shuffle = shuffle
      self.num_workers = num_workers
      self.collate_fn = collate_fn
      self.buf_size = buf_size
      self.drop_last = drop_last
      self.stream_shuffle_size = stream_shuffle_size
  
      if self.shuffle and isinstance(self.dataset, StreamDataset):
          warn_msg = (
              "The argument [shuffle] should not be True with StreamDataset. "
              "It will be ignored. "
              "You might want to set [stream_shuffle_size] with StreamDataset."
          )
          warnings.warn(warn_msg)
  
      if (
          self.stream_shuffle_size > 0
          and self.batch_size >= stream_shuffle_size
      ):
          raise ValueError(
              "stream_shuffle_size must be larger than batch_size,"
              "but got [stream_shuffle_size=%s] smaller than [batch_size=%s]"
              % (self.stream_shuffle_size, self.batch_size)
          )
  
      if self.stream_shuffle_size > 0 and isinstance(
          self.dataset, Dataset
      ):
          warn_msg = (
              "[stream_shuffle_size] should not be set with Dataset. "
              "It will be ignored. "
              "You might want to set [shuffle] with Dataset."
          )
          warnings.warn(warn_msg)
  
      if self.num_workers < 1:
          raise ValueError(
              "num_workers(default: 1) should be larger than 0, "
              "but got [num_workers=%s] < 1." % self.num_workers
          )
  
      if isinstance(self.dataset, StreamDataset):  # for stream data
          # generating a iterable sequence for produce batch data without repetition
          self.sampler = StreamSampler(
              self.dataset,
              batch_size=self.batch_size,
              drop_last=self.drop_last,
          )
      else:
          self.sampler = Sampler(
              self.dataset,
              batch_size=self.batch_size,
              drop_last=self.drop_last,
              shuffle=self.shuffle,
          )
  
      if self.num_workers == 1:
          generator = paddle.reader.buffered(
              _DataLoaderIter(self, 0), self.buf_size
          )
      else:
          worker_pool = [
              _DataLoaderIter(self, wid)
              for wid in range(self.num_workers)
          ]
          workers = mp_reader.multiprocess_reader(
              worker_pool, use_pipe=True, queue_size=1000
          )
  
          generator = paddle.reader.buffered(workers, self.buf_size)
  
      # need a more elegant method
      gen = generator()
      first_data = next(gen)
  
      def _generator():
          yield first_data
          while True:
              try:
                  data = next(gen)
                  yield data
              except:
                  break
  
      self.generator = _generator
  
imports: import os
  import sys
  import time
  import warnings
  import numpy
  from collections import namedtuple
  import paddle
  from pgl.utils import mp_reader
  from pgl.utils.logger import log
  from pgl.utils.data.dataset import Dataset, StreamDataset
  from pgl.utils.data.sampler import Sampler, StreamSampler
  from pgl.distributed import DistGraphClient, DistGraphServer
  from utils.config import prepare_config
label_desc: Use image node_id 1 to create a new DistCPUDataloader object from inherited base classes: object with example: obj = DistCPUDataloader(dataset, batch_size, drop_last, shuffle, num_workers, collate_fn, buf_size, stream_shuffle_size)