





import inspect
import os

import torch
from transformers import AutoModel
from transformers.testing_utils import mockenv_context
from transformers.trainer_utils import set_seed

import accelerate
from accelerate.accelerator import Accelerator
from accelerate.state import AcceleratorState
from accelerate.test_utils.testing import (
    AccelerateTestCase,
    TempDirTestCase,
    execute_subprocess_async,
    require_cuda,
    require_fsdp,
    require_multi_gpu,
    slow,
)
from accelerate.utils.constants import (
    FSDP_AUTO_WRAP_POLICY,
    FSDP_BACKWARD_PREFETCH,
    FSDP_SHARDING_STRATEGY,
    FSDP_STATE_DICT_TYPE,
)
from accelerate.utils.dataclasses import FullyShardedDataParallelPlugin
from accelerate.utils.other import patch_environment


set_seed(4_2)

SCREAMING_SNAKE_CASE_						       =	"""bert-base-cased"""
SCREAMING_SNAKE_CASE_						       =	"""fp16"""
SCREAMING_SNAKE_CASE_						       =	"""bf16"""
SCREAMING_SNAKE_CASE_						       =	[FPaa, BFaa]

@require_fsdp
@require_cuda
class 		UpperCamelCase__			(				lowerCAmelCase_       ):



     '''simple docstring'''

     def     SCREAMING_SNAKE_CASE__  (							self						:   Tuple	)					->		Optional[int]:




           '''simple docstring'''


           super().setUp()

           SCREAMING_SNAKE_CASE		=						dict(
               ACCELERATE_USE_FSDP="""true"""						,MASTER_ADDR="""localhost"""						,MASTER_PORT="""10999"""						,RANK="""0"""						,LOCAL_RANK="""0"""						,WORLD_SIZE="""1"""						,)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Any	)					->		List[str]:




           '''simple docstring'''


           from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

           for i, strategy in enumerate(lowerCamelCase__	):
                 SCREAMING_SNAKE_CASE		=						self.dist_env.copy()
                 SCREAMING_SNAKE_CASE		=						F"""{i + 1}"""
                 SCREAMING_SNAKE_CASE		=						strategy
                 with mockenv_context(**lowerCamelCase__	):
                       SCREAMING_SNAKE_CASE		=						FullyShardedDataParallelPlugin()
                       self.assertEqual(fsdp_plugin.sharding_strategy						,ShardingStrategy(i + 1	)	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Any	)					->		List[Any]:




           '''simple docstring'''


           from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

           for i, prefetch_policy in enumerate(lowerCamelCase__	):
                 SCREAMING_SNAKE_CASE		=						self.dist_env.copy()
                 SCREAMING_SNAKE_CASE		=						prefetch_policy
                 with mockenv_context(**lowerCamelCase__	):
                       SCREAMING_SNAKE_CASE		=						FullyShardedDataParallelPlugin()
                       if prefetch_policy == "NO_PREFETCH":
                             self.assertIsNone(fsdp_plugin.backward_prefetch	)
                       else:
                             self.assertEqual(fsdp_plugin.backward_prefetch						,BackwardPrefetch(i + 1	)	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Union[str, Any]	)					->		Optional[int]:




           '''simple docstring'''


           from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

           for i, state_dict_type in enumerate(lowerCamelCase__	):
                 SCREAMING_SNAKE_CASE		=						self.dist_env.copy()
                 SCREAMING_SNAKE_CASE		=						state_dict_type
                 with mockenv_context(**lowerCamelCase__	):
                       SCREAMING_SNAKE_CASE		=						FullyShardedDataParallelPlugin()
                       self.assertEqual(fsdp_plugin.state_dict_type						,StateDictType(i + 1	)	)
                       if state_dict_type == "FULL_STATE_DICT":
                             self.assertTrue(fsdp_plugin.state_dict_config.offload_to_cpu	)
                             self.assertTrue(fsdp_plugin.state_dict_config.ranka_only	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Optional[int]	)					->		Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						AutoModel.from_pretrained(lowerCamelCase__	)
           for policy in FSDP_AUTO_WRAP_POLICY:
                 SCREAMING_SNAKE_CASE		=						self.dist_env.copy()
                 SCREAMING_SNAKE_CASE		=						policy
                 if policy == "TRANSFORMER_BASED_WRAP":
                       SCREAMING_SNAKE_CASE		=						"""BertLayer"""
                 elif policy == "SIZE_BASED_WRAP":
                       SCREAMING_SNAKE_CASE		=						"""2000"""
                 with mockenv_context(**lowerCamelCase__	):
                       SCREAMING_SNAKE_CASE		=						FullyShardedDataParallelPlugin()
                       fsdp_plugin.set_auto_wrap_policy(lowerCamelCase__	)
                       if policy == "NO_WRAP":
                             self.assertIsNone(fsdp_plugin.auto_wrap_policy	)
                       else:
                             self.assertIsNotNone(fsdp_plugin.auto_wrap_policy	)

           SCREAMING_SNAKE_CASE		=						self.dist_env.copy()
           SCREAMING_SNAKE_CASE		=						"""TRANSFORMER_BASED_WRAP"""
           SCREAMING_SNAKE_CASE		=						"""T5Layer"""
           with mockenv_context(**lowerCamelCase__	):
                 SCREAMING_SNAKE_CASE		=						FullyShardedDataParallelPlugin()
                 with self.assertRaises(lowerCamelCase__	) as cm:
                       fsdp_plugin.set_auto_wrap_policy(lowerCamelCase__	)
                 self.assertTrue("""Could not find the transformer layer class to wrap in the model.""" in str(cm.exception	)	)

           SCREAMING_SNAKE_CASE		=						self.dist_env.copy()
           SCREAMING_SNAKE_CASE		=						"""SIZE_BASED_WRAP"""
           SCREAMING_SNAKE_CASE		=						"""0"""
           with mockenv_context(**lowerCamelCase__	):
                 SCREAMING_SNAKE_CASE		=						FullyShardedDataParallelPlugin()
                 fsdp_plugin.set_auto_wrap_policy(lowerCamelCase__	)
                 self.assertIsNone(fsdp_plugin.auto_wrap_policy	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Any	)					->		str:




           '''simple docstring'''


           from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
           from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

           for mp_dtype in dtypes:
                 SCREAMING_SNAKE_CASE		=						self.dist_env.copy()
                 SCREAMING_SNAKE_CASE		=						mp_dtype
                 with mockenv_context(**lowerCamelCase__	):
                       SCREAMING_SNAKE_CASE		=						Accelerator()
                       if mp_dtype == "fp16":
                             SCREAMING_SNAKE_CASE		=						torch.floataa
                       elif mp_dtype == "bf16":
                             SCREAMING_SNAKE_CASE		=						torch.bfloataa
                       SCREAMING_SNAKE_CASE		=						MixedPrecision(param_dtype=lowerCamelCase__						,reduce_dtype=lowerCamelCase__						,buffer_dtype=lowerCamelCase__	)
                       self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy						,lowerCamelCase__	)
                       if mp_dtype == FPaa:
                             self.assertTrue(isinstance(accelerator.scaler						,lowerCamelCase__	)	)
                       elif mp_dtype == BFaa:
                             self.assertIsNone(accelerator.scaler	)
                       AcceleratorState._reset_state(lowerCamelCase__	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Dict	)					->		Optional[int]:




           '''simple docstring'''


           from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

           for flag in [True, False]:
                 SCREAMING_SNAKE_CASE		=						self.dist_env.copy()
                 SCREAMING_SNAKE_CASE		=						str(lowerCamelCase__	).lower()
                 with mockenv_context(**lowerCamelCase__	):
                       SCREAMING_SNAKE_CASE		=						FullyShardedDataParallelPlugin()
                       self.assertEqual(fsdp_plugin.cpu_offload						,CPUOffload(offload_params=lowerCamelCase__	)	)

