
'''simple docstring'''


import warnings
from typing import List
from unittest.mock import Mock

import torch
from torch.utils.data import DataLoader, IterableDataset, TensorDataset

from accelerate.accelerator import Accelerator
from accelerate.utils.dataclasses import DistributedType

class 	__lowercase				(  __UpperCAmelCase    ):

			def __init__(self			,   A      ):
								lowerCamelCase_					:      Dict     =		data






			def __iter__(self      ):







								for element in self.data:
													yield element




def lowercase_	(   _lowercase=True				)     ->							List[Any]:
					'''simple docstring'''

					lowerCamelCase_					:      str     =		Accelerator(even_batches=__lowerCAmelCase				)
					assert accelerator.num_processes == 2, "this script expects that two GPUs are available"
					return accelerator




def lowercase_	(   _lowercase			,       _lowercase			,       _lowercase			,       _lowercase = False				)     ->							Dict:
					'''simple docstring'''

					if iterable:
										lowerCamelCase_					:      Optional[Any]     =		DummyIterableDataset(torch.as_tensor(range(__lowerCAmelCase				)				)				)
					else:
										lowerCamelCase_					:      str     =		TensorDataset(torch.as_tensor(range(__lowerCAmelCase				)				)				)

					lowerCamelCase_					:      List[str]     =		DataLoader(__lowerCAmelCase			,       batch_size=__lowerCAmelCase				)
					lowerCamelCase_					:      Dict     =		accelerator.prepare(__lowerCAmelCase				)

					return dl




def lowercase_	(   _lowercase			,       _lowercase			,       _lowercase			,       _lowercase			,       _lowercase			,       )     ->							Optional[int]:
					'''simple docstring'''

					lowerCamelCase_					:      Union[str, Any]     =		create_dataloader(accelerator=__lowerCAmelCase			,       dataset_size=__lowerCAmelCase			,       batch_size=__lowerCAmelCase				)

					lowerCamelCase_					:      Optional[Any]     =		[len(batch[0]				) for batch in dl]

					if accelerator.process_index == 0:
										assert batch_sizes == process_0_expected_batch_sizes
					elif accelerator.process_index == 1:
										assert batch_sizes == process_1_expected_batch_sizes




def lowercase_	(   )     ->							Tuple:
					'''simple docstring'''

					lowerCamelCase_					:      List[str]     =		create_accelerator()

					# without padding, we would expect a different number of batches
					verify_dataloader_batch_sizes(
					    __lowerCAmelCase			,       dataset_size=3			,       batch_size=1			,       process_0_expected_batch_sizes=[1, 1]			,       process_1_expected_batch_sizes=[1, 1]			,       )

					# without padding, we would expect the same number of batches, but different sizes
					verify_dataloader_batch_sizes(
					    __lowerCAmelCase			,       dataset_size=7			,       batch_size=2			,       process_0_expected_batch_sizes=[2, 2]			,       process_1_expected_batch_sizes=[2, 2]			,       )




def lowercase_	(   )     ->							List[str]:
					'''simple docstring'''

					lowerCamelCase_					:      Any     =		create_accelerator(even_batches=__lowerCAmelCase				)

					verify_dataloader_batch_sizes(
					    __lowerCAmelCase			,       dataset_size=3			,       batch_size=1			,       process_0_expected_batch_sizes=[1, 1]			,       process_1_expected_batch_sizes=[1]			,       )

					verify_dataloader_batch_sizes(
					    __lowerCAmelCase			,       dataset_size=7			,       batch_size=2			,       process_0_expected_batch_sizes=[2, 2]			,       process_1_expected_batch_sizes=[2, 1]			,       )




def lowercase_	(   )     ->							str:
					'''simple docstring'''

					lowerCamelCase_					:      int     =		create_accelerator(even_batches=__lowerCAmelCase				)

					lowerCamelCase_					:      Union[str, Any]     =		torch.nn.Linear(1			,       1				)
					lowerCamelCase_					:      Any     =		accelerator.prepare(__lowerCAmelCase				)

					lowerCamelCase_					:      Tuple     =		create_dataloader(__lowerCAmelCase			,       dataset_size=3			,       batch_size=1				)

					lowerCamelCase_					:      str     =		[]
					with accelerator.join_uneven_inputs([ddp_model]				):
										for batch_idx, batch in enumerate(__lowerCAmelCase				):
															lowerCamelCase_					:      Tuple     =		ddp_model(batch[0].float()				)
															lowerCamelCase_					:      Optional[int]     =		output.sum()
															loss.backward()
															batch_idxs.append(__lowerCAmelCase				)

					accelerator.wait_for_everyone()

					if accelerator.process_index == 0:
										assert batch_idxs == [0, 1]
					elif accelerator.process_index == 1:
										assert batch_idxs == [0]




