





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)

lowercase_          =		"""bert-base-cased"""
lowercase_          =		"""fp16"""
lowercase_          =		"""bf16"""
lowercase_          =		[FPaa, BFaa]
@require_fsdp
@require_cuda
class     SCREAMING_SNAKE_CASE       (UpperCAmelCase				):


   def 	SCREAMING_SNAKE_CASE_							(   self      :    int	)->       str:





     """simple docstring"""







     super().setUp()

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


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Dict	)->       List[str]:





     """simple docstring"""







     from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

     for i, strategy in enumerate(a	):
       lowercase__           =  self.dist_env.copy()
       lowercase__           =  f"""{i + 1}"""
       lowercase__           =  strategy
       with mockenv_context(**a	):
         lowercase__           =  FullyShardedDataParallelPlugin()
         self.assertEqual(fsdp_plugin.sharding_strategy      ,					ShardingStrategy(i + 1	)	)


   def 	SCREAMING_SNAKE_CASE_							(   self      :    int	)->       Any:





     """simple docstring"""







     from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

     for i, prefetch_policy in enumerate(a	):
       lowercase__           =  self.dist_env.copy()
       lowercase__           =  prefetch_policy
       with mockenv_context(**a	):
         lowercase__           =  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      :    Tuple	)->       Dict:





     """simple docstring"""







     from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

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





     """simple docstring"""







     lowercase__           =  AutoModel.from_pretrained(a	)
     for policy in FSDP_AUTO_WRAP_POLICY:
       lowercase__           =  self.dist_env.copy()
       lowercase__           =  policy
       if policy == "TRANSFORMER_BASED_WRAP":
         lowercase__           =  'BertLayer'
       elif policy == "SIZE_BASED_WRAP":
         lowercase__           =  '2000'
       with mockenv_context(**a	):
         lowercase__           =  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	)

     lowercase__           =  self.dist_env.copy()
     lowercase__           =  'TRANSFORMER_BASED_WRAP'
     lowercase__           =  'T5Layer'
     with mockenv_context(**a	):
       lowercase__           =  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	)	)

     lowercase__           =  self.dist_env.copy()
     lowercase__           =  'SIZE_BASED_WRAP'
     lowercase__           =  '0'
     with mockenv_context(**a	):
       lowercase__           =  FullyShardedDataParallelPlugin()
       fsdp_plugin.set_auto_wrap_policy(a	)
       self.assertIsNone(fsdp_plugin.auto_wrap_policy	)


   def 	SCREAMING_SNAKE_CASE_							(   self      :    List[str]	)->       List[str]:





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





     """simple docstring"""







     from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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






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


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Any	)->       Any:





     """simple docstring"""







     super().setUp()
     lowercase__           =  0.82
     lowercase__           =  [
         'fsdp_shard_grad_op_transformer_based_wrap',
         'fsdp_full_shard_transformer_based_wrap',
     ]
     lowercase__           =  {
         '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
     }
     lowercase__           =  160
     lowercase__           =  160

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


   def 	SCREAMING_SNAKE_CASE_							(   self      :    List[Any]	)->       str:





     """simple docstring"""







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





     """simple docstring"""







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

         lowercase__           =  cmd_config[:-1]
         lowercase__           =  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      :    Dict	)->       Dict:





     """simple docstring"""







     lowercase__           =  os.path.join(self.test_scripts_folder      ,					'test_peak_memory_usage.py'	)
     lowercase__           =  [
         '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():
       lowercase__           =  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 os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
      from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
      import torch_xla.core.xla_model as xm




def       __UpperCamelCase  (_SCREAMING_SNAKE_CASE		)    ->       List[str]:
  if is_torch_version('<'		,						'2.0.0'		) or not hasattr(_SCREAMING_SNAKE_CASE		,						'_dynamo'		):
    return False
  return isinstance(_SCREAMING_SNAKE_CASE		,						torch._dynamo.eval_frame.OptimizedModule		)




def       __UpperCamelCase  (_SCREAMING_SNAKE_CASE		,						_SCREAMING_SNAKE_CASE = True		)    ->       Dict:
  lowercase__           =  (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

  lowercase__           =  is_compiled_module(_SCREAMING_SNAKE_CASE		)
  if is_compiled:
    lowercase__           =  model
    lowercase__           =  model._orig_mod

  if is_deepspeed_available():
    options += (DeepSpeedEngine,)

  while isinstance(_SCREAMING_SNAKE_CASE		,						_SCREAMING_SNAKE_CASE		):
    lowercase__           =  model.module

  if not keep_fpaa_wrapper:
    lowercase__           =  getattr(_SCREAMING_SNAKE_CASE		,						'forward'		)
    lowercase__           =  model.__dict__.pop('_original_forward'		,						_SCREAMING_SNAKE_CASE		)
    if original_forward is not None:
      while hasattr(_SCREAMING_SNAKE_CASE		,						'__wrapped__'		):
        lowercase__           =  forward.__wrapped__
        if forward == original_forward:
          break
      lowercase__           =  forward
    if getattr(_SCREAMING_SNAKE_CASE		,						'_converted_to_transformer_engine'		,						_SCREAMING_SNAKE_CASE		):
      convert_model(_SCREAMING_SNAKE_CASE		,						to_transformer_engine=_SCREAMING_SNAKE_CASE		)

  if is_compiled:
    lowercase__           =  model
    lowercase__           =  compiled_model

  return model




def       __UpperCamelCase  ()    ->       Tuple:
  PartialState().wait_for_everyone()




def       __UpperCamelCase  (_SCREAMING_SNAKE_CASE		,						_SCREAMING_SNAKE_CASE		)    ->       List[Any]:
  if PartialState().distributed_type == DistributedType.TPU:
    xm.save(_SCREAMING_SNAKE_CASE		,						_SCREAMING_SNAKE_CASE		)
  elif PartialState().local_process_index == 0:
    torch.save(_SCREAMING_SNAKE_CASE		,						_SCREAMING_SNAKE_CASE		)




@contextmanager
def       __UpperCamelCase  (**_SCREAMING_SNAKE_CASE		)    ->       Optional[Any]:
  for key, value in kwargs.items():
    lowercase__           =  str(_SCREAMING_SNAKE_CASE		)

  yield

  for key in kwargs:
    if key.upper() in os.environ:
      del os.environ[key.upper()]




def       __UpperCamelCase  (_SCREAMING_SNAKE_CASE		)    ->       List[str]:
  if not hasattr(_SCREAMING_SNAKE_CASE		,						'__qualname__'		) and not hasattr(_SCREAMING_SNAKE_CASE		,						'__name__'		):
    lowercase__           =  getattr(_SCREAMING_SNAKE_CASE		,						'__class__'		,						_SCREAMING_SNAKE_CASE		)
  if hasattr(_SCREAMING_SNAKE_CASE		,						'__qualname__'		):
    return obj.__qualname__
  if hasattr(_SCREAMING_SNAKE_CASE		,						'__name__'		):
    return obj.__name__
  return str(_SCREAMING_SNAKE_CASE		)




def       __UpperCamelCase  (_SCREAMING_SNAKE_CASE		,						_SCREAMING_SNAKE_CASE		)    ->       Dict:
  for key, value in source.items():
    if isinstance(_SCREAMING_SNAKE_CASE		,						_SCREAMING_SNAKE_CASE		):
      lowercase__           =  destination.setdefault(_SCREAMING_SNAKE_CASE		,						{}		)
      merge_dicts(_SCREAMING_SNAKE_CASE		,						_SCREAMING_SNAKE_CASE		)
    else:
      lowercase__           =  value

  return destination




def       __UpperCamelCase  (_SCREAMING_SNAKE_CASE = None		)    ->       bool:
  if port is None:
    lowercase__           =  29500
  with socket.socket(socket.AF_INET		,						socket.SOCK_STREAM		) as s:
    return s.connect_ex(('localhost', port)		) == 0


