o {i7, @sddlZddlZddlZddlZddlZddlmZddlmZddl m Z m Z ddl Z ddl mZedZdadadadefddZd8defd d Zd8defd d Zd8defd dZd9ddZd9ddZd:dedefddZdefddZdede efddZ GdddeZ!GdddZ"dd d ddd!d"ed#ed$ed%ed&edBf d'd(Z#d)d*Z$d9d+d,Z%d-e e efd.d/Z&d0d1Z'defd2d3Z(defd4d5Z)defd6d7Z*dS);N) timedelta)Enum)ListSequencedinov3returncCstotS)zC Returns: True if distributed training is enabled. )dist is_availableis_initializedr r X/data/cameron/keygrip/volume_dino_tracks/dinov3/distributed/torch_distributed_wrapper.pyis_distributed_enabledr cCtsdStj|dS)z^ Returns: The rank of the current process within the specified process group. rgroup)r rget_rankrr r r r! rcCr)zR Returns: The number of processes in the specified process group. r)r rget_world_sizerr r r r+rrcCs t|dkS)zf Returns: True if the current process is the main one in the specified process group. r)rrr r r is_main_process5 rcOs,|dd}t|s dStj|i|dS)z4Utility function to save only from the main process.rN)poprtorchsave)argskwargsrr r r save_in_main_process=s rcCs ddl}|jadd}||_dS)z=This function disables printing when not in the main process.rNc_s,|dd}ts |rt|i|dSdS)NforceF)rr_BUILTIN_PRINT)rrrr r r printLs  z._restrict_print_to_main_process..print)builtinsr r) __builtin__r r r r _restrict_print_to_main_processEs r#seedcCs:d\}}tjd}|durt|}|||St|S)N)i Ni` MASTER_PORT)osenvirongetrandomRandomrandintint)r$ZMIN_MASTER_PORTZMAX_MASTER_PORTZmaster_port_strrngr r r _get_master_portTs    r.cCsNttjtj}|d|d}|WdS1s wYdS)N)rr)socketAF_INET SOCK_STREAMbind getsockname)sportr r r _get_available_port_s   $r7r5cCstjddd|gddS)NZscontrolshowZ hostnamesT)text) subprocess check_output splitlines)r5r r r _parse_slurm_node_listhsr=c@seZdZdZdZdZdS)JobTypeZ TorchElasticZSlurmmanualN)__name__ __module__ __qualname__ TORCHELASTICSLURMMANUALr r r r r>lsr>c@sVeZdZdZddZdddededdfd d Zedefd d Zd dZ ddZ dS)TorchDistributedEnvironmentz Helper class to get (and set) distributed job information from the environment. Identifies and supports (in this order): - TorchElastic, - Slurm, - Manual launch (single-node). cCsxdtjvr>tjd|_tj|_tjd|_ttjd|_ttjd|_ ttjd|_ ttjd|_ ttjd|_ nldtjvrttjd|_tj |_ttjd }ttjd }t||kseJ|d |_t|jd |_ttjd |_ ttjd|_ ttjd|_ |j ||_ nd|_tj|_d|_t|_d |_ d|_ d |_ d|_ |j |j ksJ|j |j ksJdS)NTORCHELASTIC_RUN_ID MASTER_ADDRr%RANK WORLD_SIZE LOCAL_RANKLOCAL_WORLD_SIZEZ SLURM_JOB_IDZSLURM_JOB_NUM_NODESZSLURM_JOB_NODELISTr)r$Z SLURM_PROCIDZ SLURM_NTASKSZ SLURM_LOCALIDz 127.0.0.1r)r&r'job_idr>rCjob_type master_addrr, master_portrank world_size local_ranklocal_world_sizerDr=lenr.rEr7)selfZ node_countnodesr r r __init__{s>     z$TorchDistributedEnvironment.__init__F)nccl_async_error_handling overwriterYrcCs|jt|jt|jt|jt|jt|jd}|r"|ddi|sB|D]\}}|t j vr2q(t j ||kr:q(t d|dt j ||S)N)rHr%rIrJrKrLZTORCH_NCCL_ASYNC_ERROR_HANDLING1z'Cannot export environment variables as z is already set) rOstrrPrQrRrSrTupdateitemsr&r' RuntimeError)rVrZrYenv_varskvr r r exports*   z"TorchDistributedEnvironment.exportcCs |jdkSNr)rQrVr r r rs z+TorchDistributedEnvironment.is_main_processc CsL|jjd|jrd|jdndd|jd|jd|jd|jd S) Nz job (z) r/zusing :z (rank=z , world size=))rNvaluerMrOrPrQrRrer r r __str__s z#TorchDistributedEnvironment.__str__cCs>|jjd|jd|jd|jd|jd|jd|jdS)Nz (master_addr=z ,master_port=z,rank=z ,world_size=z ,local_rank=z,local_world_size=rh) __class__r@rOrPrQrRrSrTrer r r __repr__s z$TorchDistributedEnvironment.__repr__N) r@rArB__doc__rXboolrcpropertyrrjrlr r r r rFrs. $ rFTF)set_cuda_current_devicerZrYrestrict_print_to_main_processtimeoutrprZrYrqrrcCsttdurtdt}td||j||d|r#tj|j t j d|dt |r3t tjjjadS)a{Enable distributed mode. Args: set_cuda_current_device: If True, call torch.cuda.set_device() to set the current PyTorch CUDA device to the one matching the local rank. overwrite: If True, overwrites already set variables. Else fails. nccl_async_error_handling: Enables NCCL asynchronous error handling. As a side effect, this enables timing out PyTorch distributed operations after a default 30 minutes delay). restrict_print_to_main_process: If True, the print function of non-main processes (ie rank>0) is disabled. Use print(..., force=True) to print anyway. If False, nothing is changed and all processes can print as usual. timeout: Timeout for operations executed against the process group. Default value is 10 minutes for NCCL and 30 minutes for other backends. Nz)Distributed mode has already been enabledz!PyTorch distributed environment: )rZrYnccl)backendrr)_DEFAULT_PROCESS_GROUPr_rFloggerinforcrcuda set_devicerSrinit_process_groupbarrierr# distributedrWORLD)rprZrYrqrrZ torch_envr r r enable_distributedsr~cCstSN)rur r r r get_default_process_groupsrcCsNtdur ddl}t|_tdurtjtdatdur%tjtdadSdSrd)rr!r _PROCESS_SUBGROUPrr|destroy_process_groupru)r"r r r disable_distributeds  rall_subgroup_rankscCsntdd|D}t}t|tt|ksJ||vsJtdus$J|D]}tj|}||vr4|aq&dS)aCreate new process subgroups according to the provided specification. Args: all_subgroup_ranks: a sequence of rank sequences (first rank, ..., last rank), one for each process subgroup. Example: ((0, 1), (2, 3), (4, 5, 6, 7)). Note: This is similar to the (non-documented) new_subgroups_by_enumeration(). This should be called once (and not sequentially) to create all subgroups. css|] }|D]}|VqqdSrr ).0subgroup_ranksrQr r r 4sz new_subgroups..N)tuplerrUsetrrr| new_group)rZ all_ranksrQrsubgroupr r r new_subgroups)s    rcCstptS)zG Returns: The process subgroup of this rank (or None). )rrur r r r get_process_subgroupBsrcC ttdS)zW Returns: The rank of the current process within its process subgroup. rrrr r r r get_subgroup_rankJrrcCr)zJ Returns: The number of processes in the process subgroup r)rrr r r r get_subgroup_sizeRrrcCsttddkS)zc Returns: True if the current process is the main one within its process subgroup. rrrr r r r is_subgroup_main_processZrrr)rN)r)+loggingr&r)r0r:datetimerenumrtypingrrrtorch.distributedr|r getLoggerrvrurrrnr r,rrrrr#r.r7r\r=r>rFr~rrrrrrrr r r r s`         r 0