def lowercase_	(   _lowercase				)     ->							Optional[Any]:
					'''simple docstring'''

					with warnings.catch_warnings(record=__lowerCAmelCase				) as w:
										with accelerator.join_uneven_inputs([Mock()]				):
															pass

										assert issubclass(w[-1].category			,       __lowerCAmelCase				)
										assert "only supported for multi-GPU" in str(w[-1].message				)




def lowercase_	(   )     ->							int:
					'''simple docstring'''

					lowerCamelCase_					:      Tuple     =		True
					lowerCamelCase_					:      List[str]     =		False
					lowerCamelCase_					:      Optional[Any]     =		create_accelerator(even_batches=__lowerCAmelCase				)
					lowerCamelCase_					:      Optional[int]     =		torch.nn.Linear(1			,       1				)
					lowerCamelCase_					:      Optional[int]     =		accelerator.prepare(__lowerCAmelCase				)
					lowerCamelCase_					:      Optional[Any]     =		create_dataloader(__lowerCAmelCase			,       dataset_size=3			,       batch_size=1				)
					lowerCamelCase_					:      Tuple     =		create_dataloader(__lowerCAmelCase			,       dataset_size=3			,       batch_size=1				)

					with accelerator.join_uneven_inputs([ddp_model]			,       even_batches=__lowerCAmelCase				):
										lowerCamelCase_					:      Any     =		train_dl.batch_sampler.even_batches
										lowerCamelCase_					:      Optional[Any]     =		valid_dl.batch_sampler.even_batches

					assert train_dl_overridden_value == overridden_even_batches
					assert valid_dl_overridden_value == overridden_even_batches
					assert train_dl.batch_sampler.even_batches == default_even_batches
					assert valid_dl.batch_sampler.even_batches == default_even_batches




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

					lowerCamelCase_					:      Tuple     =		True
					lowerCamelCase_					:      int     =		False
					lowerCamelCase_					:      Any     =		create_accelerator(even_batches=__lowerCAmelCase				)
					lowerCamelCase_					:      int     =		torch.nn.Linear(1			,       1				)
					lowerCamelCase_					:      Optional[Any]     =		accelerator.prepare(__lowerCAmelCase				)
					create_dataloader(__lowerCAmelCase			,       dataset_size=3			,       batch_size=1			,       iterable=__lowerCAmelCase				)
					lowerCamelCase_					:      str     =		create_dataloader(__lowerCAmelCase			,       dataset_size=3			,       batch_size=1				)

					with warnings.catch_warnings():
										warnings.filterwarnings('''ignore'''				)
										try:
															with accelerator.join_uneven_inputs([ddp_model]			,       even_batches=__lowerCAmelCase				):
																				lowerCamelCase_					:      Dict     =		batch_dl.batch_sampler.even_batches
										except AttributeError:
															# ensure attribute error is not raised when processing iterable dl
															raise AssertionError

					assert batch_dl_overridden_value == overridden_even_batches
					assert batch_dl.batch_sampler.even_batches == default_even_batches




def lowercase_	(   )     ->							int:
					'''simple docstring'''

					lowerCamelCase_					:      Optional[Any]     =		create_accelerator()
					lowerCamelCase_					:      List[str]     =		torch.nn.Linear(1			,       1				)
					lowerCamelCase_					:      Optional[Any]     =		accelerator.prepare(__lowerCAmelCase				)
					create_dataloader(__lowerCAmelCase			,       dataset_size=3			,       batch_size=1			,       iterable=__lowerCAmelCase				)

					with warnings.catch_warnings(record=__lowerCAmelCase				) as w:
										with accelerator.join_uneven_inputs([ddp_model]			,       even_batches=__lowerCAmelCase				):
															pass

										assert issubclass(w[-1].category			,       __lowerCAmelCase				)
										assert "only supported for map-style datasets" in str(w[-1].message				)




def lowercase_	(   )     ->							Tuple:
					'''simple docstring'''

					lowerCamelCase_					:      Tuple     =		create_accelerator()

					accelerator.print('''Test that even_batches variable ensures uniform batches across processes'''				)
					test_default_ensures_even_batch_sizes()

					accelerator.print('''Run tests with even_batches disabled'''				)
					test_can_disable_even_batches()

					accelerator.print('''Test joining uneven inputs'''				)
					test_can_join_uneven_inputs()

					accelerator.print('''Test overriding even_batches when joining uneven inputs'''				)
					test_join_can_override_even_batches()

					accelerator.print('''Test overriding even_batches for mixed dataloader types'''				)
					test_join_can_override_for_mixed_type_dataloaders()

					accelerator.print('''Test overriding even_batches raises a warning for iterable dataloaders'''				)
					test_join_raises_warning_for_iterable_when_overriding_even_batches()

					accelerator.print('''Test join with non DDP distributed raises warning'''				)
					lowerCamelCase_					:      str     =		accelerator.state.distributed_type
					lowerCamelCase_					:      Optional[Any]     =		DistributedType.FSDP
					test_join_raises_warning_for_non_ddp_distributed(__lowerCAmelCase				)
					lowerCamelCase_					:      List[str]     =		original_state


