




'''simple docstring'''







import math
import os
from copy import deepcopy

import datasets
import evaluate
import torch
import transformers
from datasets import load_dataset
from torch.utils.data import DataLoader
from transformers import AutoModelForSequenceClassification, AutoTokenizer

from accelerate import Accelerator
from accelerate.test_utils import RegressionDataset, RegressionModel
from accelerate.utils import is_tpu_available, set_seed


lowerCAmelCase_     						=				'''true'''







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase=82      ,UpperCAmelCase=16						):






   '''simple docstring'''







   set_seed(42						)
   A__       =     RegressionModel()
   A__       =     deepcopy(UpperCAmelCase						)
   A__       =     RegressionDataset(length=UpperCAmelCase						)
   A__       =     DataLoader(UpperCAmelCase      ,batch_size=UpperCAmelCase						)
   model.to(accelerator.device						)
   A__  ,			A__       =     accelerator.prepare(UpperCAmelCase      ,UpperCAmelCase						)
   return model, ddp_model, dataloader







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase=False						):






   '''simple docstring'''







   A__       =     AutoTokenizer.from_pretrained('hf-internal-testing/mrpc-bert-base-cased'						)
   A__       =     load_dataset('glue'      ,'mrpc'      ,split='validation'						)

   def tokenize_function(UpperCAmelCase						):
      A__       =     tokenizer(examples['sentence1']      ,examples['sentence2']      ,truncation=UpperCAmelCase      ,max_length=UpperCAmelCase						)
      return outputs

   with accelerator.main_process_first():
      A__       =     dataset.map(
          UpperCAmelCase      ,batched=UpperCAmelCase      ,remove_columns=['idx', 'sentence1', 'sentence2']      ,)

   A__       =     tokenized_datasets.rename_column('label'      ,'labels'						)

   def collate_fn(UpperCAmelCase						):
      if use_longest:
         return tokenizer.pad(UpperCAmelCase      ,padding='longest'      ,return_tensors='pt'						)
      return tokenizer.pad(UpperCAmelCase      ,padding='max_length'      ,max_length=128      ,return_tensors='pt'						)

   return DataLoader(UpperCAmelCase      ,shuffle=UpperCAmelCase      ,collate_fn=UpperCAmelCase      ,batch_size=16						)







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase						):






   '''simple docstring'''







   A__       =     Accelerator(dispatch_batches=UpperCAmelCase      ,split_batches=UpperCAmelCase						)
   A__       =     get_dataloader(UpperCAmelCase      ,not dispatch_batches						)
   A__       =     AutoModelForSequenceClassification.from_pretrained(
       'hf-internal-testing/mrpc-bert-base-cased'      ,return_dict=UpperCAmelCase						)
   A__  ,			A__       =     accelerator.prepare(UpperCAmelCase      ,UpperCAmelCase						)
   return {"ddp": [ddp_model, ddp_dataloader, "cuda:0"], "no": [model, dataloader, accelerator.device]}, accelerator







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase						):






   '''simple docstring'''







   A__       =     []
   for batch in dataloader:
      A__  ,			A__       =     batch.values()
      with torch.no_grad():
         A__       =     model(UpperCAmelCase						)
         A__  ,			A__       =     accelerator.gather_for_metrics((logit, target)						)
         logits_and_targets.append((logit, target)						)
   A__  ,			A__       =     [], []
   for logit, targ in logits_and_targets:
      logits.append(UpperCAmelCase						)
      targs.append(UpperCAmelCase						)
   A__  ,			A__       =     torch.cat(UpperCAmelCase						), torch.cat(UpperCAmelCase						)
   return logits, targs







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase=82      ,UpperCAmelCase=False      ,UpperCAmelCase=False      ,UpperCAmelCase=16						):






   '''simple docstring'''







   A__  ,			A__  ,			A__       =     get_basic_setup(UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase						)
   A__  ,			A__       =     generate_predictions(UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase						)
   assert (
       len(UpperCAmelCase						) == num_samples
   ), F"""Unexpected number of inputs:\n    Expected: {num_samples}\n    Actual: {len(UpperCAmelCase						)}"""







