






'''simple docstring'''

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      					="bert-base-cased"
UpperCamelCase      					="fp16"
UpperCamelCase      					="bf16"
UpperCamelCase      					=[FPaa, BFaa]




@require_fsdp
@require_cuda
class      A     (   __lowercase    ):






 """simple docstring"""





 def 	_UpperCAmelCase							(				self			):
       super().setUp()

       UpperCamelCase_					: Tuple            =				dict(
           ACCELERATE_USE_FSDP="""true"""      ,  MASTER_ADDR="""localhost"""      ,  MASTER_PORT="""10999"""      ,  RANK="""0"""      ,  LOCAL_RANK="""0"""      ,  WORLD_SIZE="""1"""      ,  )
 def 	_UpperCAmelCase							(				self			):
       from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

       for i, strategy in enumerate(SCREAMING_SNAKE_CASE_			):
             UpperCamelCase_					: Tuple            =				self.dist_env.copy()
             UpperCamelCase_					: Optional[int]            =				F"{i + 1}"
             UpperCamelCase_					: Dict            =				strategy
             with mockenv_context(**SCREAMING_SNAKE_CASE_			):
                   UpperCamelCase_					: Optional[Any]            =				FullyShardedDataParallelPlugin()
                   self.assertEqual(fsdp_plugin.sharding_strategy      ,  ShardingStrategy(i + 1			)			)
 def 	_UpperCAmelCase							(				self			):
       from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

       for i, prefetch_policy in enumerate(SCREAMING_SNAKE_CASE_			):
             UpperCamelCase_					: Union[str, Any]            =				self.dist_env.copy()
             UpperCamelCase_					: Optional[Any]            =				prefetch_policy
             with mockenv_context(**SCREAMING_SNAKE_CASE_			):
                   UpperCamelCase_					: List[str]            =				FullyShardedDataParallelPlugin()
                   if prefetch_policy == "NO_PREFETCH":
                         self.assertIsNone(fsdp_plugin.backward_prefetch			)
                   else:
                         self.assertEqual(fsdp_plugin.backward_prefetch      ,  BackwardPrefetch(i + 1			)			)
 def 	_UpperCAmelCase							(				self			):
       from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

       for i, state_dict_type in enumerate(SCREAMING_SNAKE_CASE_			):
             UpperCamelCase_					: str            =				self.dist_env.copy()
             UpperCamelCase_					: Optional[Any]            =				state_dict_type
             with mockenv_context(**SCREAMING_SNAKE_CASE_			):
                   UpperCamelCase_					: Optional[int]            =				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 	_UpperCAmelCase							(				self			):
       UpperCamelCase_					: Any            =				AutoModel.from_pretrained(SCREAMING_SNAKE_CASE_			)
       for policy in FSDP_AUTO_WRAP_POLICY:
             UpperCamelCase_					: str            =				self.dist_env.copy()
             UpperCamelCase_					: Tuple            =				policy
             if policy == "TRANSFORMER_BASED_WRAP":
                   UpperCamelCase_					: Optional[Any]            =				"""BertLayer"""
             elif policy == "SIZE_BASED_WRAP":
                   UpperCamelCase_					: str            =				"""2000"""
             with mockenv_context(**SCREAMING_SNAKE_CASE_			):
                   UpperCamelCase_					: int            =				FullyShardedDataParallelPlugin()
                   fsdp_plugin.set_auto_wrap_policy(SCREAMING_SNAKE_CASE_			)
                   if policy == "NO_WRAP":
                         self.assertIsNone(fsdp_plugin.auto_wrap_policy			)
                   else:
                         self.assertIsNotNone(fsdp_plugin.auto_wrap_policy			)

       UpperCamelCase_					: str            =				self.dist_env.copy()
       UpperCamelCase_					: str            =				"""TRANSFORMER_BASED_WRAP"""
       UpperCamelCase_					: Dict            =				"""T5Layer"""
       with mockenv_context(**SCREAMING_SNAKE_CASE_			):
             UpperCamelCase_					: Union[str, Any]            =				FullyShardedDataParallelPlugin()
             with self.assertRaises(SCREAMING_SNAKE_CASE_			) as cm:
                   fsdp_plugin.set_auto_wrap_policy(SCREAMING_SNAKE_CASE_			)
             self.assertTrue("""Could not find the transformer layer class to wrap in the model.""" in str(cm.exception			)			)

       UpperCamelCase_					: List[str]            =				self.dist_env.copy()
       UpperCamelCase_					: List[str]            =				"""SIZE_BASED_WRAP"""
       UpperCamelCase_					: Optional[Any]            =				"""0"""
       with mockenv_context(**SCREAMING_SNAKE_CASE_			):
             UpperCamelCase_					: Optional[Any]            =				FullyShardedDataParallelPlugin()
             fsdp_plugin.set_auto_wrap_policy(SCREAMING_SNAKE_CASE_			)
             self.assertIsNone(fsdp_plugin.auto_wrap_policy			)
 def 	_UpperCAmelCase							(				self			):
       from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
       from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

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





 def 	_UpperCAmelCase							(				self			):
       from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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






