


"""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)

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



@require_fsdp
@require_cuda
class 			_UpperCAmelCase (							lowercase__				):






		'''simple docstring'''

		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			super().setUp()

			__snake_case	:						int						=				dict(
			    ACCELERATE_USE_FSDP='''true'''				,       MASTER_ADDR='''localhost'''				,       MASTER_PORT='''10999'''				,       RANK='''0'''				,       LOCAL_RANK='''0'''				,       WORLD_SIZE='''1'''				,       )



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

			for i, strategy in enumerate(_a					):
				__snake_case	:						Union[str, Any]						=				self.dist_env.copy()
				__snake_case	:						Optional[Any]						=				f"""{i + 1}"""
				__snake_case	:						str						=				strategy
				with mockenv_context(**_a					):
					__snake_case	:						Optional[int]						=				FullyShardedDataParallelPlugin()
					self.assertEqual(fsdp_plugin.sharding_strategy				,       ShardingStrategy(i + 1					)					)



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

			for i, prefetch_policy in enumerate(_a					):
				__snake_case	:						Tuple						=				self.dist_env.copy()
				__snake_case	:						List[str]						=				prefetch_policy
				with mockenv_context(**_a					):
					__snake_case	:						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 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

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





			'''simple docstring'''

			__snake_case	:						Dict						=				AutoModel.from_pretrained(_a					)
			for policy in FSDP_AUTO_WRAP_POLICY:
				__snake_case	:						Optional[int]						=				self.dist_env.copy()
				__snake_case	:						Union[str, Any]						=				policy
				if policy == "TRANSFORMER_BASED_WRAP":
					__snake_case	:						int						=				'BertLayer'
				elif policy == "SIZE_BASED_WRAP":
					__snake_case	:						Optional[int]						=				'2000'
				with mockenv_context(**_a					):
					__snake_case	:						Union[str, 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					)

			__snake_case	:						Tuple						=				self.dist_env.copy()
			__snake_case	:						List[str]						=				'TRANSFORMER_BASED_WRAP'
			__snake_case	:						Optional[Any]						=				'T5Layer'
			with mockenv_context(**_a					):
				__snake_case	:						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					)					)

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



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''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:
				__snake_case	:						Any						=				self.dist_env.copy()
				__snake_case	:						List[str]						=				mp_dtype
				with mockenv_context(**_a					):
					__snake_case	:						Dict						=				Accelerator()
					if mp_dtype == "fp16":
						__snake_case	:						List[str]						=				torch.floataa
					elif mp_dtype == "bf16":
						__snake_case	:						Any						=				torch.bfloataa
					__snake_case	:						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 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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



@require_fsdp
@require_multi_gpu
@slow
class 			_UpperCAmelCase (							lowercase__				):






		'''simple docstring'''

		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			super().setUp()
			__snake_case	:						int						=				0.82
			__snake_case	:						Union[str, Any]						=				[
			    'fsdp_shard_grad_op_transformer_based_wrap',
			    'fsdp_full_shard_transformer_based_wrap',
			]
			__snake_case	:						Dict						=				{
			    '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
			}
			__snake_case	:						Tuple						=				1_60
			__snake_case	:						Dict						=				1_60

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



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						List[str]						=				os.path.join(self.test_scripts_folder				,       '''test_performance.py'''					)
			__snake_case	:						Any						=				['accelerate', 'launch', '--num_processes=2', '--num_machines=1', '--machine_rank=0', '--use_fsdp']
			for config in self.performance_configs:
				__snake_case	:						int						=				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 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

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

					__snake_case	:						Optional[int]						=				cmd_config[:-1]
					__snake_case	:						Tuple						=				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 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						int						=				os.path.join(self.test_scripts_folder				,       '''test_peak_memory_usage.py'''					)
			__snake_case	:						Optional[int]						=				[
			    '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():
				__snake_case	:						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()					)







'''simple docstring'''







def 					UpperCAmelCase_		(__a      :    list ,	__a      :    list ,	__a      :    int ):






   """simple docstring"""

   _a				:	Optional[Any]			=   len(__a )
   _a				:	int			=   [[0] * n for i in range(__a )]
   for i in range(__a ):
      _a				:	Tuple			=   y_points[i]

   for i in range(2 ,	__a ):
      for j in range(__a ,	__a ):
         _a				:	Tuple			=   (
             (xa - x_points[j - i + 1]) * q[j][i - 1]
             - (xa - x_points[j]) * q[j - 1][i - 1]
         ) / (x_points[j] - x_points[j - i + 1])

   return [q[n - 1][n - 1], q]


if __name__ == "__main__":
  import doctest

  doctest.testmod()



