code: def __iter_distributed__(
      self,
  ) -> Iterator[List[Union[int, ChunkedIndex]]]:
      self.indices = list(range(self._dataset_size))
      replica_size = self._dataset_size // self._num_replicas
      worker_size = self._dataset_size // (
          self._num_replicas * self._num_workers
      )
      for rank in range(self._num_replicas):
          if rank != self._global_rank:
              continue
  
          is_last_replica = rank == self._num_replicas - 1
          start_replica = rank * replica_size
          end_replica = (
              self._dataset_size
              if is_last_replica
              else (rank + 1) * replica_size
          )
          replica_indices = self.indices[start_replica:end_replica]
  
          replica_size = len(replica_indices)
  
          worker_indices = []
          for worker_idx in range(self._num_workers):
              is_last_worker = worker_idx == self._num_workers - 1
              start_worker = worker_idx * worker_size
              end_worker = (
                  replica_size
                  if is_last_worker
                  else (worker_idx + 1) * worker_size
              )
              worker_indices.append(
                  replica_indices[start_worker:end_worker]
              )
  
      assert sum([len(s) for s in worker_indices]) == len(
          replica_indices
      )
  
      worker_indices_batches = [
          self._chunk_list(indices, self._batch_size)
          for indices in worker_indices
      ]
  
      yield from self.__iter_indices_per_workers__(
          worker_indices_batches
      )
  
file_path: pytorch-lightning/src/lightning/data/streaming/sampler.py
imports: import logging
  from dataclasses import dataclass
  from typing import Any, Dict, Iterator, List, Optional, Union
  import numpy
label_desc: Use image node_id 5 for calling the CacheBatchSampler obj's underlying member method code with example usage: obj.__iter_distributed__() without return types