





'''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(4_2)

_SCREAMING_SNAKE_CASE   =      '''bert-base-cased'''
_SCREAMING_SNAKE_CASE   =      '''fp16'''
_SCREAMING_SNAKE_CASE   =      '''bf16'''
_SCREAMING_SNAKE_CASE   =      [FPaa, BFaa]






@require_fsdp
@require_cuda
class 							__lowercase    ( lowerCAmelCase__  ):







	'''simple docstring'''

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







				'''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   _UpperCAmelCase       (self	)							->							Optional[int]:







				'''simple docstring'''





				from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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


	def   _UpperCAmelCase       (self	)							->							Dict:







				'''simple docstring'''





				from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

				for i, prefetch_policy in enumerate(_lowerCamelCase	):
							__lowercase			  =							self.dist_env.copy()
							__lowercase			  =							prefetch_policy
							with mockenv_context(**_lowerCamelCase	):
										__lowercase			  =							FullyShardedDataParallelPlugin()
										if prefetch_policy == "NO_PREFETCH":
													self.assertIsNone(fsdp_plugin.backward_prefetch	)
										else:
													self.assertEqual(fsdp_plugin.backward_prefetch       ,BackwardPrefetch(i + 1	)	)


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







				'''simple docstring'''





				from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

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







				'''simple docstring'''





				__lowercase			  =							AutoModel.from_pretrained(_lowerCamelCase	)
				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(**_lowerCamelCase	):
										__lowercase			  =							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	)

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

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


	def   _UpperCAmelCase       (self	)							->							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:
							__lowercase			  =							self.dist_env.copy()
							__lowercase			  =							mp_dtype
							with mockenv_context(**_lowerCamelCase	):
										__lowercase			  =							Accelerator()
										if mp_dtype == "fp16":
													__lowercase			  =							torch.floataa
										elif mp_dtype == "bf16":
													__lowercase			  =							torch.bfloataa
										__lowercase			  =							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   _UpperCAmelCase       (self	)							->							int:







				'''simple docstring'''





				from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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






@require_fsdp
@require_multi_gpu
@slow
class 							__lowercase    ( lowerCAmelCase__  ):







	'''simple docstring'''

	def   _UpperCAmelCase       (self	)							->							Tuple:







				'''simple docstring'''





				super().setUp()
				__lowercase			  =							0.8_2
				__lowercase			  =							[
				    '''fsdp_shard_grad_op_transformer_based_wrap''',
				    '''fsdp_full_shard_transformer_based_wrap''',
				]
				__lowercase			  =							{
				    '''multi_gpu_fp16''': 3200,
				    '''fsdp_shard_grad_op_transformer_based_wrap_fp16''': 2000,
				    '''fsdp_full_shard_transformer_based_wrap_fp16''': 1900,
				    # 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   _UpperCAmelCase       (self	)							->							List[Any]:







				'''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(_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   _UpperCAmelCase       (self	)							->							List[Any]:







				'''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(_lowerCamelCase	):
							__lowercase			  =							cmd.copy()
							cmd_config.append(f"--fsdp_sharding_strategy={i+1}"	)
							if strategy != "FULL_SHARD":
										continue
							__lowercase			  =							len(_lowerCamelCase	)
							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(_lowerCamelCase       ,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(_lowerCamelCase       ,env=os.environ.copy()	)


	def   _UpperCAmelCase       (self	)							->							Any:







				'''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(_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'''



# NOTE: This file is deprecated and will be removed in a future version.
# It only exists so that temporarely `from diffusers.pipelines import DiffusionPipeline` works

from ...utils import deprecate
from ..controlnet.pipeline_flax_controlnet import FlaxStableDiffusionControlNetPipeline  # noqa: F401


deprecate(
    '''stable diffusion controlnet''',
    '''0.22.0''',
    '''Importing `FlaxStableDiffusionControlNetPipeline` from diffusers.pipelines.stable_diffusion.flax_pipeline_stable_diffusion_controlnet is deprecated. Please import `from diffusers import FlaxStableDiffusionControlNetPipeline` instead.''',
    standard_warn=False,
    stacklevel=3,
)





