




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(42)

UpperCAmelCase     :     List[str]							=	"""bert-base-cased"""
UpperCAmelCase     :     Dict							=	"""fp16"""
UpperCAmelCase     :     Any							=	"""bf16"""
UpperCAmelCase     :     Any							=	[FPaa, BFaa]







@require_fsdp
@require_cuda
class     __lowerCAmelCase	(   SCREAMING_SNAKE_CASE_):







      def        _lowercase  (   self							)       ->						Optional[int]:



        '''simple docstring'''




        super().setUp()

        a__     : List[str]  =dict(
            ACCELERATE_USE_FSDP="true"		,				MASTER_ADDR="localhost"		,				MASTER_PORT="10999"		,				RANK="0"		,				LOCAL_RANK="0"		,				WORLD_SIZE="1"		,				)







      def        _lowercase  (   self							)       ->						Optional[int]:



        '''simple docstring'''




        from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

        for i, strategy in enumerate(snake_case__							):
          a__     : List[str]  =self.dist_env.copy()
          a__     : Optional[Any]  =F'''{i + 1}'''
          a__     : Optional[Any]  =strategy
          with mockenv_context(**snake_case__							):
            a__     : Union[str, Any]  =FullyShardedDataParallelPlugin()
            self.assertEqual(fsdp_plugin.sharding_strategy		,				ShardingStrategy(i + 1							)							)







      def        _lowercase  (   self							)       ->						int:



        '''simple docstring'''




        from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

        for i, prefetch_policy in enumerate(snake_case__							):
          a__     : str  =self.dist_env.copy()
          a__     : str  =prefetch_policy
          with mockenv_context(**snake_case__							):
            a__     : List[Any]  =FullyShardedDataParallelPlugin()
            if prefetch_policy == "NO_PREFETCH":
              self.assertIsNone(fsdp_plugin.backward_prefetch							)
            else:
              self.assertEqual(fsdp_plugin.backward_prefetch		,				BackwardPrefetch(i + 1							)							)







      def        _lowercase  (   self							)       ->						List[str]:



        '''simple docstring'''




        from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

        for i, state_dict_type in enumerate(snake_case__							):
          a__     : List[str]  =self.dist_env.copy()
          a__     : str  =state_dict_type
          with mockenv_context(**snake_case__							):
            a__     : str  =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        _lowercase  (   self							)       ->						List[str]:



        '''simple docstring'''




        a__     : Union[str, Any]  =AutoModel.from_pretrained(snake_case__							)
        for policy in FSDP_AUTO_WRAP_POLICY:
          a__     : List[str]  =self.dist_env.copy()
          a__     : int  =policy
          if policy == "TRANSFORMER_BASED_WRAP":
            a__     : str  ='BertLayer'
          elif policy == "SIZE_BASED_WRAP":
            a__     : List[Any]  ='2000'
          with mockenv_context(**snake_case__							):
            a__     : Optional[Any]  =FullyShardedDataParallelPlugin()
            fsdp_plugin.set_auto_wrap_policy(snake_case__							)
            if policy == "NO_WRAP":
              self.assertIsNone(fsdp_plugin.auto_wrap_policy							)
            else:
              self.assertIsNotNone(fsdp_plugin.auto_wrap_policy							)

        a__     : Any  =self.dist_env.copy()
        a__     : str  ='TRANSFORMER_BASED_WRAP'
        a__     : Optional[Any]  ='T5Layer'
        with mockenv_context(**snake_case__							):
          a__     : int  =FullyShardedDataParallelPlugin()
          with self.assertRaises(snake_case__							) as cm:
            fsdp_plugin.set_auto_wrap_policy(snake_case__							)
          self.assertTrue("Could not find the transformer layer class to wrap in the model." in str(cm.exception							)							)

        a__     : List[Any]  =self.dist_env.copy()
        a__     : Optional[int]  ='SIZE_BASED_WRAP'
        a__     : List[str]  ='0'
        with mockenv_context(**snake_case__							):
          a__     : Tuple  =FullyShardedDataParallelPlugin()
          fsdp_plugin.set_auto_wrap_policy(snake_case__							)
          self.assertIsNone(fsdp_plugin.auto_wrap_policy							)







      def        _lowercase  (   self							)       ->						List[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:
          a__     : Dict  =self.dist_env.copy()
          a__     : int  =mp_dtype
          with mockenv_context(**snake_case__							):
            a__     : List[str]  =Accelerator()
            if mp_dtype == "fp16":
              a__     : Any  =torch.floataa
            elif mp_dtype == "bf16":
              a__     : List[Any]  =torch.bfloataa
            a__     : Tuple  =MixedPrecision(param_dtype=snake_case__		,				reduce_dtype=snake_case__		,				buffer_dtype=snake_case__							)
            self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy		,				snake_case__							)
            if mp_dtype == FPaa:
              self.assertTrue(isinstance(accelerator.scaler		,				snake_case__							)							)
            elif mp_dtype == BFaa:
              self.assertIsNone(accelerator.scaler							)
            AcceleratorState._reset_state(snake_case__							)







      def        _lowercase  (   self							)       ->						Dict:



        '''simple docstring'''




        from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

        for flag in [True, False]:
          a__     : List[Any]  =self.dist_env.copy()
          a__     : Dict  =str(snake_case__							).lower()
          with mockenv_context(**snake_case__							):
            a__     : List[Any]  =FullyShardedDataParallelPlugin()
            self.assertEqual(fsdp_plugin.cpu_offload		,				CPUOffload(offload_params=snake_case__							)							)