def 						_A      ( UpperCAmelCase = False      ,UpperCAmelCase = False						):






   '''simple docstring'''







   A__       =     evaluate.load('glue'      ,'mrpc'						)
   A__  ,			A__       =     get_mrpc_setup(UpperCAmelCase      ,UpperCAmelCase						)
   # First do baseline
   A__  ,			A__  ,			A__       =     setup['no']
   model.to(UpperCAmelCase						)
   model.eval()
   for batch in dataloader:
      batch.to(UpperCAmelCase						)
      with torch.inference_mode():
         A__       =     model(**UpperCAmelCase						)
      A__       =     outputs.logits.argmax(dim=-1						)
      metric.add_batch(predictions=UpperCAmelCase      ,references=batch['labels']						)
   A__       =     metric.compute()

   # Then do distributed
   A__  ,			A__  ,			A__       =     setup['ddp']
   model.eval()
   for batch in dataloader:
      with torch.inference_mode():
         A__       =     model(**UpperCAmelCase						)
      A__       =     outputs.logits.argmax(dim=-1						)
      A__       =     batch['labels']
      A__  ,			A__       =     accelerator.gather_for_metrics((preds, references)						)
      metric.add_batch(predictions=UpperCAmelCase      ,references=UpperCAmelCase						)
   A__       =     metric.compute()

   for key in "accuracy f1".split():
      assert math.isclose(
          baseline[key]      ,distributed[key]						), F"""Baseline and Distributed are not the same for key {key}:\n\tBaseline: {baseline[key]}\n\tDistributed: {distributed[key]}\n"""







def 						_A      ( ):






   '''simple docstring'''







   A__       =     Accelerator(split_batches=UpperCAmelCase      ,dispatch_batches=UpperCAmelCase						)
   if accelerator.is_local_main_process:
      datasets.utils.logging.set_verbosity_warning()
      transformers.utils.logging.set_verbosity_warning()
   else:
      datasets.utils.logging.set_verbosity_error()
      transformers.utils.logging.set_verbosity_error()
   # These are a bit slower so they should only be ran on the GPU or TPU
   if torch.cuda.is_available() or is_tpu_available():
      if accelerator.is_local_main_process:
         print('**Testing gather_for_metrics**'						)
      for split_batches in [True, False]:
         for dispatch_batches in [True, False]:
            if accelerator.is_local_main_process:
               print(F"""With: `split_batches={split_batches}`, `dispatch_batches={dispatch_batches}`"""						)
            test_mrpc(UpperCAmelCase      ,UpperCAmelCase						)
            accelerator.state._reset_state()
   if accelerator.is_local_main_process:
      print('**Test torch metrics**'						)
   for split_batches in [True, False]:
      for dispatch_batches in [True, False]:
         A__       =     Accelerator(split_batches=UpperCAmelCase      ,dispatch_batches=UpperCAmelCase						)
         if accelerator.is_local_main_process:
            print(F"""With: `split_batches={split_batches}`, `dispatch_batches={dispatch_batches}`, length=99"""						)
         test_torch_metrics(UpperCAmelCase      ,99						)
         accelerator.state._reset_state()
   if accelerator.is_local_main_process:
      print('**Test last batch is not dropped when perfectly divisible**'						)
   A__       =     Accelerator()
   test_torch_metrics(UpperCAmelCase      ,512						)
   accelerator.state._reset_state()







def 						_A      ( UpperCAmelCase						):






   '''simple docstring'''







   main()


if __name__ == "__main__":
    main()








'''simple docstring'''







import os

import torch

from ..logging import get_logger
from .constants import FSDP_PYTORCH_VERSION, MODEL_NAME, OPTIMIZER_NAME
from .versions import is_torch_version


if is_torch_version('''>=''', FSDP_PYTORCH_VERSION):
    import torch.distributed.checkpoint as dist_cp
    from torch.distributed.checkpoint.default_planner import DefaultLoadPlanner, DefaultSavePlanner
    from torch.distributed.checkpoint.optimizer import load_sharded_optimizer_state_dict
    from torch.distributed.fsdp.fully_sharded_data_parallel import FullyShardedDataParallel as FSDP
    from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType


