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

lowerCAmelCase_	:		Tuple			=     """bert-base-cased"""
lowerCAmelCase_	:		Union[str, Any]			=     """fp16"""
lowerCAmelCase_	:		Optional[Any]			=     """bf16"""
lowerCAmelCase_	:		str			=     [FPaa, BFaa]




@require_fsdp
@require_cuda
class 		SCREAMING_SNAKE_CASE     (					SCREAMING_SNAKE_CASE	):







				'''simple docstring'''





				def  snake_case__				(						self					:						Optional[Any]					)							->Tuple:


										'''simple docstring'''




										super().setUp()

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





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


										'''simple docstring'''




										from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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





				def  snake_case__				(						self					:						Optional[Any]					)							->List[Any]:


										'''simple docstring'''




										from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

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





				def  snake_case__				(						self					:						Tuple					)							->Tuple:


										'''simple docstring'''




										from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

										for i, state_dict_type in enumerate(lowercase__					):
																_UpperCamelCase			:			Any   			=						self.dist_env.copy()
																_UpperCamelCase			:			Any   			=						state_dict_type
																with mockenv_context(**lowercase__					):
																						_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  snake_case__				(						self					:						Optional[int]					)							->int:


										'''simple docstring'''




										_UpperCamelCase			:			str   			=						AutoModel.from_pretrained(lowercase__					)
										for policy in FSDP_AUTO_WRAP_POLICY:
																_UpperCamelCase			:			Tuple   			=						self.dist_env.copy()
																_UpperCamelCase			:			Optional[int]   			=						policy
																if policy == "TRANSFORMER_BASED_WRAP":
																						_UpperCamelCase			:			Tuple   			=						"BertLayer"
																elif policy == "SIZE_BASED_WRAP":
																						_UpperCamelCase			:			str   			=						"2000"
																with mockenv_context(**lowercase__					):
																						_UpperCamelCase			:			int   			=						FullyShardedDataParallelPlugin()
																						fsdp_plugin.set_auto_wrap_policy(lowercase__					)
																						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			:			Tuple   			=						"TRANSFORMER_BASED_WRAP"
										_UpperCamelCase			:			List[str]   			=						"T5Layer"
										with mockenv_context(**lowercase__					):
																_UpperCamelCase			:			Optional[Any]   			=						FullyShardedDataParallelPlugin()
																with self.assertRaises(lowercase__					) as cm:
																						fsdp_plugin.set_auto_wrap_policy(lowercase__					)
																self.assertTrue("Could not find the transformer layer class to wrap in the model." in str(cm.exception					)					)

										_UpperCamelCase			:			str   			=						self.dist_env.copy()
										_UpperCamelCase			:			Tuple   			=						"SIZE_BASED_WRAP"
										_UpperCamelCase			:			Any   			=						"0"
										with mockenv_context(**lowercase__					):
																_UpperCamelCase			:			str   			=						FullyShardedDataParallelPlugin()
																fsdp_plugin.set_auto_wrap_policy(lowercase__					)
																self.assertIsNone(fsdp_plugin.auto_wrap_policy					)





				def  snake_case__				(						self					:						Optional[int]					)							->Any:


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







				def  snake_case__				(						self					:						Any					)							->List[Any]:


										'''simple docstring'''




										from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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





@require_fsdp
@require_multi_gpu
@slow
class 		SCREAMING_SNAKE_CASE     (					SCREAMING_SNAKE_CASE	):







				'''simple docstring'''





				def  snake_case__				(						self					:						Any					)							->Any:


										'''simple docstring'''




										super().setUp()
										_UpperCamelCase			:			Any   			=						0.8_2
										_UpperCamelCase			:			Tuple   			=						[
										    "fsdp_shard_grad_op_transformer_based_wrap",
										    "fsdp_full_shard_transformer_based_wrap",
										]
										_UpperCamelCase			:			Optional[int]   			=						{
										    "multi_gpu_fp16": 3_200,
										    "fsdp_shard_grad_op_transformer_based_wrap_fp16": 2_000,
										    "fsdp_full_shard_transformer_based_wrap_fp16": 1_900,
										    # 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			:			int   			=						160
										_UpperCamelCase			:			Union[str, Any]   			=						160

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





				def  snake_case__				(						self					:						List[Any]					)							->Tuple:


										'''simple docstring'''




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





				def  snake_case__				(						self					:						int					)							->Any:


										'''simple docstring'''




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

																						_UpperCamelCase			:			int   			=						cmd_config[:-1]
																						_UpperCamelCase			:			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(lowercase__       ,		env=os.environ.copy()					)







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


										'''simple docstring'''




										_UpperCamelCase			:			Optional[int]   			=						os.path.join(self.test_scripts_folder       ,		"test_peak_memory_usage.py"					)
										_UpperCamelCase			:			Tuple   			=						[
										    "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(lowercase__					):
																												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(lowercase__       ,		env=os.environ.copy()					)



'''simple docstring'''





from __future__ import annotations

from cmath import sqrt
def 						__A							(     UpperCAmelCase     ,UpperCAmelCase     ,UpperCAmelCase					)     ->			tuple[complex, complex]:




						'''simple docstring'''

						if a == 0:
												raise ValueError("Coefficient 'a' must not be zero."					)
						_UpperCamelCase			:			List[str]   			=						b * b - 4 * a * c

						_UpperCamelCase			:			Dict   			=						(-b + sqrt(UpperCAmelCase					)) / (2 * a)
						_UpperCamelCase			:			Optional[int]   			=						(-b - sqrt(UpperCAmelCase					)) / (2 * a)

						return (
						    root_a.real if not root_a.imag else root_a,
						    root_a.real if not root_a.imag else root_a,
						)
def 						__A							(     )     ->			List[Any]:




						'''simple docstring'''
						_UpperCamelCase			,       _UpperCamelCase			:			Any   			=						quadratic_roots(a=5     ,b=6     ,c=1					)
						print(f'''The solutions are: {solutiona} and {solutiona}'''					)


if __name__ == "__main__":
			main()


