


# This test is meant to be run in on an instance with TPUs like this:
#
#   python examples/pytorch/xla_spawn.py --num_cores=8 tests/test_trainer_tpu.py
#
# Replace 8 with the number of TPU cores you have.
#

import sys
from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.utils import logging


snake_case			:      Optional[int] 							=   logging.get_logger(__name__)


if is_torch_available():
							import torch
							from torch import nn
							from torch.utils.data import Dataset

							from transformers import Trainer



							class    __lowercase  (       UpperCamelCase							):

														"""simple docstring"""


														def __init__(    self				, A_ = 101	)->	Optional[Any]:
																		_SCREAMING_SNAKE_CASE      = length





														def __len__(    self	)->	Optional[Any]:
																		return self.length





														def __getitem__(    self				, A_	)->	int:
																		return i







							class    __lowercase  :

														"""simple docstring"""


														def __call__(    self				, A_	)->	Tuple:
																		return {"input_ids": torch.tensor(A_	), "labels": torch.tensor(A_	)}







							class    __lowercase  (       nn.Module							):

														"""simple docstring"""


														def __init__(    self	)->	Tuple:
																		super().__init__()
																		# Add some (unused) params otherwise DDP will complain.
																		_SCREAMING_SNAKE_CASE      = nn.Linear(120				, 80	)





														def 		__magic_name__			(    self				, A_				, A_=None	)->	Tuple:
																		if labels is not None:
																						return torch.tensor(0.0				, device=input_ids.device	), input_ids
																		else:
																						return input_ids


def 				SCREAMING_SNAKE_CASE (     ):

				"""simple docstring"""






				_SCREAMING_SNAKE_CASE      = HfArgumentParser((TrainingArguments,)							)
				sys.argv += ["--output_dir", "./examples"]
				_SCREAMING_SNAKE_CASE      = parser.parse_args_into_dataclasses()[0]

				logger.warning(
				    f'''Process rank: {training_args.local_rank}, device: {training_args.device}, '''
				    f'''tpu_num_cores: {training_args.tpu_num_cores}'''   ,)

				# Essentially, what we want to verify in the distributed case is
				# that we get all samples back, in the right order.
				# (this is crucial for prediction for instance)
				for dataset_length in [1001, 256, 15]:
								_SCREAMING_SNAKE_CASE      = DummyDataset(UpperCAmelCase__							)

								def compute_metrics(UpperCAmelCase__							) -> Dict:
												_SCREAMING_SNAKE_CASE      = list(range(len(UpperCAmelCase__							)							)							)
												_SCREAMING_SNAKE_CASE      = p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
												return {"success": success}

								_SCREAMING_SNAKE_CASE      = Trainer(
								    model=DummyModel()   ,args=UpperCAmelCase__   ,data_collator=DummyDataCollator()   ,eval_dataset=UpperCAmelCase__   ,compute_metrics=UpperCAmelCase__   ,)
								_SCREAMING_SNAKE_CASE      = trainer.evaluate()
								logger.info(UpperCAmelCase__							)
								if metrics["eval_success"] is not True:
												logger.error(UpperCAmelCase__							)
												exit(1							)

								_SCREAMING_SNAKE_CASE      = trainer.predict(UpperCAmelCase__							)
								logger.info(p.metrics							)
								if p.metrics["test_success"] is not True:
												logger.error(p.metrics							)
												exit(1							)

								_SCREAMING_SNAKE_CASE      = 2

								_SCREAMING_SNAKE_CASE      = trainer.evaluate()
								logger.info(UpperCAmelCase__							)
								if metrics["eval_success"] is not True:
												logger.error(UpperCAmelCase__							)
												exit(1							)

								_SCREAMING_SNAKE_CASE      = trainer.predict(UpperCAmelCase__							)
								logger.info(p.metrics							)
								if p.metrics["test_success"] is not True:
												logger.error(p.metrics							)
												exit(1							)

								_SCREAMING_SNAKE_CASE      = None

				logger.info('🔥 All distributed tests successful'							)


def 				SCREAMING_SNAKE_CASE (     UpperCAmelCase__							):

				"""simple docstring"""






				main()


if __name__ == "__main__":
							main()




import inspect
import warnings

import torch

from .state import AcceleratorState, GradientState
from .utils import DistributedType, honor_type, is_tpu_available


if is_tpu_available(check_device=False):
							import torch_xla.core.xla_model as xm


def 				SCREAMING_SNAKE_CASE (     UpperCAmelCase__   ,UpperCAmelCase__							):

				"""simple docstring"""






				if isinstance(UpperCAmelCase__   ,(list, tuple)							):
								return honor_type(UpperCAmelCase__   ,(move_to_device(UpperCAmelCase__   ,UpperCAmelCase__							) for t in state)							)
				elif isinstance(UpperCAmelCase__   ,UpperCAmelCase__							):
								return type(UpperCAmelCase__							)({k: move_to_device(UpperCAmelCase__   ,UpperCAmelCase__							) for k, v in state.items()}							)
				elif isinstance(UpperCAmelCase__   ,torch.Tensor							):
								return state.to(UpperCAmelCase__							)
				return state



