






'''simple docstring'''



from copy import deepcopy

import torch
import torch.nn.functional as F
from torch.optim import AdamW
from torch.optim.lr_scheduler import LambdaLR
from torch.utils.data import DataLoader

from accelerate.accelerator import Accelerator
from accelerate.state import GradientState
from accelerate.test_utils import RegressionDataset, RegressionModel
from accelerate.utils import DistributedType, is_torch_version, set_seed


def 							lowercase_				(    __A :       Optional[Any]					,      __A :       List[str]					,      __A :       Any					,      __A :       Optional[int]			)       ->      Dict:



 """simple docstring"""




 for param, grad_param in zip(model_a.parameters()					,      model_b.parameters()			):
  if not param.requires_grad:
   continue
  if not did_step:
   # Grads should not be in sync
   assert (
       torch.allclose(param.grad					,      grad_param.grad			) is False
   ), F'Gradients in sync when they should not be at iteration {iteration}:\nmodel_a grad ({param.grad}) == model_b grad ({grad_param.grad})'
  else:
   # Grads should be in sync
   assert (
       torch.allclose(param.grad					,      grad_param.grad			) is True
   ), F'Gradients not in sync when they should be at iteration {iteration}:\nmodel_a grad ({param.grad}) != model_b grad ({grad_param.grad})'


def 							lowercase_				(    __A :       str					,      __A :       Any					,      __A :       Optional[int]					,      __A :       Tuple					,      __A :       Optional[Any]=True			)       ->      str:



 """simple docstring"""




 model.train()
 lowercase					:   List[str]				  =model(__A			)
 lowercase					:   str				  =F.mse_loss(__A					,      target.to(output.device			)			)
 if not do_backward:
  loss /= accelerator.gradient_accumulation_steps
  loss.backward()
 else:
  accelerator.backward(__A			)


def 							lowercase_				(    __A :       Dict					,      __A :       Optional[Any]=False			)       ->      Optional[Any]:



 """simple docstring"""




 set_seed(4_2			)
 lowercase					:   str				  =RegressionModel()
 lowercase					:   str				  =deepcopy(__A			)
 lowercase					:   Dict				  =RegressionDataset(length=8_0			)
 lowercase					:   Union[str, Any]				  =DataLoader(__A					,      batch_size=1_6			)
 model.to(accelerator.device			)
 if sched:
  lowercase					:   int				  =AdamW(params=model.parameters()					,      lr=1E-3			)
  lowercase					:   Dict				  =AdamW(params=ddp_model.parameters()					,      lr=1E-3			)
  lowercase					:   List[str]				  =LambdaLR(__A					,      lr_lambda=lambda __A			: epoch**0.65			)
  lowercase					:   str				  =LambdaLR(__A					,      lr_lambda=lambda __A			: epoch**0.65			)
 # Make a copy of `model`
 if sched:
  lowercase  ,		lowercase  ,		lowercase  ,		lowercase					:   List[Any]				  =accelerator.prepare(__A					,      __A					,      __A					,      __A			)
 else:
  lowercase  ,		lowercase					:   Any				  =accelerator.prepare(__A					,      __A			)
 if sched:
  return (model, opt, sched, dataloader, ddp_model, ddp_opt, ddp_sched)
 return model, ddp_model, dataloader


