


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







@require_fsdp
@require_cuda
class A__       (						UpperCAmelCase__      ):






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




             '''simple docstring'''
             super().setUp()

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






      def __UpperCAmelCase       ( self			:Dict	)       ->					List[str]:




             '''simple docstring'''
             from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

             for i, strategy in enumerate(SCREAMING_SNAKE_CASE	):
                    _a    : Dict     					=self.dist_env.copy()
                    _a    : int     					=f"{i + 1}"
                    _a    : List[str]     					=strategy
                    with mockenv_context(**SCREAMING_SNAKE_CASE	):
                           _a    : Dict     					=FullyShardedDataParallelPlugin()
                           self.assertEqual(fsdp_plugin.sharding_strategy     ,						ShardingStrategy(i + 1	)	)






      def __UpperCAmelCase       ( self			:Dict	)       ->					int:




             '''simple docstring'''
             from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

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




             '''simple docstring'''
             from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

             for i, state_dict_type in enumerate(SCREAMING_SNAKE_CASE	):
                    _a    : Optional[int]     					=self.dist_env.copy()
                    _a    : Optional[int]     					=state_dict_type
                    with mockenv_context(**SCREAMING_SNAKE_CASE	):
                           _a    : Dict     					=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[int]	)       ->					Optional[int]:




             '''simple docstring'''
             _a    : List[Any]     					=AutoModel.from_pretrained(SCREAMING_SNAKE_CASE	)
             for policy in FSDP_AUTO_WRAP_POLICY:
                    _a    : int     					=self.dist_env.copy()
                    _a    : Optional[Any]     					=policy
                    if policy == "TRANSFORMER_BASED_WRAP":
                           _a    : int     					="""BertLayer"""
                    elif policy == "SIZE_BASED_WRAP":
                           _a    : Any     					="""2000"""
                    with mockenv_context(**SCREAMING_SNAKE_CASE	):
                           _a    : Dict     					=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	)

             _a    : Union[str, Any]     					=self.dist_env.copy()
             _a    : Tuple     					="""TRANSFORMER_BASED_WRAP"""
             _a    : List[str]     					="""T5Layer"""
             with mockenv_context(**SCREAMING_SNAKE_CASE	):
                    _a    : Optional[int]     					=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	)	)

             _a    : Optional[Any]     					=self.dist_env.copy()
             _a    : Union[str, Any]     					="""SIZE_BASED_WRAP"""
             _a    : int     					="""0"""
             with mockenv_context(**SCREAMING_SNAKE_CASE	):
                    _a    : str     					=FullyShardedDataParallelPlugin()
                    fsdp_plugin.set_auto_wrap_policy(SCREAMING_SNAKE_CASE	)
                    self.assertIsNone(fsdp_plugin.auto_wrap_policy	)






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




             '''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    : Dict     					=self.dist_env.copy()
                    _a    : List[str]     					=mp_dtype
                    with mockenv_context(**SCREAMING_SNAKE_CASE	):
                           _a    : Dict     					=Accelerator()
                           if mp_dtype == "fp16":
                                  _a    : int     					=torch.floataa
                           elif mp_dtype == "bf16":
                                  _a    : int     					=torch.bfloataa
                           _a    : str     					=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 __UpperCAmelCase       ( self			:Any	)       ->					Any:




             '''simple docstring'''
             from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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







@require_fsdp
@require_multi_gpu
@slow
class A__       (						UpperCAmelCase__      ):






      def __UpperCAmelCase       ( self			:Optional[int]	)       ->					List[str]:




             '''simple docstring'''
             super().setUp()
             _a    : str     					=0.82
             _a    : Optional[int]     					=[
                 """fsdp_shard_grad_op_transformer_based_wrap""",
                 """fsdp_full_shard_transformer_based_wrap""",
             ]
             _a    : Union[str, Any]     					={
                 """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    : Dict     					=1_6_0
             _a    : Optional[int]     					=1_6_0

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






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




             '''simple docstring'''
             _a    : Union[str, Any]     					=os.path.join(self.test_scripts_folder     ,						"""test_performance.py"""	)
             _a    : List[str]     					=["""accelerate""", """launch""", """--num_processes=2""", """--num_machines=1""", """--machine_rank=0""", """--use_fsdp"""]
             for config in self.performance_configs:
                    _a    : Dict     					=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 __UpperCAmelCase       ( self			:Union[str, Any]	)       ->					int:




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

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






      def __UpperCAmelCase       ( self			:Optional[int]	)       ->					List[Any]:




             '''simple docstring'''
             _a    : str     					=os.path.join(self.test_scripts_folder     ,						"""test_peak_memory_usage.py"""	)
             _a    : Optional[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():
                    _a    : 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(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 contextlib
import os
import sqlitea

import pytest

from datasets import Dataset, Features, Value
from datasets.io.sql import SqlDatasetReader, SqlDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases, require_sqlalchemy


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Any    ,_UpperCAmelCase	:	str  )  ->			Dict:
       assert isinstance(_UpperCAmelCase    ,_UpperCAmelCase  )
       assert dataset.num_rows == 4
       assert dataset.num_columns == 3
       assert dataset.column_names == ["col_1", "col_2", "col_3"]
       for feature, expected_dtype in expected_features.items():
              assert dataset.features[feature].dtype == expected_dtype


@require_sqlalchemy
@pytest.mark.parametrize("""keep_in_memory"""    ,[False, True]  )
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[Any]    ,_UpperCAmelCase	:	Optional[int]    ,_UpperCAmelCase	:	Optional[int]    ,_UpperCAmelCase	:	str  )  ->			Optional[Any]:
       _a    : Any     					=tmp_path / """cache"""
       _a    : int     					={"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""}
       with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
              _a    : Tuple     					=SqlDatasetReader(
                  """dataset"""    ,"""sqlite:///""" + sqlite_path    ,cache_dir=_UpperCAmelCase    ,keep_in_memory=_UpperCAmelCase  ).read()
       _check_sql_dataset(_UpperCAmelCase    ,_UpperCAmelCase  )


@require_sqlalchemy
@pytest.mark.parametrize(
    """features"""    ,[
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]    ,)
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	List[Any]    ,_UpperCAmelCase	:	Dict    ,_UpperCAmelCase	:	Optional[Any]    ,_UpperCAmelCase	:	int  )  ->			List[Any]:
       _a    : Union[str, Any]     					=tmp_path / """cache"""
       _a    : str     					={"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""}
       _a    : Optional[int]     					=features.copy() if features else default_expected_features
       _a    : Union[str, Any]     					=(
           Features({feature: Value(_UpperCAmelCase  ) for feature, dtype in features.items()}  ) if features is not None else None
       )
       _a    : Optional[Any]     					=SqlDatasetReader("""dataset"""    ,"""sqlite:///""" + sqlite_path    ,features=_UpperCAmelCase    ,cache_dir=_UpperCAmelCase  ).read()
       _check_sql_dataset(_UpperCAmelCase    ,_UpperCAmelCase  )


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[Any]  )  ->			List[str]:
       with contextlib.closing(sqlitea.connect(_UpperCAmelCase  )  ) as con:
              _a    : Any     					=con.cursor()
              cur.execute("""SELECT * FROM dataset"""  )
              for row in cur:
                     yield row


@require_sqlalchemy
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Dict    ,_UpperCAmelCase	:	Tuple    ,_UpperCAmelCase	:	List[str]  )  ->			Union[str, Any]:
       _a    : Union[str, Any]     					=tmp_path / """cache"""
       _a    : Union[str, Any]     					=os.path.join(_UpperCAmelCase    ,"""tmp.sql"""  )
       _a    : Tuple     					=SqlDatasetReader("""dataset"""    ,"""sqlite:///""" + sqlite_path    ,cache_dir=_UpperCAmelCase  ).read()
       SqlDatasetWriter(_UpperCAmelCase    ,"""dataset"""    ,"""sqlite:///""" + output_sqlite_path    ,num_proc=1  ).write()

       _a    : Tuple     					=iter_sql_file(_UpperCAmelCase  )
       _a    : List[Any]     					=iter_sql_file(_UpperCAmelCase  )

       for rowa, rowa in zip(_UpperCAmelCase    ,_UpperCAmelCase  ):
              assert rowa == rowa


@require_sqlalchemy
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[int]    ,_UpperCAmelCase	:	Any    ,_UpperCAmelCase	:	List[Any]  )  ->			Optional[int]:
       _a    : int     					=tmp_path / """cache"""
       _a    : Any     					=os.path.join(_UpperCAmelCase    ,"""tmp.sql"""  )
       _a    : Union[str, Any]     					=SqlDatasetReader("""dataset"""    ,"""sqlite:///""" + sqlite_path    ,cache_dir=_UpperCAmelCase  ).read()
       SqlDatasetWriter(_UpperCAmelCase    ,"""dataset"""    ,"""sqlite:///""" + output_sqlite_path    ,num_proc=2  ).write()

       _a    : List[Any]     					=iter_sql_file(_UpperCAmelCase  )
       _a    : str     					=iter_sql_file(_UpperCAmelCase  )

       for rowa, rowa in zip(_UpperCAmelCase    ,_UpperCAmelCase  ):
              assert rowa == rowa


@require_sqlalchemy
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Union[str, Any]    ,_UpperCAmelCase	:	str    ,_UpperCAmelCase	:	List[Any]  )  ->			List[str]:
       _a    : List[str]     					=tmp_path / """cache"""
       _a    : Dict     					=os.path.join(_UpperCAmelCase    ,"""tmp.sql"""  )
       _a    : Optional[Any]     					=SqlDatasetReader("""dataset"""    ,"""sqlite:///""" + sqlite_path    ,cache_dir=_UpperCAmelCase  ).read()
       with pytest.raises(_UpperCAmelCase  ):
              SqlDatasetWriter(_UpperCAmelCase    ,"""dataset"""    ,"""sqlite:///""" + output_sqlite_path    ,num_proc=0  ).write()