@require_fsdp
@require_multi_gpu
@slow
class 		UpperCamelCase__			(				lowerCAmelCase_       ):



     '''simple docstring'''

     def     SCREAMING_SNAKE_CASE__  (							self						:   List[str]	)					->		Optional[Any]:




           '''simple docstring'''


           super().setUp()
           SCREAMING_SNAKE_CASE		=						0.82
           SCREAMING_SNAKE_CASE		=						[
               """fsdp_shard_grad_op_transformer_based_wrap""",
               """fsdp_full_shard_transformer_based_wrap""",
           ]
           SCREAMING_SNAKE_CASE		=						{
               """multi_gpu_fp16""": 3200,
               """fsdp_shard_grad_op_transformer_based_wrap_fp16""": 2000,
               """fsdp_full_shard_transformer_based_wrap_fp16""": 1900,
               # Disabling below test as it overwhelms the RAM memory usage
               # on CI self-hosted runner leading to tests getting killed.
               # "fsdp_full_shard_cpu_offload_transformer_based_wrap_fp32": 1500,  # fp16 was leading to indefinite hang
           }
           SCREAMING_SNAKE_CASE		=						160
           SCREAMING_SNAKE_CASE		=						160

           SCREAMING_SNAKE_CASE		=						inspect.getfile(accelerate.test_utils	)
           SCREAMING_SNAKE_CASE		=						os.path.sep.join(mod_file.split(os.path.sep	)[:-1] + ["""scripts""", """external_deps"""]	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Any	)					->		Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						os.path.join(self.test_scripts_folder						,"""test_performance.py"""	)
           SCREAMING_SNAKE_CASE		=						["""accelerate""", """launch""", """--num_processes=2""", """--num_machines=1""", """--machine_rank=0""", """--use_fsdp"""]
           for config in self.performance_configs:
                 SCREAMING_SNAKE_CASE		=						cmd.copy()
                 for i, strategy in enumerate(lowerCamelCase__	):
                       if strategy.lower() in config:
                             cmd_config.append(F"""--fsdp_sharding_strategy={i+1}"""	)
                             break

                 if "fp32" in config:
                       cmd_config.append("""--mixed_precision=no"""	)
                 else:
                       cmd_config.append("""--mixed_precision=fp16"""	)

                 if "cpu_offload" in config:
                       cmd_config.append("""--fsdp_offload_params=True"""	)

                 for policy in FSDP_AUTO_WRAP_POLICY:
                       if policy.lower() in config:
                             cmd_config.append(F"""--fsdp_auto_wrap_policy={policy}"""	)
                             break

                 if policy == "TRANSFORMER_BASED_WRAP":
                       cmd_config.append("""--fsdp_transformer_layer_cls_to_wrap=BertLayer"""	)
                 elif policy == "SIZE_BASED_WRAP":
                       cmd_config.append("""--fsdp_min_num_params=2000"""	)

                 cmd_config.extend(
                     [
                         self.test_file_path,
                         F"""--output_dir={self.tmpdir}""",
                         F"""--performance_lower_bound={self.performance_lower_bound}""",
                     ]	)
                 with patch_environment(omp_num_threads=1	):
                       execute_subprocess_async(lowerCamelCase__						,env=os.environ.copy()	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Union[str, Any]	)					->		str:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						os.path.join(self.test_scripts_folder						,"""test_checkpointing.py"""	)
           SCREAMING_SNAKE_CASE		=						[
               """accelerate""",
               """launch""",
               """--num_processes=2""",
               """--num_machines=1""",
               """--machine_rank=0""",
               """--use_fsdp""",
               """--mixed_precision=fp16""",
               """--fsdp_transformer_layer_cls_to_wrap=BertLayer""",
           ]

           for i, strategy in enumerate(lowerCamelCase__	):
                 SCREAMING_SNAKE_CASE		=						cmd.copy()
                 cmd_config.append(F"""--fsdp_sharding_strategy={i+1}"""	)
                 if strategy != "FULL_SHARD":
                       continue
                 SCREAMING_SNAKE_CASE		=						len(lowerCamelCase__	)
                 for state_dict_type in FSDP_STATE_DICT_TYPE:
                       SCREAMING_SNAKE_CASE		=						cmd_config[:state_dict_config_index]
                       cmd_config.append(F"""--fsdp_state_dict_type={state_dict_type}"""	)
                       cmd_config.extend(
                           [
                               self.test_file_path,
                               F"""--output_dir={self.tmpdir}""",
                               """--partial_train_epoch=1""",
                           ]	)
                       with patch_environment(omp_num_threads=1	):
                             execute_subprocess_async(lowerCamelCase__						,env=os.environ.copy()	)

                       SCREAMING_SNAKE_CASE		=						cmd_config[:-1]
                       SCREAMING_SNAKE_CASE		=						os.path.join(self.tmpdir						,"""epoch_0"""	)
                       cmd_config.extend(
                           [
                               F"""--resume_from_checkpoint={resume_from_checkpoint}""",
                           ]	)
                       with patch_environment(omp_num_threads=1	):
                             execute_subprocess_async(lowerCamelCase__						,env=os.environ.copy()	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Dict	)					->		str:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						os.path.join(self.test_scripts_folder						,"""test_peak_memory_usage.py"""	)
           SCREAMING_SNAKE_CASE		=						[
               """accelerate""",
               """launch""",
               """--num_processes=2""",
               """--num_machines=1""",
               """--machine_rank=0""",
           ]
           for spec, peak_mem_upper_bound in self.peak_memory_usage_upper_bound.items():
                 SCREAMING_SNAKE_CASE		=						cmd.copy()
                 if "fp16" in spec:
                       cmd_config.extend(["""--mixed_precision=fp16"""]	)
                 else:
                       cmd_config.extend(["""--mixed_precision=no"""]	)

                 if "multi_gpu" in spec:
                       continue
                 else:
                       cmd_config.extend(["""--use_fsdp"""]	)
                       for i, strategy in enumerate(lowerCamelCase__	):
                             if strategy.lower() in spec:
                                   cmd_config.append(F"""--fsdp_sharding_strategy={i+1}"""	)
                                   break

                       if "cpu_offload" in spec:
                             cmd_config.append("""--fsdp_offload_params=True"""	)

                       for policy in FSDP_AUTO_WRAP_POLICY:
                             if policy.lower() in spec:
                                   cmd_config.append(F"""--fsdp_auto_wrap_policy={policy}"""	)
                                   break

                       if policy == "TRANSFORMER_BASED_WRAP":
                             cmd_config.append("""--fsdp_transformer_layer_cls_to_wrap=BertLayer"""	)
                       elif policy == "SIZE_BASED_WRAP":
                             cmd_config.append("""--fsdp_min_num_params=2000"""	)

                 cmd_config.extend(
                     [
                         self.test_file_path,
                         F"""--output_dir={self.tmpdir}""",
                         F"""--peak_memory_upper_bound={peak_mem_upper_bound}""",
                         F"""--n_train={self.n_train}""",
                         F"""--n_val={self.n_val}""",
                     ]	)
                 with patch_environment(omp_num_threads=1	):
                       execute_subprocess_async(lowerCamelCase__						,env=os.environ.copy()	)









