class_docstr: Open dataset in readonly mode.
  
  `DistributeDataset` is used to read datasets in a distributed manner.
  Data in this dataset will be distributed evenly in blocks to each worker in the `distributed communicator`.
  
  **Note** When all data has been read, reading dataset again will revert back to the first data.
  
  Args:
      path (str): Path to dataset.
      rank (int): Rank in distributed communicator. See: bmtrain.rank()
      world_size (int): Total workers in distributed communicator. See: bmtrain.world_size()
      block_size (int): Size of each block in bytes. All files in the same dataset should have the same block size. Default: 16MB
  
  Example:
      >>> dataset = DistributedDataset("/path/to/dataset")
      >>> for i in range(10):
      >>>     dataset.read()
class_docstr_tok: ["Open","dataset","in","readonly","mode",".","`","DistributeDataset","`","is","used","to","read","datasets","in","a","distributed","manner",".","Data","in","this","dataset","will","be","distributed","evenly","in","blocks","to","each","worker","in","the","`","distributed","communicator","`",".","*","*","Note","*","*","When","all","data","has","been","read",",","reading","dataset","again","will","revert","back","to","the","first","data",".","Args",":","path","(","str",")",":","Path","to","dataset",".","rank","(","int",")",":","Rank","in","distributed","communicator",".","See",":","bmtrain.rank","(",")","world_size","(","int",")",":","Total","workers","in","distributed","communicator",".","See",":","bmtrain.world_size","(",")","block_size","(","int",")",":","Size","of","each","block","in","bytes",".","All","files","in","the","same","dataset","should","have","the","same","block","size",".","Default",":","16MB","Example",":",">",">",">","dataset","=","DistributedDataset","(","``","\/path\/to\/dataset","''",")",">",">",">","for","i","in","range","(","10",")",":",">",">",">","dataset.read","(",")"]
docstr: Read a piece of data from dataset.
  
  Workers in different ranks will read different data.
code: def read(self):
      """Read a piece of data from dataset.
  
      Workers in different ranks will read different data.
      """
      if self._curr_block is None:
          next_block_id = self._get_next_block()
          f_info = self._get_block_file(next_block_id)
          try:
              self._open_file(
                  f_info.file_name,
                  (next_block_id - f_info.block_begin)
                  * f_info.block_size,
              )
              self._curr_block = next_block_id
          except FileNotFoundError:
              print("ERR: reading again!")
              self._mask_file(f_info)
              return self.read()  # read again
  
      if self._fp is None:
          raise RuntimeError("Dataset is not initialized")
  
      MAGIC = self._fp.read(1)
      if MAGIC == b"\x1F":
          # correct
          size = struct.unpack("I", self._fp.read(4))[0]
          data = self._fp.read(size)
          return self.serializer.deserialize(data)
      elif MAGIC == b"\x00":
          # end of block
          self._curr_block = None
          return self.read()  # read next block
      else:
          raise ValueError("Invalid magic header")
  
imports: import io
  import os
  import struct
  from typing import List, Optional, Set
  import torch
  import bisect
  import bmtrain
  import json
  from .serializer import Serializer, PickleSerializer
  import random
  import string
  import time
label_desc: Use image node_id 12 for calling the DistributedDataset obj's underlying member method code with example usage: obj.read() and returns: self, self, self