o ?߱i_( @sddlZddlmZz ddlmmZdZWn ey"dZYnwddlZddlm Z ddl m Z m Z m Z mZmZddlmZddlmZd e d ed e d e fd dZejjd e d ed e d e fddZd e d ed e d e fddZejjd'dej dede ded ej f ddZ d(dej eBdBdee d ej eBdBfddZ d(dej d edee d ej fddZ d)d!ejd"ed#e eeefd$ed ejf d%d&Z!dS)*N)OptionalTF)Tensor) ProcessGroup all_gatherbroadcast_object_listget_process_group_ranksget_world_size)$_verify_param_shape_across_processes) distributedxseq_dimcp_groupreturncCst|}t|}|j||dksJ|j|d||jg|jd|||j|||j|ddR}tj|g|jd}|||}|jg|jd|d|j|ddR}|S)a Split input tensor along the sequence dimension for checkpoint parallelism. This function divides the input tensor into equal parts along the specified sequence dimension, based on the number of ranks in the checkpoint parallelism group. It then selects the part corresponding to the current rank. Args: x: Input tensor to be split. seq_dim: The dimension along which to split the input (sequence dimension). cp_group: The process group for checkpoint parallelism. Returns: A slice of the input tensor corresponding to the current rank. Raises: AssertionError: If the sequence dimension is not divisible by the number of ranks. rz cannot divide cp_size N)device) rlenshapeviewtorchtensorrankr index_select)r r r Zcp_rankscp_sizeseq_idxr`/data/cameron/vidgen/cosmos-predict2.5/cosmos_predict2/_src/imaginaire/utils/context_parallel.pysplit_inputs_cp"s*> 0rc sft|}fddt|D}z t||dWnty+}ztd|d}~wwtj||dS)aa Concatenate outputs from different ranks in the checkpoint parallelism group. This function gathers tensors from all ranks in the checkpoint parallelism group and concatenates them along the specified sequence dimension. The function is decorated with @torch.compiler.disable because it contains distributed operations and dynamic tensor creation based on runtime rank information that seem to be incompatible with torch.compile's static graph compilation. Args: x: Input tensor to be concatenated. seq_dim: The dimension along which to concatenate the tensors (sequence dimension). cp_group: The process group for checkpoint parallelism. Returns: A tensor that is the concatenation of tensors from all ranks in the cp_group. Raises: RuntimeError: If the gather operation fails. cg|]}tqSrr zeros_like.0_r rr \z"cat_outputs_cp..groupFailed to gather tensors: Ndim)rranger RuntimeErrorrcat)r r r world_sizegathered_tensorserr%rcat_outputs_cpAsr3c s|}|dks Jdfddt|D}z t||dWnty3}ztd|d}~ww|}||<tj||dS) a Concatenate outputs from different ranks in the context parallelism group. This function gathers tensors from all ranks in the checkpoint parallelism group and concatenates them along the specified sequence dimension. It retains computational graph locally for each rank by replacing the portion of the tensor with original output. Args: x: Input tensor to be concatenated. seq_dim: The dimension along which to concatenate the tensors (sequence dimension). cp_group: The process group for checkpoint parallelism. Returns: A tensor that is the concatenation of tensors from all ranks in the cp_group. Raises: RuntimeError: If the gather operation fails. rz cp_size should be greater than 0crrr r"r%rrr&r'z,cat_outputs_cp_with_grad..r(r*Nr+)sizer-rr.rrr/)r r r rr1r2rrr%rcat_outputs_cp_with_gradhsr5rsrcpgis_check_shapecCst|krtj|jtjd}n tj|tjd}|r&t ||gtjj |||dt|kr?| |  |}tjj |||d|S)a] Perform a robust broadcast operation that works regardless of tensor shapes on different ranks. The function is decorated with @torch.compiler.disable because it contains distributed operations and dynamic tensor creation based on runtime rank information that seem to be incompatible with torch.compile's static graph compilation. Args: tensor (torch.Tensor): The tensor to broadcast (on src rank) or receive (on other ranks). src (int): The source rank for the broadcast. Defaults to 0. Returns: torch.Tensor: The broadcasted tensor on all ranks. )dtyper()r get_rankrrrlongcudaemptyr,r broadcast new_emptytolisttype_as)rr6r7r8rrrrrobust_broadcasts   rBitem process_groupcCs\|dur|Stt|}t|tjrt|||}|S|dur,|g}t|||d|d}|S)zM Broadcast the item from the minimum rank in the specified group(s). Nr(r)minr isinstancerrrBr)rCrDmin_rankZbroadcastable_listrrrr>s   r>cCs0|dur|Stt|}t|||}t|||S)zO Broadcast the tensor from the minimum rank in the specified group(s). N)rErrBr)rr rDrGrrrbroadcast_split_tensors    rHrrrr shape_tensorr patch_values view_factorcCststd|\}}}}}g} ||dksJ||}d} t|||gD]1\} } | dkr8|dkr8td|d|| } t| | |}||}| dkrN|} | | |q$| t_t | S)a Find the shape of input tensor for post-CP split, taking into account both temporal and spatial split, as well as patching values. The split by width is not possible currently, due to memory stride issues, which break quality. This is checked by an assert. The spatial split is achieved by flattening the input video into a single dimension before CP split is performed, and rearranging it back into [T, H, W] format after the CP split, since the input passed to the model must still be in [T, H, W] format. Args: shape_tensor (torch.Size): The shape of the Tensor that we want to split. Needs to be in [B, C, T, H, W] format. cp_size (int): The Context Parallelism size that we want to use. patch_values (tuple[int, int, int], optional): The patch values that are applied inside the Diffusion model. First element of the tuple is temporal patch size. Two next elements are the spatial patch sizes. The default value is (1, 2, 2) view_factor (int, optional): The number of views that are present in the temporal dimension. Default value is 1. Returns: The torch.Size of how the post-split tensor should look like in [T, H, W] dimensions. zPNo megatron.core package found, which is required for Context Parallelism usage.rrrzySplit by width dimension is not currently supported due to quality issues. Width dimension would be split by a factor of z0. Lower the CP size to avoid splitting by width.) USE_MEGATRON ImportError enumerate ValueErrormathgcdappendparallel_state cp_size_trSize)rJrrKrLBCTHWretrUir4 patch_sizerRrrr find_splits(  r_)F)N)rIr)"rQtypingrmegatron.core.parallel_statecorerTrMrNrrtorch.distributedrrrrrtorch.distributed.utilsr %cosmos_predict2._src.imaginaire.utilsr intrcompilerdisabler3r5boolrBstrr>rHrVtupler_rrrrsb     &'&#