

'''simple docstring'''



import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment
def 		__lowerCamelCase			(  __snake_case	:						Any,						__snake_case	:						List[str]=(),						__snake_case	:						Union[str, Any]=None,						__snake_case	:						Dict="no",						__snake_case	:						int="29500"			)			->				Optional[Any]:






						"""simple docstring"""





						A__	:  Optional[int]           =False
						A__	:  Tuple           =False
						if any(key.startswith("""KAGGLE"""			) for key in os.environ.keys()			):
												A__	:  Optional[int]           =True
						elif "IPython" in sys.modules:
												A__	:  int           ='''google.colab''' in str(sys.modules["""IPython"""].get_ipython()			)

						try:
												A__	:  str           =PrecisionType(mixed_precision.lower()			)
						except ValueError:
												raise ValueError(
												    f"Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}."			)

						if (in_colab or in_kaggle) and (os.environ.get("""TPU_NAME""",						__snake_case			) is not None):
												# TPU launch
												import torch_xla.distributed.xla_multiprocessing as xmp

												if len(AcceleratorState._shared_state			) > 0:
																		raise ValueError(
																		    """To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside """
																		    """your training function. Restart your notebook and make sure no cells initializes an """
																		    """`Accelerator`."""			)
												if num_processes is None:
																		A__	:  int           =8

												A__	:  Optional[Any]           =PrepareForLaunch(__snake_case,						distributed_type="""TPU"""			)
												print(f"Launching a training on {num_processes} TPU cores."			)
												xmp.spawn(__snake_case,						args=__snake_case,						nprocs=__snake_case,						start_method="""fork"""			)
						elif in_colab:
												# No need for a distributed launch otherwise as it's either CPU or one GPU.
												if torch.cuda.is_available():
																		print("""Launching training on one GPU."""			)
												else:
																		print("""Launching training on one CPU."""			)
												function(*__snake_case			)
						else:
												if num_processes is None:
																		raise ValueError(
																		    """You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call."""			)

												if num_processes > 1:
																		# Multi-GPU launch
																		from torch.multiprocessing import start_processes
																		from torch.multiprocessing.spawn import ProcessRaisedException

																		if len(AcceleratorState._shared_state			) > 0:
																								raise ValueError(
																								    """To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized """
																								    """inside your training function. Restart your notebook and make sure no cells initializes an """
																								    """`Accelerator`."""			)

																		if torch.cuda.is_initialized():
																								raise ValueError(
																								    """To launch a multi-GPU training from your notebook, you need to avoid running any instruction """
																								    """using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA """
																								    """function."""			)

																		# torch.distributed will expect a few environment variable to be here. We set the ones common to each
																		# process here (the other ones will be set be the launcher).
																		with patch_environment(
																		    world_size=__snake_case,						master_addr="""127.0.01""",						master_port=__snake_case,						mixed_precision=__snake_case			):
																								A__	:  List[str]           =PrepareForLaunch(__snake_case,						distributed_type="""MULTI_GPU"""			)
																								print(f"Launching training on {num_processes} GPUs."			)
																								try:
																														start_processes(__snake_case,						args=__snake_case,						nprocs=__snake_case,						start_method="""fork"""			)
																								except ProcessRaisedException as e:
																														if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
																																				raise RuntimeError(
																																				    """CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. """
																																				    """This likely stems from an outside import causing issues once the `notebook_launcher()` is called. """
																																				    """Please review your imports and test them when running the `notebook_launcher()` to identify """
																																				    """which one is problematic."""			) from e

												else:
																		# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
																		if is_mps_available():
																								A__	:  Optional[Any]           ='''1'''
																								print("""Launching training on MPS."""			)
																		elif torch.cuda.is_available():
																								print("""Launching training on one GPU."""			)
																		else:
																								print("""Launching training on CPU."""			)
																		function(*__snake_case			)
def 		__lowerCamelCase			(  __snake_case	:						str,						__snake_case	:						Optional[int]=(),						__snake_case	:						Optional[Any]=2			)			->				Any:






						"""simple docstring"""





						from torch.multiprocessing import start_processes

						with tempfile.NamedTemporaryFile() as tmp_file:
												# torch.distributed will expect a few environment variable to be here. We set the ones common to each
												# process here (the other ones will be set be the launcher).
												with patch_environment(
												    world_size=__snake_case,						master_addr="""127.0.01""",						master_port="""29500""",						accelerate_mixed_precision="""no""",						accelerate_debug_rdv_file=tmp_file.name,						accelerate_use_cpu="""yes""",						):
																		A__	:  Tuple           =PrepareForLaunch(__snake_case,						debug=__snake_case			)
																		start_processes(__snake_case,						args=__snake_case,						nprocs=__snake_case,						start_method="""fork"""			)








"""simple docstring"""





import inspect
import os
import unittest

import torch

import accelerate
from accelerate import debug_launcher
from accelerate.test_utils import (
    execute_subprocess_async,
    require_cpu,
    require_huggingface_suite,
    require_multi_gpu,
    require_single_gpu,
)
from accelerate.utils import patch_environment
@require_huggingface_suite
class 		SCREAMING_SNAKE_CASE_				(     unittest.TestCase ):
				"""simple docstring"""




				def        __lowercase	(  self	:Dict      ):
											__lowerCamelCase   :   Union[str, Any]       =inspect.getfile(accelerate.test_utils      )
											__lowerCamelCase   :   Union[str, Any]       =os.path.sep.join(
											    mod_file.split(os.path.sep      )[:-1] + ['''scripts''', '''external_deps''', '''test_metrics.py''']      )

											from accelerate.test_utils.scripts.external_deps import test_metrics  # noqa: F401

											__lowerCamelCase   :   List[str]       =test_metrics




				@require_cpu
				def        __lowercase	(  self	:Optional[Any]      ):
											debug_launcher(self.test_metrics.main    ,     num_processes=1      )




				@require_cpu
				def        __lowercase	(  self	:int      ):
											debug_launcher(self.test_metrics.main      )




				@require_single_gpu
				def        __lowercase	(  self	:int      ):
											self.test_metrics.main()




				@require_multi_gpu
				def        __lowercase	(  self	:Dict      ):
											print(f'Found {torch.cuda.device_count()} devices.'      )
											__lowerCamelCase   :   Union[str, Any]       =['''torchrun''', f'--nproc_per_node={torch.cuda.device_count()}', self.test_file_path]
											with patch_environment(omp_num_threads=1      ):
																		execute_subprocess_async(__lowercase    ,     env=os.environ.copy()      )




