o -i-m@sDdZddlmZmZddlmZddlZddlZddl m Z m Z m Z m Z mZddlmZeeZeGdddZeejejefZd ejd ed ejd ejd eeeeefeeeefff ddZd ed ejd ejdeejdeejdejjdBde d efddZdeejdeejdejdejded ejded dfddZ    d-dejdejdeeejdeejde d e d!edejjdBd"eeefdBd efd#d$Z!  d.dejdejdeeejde d e d"eeefdBd dfd%d&Z"dejd"eeefd'ed ejfd(d)Z#dejd"eeefd ejfd*d+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@s>eZdZUdZejed< eed< ejed< ejed<dS) RecvMetadataz}||||d}}t ||}|||}tj|dd\}}|t|}t||gfdd|D}|||<q|D]}t |}||vr+g||<||vr4g||<q||fS)a Get the ranks of the experts that need to be exchanged. Args: expert_ids: 1D array of expert indices to query. num_local_experts: The number of local experts. old_indices: The old indices of the experts. new_indices: The new indices of the experts. Returns: A tuple of two dictionaries mapping expert_id to: - ranks_to_send: The ranks that have this expert and need to send. - ranks_to_recv: The ranks that need to receive this expert. rdtypeT return_indexcsg|] }|vrt|qSr)r).0rsend_ranks_setrr sz3get_ep_ranks_with_experts_batch..)sizeruniquelenarangeint32isinanylexsort concatenatewherediffrangersorttolistsetget)rrrrranks_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_actualrr&rget_ep_ranks_with_experts_batch-sn     (   (      rSexpert_weightsexpert_weights_buffers cuda_streamep_groupc>s2|j|jksJ}tj|ftjd}tj|fdtjd} tj|fdtjd} tj|fdtjd} tj|fdtjd} ||} tj|tjd}| |}||}||}||k}|dk}tj ||dd}t |t ||}d}|dk}t |rtj ||dd\}}||}||}t|jd}|| d|<|| d|<d}t ||}t |r||}||}tj |dd\} }!||!}"t| jd}| | d|<|"| d|<d||"<t ||}#t|# r0|dkr0t|#d}$tt| d|| d|}%|$D](|}&|%|&d}'|'dkr.t||D]\}(})|)j|(|'dd qqg}*}+fd d t|+D},|dkr| d|}-| d|}.tj|-d d }/|-|/}-|.|/}.t|-|||\}0}1t|-|.D]^\}&|0|&}2|1|&}3|2r|3sqst|3t|2}4|2|}5|5|4}6|6|4}7|3|6|7}8t|2|4}9|9|5}:|:t|3kr|8|3|:|8D]|,|*fdd|D7}*qqs|dkrO| d|}-| d|};tj|-d d }/|-|/}-|;|/};t|-|||\}0}1t|-|;D]J\}&|0|&}2|1|&}3|2r|3sqt|3t|2}4|3|}:t|2|4}9|:|9kr7|2|:|4n|2|:|9|,|*fdd|D7}*q|*r}|dur}tj|t |*}<|sz"move_to_buffer..stablekindcs g|] }ttjj|qSr)rtorch distributedisendr$w) dst_globalsrcrrr( z"move_to_buffer..cs g|] }ttjj|qSr)rrarbirecv)r$b)dst src_globalrrr(Erh)r r rr)#shaper\rzerosbool_fullint64r-r,r. logical_or logical_andr/r*rboolnonzeror6dictzipr8copy_r)r4argsortrSr+indexappendracudastreamrwaitr )>rrrrTrUrVrWep_rankr send_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_rowsr need_recv_maskdesired_experts desired_dstsuniq_recv_experts uniq_indicesdst_rowseligible_local_buffer_mask dest_indicesexpert_to_src_maprJ src_localrerjp2p_opsep_sizerank_to_globalexpertssrcsordersend_maprecv_map ranks_to_send ranks_to_recvnum_dst_per_sender sender_pos recv_beginrecv_end recv_ranksremainder_start recver_posdstsreqsreqr)rkrfrWrgrlrmove_to_buffers                             rrr recv_metadatarc! Cs|j}|j}|j} |j} |jd} t||} t|| } t| rGt | d }|D]}t ||D]\}}||j ||ddq6q/|dkrMdS|| }||tj| tjd}tt||t||dk}t| sxdSt |d}||}| d|}| d|}tj|dd}||}||}t||}t||jdk|t||jdd |k}t| sdS||}|||}t | | D]\}} |D] }||j || ddqqdS) 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. rTrZNrrXr^r_r!)r r rrrmrrrrsrtr/rur6rwrxr,r-ry searchsortedminimum)!rTrUrrrrrr r rrr copy_mask dest_mask_nprrkrerjr local_expertsduplicate_mask dup_dst_rows dup_experts prim_experts prim_dstsrprim_experts_sortedprim_dsts_sortedposvalidmatched_dst_rowsmatched_src_rowsrgrrrmove_from_bufferes\          rFold_global_expert_indicesnew_global_expert_indicesexpert_weights_buffer is_profilelayer rank_mappingc  s|} |durt||krt||}nt|||}|jd|jdks+J|j\} } t|| ks8Jtt|djd} |j| | fksLJ| | | ksTJ|} |}t | | ||||||||d\}}}|||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!rrrrrTrUrVrW) r)r+)_map_new_expert_indices_with_rank_mapping)_map_old_expert_indices_with_rank_mappingrmnextitercpunumpyr)rrrTrrWrrrVrrnum_moe_layersnum_physical_expertsnum_local_physical_expertsold_global_expert_indices_npnew_global_expert_indices_nprrrrrrtransfer_layers>$    rc s|durt||krt||}nt|||}|jd|jdks&J|j\}}t||ks3Jtt|djd}|j||fksGJ|} || |ksSJt|d} dd| D} |rt|d| D]\} fddt | D} t j t | | |dqidSt j|}|}t |D]'}t|||||||| d|d\}}}t||| ||||||d qdS) 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!rcSsg|]}t|qSr)ra empty_likerdrrrr(7s z4rearrange_expert_weights_inplace..csg|]}qSrr)r$rLbufferrrr(=s)groupr)rTrUrrrrr)r+r)rrrmrrlistrwr4rarbbarrierrr| synchronizerrrrr\)rrrTrWrrrrrrfirst_layer_weightsweights_bufferweightdummy_recv_bufferold_global_expert_indices_cpunew_global_expert_indices_cpu layer_idxrrrrrr rearrange_expert_weights_inplacesn         r new_ep_sizecCs|j\}}|s Jdt|}||}||}tj||fd|j|jd}t|D]9} || } | dura| dkra| |kra| |} | d|} | |} | d|}|dd| | f|dd| |f<q(|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 requiredrX fill_valuer deviceNrr!)rmr+rarpr rr4r8)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_idxrrrrcs,      rcCs|j\}}|s Jdt|}tdd|D}||}||}tj||fd|j|jd}t|D]4} || } | dkrg| |krg| |} | d|} | |} | d|}|dd| |f|dd| | f<q3|S)Nrcss|]}|dkVqdS)rXNr)r$rrrr sz<_map_new_expert_indices_with_rank_mapping..rXrrr!) rmr+sumvaluesrarpr rr4)rrrrrrrrrrrrrrrrrrrs.     r)rrr )FrNN)FN)&rcollections.abcrr dataclassesrrrratorch.distributedrrrrr vllm.loggerr rloggerr tuplerMoveToBufferResultrrvrrSTensorr|Streamrrrtrrrr__all__rrrrs  " k  N U    O  e  5  #