

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)

_lowerCamelCase    :       List[str]        =							"bert-base-cased"
_lowerCamelCase    :       str        =							"fp16"
_lowerCamelCase    :       Optional[int]        =							"bf16"
_lowerCamelCase    :       List[str]        =							[FPaa, BFaa]





@require_fsdp
@require_cuda
class  __UpperCAmelCase					(	lowerCamelCase__						):



			def 	__magic_name__   (							self							:					Tuple						):
					super().setUp()

					UpperCAmelCase	:							Optional[Any]						=			dict(
					    ACCELERATE_USE_FSDP='''true''',		MASTER_ADDR='''localhost''',		MASTER_PORT='''10999''',		RANK='''0''',		LOCAL_RANK='''0''',		WORLD_SIZE='''1''',		)



			def 	__magic_name__   (							self							:					int						):
					from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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



			def 	__magic_name__   (							self							:					Dict						):
					from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

					for i, prefetch_policy in enumerate(__A						):
							UpperCAmelCase	:							int						=			self.dist_env.copy()
							UpperCAmelCase	:							Dict						=			prefetch_policy
							with mockenv_context(**__A						):
									UpperCAmelCase	:							Optional[int]						=			FullyShardedDataParallelPlugin()
									if prefetch_policy == "NO_PREFETCH":
											self.assertIsNone(fsdp_plugin.backward_prefetch						)
									else:
											self.assertEqual(fsdp_plugin.backward_prefetch,		BackwardPrefetch(i + 1						)						)



			def 	__magic_name__   (							self							:					Any						):
					from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

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

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

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



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





			def 	__magic_name__   (							self							:					Optional[int]						):
					from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

					for flag in [True, False]:
							UpperCAmelCase	:							Any						=			self.dist_env.copy()
							UpperCAmelCase	:							int						=			str(__A						).lower()
							with mockenv_context(**__A						):
									UpperCAmelCase	:							Union[str, Any]						=			FullyShardedDataParallelPlugin()
									self.assertEqual(fsdp_plugin.cpu_offload,		CPUOffload(offload_params=__A						)						)






@require_fsdp
@require_multi_gpu
@slow
class  __UpperCAmelCase					(	lowerCamelCase__						):



			def 	__magic_name__   (							self							:					List[Any]						):
					super().setUp()
					UpperCAmelCase	:							int						=			0.8_2
					UpperCAmelCase	:							List[str]						=			[
					    '''fsdp_shard_grad_op_transformer_based_wrap''',
					    '''fsdp_full_shard_transformer_based_wrap''',
					]
					UpperCAmelCase	:							int						=			{
					    '''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
					}
					UpperCAmelCase	:							List[str]						=			1_6_0
					UpperCAmelCase	:							Optional[int]						=			1_6_0

					UpperCAmelCase	:							Union[str, Any]						=			inspect.getfile(accelerate.test_utils						)
					UpperCAmelCase	:							Dict						=			os.path.sep.join(mod_file.split(os.path.sep						)[:-1] + ['''scripts''', '''external_deps''']						)



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



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

									UpperCAmelCase	:							Tuple						=			cmd_config[:-1]
									UpperCAmelCase	:							List[str]						=			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(__A,		env=os.environ.copy()						)





			def 	__magic_name__   (							self							:					Any						):
					UpperCAmelCase	:							Optional[Any]						=			os.path.join(self.test_scripts_folder,		'''test_peak_memory_usage.py'''						)
					UpperCAmelCase	:							str						=			[
					    '''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	:							List[Any]						=			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(__A						):
											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(__A,		env=os.environ.copy()						)








def       a__					(		UpperCAmelCase							:      int      )  ->						bool:
  if not isinstance(UpperCAmelCase    ,			UpperCAmelCase      ):
    UpperCAmelCase	:							List[str]						=			f'''Input value of [number={number}] must be an integer'''
    raise TypeError(UpperCAmelCase      )
  if number < 0:
    return False
  UpperCAmelCase	:							List[str]						=			number * number
  while number > 0:
    if number % 10 != number_square % 10:
      return False
    number //= 10
    number_square //= 10
  return True


if __name__ == "__main__":
     import doctest

     doctest.testmod()





