






"""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__	:    Optional[int]			 = 'bert-base-cased'
A__	:    List[Any]			 = 'fp16'
A__	:    Any			 = 'bf16'
A__	:    List[str]			 = [FPaa, BFaa]







@require_fsdp
@require_cuda
class        lowercase__      (							snake_case__       ):


			def      UpperCAmelCase__						(				self					:  Optional[Any] ):
										super().setUp()

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


			def      UpperCAmelCase__						(				self					:  str ):
										from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

										for i, strategy in enumerate(snake_case__ ):
																	lowerCamelCase_	:							int									=self.dist_env.copy()
																	lowerCamelCase_	:							Any									=F"""{i + 1}"""
																	lowerCamelCase_	:							Optional[Any]									=strategy
																	with mockenv_context(**snake_case__ ):
																								lowerCamelCase_	:							str									=FullyShardedDataParallelPlugin()
																								self.assertEqual(fsdp_plugin.sharding_strategy       ,	ShardingStrategy(i + 1 ) )


			def      UpperCAmelCase__						(				self					:  Optional[int] ):
										from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

										for i, prefetch_policy in enumerate(snake_case__ ):
																	lowerCamelCase_	:							List[Any]									=self.dist_env.copy()
																	lowerCamelCase_	:							Tuple									=prefetch_policy
																	with mockenv_context(**snake_case__ ):
																								lowerCamelCase_	:							Any									=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					:  str ):
										from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

										for i, state_dict_type in enumerate(snake_case__ ):
																	lowerCamelCase_	:							str									=self.dist_env.copy()
																	lowerCamelCase_	:							int									=state_dict_type
																	with mockenv_context(**snake_case__ ):
																								lowerCamelCase_	:							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      UpperCAmelCase__						(				self					:  Optional[Any] ):
										lowerCamelCase_	:							int									=AutoModel.from_pretrained(snake_case__ )
										for policy in FSDP_AUTO_WRAP_POLICY:
																	lowerCamelCase_	:							Any									=self.dist_env.copy()
																	lowerCamelCase_	:							List[Any]									=policy
																	if policy == "TRANSFORMER_BASED_WRAP":
																								lowerCamelCase_	:							Tuple									="BertLayer"
																	elif policy == "SIZE_BASED_WRAP":
																								lowerCamelCase_	:							Any									="2000"
																	with mockenv_context(**snake_case__ ):
																								lowerCamelCase_	:							Dict									=FullyShardedDataParallelPlugin()
																								fsdp_plugin.set_auto_wrap_policy(snake_case__ )
																								if policy == "NO_WRAP":
																															self.assertIsNone(fsdp_plugin.auto_wrap_policy )
																								else:
																															self.assertIsNotNone(fsdp_plugin.auto_wrap_policy )

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

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


			def      UpperCAmelCase__						(				self					:  List[str] ):
										from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
										from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

										for mp_dtype in dtypes:
																	lowerCamelCase_	:							Tuple									=self.dist_env.copy()
																	lowerCamelCase_	:							Dict									=mp_dtype
																	with mockenv_context(**snake_case__ ):
																								lowerCamelCase_	:							int									=Accelerator()
																								if mp_dtype == "fp16":
																															lowerCamelCase_	:							str									=torch.floataa
																								elif mp_dtype == "bf16":
																															lowerCamelCase_	:							Any									=torch.bfloataa
																								lowerCamelCase_	:							int									=MixedPrecision(param_dtype=snake_case__       ,	reduce_dtype=snake_case__       ,	buffer_dtype=snake_case__ )
																								self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy       ,	snake_case__ )
																								if mp_dtype == FPaa:
																															self.assertTrue(isinstance(accelerator.scaler       ,	snake_case__ ) )
																								elif mp_dtype == BFaa:
																															self.assertIsNone(accelerator.scaler )
																								AcceleratorState._reset_state(snake_case__ )


			def      UpperCAmelCase__						(				self					:  Optional[Any] ):
										from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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







@require_fsdp
@require_multi_gpu
@slow
class        lowercase__      (							snake_case__       ):


			def      UpperCAmelCase__						(				self					:  Optional[Any] ):
										super().setUp()
										lowerCamelCase_	:							List[str]									=0.82
										lowerCamelCase_	:							str									=[
										    "fsdp_shard_grad_op_transformer_based_wrap",
										    "fsdp_full_shard_transformer_based_wrap",
										]
										lowerCamelCase_	:							Optional[Any]									={
										    "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
										}
										lowerCamelCase_	:							List[Any]									=160
										lowerCamelCase_	:							Optional[int]									=160

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


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


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

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


			def      UpperCAmelCase__						(				self					:  Tuple ):
										lowerCamelCase_	:							Dict									=os.path.join(self.test_scripts_folder       ,	"test_peak_memory_usage.py" )
										lowerCamelCase_	:							Optional[Any]									=[
										    "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():
																	lowerCamelCase_	:							List[str]									=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(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(snake_case__       ,	env=os.environ.copy() )











"""simple docstring"""



def 			_snake_case    (					lowerCamelCase__     :   Optional[Any]  )    ->     Optional[int]:
							if not head:
														return True
							# split the list to two parts
							lowerCamelCase_							,						lowerCamelCase_	:							Union[str, Any]									=head.next, head
							while fast and fast.next:
														lowerCamelCase_	:							Optional[Any]									=fast.next.next
														lowerCamelCase_	:							str									=slow.next
							lowerCamelCase_	:							Tuple									=slow.next
							lowerCamelCase_	:							Any									=None  # Don't forget here! But forget still works!
							# reverse the second part
							lowerCamelCase_	:							List[str]									=None
							while second:
														lowerCamelCase_	:							Any									=second.next
														lowerCamelCase_	:							Union[str, Any]									=node
														lowerCamelCase_	:							Union[str, Any]									=second
														lowerCamelCase_	:							Optional[Any]									=nxt
							# compare two parts
							# second part has the same or one less node
							while node:
														if node.val != head.val:
																					return False
														lowerCamelCase_	:							List[str]									=node.next
														lowerCamelCase_	:							Optional[Any]									=head.next
							return True

def 			_snake_case    (					lowerCamelCase__     :   str  )    ->     Optional[int]:
							if not head or not head.next:
														return True

							# 1. Get the midpoint (slow)
							lowerCamelCase_	:							List[str]									=head
							while fast and fast.next:
														lowerCamelCase_							,						lowerCamelCase_	:							Union[str, Any]									=fast.next.next, slow.next

							# 2. Push the second half into the stack
							lowerCamelCase_	:							List[Any]									=[slow.val]
							while slow.next:
														lowerCamelCase_	:							List[Any]									=slow.next
														stack.append(slow.val  )

							# 3. Comparison
							while stack:
														if stack.pop() != cur.val:
																					return False
														lowerCamelCase_	:							Union[str, Any]									=cur.next

							return True

def 			_snake_case    (					lowerCamelCase__     :   Dict  )    ->     Optional[Any]:
							if not head or not head.next:
														return True
							lowerCamelCase_	:							Union[str, Any]									={}
							lowerCamelCase_	:							List[Any]									=0
							while head:
														if head.val in d:
																					d[head.val].append(lowerCamelCase__  )
														else:
																					lowerCamelCase_	:							List[str]									=[pos]
														lowerCamelCase_	:							Optional[int]									=head.next
														pos += 1
							lowerCamelCase_	:							Union[str, Any]									=pos - 1
							lowerCamelCase_	:							Optional[int]									=0
							for v in d.values():
														if len(lowerCamelCase__  ) % 2 != 0:
																					middle += 1
														else:
																					lowerCamelCase_	:							Optional[Any]									=0
																					for i in range(0    ,						len(lowerCamelCase__  )  ):
																												if v[i] + v[len(lowerCamelCase__  ) - 1 - step] != checksum:
																																			return False
																												step += 1
														if middle > 1:
																					return False
							return True