@require_fsdp
@require_multi_gpu
@slow
class     __lowerCAmelCase	(   SCREAMING_SNAKE_CASE_):







      def        _lowercase  (   self							)       ->						Any:



        '''simple docstring'''




        super().setUp()
        a__     : List[str]  =0.82
        a__     : int  =[
            'fsdp_shard_grad_op_transformer_based_wrap',
            'fsdp_full_shard_transformer_based_wrap',
        ]
        a__     : Any  ={
            'multi_gpu_fp16': 3_2_0_0,
            'fsdp_shard_grad_op_transformer_based_wrap_fp16': 2_0_0_0,
            'fsdp_full_shard_transformer_based_wrap_fp16': 1_9_0_0,
            # 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
        }
        a__     : str  =1_6_0
        a__     : Any  =1_6_0

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







      def        _lowercase  (   self							)       ->						List[str]:



        '''simple docstring'''




        a__     : Dict  =os.path.join(self.test_scripts_folder		,				"test_performance.py"							)
        a__     : Dict  =['accelerate', 'launch', '--num_processes=2', '--num_machines=1', '--machine_rank=0', '--use_fsdp']
        for config in self.performance_configs:
          a__     : List[Any]  =cmd.copy()
          for i, strategy in enumerate(snake_case__							):
            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(snake_case__		,				env=os.environ.copy()							)







      def        _lowercase  (   self							)       ->						int:



        '''simple docstring'''




        a__     : Tuple  =os.path.join(self.test_scripts_folder		,				"test_checkpointing.py"							)
        a__     : Any  =[
            '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(snake_case__							):
          a__     : List[str]  =cmd.copy()
          cmd_config.append(F'''--fsdp_sharding_strategy={i+1}'''							)
          if strategy != "FULL_SHARD":
            continue
          a__     : Any  =len(snake_case__							)
          for state_dict_type in FSDP_STATE_DICT_TYPE:
            a__     : List[Any]  =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(snake_case__		,				env=os.environ.copy()							)

            a__     : Any  =cmd_config[:-1]
            a__     : Optional[Any]  =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(snake_case__		,				env=os.environ.copy()							)







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



        '''simple docstring'''




        a__     : int  =os.path.join(self.test_scripts_folder		,				"test_peak_memory_usage.py"							)
        a__     : Dict  =[
            '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():
          a__     : str  =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(snake_case__							):
              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(snake_case__		,				env=os.environ.copy()							)













import os
def 					_A	(		):





  """simple docstring"""



  a__     : Optional[int]  =os.path.join(os.path.dirname(SCREAMING_SNAKE_CASE						)	,	"num.txt"						)
  with open(SCREAMING_SNAKE_CASE						) as file_hand:
    return str(sum(int(SCREAMING_SNAKE_CASE						) for line in file_hand						)						)[:10]


if __name__ == "__main__":
     print(solution())






