class_docstr: FullShuffle shuffles the chunks and associates them to the ranks.
  
  As the number of items in a chunk varies, it is possible for a rank to end up with more or less items.
  
  To ensure the same fixed dataset length for all ranks while dropping as few items as possible,
  
  we adopt the following strategy.
  
  We compute the maximum number of items per rank (M) and iterate through the chunks and ranks
  
  until we have associated at least M items per rank.
  
  As a result, we lose at most (number of ranks) items. However, as some chunks are shared across ranks. This leads to
  the same chunk to be downloaded multiple times.
class_docstr_tok: ["FullShuffle","shuffles","the","chunks","and","associates","them","to","the","ranks",".","As","the","number","of","items","in","a","chunk","varies",",","it","is","possible","for","a","rank","to","end","up","with","more","or","less","items",".","To","ensure","the","same","fixed","dataset","length","for","all","ranks","while","dropping","as","few","items","as","possible",",","we","adopt","the","following","strategy",".","We","compute","the","maximum","number","of","items","per","rank","(","M",")","and","iterate","through","the","chunks","and","ranks","until","we","have","associated","at","least","M","items","per","rank",".","As","a","result",",","we","lose","at","most","(","number","of","ranks",")","items",".","However",",","as","some","chunks","are","shared","across","ranks",".","This","leads","to","the","same","chunk","to","be","downloaded","multiple","times","."]
code: def get_chunks_and_intervals_per_ranks(
      self, distributed_env: _DistributedEnv, current_epoch: int
  ) -> Any:
      # 1. Get the intervals
      chunk_intervals = self.cache.get_chunk_intervals()
  
      # 2. Shuffle them
      indexes = range(len(chunk_intervals))
  
      # FIXME: Shuffling should be done only within the nodes to benefit
      # from cache if the dataset doesn't fit on the node.
      shuffled_indexes = np.random.RandomState(
          seed=self.seed + current_epoch
      ).permutation(indexes)
      shuffled_chunk_intervals = np.asarray(chunk_intervals)[
          shuffled_indexes
      ]
  
      # 3. Compute the items budget of each rank
      num_items = sum(
          [(interval[-1] - interval[0]) for interval in chunk_intervals]
      )
      num_items_per_ranks: List[int] = [
          num_items // distributed_env.world_size
          + num_items % distributed_env.world_size
          if rank == distributed_env.world_size - 1
          and not self.drop_last
          else num_items // distributed_env.world_size
          for rank in range(distributed_env.world_size)
      ]
      chunks_per_ranks: List[List[int]] = [
          [] for _ in range(distributed_env.world_size)
      ]
      intervals_per_ranks: List[List[List[int]]] = [
          [] for _ in range(distributed_env.world_size)
      ]
  
      # 4. Assign the chunk & intervals to each rank
      for chunk_index, chunk_interval in zip(
          shuffled_indexes, shuffled_chunk_intervals
      ):
          rank = 0
  
          while True:
              if rank == len(num_items_per_ranks):
                  break
  
              items_left_to_assign = num_items_per_ranks[rank]
  
              if items_left_to_assign == 0:
                  rank += 1
                  continue
  
              items_in_chunk = chunk_interval[-1] - chunk_interval[0]
  
              if items_in_chunk == 0:
                  break
  
              if items_in_chunk > items_left_to_assign:
                  chunks_per_ranks[rank].append(chunk_index)
                  begin, end = chunk_interval
                  intervals_per_ranks[rank].append(
                      [begin, begin + items_left_to_assign]
                  )
                  chunk_interval = (begin + items_left_to_assign, end)
                  num_items_per_ranks[rank] = 0
                  rank += 1
              else:
                  chunks_per_ranks[rank].append(chunk_index)
                  intervals_per_ranks[rank].append(chunk_interval)
                  num_items_per_ranks[rank] -= items_in_chunk
                  break
  
      return chunks_per_ranks, intervals_per_ranks
  
file_path: pytorch-lightning/src/lightning/data/streaming/shuffle.py
imports: from abc import ABC, abstractmethod
  from functools import lru_cache
  from typing import Any, List, Tuple
  import numpy
  from lightning.data.streaming import Cache
  from lightning.data.utilities.env import _DistributedEnv
label_desc: Use image node_id 1 for calling the FullShuffle obj's underlying member method code with example usage: obj.get_chunks_and_intervals_per_ranks(distributed_env, current_epoch) and returns: chunks_per_ranks,  intervals_per_ranks