o Tix @sPdZddlZddlZddlmZmZddlTddlTddlmZddl m Z dd l m Z dd l mZmZdd lmZdd lmZdd lmZdaeZiZe ZdadadadaddlTGdddZ ddZ!      d~ddZ"ddZ#ddZ$ddZ%dddZ&ddZ'de(fdd Z)d!d"Z*e#dd#d#d$efd%d$Z+e#dd&d'Z,e#dd#d#d(efd)d(Z-d*d+Z.e/j0dd#d#efd,d-Z1e#e/j0dd#d#d.efd/d.Z2e#dd#d#d0efd1d0Z3d2d3Z4dd#efd4d5Z5e#dddd#d#d6efd7d6Z6e#dd8d9Z7e#ddd#d:efd;d:Z8e#dddd#defd?d>Z:e#dddd#d@efdAd@Z;e#dddd#d#dBefdCdBZe#ddd#d#dHefdIdHZ?ddJdKZ@e#e/j0dd#d#dLefdMdLZAe#e/j0dd#d#dNefdOdNZBdPdQZCdRdSZDddTdUZEe#e/j0dd#d#dVefdWdVZFe#e/j0dd#d#dVefdXdYZGe#e/j0dd#d#dVefdZd[ZHd\d]ZIddeJfd^d_ZKdd`daZLdbdcZMddddeZNddfdgZOdhdiZPdjeQfdkdlZRddmedmeddddndnf dodpZSedmfdqdrZTdsdtZUdudvZVdwdxZWddzd{ZXdd|d}ZYdS)a DeepSpeed Communication Package: deepspeed.comm deepspeed.comm -- import and use deepspeed.ops.comm -- use torch.distributed directly if both this package and torch.distributed use the same NCCL version -- use custom collectives -- can either use torch.dist or ds.ops.comm? Note: the old 1-bit compressed allreduce variants that resided in deepspeed.runtime.comm will be moved here as well. deepspeed.comm API -- must be kept fully compatible (same signatures) as torch.dist API to ensure backward/cross-framework compatibility. -- e.g. if a client code used from deepspeed import comm as dist instead of import torch.distributed as dist The code should work without breaking any of the public torch.distributed functionality Future: -- deepspeed groups API should be brought into ds.comm N)TORCH_DISTRIBUTED_DEFAULT_PORTdefault_pg_timeout)*)get_accelerator) CCLBackend) CommsLogger)timerget_caller_func) TorchBackend)utils) timedeltac@seZdZgfddZdS) ProcessGroupcCs||_||_t||_dSN)rankscomm_idlensize)selfrrrG/home/ubuntu/.local/lib/python3.10/site-packages/deepspeed/comm/comm.py__init__=szProcessGroup.__init__N)__name__ __module__ __qualname__rrrrrr;srcCs|jr t|dSdSr)comms_logger_enabled comms_logger configure)configrrr_configure_using_config_fileCsr cCs`|dur t|j|dur|t_|dur|t_|dur|t_|dur%|t_|dur.|t_dSdSr)r comms_configrenabledprof_allprof_opsverbosedebug)deepspeed_configr"r#r$r%r&rrrrHs  rcsfdd}|S)Nc stjr9d|vr |dstjsd|vr9|dtjvr9t}||tg|Ri|}t|tj}t | zO|i|Wtjrt t jrPt d|vrX|dsftjsfd|vr|dtjvrt|tj}j}t |t |jdd}t||||SSSStjrt t jrt d|vr|dstjsd|vr|dtjvrt|tj}j}t |t |jdd}t||||wwww)Nproflog_nameF)reset)rr"r#r$get_default_argsupdateget_msg_size_from_argsget_debug_log_namer&timersstartr synchronizecdb using_mpibarrierrstopelapsedappend)argskwargs func_argsmsg_sizer)raw_name time_elapsedfuncrr log_wrappergsR         ztimed_op..log_wrapperr)r?r@rr>rtimed_opes rAcCsttdd}ttdd}|tkrtjddS|tkr(tjddS|tkr4tjddS|t krLt ||||da tj d|d dS|t krXtjd dStjd |d dS) NRANKz-1 WORLD_SIZEz-NCCL backend in DeepSpeed not yet implementedz,MPI backend in DeepSpeed not yet implementedz-Gloo backend in DeepSpeed not yet implemented)rank world_sizetimeout init_methodz Initialize z backendz-HCCL backend in DeepSpeed not yet implementedzDeepSpeed does not support )intosgetenv NCCL_BACKENDr loggerr& MPI_BACKEND GLOO_BACKEND CCL_BACKENDr ccl_backendinfo HCCL_BACKEND) ds_backendrFrGrDrrrrinit_deepspeed_backendsrTcCstdurdStSNF)r2is_initializedrrrrrVsrVcCs tj|dS)Ngroup)r2destroy_process_grouprWrrrrYs rYcCs"tdurts Jdt|SNJDeepSpeed backend not set, please initialize it using init_process_group())r2rV new_group)rrrrr\s r\returncCsdS)NTrrrrr is_availablesr^cCst}|tkrtdurtrtadSdSdS|tkr-tdur)tr+tadSdSdS|tkrAt dur=t r?t adSdSdS|t krQt durSt rUt adSdSdSdSr) rcommunication_backend_namerK nccl_backendrVr2rM mpi_backendrOrPrR hccl_backend) backend_namerrr set_backends$ rdF broadcastcCtj||||dS)N)tensorsrcrXasync_op)r2re)rgrhrXrir(r)r&rrrrecCrf)N object_listrhrXdevice)r2broadcast_object_listrkrrrrnrjrn all_gathercCrf)N) tensor_listrgrXri)r2ro)rprgrXrir(r)r&rrrro cC tdurts JdtSrZ)r2rVhas_reduce_scatter_tensorrrrrrsrsc Csxtdurts Jdtrt|||||||dStdkr&tjdtt |t |}t |||||||dS)Nr[)oprXrir(r&rzunable to find torch.distributed.reduce_scatter_tensor. will fall back to torch.distributed.reduce_scatter which will result in suboptimal performance. please consider upgrading your pytorch installation.) r2rVrsreduce_scatter_tensorget_rankr rL warning_oncelisttorchchunkget_world_sizereduce_scatter) output_tensorrgrurXrir(r&input_tensor_lstrrrreduce_scatter_fns.  rrvcCtj|||||dS)N)r~ input_tensorrurXri)r2rv)r~rgrurXrir(r)r&rrrrvs  all_gather_into_tensorcCrf)N)r~rrXri)r2r)r~rgrXrir(r)r&rrrr.rqcCrrrZ)r2rVhas_all_gather_into_tensorrrrrr:rtrcCsptdurts Jdtrt|||||dStdkr$tjdtt |t |}t |||||dS)Nr[)rXrir&rzunable to find torch.distributed.all_gather_into_tensor. will fall back to torch.distributed.all_gather which will result in suboptimal performance. please consider upgrading your pytorch installation.) r2rVrrrwr rLrxryrzr{r|ro)r~rrXrir&output_tensorsrrr allgather_fnAs  rall_to_all_singlec Cstj||||||dS)N)outputinputoutput_split_sizesinput_split_sizesrXri)r2r) rrgrrrXrir(r)r&rrrrPs cCrfNrXri)r2 all_to_all)output_tensor_listinput_tensor_listrXrirrrrcrjrsendcCrfN)rgdstrXtagr2rrgrrXrr(r)r&rrrrirjrecvcCrfN)rgrhrXrr2rrgrhrXrr(r)r&rrrrorjisendcCrfrrrrrrrurjirecvcCrfrrrrrrr{rjgathercCr)N)rg gather_listrrXri)r2r)rgrrrXrir(r)r&rrrr scattercCr)N)rg scatter_listrhrXri)r2r)rgrrhrXrir(r)r&rrrrrr4cCstj||dSr)r2r4)rXri device_idsr(r)r&rrrr4smonitored_barriercCstj|||dS)N)rXrFwait_all_ranks)r2r)rXrFrr(r)r&rrrrscCsBtddtdkrtjd|dntjd|dtdddS)Nlog_summary_barrier)r)rT) print_logshow_stragglerF)r4r2rwrlog_all)rrrr log_summarys  rreducecCr)N)rgrrurXri)r2r)rgrrurXrir(r)r&rrrrrr}cCr)N)r input_listrurXri)r2r})rrrurXrir(r)r&rrrr}rcCs0tdurts JdtjdusJdtjS)Nr[z+has_all_reduce_coalesced is not yet defined)r2rVhas_all_reduce_coalescedrrrrrs rcCs0tdurts JdtjdusJdtjS)Nr[z)has_coalescing_manager is not yet defined)r2rVhas_coalescing_managerrrrrrs rcCs*tdurts Jdtj||||dS)Nr[r)r2rVall_gather_coalesced)r input_tensorsrXrirrrrsr all_reducecCt||||Sr)r2rrgrurXrir(r)r&rrrrscCst|||Sr)r2inference_all_reducerrrrrs rcCrr)r2all_reduce_coalesced)tensorsrurXrir(r)r&rrrrs rcCrrrZ)r2rVget_world_grouprrrrrrtrcC"tdurts Jdt|S)a1 Returns the number of processes in the current process group Args: group (ProcessGroup, optional): The process group to work on. If None, the default process group will be used. Returns: The world size of the process group -1, if not part of the group Nr[)r2rVr|rWrrrr|s  r|cCr)a  Returns the rank of the current process in the provided ``group`` or the default group if none was provided. Rank is a unique identifier assigned to each process within a distributed process group. They are always consecutive integers ranging from 0 to ``world_size``. Args: group (ProcessGroup, optional): The process group to work on. If None, the default process group will be used. Returns: The rank of the process group -1, if not part of the group Nr[)r2rVrwrWrrrrw)s rwcCstdurts JdtS)z Helper function to get local rank after a backend has been set and initialized Args: None Returns: local rank (= GPU device ID) Nr[)r2rVget_local_rank_from_launcherrrrrget_local_rank=s rcCs$tdurts Jdt||SrZ)r2rVget_global_rank)rX group_rankrrrrKs rc CsXtdurts Jdd}g}z |t|||d7}qttfy+Y|Sw)Nr[rTr)r2rVr7r RuntimeError ValueError)rXrD group_ranksrrrget_all_ranks_from_groupRsrcCsztdurts Jdd}ttdr+tjdtjd|d|t||}|Stdkr;tj dtjd|S) Nr[init_device_meshz&Initializing mesh device with backend z with shape z and dim names rBackend z, does not support mesh device initialization) r2rVhasattrr rLrQnamerrwrx) mesh_shapemesh_dim_names mesh_devicerrrinitialize_mesh_deviceas   r group_namecCsBtdurts Jdttdrt|dStdtjd)Nr[enable_symm_mem_for_grouprz1 does not support symmetric memory initialization)r2rVrrrr)rrrrrps  rTc Cst|d|durtdupt }tdur+tt||ttj dttdurs z"init_distributed..zRNot using the DeepSpeed or dist launchers, attempting to detect MPI environment...r%)distributed_portr%rB0rz'Distributed backend already initializedz6Initializing TorchBackend in DeepSpeed with backend {})rr2rVrTrr_rdr rLrQrz distributedr allmapin_amlin_dlts$patch_aml_env_for_torch_nccl_backend in_aws_sm'patch_aws_sm_env_for_torch_nccl_backend mpi_discoveryrHrIrJ isinstancerformat) dist_backendauto_mpi_discoveryrr%rFrGdist_init_requiredrrDrE required_envrrrinit_distributed|sB        rc sddlm}ddl}|j}|}|}d}|dkrJddl}z|d} || } | dd}Wn|j yIddl } | | }Ynw|j|dd}||} tfdd| d|D} t|tjd <t|tjd <t| tjd <|tjd <t|tjd <|rtjdtjd tjd tjd tjd tjd tdurtrt|ksJd|tt|ksJd|tdSdSdS)zM Discovery MPI environment via mpi4py and map to relevant dist state r)MPINz hostname -Izutf-8)rootcsg|]}|kqSrr).0i proc_namerr sz!mpi_discovery..rBrCrrrzfDiscovered MPI settings of world_rank={}, local_rank={}, world_size={}, master_addr={}, master_port={}z(MPI rank {} does not match torch rank {}z4MPI world size {} does not match torch world size {})mpi4pyr subprocess COMM_WORLDGet_rankGet_sizeshlexsplit check_outputdecodeCalledProcessErrorsocket gethostbyname gethostnamebcastGet_processor_name allgathersumstrrIrr rLrQrr2rVrwr|)rr%rrcommrDrE master_addrr hostname_cmdresultr all_procs local_rankrrrrsN     " rcC dtjvS)NAZUREML_EXPERIMENT_IDrrrrrr rcCr)NSM_TRAINING_ENVrrrrrrrrcCr)N DLTS_JOB_IDrrrrrrrrc Cs tjdtjd<tjdtjd<ttjdttjdk}|ssz       *       !                          E-