from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


SCREAMING_SNAKE_CASE_						       =	logging.get_logger(__name__)


SCREAMING_SNAKE_CASE_						       =	TypeVar("""DatasetType""", Dataset, IterableDataset)



def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = "first_exhausted"			,						)			->       DatasetType:


      '''simple docstring'''


      from .arrow_dataset import Dataset
      from .iterable_dataset import IterableDataset

      if not datasets:
            raise ValueError("""Unable to interleave an empty list of datasets."""     )
      for i, dataset in enumerate(_SCREAMING_SNAKE_CASE     ):
            if not isinstance(_SCREAMING_SNAKE_CASE			,						(Dataset, IterableDataset)     ):
                  if isinstance(_SCREAMING_SNAKE_CASE			,						(DatasetDict, IterableDatasetDict)     ):
                        if not dataset:
                              raise ValueError(
                                  F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
                                  """is an empty dataset dictionary."""     )
                        raise ValueError(
                            F"""Dataset at position {i} has at least one split: {list(_SCREAMING_SNAKE_CASE     )}\n"""
                            F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(_SCREAMING_SNAKE_CASE     )     )}']"""     )
                  raise ValueError(
                      F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(_SCREAMING_SNAKE_CASE     ).__name__}."""     )
            if i == 0:
                  SCREAMING_SNAKE_CASE, SCREAMING_SNAKE_CASE		=						(
                      (Dataset, IterableDataset) if isinstance(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     ) else (IterableDataset, Dataset)
                  )
            elif not isinstance(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     ):
                  raise ValueError(
                      F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""     )
      if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
            raise ValueError(F"""{stopping_strategy} is not supported. Please enter a valid stopping_strategy."""     )
      if dataset_type is Dataset:
            return _interleave_map_style_datasets(
                _SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						info=_SCREAMING_SNAKE_CASE			,						split=_SCREAMING_SNAKE_CASE			,						stopping_strategy=_SCREAMING_SNAKE_CASE     )
      else:
            return _interleave_iterable_datasets(
                _SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						info=_SCREAMING_SNAKE_CASE			,						split=_SCREAMING_SNAKE_CASE			,						stopping_strategy=_SCREAMING_SNAKE_CASE     )



def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = 0			,						)			->       DatasetType:


      '''simple docstring'''


      if not dsets:
            raise ValueError("""Unable to concatenate an empty list of datasets."""     )
      for i, dataset in enumerate(_SCREAMING_SNAKE_CASE     ):
            if not isinstance(_SCREAMING_SNAKE_CASE			,						(Dataset, IterableDataset)     ):
                  if isinstance(_SCREAMING_SNAKE_CASE			,						(DatasetDict, IterableDatasetDict)     ):
                        if not dataset:
                              raise ValueError(
                                  F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
                                  """is an empty dataset dictionary."""     )
                        raise ValueError(
                            F"""Dataset at position {i} has at least one split: {list(_SCREAMING_SNAKE_CASE     )}\n"""
                            F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(_SCREAMING_SNAKE_CASE     )     )}']"""     )
                  raise ValueError(
                      F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(_SCREAMING_SNAKE_CASE     ).__name__}."""     )
            if i == 0:
                  SCREAMING_SNAKE_CASE, SCREAMING_SNAKE_CASE		=						(
                      (Dataset, IterableDataset) if isinstance(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     ) else (IterableDataset, Dataset)
                  )
            elif not isinstance(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     ):
                  raise ValueError(
                      F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""     )
      if dataset_type is Dataset:
            return _concatenate_map_style_datasets(_SCREAMING_SNAKE_CASE			,						info=_SCREAMING_SNAKE_CASE			,						split=_SCREAMING_SNAKE_CASE			,						axis=_SCREAMING_SNAKE_CASE     )
      else:
            return _concatenate_iterable_datasets(_SCREAMING_SNAKE_CASE			,						info=_SCREAMING_SNAKE_CASE			,						split=_SCREAMING_SNAKE_CASE			,						axis=_SCREAMING_SNAKE_CASE     )


