tjN &UddlmZddlZddlZddlZddlZddlmZddlm Z ddl m Z m Z ddl mZddlmZer)ddlZddlmZdd lmZejZe jeZd Z dndodZdpdZdqdrdZerEejejej ej!ej"ej#ej$ej%ej&ej'ej(d Z)dsdZ*d Z+ dtdud&Z,dvdwd(Z-d)Z.Gd*d+ej/j0Z1Gd,d-ej/j0Z2Gd.d/ej/j0Z3Gd0d1ej/j0Z4Gd2d3ej/j0Z5d4Z6d5Z7d6Z8d7Z9d8Z: dndxd;Z;Gd<d=Z<Gd>d?e<Z=Gd@dAe<Z>GdBdCe<Z?GdDdEe<Z@GdFdGe<ZAGdHdIe=ZBGdJdKeAZCGdLdMe<ZDGdNdOe<ZEGdPdQe<ZFGdRdSe<ZGGdTdUeGZHGdVdWe<ZIGdXdYe<ZJGdZd[e<ZKGd\d]eZLeLZMd]eNd^<dydcZOdzdfZPdgZQdhZRd{dlZSdmZTdS)|) annotationsN)reduce)DistributedConfig)is_torch_greater_or_equallogging)GeneralInterface)is_torch_available)nncttjdrWttjjdr8t |tjjjr|S|S)aUnwrap a `DTensor` to its local shard if needed; pass through otherwise. Custom kernels (CUTLASS, CuteDSL, Triton) take raw tensor pointers and don't understand `DTensor`, so weights wrapped by FSDP2 / EP need this unwrap before they can be fed to the kernel. ``to_local()`` is autograd-aware on the train path: backward rewraps the gradient as a DTensor matching each parameter's placements. tensorDTensor)hasattrtorch distributedr isinstancerto_local)ts g/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/transformers/integrations/tensor_parallel.pyrr(s_u (++ 8I8PR[0\0\ a*19 : : ::<<  Htp_planstr | dict[str, str] | Nonetp_size int | Nonec||td||td|?tdstdtjj}|dkrtdtt|}tj s ttj d}ttj d }ttj d }d d d dddd} | |} tj | ||tt|}|dkr||n"#t"$r} td| d} ~ wwxYw|dkr^|ttj d |} tj|| } | }ntj|} |pi}||ntj }tj | j|f}nz|jdkr d|jvrtd|d}|}tj|jdttj d }|||fS)z Sets up the device mesh and initialized the backend for tensor parallelism. This function is called when the model is loaded and the TP plan is set to 'auto'. Nz-tp_plan has to be set when tp_size is passed.zY`tp_plan` and `device_map` are mutually exclusive. Choose either one for parallelization.z2.5z3Tensor parallel is only supported for `torch>=2.5`.mpsz3Tensor parallelism is not supported on MPS devices.RANK LOCAL_RANK WORLD_SIZEncclglooxcclhcclneurontpu_dist)cudacpuxpuhpur$tpu)backendrank world_sizer'zWe tried to initialize torch.distributed for you, but it failed. Make sure you init torch distributed in your script to use `tp_plan`.tpzsWhen using `tp_plan` and n-d `device_mesh`, it must contain a 'tp' dimension. Please provide a valid `device_mesh`.:) ValueErrorrOSErrorr_C_get_acceleratortype RuntimeErrorgetattrris_initializedintosenvirongetinit_process_group set_device Exceptioncurrent_devicedeviceget_world_sizeinit_device_meshndimmesh_dim_namessize device_type)rr device_mesh device_maprGr@r, local_rankr- backend_mapr+eindex tp_devices rinitialize_tensor_parallelismrO7swHIIIz5tuuu(// QOPP Ph//116 %  TUU U  44 //11  2:f-.. L!9::  L!9:: #!!!&%  &//+66!44W4\f4ggg!( !z2replace_layer_number_by_wildcard..sD1771::,=r)resub)rPs r replace_layer_number_by_wildcardr[s 6."="=t D DDrTparameter_namedict[str, str] str | Nonect|}||vr||S|r,d|vr(|dddx}|vr||SdS)a Get the TP style for a parameter from the TP plan. The TP plan is a dictionary that maps parameter names to TP styles. The parameter name can be a generic name with wildcards (e.g. "*.weight") or a specific name (e.g. "layer_1.weight"). The `is_weight` is important because for weights, we want to support `.weights` and `.bias` cases seamlessly! but not parent classes for `post_init` calls .r.rN)r[rsplit)r\r is_weightgeneric_param_name module_names r_get_parameter_tp_planresu:.IIW$$)** $s000EWE^E^_bdeEfEfghEi6iknu5u5u{## 4r) BOOLU8I8I16F16BF16I32F32F64I64F8_E4M3 total_sizer9blocksint | list[int] list[int]ct|tr;t|}||zdksJd|d|||zfd|DS||zdks Jd|||z}|g|zS)a Convert block count or proportions to block sizes. This function accepts - The number of blocks (int), in which case the block size is total_size//blocks; or - A list of block sizes (list[int]). In the second case, if sum(blocks) < total_size, the ratios between the block sizes will be preserved. For instance, if blocks is [2, 1, 1] and total_size is 1024, the returned block sizes are [512, 256, 256]. rz Cannot split z in proportional blocks: cg|]}|zSrw).0block part_sizes r z*_blocks_to_block_sizes..s666e E!666rzPrepacked is not divisible by )rlistsum)rqrr total_blocks single_sizerzs @r_blocks_to_block_sizesrs&$&6{{ L(A---/lz/l/ldj/l/l---,. 6666v6666F"a''')R&)R)R''' F* }v%%rc`|}|j|}|}t|d}g} d} |D]2} | |z} || z} |dz| z}| t| | z| |zz } | | z } 3|}d}|dks|dkr'|dt j}d }|dkr || df}nD|dks|d kr|d d | df}n*|dks|d kr |d| f}ntd |d|r|S|t|S)u When weights are packed (gate_up_proj), we need to make sure each shard gets its correct share. So if you have: gate_proj ( 16, 5120, 8190) and up_proj ( 16, 5120, 8190) packed as gate_up_proj ( 16, 5120, 2 * 8190) And you shard along the last dimension, you need to interleave the gate and up values: Now, if we shard along the last dimension across TP_size (Tensor Parallelism size), we must interleave the values from gate and up projections correctly. Let's take TP_size = 4 for an example: Packed tensor `gate_up_proj` --------------------------------------------------------------- [ G0 G1 G2 G3 | G4 G5 G6 G7 | ... | U0 U1 U2 U3 | U4 U5 U6 U7 | ... ] ↑─────────────↑ ↑─────────────↑ ↑─────────────↑ ↑─────────────↑ Gate Slice 0 Gate Slice 1 Up Slice 0 Up Slice 1 Explanation: - The first half of the tensor (left of the center) holds the gate_proj values. - The second half (right of the center) holds the up_proj values. - For TP=4, we divide each half into 4 slices. In this example, we show two slices for brevity. - Each shard receives one slice from the gate part and the corresponding slice from the up part. For instance: • Shard 0 gets: [ Gate Slice 0, Up Slice 0 ] = [ G0, G1, G2, G3, U0, U1, U2, U3 ] • Shard 1 gets: [ Gate Slice 1, Up Slice 1 ] = [ G4, G5, G6, G7, U4, U5, U6, U7 ] • … and so on. This ensures that each shard receives an equal portion of both gate and up projections, maintaining consistency across tensor parallelism. r)rqrrrr.FrpF8_E5M2.TNzUnsupported dim z", only dim 0, 1 or 2 are supported) shaperFrrange get_dtypetorfloat16r1 str_to_dtype)param empty_paramrHr,dimslice_rqr- block_sizestensors_slices block_offset block_sizeshard_block_sizestartstop slice_dtypecastedr s rget_packed_weightsrs>F"3'J!!##J(JqIIIKNL!## %3''q,,% u 4lT6IJJJ " ""$$KFi;)#;#; .. axx+, SBYY>3./ SBYY^+,SCSSSTTT 4 yyk2333rpacked_parameter torch.Tensor sharded_dimr- num_blocksc|dkrtd|dkr|n ||jz}|j|}||z}||z}|jd|}|j|dzd} |jg||||| R} t |} t |dz} t t | j} | | | | c| | <| | <| j| }||}|S)as Reorders a tensor that was reconstructed from sharded packed weights into its canonical packed format. For example, if a weight was packed (e.g., gate_proj and up_proj) and then sharded, DTensor.full_tensor() might produce an interleaved layout like [G0, U0, G1, U1, ...] along the sharded dimension. This function reorders it to [G0, G1, ..., U0, U1, ...]. This is an inverse operation to get_packed_weights. Args: reconstructed_tensor: The tensor reconstructed from DTensor (e.g., via .full_tensor().contiguous()). sharded_dim: The dimension index in the reconstructed_tensor that was originally sharded. world_size: The tensor parallel world size. num_packed_projs: The number of projections that were packed together (e.g., 2 for gate_up_proj). Returns: The reordered tensor in canonical packed format. rzNum blocks different from 2 is not supported yet. This is most likely a bug in your implementation as we only pack gate and up projections together.rNr.) r1rDrviewlenr|rpermute reshape_as)rrr-ractual_sharded_dimtotal_size_on_sharded_dimoriginal_block_size_on_dimshard_chunk_size prefix_shape suffix_shape tensor_view axis_ws_abs axis_npp_abs permute_ordertensor_permutedfinal_ordered_tensors rrepack_weightsrse0Q c   )4q(8(8kL\La>a 0 67I J!:j!H1Z?#)*=+=*=>L#)*KL>Y[hit[u;M+ l ;)k)=9O+556FGG r tensor_idxc|j}|j}ttj|}t |t jrt|jn| } |dkr||z}| dkr|dkrt| dkrd}n3| dkr|dkrt| dkrd}tj | ||z } || z} t| | z| |} ||krtd|d|||krtd|d||l| dkrT|dkrNt| dkr;| |cxkr| kr nn |d d St jgt j| St%d gt| z} | | |krKt%| | | |<|t'| }t |tr d |D}|Sd| |<t jt'| t j S) a Generalized tensor sharding across a multi-dimensional device mesh. Extract only the fraction of the parameter owned by the given `rank` when the parameter would have gone sharding at provided `dim`. Extraction follows the pytorch `Shard` placement so that sharding and materializing back to full tensor follows `Shard` semantics. `Shard` follows torch.chunk style sharding of the tensor. We demonstrate some cases below on how sharding happens including some edge cases such as some ranks having an empty tensor as shard. Below implementation is robut to all these cases. Case (1) empty_param (16, 5120, 8190) dim 0 device_mesh.size() 4 rank 0 gets (4, 5120, 8190) (0 ... 4, 5120, 8190) rank 1 gets (4, 5120, 8190) (4 ... 8, 5120, 8190) rank 2 gets (4, 5120, 8190) (8 ... 12, 5120, 8190) rank 3 gets (4, 5120, 8190) (12 ... 16, 5120, 8190) Case (2) empty_param (16, 5120, 8190) dim 0 device_mesh.size() 14 rank 0 gets (2, 5120, 8190) (0 ... 2, 5120, 8190) rank 1 gets (2, 5120, 8190) (2 ... 4, 5120, 8190) rank 2 gets (2, 5120, 8190) (4 ... 6, 5120, 8190) rank 3 gets (2, 5120, 8190) (6 ... 8, 5120, 8190) rank 4 gets (2, 5120, 8190) (8 ... 10, 5120, 8190) rank 5 gets (2, 5120, 8190) (10 ... 12, 5120, 8190) rank 6 gets (2, 5120, 8190) (12 ... 14, 5120, 8190) rank 7 gets (2, 5120, 8190) (14 ... 16, 5120, 8190) rank 8 gets (0, 5120, 8190) rank 9 gets (0, 5120, 8190) rank 10 gets (0, 5120, 8190) rank 11 gets (0, 5120, 8190) rank 12 gets (0, 5120, 8190) rank 13 gets (0, 5120, 8190) Case (3) empty_param (16, 5120, 8190) dim 0 device_mesh.size() 3 rank 0 gets (6, 5120, 8190) (0 ... 6, 5120, 8190) rank 1 gets (6, 5120, 8190) (6 ... 12, 5120, 8190) rank 2 gets (4, 5120, 8190) (12 ... 16, 5120, 8190) In case (2), empty shards are returned with appropriate dimension to allow for operations to work smoothly. Args: param (torch.Tensor): The tensor to shard. empty_param (torch.Tensor): A tensor used for shape reference. device_mesh (torch.Tensor): Shape [d_0, ..., d_n] representing the mesh. rank (int): Global rank of the current process/device. dim (int): Dimension along which to shard the tensor. rr.rzdim z* is out of bounds for tensor of dimension zRank z is out of bounds for mesh size NdtyperAc"g|] }|dd SNrw)rxps rr{z$get_tensor_shard..s )))aQqqqT)))r)r)rDrroperatormulrrTensorr| get_shaperrmathceilminr1emptyint64slicetuple)rrrHr,rr param_dim mesh_shaper- param_shape shard_sizerend slice_indicess rget_tensor_shardrOsh I"J j11J'1%'F'F]$u{###EOOL]L]K Qww#oA#((s;/?/?1/D/D   a  C1HH[1A1AQ1F1F;s+j899J : E ej +c"2 3 3C iZZZyZZ[[[ zSSSzSSTTT+//"3"3q"8"8SAXX#kJZJZ^_J_J_ J $ $ $ $ $ $ $ $ $8O;rTBBB B4[[MC $4$44M {3"5#.. ceM**+ eT " " *))5)))E K ;u[)) = = ==rc0tj||dS)z9Split tensor along last dimension into world_size chunks.rr)rchunk)xr-s r_split_along_last_dimrs ;q*" - - --rc>eZdZdZedZedZdS)_AllReduceBackwardzRIdentity forward, all-reduce backward. Used before colwise layers (f in Megatron).c||_|Sr)rHctxrrHs rforwardz_AllReduceBackward.forwards%rc|j}|dkr|dfS|}tj|tjj||dfSNr.oprV)rHrF contiguousdist all_reduceReduceOpSUM get_group)r grad_outputrHs rbackwardz_AllReduceBackward.backwardsoo      " "$ $!,,..    (9AVAVAXAXYYYYD  rN__name__ __module__ __qualname____doc__ staticmethodrrrwrrrrsN\\\!!\!!!rrc>eZdZdZedZedZdS)_AllReduceForwardzQAll-reduce forward, identity backward. Used after rowwise layers (g in Megatron).c|dkr|Stj|tjj||Sr)rFrrrrrrs rrz_AllReduceForward.forwardsK      " "H dm/{7L7L7N7NOOOOrc |dfSrrw)rrs rrz_AllReduceForward.backwards D  rNrrwrrrrsN[[\ !!\!!!rrc>eZdZdZedZedZdS) _AllGatherz.s$FFFqu'**FFFrrUr rHrFrget_local_rankrrrr all_gatherrcat)rrrHr-last_dimr,rV tensor_lists ` rrz_AllGather.forwards% %%'' ??H5577Q;))++%%'' LLNNFFFFE*4E4EFFF  D  Qe4444y(333>>@@@rc|j}|}|dkr|dfS|}t||}||dfSNr.rHrFrrr)rrrHr-r,chunkss rrz_AllGather.backwardsio  %%'' ??$ $))++&{J??d|&&(($..rNrrwrrrrsQFFAA\A" / /\ / / /rrc>eZdZdZedZedZdS)_Splitz>Split forward, all-gather backward. Scatters replicated input.c||_|}|dkr|S|}t||}||Srr)rrrHr-r,rs rrz_Split.forwards^% %%'' ??H))++&q*55d|&&(((rc|j}|}|dkrdfSdz }|}|}fdt |D}||<tj||tj ||dfS)Nr.c8g|]}tjSrwrrxrrs rr{z#_Split.backward..0$PPPu' 44PPPrrUrrrrrHr-rr,rVrs ` rrz_Split.backward#o  %%'' ??$ $??$$q())++%%''!,,.. PPPPeJ>O>OPPP ' D  [>>>>y(333>>@@$FFrNrrwrrrrsSHH ) )\ )GG\GGGrrc>eZdZdZedZedZdS)_ReduceScatterzCReduce-scatter forward, all-gather backward. For sequence parallel.c||_|}|dkr|S|dz }|}t |||}t |j}||xx|zcc<tj||j |j }tj ||tj j||S)Nr.rrr)rHrFrrr|rrrrrrArreduce_scatterrr) rrrHr-rrV input_chunks output_shapeoutputs rrz_ReduceScatter.forward9s% %%'' ??H5577Q;%%''AGGJHG==>> AG}} X:-\JJJ FLT]5FeTTTT rc|j}|}|dkrdfSdz }|}|}fdt |D}||<tj||tj ||dfS)Nr.c8g|]}tjSrwrrs rr{z+_ReduceScatter.backward..YrrrUrrrs ` rrz_ReduceScatter.backwardLrrNrrwrrrr6sSMM\$GG\GGGrrc8t||S)zAIdentity forward, all-reduce backward. Use before colwise layers.)rapplyrrHs rall_reduce_backwardrds  # #A{ 3 33rc8t||S)z@All-reduce forward, identity backward. Use after rowwise layers.)rrrs rall_reduce_forwardris  " "1k 2 22rc8t||S)z#All-gather forward, split backward.)rrrs rrrns   A{ + ++rc8t||S)z#Split forward, all-gather backward.)rrrs rsplitrss <<; ' ''rc8t||S)z,Reduce-scatter forward, all-gather backward.)rrrs rrrxs   ; / //rmodule nn.Modulecx|fd|fd|S)z Copy pasted from torch's function but we remove the communications (partitioning) as well as buffer registering that is similarly not efficient. Nc||Srrw)modinputsrHinput_fns rrXz#distribute_module..sXXc6S^=_=_rc||Srrw)rroutputsrH output_fns rrXz#distribute_module..s))CQXZeBfBfr)register_forward_pre_hookregister_forward_hook)rrHrr s ```rdistribute_moduler#}sZ(()_)_)_)_)_```$$%f%f%f%f%fggg Mrc`eZdZdZdZdZdZddZdZdZ ddd Z dddZ ddZ ddZ ddZdS)TensorParallelLayerz.General tensor parallel layer for transformersNc0||_||_||_dSr)r,rHr)selfrHr,rs r__init__zTensorParallelLayer.__init__s &&rctrNotImplementedErrorr'rrrHs r_prepare_input_fnz%TensorParallelLayer._prepare_input_fn!!rctrr*r'rrrHs r_prepare_output_fnz&TensorParallelLayer._prepare_output_fnr.rrrrrrRctrr*r'rrrArs r shard_tensorz TensorParallelLayer.shard_tensors "!rrr layer_namerQcdS)zHRaise if the module cannot be sharded with this style on the given mesh.Nrw)r'rrHr6s rvalidate_modulez#TensorParallelLayer.validate_modules rc >t|||j|jdSr)r#r-r1r'rrHkwargss rprepare_module_tpz%TensorParallelLayer.prepare_module_tps0    "  #      r full_shapetuple[int, ...] | torch.Sizetuple[int, ...]c t|S)z Compute the expected shape after TP sharding for a given full shape. Args: full_shape: The full (unsharded) parameter shape Returns: The expected sharded shape for this rank )r)r'r=s rget_expected_sharded_shapez.TensorParallelLayer.get_expected_sharded_shapesZ   rcdS)z Update module attributes (e.g. in_features, out_features) to reflect sharded dimensions. Args: module: The module to update Returns: None, update the module in-place Nrwr'rs rupdate_module_attributesz,TensorParallelLayer.update_module_attributess  rNNNrrrrrRrr5rrr6rQrrrRrr=r>rRr?rr)rrrrrHr,rr(r-r1r4r8r<rArDrwrrr%r%s88K DK'''' """"""VZ"""""           ! ! ! !       rr%cXeZdZdZddfd Zddd Zd Zd Z dddZd dZ d!dZ xZ S)"ColwiseParallelz Column-wise parallel: weight is sharded on dim -2 (output features). Forward: input replicated -> output sharded on last dim. If gather_output=True, output is all-gathered to produce full tensor. F gather_outputboolc Htjdi|||_dSNrw)superr(rN)r'rNr; __class__s rr(zColwiseParallel.__init__s.""6"""*rr5rrr6rQc t|dd}|jr]|]||zdkrDtd|dt |jd|d|d dSdSdS)N out_featuresr`z` (z with out_features=zo) is sharded with 'colwise_gather_output', which requires out_features to be divisible by the number of ranks (z) to all-gather equal-size shards. Resize the weight (e.g. `model.resize_token_embeddings` for LM heads) or override this module's entry in the tp_plan.)r7rNrFr1r5r)r'rrHr6rUs rr8zColwiseParallel.validate_modulesv~t<<   ,":|kN^N^N`N`?`de?e?eqJqq4<<#8qq\qq$$&&qqq   ":":?e?erc:|r|dn|}t||SNrrr'rrrH input_tensors rr-z!ColwiseParallel._prepare_input_fns$$*6vayy "<===rc4|jrt||S|Sr)rNrr0s rr1z"ColwiseParallel._prepare_output_fns"   4g{33 3rNrrrrrRcdt|tjr|n t |}|dkr#t ||j|j|j d}n"t ||j|j|j d}| ||SNr.rrrAr rrrrrrrrrHr,rr'rrrArr parameters rr4zColwiseParallel.shard_tensors(u|<<Xeiikkk#eooFWFWBXBX !88(0@$BRTXT]_abbII(0@$BRTXT]_abbI||6|777rr=r>r?cb|j}t|}t|dkrdnd}|dkrt||zn|}t j|||z }|j|z}t||z||}||z ||<t|S)Nr.rrr) rHrFr|rrrr,rrr'r=r-rrrrrs rrAz*ColwiseParallel.get_expected_sharded_shapes%**,, Z  JJ!OObb"%''c%jj3sYuSzJ677  J&%*$eCj115[c U||rc|js8t|dr*||jfd|_dSdSdS)NrUr)rNrrArUrCs rrDz(ColwiseParallel.update_module_attributessb! ]gfn&E&E ]"&"A"A6CVBX"Y"YZ["\F    ] ] ] ]rF)rNrOrGrHrErFrJrK) rrrrr(r8r-r1r4rArD __classcell__rSs@rrMrMs +++++++>>> VZ 8 8 8 8 8    ]]]]]]]]rrMc,eZdZdZdZdZddZdZdS)ReplicatedWithGradAllReduceaM Replicated parameter with gradient all-reduce. For parameters like q_norm/k_norm that sit between colwise and rowwise layers. The parameter is replicated (not sharded), but its gradient accumulates from local heads only in TP mode. This class registers a backward hook to all-reduce the parameter gradient. c|Srrwr,s rr-z-ReplicatedWithGradAllReduce._prepare_input_fn rc|Srrwr0s rr1z.ReplicatedWithGradAllReduce._prepare_output_fnrNc<|d||SN.r_rr3s rr4z(ReplicatedWithGradAllReduce.shard_tensorSz}}F%}888rc :|fd}||dS)Ncl|D]}|jt|j|dSr) parametersgradr)r grad_inputrmeshrs r_backward_hookzEReplicatedWithGradAllReduce.prepare_module_tp.._backward_hooks@)) 9 9:)&uz4888 9 9r)register_full_backward_hook)r'rrHr;rys rr<z-ReplicatedWithGradAllReduce.prepare_module_tps8?J 9 9 9 9 **>:::::rrErrrrr-r1r4r<rwrrrjrjs_9999;;;;;rrjc,eZdZdZdZdZddZdZdS)AllReduceParallelzAll-reduce a module's forward output across the TP mesh. Use as a declarative sync point at the boundary of a multi-arg module whose compute ends in a partial sum (e.g. the lightning indexer's score sum before its top-k). c|Srrwr,s rr-z#AllReduceParallel._prepare_input_fn+rlrc"t||Srrr0s rr1z$AllReduceParallel._prepare_output_fn.s!';777rNc<|d||Srprqr3s rr4zAllReduceParallel.shard_tensor1rrrc 4t|||jdSN)r )r#r1r:s rr<z#AllReduceParallel.prepare_module_tp4s &+9PQQQQQQrrEr{rwrrr}r}%sd 8889999RRRRRrr}c(eZdZdZdZddZddZdS)MlaKvAProjParallela# For MLA attention used in DeepSeek-V2 style models (deepseek_v2, longcat_flash, glm_moe_dsa, glm4_moe_lite): kv_a_proj_with_mqa output is [kv_lora_rank + qk_rope_head_dim] (can have different naming but important thing to understand is that it is split) Example below (from modeling_longcat_flash.py): kv_a_proj_with_mqa | split / k_pass k_rot <-- "bypasses kv_b_proj" | | (goes straight to attention, kv_a_layernorm | never touches kv_b_proj) | | kv_b_proj | (colwise) | | | k_pass k_rot \ / cat | key_states k_pass is passed to kv_b_proj (colwise) which has built-in all_reduce_backward so we don't have a partial gradient for it. However, k_rot goes straight to attention, never touches kv_b_proj. So we need to average gradient across all ranks otherwise we only get gradient for one rank (partial gradient). c2t|jds%tdt|jd|jj}||jd|z |gd\}}t||}tj ||gdS)Nqk_rope_head_dimz Config for z does not have `qk_rope_head_dim`. MlaKvAProjParallel requires `qk_rope_head_dim` to be defined in the model config. Please add it to the model's config or update the TP plan mapping.rr) rconfigAttributeErrorr5rrrrrrr)r'rr rHrope_dim pass_output rope_outputs rr1z%MlaKvAProjParallel._prepare_output_fnTssz#566  Ud3ii0UUU  :.#)<<b1AH1Lh0W]_<#`#` [)+{CC y+{3<<< output partial -> all-reduce to replicate. Args: split_input: If True, splits replicated input before matmul. Use when input comes from a non-parallelizable operation (chunk/slice). Default False (expects pre-sharded input from colwise layer). F split_inputrOc Htjdi|||_dSrQ)rRr(r)r'rr;rSs rr(zRowwiseParallel.__init__ss.""6"""&rct|dr|j|j|_d|_|r|dn|}|jrt ||S|S)Nbiasr)rr_biasrrrZs rr-z!RowwiseParallel._prepare_input_fnws^ 3   CH$8CICH$*6vayy   4{33 3rcht||}t|dr|j ||jz}|S)Nr)rrrr0s rr1z"RowwiseParallel._prepare_output_fns;$Wk:: 3  *SY%: )GrNrrrrrRc0t|tjr|n t |}|dkr |d}n"t ||j|j|j d}| ||S)Nr..rr_r`ras rr4zRowwiseParallel.shard_tensors(u|<<Xeiikkk#eooFWFWBXBX !88c II(0@$BRTXT]_abbI||6|777rr=r>r?c|t|dkrt|S|j}t |}d}|dkrt||zn|}t j|||z }|j|z}t||z||}||z ||<t|SNr.rr) rrrHrFr|rrr,rrds rrAz*RowwiseParallel.get_expected_sharded_shapes z??a  $$ $%**,, Z  "%''c%jj3sYuSzJ677  J&%*$eCj115[c U||rrrc|t|dr+d|jf}||d|_dSdS)N in_featuresr.)rrrA)r'rrs rrDz(RowwiseParallel.update_module_attributessQ 6= ) ) K*+E!%!@!@!G!G!JF     K Krrf)rrOrErFrJrK rrrrr(r-r1r4rArDrgrhs@rrrhs'''''''   VZ 8 8 8 8 8    KKKKKKKKrrceZdZdZ d d dZdS) PackedColwiseParallelz@Packed column-wise parallel for fused weights like gate_up_proj.NrrrrrRct|tjr|n t |}|dkr#t ||j|j|j d}nw| |jj }|t |kr#t ||j|j|j d}n"t||j|j|j d}| ||Sr^)rrrrrrrrrHr,rArrr)r'rrrArrrbexpected_shapes rr4z"PackedColwiseParallel.shard_tensors(u|<<Xeiikkk#eooFWFWBXBX !88(0@$BRTXT]_abbII!<r?cl|j}t|}t|dkrdn|j}|dkrt||zn|}t j|||z }|j|z}t||z||}||z ||<t|Sr) rHrFr|rrrrr,rrrds rrAz,EmbeddingParallel.get_expected_sharded_shapes%**,, Z  JJ!OObb)D"%''c%jj3sYuSzJ677  J&%*$eCj115[c U||rrrct|dr1|jdkr&||jfd|_t|dr3|jdkr*||jfd|_dSdSdS)Nnum_embeddingsr embedding_dimr.)rrrArrrCs rrDz*EmbeddingParallel.update_module_attributes$s 6+ , , a1LPQ1Q1Q$($C$CVEZD\$]$]^_$`F ! 6? + + _0Kq0P0P#'#B#BFDXCZ#[#[\]#^F  _ _0P0Pr)rr9rErFrJrKrrhs@rrrsbb89========4 8 8 8VZ88888"    ________rrc>eZdZdZddfd Zd Zd Z dddZxZS)SequenceParallelzd Sequence Parallel: input/output sharded on sequence dimension. Weights are replicated. r.F sequence_dimr9use_local_outputrOc Htjdi|||_dSrQ)rRr(r)r'rr use_dtensorr;rSs rr(zSequenceParallel.__init__1s.""6"""(rc:|r|dn|}t||SrX)rrZs rr-z"SequenceParallel._prepare_input_fn5s&$*6vayy , 444rc"t||Sr)rr0s rr1z#SequenceParallel._prepare_output_fn;sg{333rNrrrrrRc<|d||Srprqr3s rr4zSequenceParallel.shard_tensor>Sz}}F%}888r)r.FF)rr9rrOrErF rrrrr(r-r1r4rgrhs@rrr+s )))))))555 444VZ999999999rrc>eZdZdZfdZ ddd Zdd ZddZxZS)GroupedGemmParallelzb Applies Expert Parallelism to MoE experts by loading the correct experts on each device. c :tjdi|dSrQrRr(r'r;rSs rr(zGroupedGemmParallel.__init__I&""6"""""rNrrrrrRc|jjd}||jzdkr-t d|d|jd||jz}|}|j|z}|jdz|z} t |tjs| n|j} |.||cxkr| kr!nn|dd |S||||  ||St| dkr|dS|dd ||S)NrzAGlobal number of experts must be divisible by number of devices:  %  != 0r.)rAr_) rrrHrFr1r,rrrrrr) r'rrrArglobal_num_expertslocal_num_expertsrrrrs rr4z GroupedGemmParallel.shard_tensorLs"-3A6  0 5 5 7 7 71 < <JTfJJkok{lAlAlClCJJJ /$2B2G2G2I2II&  J&y1} *)3E5<)H)HY!!!ek  !ez&?&?&?&?C&?&?&?&?&?8;;f;-- -  s#&&fE&BB B ZZ1__!748;;fE;:: :rr=r>r?c|j}t|}|d|z}||d<t|SrX)rHrFr|r)r'r=r-rrs rrAz.GroupedGemmParallel.get_expected_sharded_shapedsG%**,, Z  !!H 2$aU||rrrct|dr3||jjdfd|_dSdS)N num_expertsr)rrArrrrCs rrDz,GroupedGemmParallel.update_module_attributeslsR 6= ) ) b!%!@!@$BRBXYZB[A]!^!^_`!aF    b brrErFrJrK) rrrrr(r4rArDrgrhs@rrrDs#####VZ;;;;;0bbbbbbbbrrc:eZdZdZfdZdZdZ d d d ZxZS)RouterParallelzQ Allows to reshape the router scores to support running expert parallel. c :tjdi|dSrQrrs rr(zRouterParallel.__init__vrrc|Srrwr,s rr-z RouterParallel._prepare_input_fnyrlrc||}}t|dd}| tt|dddd}|%tdt |jd||zdkrt d|d|d ||z}|^}} } } | |z|k} | | d } | | d } |d krtj | |} n2| | dkd| dkd } | | d k|} || | g| RS) u  Remap global expert indices to local and zero out non-local scores. Example: 4 tokens, top_k=4, 128 experts, EP=8. num_local_experts = 128/8 = 16. Router produces (all ranks see the same values): router_scores: (4, 4) — top-k routing weights router_indices: (4, 4) — global expert IDs [ 52, 42, 119, 67], [102, 89, 61, 40], [ 82, 103, 4, 34], [ 93, 23, 109, 11], Each index maps to a rank: index // 16 gives the owning rank. [3, 2, 7, 4], [6, 5, 3, 2], [5, 6, 0, 2], [5, 1, 6, 0], For rank 0 (owns experts 0-15), we remap local indices with fmod and fill non-local with sentinel=16 (used for one_hot masking): router_indices (rank 0): [ 16, 16, 16, 16], [ 16, 16, 16, 16], [ 16, 16, 4, 16], [ 16, 16, 16, 11], Scores for non-local experts are zeroed out via masked_fill: router_scores (rank 0): [0.0, 0.0, 0.0, 0.0], [0.0, 0.0, 0.0, 0.0], [0.0, 0.0, 0.3, 0.0], ← only expert 4 (local) keeps its score [0.0, 0.0, 0.0, 0.1], ← only expert 11 (local) keeps its score both router_scores and router_indices stay (seq, top_k) shape. They are paired element-wise: scores[i] is the weight for indices[i]. All expert forward implementations (grouped_mm, batched_mm, eager) flatten both with reshape(-1) and rely on this pairing. Changing the shape of one without the other breaks routing! Each rank believes it is alone and computes only its part of the hidden states. The sentinel index (num_local_experts) is skipped by one_hot encoding or clamped + masked in grouped_mm/batched_mm. After the expert forward, an all_reduce sums partial outputs across EP ranks to produce the full result. rNrzRouter module z. is missing num_experts and config.num_expertsrz>The number of experts must be divisible by number of ep_size: rrgrr.) rrFr7rr5rr1 masked_fillrfmod) r'rrrHep_rankep_sizernum_local_experts router_logits router_scoresrouter_indices extra_outputsnon_local_masks rr1z!RouterParallel._prepare_output_fn|s\'55779I9I9K9Kc=$77  !'#x">"> tTTK   !t$s))2D!t!t!tuu u  A % %oQ\ooahooo (72GND }n}(,=='I%11.#FF '33NBGG q "Z8IJJNN+778JANNZZ[ilm[moqrrN'33Nb4HJ[\\m^KmKKKrNrrrrrRc<|d||Srprqr3s rr4zRouterParallel.shard_tensorrrrErFrrhs@rrrqs#####ELELELPVZ999999999rrceZdZdZdZdS)RouterParallelMegaMoeuRouter TP plan used with DeepGEMM Mega MoE. Mega MoE handles EP dispatch inside the kernel and wants raw global expert ids with unmasked routing weights, so the router doesn't pre-shard per EP rank like `RouterParallel._prepare_output_fn` does. The quantizer's `update_tp_plan` swaps `"ep_router"` → `"megamoe_router"` when `experts_implementation == "deepgemm_megamoe"`. c|Srrwr0s rr1z(RouterParallelMegaMoe._prepare_output_fnrnrN)rrrrr1rwrrrrs-rrc:eZdZdZfdZdZdZ d d d ZxZS)MoeTensorParalellExpertsa7 Note: For tensor parallel, the MoEExpertsParallel TP layer handles gradient sync: - all_reduce_backward on hidden_states (for colwise gate_up_proj gradient) - all_reduce_backward on top_k_weights (for router gradient) - all_reduce_forward on output (for partial expert outputs) c :tjdi|dSrQrrs rr(z!MoeTensorParalellExperts.__init__rrc||d}|d}|d}t||}t||}|||fSNrr.rrYr'rrrH hidden_states top_k_index top_k_weightss rr-z*MoeTensorParalellExperts._prepare_input_fnsLq Qi q ,M;GG ,M;GG k=88rc"t||Srrr0s rr1z+MoeTensorParalellExperts._prepare_output_fns!';777rNrrrrrRc<|d||Srprqr3s rr4z%MoeTensorParalellExperts.shard_tensors Sz}}F%}888rrErFrrhs@rrrs#####999 888 VZ999999999rrc*eZdZdZdZdZ d d d ZdS) MoeTensorParalellMegaMoeExpertsuITP layer for DeepGEMM Mega MoE experts. Mega MoE is inference-only (the kernel has no backward) and handles EP dispatch + combine + per-rank token sharding internally — so we skip the gradient-sync hooks that the regular `MoeTensorParalellExperts` would apply, and we forward the EP `process_group` into the module so the symm-buffer rendezvous can run on first forward. The quantizer's `update_tp_plan` swaps the experts plan key from `"moe_tp_experts"` to `"megamoe_experts"` when `from_pretrained(..., experts_implementation="deepgemm_megamoe")`. cb|d|d|d}}}||||fSr)rrs rr-z1MoeTensorParalellMegaMoeExperts._prepare_input_fn s74:1Ivay&QR)M{ k=+:O:O:Q:QQQrc|Srrwr0s rr1z2MoeTensorParalellMegaMoeExperts._prepare_output_fn srNrrrrrRc<|d||Srprqr3s rr4z,MoeTensorParalellMegaMoeExperts.shard_tensorrrrErF)rrrrr-r1r4rwrrrrs_  RRR VZ9999999rrc&eZdZdZdZddZdZdS)MoeIdentityExpertParallelaM TP class for zero/identity experts in MoE layers. Under TP, the parent MoeTensorParalellExperts does all_reduce_forward (sum) on the expert module output. Identity experts produce the same output on every rank, so the sum gives world_size * output. This class divides the input by world_size to compensate. cH|r|dn|}||z SrX)rFrZs rr-z+MoeIdentityExpertParallel._prepare_input_fn!s+$*6vayy k..0000rNc<|d||Srprqr3s rr4z&MoeIdentityExpertParallel.shard_tensor&rrrc 4t|||jdS)N)r)r#r-r:s rr<z+MoeIdentityExpertParallel.prepare_module_tp)s &+8NOOOOOOrrE)rrrrr-r4r<rwrrrrsU111 9999PPPPPrrc NeZdZUererideddeddedd ed ed ed d ede de de de de dedededededeiniZddddddddddddd Zded<ddddddddddddd Zded<ed&d$Zed&d%ZdS)'ParallelInterfaceembedding_rowwiserrembedding_colwiser.colwise_gather_outputT)rNcolwiserowwiserowwise_split_input)rpacked_colwisepacked_rowwisesequence_parallel grouped_gemm ep_routermegamoe_routermoe_tp_expertsmegamoe_expertsmoe_identity_expertreplicated_with_grad_allreduce mla_kv_a_projrrrN) rrrrrrrrrrrrzdict[str, int | None]plan_to_weight_dimplan_to_bias_dimkeyrQvaluerc||j|<dSr)rclsrrs rregister_plan_to_weight_dimz-ParallelInterface.register_plan_to_weight_dimks&+s###rc||j|<dSr)rrs rregister_plan_to_bias_dimz+ParallelInterface.register_plan_to_bias_dimos$)S!!!r)rrQrr)rrrr _torch_distributed_availablerrMrrrrrrrrrrrjrr}_global_mappingr__annotations__r classmethodrrrwrrrr-s0    ) (%A)  !2!2!!L!L!L !2!2!!L!L!L  $__4%H%H%H  ((  ((  "??t#D#D#D   3355  3355  !1!1!3!3  //11  ))  3355  6688  >>@@  "#<#<#>#> -.I.I.K.K! " //11# $ ++--%   *-:!#!!*. 1 1    "!##!!!*. / /    ,,,[,***[***rrALL_PARALLEL_STYLES local_tensor shard_dimrHdist.device_mesh.DeviceMeshcD|}d|jpivr|dnd}|dkr j|z}fdt |D}t j||tj ||S)a~ All-gather a sharded tensor along the specified dimension to reconstruct the full tensor. Args: local_tensor: The local shard of the tensor on this rank shard_dim: The dimension along which the tensor was sharded device_mesh: The device mesh for distributed communication Returns: The full reconstructed tensor (same on all ranks) r/Nrc8g|]}tjSrwr)rxrr s rr{z&gather_full_tensor..s$RRR1(66RRRrrUr) rFrErrDrrrrrr)r r!rHr- process_groupgathered_tensorss` rgather_full_tensorr'|s!!##J37Kt}||| |||jn:#t$r-}t d|dd|Yd}~nd}~wwxYw_|_fd_ dSdS)a This function is called in `PretrainedModel.post_init()`. It is responsible of adding hooks to the modules of the `model`, based on the `PretrainedModel._tp_plan`. This is the place where we add the `pre_forward` and `post_forwards` hooks. These are defined for each `TensorParallelLayer` as `_prepare_input_fn` and `_prepare_output_fn`. Args: model (`PretrainedModel`): The model containing the modules. module (`nn.Module`): The current module to which we want to add the hooks. current_module_plan (`str` or `None`): The tensor parallel plan for the current module, if any. layer_name (`str`): The qualified name of the current module. device_mesh (`dist.device_mesh.DeviceMesh`): The device mesh for distributed communication. N)rTrying to prepare 0, but it's not supported. Corresponding module: z Fix it's TP plan: c6dS)Nz TP Plan: )__repr__)current_module_planrsrrXz5add_tensor_parallel_hooks_to_module..s V__%6%6"X"XCV"X"Xr) rr8r<rr+loggerwarning _hf_tp_plan_hf_device_meshr:)modelrr;r6rHtp_layerrLs `` r#add_tensor_parallel_hooks_to_modulerBs,&&':;  jAAA   & &v{5< & P P P P"    NNZag          1!,XXXXX'&sA A?#A::A?c d|vr|ddn|\}} |jpi} ||} t|}t || } t jdkrA| td|dn td|d| d} | t| } || _ || _ || _ | |d||}|r|}nO#t$r%}t!d |d | d | d |Yd}~n%d}~wwxYw|dd|}t%|t&jjs3t&j|| }t/| | || | | |S)a This function is called in `from_pretrained` when loading a model's checkpoints. It receives the pointer to the parameter (or the parameter itself) and takes care of "sharding". All process run this function, so they just load the partition of the tensor that they require. Main uses cases: - column / rowise parallelism, you just shard all the weights of the layer (weight and bias) - packed layers: you slice the weights, then shard like above - custom operation: - you want to add an all-gather at the end of a local layer. - you want to have a layer that is isolated from the rest of the world (because torch.DTensor does not work well with `.view` for instance) r`r.rNzTensor sharding plan for z+ not found, using default 'replicate' plan.z: )rrrAr7r8z" Fix it's TP plan, current layer: z : ) requires_grad)rar get_submoduler9rerget_rankr<inforrrHr,r4rr+printrrrr Parameteris_floating_pointsetattrrD)r@rrr\param_casting_dtype is_contiguousr,rHr/r0r module_to_tpcurrent_shard_planrArLs rshard_and_distribute_modulerPsO ?B^>S>S^223:::YgJ m!rG&&z22L t99D/HH }!  % KKkJkkk l l l l KKVJVVBTVV W W WH% *+=>H#.H #.H  HM))%DH[dh)iiE +((**"    f^ffeqffV^ffcdff          aaa /00 eUX/ 0 0Y""5 8U8U8W8W"XX L*e,,,)),777 Ls:AD D;D66D; expected_keys list[str]dict[str, str] | Nonec|dSd|D}t|}|}|D]}d|vr|dddn|}tjdd|}||vr,||d||jd|vrK|dddx}|vr+||d||t|dkrt d|t|dkr2t d d |dSdS) z Verify the TP plan of the model, log a warning if the layers that were not sharded and the rules that were not applied. Nc,h|]}t|Srw)r[)rxrs r z!verify_tp_plan..Fs!SSSc4S99SSSrr`r.rr+r,z>The following TP rules were not applied on any of the layers: z'The following layers were not sharded: z, ) setcopyrarYrZpopdiscardrr<r=join) rQr generic_keysunsharded_layers unused_rulesrr/rcr3s rverify_tp_planr_>s SS]SSSL<((<<>>L * *.1SjjSZZQ''**c VFC<<  ( (   / 6 6 6  $ $S ) ) ) ) & & &ASAZAZ[^`aAbAbcdAe,e,=jq+q+q   . 5 5 5  $ $S ) ) ) <1fXdffggg q  ^K[A\A\^^_____! rc ||_||_|5t|trt j|}||j_t|tr||_|j}|tr| D]%}|tvrtd|dt&| D]B\}}t|dds%t||d} t!||| ||d|_C|S)z,Distribute a model according to the TP plan.Nz"Unsupported tensor parallel style z. Supported styles are _is_hookedF)r\rrbT)_tp_size _device_meshrdictr from_dictrdistributed_configrrvaluesrr1 named_modulesr7rerBra) r@rrfrHr model_planvrPrplans rdistribute_modelrl[s<EN$E% ($ / / Q!2!<=O!P!P *< ''4    J">""$$ w wA+++ !ua!u!u`s!u!uvvv,!//11 % %LD&6<77 -T:afggg3 !%F   LrrE)rrrr)rPrQrRrQ)T)r\rQrr]rRr^)rqr9rrrsrRrt)r) rrrr9r-r9rr9rRrr)rrrI)r rr!r9rHr"rRr)r(r)rr]rr9rRr))rQrRrrS)U __future__rrrr:rY functoolsrrrutilsrr utils.genericr utils.import_utilsr rtorch.distributedrr is_availabler get_loggerrr<rrOr[rerOuint8int8int16rbfloat16int32float32float64r float8_e4m3fnrrrrrrautogradFunctionrrrrrrrrrrr#r%rMrjr}rrrrrrrrrrrrrrrr'r5rBrPr_rlrwrrrsv#""""""  ++++++66666666,,,,,,333333DLLL$$$$$$$)#4#A#A#C#C   H % %    dhE,E,E,E,E,PEEEE.  kj{}{}}{&  L&&&&4A4A4A4P > > > > > Be>e>e>e>e>P...4!!!!!0!!!$ ! ! ! ! !/ ! ! !/////(///DGGGGGU^ $GGGD&G&G&G&G&GU^,&G&G&G\444 333 ,,, ((( 000  ": : : : : : : : z9]9]9]9]9])9]9]9]x;;;;;"5;;;<RRRRR+RRR&-R-R-R-R-R,-R-R-R`?K?K?K?K?K)?K?K?KD88888O888,88888O8884N_N_N_N_N_+N_N_N_b99999*9992*b*b*b*b*b-*b*b*bZU9U9U9U9U9(U9U9U9p     N   $9$9$9$9$92$9$9$9N99999&99994PPPPP 3PPP,D*D*D*D*D*(D*D*D*N*;):)<)<<<<<6666