if __name__ == "__main__":
					main()




'''simple docstring'''


import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


__lowercase							:   Union[str, Any]					=     logging.get_logger(__name__)




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

					lowerCamelCase_					:      Optional[Any]     =		os.getenv('''SM_HP_MP_PARAMETERS'''			,       '''{}'''				)
					try:
										# Parse it and check the field "partitions" is included, it is required for model parallel.
										lowerCamelCase_					:      Optional[Any]     =		json.loads(_lowercase				)
										if "partitions" not in smp_options:
															return False
					except json.JSONDecodeError:
										return False

					# Get the sagemaker specific framework parameters from mpi_options variable.
					lowerCamelCase_					:      List[str]     =		os.getenv('''SM_FRAMEWORK_PARAMS'''			,       '''{}'''				)
					try:
										# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
										lowerCamelCase_					:      List[str]     =		json.loads(_lowercase				)
										if not mpi_options.get('''sagemaker_mpi_enabled'''			,       _lowercase				):
															return False
					except json.JSONDecodeError:
										return False
					# Lastly, check if the `smdistributed` module is present.
					return importlib.util.find_spec('''smdistributed'''				) is not None


if is_sagemaker_model_parallel_available():
					import smdistributed.modelparallel.torch as smp

					smp.init()






@dataclass
class 	__lowercase				(  _lowercase    ):
			lowerCamelCase			:		str	      =			field(
			    default=""      ,     metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}      ,     )

			def        UpperCAmelCase__ (self      ):
								super().__post_init__()
								warnings.warn(
								    '''`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use '''
								    '''`TrainingArguments` instead.'''			,   A			,   )

			@cached_property
			def        UpperCAmelCase__ (self      ):
								logger.info('''PyTorch: setting up devices'''      )
								if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
													logger.warning(
													    '''torch.distributed process group is initialized, but local_rank == -1. '''
													    '''In order to use Torch DDP, launch your script with `python -m torch.distributed.launch'''      )
								if self.no_cuda:
													lowerCamelCase_					:      Optional[int]     =		torch.device('''cpu'''      )
													lowerCamelCase_					:      Dict     =		0
								elif is_sagemaker_model_parallel_available():
													lowerCamelCase_					:      int     =		smp.local_rank()
													lowerCamelCase_					:      str     =		torch.device('''cuda'''			,   A      )
													lowerCamelCase_					:      Tuple     =		1
								elif is_sagemaker_dp_enabled():
													import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

													torch.distributed.init_process_group(backend='''smddp'''			,   timeout=self.ddp_timeout_delta      )
													lowerCamelCase_					:      List[Any]     =		int(os.getenv('''SMDATAPARALLEL_LOCAL_RANK'''      )      )
													lowerCamelCase_					:      Union[str, Any]     =		torch.device('''cuda'''			,   self.local_rank      )
													lowerCamelCase_					:      List[str]     =		1
								elif self.local_rank == -1:
													# if n_gpu is > 1 we'll use nn.DataParallel.
													# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
													# Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
													# trigger an error that a device index is missing. Index 0 takes into account the
													# GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
													# will use the first GPU in that env, i.e. GPU#1
													lowerCamelCase_					:      List[str]     =		torch.device('''cuda:0''' if torch.cuda.is_available() else '''cpu'''      )
													# Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
													# the default value.
													lowerCamelCase_					:      Any     =		torch.cuda.device_count()
								else:
													# Here, we'll use torch.distributed.
													# Initializes the distributed backend which will take care of synchronizing nodes/GPUs
													if not torch.distributed.is_initialized():
																		torch.distributed.init_process_group(backend='''nccl'''			,   timeout=self.ddp_timeout_delta      )
													lowerCamelCase_					:      Any     =		torch.device('''cuda'''			,   self.local_rank      )
													lowerCamelCase_					:      List[str]     =		1

								if device.type == "cuda":
													torch.cuda.set_device(A      )

								return device

			@property
			def        UpperCAmelCase__ (self      ):
								if is_sagemaker_model_parallel_available():
													return smp.dp_size()

								return super().world_size

			@property
			def        UpperCAmelCase__ (self      ):
								return not is_sagemaker_model_parallel_available()






			@property
			def        UpperCAmelCase__ (self      ):
								return False