def 							lowercase_				(    __A :       List[str]			)       ->      Tuple:



 """simple docstring"""




 lowercase  ,		lowercase  ,		lowercase					:   str				  =get_training_setup(__A			)
 # Use a single batch
 lowercase  ,		lowercase					:   Optional[int]				  =next(iter(__A			)			).values()
 for iteration in range(3			):
  # Gather the distributed inputs and targs for the base model
  lowercase  ,		lowercase					:   List[Any]				  =accelerator.gather((ddp_input, ddp_target)			)
  lowercase  ,		lowercase					:   Any				  =input.to(accelerator.device			), target.to(accelerator.device			)
  # Perform our initial ground truth step in non "DDP"
  step_model(__A					,      __A					,      __A					,      __A			)
  # Do "gradient accumulation" (noop)
  if iteration % 2 == 0:
   # Accumulate grads locally
   with accelerator.no_sync(__A			):
    step_model(__A					,      __A					,      __A					,      __A			)
  else:
   # Sync grads
   step_model(__A					,      __A					,      __A					,      __A			)

  # Since `no_sync` is a noop, `ddp_model` and `model` grads should always be in sync
  check_model_parameters(__A					,      __A					,      __A					,      __A			)
  for param, ddp_param in zip(model.parameters()					,      ddp_model.parameters()			):
   if not param.requires_grad:
    continue
   assert torch.allclose(
       param.grad					,      ddp_param.grad			), F'Gradients not in sync when they should be:\nModel grad ({param.grad}) != DDP grad ({ddp_param.grad})'

  # Shuffle ddp_input on each iteration
  torch.manual_seed(1_3_3_7 + iteration			)
  lowercase					:   Any				  =ddp_input[torch.randperm(len(__A			)			)]


def 							lowercase_				(    __A :       Optional[int]			)       ->      Optional[int]:



 """simple docstring"""




 lowercase  ,		lowercase  ,		lowercase					:   List[Any]				  =get_training_setup(__A			)
 # Use a single batch
 lowercase  ,		lowercase					:   Optional[Any]				  =next(iter(__A			)			).values()
 for iteration in range(3			):
  # Gather the distributed inputs and targs for the base model
  lowercase  ,		lowercase					:   Tuple				  =accelerator.gather((ddp_input, ddp_target)			)
  lowercase  ,		lowercase					:   List[str]				  =input.to(accelerator.device			), target.to(accelerator.device			)
  # Perform our initial ground truth step in non "DDP"
  step_model(__A					,      __A					,      __A					,      __A			)
  # Do "gradient accumulation" (noop)
  if iteration % 2 == 0:
   # Accumulate grads locally
   with accelerator.no_sync(__A			):
    step_model(__A					,      __A					,      __A					,      __A			)
  else:
   # Sync grads
   step_model(__A					,      __A					,      __A					,      __A			)

  # DDP model and model should only be in sync when not (iteration % 2 == 0)
  for param, ddp_param in zip(model.parameters()					,      ddp_model.parameters()			):
   if not param.requires_grad:
    continue
   if iteration % 2 == 0:
    # Grads should not be in sync
    assert (
        torch.allclose(param.grad					,      ddp_param.grad			) is False
    ), F'Gradients in sync when they should not be:\nModel grad ({param.grad}) == DDP grad ({ddp_param.grad})'
   else:
    # Grads should be in sync
    assert (
        torch.allclose(param.grad					,      ddp_param.grad			) is True
    ), F'Gradients not in sync when they should be:\nModel grad ({param.grad}) != DDP grad ({ddp_param.grad})'

        # Shuffle ddp_input on each iteration
  torch.manual_seed(1_3_3_7 + iteration			)
  lowercase					:   int				  =ddp_input[torch.randperm(len(__A			)			)]


def 							lowercase_				(    __A :       Dict=False					,      __A :       Tuple=False			)       ->      Union[str, Any]:



 """simple docstring"""




 lowercase					:   List[Any]				  =Accelerator(
     split_batches=__A					,      dispatch_batches=__A					,      gradient_accumulation_steps=2			)
 # Test that context manager behaves properly
 lowercase  ,		lowercase  ,		lowercase					:   Dict				  =get_training_setup(__A			)
 for iteration, batch in enumerate(__A			):
  lowercase  ,		lowercase					:   Optional[int]				  =batch.values()
  # Gather the distributed inputs and targs for the base model
  lowercase  ,		lowercase					:   Tuple				  =accelerator.gather((ddp_input, ddp_target)			)
  lowercase  ,		lowercase					:   Dict				  =input.to(accelerator.device			), target.to(accelerator.device			)
  # Perform our initial ground truth step in non "DDP"
  step_model(__A					,      __A					,      __A					,      __A					,      __A			)
  # Do "gradient accumulation" (noop)
  with accelerator.accumulate(__A			):
   step_model(__A					,      __A					,      __A					,      __A			)

  # DDP model and model should only be in sync when not (iteration % 2 == 0)
  for param, ddp_param in zip(model.parameters()					,      ddp_model.parameters()			):
   if not param.requires_grad:
    continue
   if ((iteration + 1) % 2 == 0) or (iteration == len(__A			) - 1):
    # Grads should be in sync
    assert (
        torch.allclose(param.grad					,      ddp_param.grad			) is True
    ), F'Gradients not in sync when they should be at iteration {iteration}:\nModel grad ({param.grad}) != DDP grad ({ddp_param.grad})'
   else:
    # Grads should not be in sync
    assert (
        torch.allclose(param.grad					,      ddp_param.grad			) is False
    ), F'Gradients in sync when they should not be at iteration {iteration}:\nModel grad ({param.grad}) == DDP grad ({ddp_param.grad})'

        # Shuffle ddp_input on each iteration
  torch.manual_seed(1_3_3_7 + iteration			)
  lowercase					:   Optional[int]				  =ddp_input[torch.randperm(len(__A			)			)]
 GradientState._reset_state()


