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

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


@require_fsdp
@require_cuda
class  _UpperCAmelCase ( __lowerCAmelCase    ):

     """simple docstring"""

     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       super().setUp()

       a_      :	str      =							dict(
           ACCELERATE_USE_FSDP="""true"""							,		MASTER_ADDR="""localhost"""							,		MASTER_PORT="""10999"""							,		RANK="""0"""							,		LOCAL_RANK="""0"""							,		WORLD_SIZE="""1"""							,		)


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

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


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

       for i, state_dict_type in enumerate(lowerCAmelCase_ ):
         a_      :	Tuple      =							self.dist_env.copy()
         a_      :	Tuple      =							state_dict_type
         with mockenv_context(**lowerCAmelCase_ ):
           a_      :	List[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 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	str      =							AutoModel.from_pretrained(lowerCAmelCase_ )
       for policy in FSDP_AUTO_WRAP_POLICY:
         a_      :	Union[str, Any]      =							self.dist_env.copy()
         a_      :	str      =							policy
         if policy == "TRANSFORMER_BASED_WRAP":
           a_      :	List[str]      =							"""BertLayer"""
         elif policy == "SIZE_BASED_WRAP":
           a_      :	List[Any]      =							"""2000"""
         with mockenv_context(**lowerCAmelCase_ ):
           a_      :	int      =							FullyShardedDataParallelPlugin()
           fsdp_plugin.set_auto_wrap_policy(lowerCAmelCase_ )
           if policy == "NO_WRAP":
             self.assertIsNone(fsdp_plugin.auto_wrap_policy )
           else:
             self.assertIsNotNone(fsdp_plugin.auto_wrap_policy )

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

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


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






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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







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

     """simple docstring"""

     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       super().setUp()
       a_      :	Any      =							0.82
       a_      :	Any      =							[
           """fsdp_shard_grad_op_transformer_based_wrap""",
           """fsdp_full_shard_transformer_based_wrap""",
       ]
       a_      :	Optional[int]      =							{
           """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
       }
       a_      :	Optional[int]      =							1_60
       a_      :	int      =							1_60

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


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       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_      :	Dict      =							cmd.copy()
         for i, strategy in enumerate(lowerCAmelCase_ ):
           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(lowerCAmelCase_							,		env=os.environ.copy() )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

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






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							os.path.join(self.test_scripts_folder							,		"""test_peak_memory_usage.py""" )
       a_      :	List[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():
         a_      :	Union[str, 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(lowerCAmelCase_ ):
             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(lowerCAmelCase_							,		env=os.environ.copy() )




'''simple docstring'''

import logging
import os

import quant_trainer
import torch
from torch.utils.data import DataLoader

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput


__snake_case:  Dict								=			logging.getLogger(__name__)

if is_torch_tpu_available(check_device=False):
    import torch_xla.core.xla_model as xm
    import torch_xla.debug.metrics as met


class  _UpperCAmelCase ( lowerCAmelCase__    ):

     """simple docstring"""

     def __init__(      self							,		*lowerCAmelCase_							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		**lowerCAmelCase_ ):

       '''simple docstring'''

       super().__init__(*lowerCAmelCase_							,		**lowerCAmelCase_ )
       a_      :	Tuple      =							eval_examples
       a_      :	Optional[Any]      =							post_process_function
       a_      :	List[str]      =							quant_trainer_args
       a_      :	List[str]      =							1_28  # default number of calibration samples


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_=None ):

       '''simple docstring'''

       if calib_dataset is None and self.calib_dataset is None:
         raise ValueError("""Trainer: calibration requires an calib_dataset.""" )
       a_      :	str      =							calib_dataset if calib_dataset is not None else self.calib_dataset

       a_      :	Dict      =							self._remove_unused_columns(lowerCAmelCase_							,		description="""Calibration""" )

       return DataLoader(
           lowerCAmelCase_							,		batch_size=self.args.eval_batch_size							,		collate_fn=self.data_collator							,		drop_last=self.args.dataloader_drop_last							,		num_workers=self.args.dataloader_num_workers							,		pin_memory=self.args.dataloader_pin_memory							,		shuffle=lowerCAmelCase_							,		)


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_=None ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							self.train_dataset if calib_dataset is None else calib_dataset
       a_      :	int      =							self.get_calib_dataloader(lowerCAmelCase_ )

       a_      :	Dict      =							self.model
       quant_trainer.configure_model(lowerCAmelCase_							,		self.quant_trainer_args							,		calib=lowerCAmelCase_ )
       model.eval()
       quant_trainer.enable_calibration(lowerCAmelCase_ )

       logger.info("""***** Running calibration *****""" )
       logger.info(f'''  Num examples = {self.calib_num}''' )
       logger.info(f'''  Batch size = {calib_dataloader.batch_size}''' )

       for step, inputs in enumerate(lowerCAmelCase_ ):
         # Prediction step
         a_      , a_      , a_      :	Optional[int]      =							self.prediction_step(lowerCAmelCase_							,		lowerCAmelCase_							,		prediction_loss_only=lowerCAmelCase_ )
         if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
           break

       quant_trainer.finish_calibration(lowerCAmelCase_							,		self.quant_trainer_args )
       a_      :	Optional[int]      =							model


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		lowerCAmelCase_ = "eval" ):

       '''simple docstring'''

       a_      :	str      =							self.eval_dataset if eval_dataset is None else eval_dataset
       a_      :	List[str]      =							self.get_eval_dataloader(lowerCAmelCase_ )
       a_      :	str      =							self.eval_examples if eval_examples is None else eval_examples

       # Temporarily disable metric computation, we will do it in the loop here.
       a_      :	List[str]      =							self.compute_metrics
       a_      :	Tuple      =							None
       a_      :	Optional[Any]      =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
       try:
         a_      :	List[str]      =							eval_loop(
             lowerCAmelCase_							,		description="""Evaluation"""							,		prediction_loss_only=True if compute_metrics is None else None							,		ignore_keys=lowerCAmelCase_							,		)
       finally:
         a_      :	int      =							compute_metrics

       if self.post_process_function is not None and self.compute_metrics is not None:
         a_      :	Optional[Any]      =							self.post_process_function(lowerCAmelCase_							,		lowerCAmelCase_							,		output.predictions )
         a_      :	Tuple      =							self.compute_metrics(lowerCAmelCase_ )

         # Prefix all keys with metric_key_prefix + '_'
         for key in list(metrics.keys() ):
           if not key.startswith(f'''{metric_key_prefix}_''' ):
             a_      :	Optional[int]      =							metrics.pop(lowerCAmelCase_ )

         self.log(lowerCAmelCase_ )
       else:
         a_      :	Any      =							{}

       if self.args.tpu_metrics_debug or self.args.debug:
         # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
         xm.master_print(met.metrics_report() )

       a_      :	Optional[Any]      =							self.callback_handler.on_evaluate(self.args							,		self.state							,		self.control							,		lowerCAmelCase_ )
       return metrics


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_=None							,		lowerCAmelCase_ = "test" ):

       '''simple docstring'''

       a_      :	List[Any]      =							self.get_test_dataloader(lowerCAmelCase_ )

       # Temporarily disable metric computation, we will do it in the loop here.
       a_      :	Any      =							self.compute_metrics
       a_      :	Dict      =							None
       a_      :	Dict      =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
       try:
         a_      :	Dict      =							eval_loop(
             lowerCAmelCase_							,		description="""Prediction"""							,		prediction_loss_only=True if compute_metrics is None else None							,		ignore_keys=lowerCAmelCase_							,		)
       finally:
         a_      :	int      =							compute_metrics

       if self.post_process_function is None or self.compute_metrics is None:
         return output

       a_      :	Tuple      =							self.post_process_function(lowerCAmelCase_							,		lowerCAmelCase_							,		output.predictions							,		"""predict""" )
       a_      :	List[Any]      =							self.compute_metrics(lowerCAmelCase_ )

       # Prefix all keys with metric_key_prefix + '_'
       for key in list(metrics.keys() ):
         if not key.startswith(f'''{metric_key_prefix}_''' ):
           a_      :	Tuple      =							metrics.pop(lowerCAmelCase_ )

       return PredictionOutput(predictions=predictions.predictions							,		label_ids=predictions.label_ids							,		metrics=lowerCAmelCase_ )






     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_="./" ):

       '''simple docstring'''

       a_      :	Any      =							self.eval_dataset
       a_      :	Tuple      =							self.get_eval_dataloader(lowerCAmelCase_ )

       a_      :	List[str]      =							next(iter(lowerCAmelCase_ ) )

       # saving device - to make it consistent
       a_      :	Dict      =							torch.device("""cuda""" if torch.cuda.is_available() else """cpu""" )

       # convert to tuple
       a_      :	Optional[int]      =							tuple(v.to(lowerCAmelCase_ ) for k, v in batch.items() )

       logger.info("""Converting model to be onnx compatible""" )
       from pytorch_quantization.nn import TensorQuantizer

       a_      :	List[str]      =							True

       a_      :	str      =							self.model.to(lowerCAmelCase_ )

       model.eval()
       model.float()

       a_      :	Optional[int]      =							model.module if hasattr(lowerCAmelCase_							,		"""module""" ) else model
       quant_trainer.configure_model(lowerCAmelCase_							,		self.quant_trainer_args )

       a_      :	Union[str, Any]      =							os.path.join(lowerCAmelCase_							,		"""model.onnx""" )
       logger.info(f'''exporting model to {output_model_file}''' )

       a_      :	Dict      =							{0: """batch_size""", 1: """seq_len"""}

       torch.onnx.export(
           lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		export_params=lowerCAmelCase_							,		opset_version=13							,		do_constant_folding=lowerCAmelCase_							,		input_names=["""input_ids""", """attention_mask""", """token_type_ids"""]							,		output_names=["""output_start_logits""", """output_end_logits"""]							,		dynamic_axes={
               """input_ids""": axes,
               """attention_mask""": axes,
               """token_type_ids""": axes,
               """output_start_logits""": axes,
               """output_end_logits""": axes,
           }							,		verbose=lowerCAmelCase_							,		)
       logger.info("""onnx export finished""" )


