





'''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  a__    (    lowercase :					Optional[int], lowercase :					Optional[Any]=(), lowercase :					Union[str, Any]=None, lowercase :					List[Any]="no", lowercase :					List[Any]="29500"				)		->		Tuple:





     """simple docstring"""

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

     try:
          _UpperCamelCase       =	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''', lowercase				) 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:
               _UpperCamelCase       =	8

          _UpperCamelCase       =	PrepareForLaunch(lowercase, distributed_type='''TPU'''				)
          print(F"""Launching a training on {num_processes} TPU cores."""				)
          xmp.spawn(lowercase, args=lowercase, nprocs=lowercase, 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(*lowercase				)
     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=lowercase, master_addr='''127.0.01''', master_port=lowercase, mixed_precision=lowercase				):
                    _UpperCamelCase       =	PrepareForLaunch(lowercase, distributed_type='''MULTI_GPU'''				)
                    print(F"""Launching training on {num_processes} GPUs."""				)
                    try:
                         start_processes(lowercase, args=lowercase, nprocs=lowercase, 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():
                    _UpperCamelCase       =	'''1'''
                    print('''Launching training on MPS.'''				)
               elif torch.cuda.is_available():
                    print('''Launching training on one GPU.'''				)
               else:
                    print('''Launching training on CPU.'''				)
               function(*lowercase				)







def  a__    (    lowercase :					List[Any], lowercase :					Optional[Any]=(), lowercase :					Optional[int]=2				)		->		List[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=lowercase, master_addr='''127.0.01''', master_port='''29500''', accelerate_mixed_precision='''no''', accelerate_debug_rdv_file=tmp_file.name, accelerate_use_cpu='''yes''', ):
               _UpperCamelCase       =	PrepareForLaunch(lowercase, debug=lowercase				)
               start_processes(lowercase, args=lowercase, nprocs=lowercase, start_method='''fork'''				)










'''simple docstring'''

def  a__    (    lowercase :					int				)		->		int:





     """simple docstring"""

     if not isinstance(lowercase, lowercase				):
          raise TypeError('''Input value must be an \'int\' type'''				)

     _UpperCamelCase       =	0
     while number:
          position += 1
          number >>= 1

     return position


if __name__ == "__main__":
   import doctest

   doctest.testmod()