def 							lowercase_				(    __A :       Optional[Any]=False					,      __A :       List[str]=False			)       ->      Union[str, Any]:



 """simple docstring"""




 lowercase					:   int				  =Accelerator(
     split_batches=__A					,      dispatch_batches=__A					,      gradient_accumulation_steps=2			)
 # Test that context manager behaves properly
 lowercase  ,		lowercase  ,		lowercase  ,		lowercase  ,		lowercase  ,		lowercase  ,		lowercase					:   Optional[int]				  =get_training_setup(__A					,      __A			)
 for iteration, batch in enumerate(__A			):
  lowercase  ,		lowercase					:   int				  =batch.values()
  # Gather the distributed inputs and targs for the base model
  lowercase  ,		lowercase					:   Any				  =accelerator.gather((ddp_input, ddp_target)			)
  lowercase  ,		lowercase					:   Optional[int]				  =input.to(accelerator.device			), target.to(accelerator.device			)
  # Perform our initial ground truth step in non "DDP"
  model.train()
  ddp_model.train()
  step_model(__A					,      __A					,      __A					,      __A					,      __A			)
  opt.step()

  if ((iteration + 1) % 2 == 0) or ((iteration + 1) == len(__A			)):
   if split_batches:
    sched.step()
   else:
    for _ in range(accelerator.num_processes			):
     sched.step()
  opt.zero_grad()
  # Perform gradient accumulation under wrapper
  with accelerator.accumulate(__A			):
   step_model(__A					,      __A					,      __A					,      __A			)
   ddp_opt.step()
   ddp_sched.step()
   ddp_opt.zero_grad()

  # Learning rates should be the same
  assert (
      opt.param_groups[0]["lr"] == ddp_opt.param_groups[0]["lr"]
  ), F'Learning rates found in each optimizer did not align\nopt: {opt.param_groups[0]["lr"]}\nDDP opt: {ddp_opt.param_groups[0]["lr"]}\n'
  lowercase					:   Union[str, Any]				  =(((iteration + 1) % 2) == 0) or ((iteration + 1) == len(__A			))
  if accelerator.num_processes > 1:
   check_model_parameters(__A					,      __A					,      __A					,      __A			)
  # Shuffle ddp_input on each iteration
  torch.manual_seed(1_3_3_7 + iteration			)
 GradientState._reset_state()


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



 """simple docstring"""




 lowercase					:   Tuple				  =Accelerator()

 lowercase					:   str				  =RegressionDataset(length=8_0			)
 lowercase					:   Tuple				  =DataLoader(__A					,      batch_size=1_6			)
 lowercase					:   Tuple				  =RegressionDataset(length=9_6			)
 lowercase					:   Tuple				  =DataLoader(__A					,      batch_size=1_6			)
 lowercase  ,		lowercase					:   int				  =accelerator.prepare(__A					,      __A			)
 assert accelerator.gradient_state.active_dataloader is None
 for iteration, _ in enumerate(__A			):
  assert id(accelerator.gradient_state.active_dataloader			) == id(__A			)
  if iteration < len(__A			) - 1:
   assert not accelerator.gradient_state.end_of_dataloader
   if iteration == 1:
    for batch_num, _ in enumerate(__A			):
     assert id(accelerator.gradient_state.active_dataloader			) == id(__A			)
     if batch_num < len(__A			) - 1:
      assert not accelerator.gradient_state.end_of_dataloader
     else:
      assert accelerator.gradient_state.end_of_dataloader
  else:
   assert accelerator.gradient_state.end_of_dataloader
 assert accelerator.gradient_state.active_dataloader is None


