o ॵiy+@sddlZddlZddlZddlZddlZddlZddlZddlmZm Z m Z m Z ddl Z ddlZddlmZddlmZddlmZdefddZdedefd d Zd d Zd9dededdfddZdeddfddZdeddfddZd:dede eddfddZd:de eeffddZ ddZ!ddZ"ddZ#d d!Z$d"d#Z%d:d$d%Z&d:d&d'Z'd(d)Z(d*d+Z)d,d-Z*e+d.d/Z,d0d1Z-d2d3Z.d:d4d5Z/d6ej0j1defd7d8Z2dS);N)CallableListOptionalTuple)version) distributedreturncCs2ttjtj}|d|d}||S)N)r)socketAF_INET SOCK_STREAMbind getsocknameclose)sockportrP/home/ubuntu/.local/lib/python3.10/site-packages/modelscope/utils/torch_utils.py_find_free_ports   rrcshttd}|dttjtjtfdd|DWdS1s-wYdS)N localhostc3s"|] }|fdkVqdS)rN) connect_ex).0iprsrr "s z _is_free_port..)r gethostbyname_ex gethostnameappendr r all)ripsrrr _is_free_ports  $r#cKs`t|dr|jdi|}|Sttjtdkr%tj|fi|}|Stdtjd|S)Ncompilez 2.0.0.devzDCompiling model needs torch version > 2.0.0, your torch version is: z , origin model will be returned.r)hasattrr$rparsetorch __version__print)modelcompile_optionsrrr compile_model%s r,nccllauncherbackendcKs|tjdddur td|dkrt|fi|dS|dkr)t|fi|dS|dkr7t|fi|dStd|)NT) allow_nonespawnpytorchmpislurmzInvalid launcher type: )mpget_start_methodset_start_method_init_dist_pytorch_init_dist_mpi_init_dist_slurm ValueError)r.r/kwargsrrr init_dist3s r=cKs2ttjd}tj|tjdd|i|dS)N LOCAL_RANKr/r)intosenvironr'cuda set_devicedistinit_process_groupr/r< local_rankrrrr8@s r8cKsxttjd}tj|dtjvrdtjd<dtjvr tdtjdtjd<tjdtjd <tjd d |i|dS) NOMPI_COMM_WORLD_LOCAL_RANK MASTER_PORT29500 MASTER_ADDRz/The environment variable MASTER_ADDR is not setOMPI_COMM_WORLD_SIZE WORLD_SIZEOMPI_COMM_WORLD_RANKRANKr/r) r?r@rAr'rBrCKeyErrorrDrErFrrrr9Gs    r9cCsttjd}ttjd}tjd}tj}tj||td|d}|dur5t |tjd<ndtjvr;nt drEd tjd<nt t tjd<d tjvrW|tjd <t |tjd <t ||tjd <t |tjd <t j |ddS)aInitialize slurm distributed training environment. If argument ``port`` is not specified, then the master port will be system environment variable ``MASTER_PORT``. If ``MASTER_PORT`` is not in system environment variable, then a default port ``29500`` will be used. Args: backend (str): Backend of torch.distributed. port (int, optional): Master port. Defaults to None. SLURM_PROCID SLURM_NTASKSSLURM_NODELISTzscontrol show hostname z | head -n1NrIirOr/)r?r@rAr'rB device_countrC subprocess getoutputstrr#rrDrE)r/rproc_idntasks node_listnum_gpusaddrrrrr:Ts*        r:cCs`tr(ddlm}|dur|rddlm}|}t|}t|}||fSd}d}||fS)zGet dist info of a specified group Args: group: The parallel group, default None, for the global group Returns: A tuple of the current rank and world_size of the group r)is_megatron_initializedN)mpur ) is_distmodelscope.utils.megatron_utilsr^ megatron_utilr_get_data_parallel_grouprDget_rankget_world_size)groupr^r_rank world_sizerrr get_dist_info{s    ricCsttjddS)Nr>r)r?r@rAgetrrrrget_local_ranksrkcC tsdSts dStS)Nr)rD is_availableis_initializedrdrrrrrd rdcCrl)Nr )rDrmrnrerrrrrerorecCs8tsdSts dSt}|dkrdStdS)zj Helper function to synchronize (barrier) among all processes when using distributed training Nr )rDrmrnrebarrier)rhrrr synchronizes rqcCstotSN)rDrmrnrrrrr`sr`cCstr t|dkSdS)NrT)r`rDrdrfrrr is_mastersrtcsdtdtffdd }|S)Nfuncrcstfdd}|S)Ncstr |i|SdSrr)rt)argsr<)rurfrrwrappersz.master_only..decorate..wrapper) functoolswraps)rurwrs)rurdecorateszmaster_only..decorate)r)rfrzrrsr master_onlys r{cCs6tstSd}trt}tt|d}|S)zRMake sure each rank has the same temporary directory on the distributed mode. Nr)r`tempfilemkdtemprtrDrp broadcast)tmpdirrrr make_tmp_dirs rcCst}tjdgdd}||kr&tjtt|tjdd}tj|jdd}t t ||||krAtj | fdtjdd}t t ||t |S)z Broadcasts the inputs to all ranks. Arguments: inputs : Any objects that can be serialized by pickle. src (int): Source rank. Returns: Each rank returns the same value as src. rrBdevicedtyper)rDrdr'tensor bytearraypickledumpsuint8shaperpr~fullitemloadscpunumpytobytes)inputssrcrg shape_tensor inputs_tensorrrrr~s$    r~cCsN|dur |dkr t|tj|t|tj|dStd|)Nrz0Random seed should be positive, current seed is )randomseednpr' manual_seedrBmanual_seed_allr;)rrrrset_random_seeds   rcCs tdkr tjddStjjS)zj Return a process group based on gloo backend, containing all the ranks The result is cached. r-gloorT)rD get_backend new_grouprfWORLDrrrr_get_global_gloo_groups  rcCst|}|dvs Jt|dkrdnd}t|}t|dkr0td t t|d|tj |}t |j|d}|S)N)rr-rrrBi@z;Rank {} trying to all-gather {:.2f} GB of data on device {}r)rDrr'rrrlenloggerwarningformatrd ByteStorage from_buffer ByteTensorto)datarfr/rbufferstoragerrrr_serialize_to_tensors      rcstj|d}|dksJdtjgtjjd}fddt|D}tj|||ddd|D}t |}||krStj ||ftj jd}tj |fdd |fS) zz Returns: list[int]: size of the tensor, on each rank Tensor: padded tensor that has the max size rsr zBcomm.gather/all_gather must be called from ranks within the group!rcs"g|] }tjdgtjjdqS)r r)r'zerosint64rr_rrr /z*_pad_to_largest_tensor..cSsg|]}t|qSr)r?r)rsizerrrr4sr)dim) rDrer'rnumelrrrange all_gathermaxrrcat)rrfrh local_size size_listmax_sizepaddingrrr_pad_to_largest_tensor"s*     rcstdkr|gS|durt}t|dkr|gSt||t|\}t|fdd|D}tj||dg}t||D]\} d|}| t |qB|S)a; Run all_gather on arbitrary picklable data (not necessarily tensors). Args: data: any picklable object group: a torch process group. By default, will use a group which contains all ranks on gloo backend. Returns: list[data]: list of data gathered from each rank r Ncs"g|] }tjftjjdqS)r)r'emptyrrrrrrrrYrzall_gather..rs)rerrDrrrrziprrrr rr)rrfr tensor_list data_listrrrrrrBs$   rr*cCs(tdd|Ddh}t|dkS)Ncss|]}t|jVqdSrr)rXr)rprrrrhsz$is_on_same_device..rr )set parametersr)r* device_setrrris_on_same_devicegs r)r-rr)3rxr@rrr rVr|typingrrrrrrr'torch.multiprocessingmultiprocessingr5 packagingrrrDrXrr?boolr#r,r=r8r9r:rirkrdrerqr`rtr{rr~r lru_cacherrrrnnModulerrrrrsJ      '  !   %