lowerCAmelCase_     						=				get_logger(__name__)







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase=0						):






   '''simple docstring'''







   os.makedirs(UpperCAmelCase      ,exist_ok=UpperCAmelCase						)
   with FSDP.state_dict_type(
       UpperCAmelCase      ,fsdp_plugin.state_dict_type      ,fsdp_plugin.state_dict_config      ,fsdp_plugin.optim_state_dict_config						):
      A__       =     model.state_dict()
      if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
         A__       =     F"""{MODEL_NAME}.bin""" if model_index == 0 else F"""{MODEL_NAME}_{model_index}.bin"""
         A__       =     os.path.join(UpperCAmelCase      ,UpperCAmelCase						)
         if accelerator.process_index == 0:
            logger.info(F"""Saving model to {output_model_file}"""						)
            torch.save(UpperCAmelCase      ,UpperCAmelCase						)
            logger.info(F"""Model saved to {output_model_file}"""						)
      elif fsdp_plugin.state_dict_type == StateDictType.LOCAL_STATE_DICT:
         A__       =     (
             F"""{MODEL_NAME}_rank{accelerator.process_index}.bin"""
             if model_index == 0
             else F"""{MODEL_NAME}_{model_index}_rank{accelerator.process_index}.bin"""
         )
         A__       =     os.path.join(UpperCAmelCase      ,UpperCAmelCase						)
         logger.info(F"""Saving model to {output_model_file}"""						)
         torch.save(UpperCAmelCase      ,UpperCAmelCase						)
         logger.info(F"""Model saved to {output_model_file}"""						)
      elif fsdp_plugin.state_dict_type == StateDictType.SHARDED_STATE_DICT:
         A__       =     os.path.join(UpperCAmelCase      ,F"""{MODEL_NAME}_{model_index}"""						)
         os.makedirs(UpperCAmelCase      ,exist_ok=UpperCAmelCase						)
         logger.info(F"""Saving model to {ckpt_dir}"""						)
         A__       =     {'model': state_dict}

         dist_cp.save_state_dict(
             state_dict=UpperCAmelCase      ,storage_writer=dist_cp.FileSystemWriter(UpperCAmelCase						)      ,planner=DefaultSavePlanner()      ,)
         logger.info(F"""Model saved to {ckpt_dir}"""						)







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase=0						):






   '''simple docstring'''







   accelerator.wait_for_everyone()
   with FSDP.state_dict_type(
       UpperCAmelCase      ,fsdp_plugin.state_dict_type      ,fsdp_plugin.state_dict_config      ,fsdp_plugin.optim_state_dict_config						):
      if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
         if type(UpperCAmelCase						) != FSDP and accelerator.process_index != 0:
            if not fsdp_plugin.sync_module_states:
               raise ValueError(
                   'Set the `sync_module_states` flag to `True` so that model states are synced across processes when '
                   'initializing FSDP object'						)
            return
         A__       =     F"""{MODEL_NAME}.bin""" if model_index == 0 else F"""{MODEL_NAME}_{model_index}.bin"""
         A__       =     os.path.join(UpperCAmelCase      ,UpperCAmelCase						)
         logger.info(F"""Loading model from {input_model_file}"""						)
         A__       =     torch.load(UpperCAmelCase						)
         logger.info(F"""Model loaded from {input_model_file}"""						)
      elif fsdp_plugin.state_dict_type == StateDictType.LOCAL_STATE_DICT:
         A__       =     (
             F"""{MODEL_NAME}_rank{accelerator.process_index}.bin"""
             if model_index == 0
             else F"""{MODEL_NAME}_{model_index}_rank{accelerator.process_index}.bin"""
         )
         A__       =     os.path.join(UpperCAmelCase      ,UpperCAmelCase						)
         logger.info(F"""Loading model from {input_model_file}"""						)
         A__       =     torch.load(UpperCAmelCase						)
         logger.info(F"""Model loaded from {input_model_file}"""						)
      elif fsdp_plugin.state_dict_type == StateDictType.SHARDED_STATE_DICT:
         A__       =     (
             os.path.join(UpperCAmelCase      ,F"""{MODEL_NAME}_{model_index}"""						)
             if F"""{MODEL_NAME}""" not in input_dir
             else input_dir
         )
         logger.info(F"""Loading model from {ckpt_dir}"""						)
         A__       =     {'model': model.state_dict()}
         dist_cp.load_state_dict(
             state_dict=UpperCAmelCase      ,storage_reader=dist_cp.FileSystemReader(UpperCAmelCase						)      ,planner=DefaultLoadPlanner()      ,)
         A__       =     state_dict['model']
         logger.info(F"""Model loaded from {ckpt_dir}"""						)
      model.load_state_dict(UpperCAmelCase						)







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase=0						):






   '''simple docstring'''







   os.makedirs(UpperCAmelCase      ,exist_ok=UpperCAmelCase						)
   with FSDP.state_dict_type(
       UpperCAmelCase      ,fsdp_plugin.state_dict_type      ,fsdp_plugin.state_dict_config      ,fsdp_plugin.optim_state_dict_config						):
      A__       =     FSDP.optim_state_dict(UpperCAmelCase      ,UpperCAmelCase						)
      if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
         if accelerator.process_index == 0:
            A__       =     (
                F"""{OPTIMIZER_NAME}.bin""" if optimizer_index == 0 else F"""{OPTIMIZER_NAME}_{optimizer_index}.bin"""
            )
            A__       =     os.path.join(UpperCAmelCase      ,UpperCAmelCase						)
            logger.info(F"""Saving Optimizer state to {output_optimizer_file}"""						)
            torch.save(UpperCAmelCase      ,UpperCAmelCase						)
            logger.info(F"""Optimizer state saved in {output_optimizer_file}"""						)
      else:
         A__       =     os.path.join(UpperCAmelCase      ,F"""{OPTIMIZER_NAME}_{optimizer_index}"""						)
         os.makedirs(UpperCAmelCase      ,exist_ok=UpperCAmelCase						)
         logger.info(F"""Saving Optimizer state to {ckpt_dir}"""						)
         dist_cp.save_state_dict(
             state_dict={'optimizer': optim_state}      ,storage_writer=dist_cp.FileSystemWriter(UpperCAmelCase						)      ,planner=DefaultSavePlanner()      ,)
         logger.info(F"""Optimizer state saved in {ckpt_dir}"""						)







