






'''simple docstring'''





# Copyright 2022 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
from typing import List






def  lowerCamelCase_  (     lowercase__					,  lowercase__):
 if name != "training_function" and name != "main":
  raise ValueError(F'''Incorrect function name passed: {name}, choose either \'main\' or \'training_function\'''')
 lowerCamelCase__	,						lowerCamelCase__	=  [], False
 for line in lines:
  if not found_start and F'''def {name}''' in line:
   lowerCamelCase__	=  True
   good_lines.append(lowercase__)
   continue
  if found_start:
   if name == "training_function" and "def main" in line:
    return good_lines
   if name == "main" and "if __name__" in line:
    return good_lines
   good_lines.append(lowercase__)






def  lowerCamelCase_  (     lowercase__):
 return [line for line in lines if not line.lstrip().startswith("#") and line != "\n"]






def  lowerCamelCase_  (     lowercase__					,  lowercase__					,  lowercase__					,  lowercase__ = None):
 with open(lowercase__					,  "r") as f:
  lowerCamelCase__	=  f.readlines()
 with open(os.path.abspath(os.path.join("examples"					,  "nlp_example.py"))					,  "r") as f:
  lowerCamelCase__	=  f.readlines()
 with open(lowercase__					,  "r") as f:
  lowerCamelCase__	=  f.readlines()
 if secondary_filename is not None:
  with open(lowercase__					,  "r") as f:
   lowerCamelCase__	=  f.readlines()

    # This is our base, we remove all the code from here in our `full_filename` and `feature_filename` to find the new content
 if parser_only:
  lowerCamelCase__	=  clean_lines(get_function_contents_by_name(lowercase__					,  "main"))
  lowerCamelCase__	=  clean_lines(get_function_contents_by_name(lowercase__					,  "main"))
  lowerCamelCase__	=  clean_lines(get_function_contents_by_name(lowercase__					,  "main"))
  if secondary_filename is not None:
   lowerCamelCase__	=  clean_lines(get_function_contents_by_name(lowercase__					,  "main"))
 else:
  lowerCamelCase__	=  clean_lines(get_function_contents_by_name(lowercase__					,  "training_function"))
  lowerCamelCase__	=  clean_lines(get_function_contents_by_name(lowercase__					,  "training_function"))
  lowerCamelCase__	=  clean_lines(get_function_contents_by_name(lowercase__					,  "training_function"))
  if secondary_filename is not None:
   lowerCamelCase__	=  clean_lines(
       get_function_contents_by_name(lowercase__					,  "training_function"))

 lowerCamelCase__	=  "train_dataloader, eval_dataloader = get_dataloaders(accelerator, batch_size)\n"

 # Specific code in our script that differs from the full version, aka what is new
 lowerCamelCase__	=  []
 lowerCamelCase__	=  []  # We keep track of the idxs just in case it's a repeated statement
 lowerCamelCase__	=  iter(lowercase__)
 for i in range(len(lowercase__) - 1):
  if i not in passed_idxs:
   lowerCamelCase__	=  next(lowercase__)
   if (line not in full_file_func) and (line.lstrip() != _dl_line):
    if "TESTING_MOCKED_DATALOADERS" not in line:
     new_feature_code.append(lowercase__)
     passed_idxs.append(lowercase__)
    else:
     # Skip over the `config['num_epochs'] = 2` statement
     lowerCamelCase__	=  next(lowercase__)

    # Extract out just the new parts from the full_file_training_func
 lowerCamelCase__	=  []
 lowerCamelCase__	=  []  # We keep track of the idxs just in case it's a repeated statement
 for i, line in enumerate(lowercase__):
  if i not in passed_idxs:
   if (line not in full_file_func) and (line.lstrip() != _dl_line):
    if "TESTING_MOCKED_DATALOADERS" not in line:
     new_full_example_parts.append(lowercase__)
     passed_idxs.append(lowercase__)

    # Finally, get the overall diff
 lowerCamelCase__	=  [line for line in new_feature_code if line not in new_full_example_parts]
 if secondary_filename is not None:
  lowerCamelCase__	=  [line for line in full_file_contents if line not in secondary_file_func]
  lowerCamelCase__	=  [line for line in diff_from_example if line not in diff_from_two]

 return diff_from_example