class    __lowercase  (       torch.optim.Optimizer							):

							"""simple docstring"""


							def __init__(    self				, A_				, A_=True				, A_=None	)->	List[str]:
											_SCREAMING_SNAKE_CASE      = optimizer
											_SCREAMING_SNAKE_CASE      = scaler
											_SCREAMING_SNAKE_CASE      = AcceleratorState()
											_SCREAMING_SNAKE_CASE      = GradientState()
											_SCREAMING_SNAKE_CASE      = device_placement
											_SCREAMING_SNAKE_CASE      = False
											_SCREAMING_SNAKE_CASE      = None

											# Handle device placement
											if device_placement:
															_SCREAMING_SNAKE_CASE      = self.optimizer.state_dict()
															if self.accelerator_state.distributed_type == DistributedType.TPU:
																			xm.send_cpu_data_to_device(A_				, self.accelerator_state.device	)
															else:
																			_SCREAMING_SNAKE_CASE      = move_to_device(A_				, self.accelerator_state.device	)
															self.optimizer.load_state_dict(A_	)





							@property
							def 		__magic_name__			(    self	)->	Tuple:
											return self.optimizer.state





							@state.setter
							def 		__magic_name__			(    self				, A_	)->	str:
											_SCREAMING_SNAKE_CASE      = state





							@property
							def 		__magic_name__			(    self	)->	Optional[Any]:
											return self.optimizer.param_groups





							@param_groups.setter
							def 		__magic_name__			(    self				, A_	)->	Tuple:
											_SCREAMING_SNAKE_CASE      = param_groups





							@property
							def 		__magic_name__			(    self	)->	Any:
											return self.optimizer.defaults





							@defaults.setter
							def 		__magic_name__			(    self				, A_	)->	Dict:
											_SCREAMING_SNAKE_CASE      = defaults





							def 		__magic_name__			(    self				, A_	)->	Any:
											self.optimizer.add_param_group(A_	)





							def 		__magic_name__			(    self				, A_	)->	int:
											if self.accelerator_state.distributed_type == DistributedType.TPU and self.device_placement:
															xm.send_cpu_data_to_device(A_				, self.accelerator_state.device	)
											self.optimizer.load_state_dict(A_	)





							def 		__magic_name__			(    self	)->	List[str]:
											return self.optimizer.state_dict()





							def 		__magic_name__			(    self				, A_=None	)->	Any:
											if self.gradient_state.sync_gradients:
															_SCREAMING_SNAKE_CASE      = 'set_to_none' in inspect.signature(self.optimizer.zero_grad	).parameters
															if accept_arg:
																			if set_to_none is None:
																							_SCREAMING_SNAKE_CASE      = False
																			self.optimizer.zero_grad(set_to_none=A_	)
															else:
																			if set_to_none is not None:
																							raise ValueError('`set_to_none` for Optimizer.zero_grad` is not supported by this optimizer.'	)
																			self.optimizer.zero_grad()





							def 		__magic_name__			(    self				, A_=None	)->	Optional[int]:
											if self.gradient_state.sync_gradients:
															if self.accelerator_state.distributed_type == DistributedType.TPU:
																			_SCREAMING_SNAKE_CASE      = {'closure': closure} if closure is not None else {}
																			xm.optimizer_step(self.optimizer				, optimizer_args=A_	)
															elif self.scaler is not None:
																			_SCREAMING_SNAKE_CASE      = False
																			if self._last_scale is None:
																							# `get_scale` is an async operation requiring full synchronization
																							# on CPU and GPUs before finishing. As a result, we store away
																							# the prior one to reduce the call overhead
																							_SCREAMING_SNAKE_CASE      = self.scaler.get_scale()
																							_SCREAMING_SNAKE_CASE      = True
																			self.scaler.step(self.optimizer				, A_	)
																			self.scaler.update()
																			_SCREAMING_SNAKE_CASE      = self.scaler.get_scale()
																			if not new_scale:
																							# If we reduced the loss scale, it means the optimizer step was skipped because of gradient overflow.
																							_SCREAMING_SNAKE_CASE      = scale_after < self._last_scale
																			_SCREAMING_SNAKE_CASE      = scale_after

															else:
																			self.optimizer.step(A_	)





							def 		__magic_name__			(    self				, A_	)->	Dict:
											for param_group in self.optimizer.param_groups:
															_SCREAMING_SNAKE_CASE      = [parameters_map.get(A_				, A_	) for p in param_group['params']]





							@property
							def 		__magic_name__			(    self	)->	str:
											warnings.warn(
											    'The `is_overflow` property is deprecated and will be removed in version 1.0 of Accelerate use '
											    '`optimizer.step_was_skipped` instead.'				, A_				, )
											return self._is_overflow





							@property
							def 		__magic_name__			(    self	)->	int:
											return self._is_overflow





							def __getstate__(    self	)->	Dict:
											return self.__dict__.copy()





							def __setstate__(    self				, A_	)->	List[str]:
											self.__dict__.update(A_	)