def 						_A      ( UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase=0						):






   '''simple docstring'''







   accelerator.wait_for_everyone()
   with FSDP.state_dict_type(
       UpperCAmelCase      ,fsdp_plugin.state_dict_type      ,fsdp_plugin.state_dict_config      ,fsdp_plugin.optim_state_dict_config						):
      if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
         A__       =     None
         # below check should work but currently it isn't working (mostly opytorch issue),
         # in the meantime disabling it at the cost of excess memory usage
         # if accelerator.process_index == 0 or not fsdp_plugin.optim_state_dict_config.rank0_only:
         A__       =     (
             F"""{OPTIMIZER_NAME}.bin""" if optimizer_index == 0 else F"""{OPTIMIZER_NAME}_{optimizer_index}.bin"""
         )
         A__       =     os.path.join(UpperCAmelCase      ,UpperCAmelCase						)
         logger.info(F"""Loading Optimizer state from {input_optimizer_file}"""						)
         A__       =     torch.load(UpperCAmelCase						)
         logger.info(F"""Optimizer state loaded from {input_optimizer_file}"""						)
      else:
         A__       =     (
             os.path.join(UpperCAmelCase      ,F"""{OPTIMIZER_NAME}_{optimizer_index}"""						)
             if F"""{OPTIMIZER_NAME}""" not in input_dir
             else input_dir
         )
         logger.info(F"""Loading Optimizer from {ckpt_dir}"""						)
         A__       =     load_sharded_optimizer_state_dict(
             model_state_dict=model.state_dict()      ,optimizer_key='optimizer'      ,storage_reader=dist_cp.FileSystemReader(UpperCAmelCase						)      ,)
         A__       =     optim_state['optimizer']
         logger.info(F"""Optimizer loaded from {ckpt_dir}"""						)
      A__       =     FSDP.optim_state_dict_to_load(UpperCAmelCase      ,UpperCAmelCase      ,UpperCAmelCase						)
      optimizer.load_state_dict(UpperCAmelCase						)


