code: 
  
  """simple docstring"""
  
  
  
  
  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		(		):
  					"""simple docstring"""
  					A    =      Dataset.from_dict({"i": range(17   )}   )
  					A    =      len(_A   )
  					A    =      3
  					A    =      [
  					    split_dataset_by_node(_A    ,							rank=_A    ,							world_size=_A   ) for rank in range(_A   )
  					]
  					assert sum(len(_A   ) 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		(		):
  					"""simple docstring"""
  
  
  					def gen():
  										return ({"i": i} for i in range(17   ))
  
  					A    =      3
  					A    =      IterableDataset.from_generator(_A   )
  					A    =      len(list(_A   )   )
  					A    =      [
  					    split_dataset_by_node(_A    ,							rank=_A    ,							world_size=_A   ) for rank in range(_A   )
  					]
  					assert sum(len(list(_A   )   ) 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		(		lowercase__   ):
  					"""simple docstring"""
  
  
  					def gen(lowercase__   ):
  										for shard in shards:
  															yield from ({"i": i, "shard": shard} for i in range(17   ))
  
  					A    =      3
  					A    =      shards_per_node * world_size
  					A    =      {"shards": [F"""shard_{shard_idx}.txt""" for shard_idx in range(_A   )]}
  					A    =      IterableDataset.from_generator(_A    ,							gen_kwargs=_A   )
  					A    =      len(list(_A   )   )
  					assert full_ds.n_shards == world_size * shards_per_node
  					A    =      [
  					    split_dataset_by_node(_A    ,							rank=_A    ,							world_size=_A   ) for rank in range(_A   )
  					]
  					assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
  					assert sum(len(list(_A   )   ) 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		(		):
  					"""simple docstring"""
  
  
  					def gen():
  										return ({"i": i} for i in range(17   ))
  
  					A    =      2
  					A    =      IterableDataset.from_generator(_A   )
  					A    =      len(list(_A   )   )
  
  					A    =      split_dataset_by_node(_A    ,							rank=0    ,							world_size=_A   ).shuffle(seed=42   )
  					assert len(list(_A   )   ) == 1 + full_size // world_size
  					with pytest.raises(_A   ):
  										split_dataset_by_node(_A    ,							rank=0    ,							world_size=_A   ).shuffle()
  
  					A    =      split_dataset_by_node(full_ds.shuffle(seed=42   )    ,							rank=0    ,							world_size=_A   )
  					assert len(list(_A   )   ) == 1 + full_size // world_size
  					with pytest.raises(_A   ):
  										split_dataset_by_node(full_ds.shuffle()    ,							rank=0    ,							world_size=_A   )
  @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		(		lowercase__   ):
  					"""simple docstring"""
  					A    =      2
  					A    =      get_torch_dist_unique_port()
  					A    =      Path(_A   ).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
  					A    =      F"""
          -m torch.distributed.launch
          --nproc_per_node={nproc_per_node}
          --master_port={master_port}
          {test_script}
      """.split()
  					A    =      F"""
          --streaming={streaming}
      """.split()
  					A    =      [sys.executable] + distributed_args + args
  					execute_subprocess_async(_A    ,							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		(		lowercase__    ,							lowercase__   ):
  					"""simple docstring"""
  					A    =      True
  					A    =      get_torch_dist_unique_port()
  					A    =      Path(_A   ).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
  					A    =      F"""
          -m torch.distributed.launch
          --nproc_per_node={nproc_per_node}
          --master_port={master_port}
          {test_script}
      """.split()
  					A    =      F"""
          --streaming={streaming}
          --num_workers={num_workers}
      """.split()
  					A    =      [sys.executable] + distributed_args + args
  					execute_subprocess_async(_A    ,							env=os.environ.copy()   )
  
  
  
  
  
  
style_context: 
  
  """simple docstring"""
  
  
  
  
  import functools
  from math import sqrt
  from time import time
  def      __SCREAMING_SNAKE_CASE		(		lowercase__    ,							*lowercase__    ,							**lowercase__   ):
  					"""simple docstring"""
  					A    =      time()
  					A    =      func(*lowercase__    ,							**lowercase__   )
  					A    =      time()
  					if int(end - start   ) > 0:
  										print(F"""{func.__name__} runtime: {(end - start):0.4f} s"""   )
  					else:
  										print(F"""{func.__name__} runtime: {(end - start) * 1_000:0.4f} ms"""   )
  					return output
  def      __SCREAMING_SNAKE_CASE		(		lowercase__   ):
  					"""simple docstring"""
  					if n < 0:
  										raise Exception("n is negative"   )
  					if n == 0:
  										return [0]
  					A    =      [0, 1]
  					for _ in range(n - 1   ):
  										fib.append(fib[-1] + fib[-2]   )
  					return fib
  def      __SCREAMING_SNAKE_CASE		(		lowercase__   ):
  					"""simple docstring"""
  
  					def fib_recursive_term(lowercase__   ) -> int:
  										if i < 0:
  															raise Exception("n is negative"   )
  										if i < 2:
  															return i
  										return fib_recursive_term(i - 1   ) + fib_recursive_term(i - 2   )
  
  					if n < 0:
  										raise Exception("n is negative"   )
  					return [fib_recursive_term(lowercase__   ) for i in range(n + 1   )]
  def      __SCREAMING_SNAKE_CASE		(		lowercase__   ):
  					"""simple docstring"""
  
  					@functools.cache
  					def fib_recursive_term(lowercase__   ) -> int:
  										if i < 0:
  															raise Exception("n is negative"   )
  										if i < 2:
  															return i
  										return fib_recursive_term(i - 1   ) + fib_recursive_term(i - 2   )
  
  					if n < 0:
  										raise Exception("n is negative"   )
  					return [fib_recursive_term(lowercase__   ) for i in range(n + 1   )]
  def      __SCREAMING_SNAKE_CASE		(		lowercase__   ):
  					"""simple docstring"""
  					if n < 0:
  										raise Exception("n is negative"   )
  					# Cache must be outside recursuive function
  					# other it will reset every time it calls itself.
  					A    =      {0: 0, 1: 1, 2: 1}  # Prefilled cache
  
  					def rec_fn_memoized(lowercase__   ) -> int:
  										if num in cache:
  															return cache[num]
  
  										A    =      rec_fn_memoized(num - 1   ) + rec_fn_memoized(num - 2   )
  										A    =      value
  										return value
  
  					return [rec_fn_memoized(lowercase__   ) for i in range(n + 1   )]
  def      __SCREAMING_SNAKE_CASE		(		lowercase__   ):
  					"""simple docstring"""
  					if n < 0:
  										raise Exception("n is negative"   )
  					if n >= 1_475:
  										raise Exception("n is too large"   )
  					A    =      sqrt(5   )
  					A    =      (1 + sqrt_a) / 2
  					return [round(phi**i / sqrt_a   ) for i in range(n + 1   )]
  
  
  if __name__ == "__main__":
  	__A  :					Optional[Any]												=	30
  	time_func(fib_iterative, num)
  	time_func(fib_recursive, num)  # Around 3s runtime
  	time_func(fib_recursive_cached, num)  # Around 0ms runtime
  	time_func(fib_memoization, num)
  	time_func(fib_binet, num)
  
  
  
  
  
  