def 							lowercase_				(    )       ->      Tuple:



 """simple docstring"""




 lowercase					:   List[str]				  =Accelerator()
 lowercase					:   int				  =accelerator.state
 if state.local_process_index == 0:
  print('''**Test `accumulate` gradient accumulation with dataloader break**'''			)
 test_dataloader_break()
 if state.distributed_type == DistributedType.NO:
  if state.local_process_index == 0:
   print('''**Test NOOP `no_sync` context manager**'''			)
  test_noop_sync(__A			)
 if state.distributed_type in (DistributedType.MULTI_GPU, DistributedType.MULTI_CPU):
  if state.local_process_index == 0:
   print('''**Test Distributed `no_sync` context manager**'''			)
  test_distributed_sync(__A			)
 if state.distributed_type == DistributedType.MULTI_GPU:
  for split_batch in [True, False]:
   for dispatch_batches in [True, False]:
    if state.local_process_index == 0:
     print(
         '''**Test `accumulate` gradient accumulation, '''					,      F'`split_batches={split_batch}` and `dispatch_batches={dispatch_batches}`**'					,      )
    test_gradient_accumulation(__A					,      __A			)

    # Currently will break on torch 2.0 +, need to investigate why
 if is_torch_version('''<'''					,      '''2.0'''			) or state.distributed_type == DistributedType.NO:
  if state.local_process_index == 0:
   print(
       '''**Test `accumulate` gradient accumulation with optimizer and scheduler, '''					,      '''`split_batches=False`, `dispatch_batches=False`**'''					,      )
  test_gradient_accumulation_with_opt_and_scheduler()
  if state.distributed_type == DistributedType.MULTI_GPU:
   for split_batch in [True, False]:
    for dispatch_batches in [True, False]:
     if not split_batch and not dispatch_batches:
      continue
     if state.local_process_index == 0:
      print(
          '''**Test `accumulate` gradient accumulation with optimizer and scheduler, '''					,      F'`split_batches={split_batch}` and `dispatch_batches={dispatch_batches}`**'					,      )
     test_gradient_accumulation_with_opt_and_scheduler(__A					,      __A			)


def 							lowercase_				(    __A :       Tuple			)       ->      Tuple:



 """simple docstring"""




 main()


if __name__ == "__main__":
      main()











'''simple docstring'''



from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


SCREAMING_SNAKE_CASE						=	{
    'configuration_mega': ['MEGA_PRETRAINED_CONFIG_ARCHIVE_MAP', 'MegaConfig', 'MegaOnnxConfig'],
}

try:
      if not is_torch_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      SCREAMING_SNAKE_CASE						=	[
          'MEGA_PRETRAINED_MODEL_ARCHIVE_LIST',
          'MegaForCausalLM',
          'MegaForMaskedLM',
          'MegaForMultipleChoice',
          'MegaForQuestionAnswering',
          'MegaForSequenceClassification',
          'MegaForTokenClassification',
          'MegaModel',
          'MegaPreTrainedModel',
      ]

if TYPE_CHECKING:
      from .configuration_mega import MEGA_PRETRAINED_CONFIG_ARCHIVE_MAP, MegaConfig, MegaOnnxConfig

      try:
            if not is_torch_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_mega import (
                MEGA_PRETRAINED_MODEL_ARCHIVE_LIST,
                MegaForCausalLM,
                MegaForMaskedLM,
                MegaForMultipleChoice,
                MegaForQuestionAnswering,
                MegaForSequenceClassification,
                MegaForTokenClassification,
                MegaModel,
                MegaPreTrainedModel,
            )

else:
      import sys

      SCREAMING_SNAKE_CASE						=	_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)