'''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  lowerCamelCase_  (     ):
 lowerCamelCase__	=  Dataset.from_dict({"i": range(17)})
 lowerCamelCase__	=  len(lowercase__)
 lowerCamelCase__	=  3
 lowerCamelCase__	=  [
     split_dataset_by_node(lowercase__					,  rank=lowercase__					,  world_size=lowercase__) for rank in range(lowercase__)
 ]
 assert sum(len(lowercase__) 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  lowerCamelCase_  (     ):
 def gen():
  return ({"i": i} for i in range(17))

 lowerCamelCase__	=  3
 lowerCamelCase__	=  IterableDataset.from_generator(lowercase__)
 lowerCamelCase__	=  len(list(lowercase__))
 lowerCamelCase__	=  [
     split_dataset_by_node(lowercase__					,  rank=lowercase__					,  world_size=lowercase__) for rank in range(lowercase__)
 ]
 assert sum(len(list(lowercase__)) 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  lowerCamelCase_  (     lowercase__):
 def gen(lowercase__):
  for shard in shards:
   yield from ({"i": i, "shard": shard} for i in range(17))

 lowerCamelCase__	=  3
 lowerCamelCase__	=  shards_per_node * world_size
 lowerCamelCase__	=  {"shards": [F'''shard_{shard_idx}.txt''' for shard_idx in range(lowercase__)]}
 lowerCamelCase__	=  IterableDataset.from_generator(lowercase__					,  gen_kwargs=lowercase__)
 lowerCamelCase__	=  len(list(lowercase__))
 assert full_ds.n_shards == world_size * shards_per_node
 lowerCamelCase__	=  [
     split_dataset_by_node(lowercase__					,  rank=lowercase__					,  world_size=lowercase__) for rank in range(lowercase__)
 ]
 assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
 assert sum(len(list(lowercase__)) 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  lowerCamelCase_  (     ):
 def gen():
  return ({"i": i} for i in range(17))

 lowerCamelCase__	=  2
 lowerCamelCase__	=  IterableDataset.from_generator(lowercase__)
 lowerCamelCase__	=  len(list(lowercase__))

 lowerCamelCase__	=  split_dataset_by_node(lowercase__					,  rank=0					,  world_size=lowercase__).shuffle(seed=42)
 assert len(list(lowercase__)) == 1 + full_size // world_size
 with pytest.raises(lowercase__):
  split_dataset_by_node(lowercase__					,  rank=0					,  world_size=lowercase__).shuffle()

 lowerCamelCase__	=  split_dataset_by_node(full_ds.shuffle(seed=42)					,  rank=0					,  world_size=lowercase__)
 assert len(list(lowercase__)) == 1 + full_size // world_size
 with pytest.raises(lowercase__):
  split_dataset_by_node(full_ds.shuffle()					,  rank=0					,  world_size=lowercase__)






@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  lowerCamelCase_  (     lowercase__):
 lowerCamelCase__	=  2
 lowerCamelCase__	=  get_torch_dist_unique_port()
 lowerCamelCase__	=  Path(lowercase__).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
 lowerCamelCase__	=  F'''
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    '''.split()
 lowerCamelCase__	=  F'''
        --streaming={streaming}
    '''.split()
 lowerCamelCase__	=  [sys.executable] + distributed_args + args
 execute_subprocess_async(lowercase__					,  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  lowerCamelCase_  (     lowercase__					,  lowercase__):
 lowerCamelCase__	=  True
 lowerCamelCase__	=  get_torch_dist_unique_port()
 lowerCamelCase__	=  Path(lowercase__).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
 lowerCamelCase__	=  F'''
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    '''.split()
 lowerCamelCase__	=  F'''
        --streaming={streaming}
        --num_workers={num_workers}
    '''.split()
 lowerCamelCase__	=  [sys.executable] + distributed_args + args
 execute_subprocess_async(lowercase__					,  env=os.environ.copy())

