






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

UpperCamelCase							       =    """bert-base-cased"""
UpperCamelCase							       =    """fp16"""
UpperCamelCase							       =    """bf16"""
UpperCamelCase							       =    [FPaa, BFaa]


@require_fsdp
@require_cuda
class    lowercase_    (_UpperCAmelCase     ):






      def 			lowerCamelCase__    (    self	)	->str:







        '''simple docstring'''


        super().setUp()

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






      def 			lowerCamelCase__    (    self	)	->Optional[Any]:







        '''simple docstring'''


        from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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






      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


        from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

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







        '''simple docstring'''


        from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

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







        '''simple docstring'''


        _a							=	AutoModel.from_pretrained(_UpperCAmelCase	)
        for policy in FSDP_AUTO_WRAP_POLICY:
          _a							=	self.dist_env.copy()
          _a							=	policy
          if policy == "TRANSFORMER_BASED_WRAP":
            _a							=	"BertLayer"
          elif policy == "SIZE_BASED_WRAP":
            _a							=	"2000"
          with mockenv_context(**_UpperCAmelCase	):
            _a							=	FullyShardedDataParallelPlugin()
            fsdp_plugin.set_auto_wrap_policy(_UpperCAmelCase	)
            if policy == "NO_WRAP":
              self.assertIsNone(fsdp_plugin.auto_wrap_policy	)
            else:
              self.assertIsNotNone(fsdp_plugin.auto_wrap_policy	)

        _a							=	self.dist_env.copy()
        _a							=	"TRANSFORMER_BASED_WRAP"
        _a							=	"T5Layer"
        with mockenv_context(**_UpperCAmelCase	):
          _a							=	FullyShardedDataParallelPlugin()
          with self.assertRaises(_UpperCAmelCase	) as cm:
            fsdp_plugin.set_auto_wrap_policy(_UpperCAmelCase	)
          self.assertTrue("Could not find the transformer layer class to wrap in the model." in str(cm.exception	)	)

        _a							=	self.dist_env.copy()
        _a							=	"SIZE_BASED_WRAP"
        _a							=	"0"
        with mockenv_context(**_UpperCAmelCase	):
          _a							=	FullyShardedDataParallelPlugin()
          fsdp_plugin.set_auto_wrap_policy(_UpperCAmelCase	)
          self.assertIsNone(fsdp_plugin.auto_wrap_policy	)






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






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







        '''simple docstring'''


        from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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







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






      def 			lowerCamelCase__    (    self	)	->Optional[int]:







        '''simple docstring'''


        super().setUp()
        _a							=	0.82
        _a							=	[
            "fsdp_shard_grad_op_transformer_based_wrap",
            "fsdp_full_shard_transformer_based_wrap",
        ]
        _a							=	{
            "multi_gpu_fp16": 3_2_0_0,
            "fsdp_shard_grad_op_transformer_based_wrap_fp16": 2_0_0_0,
            "fsdp_full_shard_transformer_based_wrap_fp16": 1_9_0_0,
            # 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							=	1_6_0
        _a							=	1_6_0

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






      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


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






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







        '''simple docstring'''


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

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






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







        '''simple docstring'''


        _a							=	os.path.join(self.test_scripts_folder     ,     "test_peak_memory_usage.py"	)
        _a							=	[
            "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							=	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(_UpperCAmelCase	):
              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(_UpperCAmelCase     ,     env=os.environ.copy()	)










"""simple docstring"""




import tempfile

import torch

from diffusers import PNDMScheduler

from .test_schedulers import SchedulerCommonTest




class    lowercase_    (_UpperCAmelCase     ):
      A__	:		Union[str, Any]       			=				(PNDMScheduler,)
      A__	:		Optional[int]       			=				(('''num_inference_steps''', 50),)






      def 			lowerCamelCase__    (    self     ,     **a_	)	->Optional[Any]:







        '''simple docstring'''


        _a							=	{
            "num_train_timesteps": 1_0_0_0,
            "beta_start": 0.0_001,
            "beta_end": 0.02,
            "beta_schedule": "linear",
        }

        config.update(**a_	)
        return config






      def 			lowerCamelCase__    (    self     ,     a_=0     ,     **a_	)	->Tuple:







        '''simple docstring'''


        _a							=	dict(self.forward_default_kwargs	)
        _a							=	kwargs.pop("num_inference_steps"     ,     a_	)
        _a							=	self.dummy_sample
        _a							=	0.1 * sample
        _a							=	[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

        for scheduler_class in self.scheduler_classes:
          _a							=	self.get_scheduler_config(**a_	)
          _a							=	scheduler_class(**a_	)
          scheduler.set_timesteps(a_	)
          # copy over dummy past residuals
          _a							=	dummy_past_residuals[:]

          with tempfile.TemporaryDirectory() as tmpdirname:
            scheduler.save_config(a_	)
            _a							=	scheduler_class.from_pretrained(a_	)
            new_scheduler.set_timesteps(a_	)
            # copy over dummy past residuals
            _a							=	dummy_past_residuals[:]

          _a							=	scheduler.step_prk(a_     ,     a_     ,     a_     ,     **a_	).prev_sample
          _a							=	new_scheduler.step_prk(a_     ,     a_     ,     a_     ,     **a_	).prev_sample

          assert torch.sum(torch.abs(output - new_output	)	) < 1E-5, "Scheduler outputs are not identical"

          _a							=	scheduler.step_plms(a_     ,     a_     ,     a_     ,     **a_	).prev_sample
          _a							=	new_scheduler.step_plms(a_     ,     a_     ,     a_     ,     **a_	).prev_sample

          assert torch.sum(torch.abs(output - new_output	)	) < 1E-5, "Scheduler outputs are not identical"






      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


        pass






      def 			lowerCamelCase__    (    self     ,     a_=0     ,     **a_	)	->Tuple:







        '''simple docstring'''


        _a							=	dict(self.forward_default_kwargs	)
        _a							=	kwargs.pop("num_inference_steps"     ,     a_	)
        _a							=	self.dummy_sample
        _a							=	0.1 * sample
        _a							=	[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

        for scheduler_class in self.scheduler_classes:
          _a							=	self.get_scheduler_config()
          _a							=	scheduler_class(**a_	)
          scheduler.set_timesteps(a_	)

          # copy over dummy past residuals (must be after setting timesteps)
          _a							=	dummy_past_residuals[:]

          with tempfile.TemporaryDirectory() as tmpdirname:
            scheduler.save_config(a_	)
            _a							=	scheduler_class.from_pretrained(a_	)
            # copy over dummy past residuals
            new_scheduler.set_timesteps(a_	)

            # copy over dummy past residual (must be after setting timesteps)
            _a							=	dummy_past_residuals[:]

          _a							=	scheduler.step_prk(a_     ,     a_     ,     a_     ,     **a_	).prev_sample
          _a							=	new_scheduler.step_prk(a_     ,     a_     ,     a_     ,     **a_	).prev_sample

          assert torch.sum(torch.abs(output - new_output	)	) < 1E-5, "Scheduler outputs are not identical"

          _a							=	scheduler.step_plms(a_     ,     a_     ,     a_     ,     **a_	).prev_sample
          _a							=	new_scheduler.step_plms(a_     ,     a_     ,     a_     ,     **a_	).prev_sample

          assert torch.sum(torch.abs(output - new_output	)	) < 1E-5, "Scheduler outputs are not identical"






      def 			lowerCamelCase__    (    self     ,     **a_	)	->Optional[int]:







        '''simple docstring'''


        _a							=	self.scheduler_classes[0]
        _a							=	self.get_scheduler_config(**a_	)
        _a							=	scheduler_class(**a_	)

        _a							=	1_0
        _a							=	self.dummy_model()
        _a							=	self.dummy_sample_deter
        scheduler.set_timesteps(a_	)

        for i, t in enumerate(scheduler.prk_timesteps	):
          _a							=	model(a_     ,     a_	)
          _a							=	scheduler.step_prk(a_     ,     a_     ,     a_	).prev_sample

        for i, t in enumerate(scheduler.plms_timesteps	):
          _a							=	model(a_     ,     a_	)
          _a							=	scheduler.step_plms(a_     ,     a_     ,     a_	).prev_sample

        return sample






      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


        _a							=	dict(self.forward_default_kwargs	)

        _a							=	kwargs.pop("num_inference_steps"     ,     a_	)

        for scheduler_class in self.scheduler_classes:
          _a							=	self.get_scheduler_config()
          _a							=	scheduler_class(**a_	)

          _a							=	self.dummy_sample
          _a							=	0.1 * sample

          if num_inference_steps is not None and hasattr(a_     ,     "set_timesteps"	):
            scheduler.set_timesteps(a_	)
          elif num_inference_steps is not None and not hasattr(a_     ,     "set_timesteps"	):
            _a							=	num_inference_steps

          # copy over dummy past residuals (must be done after set_timesteps)
          _a							=	[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]
          _a							=	dummy_past_residuals[:]

          _a							=	scheduler.step_prk(a_     ,     0     ,     a_     ,     **a_	).prev_sample
          _a							=	scheduler.step_prk(a_     ,     1     ,     a_     ,     **a_	).prev_sample

          self.assertEqual(output_a.shape     ,     sample.shape	)
          self.assertEqual(output_a.shape     ,     output_a.shape	)

          _a							=	scheduler.step_plms(a_     ,     0     ,     a_     ,     **a_	).prev_sample
          _a							=	scheduler.step_plms(a_     ,     1     ,     a_     ,     **a_	).prev_sample

          self.assertEqual(output_a.shape     ,     sample.shape	)
          self.assertEqual(output_a.shape     ,     output_a.shape	)






      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        for timesteps in [1_0_0, 1_0_0_0]:
          self.check_over_configs(num_train_timesteps=a_	)






      def 			lowerCamelCase__    (    self	)	->Optional[Any]:







        '''simple docstring'''


        for steps_offset in [0, 1]:
          self.check_over_configs(steps_offset=a_	)

        _a							=	self.scheduler_classes[0]
        _a							=	self.get_scheduler_config(steps_offset=1	)
        _a							=	scheduler_class(**a_	)
        scheduler.set_timesteps(1_0	)
        assert torch.equal(
            scheduler.timesteps     ,     torch.LongTensor(
                [9_0_1, 8_5_1, 8_5_1, 8_0_1, 8_0_1, 7_5_1, 7_5_1, 7_0_1, 7_0_1, 6_5_1, 6_5_1, 6_0_1, 6_0_1, 5_0_1, 4_0_1, 3_0_1, 2_0_1, 1_0_1, 1]	)     ,     )






      def 			lowerCamelCase__    (    self	)	->Tuple:







        '''simple docstring'''


        for beta_start, beta_end in zip([0.0_001, 0.001]     ,     [0.002, 0.02]	):
          self.check_over_configs(beta_start=a_     ,     beta_end=a_	)






      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


        for schedule in ["linear", "squaredcos_cap_v2"]:
          self.check_over_configs(beta_schedule=a_	)






      def 			lowerCamelCase__    (    self	)	->str:







        '''simple docstring'''


        for prediction_type in ["epsilon", "v_prediction"]:
          self.check_over_configs(prediction_type=a_	)






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        for t in [1, 5, 1_0]:
          self.check_over_forward(time_step=a_	)






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







        '''simple docstring'''


        for t, num_inference_steps in zip([1, 5, 1_0]     ,     [1_0, 5_0, 1_0_0]	):
          self.check_over_forward(num_inference_steps=a_	)






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







        '''simple docstring'''


        _a							=	2_7

        for scheduler_class in self.scheduler_classes:
          _a							=	self.dummy_sample
          _a							=	0.1 * sample

          _a							=	self.get_scheduler_config()
          _a							=	scheduler_class(**a_	)

          scheduler.set_timesteps(a_	)

          # before power of 3 fix, would error on first step, so we only need to do two
          for i, t in enumerate(scheduler.prk_timesteps[:2]	):
            _a							=	scheduler.step_prk(a_     ,     a_     ,     a_	).prev_sample






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        with self.assertRaises(a_	):
          _a							=	self.scheduler_classes[0]
          _a							=	self.get_scheduler_config()
          _a							=	scheduler_class(**a_	)

          scheduler.step_plms(self.dummy_sample     ,     1     ,     self.dummy_sample	).prev_sample






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        _a							=	self.full_loop()
        _a							=	torch.sum(torch.abs(a_	)	)
        _a							=	torch.mean(torch.abs(a_	)	)

        assert abs(result_sum.item() - 198.1_318	) < 1E-2
        assert abs(result_mean.item() - 0.2_580	) < 1E-3






      def 			lowerCamelCase__    (    self	)	->Optional[Any]:







        '''simple docstring'''


        _a							=	self.full_loop(prediction_type="v_prediction"	)
        _a							=	torch.sum(torch.abs(a_	)	)
        _a							=	torch.mean(torch.abs(a_	)	)

        assert abs(result_sum.item() - 67.3_986	) < 1E-2
        assert abs(result_mean.item() - 0.0_878	) < 1E-3






      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        _a							=	self.full_loop(set_alpha_to_one=a_     ,     beta_start=0.01	)
        _a							=	torch.sum(torch.abs(a_	)	)
        _a							=	torch.mean(torch.abs(a_	)	)

        assert abs(result_sum.item() - 230.0_399	) < 1E-2
        assert abs(result_mean.item() - 0.2_995	) < 1E-3






      def 			lowerCamelCase__    (    self	)	->str:







        '''simple docstring'''


        _a							=	self.full_loop(set_alpha_to_one=a_     ,     beta_start=0.01	)
        _a							=	torch.sum(torch.abs(a_	)	)
        _a							=	torch.mean(torch.abs(a_	)	)

        assert abs(result_sum.item() - 186.9_482	) < 1E-2
        assert abs(result_mean.item() - 0.2_434	) < 1E-3