@require_fsdp
@require_multi_gpu
@slow
class      A     (   __lowercase    ):






 """simple docstring"""





 def 	_UpperCAmelCase							(				self			):
       super().setUp()
       UpperCamelCase_					: Tuple            =				0.82
       UpperCamelCase_					: Optional[int]            =				[
           """fsdp_shard_grad_op_transformer_based_wrap""",
           """fsdp_full_shard_transformer_based_wrap""",
       ]
       UpperCamelCase_					: Union[str, Any]            =				{
           """multi_gpu_fp16""": 32_00,
           """fsdp_shard_grad_op_transformer_based_wrap_fp16""": 20_00,
           """fsdp_full_shard_transformer_based_wrap_fp16""": 19_00,
           # 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
       }
       UpperCamelCase_					: Optional[Any]            =				1_60
       UpperCamelCase_					: str            =				1_60

       UpperCamelCase_					: List[Any]            =				inspect.getfile(accelerate.test_utils			)
       UpperCamelCase_					: str            =				os.path.sep.join(mod_file.split(os.path.sep			)[:-1] + ["""scripts""", """external_deps"""]			)
 def 	_UpperCAmelCase							(				self			):
       UpperCamelCase_					: Dict            =				os.path.join(self.test_scripts_folder      ,  """test_performance.py"""			)
       UpperCamelCase_					: Dict            =				["""accelerate""", """launch""", """--num_processes=2""", """--num_machines=1""", """--machine_rank=0""", """--use_fsdp"""]
       for config in self.performance_configs:
             UpperCamelCase_					: int            =				cmd.copy()
             for i, strategy in enumerate(SCREAMING_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(SCREAMING_SNAKE_CASE_      ,  env=os.environ.copy()			)
 def 	_UpperCAmelCase							(				self			):
       UpperCamelCase_					: List[Any]            =				os.path.join(self.test_scripts_folder      ,  """test_checkpointing.py"""			)
       UpperCamelCase_					: Union[str, 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(SCREAMING_SNAKE_CASE_			):
             UpperCamelCase_					: Any            =				cmd.copy()
             cmd_config.append(F"--fsdp_sharding_strategy={i+1}"			)
             if strategy != "FULL_SHARD":
                   continue
             UpperCamelCase_					: Union[str, Any]            =				len(SCREAMING_SNAKE_CASE_			)
             for state_dict_type in FSDP_STATE_DICT_TYPE:
                   UpperCamelCase_					: Tuple            =				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(SCREAMING_SNAKE_CASE_      ,  env=os.environ.copy()			)

                   UpperCamelCase_					: List[Any]            =				cmd_config[:-1]
                   UpperCamelCase_					: Dict            =				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(SCREAMING_SNAKE_CASE_      ,  env=os.environ.copy()			)





 def 	_UpperCAmelCase							(				self			):
       UpperCamelCase_					: Dict            =				os.path.join(self.test_scripts_folder      ,  """test_peak_memory_usage.py"""			)
       UpperCamelCase_					: Union[str, Any]            =				[
           """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():
             UpperCamelCase_					: Tuple            =				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(SCREAMING_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(SCREAMING_SNAKE_CASE_      ,  env=os.environ.copy()			)








"""simple docstring"""




# Copyright 2021 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 argparse
import os

from accelerate.test_utils import execute_subprocess_async



def 							_lowerCamelCase			(			__a=None     ):
		if subparsers is not None:
				SCREAMING_SNAKE_CASE_    			=			subparsers.add_parser('''test'''     )
		else:
				SCREAMING_SNAKE_CASE_    			=			argparse.ArgumentParser('''Accelerate test command'''     )

		parser.add_argument(
		    '''--config_file''',				default=__a,				help=(
		        '''The path to use to store the config file. Will default to a file named default_config.yaml in the cache '''
		        '''location, which is the content of the environment `HF_HOME` suffixed with \'accelerate\', or if you don\'t have '''
		        '''such an environment variable, your cache directory (\'~/.cache\' or the content of `XDG_CACHE_HOME`) suffixed '''
		        '''with \'huggingface\'.'''
		    ),				)

		if subparsers is not None:
				parser.set_defaults(func=__a     )
		return parser



def 							_lowerCamelCase			(			__a     ):
		SCREAMING_SNAKE_CASE_    			=			os.path.sep.join(__file__.split(os.path.sep     )[:-2] + ['''test_utils''', '''scripts''', '''test_script.py''']     )

		if args.config_file is None:
				SCREAMING_SNAKE_CASE_    			=			script_name
		else:
				SCREAMING_SNAKE_CASE_    			=			F'--config_file={args.config_file} {script_name}'

		SCREAMING_SNAKE_CASE_    			=			['''accelerate-launch'''] + test_args.split()
		SCREAMING_SNAKE_CASE_    			=			execute_subprocess_async(__a,				env=os.environ.copy()     )
		if result.returncode == 0:
				print('''Test is a success! You are ready for your distributed training!'''     )



def 							_lowerCamelCase			(			):
		SCREAMING_SNAKE_CASE_    			=			test_command_parser()
		SCREAMING_SNAKE_CASE_    			=			parser.parse_args()
		test_command(__a     )


if __name__ == "__main__":
						main()