code: 
  
  
  
  
  
  
  import os
  import sys
  from pathlib import Path
  
  import pytest
  
  from datasets import Dataset, IterableDataset
  from datasets.distributed import split_dataset_by_node
  
  from .utils import execute_subprocess_async, get_torch_dist_unique_port, require_torch
  
  
  
  
  
  
  
  def 	SCREAMING_SNAKE_CASE							(      )				->			Optional[int]:
       UpperCamelCase__    :     Union[str, Any]        =						Dataset.from_dict({"i": range(17    )}    )
       UpperCamelCase__    :     Union[str, Any]        =						len(__snake_case    )
       UpperCamelCase__    :     List[str]        =						3
       UpperCamelCase__    :     Tuple        =						[
           split_dataset_by_node(__snake_case      ,					rank=__snake_case      ,					world_size=__snake_case    ) for rank in range(__snake_case    )
       ]
       assert sum(len(__snake_case    ) for ds in datasets_per_rank    ) == full_size
       assert len({tuple(x.values()    ) for ds in datasets_per_rank for x in ds}    ) == full_size
  
  
  
  
  
  
  
  def 	SCREAMING_SNAKE_CASE							(      )				->			Dict:
       def gen():
            return ({"i": i} for i in range(17    ))
  
       UpperCamelCase__    :     Union[str, Any]        =						3
       UpperCamelCase__    :     Tuple        =						IterableDataset.from_generator(__snake_case    )
       UpperCamelCase__    :     List[Any]        =						len(list(__snake_case    )    )
       UpperCamelCase__    :     Dict        =						[
           split_dataset_by_node(__snake_case      ,					rank=__snake_case      ,					world_size=__snake_case    ) for rank in range(__snake_case    )
       ]
       assert sum(len(list(__snake_case    )    ) for ds in datasets_per_rank    ) == full_size
       assert len({tuple(x.values()    ) for ds in datasets_per_rank for x in ds}    ) == full_size
  
  
  
  
  
  
  
  @pytest.mark.parametrize("shards_per_node"      ,					[1, 2, 3]    )
  def 	SCREAMING_SNAKE_CASE							(      __lowerCAmelCase    )				->			List[Any]:
       def gen(__lowerCAmelCase    ):
            for shard in shards:
                 yield from ({"i": i, "shard": shard} for i in range(17    ))
  
       UpperCamelCase__    :     Optional[Any]        =						3
       UpperCamelCase__    :     int        =						shards_per_node * world_size
       UpperCamelCase__    :     Optional[int]        =						{"shards": [f'shard_{shard_idx}.txt' for shard_idx in range(__snake_case    )]}
       UpperCamelCase__    :     List[str]        =						IterableDataset.from_generator(__snake_case      ,					gen_kwargs=__snake_case    )
       UpperCamelCase__    :     List[str]        =						len(list(__snake_case    )    )
       assert full_ds.n_shards == world_size * shards_per_node
       UpperCamelCase__    :     Union[str, Any]        =						[
           split_dataset_by_node(__snake_case      ,					rank=__snake_case      ,					world_size=__snake_case    ) for rank in range(__snake_case    )
       ]
       assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
       assert sum(len(list(__snake_case    )    ) for ds in datasets_per_rank    ) == full_size
       assert len({tuple(x.values()    ) for ds in datasets_per_rank for x in ds}    ) == full_size
  
  
  
  
  
  
  
  def 	SCREAMING_SNAKE_CASE							(      )				->			int:
       def gen():
            return ({"i": i} for i in range(17    ))
  
       UpperCamelCase__    :     Any        =						2
       UpperCamelCase__    :     Union[str, Any]        =						IterableDataset.from_generator(__snake_case    )
       UpperCamelCase__    :     Optional[int]        =						len(list(__snake_case    )    )
  
       UpperCamelCase__    :     int        =						split_dataset_by_node(__snake_case      ,					rank=0      ,					world_size=__snake_case    ).shuffle(seed=42    )
       assert len(list(__snake_case    )    ) == 1 + full_size // world_size
       with pytest.raises(__snake_case    ):
            split_dataset_by_node(__snake_case      ,					rank=0      ,					world_size=__snake_case    ).shuffle()
  
       UpperCamelCase__    :     Dict        =						split_dataset_by_node(full_ds.shuffle(seed=42    )      ,					rank=0      ,					world_size=__snake_case    )
       assert len(list(__snake_case    )    ) == 1 + full_size // world_size
       with pytest.raises(__snake_case    ):
            split_dataset_by_node(full_ds.shuffle()      ,					rank=0      ,					world_size=__snake_case    )
  
  
  
  
  
  
  
  @pytest.mark.parametrize("streaming"      ,					[False, True]    )
  @require_torch
  @pytest.mark.skipif(os.name == "nt"      ,					reason="execute_subprocess_async doesn\'t support windows"    )
  @pytest.mark.integration
  def 	SCREAMING_SNAKE_CASE							(      __lowerCAmelCase    )				->			str:
       UpperCamelCase__    :     Optional[Any]        =						2
       UpperCamelCase__    :     Union[str, Any]        =						get_torch_dist_unique_port()
       UpperCamelCase__    :     int        =						Path(__snake_case    ).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
       UpperCamelCase__    :     List[Any]        =						f'\n        -m torch.distributed.launch\n        --nproc_per_node={nproc_per_node}\n        --master_port={master_port}\n        {test_script}\n    '.split()
       UpperCamelCase__    :     str        =						f'\n        --streaming={streaming}\n    '.split()
       UpperCamelCase__    :     Tuple        =						[sys.executable] + distributed_args + args
       execute_subprocess_async(__snake_case      ,					env=os.environ.copy()    )
  
  
  
  
  
  
  
  @pytest.mark.parametrize(
      "nproc_per_node, num_workers"      ,					[
          (2, 2),  # each node has 2 shards and each worker has 1 shards
          (3, 2),  # each node uses all the shards but skips examples, and each worker has 2 shards
      ]      ,					)
  @require_torch
  @pytest.mark.skipif(os.name == "nt"      ,					reason="execute_subprocess_async doesn\'t support windows"    )
  @pytest.mark.integration
  def 	SCREAMING_SNAKE_CASE							(      __lowerCAmelCase      ,					__lowerCAmelCase    )				->			Any:
       UpperCamelCase__    :     Tuple        =						True
       UpperCamelCase__    :     Optional[int]        =						get_torch_dist_unique_port()
       UpperCamelCase__    :     str        =						Path(__snake_case    ).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
       UpperCamelCase__    :     Any        =						f'\n        -m torch.distributed.launch\n        --nproc_per_node={nproc_per_node}\n        --master_port={master_port}\n        {test_script}\n    '.split()
       UpperCamelCase__    :     int        =						f'\n        --streaming={streaming}\n        --num_workers={num_workers}\n    '.split()
       UpperCamelCase__    :     List[Any]        =						[sys.executable] + distributed_args + args
       execute_subprocess_async(__snake_case      ,					env=os.environ.copy()    )
style_context: 
  
  
  
  
  
  
  def 	SCREAMING_SNAKE_CASE							(      __lowerCAmelCase    )				->			str:
       return "".join(chr(ord(__lowerCAmelCase    ) + 32    ) if "A" <= char <= "Z" else char for char in word    )
  
  
  if __name__ == "__main__":
   from doctest import testmod
  
   testmod()