





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						:   Optional[int]			  =     'bert-base-cased'
UpperCAmelCase						:   Union[str, Any]			  =     'fp16'
UpperCAmelCase						:   Dict			  =     'bf16'
UpperCAmelCase						:   Tuple			  =     [FPaa, BFaa]





@require_fsdp
@require_cuda
class 						_A(					_lowerCamelCase   ):


				"""simple docstring"""


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

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

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

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





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

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





@require_fsdp
@require_multi_gpu
@slow
class 						_A(					_lowerCamelCase   ):


				"""simple docstring"""


				def 					UpperCAmelCase_		(			self     ):
									super().setUp()
									__A	:						Tuple       =				0.8_2
									__A	:						Any       =				[
									    'fsdp_shard_grad_op_transformer_based_wrap',
									    'fsdp_full_shard_transformer_based_wrap',
									]
									__A	:						List[str]       =				{
									    '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
									}
									__A	:						Any       =				160
									__A	:						int       =				160

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



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



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

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





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











def _a				(	lowerCamelCase:   dict )	->						bool:







      '''simple docstring'''






      __A          =				set()
      # To detect a back edge, keep track of vertices currently in the recursion stack
      __A          =				set()
      return any(
          node not in visited and depth_first_search(lowerCamelCase					,				lowerCamelCase					,				lowerCamelCase					,				lowerCamelCase )
          for node in graph )



def _a				(	lowerCamelCase:   dict					,				lowerCamelCase:   int					,				lowerCamelCase:   set					,				lowerCamelCase:   set )	->						bool:







      '''simple docstring'''






      visited.add(lowerCamelCase )
      rec_stk.add(lowerCamelCase )

      for node in graph[vertex]:
            if node not in visited:
                  if depth_first_search(lowerCamelCase					,				lowerCamelCase					,				lowerCamelCase					,				lowerCamelCase ):
                        return True
            elif node in rec_stk:
                  return True

    # The node needs to be removed from recursion stack before function ends
      rec_stk.remove(lowerCamelCase )
      return False


if __name__ == "__main__":
       from doctest import testmod

       testmod()





