code: def test_streaming_dataset_distributed_no_shuffle(drop_last, tmpdir):
      seed_everything(42)
  
      cache = Cache(str(tmpdir), chunk_size=10)
      for i in range(101):
          cache[i] = i
  
      cache.done()
      cache.merge()
  
      dataset = StreamingDataset(
          input_dir=str(tmpdir), shuffle=False, drop_last=drop_last
      )
      assert not dataset.shuffle
      _ = dataset[0]  # init shuffler
      assert isinstance(dataset.shuffler, NoShuffle)
  
      for i in range(101):
          assert dataset[i] == i
  
      dataset.distributed_env = _DistributedEnv(1, 0)
      assert len(dataset) == 101
  
      dataset.distributed_env = _DistributedEnv(2, 0)
      assert len(dataset) == 50 + int(not drop_last)
      dataset_iter = iter(dataset)
      assert len(dataset_iter) == 50 + int(not drop_last)
      process_1_1 = list(dataset_iter)
      assert len(process_1_1) == 50 + int(not drop_last)
      assert process_1_1[:10] == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
      dataset_iter = iter(dataset)
      assert len(dataset_iter) == 50 + int(not drop_last)
      process_1_2 = list(dataset_iter)
      assert process_1_2[:10] == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
      assert len(process_1_2) == 50 + int(not drop_last)
  
      dataset = StreamingDataset(
          input_dir=str(tmpdir), shuffle=False, drop_last=drop_last
      )
      dataset.distributed_env = _DistributedEnv(2, 1)
      assert len(dataset) == 50
      dataset_iter = iter(dataset)
      process_2_1 = list(dataset_iter)
      assert process_2_1[:10] == [
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
      ]
      assert len(process_2_1) == 50
      dataset_iter = iter(dataset)
      assert len(dataset_iter) == 50
      process_2_2 = list(dataset_iter)
      assert process_2_2[:10] == [
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
      ]
  
      assert len(process_2_2) == 50
  
      (
          _,
          intervals_per_ranks,
      ) = dataset.shuffler.get_chunks_and_intervals_per_ranks(
          dataset.distributed_env, dataset.current_epoch
      )
  
      assert process_1_1 == process_1_2
  
      found_list = []
      for i in process_1_1:
          found = False
          for interval in intervals_per_ranks[0]:
              if interval[0] <= i <= interval[1]:
                  found = True
                  break
          found_list.append(found)
  
      assert all(found_list) is True
  
      found_list = []
      for i in process_2_1:
          found = False
          for interval in intervals_per_ranks[1]:
              if interval[0] <= i <= interval[1]:
                  found = True
                  break
          found_list.append(found)
  
      assert all(found_list) is True
  
      assert len([i for i in process_1_1 if i in process_2_1]) == 0
      assert len([i for i in process_1_2 if i in process_2_2]) == 0
  
imports: import os
  import sys
  from time import sleep
  from unittest import mock
  import numpy
  import pytest
  import torch
  from lightning import seed_everything
  from lightning.data.streaming import Cache, functions
  from lightning.data.streaming.dataloader import StreamingDataLoader
  from lightning.data.streaming.dataset import _INDEX_FILENAME, Dir, RemoteDir, StreamingDataset, _associate_chunks_to_workers, _replay_chunks_sampling, _replay_sampling, _should_replace_path, _try_create_cache_dir
  from lightning.data.streaming.item_loader import TokensLoader
  from lightning.data.streaming.shuffle import FullShuffle, NoShuffle
  from lightning.data.utilities.env import Environment, _DistributedEnv, _WorkerEnv
  from torch.utils.data import DataLoader
label_desc: Use image node_id 3 for calling a global function with example usage: test_streaming_dataset_distributed_no_shuffle(drop_last, tmpdir) without return types