




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

__A					=      "bert-base-cased"
__A					=      "fp16"
__A					=      "bf16"
__A					=      [FPaa, BFaa]
@require_fsdp
@require_cuda
class       lowerCamelCase__     (	UpperCamelCase__      ):


			def 				lowerCamelCase_					(  self				):



					"""simple docstring"""







					super().setUp()

					snake_case		:    Dict             =					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(SCREAMING_SNAKE_CASE				):
							snake_case		:    List[Any]             =					self.dist_env.copy()
							snake_case		:    Union[str, Any]             =					F'''{i + 1}'''
							snake_case		:    List[str]             =					strategy
							with mockenv_context(**SCREAMING_SNAKE_CASE				):
									snake_case		:    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(SCREAMING_SNAKE_CASE				):
							snake_case		:    Any             =					self.dist_env.copy()
							snake_case		:    Tuple             =					prefetch_policy
							with mockenv_context(**SCREAMING_SNAKE_CASE				):
									snake_case		:    Any             =					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(SCREAMING_SNAKE_CASE				):
							snake_case		:    Optional[int]             =					self.dist_env.copy()
							snake_case		:    List[Any]             =					state_dict_type
							with mockenv_context(**SCREAMING_SNAKE_CASE				):
									snake_case		:    Tuple             =					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"""







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

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

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



			def 				lowerCamelCase_					(  self				):



					"""simple docstring"""







					from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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







@require_fsdp
@require_multi_gpu
@slow
class       lowerCamelCase__     (	UpperCamelCase__      ):


			def 				lowerCamelCase_					(  self				):



					"""simple docstring"""







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

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


			def 				lowerCamelCase_					(  self				):



					"""simple docstring"""







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


			def 				lowerCamelCase_					(  self				):



					"""simple docstring"""







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

									snake_case		:    Optional[int]             =					cmd_config[:-1]
									snake_case		:    Union[str, 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(SCREAMING_SNAKE_CASE					,			env=os.environ.copy()				)



			def 				lowerCamelCase_					(  self				):



					"""simple docstring"""







					snake_case		:    str             =					os.path.join(self.test_scripts_folder					,			"test_peak_memory_usage.py"				)
					snake_case		:    int             =					[
					    "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():
							snake_case		:    Tuple             =					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(SCREAMING_SNAKE_CASE				):
											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(SCREAMING_SNAKE_CASE					,			env=os.environ.copy()				)



'''simple docstring'''
import unittest

import torch

from diffusers import DDIMScheduler, DDPMScheduler, UNetaDModel
from diffusers.training_utils import set_seed
from diffusers.utils.testing_utils import slow


lowercase__											=False







class   a_ (      unittest.TestCase					):




			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase=32	):
				set_seed(0	)
				a_      				=						UNetaDModel(sample_size=UpperCAmelCase					,   in_channels=3					,   out_channels=3	)
				a_      				=						torch.optim.SGD(model.parameters()					,   lr=0.00_01	)
				return model, optimizer






			@slow
			def 					lowerCAmelCase__					(       self	):
				a_      				=						"""cpu"""  # ensure full determinism without setting the CUBLAS_WORKSPACE_CONFIG env variable
				a_      				=						DDPMScheduler(
				    num_train_timesteps=10_00					,   beta_start=0.00_01					,   beta_end=0.02					,   beta_schedule="""linear"""					,   clip_sample=UpperCAmelCase					,   )
				a_      				=						DDIMScheduler(
				    num_train_timesteps=10_00					,   beta_start=0.00_01					,   beta_end=0.02					,   beta_schedule="""linear"""					,   clip_sample=UpperCAmelCase					,   )

				assert ddpm_scheduler.config.num_train_timesteps == ddim_scheduler.config.num_train_timesteps

				# shared batches for DDPM and DDIM
				set_seed(0	)
				a_      				=						[torch.randn((4, 3, 32, 32)	).clip(-1					,   1	).to(UpperCAmelCase	) for _ in range(4	)]
				a_      				=						[torch.randn((4, 3, 32, 32)	).to(UpperCAmelCase	) for _ in range(4	)]
				a_      				=						[torch.randint(0					,   10_00					,   (4,)	).long().to(UpperCAmelCase	) for _ in range(4	)]

				# train with a DDPM scheduler
				a_				,  a_      				=						self.get_model_optimizer(resolution=32	)
				model.train().to(UpperCAmelCase	)
				for i in range(4	):
					optimizer.zero_grad()
					a_      				=						ddpm_scheduler.add_noise(clean_images[i]					,   noise[i]					,   timesteps[i]	)
					a_      				=						model(UpperCAmelCase					,   timesteps[i]	).sample
					a_      				=						torch.nn.functional.mse_loss(UpperCAmelCase					,   noise[i]	)
					loss.backward()
					optimizer.step()
				del model, optimizer

				# recreate the model and optimizer, and retry with DDIM
				a_				,  a_      				=						self.get_model_optimizer(resolution=32	)
				model.train().to(UpperCAmelCase	)
				for i in range(4	):
					optimizer.zero_grad()
					a_      				=						ddim_scheduler.add_noise(clean_images[i]					,   noise[i]					,   timesteps[i]	)
					a_      				=						model(UpperCAmelCase					,   timesteps[i]	).sample
					a_      				=						torch.nn.functional.mse_loss(UpperCAmelCase					,   noise[i]	)
					loss.backward()
					optimizer.step()
				del model, optimizer

				self.assertTrue(torch.allclose(UpperCAmelCase					,   UpperCAmelCase					,   atol=1e-5	)	)
				self.assertTrue(torch.allclose(UpperCAmelCase					,   UpperCAmelCase					,   atol=1e-5	)	)
