o ?߱iF@sddlmZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z mZmZmZddlZddlZddlmZddlmZddlmZer`ddlmZdd lmZmZdd lm Z e rndd l!m"Z"dMddZ#dNdOddZ$dNdOddZ%dPddZ&dPddZ'dQddZ(dRd!d"Z)dQd#d$Z*dSd*d+Z+Gd,d-d-ej,j-j.Z.ed.d/Z/dTd3d4Z0e1dUd8d9Z2dVd=d>Z3dWd@dAZ4dXdCdDZ5   EdYdZdKdLZ6dS)[) annotationsN)contextmanager) timedelta) TYPE_CHECKINGAnyCallable ContainerOptional)get_process_group_ranks)Device)_get_default_group)_sync_module_states$_verify_param_shape_across_processes)log) DDPConfigreturn int | Nonec CsRtr tjSttt dd}zt |}t d| Wntj y>}ztd|WYd}~nd}~wwdtjd<dtjd<trutj|t d d }tt|d }tjd d |dtjd|d|ddtd}ttjdttj}|tdtd||tdtdtddS)z Initialize distributed training. LOCAL_RANKrzFailed to set device affinity: N0TORCH_NCCL_BLOCKING_WAIT1TORCH_NCCL_ASYNC_ERROR_HANDLINGZ TORCH_NCCL_HEARTBEAT_TIMEOUT_SECi)secondsncclzenv://)backend init_methodtimeoutz1Initialized distributed training with local rank z with timeout F) rank0_onlyz libcudart.sozTraining with z GPUs.)distis_initializedtorchcudacurrent_devicepynvmlnvmlInitintosgetenvr sched_setaffinityZget_cpu_affinity NVMLErrorrwarningenviron is_available set_devicerinit_process_groupcriticalctypesCDLLcastc_intPOINTERZcudaDeviceSetLimitZcudaDeviceGetLimitinfoget_world_size) local_rankdeviceeZtimeout_secondsZtimeout_timedeltaZ _libcudartp_valuer>[/data/cameron/vidgen/cosmos-predict2.5/cosmos_predict2/_src/imaginaire/utils/distributed.pyinit,s6      r@groupOptional[dist.ProcessGroup]r(cC"d}trtrt|}|S)zgGet the rank (GPU device) of the worker. Returns: rank (int): The rank of the worker. r)r!r/r"get_rank)rArankr>r>r?rDP rDcCrC)zGet world size. How many GPUs are available in this job. Returns: world_size (int): The total number of GPUs available in this job. r)r!r/r"r9)rA world_sizer>r>r?r9\rFr9boolcCs tdkS)zCheck if current process is the master GPU. Returns: (bool): True if this function is called from the master GPU, else False. r)rDr>r>r>r?is_rank0hs rIcCstjdkS)zCheck if current process is the local master GPU in the current node. Returns: (bool): True if this function is called from the local master GPU, else False. r)r#r$r%r>r>r>r?is_local_rank0qsrJfuncrctfdd}|S)aApply this function only to the master GPU. Example usage: @rank0_only def func(x): return x + 3 Args: func (Callable): a function. Returns: (Callable): A function wrapper executing the function only on the master GPU. cstr |i|SdSN)rI)argskwargsrKr>r?wrapperszrank0_only..wrapper functoolswrapsrKrQr>rPr?rzsrNonecCs$trtrtdSdSdS)zBarrier for all GPUs.N)r!r/r"barrierr>r>r>r?rWs rWcrL)z6run the function on rank 0 first, then on other ranks.cs2tr |i|}tts|i|}|SrM)rIrW)rNrOresultrPr>r?rQs zrank0_first..wrapperrRrUr>rPr? rank0_firstsrY config_ddprmodeltorch.nn.Module)torch.nn.Module | DistributedDataParallelc CstrLtrLttdd}zddlm}|jdd}Wnt y<}zt |t dd}WYd}~nd}~wwt ||g||j |j|j|d}|S) aWraps the model to enable data parallalism for training across multiple GPU devices. Args: config_ddp (DDPConfig): The data parallel config. model (torch.nn.Module): The PyTorch module. Returns: model (torch.nn.Module | DistributedDataParallel): The data parallel model wrapper if distributed environment is available, otherwise return the original model. rr)parallel_stateT)Zwith_context_parallelzAparallel_state not initialized, treating all GPUs equally for DDPN) device_ids output_devicefind_unused_parameters static_graphbroadcast_buffers process_group)r!r/r"r(r)r* megatron.corer^Zget_data_parallel_group Exceptionrr8DistributedDataParallelrarbrc)rZr[r:r^Z ddp_groupr<r>r>r?parallel_model_wrappers*     rhcs,eZdZdZd fdd Zd dd ZZS) rgaThis extends torch.nn.parallel.DistributedDataParallel with .training_step(). This borrows the concept of `forward-redirection` from Pytorch lightning. It wraps an ImaginaireModel such that model.training_step() would be executed when calling self.training_step(), while preserving the behavior of calling model() for Pytorch modules. Internally, this is a double rerouting mechanism (training_step -> forward -> training_step), allowing us to preserve the function names and signatures. r[r\cs$tj|g|Ri|d|_dS)NT)super__init__#show_sync_grad_static_graph_warning)selfr[rNrO __class__r>r?rjs z DistributedDataParallel.__init__rrcs,jjfdd}|j_|i|S)Ncsj_jj|i|SrM)moduleforward training_step)_args_kwargsoriginal_forwardrlr>r?wrapped_training_stepszDDistributedDataParallel.training_step..wrapped_training_step)rorp)rlrNrOrvr>rtr?rqsz%DistributedDataParallel.training_step)r[r\)rr)__name__ __module__ __qualname____doc__rjrq __classcell__r>r>rmr?rgsrgc cst|tjjs Jt|tr)|j}|jr&|j|kr&|jr%t dd|_n||_zdVWt|tr8||_dSdSt|trB||_w)a Context manager to enable/disable gradient synchronizations across DDP processes for DDP model. Modified from: https://pytorch.org/docs/stable/_modules/torch/nn/parallel/distributed.html#DistributedDataParallel.no_sync Note that this is incompatible with static_graph=True and will be an no-op if static_graph=True. Within this context, gradients will be accumulated on module variables, which will later be synchronized in the first forward-backward pass exiting the context. .. warning:: The forward pass should be included inside the context manager, or else gradients will still be synchronized. zTDDP static_graph=True is incompatible with sync_grad(). Performance will be reduced.FN) isinstancer#nnModulergrequire_backward_grad_syncrbrkrr-)r[enabledold_require_backward_grad_syncr>r>r? ddp_sync_grads     r data_batcheslist[dict[str, torch.Tensor]]&torch.Tensor | dict[str, torch.Tensor]cs@t|dtjrztj|dd}tjt|dd}tj|tjj dt||krKt|d|ks1Jt |dd}tj||gdd}tjddd}ntjddd}tj|tjj dt | }tj|ddjddd}|dkrx|d| }|St|dtjjrt}|dD]tfd d |D|<q|St) a6Aggregate the list of data batches from all devices and process the results. This is used for gathering validation data batches with cosmos_predict2._src.imaginaire.utils.dataloader.DistributedEvalSampler. It will return the data/output of the entire validation set in its original index order. The sizes of data_batches in different ranks may differ by 1 (if dataset size is not evenly divisible), in which case a dummy sample will be created before calling dis.all_gather(). Args: data_batches (list[dict[str, torch.Tensor]]): List of tensors or (hierarchical) dictionary where leaf entries are tensors. Returns: data_gather (torch.Tensor | dict[str, torch.Tensor]): tensors or (hierarchical) dictionary where leaf entries are concatenated tensors. r)dimr$)r;)oprN) start_dimend_dimcsg|]}|qSr>r>).0datakeyr>r? /z#collate_batches..)r|r#Tensorcattensorlenr! all_reduceReduceOpMAX empty_likeSUMall_gather_tensor contiguousstackflatten collectionsabcMappingdictkeyscollate_batches TypeError)rZ data_concatZmax_num_local_samplesdummyZ dummy_countZ data_collater>rr?rs,  rr torch.Tensorlist[torch.Tensor]cs(fddttD}t||S)zGather the corresponding tensor from all GPU devices to a list. Args: tensor (torch.Tensor): Pytorch tensor. Returns: tensor_list (list[torch.Tensor]): A list of Pytorch tensors gathered from all GPU devices. csg|]}tqSr>)r# zeros_like)r_rr>r?r?sz%all_gather_tensor..)ranger9r! all_gather)r tensor_listr>rr?r5s rpayloadrlist[Any] | NonecCs8tt}}|dkrdg|nd}tj||dd|S)a#Gather the corresponding object from all GPU devices to a rank 0 hosted list. Args: payload: Any pickle-able object. Returns: payload_list (list[Any]) | None: Rank 0: A list of Pytorch tensors gathered from all RANK process. Rest : None rN)object_gather_listdst)rDr9r! gather_object)rrErGZpayload_gatheredr>r>r?rDs rFcCs(t}|dkr |Stj||||ddS)N)srcrAasync_op)r9r! broadcast)rrrArrGr>r>r?rUsrmeancCst}|dkr |St5tj||dt|kr*|dkr#||}n|dkr(ntWd|SWd|SWd|S1sEwY|S)zReduce to rank 0r)rrsumN)r9r#no_gradr!reducerDNotImplementedError)rrErrGr>r>r?dist_reduce_tensor\s,          rTrdrparams_and_buffers_to_ignoreOptional[Container[str]]rccstrts dS|durt}sttd|dt|dfdd|D}tfdd|D}dd|D}t |d krJdSt ||t ||t d ||d dS) aD Modify based on DDP source code Synchronizes the parameters and buffers of a model across different processes in a distributed setting. This function ensures that all processes in the specified process group have the same initial parameters and buffers from the source rank, typically rank 0. It is useful when different processes start with different model states and a synchronization is required to ensure consistency across all ranks. Args: model (nn.Module): The model whose parameters and buffers are to be synchronized. process_group (dist.ProcessGroup, optional): The process group for communication. If None, the default group is used. Defaults to None. src (int, optional): The source rank from which parameters and buffers will be broadcasted. Defaults to 0. params_and_buffers_to_ignore (Optional[Container[str]], optional): A container of parameter and buffer names to exclude from synchronization. Defaults to None, which means all parameters and buffers are included. broadcast_buffers (bool, optional): Whether to broadcast buffers or not. Defaults to True. Side Effects: This function modifies the state of the model in-place to synchronize it with the source rank's model state. Raises: RuntimeError: If the shapes of parameters across processes do not match, a runtime error will be raised. Examples: >>> # downloading duplicated model weights from s3 in each rank and save network bandwidth >>> # useful and save our time when model weights are huge >>> if dist.get_rank == 0: >>> model.load_state_dict(network_bound_weights_download_fn(s3_weights_path)) >>> dist.barrir() >>> sync_model_states(model) # sync rank0 weights to other ranks Nz%Synchronizing model states from rank z to all ranks in process group .cs:g|]\}fdd|jddDD]}||fqqS)cs&g|]\}}d|vr|qS)rr>)r param_nameparam) module_namerr>r?rs z0sync_model_states...F)recurse)named_parameters)rro parameter)r)rr?rs  z%sync_model_states..cs*g|]\}}|vr|s||fqSr>)add)rmp)memor>r?rs cSsg|]\}}|qSr>r>)rrrr>r>r?rrri)rordbroadcast_bucket_sizerrrc) r!r/r"r setrr8r named_modulesrrr r()r[rdrrrcmodules_and_parameters parametersr>)rrr?sync_model_statesms8(     r)rrrM)rArBrr()rrH)rKrrr)rrV)rZrr[r\rr])rrrr)rrrr)rrrr)NF)rr)NrNT) r[r\rdrBrr(rrrcrH)7 __future__rrcollections.abcr3rSr) contextlibrdatetimertypingrrrrr r&r#torch.distributed distributedr!r Z,cosmos_predict2._src.imaginaire.utils.devicer r/"torch.distributed.distributed_c10dr torch.distributed.utilsr r%cosmos_predict2._src.imaginaire.utilsrZ&cosmos_predict2._src.imaginaire.configrr@rDr9rIrJrrWrYrhr}parallelrgrrrrrrrrr>r>r>r?sV           $   "  -