S^(hr ddlmZddlZddlmZmZddlmZmZm Z m Z ddl Z ddl m Z ddl mZmZe j gZej$eZe j*j-Zedrer dd lmZmZmZmZd#d Ze j<e j>e j@e jBe jDe jFe jHe jJe jLe jNe jPd Z)d Z*d Z+ d$ d%dZ,GddZ-Gdde-Z.Gdde-Z/Gdde-Z0Gdde0Z1Gdde-Z2Gdde2Z3Gdde-Z4hdZ5ed&d Z6d!Z7d"Z8y)') annotationsN) lru_cachepartial)ListOptionalTupleUnion)nn)is_torch_greater_or_equalloggingz2.5)DTensor Placement ReplicateShardct|tr8t|}||zdk(s Jd|d|||z}|Dcgc]}||z c}S||zdk(s Jd|||z}|g|zScc}w)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: zPrepacked is not divisible by ) isinstancelistsum) total_sizeblocks total_blocks part_sizeblock single_sizes g/var/www/skyplay_api_hub/venv/lib/python3.12/site-packages/transformers/integrations/tensor_parallel.py_blocks_to_block_sizesr&s&$6{ L(A-lzlJcdjck/ll-,. /56e E!66F"a'R+I&)RR' F* }v%% 7s A&) BOOLU8I8I16F16BF16I32F32F64I64F8_E4M3c|}|j|}|j}t|d}g} d} |D].} | |z} || z} |dz| z}| t| | z| |zz } | | z } 0|j }|dk(r"|dj t j}|dk(r|| df}n6|dk(s|dk(r |dd| df}n!|dk(s|d k(r|d| f}ntd |d |j 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 )rrrr(.NUnsupported dim ", only dim 0, 1 or 2 are supported) shapesizerrange get_dtypetotorchfloat16 ValueErrorstr_to_torch_dtype)param empty_param device_meshrankdimslice_r world_size block_sizestensors_slices block_offset block_sizeshard_block_sizestartstop slice_dtypetensors rget_packed_weightsrHOsA>F""3'J!!#J(JqIKNL!# %3''q,,% u 4lT6IJJ " #""$Ki . ax+, SBY>3./ SBY^+,+C50RSTT 99' 4 55c|dk(rE|jd}||||jzz|dz||jzzdf}|S|dk(s|dk(rH|jd}|d|||jzz|dz||jzzddf}|S|dk(s|dk(rE|jd}|d|||jzz|dz||jzzf}|Std|d) Nrr*.r+r r,r-r.)r/r0r6)r8r9r:r;r<size_s rget_tensor_shardrLsG ax!!!$de{'7'7'99:dQh5T_TdTdTfKf=ggillm L SBY!!"%c45K,<,<,>#>?4!8PUYdYiYiYkPkBllnoop L SBY!!"%c45K,<,<,>#>?4!8PUYdYiYiYkPkBlllm L+C50RSTTrIct|jdk(r.|jfd|jfd|S)z Copy pasted from torch's function but we remove the communications (partitionning) as well as buffer registering that is similarly not efficient. rc||SN)modinputsr:input_fns rz#distribute_module..s#vWbAcrIc||SrOrP)rQrRoutputsr: output_fns rrTz#distribute_module..siPSU\^iFjrI)len_forward_pre_hooksregister_forward_pre_hookregister_forward_hook)moduler:rSrWs ```rdistribute_moduler]sG 6 $ $%*    , ,-c d   ( ()j k MrIcBeZdZdZdZedZedZdZddZ y) TensorParallelLayer9 General tensor parallel layer for transformers. TcyrOrP input_layoutsdesired_input_layoutsrQrRr:s r_prepare_input_fnz%TensorParallelLayer._prepare_input_fns[^rIcyrOrPoutput_layoutsuse_local_outputrQrVr:s r_prepare_output_fnz&TensorParallelLayer._prepare_output_fnsY\rIctrO)NotImplementedError)selfr8r9 param_typeparam_casting_dtype to_contiguousr;r:s rpartition_tensorz$TensorParallelLayer.partition_tensors!!rIc |jrat||t|j|j|j t|j |j|jyyrO) use_dtensorr]rrercrdrjrhrirmr\r:s rprepare_module_tpz%TensorParallelLayer.prepare_module_tpsZ    ..0B0BDD^D^_//1D1DdF[F[\   rINr\ nn.Modulereturnrw) __name__ __module__ __qualname____doc__rs staticmethodrerjrqrurPrIrr_r_s4K^^\\"rIr_cXeZdZdZdddd dfdZedZedZxZS) GatherParallelzi Simple class used to define the hooks to add to a layer when we just want to gather the outputs NT)rcrhrict||xs tf|_||_tf|_||_yrO)super__init__rrcrhrdri)rmrcrhri __class__s rrzGatherParallel.__init__s> +:y{<,&/k^" 0rIcV|r&t|dtr|dj}|SNrrrto_localrbs rrez GatherParallel._prepare_input_fns) jG4AY'')F rIctjj|dtjjjd|S)NrF)opasync_op)r4 distributed all_reduceReduceOpSUMrgs rrjz!GatherParallel._prepare_output_fns; $$WQZE4E4E4N4N4R4R]b$crIrcOptional[Placement]rhrribool) ryrzr{r|rr}rerj __classcell__rs@rrrs^.2.2!% 1+ 1, 1  1 rIrc<eZdZdZeddZeddZddZy)IsolatedParallelz This class is used to isolate computation in a TP layer from the rest of the world. Parameters need to be LOCAL, so not dtensors NcP|d}t|tr|j}|SrrrcrdrQrRr: input_tensors rrez"IsolatedParallel._prepare_input_fns+ay lG ,'002LrIc|SrOrPrgs rrjz#IsolatedParallel._prepare_output_fns rIc tt||t|jddt|jddyrO)r]rrerjrts rruz"IsolatedParallel.prepare_module_tps4   D**D$ 7 D++T4 8  rIrOrv)ryrzr{r|r}rerjrurPrIrrrs4  rIrc`eZdZdZddddd d fdZedZdZedZxZ S) ColwiseParallelr`NTrcrhrirsct||xs tf|_|xs t df|_tf|_||_||_yNr,) rrrrcrrhrdrirsrmrcrhrirsrs rrzColwiseParallel.__init__ sR +:y{<-:r<&/k^" 0&rIc|d}t|tstj|||d}||k7r|j|d}|S)NrF run_check placementsrrr from_local redistributers rrez!ColwiseParallel._prepare_input_fnsVay ,0"--lKbghL 1 1'44@U`e4fLrIc4|dk(rt||||d}tdg} ntdg} t||||d}|j|}|r|j}|jrt j ||| d}tj|S)Nbiasr,r+Fr) rLrr3 contiguousrsrrr Parameter rmr8r9rnrorpr;r: parametershards rrqz ColwiseParallel.partition_tensor&s  ( [$PRSI2YKE2YKE( [$PRSILL!45 !,,.I   **9k5TYZI||I&&rIcn|j|k7r|j|d}|r|jS|S)NFr)rrrrgs rrjz"ColwiseParallel._prepare_output_fn8s=    /**nu*UG%5w!B7BrIr) ryrzr{r|rr}rerqrjrrs@rrrsh.2.2!% '+ ', '  '  '$CCrIrceZdZdZy)PackedColwiseParallelct||||d}|j|}|r|j}|jr#t j ||t dgd}tj|S)Nr+Fr rHr3rrsrrrr r rmr8r9rnrorpr;r:rs rrqz&PackedColwiseParallel.partition_tensorBk'uk;bQ LL!45 !,,.I   **9kE"I;Z_`I||I&&rINryrzr{rqrPrIrrrA 'rIrcheZdZdZddddd d fdZdZedZedZd d Z xZ S) RowwiseParallela Partition a compatible nn.Module in a row-wise fashion. Currently supports nn.Linear and nn.Embedding. Users can compose it with ColwiseParallel to achieve the sharding of more complicated modules. (i.e. MLP, Attention) Keyword Args: input_layouts (Placement, optional): The DTensor layout of input tensor for the nn.Module, this is used to annotate the input tensor to become a DTensor. If not specified, we assume the input tensor to be sharded on the last dimension. output_layouts (Placement, optional): The DTensor layout of the output for the nn.Module, this is used to ensure the output of the nn.Module with the user desired layout. If not specified, the output tensor is replicated. use_local_output (bool, optional): Whether to use local :class:`torch.Tensor` instead of :class:`DTensor` for the module output, default: True. Returns: A :class:`ParallelStyle` object that represents Rowwise sharding of the nn.Module. NTrct||xs tdf|_|xs t f|_||_||_yr)rrrrcrrhrirsrs rrzRowwiseParallel.__init__bsF +8uRy:-<> 0&rIc|dk7rt||||d}tdg} ntg} |dd}|j|}|r|j }|j rt j||| d}tj|S)Nrr,Fr) rLrrr3rrsrrr rrs rrqz RowwiseParallel.partition_tensorps  ( [$PRSI2YKE[MEaILL!45 !,,.I   **9k5TYZI||I&&rIct|dr$|j|j|_d|_|d}t|tst j |||d}||k7r|j |d}|S)NrrFrTr)hasattrr_biasrrrrrs rrez!RowwiseParallel._prepare_input_fnsw 3 CHH$8CICHay ,0"--lKbghL 1 1'44@U`d4eLrIc|j|k7r|j|d}t|dr||jz }|r|j S|S)NTrr)rrrrrrgs rrjz"RowwiseParallel._prepare_output_fnsU    /**nt*TG 3 syy G%5w!B7BrIc d|_|jrt|tjrt df|_nbt|tjrtf|_n7t|tjrt df|_n tdt||t|j|j|j t|j|j |j"yy)NTr,zBRowwiseParallel currently only support nn.Linear and nn.Embedding!)_distribute_module_appliedrsrr Linearrrd Embeddingrrrlr]rrercrjrhrirts rruz!RowwiseParallel.prepare_module_tps,0)   &")),EJ2YL*FBLL1.7k^*FBLL1.3Bi\*)*noo ..0B0BDD^D^_//1D1DdF[F[\   rIrrv) ryrzr{r|rrqr}rerjrurrs@rrrOsm*.2.2!% '+ ', '  ''$   C CrIrceZdZdZy)PackedRowwiseParallelct||||d}|j|}|r|j}|jr#t j ||t dgd}tj|S)Nr,Frrrs rrqz&PackedRowwiseParallel.partition_tensorrrINrrPrIrrrrrIrcTeZdZdZddddd fdZedZedZdZxZ S) SequenceParallelaa SequenceParallel replicates a compatible ``nn.Module`` parameters and runs the sharded computation with input sharded on the sequence dimension. This currently supports ``nn.LayerNorm``, ``nn.Dropout``, and the `RMSNorm python implementation `__ This style implements the operation that is described in the paper `Reducing Activation Recomputation in Large Transformer Models `__ If the input passed in to this ``nn.Module`` is a :class:`torch.Tensor`, it assumes that the input is already sharded on the sequence dimension and converts the input to a :class:`DTensor` sharded on the sequence dimension. If the input passed in to this ``nn.Module`` is already a :class:`DTensor` but is not sharded on the sequence dimension, it would redistribute the input to be sharded on the sequence dimension. The output of the ``nn.Module`` will be sharded on the sequence dimension. Keyword Args: sequence_dim (int, optional): The sequence dimension of the input tensor for the ``nn.Module``, this is used to annotate the input tensor to become a DTensor that is sharded on the sequence dimension, default: 1. use_local_output (bool, optional): Whether to use local :class:`torch.Tensor` instead of :class:`DTensor` for the module output, default: False. Returns: A :class:`ParallelStyle` object that represents Sequence Parallel of the ``nn.Module``. Example:: >>> # xdoctest: +SKIP(failing) >>> from torch.distributed.tensor.parallel import parallelize_module, SequenceParallel >>> from torch.distributed.device_mesh import init_device_mesh >>> ... >>> m = Model(...) # m is a nn.Module that contains a "norm" nn.LayerNorm submodule >>> tp_mesh = init_device_mesh("cuda", (8,)) >>> >>> # By default, the input of the "norm" will be converted to DTensor that shards on the sequence dim >>> # and the output of "norm" will return a sharded on sequence dimension :class:`DTensor`. >>> >>> sharded_mod = parallelize_module(m, tp_mesh, {"norm": SequenceParallel()}), >>> ... .. note:: SequenceParallel style assumes ones initialization if there are weights in the nn.Module (i.e. ``nn.LayerNorm`` or ``RMSNorm``, and they by default have ones initialization). If you have custom inits for the weights on those modules, you need to broadcast the weights before/after parallelizing to ensure that they are replicated. r*F) sequence_dimrirsct|tf|_t df|_tf|_||_d|_t |f|_ ||_y)Nr*T) rrrrcrrdrhrirssequence_sharding)rmrrirsrs rrzSequenceParallel.__init__s\ 'k^&+Ah["({n 0"' "5!7 0rIc|d}t|tstj|||d}||k7r|j|d}|S)NrFrTrrrs rrez"SequenceParallel._prepare_input_fnsRay ,0"--lKbghL 1 1'44@U`d4eLrIcZ|jtfd}|jS)NTr)rrrrgs rrjz#SequenceParallel._prepare_output_fns1&&! ~' !!rIc|dd}|j|}|r|j}|jr"tj||t gd}t j|S)NFr)r3rrsrrrr rrs rrqz!SequenceParallel.partition_tensors`!H LL!45 !,,.I   **9kIK=\abI||I&&rI)rintrir) ryrzr{r|rr}rerjrqrrs@rrrsE*X/0%]b1"" 'rIr> localgathercolwiserowwise colwise_rep rowwise_rep local_colwise local_rowwisesequence_parallellocal_packed_rowwisect|tstdt|d|dk(r t S|dk(r t S|dk(rt t S|dk(rt t S|d k(r t d S|d k(r t d S|d k(r tS|dk(r tS|dk(r td S|dk(r tStd|)z In model configurations, we use a neutral type (string) to specify parallel styles, here we translate them into torch.distributed tensor-parallel types. z Unsupported parallel style type z, expected strrrr)rhr)rcrF)rsrrrrrz"Unsupported parallel style value: ) rstrr6typerrrrrrr)styles r!translate_to_torch_parallel_styler#s eS !;DK=WXX   )   - ik:: - Y[99 / !511 / !511 ' !! (  ( ($77 % %!!=eWEFFrIc |t|} |j||d|vrp|j ddd}t j dd |} |j| d x} r/t| }|j|} |j| |yyy#t$r}td|d|d|Yd}~d}~wwxYw) a Add hooks to the module holding the layer. Meaning: ``` class MyModel(nn.Module): def __init__(self): self.layer = nn.Linear(10, 10) ``` has state_dict like: ``` { "layer.weight": torch.Tensor, "layer.bias": torch.Tensor } ``` we add hooks to `MyModel` as well as `layer` to make sure that the tensors are correctly sharded and gathered. NTrying to prepare 0, but it's not supported. Corresponding module: z Fix it's TP plan: .r*r\d+*F) rrurlprintrsplitresubget get_submodule) modelr\tp_plan layer_namecurrent_module_planr:tp_layereparrent_layer_name generic_name module_plan module_to_tp_s r#add_tensor_parallel_hooks_to_modulerEs&&45HI   & &v{ ; j'..sA6q9vvfc+=> !++lE: :; :8EH!//0BCM  & &}k B ;  #  $ZL0`ag`hh{|}{~   sB B=B88B=c d|vr|jddn|\}} |j} |j|} d} t|}t j dd|} | | vr| | } n3d| vr/| jddd| vr| | jddd} t | ddst|| | || |d | _| $ t| }|j||| ||||}n&|dj|}|r|j}t|t j"j$st j"j%|}t'| | ||S#t$r!}td |d | d d |Yd}~wd}~wwxYw)a 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) rr*Nrrr _is_hookedFTrrz" Fix it's TP plan, current layer: z : .)r_tp_planrrrrgetattrrrrrqrlrr3rrr4r rsetattr)rr8r9parameter_namero is_contiguousr;r: param_namernr module_to_tprgeneric_param_namerrs rshard_and_distribute_modulerls?B^>S^223:YgJ nnG&&z2L t9D^<W$%&89 " "'9'@'@a'H'Kw'V%&8&?&?Q&G&JK <u 5+E<*Vikvw"& & 89LMH--{J0C]TXZeEc 12 $$&E eUXX// 0""5) L*e, L##  $^$44deqdrsUV^U__bcdbef   s"E F%FF)rrrzUnion[int, List[int]]rxz List[int])NNNrv)rr)9 __future__rr functoolsrrtypingrrrr r4r utilsr r LayerNormALL_LAYERNORM_LAYERS get_loggerryloggerr is_available_torch_distributed_availabletorch.distributed.tensorrrrrrruint8int8int16r5bfloat16int32float32float64int64 float8_e4m3fnr7rHrLr]r_rrrrrrrSUPPORTED_TP_STYLESrrrrPrIrrs# (// 6 ~   H % %00==?U#(DMM&6 JJ ++ ** ;; == NN ;; == == ;;"" ;6| "     $8(> * 89C)9Cx 'O 'a)aH 'O 'Q'*Q'h  G GB$CN4rI