o bib@sdZddlZddlmZddlmZddlmZddlm Z m Z m Z ddl Z ddl mZddlZddlmZddlmZdd lmZGd d d eZd ed ededededededede efddZddZ  d#de ed ed ede efddZdd Zde efd!d"ZdS)$aThis file is modeled after ray/python/ray/train/torch/config.py The logics are duplicated right now to allow maximum flexibility for setting up PyTorch DDP process groups outside the context of Ray Train. Eventually, these use cases should be consolidated. N)ABC) defaultdict) timedelta)CallableListT) ActorHandle) get_devices)get_address_and_portc@s*eZdZdZdedefdefddZdS)TorchDistributedWorkerzDefines the interfaces required by the init_torch_dist_process_group(). This is modeled after RayTrainerWorker, which allows arbitrary functions to be executed on a remote DDP worker. func.returncOs||i|S)zExecutes the input function and returns the output. Args: func: The function to execute. args, kwargs: The arguments to pass into func. )selfr argskwargsrrK/home/ubuntu/.local/lib/python3.10/site-packages/ray/air/util/torch_dist.pyexecuteszTorchDistributedWorker.executeN)__name__ __module__ __qualname____doc__rrrrrrrr sr init_methodbackendrank world_size local_ranklocal_world_size master_addr master_portgpu_idsc Ks|dkrt|tjd<t|tjd<d} n|dkr"d|d|} ntd|d |d kr@d tjd <d dd|Dtjd<| t|| ||d| dtddt j di| t|tjd<t|tjd<t|tjd<t|tjd<dS)z$Initialize torch distributed backendenv MASTER_ADDR MASTER_PORTzenv://tcpztcp://:zThe provided init_method (z2) is not supported. Must be either 'env' or 'tcp'.nccl1NCCL_ASYNC_ERROR_HANDLING,css|]}t|VqdS)N)str).0gidrrr Gsz*_init_torch_distributed..CUDA_VISIBLE_DEVICES)rrrrtimeouti)secondsRANK LOCAL_RANK WORLD_SIZELOCAL_WORLD_SIZENr) r*osenviron ValueErrorjoinupdatedict setdefaultrdistinit_process_group) rrrrrrrrr init_process_group_kwargsurlrrr_init_torch_distributed(s8  r@cCst}t}||fS)z0Returns the node_id and gpu_ids for this worker.)rayget_runtime_context get_node_id get_gpu_ids)node_idr rrr_get_node_and_gpu_ids[s rFgloor!workersr cKs0tstdtdd|D}tt}tt}t|D]!\}\}} || |t | ts3| g} | D] } || | q5qt|dj t\} } g} t|}g}t|D]7\}}||d}|||}t||}| |j j tf||||||| | t||d || |qYt| |S)aInitialize a torch distributed process group. Note: this util assumes that the order of the workers passed in are their global ranks. Args: workers: A list of TorchDistributedWorker actors. backend: The torch distributed backend to use, possible choices are "gloo" or "nccl". init_method: The initialization method to use, possible choices are "env" or "tcp". init_process_group_kwargs: Additional kwargs to pass to the call to :meth:`torch.distributed.init_process_group`. Returns: Local ranks on their respective nodes for the list of workers. z#Distributed torch is not available.cSg|]}|jtqSr)rremoterFr+wrrr ~z1init_torch_dist_process_group..r) rrrrrrrrr )r< is_available RuntimeErrorrAgetrlistset enumerateappend isinstanceaddrrJr lenindexr@)rHrrr>node_and_gpu_idsnode_to_workersnode_to_gpu_idsirEr gpu_idrr setup_futuresr local_ranksrworkerrrrrrinit_torch_dist_process_groupbsT        rbc Cs`ttjs dSt}|D]}tj| tjWdn1s(wYqdS)z"Shutdown torch distributed backendN)r<destroy_process_grouptorchcudarOr device empty_cache)devicesrfrrr_shutdown_torch_distributeds  ricCstdd|DdS)NcSrIr)rrJrirKrrrrMrNz5shutdown_torch_dist_process_group..)rArQ)rHrrr!shutdown_torch_dist_process_groupsrj)rGr!)rr5abcr collectionsrdatetimertypingrrrrdtorch.distributed distributedr<rA ray.actorrray.air._internal.torch_utilsr ray.train._internal.utilsr r r*intr@rFrbrirjrrrrs\        3  N