-`i-mdZddlmZmZddlmZddlZddlZddl m Z m Z m Z m Z mZddlmZeeZeGddZeejejefZd ejd ed ejd ejd eeeeefeeeefff dZd ed ejd ejdeejdeejdejjdzde d efdZdeejdeejdejdejded ejded dfdZ d&dejdejdeeejdeejde de dedejjdzdeeefdzd efd Z! d'dejdejdeeejde de deeefdzd dfd!Z"dejdeeefd"ed ejfd#Z#dejdeeefd ejfd$Z$gd%Z%dS)(zh The actual execution of the rearrangement. This involves the exchange of expert weights between GPUs. )IterableSequence) dataclassN)P2POp ProcessGroup all_gatherbatch_isend_irecvget_global_rank) init_loggerc`eZdZUdZejed< eed< ejed< ejed<dS) RecvMetadatazz3get_ep_ranks_with_experts_batch..s0!!!0G0GA0G0G0Gr)sizeruniquelenarangeint32isinanylexsort concatenatewherediffrangersorttolistsetget)rrrr ranks_to_send_mapranks_to_recv_mapunique_experts num_positionsposition_indicesold_relevant_masknew_relevant_maskold_relevant_positionsold_relevant_expertsold_relevant_ranks sort_ordersorted_experts sorted_ranksexpert_boundariesistartendexpert expert_ranks_ unique_idx unique_ranksnew_relevant_positionsnew_relevant_expertsnew_relevant_ranksrecv_ranks_actualr+s @rget_ep_ranks_with_experts_batchrW-s(/1.0! "333Yz**N $$Mybh??? ^<< ^<< v  >!12C!D*+<=37HHZ!79M NOO -j9)*5 NS"(27>22a788;a?#nBUBUAV W   s,--122 > >A*1-/@Q/G3E.//F'c 2LIlFFFMAz' (;(;22a788;a?#nBUBUAV W   s,--122 : :A*1-/@Q/G3E.//F'c 2LIlFFFMAz' (;(;?@A|j|jksJ}tj|ftj}tj|fdtj} tj|fdtj} tj|fdtj} tj|fdtj} ||z} tj|tj}| |z}||}||}||k}|dk}tj ||d}tj |tj ||}d}|dk}tj |rWtj ||d\}}||}||}t|jd}|| d|<|| d|<d}tj ||}tj |r^||}||}tj |d\} }!||!}"t| jd}| | d|<|"| d|<d||"<tj ||}#t|# r|dkrtj|#d}$t%t'| d|| d|}%|$D]_>|>}&|%|&d}'|'dkr9t'||D](\}(})|)>|(|'d )`g}*}+fd t/|+D},|dkrJ| d|}-| d|}.tj|-d }/|-|/}-|.|/}.t3|-|||\}0}1t'|-|.D]\}&@|0|&}2|1|&}3|2r|3st5|3t5|2z}4|2|}5|5|4z}6|6|4z}7|3|6|7}8t5|2|4z}9|9|5z}:|:t5|3kr|8|3|:|8D]>|,>?|*?@fd |Dz }*|dkr| d|}-| d|};tj|-d }/|-|/}-|;|/};t3|-|||\}0}1t'|-|;D]\}&>|0|&}2|1|&}3|2r|3st5|3t5|2z}4|3|}:t5|2|4z}9|:|9kr |2|:|4z@n |2|:|9z @|,@A|*>Afd|Dz }*|*rc|at:j|5tA|*}<|z"move_to_buffer..s%WWWdOHd;;WWWrstablekindc\g|](}ttjj|)Sr)rtorch distributedisend)r)w dst_globalsrcs rr,z"move_to_buffer.. sH  )/#"rc\g|](}ttjj|)Sr)rrhriirecv)r)bdst src_globals rr,z"move_to_buffer..EsH  %+cFr)rrrr)#shaperbrzerosbool_fullint64r1r0r2 logical_or logical_andr3r.rboolnonzeror:dictzipr<copy_r-r8argsortrWr/indexappendrhcudastreamr waitr )Brrr rXrYrZr[ep_rankrsend_expert_ids send_src_rowsrrbase local_rows local_globalold_local_expert_idsnew_local_expert_ids is_unchanged new_validcan_recv_localis_received_locally send_count valid_old uniq_experts first_idx filtered_rowssrc_rowsrneed_recv_maskdesired_experts desired_dstsuniq_recv_experts uniq_indicesdst_rowseligible_local_buffer_mask dest_indicesexpert_to_src_maprN src_localrkrpp2p_opsep_sizerank_to_globalexpertssrcsordersend_maprecv_map ranks_to_send ranks_to_recvnum_dst_per_sender sender_pos recv_beginrecv_end recv_ranksremainder_start recver_posdstsreqsreqrqrlrmrrsB ` @@@@rmove_to_bufferrs<   1 1 1 1 1mmooG"3!5RXFFFg02BbhGGGOG.0"BHEEEMg02BbhGGGOG.0"BHEEEM & &D,BH===J*$L&|4&|4(+??L%*IW2%N-bnY?? J$*I vi."$)  +$# # #  i#9-  ++A.// '3  $%- kzk"J^%8$8)DDN vn+.~>!.1 *,)ORV*W*W*W'< -*0344 '8  $%- kzk"&*(#!# ?R!S!S & * * , ,-- B*q..z"<==a@GGII    ,mKZK.H I I    B BC)#.F)--fb99IB0FGGBBDAqcFLL9DLAAAAGmmooGWWWWgWWWNA~~!+:+.[j[) 7222%.E{<       (w~~//??  KFC$V,M$V,M    !$]!3!3s=7I7I!I &,,W55J#&88J!$66H&z(':;J!-003EEO(:5JC ....!!- ";<<<!  +C0  ,  A~~!+:+.[j[) 7222%.E{<       (w~~//??  KFC$V,M$V,M    !$]!3!3s=7I7I!I &,,W55J!-003EEOO++#J2D$DE#J$@A',J  0  GG;* Z  { + +  $W--D                     ))  C HHJJJJ /!+'      s)Y  YYrr recv_metadatarc |j}|j}|j} |j} |jd} t j||} t j|| } t| rjt j | d }|D];}t||D](\}}|| ||d)<|dkrdS|| z}||t j| t jz}t jt j||t j||dk}t| sdSt j |d}||}| d|}| d|}t j|d}||}||}t j||}t j||jdk|t j||jdd z |k}t| sdS||}|||}t| | D]-\}} |D]%}|| || d&.dS) a Copies expert weights from communication buffers back to the target weight tensors after EPLB rebalancing. Args: expert_weights: List of the actual MoE layer weights used in the execution. expert_weights_buffers: Intermediate buffers containing the experts weights after the transfer is completed. is_unchanged: (num_local_experts,), True where an expert row is unchanged. is_received_locally: (num_local_experts,), True where a row is updated locally. recv_metadata: RecvMetadata containing remote receive metadata. new_indices: (num_experts_total,) mapping from local rows to desired (possibly global) expert id, after rebalance. ep_rank: Rank of the process in the expert parallel group. rTr_Nr#r]rdrer%)rrrrrsrrxryrzr3r{r:r}r~r0r1r searchsortedminimum)!rXrYrrrr rrrrrr copy_mask dest_mask_nprrqrkrpr local_expertsduplicate_mask dup_dst_rows dup_experts prim_experts prim_dstsrprim_experts_sortedprim_dsts_sortedposvalidmatched_dst_rowsmatched_src_rowsrms! rmove_from_bufferres0&7)J#3O!/M$*1- 13DEEI><-;;L L    8z,//299;;  8 8CN,BCC 8 81# QsV$ 7777 8Q & &Dry1B"('S'S'S STM^  }':&:;; ))=B+>??N ""$$ % %:n--a0L -K";J;/Lkzk*I J|( 3 3 3E&u- ' /-{ ; ;C N !'**BJs,?,Ea,H1,LMMN    E   #E*'E 3(//113C3J3J3L3LMM44S 4 4A cFLL3dL 3 3 3 3 444rFold_global_expert_indicesnew_global_expert_indicesexpert_weights_buffer is_profilelayer rank_mappingc K|} |Yt||krt||}n#t|||}|jd|jdksJ|j\} } t|| ksJt t |djd} |j| | fksJ| | | zksJ|} |}t| | ||||||||\}}}|||fS)a Rearranges the expert weights in place according to the new expert indices. The value of the indices arguments are logical indices of the experts, while keys are physical. Args: old_global_expert_indices: Shape (num_moe_layers, num_physical_experts). new_global_expert_indices: Shape (num_moe_layers, num_physical_experts). expert_weights: A sequence of shape (num_moe_layers)(weight_count) of tensors of shape (num_local_physical_experts, hidden_size_i). For example, a linear layer may have up and down projection, so weight_count = 2. Each weight's hidden size can be different. ep_group: The device process group for expert parallelism. is_profile (bool): If `True`, do not perform any actual weight copy. This is used during profile run, where we only perform dummy communications to reserve enough memory for the buffers. Returns: is_unchanged (np.ndarray): (1, num_local_experts), True where expert is left unchanged. is_received_locally (np.ndarray): (1, num_local_experts), True where expert can be received locally. RecvMetadata: Metadata needed for completing remote weight transfers. Nr%rrrr rXrYrZr[) r-r/)_map_new_expert_indices_with_rank_mapping)_map_old_expert_indices_with_rank_mappingrsnextitercpunumpyr)rrrXrr[rrrZrrnum_moe_layersnum_physical_expertsnum_local_physical_expertsold_global_expert_indices_npnew_global_expert_indices_nprrrs rtransfer_layerrsHmmooG |    / /(Q))) % % )R) )) % % *1 -1J1PQR1S S S S S+D+J(N( ~  . 0 0 0 0!%d>!+<&=&=!>!>!DQ!G $ *~?S.T T T T T 7-G#G G G G G#<#@#@#B#B#H#H#J#J #<#@#@#B#B#H#H#J#J 7E40707%e,48884L%} ,m ;;rc |Yt||krt||}n#t|||}|jd|jdksJ|j\}}t||ksJt t |djd}|j||fksJ|} || |zksJt|d} d| D} |rit|d| D]P\} fdt| D} tj t| | |QdStj|}|}t|D]b}t%|||||||| d|\}}}t'||| ||||||cdS) a Rearranges the expert weights in place according to the new expert indices. The value of the indices arguments are logical indices of the experts, while keys are physical. Args: old_global_expert_indices: Shape (num_moe_layers, num_physical_experts). new_global_expert_indices: Shape (num_moe_layers, num_physical_experts). expert_weights: A sequence of shape (num_moe_layers)(weight_count) of tensors of shape (num_local_physical_experts, hidden_size_i). For example, a linear layer may have up and down projection, so weight_count = 2. Each weight's hidden size can be different. ep_group: The device process group for expert parallelism. is_profile (bool): If `True`, do not perform any actual weight copy. This is used during profile run, where we only perform dummy communications to reserve enough memory for the buffers. rank_mapping: A dictionary mapping old rank to new rank. Nr%rc6g|]}tj|Sr)rh empty_like)r)rks rr,z4rearrange_expert_weights_inplace..7s0*** !***rcg|]}Srr)r)rPbuffers rr,z4rearrange_expert_weights_inplace..=s @ @ @A @ @ @r)groupr)rXrYrrrr r)r/r-rrrsrrlistr}r8rhribarrierrr synchronizerrrrrb)rrrXr[rrrrrrfirst_layer_weightsweights_bufferweightdummy_recv_bufferold_global_expert_indices_cpunew_global_expert_indices_cpu layer_idxrrrrs @r rearrange_expert_weights_inplacers6 |    / /(Q))) % % )R) )) % % *1 -1J1PQR1S S S S S+D+J(N( ~  . 0 0 0 0!%d>!+<&=&=!>!>!DQ!G $ *~?S.T T T T TmmooG 7-G#G G G G G~a011**%8***N !."3^DD  NFF @ @ @ @w @ @ @    % % ' ' ' !       J$=$A$A$C$C$I$I$K$K!$=$A$A$C$C$I$I$K$K!>**  ;I85i@5i@))4#1< < < 8 )= ))4#1% 3'5i@MMOO       r new_ep_sizecx|j\}}|s Jdt|}||z}||z}tj||fd|j|j}t |D]V} || } | =| dkr7| |kr1| |z} | dz|z} | |z} | dz|z}|dd| | f|dd| |f<W|S)a Map the old global expert indices to the new global expert indices. Args: old_global_expert_indices: Shape (num_layers, old_ep_size * num_local_physical_experts). rank_mapping: Mapping from old rank to new rank. new_ep_size: New expert parallelism size. Returns: Mapped expert indices with shape (num_layers, new_ep_size * num_local_physical_experts). Rank mapping is requiredr] fill_valuer$deviceNrr%)rsr/rhrvr$rr8r<)rrr num_layersold_num_physical_experts old_ep_sizernew_num_physical_expertsmapped_expert_indicesold_ranknew_rank old_start_idx old_end_idx new_start_idx new_end_idxs rrrcs/$,E+J(J( 33333<l##K!9[!H*-GG"J -.'-(/ +&&  ##H--  HMMh6L6L$'AAM#a<+EEK$'AAM#a<+EEK*!!!];-F*FG "!!!];%>"> ? ! rc|j\}}|s Jdt|}td|D}||z}||z}t j||fd|j|j}t|D]G} || } | dkr7| |kr1| |z} | dz|z} | |z} | dz|z}|dd| |f|dd| | f<H|S)Nrc3"K|] }|dkV dS)r]Nr)r)rs r z<_map_new_expert_indices_with_rank_mapping..s&KKh"nKKKKKKrr]rrr%) rsr/sumvaluesrhrvr$rr8)rrrrrrrrrrrrrrrs rrrsG,E+J(J( 33333<l##KKK\5H5H5J5JKKKKKK!9[!H*-GG!J -.'-(/ +&&  ) q==X 33$'AAM#a<+EEK$'AAM#a<+EEK*!!!];-F*FG "!!!];%>"> ? ! r)rrr )FrNN)FN)&rcollections.abcrr dataclassesrrrrhtorch.distributedrrrr r vllm.loggerr rloggerr tuplerMoveToBufferResultrr|rrWTensorrStreamrrrzrrrr__all__rrrrsq /.......!!!!!! $##### X    O O O O O O O  O2:rz<?@h0 h0h0h0 h0  4T#Y c49n!5 56 h0h0h0h0VJJJJU\* J %U\2 J "T) JJJJJJZL4U\*L4 .L4*L4 L4  L4  L4L4 L4L4L4L4j,0*.G<G<$|G<$|G<Xel34G<$EL1 G<  G<  G< G<"T)G<sCx.4'G<G<G<G<G<^*. b b $|b $|b Xel34b  b  b sCx.4' b  b b b b J2!$|2!sCx.2!2! \ 2!2!2!2!j !$| !sCx. ! \ ! ! ! !F A @ @r