o Tix@stddlZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z mZddlmZdd lmZmZdd lmZmZmZmZmZd d lmZmZmZdd lmZddl m!Z!ddlm"Z"gdZ#ej$a%dZ&dZ'ddZ(ddZ)dHddZ*Gdddej+j,Z-Gdddej+j,Z.Gdddej+j,Z/Gd d!d!ej0eZ1d"d#Z2Gd$d%d%Z3Gd&d'd'e1Z4Gd(d)d)e1Z5Gd*d+d+Z6Gd,d-d-e5Z7Gd.d/d/e5Z8Gd0d1d1e4Z9Gd2d3d3e5Z:Gd4d5d5e5Z;Gd6d7d7e4ZGdZ?Gd>d?d?e>Z@Gd@dAdAej0ZAGdBdCdCej0ZBGdDdEdEeBZCGdFdGdGej0ZDdS)IN)comm)nn) functional) Parameter)get_accelerator)get_shard_sizeget_shard_size_list) is_zero_param)ABCabstractmethod)IterableAnyOptionalListTuple)shard_value_with_share_qkshard_chunk_mlpprepare_tp_fused_qkvw) AUTOTP_MODE)deepcopy)Union) TensorParallel_LayerLinearAllreduce LinearLayerLmHeadLinearAllreduceYuan_LinearAllreduceYuan_LinearLayerGateUpPack_LinearLayerConv_LinearALlreducefused_LinearLayerconv_LinearLayerds_is_replaced_moduletensor_model_parallelcCstSN)DEEPSPEED_AUTOTP_MODEr&r&R/home/ubuntu/.local/lib/python3.10/site-packages/deepspeed/module_inject/layers.pyget_auto_tp_modesr(cC ttjkSr$r%rTRAININGr&r&r&r'is_autotp_training_mode$ r,FcCs|rtjadStjadS)zB Set the DEEPSPEED_AUTOTP_MODE based on the training flag N)rr+r% INFERENCE)trainingr&r&r'set_autotp_mode)s  r0c @sleZdZdZeddZededejde j de de j f d d Z eded e j de d e j d ffd dZd S) RowParallelzI A custom autograd function for performing row-wise parallelism. cCs|SzSymbolic function for tracing.r&graphinputr&r&r'symbolic9szRowParallel.symbolicctxgroupr5is_inference_modereturncCs>||_|dkr |S|rtj||d|Stj||d|S) Forward pass. Nr8)r8distinference_all_reduce all_reduce contiguous)r7r8r5r9r&r&r'forward>szRowParallel.forward grad_outputNcCs d|dfS) Backward pass. Nr&r7rBr&r&r'backwardLs zRowParallel.backward)__name__ __module__ __qualname____doc__ staticmethodr6r r= ProcessGrouptorchTensorboolrArrEr&r&r&r'r14s & *r1c @sVeZdZededejdejdejfddZ ededejde dejffd d Z dS) AsyncColumnParallelr7r8r5r:cCsD|du|_||_t||dd}|dur||7}||||S)r;N)use_biasr8rLmatmul transposesave_for_backward)r7r8r5weightbiasoutputr&r&r'rAVs  zAsyncColumnParallel.forwardrBNcCs||j\}}||}tj||jdd}|d|jd|d|jd}|j r2| dnd}| d|||fS)NT)r8async_oprPr) saved_tensorsrSr=r?r@r8viewshapetrRsumwait)r7rBr5rV grad_inputhandle grad_weight grad_biasr&r&r'rEes  * zAsyncColumnParallel.backward) rFrGrHrJr r=rKrLrMrArrEr&r&r&r'rOTs "(rOc @sfeZdZdZeddZededejde j de j fdd Z eded e j de d e j ffd d Z d S)ColumnParallelz? Custom autograd function for column-wise parallelism. cCst|tSr2)r=r?r@get_tensor_model_parallel_groupr3r&r&r'r6vszColumnParallel.symbolicr7r8r5r:cCs ||_|S)r;r<)r7r8r5r&r&r'rA{szColumnParallel.forwardrBNcCs.|jdkr d|fStj||jdd|fS)rCNr<)r8r=r?r@rDr&r&r'rEs zColumnParallel.backward)rFrGrHrIrJr6r r=rKrLrMrArrEr&r&r&r'rdqs "(rdcseZdZUdZdZeed<dZeed< dee j de ffdd Z e d efd d Zed d ZeddZedeejfddZddZddZddZddZddZZS)ra A base class for model layers with tensor parallelism support. This class is designed to be extended by specific layers that require distributed operations and parameter gather/partitioning during inference or training. Attributes: mode (str): The mode of operation[INFERENCE or TRAINING], default is "INFERENCE". mp_group (Optional[dist.ProcessGroup]): The process group used for model parallelism. tp_world_size (int): The world size of tensor parallelism, i.e., the number of parallel workers. tp_index (int): The rank (ID) of the current worker in tensor parallelism. support_training (bool): Flag indicating whether the layer supports training (default: False). name (Optional[str]): The name of the layer, if provided. Fkeep_module_on_hosttp_overlap_commmp_groupkwargsc sxtd|_|dur$||_t|j|_t||_|j|_ |j|_ t |dd|_ | ddur:| d|_ dSdS)a2 Initializes the TensorParallel_Layer with optional model parallelism group and layer name. Args: mp_group (Optional[dist.ProcessGroup]): The process group for model parallelism. If None, no model parallelism is set. FNname)super__init__support_trainingrhr=get_world_size tp_world_sizeget_ranktp_index world_sizerankgetattrrjget)selfrhri __class__r&r'rls  zTensorParallel_Layer.__init__valuecCs ||_dS)z Set the static variable keep_module_on_host. Args: value (bool): The new value for keep_module_on_host. N)rf)clsryr&r&r'set_keep_module_on_hosts z,TensorParallel_Layer.set_keep_module_on_hostcCdS)zm Forward pass method. Must be implemented by subclasses to define layer-specific operations. Nr&rvr5r&r&r'rAzTensorParallel_Layer.forwardcCr|)z Gathers parameters across devices for distributed training. Must be implemented by subclasses in "TRAINING" mode. Nr&rv params_listr&r&r' gather_paramsr~z"TensorParallel_Layer.gather_paramsrcCr|)z Partitions the parameters for tensor parallelism. It is necessary to ensure that this function only involves the logic of params partitioning. Nr&rr&r&r' _tp_partitionsz"TensorParallel_Layer._tp_partitioncCsn|r |js Jd|dur5|r|jdurd|_nd|_t|tdt|td|j|_|j|_dSdS)a Configures the weight tensor for training with tensor parallelism. This includes enabling gradients and associating necessary methods for parameter gathering and partitioning. Args: weight (Optional[torch.Tensor]): The weight tensor to configure for tensor parallelism. If None, no action is taken. zNo implementation of backward.NTF)is_training_moderm requires_gradsetattrDS_TENSOR_MODEL_PARALLELDS_IS_REPLACED_MODULErr)rvrVr&r&r'config_tp_paramss     z%TensorParallel_Layer.config_tp_paramscCr)r$r*rvr&r&r'rr-z%TensorParallel_Layer.is_training_modecCsZ|j}||}t|D]\}}|dkr|j|_qt||t||q||t|<|S)Nrh)rx__new__varsitemsrhrrid)rvmemorznew_objkeyryr&r&r' __deepcopy__s   z!TensorParallel_Layer.__deepcopy__cCsld\}}|jdur t|jr|jjddn|jjdd\}}|jdur)|jjnd}d|||jdu|S)N)NNrQz2in_features={}, out_features={}, bias={}, dtype={})rVr ds_shaper\dtypeformatrW)rv out_features in_featuresrr&r&r' extra_repr s  zTensorParallel_Layer.extra_reprcCsR|jr|S|jjr dnt}|jj }|j||d}|r'tjd|jd|_ |S)Ncpu)copyr)device) is_metarxrfrcurrent_device_nametorLemptyrdata)rvtensorrreturn_new_copy cloned_tensorr&r&r'moves zTensorParallel_Layer.move)rFrGrHrIrfrN__annotations__rgrr=rKr rl classmethodr{r rArrrLrMrrrrrr __classcell__r&r&rwr'rs&      rcCs0dg}|D]}t||rtt|t||qdS)Nrg)hasattrrrrt)config runtime_keysrr&r&r'!configure_tensor_parallel_runtime,s  rc@sfeZdZdZ ddeeejejfdejj de fddZ dejd e fd d Z dd dZ dddZd S)GatherReplacedLayerParamsz A context manager for gathering parameters of a replaced layer, enabling partitioning and gathering functionality based on the configuration of the model. Tparamsmoduleenabledcsb|_|_|s dSt|trt|tjst|_n|g_tfdd|Ds/d_dSdS)a Initialize the context manager to handle parameter gathering and partitioning for a replaced layer. Args: params (Iterable or torch.Tensor): A collection or single parameter to manage. module (torch.nn.Module): The module that these parameters belong to. enabled (bool): Flag indicating whether the parameter management is enabled (default: True). Nc3s|]}|VqdSr$)_is_replaced_module_weight).0prr&r' Qsz5GatherReplacedLayerParams.__init__..F) rr isinstancer rLrMlistrany)rvrrrr&rr'rl9s  z"GatherReplacedLayerParams.__init__paramr:cCs t|tdS)a  Helper function to determine if a parameter belongs to a replaced module. Args: param (torch.Tensor): The parameter to check. Returns: bool: True if the parameter belongs to a replaced module, False otherwise. F)rtr)rvrr&r&r'rUs z4GatherReplacedLayerParams._is_replaced_module_weightNcC |jr|jd|jdSdS)zc Enter the context manager. If enabled, gather parameters for the replaced module. rN)rrrrr&r&r' __enter__asz#GatherReplacedLayerParams.__enter__cCr)zi Exit the context manager. If enabled, partition the parameters for the replaced module. rN)rrr)rvexc_type exc_value tracebackr&r&r'__exit__hsz"GatherReplacedLayerParams.__exit__)T)r:N)rFrGrHrIrr rLrMrModulerNrlrrrr&r&r&r'r3s  rcsLeZdZfddZddZeddZeddZd d Z Z S) rc sjtt|j|fi||j|_|j|_||j|jgd|_||j|jdur3||jdSdSNT)rkrrlrVrWrrmrrvrrhrirwr&r'rlss  zLinearAllreduce.__init__cCsDt||jdd}t|j|| }|jdur ||j7}|S)NrPrQ) rLrSrVrTr1applyrhrrWrvr5rXr&r&r'rA~s   zLinearAllreduce.forwardcCst|D]E\}}|dus|dkrdS|j||_|dd}tj|j|jd|jd|j |j d}t j |||j d|dd||_qdSNrrrrr<) enumeraterdata_partitionrTr@rLrror\rrr=all_gather_into_tensorrhrvridxr output_paramr&r&r'rs zLinearAllreduce.gather_paramscCsr|s ||dSt|D]'\}}|dus|dkrdStj||jdd|j}||}|||_ qdS)NrrPdim runeven_partitionrrLchunkrorqrdetachrrvrrr _partitionr&r&r'rs  zLinearAllreduce._tp_partitioncCst|D];\}}|dus|dkrdS|jdusJd||jt||jd|j|jdd|j}||}|||_ qdS)Nr7The module name must be provided in the initialization.rr rrjsplitrr\rorqrrrrr&r&r'rs z LinearAllreduce.uneven_partition) rFrGrHrlrArLno_gradrrrrr&r&rwr'rqs    rcsfeZdZdfdd ZddZeddZed d Zd d Z e dej ddfd dZ Z S)rNFc sntt|j|fi||j|_|j|_|s||j|jgd|_||j|jdur5||jdSdSr)rkrrlrVrWrrmrrvrrhskip_partitionrirwr&r'rls  zLinearLayer.__init__cCsn|jjs*t|dddurt|j|}t||j dd}|j dur(||j 7}|St |j||j|j }|S)NrhrPrQ) rxrgrtrdrrhrLrSrVrTrWrOrr&r&r'rAs  zLinearLayer.forwardcCsjt|D].\}}|j||_tj|j|jd|jd|j|jd}t j |||j d| ||_qdSr) rrrrLrror\rrr=rrhr@rr&r&r'rs zLinearLayer.gather_paramscCsj|s ||dSt|D]#\}}|durdStj||jdd|j}||}|||_ qdS)Nrrrrr&r&r'rs  zLinearLayer._tp_partitioncCs|t|D]7\}}|durdS|jdusJd||jt||jd|j|jdd|j}||}|||_ qdS)Nrrrrrr&r&r'rs zLinearLayer.uneven_partitioncCsz|dur%|jd}|jd}tj|||dud}||j_|dur$||j_n|d}|d}tj|||dud}||ddS)Nrr)rWT)r)r\rLinearrVrrW)rz weight_shaperrVrWrrlinearr&r&r' from_weightss   zLinearLayer.from_weightsNF)rFrGrHrlrArLrrrrrhalfrrr&r&rwr'rs   rc@s$eZdZdejfddZddZdS)FusedModuleWrapper fused_modulecCs ||_dSr$r)rvrr&r&r'rl zFusedModuleWrapper.__init__cCs|jSr$r)rvrr&r&r' __getattr__szFusedModuleWrapper.__getattr__N)rFrGrHrrrlrr&r&r&r'r s rcs.eZdZdfdd ZeddZZS)r Fc sB|ddus Jdt|d|_tj|||fi|dS)Nrz+'fused_module' is required but not provided)rurrrkrlrrwr&r'rlszfused_LinearLayer.__init__cCsRt|D]"\}}|durdSt|jj||j|j}||}|||_qdSr$) rrrrrorqrrrrr&r&r'rs zfused_LinearLayer._tp_partitionF)rFrGrHrlrLrrrr&r&rwr'r sr c@eZdZeddZdS)r!cCsd}d}t|dkr|d}nt|dkr|d|d}}|jjt|jd|j|jdd|j}|| }||_|dur`|jjt|jd|j|jdd|j}|| }||_dSdS)Nrrr) lenrrrr\rorjrqrr)rvrrVrWrr&r&r'r-s,    zconv_LinearLayer._tp_partitionNrFrGrHrLrrr&r&r&r'r!+r!c@r)rcCsFt|dj|d|j|jd\}}||d_|dur!||d_dSdS)NrrF)rrrqrorvrrVrWr&r&r'rFs z"Yuan_LinearAllreduce._tp_partitionNrr&r&r&r'rCsrc@r)rcCsZt|dj|d|j|jd\}}|||d_|dur+|||d_dSdS)NrrT)rrrqrorrrr&r&r'rQszYuan_LinearLayer._tp_partitionNrr&r&r&r'rOrrc@r)rcCsXt|dj|d|j|j\}}|||d_|dur*|||d_dSdS)Nrr)rrrqrorrrr&r&r'r\s z$GateUpPack_LinearLayer._tp_partitionNrr&r&r&r'rZrrc@r)rcCsvt|D]4\}}|durdS|jdd|_|jt|jd|j|jdd|j }| | }|||_qdS)NrPrQrrr) rrrTr@rrr\rorjrqrrrr&r&r'rfs z"Conv_LinearALlreduce._tp_partitionNrr&r&r&r'rdrrcs$eZdZfddZddZZS)rc s`d|_t|j|_t|dr#|jdur#t|j|_tj ||fi|dS)Nlm_headrW) rjrrrVclonerrrWrkrlrrwr&r'rlxs zLmHeadLinearAllreduce.__init__cCst|jd|jd}tt|jd|jdd|j}t|dddd|||f|j dd}|j dur@t j ||j d|j durJ||j 7}|S)NrPrrrQr<)rr\ror^rrqrLrSrVrTrhr=r>rW)rvr5input_shard_sizeinput_shard_offsetrXr&r&r'rAs"     zLmHeadLinearAllreduce.forward)rFrGrHrlrArr&r&rwr'rvs rcs:eZdZfddZddZdejdejfddZZS) TensorParallelConv2dcs*t||_||_||_||dSr$)rkrlrsrr shard_by_oc shard_weights)rvconvrsrrrrwr&r'rls zTensorParallelConv2d.__init__c Cst|jr |jjd}n|jjd}d}dg}t|jdddD]}||j}|||jkr0|d7}||d|q|jj}|jre|||j||jd}|jdurd|jj||j||jd}n |dd||j||jdf}|jdur|jjt |j}t |jd|jd|j |j |j|j|j|jdu|j |_tj ||j_|jdurtj ||j_~dS)NrrrP)rrVr\rangerrappendrrsrWfloatrConv2d kernel_sizestridepaddingdilationgroups padding_moderrLr)rvr total_size bias_data cols_per_rankicols weight_datar&r&r'rs4    $   z"TensorParallelConv2d.shard_weightsr5r:cCs ||Sr$)rr}r&r&r'rArzTensorParallelConv2d.forward) rFrGrHrlrrLrMrArr&r&rwr'rs rcseZdZfddZZS)TensorParallelOcShardConv2dct|||ddSrrkrlrvrrsrrrwr&r'rlz$TensorParallelOcShardConv2d.__init__)rFrGrHrlrr&r&rwr'rsrcs2eZdZfddZdejdejfddZZS)TensorParallelIcShardConv2dcrrrrrwr&r'rlrz$TensorParallelIcShardConv2d.__init__r5r:cCs"||}|jdkrt||S)Nr)rrrr=r>)rvr5outr&r&r'rAs   z#TensorParallelIcShardConv2d.forward)rFrGrHrlrLrMrArr&r&rwr'rs rcs2eZdZdejdddffdd ZddZZS) NormalizeNh㈵>csdtt||dur||_||_ntj||d|t |_ |j j|_|j j|_||_ dS)Neps) rkr rlrVrWr LayerNormrrrnormr )rvrrr rVrWrwr&r'rls"   zNormalize.__init__cCs&tjj||jdd|j|j|jdS)NrPr )rr layer_normr\rVrWr r}r&r&r'rAs&zNormalize.forwardrFrGrHrLrrlrArr&r&rwr'r s r cs0eZdZdejddffdd ZddZZS)EmbeddingLayerNcsJtt||dur ttj|d|d|td|_dS||_dS)Nrrr) rkrrlrrLrrrrV)rvrrrVrWrwr&r'rls  zEmbeddingLayer.__init__cCst||jSr$)F embeddingrVr}r&r&r'rAszEmbeddingLayer.forward)rFrGrHrLrrlrArr&r&rwr'rs rcs@eZdZdZd fdd Zd dejdedeffd d ZZ S) OPTEmbeddingzN This module learns positional embeddings up to a fixed maximum size. Ncsd|_tj||ddS)Nr)rV)offsetrkrl)rvrrVrWrwr&r'rlszOPTEmbedding.__init__rattention_maskpast_key_values_length position_idscsN|}tj|dd||d}|dd|df}t||jS)z3`input_ids_shape` is expected to be [bsz x seqlen].rrN)longrLcumsumtype_asrkrAr)rvrrr positionsrwr&r'rAs zOPTEmbedding.forward)NNN)rr) rFrGrHrIrlrL LongTensorintrArr&r&rwr'rs&rcs0eZdZdejddffdd ZddZZS) RMSNormalizeNr csFtt||dur||_nttj||t d|_||_ dS)Nr) rkrrlrVrrrLonesrrr )rvrrr rVrwr&r'rl s  zRMSNormalize.__init__cCs\|tjdjddd}|t||j}|jjtj tj fvr)||jj}||jS)NrrPT)keepdim) rrLfloat32powmeanrsqrtr rVrfloat16bfloat16)rv hidden_statesvariancer&r&r'rAs  zRMSNormalize.forwardrr&r&rwr'r s rr)ErL deepspeedrr=rtorch.nnrrtorch.nn.parameterrdeepspeed.acceleratorr deepspeed.module_inject.tp_shardrrdeepspeed.runtime.zero.utilsr abcr r typingr r rrrfusedqkv_utilsrrr!deepspeed.runtime.tensor_parallelrrrr__all__r.r%rrr(r,r0autogradFunctionr1rOrdrrrrrrrr r!rrrrrrrrr rrrr&r&r&r'sZ            >FV    ,