


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


@require_fsdp
@require_cuda
class  SCREAMING_SNAKE_CASE__							(						__snake_case							):
       """simple docstring"""


       def 			lowercase__				(		self		):






         """simple docstring"""





         super().setUp()

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

       def 			lowercase__				(		self		):






         """simple docstring"""





         from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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

       def 			lowercase__				(		self		):






         """simple docstring"""





         from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

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

       def 			lowercase__				(		self		):






         """simple docstring"""





         from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

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






         """simple docstring"""





         lowerCAmelCase	:	int								=				AutoModel.from_pretrained(a__		)
         for policy in FSDP_AUTO_WRAP_POLICY:
           lowerCAmelCase	:	Tuple								=				self.dist_env.copy()
           lowerCAmelCase	:	Optional[int]								=				policy
           if policy == "TRANSFORMER_BASED_WRAP":
             lowerCAmelCase	:	Optional[Any]								=				"BertLayer"
           elif policy == "SIZE_BASED_WRAP":
             lowerCAmelCase	:	List[str]								=				"2000"
           with mockenv_context(**a__		):
             lowerCAmelCase	:	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		)

         lowerCAmelCase	:	List[str]								=				self.dist_env.copy()
         lowerCAmelCase	:	Any								=				"TRANSFORMER_BASED_WRAP"
         lowerCAmelCase	:	List[str]								=				"T5Layer"
         with mockenv_context(**a__		):
           lowerCAmelCase	:	List[Any]								=				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		)		)

         lowerCAmelCase	:	int								=				self.dist_env.copy()
         lowerCAmelCase	:	str								=				"SIZE_BASED_WRAP"
         lowerCAmelCase	:	Tuple								=				"0"
         with mockenv_context(**a__		):
           lowerCAmelCase	:	Union[str, Any]								=				FullyShardedDataParallelPlugin()
           fsdp_plugin.set_auto_wrap_policy(a__		)
           self.assertIsNone(fsdp_plugin.auto_wrap_policy		)

       def 			lowercase__				(		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:
           lowerCAmelCase	:	str								=				self.dist_env.copy()
           lowerCAmelCase	:	Union[str, Any]								=				mp_dtype
           with mockenv_context(**a__		):
             lowerCAmelCase	:	List[str]								=				Accelerator()
             if mp_dtype == "fp16":
               lowerCAmelCase	:	Dict								=				torch.floataa
             elif mp_dtype == "bf16":
               lowerCAmelCase	:	Optional[Any]								=				torch.bfloataa
             lowerCAmelCase	:	List[str]								=				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 			lowercase__				(		self		):






         """simple docstring"""





         from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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


@require_fsdp
@require_multi_gpu
@slow
class  SCREAMING_SNAKE_CASE__							(						__snake_case							):
       """simple docstring"""


       def 			lowercase__				(		self		):






         """simple docstring"""





         super().setUp()
         lowerCAmelCase	:	str								=				0.82
         lowerCAmelCase	:	str								=				[
             "fsdp_shard_grad_op_transformer_based_wrap",
             "fsdp_full_shard_transformer_based_wrap",
         ]
         lowerCAmelCase	:	Optional[Any]								=				{
             "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
         }
         lowerCAmelCase	:	Dict								=				160
         lowerCAmelCase	:	str								=				160

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

       def 			lowercase__				(		self		):






         """simple docstring"""





         lowerCAmelCase	:	str								=				os.path.join(self.test_scripts_folder						,	"test_performance.py"		)
         lowerCAmelCase	:	Tuple								=				["accelerate", "launch", "--num_processes=2", "--num_machines=1", "--machine_rank=0", "--use_fsdp"]
         for config in self.performance_configs:
           lowerCAmelCase	:	Optional[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 			lowercase__				(		self		):






         """simple docstring"""





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

             lowerCAmelCase	:	List[Any]								=				cmd_config[:-1]
             lowerCAmelCase	:	List[Any]								=				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 			lowercase__				(		self		):






         """simple docstring"""





         lowerCAmelCase	:	Dict								=				os.path.join(self.test_scripts_folder						,	"test_peak_memory_usage.py"		)
         lowerCAmelCase	:	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():
           lowerCAmelCase	:	int								=				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()		)









import numpy as np

from transformers import Pipeline





def 	_lowerCAmelCase					( UpperCamelCase__:  Optional[Any]    )			-> Optional[int]:





	"""simple docstring"""







	A             =					np.max(UpperCamelCase__				,       axis=-1				,       keepdims=UpperCamelCase__    )
	A             =					np.exp(outputs - maxes    )
	return shifted_exp / shifted_exp.sum(axis=-1				,       keepdims=UpperCamelCase__    )






class    _UpperCamelCase     (       __snake_case					):

				"""simple docstring"""







				def   _UpperCAmelCase       (    self					, **a__ )						->					Union[str, Any]:
					A             =					{}
					if "second_text" in kwargs:
						A             =					kwargs["""second_text"""]
					return preprocess_kwargs, {}, {}

				def   _UpperCAmelCase       (    self					, a__					, a__=None )						->					str:
					return self.tokenizer(a__					, text_pair=a__					, return_tensors=self.framework )

				def   _UpperCAmelCase       (    self					, a__ )						->					Tuple:
					return self.model(**a__ )

				def   _UpperCAmelCase       (    self					, a__ )						->					Optional[int]:
					A             =					model_outputs.logits[0].numpy()
					A             =					softmax(a__ )

					A             =					np.argmax(a__ )
					A             =					self.model.config.idalabel[best_class]
					A             =					probabilities[best_class].item()
					A             =					logits.tolist()
					return {"label": label, "score": score, "logits": logits}



