



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

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

@require_fsdp
@require_cuda
class       _a  ( SCREAMING_SNAKE_CASE__):



			def 				__lowercase					(			self							:				List[Any]		)							->					Optional[int]:
									super().setUp()

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



			def 				__lowercase					(			self							:				int		)							->					str:
									from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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



			def 				__lowercase					(			self							:				List[Any]		)							->					Any:
									from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

									for i, prefetch_policy in enumerate(_lowercase		):
															snake_case   :			List[Any]						    = self.dist_env.copy()
															snake_case   :			Optional[int]						    = prefetch_policy
															with mockenv_context(**_lowercase		):
																					snake_case   :			Optional[int]						    = FullyShardedDataParallelPlugin()
																					if prefetch_policy == "NO_PREFETCH":
																											self.assertIsNone(fsdp_plugin.backward_prefetch		)
																					else:
																											self.assertEqual(fsdp_plugin.backward_prefetch   ,	BackwardPrefetch(i + 1		)		)



			def 				__lowercase					(			self							:				Optional[Any]		)							->					int:
									from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

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

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

									snake_case   :			Dict						    = self.dist_env.copy()
									snake_case   :			List[str]						    = "SIZE_BASED_WRAP"
									snake_case   :			Union[str, Any]						    = "0"
									with mockenv_context(**_lowercase		):
															snake_case   :			Any						    = FullyShardedDataParallelPlugin()
															fsdp_plugin.set_auto_wrap_policy(_lowercase		)
															self.assertIsNone(fsdp_plugin.auto_wrap_policy		)



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



			def 				__lowercase					(			self							:				List[Any]		)							->					Union[str, Any]:
									from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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



@require_fsdp
@require_multi_gpu
@slow
class       _a  ( SCREAMING_SNAKE_CASE__):



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

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



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



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

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



			def 				__lowercase					(			self							:				Tuple		)							->					Union[str, Any]:
									snake_case   :			Optional[Any]						    = os.path.join(self.test_scripts_folder   ,	"test_peak_memory_usage.py"		)
									snake_case   :			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():
															snake_case   :			Union[str, Any]						    = 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(_lowercase		):
																											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(_lowercase   ,	env=os.environ.copy()		)








"""simple docstring"""







import importlib
import shutil
import threading
import warnings
from typing import List

import fsspec
import fsspec.asyn

from . import compression
from .hffilesystem import HfFileSystem


A		=						importlib.util.find_spec('s3fs') is not None

if _has_safs:
			from .safilesystem import SaFileSystem  # noqa: F401

A		=						[
    compression.BzaFileSystem,
    compression.GzipFileSystem,
    compression.LzaFileSystem,
    compression.XzFileSystem,
    compression.ZstdFileSystem,
]

# Register custom filesystems
for fs_class in COMPRESSION_FILESYSTEMS + [HfFileSystem]:
			if fs_class.protocol in fsspec.registry and fsspec.registry[fs_class.protocol] is not fs_class:
						warnings.warn(F'''A filesystem protocol was already set for {fs_class.protocol} and will be overwritten.''')
			fsspec.register_implementation(fs_class.protocol, fs_class, clobber=True)

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							str ):
						"""simple docstring"""






						if "://" in dataset_path:
												snake_case   :			List[str]						    = dataset_path.split("://" )[1]
						return dataset_path

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							fsspec.AbstractFileSystem ):
						"""simple docstring"""






						if fs is not None and fs.protocol != "file":
												return True
						else:
												return False

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							fsspec.AbstractFileSystem     ,		lowerCamelCase_:							str     ,		lowerCamelCase_:							str ):
						"""simple docstring"""






						snake_case   :			int						    = not is_remote_filesystem(lowerCamelCase_ )
						if is_local:
												# LocalFileSystem.mv does copy + rm, it is more efficient to simply move a local directory
												shutil.move(fs._strip_protocol(lowerCamelCase_ )     ,		fs._strip_protocol(lowerCamelCase_ ) )
						else:
												fs.mv(lowerCamelCase_     ,		lowerCamelCase_     ,		recursive=lowerCamelCase_ )

def  __SCREAMING_SNAKE_CASE		(			):
						"""simple docstring"""






						if hasattr(fsspec.asyn     ,		"reset_lock" ):
												# for future fsspec>2022.05.0
												fsspec.asyn.reset_lock()
						else:
												snake_case   :			Union[str, Any]						    = None
												snake_case   :			Optional[int]						    = None
												snake_case   :			List[str]						    = threading.Lock()



