)`icdZddlmZddlmZddlmZddlZddlZddl m Z dd l m Z m Z dd lmZdd lmZdd lmZeGd dZejdZe dejdededefdZe dejdeedededejdejf dZe dejdeejdejdejdedededed efd!Ze d1d#ejd$edejdejdedededed%ed&edejfd'Ze d(ejdejdejded)ef d*Z e deded+ed,efd-Z!Gd.d/Z"gd0Z#dS)2z MoE All-to-All Operations (Throughput Backend) This module provides the throughput-optimized all-to-all backend for MoE expert parallelism, supporting multiple payloads per collective operation. ) dataclass)SimpleNamespace)OptionalN)flashinfer_api) MnnvlMemory MnnvlConfig)Mapping)gen_moe_alltoall_module)register_custom_opcVeZdZUdZdZeed<dZee ed<dZ ee ed<dS) _A2AStatez3Internal state tracking for MoeAlltoAll operations.idlephaseNlocal_num_tokenscombine_payload_offset) __name__ __module__ __qualname____doc__rstr__annotations__rrintrw/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/flashinfer/comm/trtllm_moe_alltoall.pyrrsR==E3&*hsm***,0HSM00000rrc ttdddtjdt dt dt ffd }td dd tjd t tjdtjd tjd t dt dt dt dt ffd }tdd d#dtjdt dtjd tjd t dt dt dt dt dtdtjffd }tdddtjdtjd tjdt dt f fd }tdgfd}td gdt dt ffd! }t||||||"S)$z$Get or build the MOE A2A JIT module.zflashinfer::moe_a2a_initialize) workspace) mutates_argsrep_rankep_sizemax_num_tokensc4||||SN)moe_a2a_initialize)rr!r"r#modules rr&z3get_moe_alltoall_module..moe_a2a_initialize%s((GWnUUUrzflashinfer::moe_a2a_dispatchtoken_selected_expertsinput_payloadsmetainforuntime_max_tokens_per_ranktop_k num_expertsc > ||||||||| S)aW Dispatch tokens and payloads to expert ranks. Args: token_selected_experts: [local_num_tokens, top_k] int32 tensor input_payloads: List of [local_num_tokens, *] tensors to dispatch workspace: [ep_size, size_per_rank] workspace tensor metainfo: Metadata tensor from initialize runtime_max_tokens_per_rank: Max tokens per rank in this batch ep_rank: Current expert parallel rank ep_size: Total expert parallel size top_k: Number of experts per token num_experts: Total number of experts Returns: recv_offsets: List of offsets for each payload in the workspace recv_sizes: List of sizes for each payload in the workspace combine_payload_offset: Offset for combine payload region )moe_a2a_dispatch) r(r)rr*r+r!r"r,r-r's rr/z1get_moe_alltoall_module..moe_a2a_dispatch1s9D&& "    '       rzflashinfer::moe_a2a_combineFpayloadrrpayload_in_workspacereturnc @ |||||||||| S)a Combine expert outputs back to originating tokens. Args: payload: [ep_size, max_tokens, elements_per_token] tensor local_num_tokens: Number of tokens on this rank workspace: [ep_size, size_per_rank] workspace tensor metainfo: Metadata tensor from initialize runtime_max_tokens_per_rank: Max tokens per rank in this batch ep_rank: Current expert parallel rank ep_size: Total expert parallel size top_k: Number of experts per token combine_payload_offset: Offset from dispatch payload_in_workspace: If True, payload is workspace-backed Returns: output: [local_num_tokens, elements_per_token] tensor )moe_a2a_combine) r0rrr*r+r!r"r,rr1r's rr4z0get_moe_alltoall_module..moe_a2a_combine_s<D%%     '    "    rz'flashinfer::moe_a2a_sanitize_expert_ids) expert_idsr5invalid_expert_idc6|||||Sr%)moe_a2a_sanitize_expert_ids)r5rr*r!r6r's rr8z.moe_a2a_sanitize_expert_idss)11  8W6G   rz,flashinfer::moe_a2a_get_metainfo_index_pairsc,S)z Get all metainfo index constants from C++. Returns: Tuple of (names, values) where names is a list of constant names and values is a list of their corresponding integer values ) moe_a2a_get_metainfo_index_pairs)r'srr:zAget_moe_alltoall_module..moe_a2a_get_metainfo_index_pairss66888rz%flashinfer::moe_a2a_get_aux_data_sizec0||S)a1 Get the auxilary datasize per rank for the MoeAlltoAll operation. Args: ep_size: Total expert parallel size max_num_tokens: Maximum number of tokens across all ranks Returns: aux_data_size: Size of the auxilary data per rank in bytes )moe_a2a_get_aux_data_size)r"r#r's rr<z:get_moe_alltoall_module..moe_a2a_get_aux_data_sizes$//HHHr)r&r/r4r8r:r<F) r build_and_loadr torchTensorrlistboolr)r&r/r4r8r:r<r's @rget_moe_alltoall_modulerC s)% & & 5 5 7 7F(#V<VVV VVVVV V&#( % ( U\*( <( , ( &) (  ( ( ( ( ( ( ( ( ( T%#&+) ) ) ) <) , ) &) )  ) ) ) !$) #)  ) ) ) ) ) ) V1$  L  <  ,            69999 9/IIIIIII I -)'$?)I";    rrr!r"r#cJt||||Sr%)rCr&)rr!r"r#s rr&r&s+ # $ $ 7 77G^  r leading_shape slice_start slice_enddtyper2c|jdkr|d}|tj}|jdks Jd||z |jdks Jd||dz}||dz}||z }||z} | |jdks Jd|||| f|jg|dR} | S) a Wrap an offset in the workspace into a tensor. Args: workspace: [ep_size, size_per_rank] or [size_per_rank] workspace tensor leading_shape: The leading shape to wrap the tensor with slice_start: The start of the slice in the workspace slice_end: The end of the slice in the workspace dtype: Data type for the output tensor Returns: tensor: [leading_shape, *] workspace-backed tensor rrrHrz0workspace must be shape [ep_size, size_per_rank]z4slice_end - slice_start must belong to a single rankz2slice must fall within the workspace size per rank)ndim unsqueezeviewr?uint8shapestride) rrErFrGrHworkspace_base slice_ranklocal_slice_start slice_lengthlocal_slice_endresults r(moe_a2a_wrap_payload_tensor_in_workspacerXs=*~''** ^^%+^66N >Q    R    { "n&:1&= = = => > = = 5 5a 8 88J#n&;&;A&>&>>{*L',6O n215 5 5 5< 6 5 5 z#4_#DDE E    " "  " " " Mrr(r)r*r+r,r-c t||||||||| \} } } g} t|| | dD]6\} }}| t |||g|||z| j7| | fS)a Dispatch tokens and payloads to expert ranks. Args: token_selected_experts: [local_num_tokens, top_k] int32 tensor input_payloads: List of [local_num_tokens, *] tensors to dispatch workspace: [ep_size, size_per_rank] workspace tensor metainfo: Metadata tensor from initialize runtime_max_tokens_per_rank: Max tokens per rank in this batch ep_rank: Current expert parallel rank ep_size: Total expert parallel size top_k: Number of experts per token num_experts: Total number of experts Returns: output_payloads: List of payloads for this rank, backed by data in the workspace combine_payload_offset: The offset to place the combine payload in the workspace Tstrict)rCr/zipappendrXrH)r(r)rr*r+r!r"r,r- recv_offsets recv_sizesroutput_payloads input_payloadoffsetsizes rr/r/s> !!22 "    '     5L*4O'* j(((   # vt  456 #        2 22rFr0rrr1c Vt|||||||||| Sr%)rCr4) r0rrr*r+r!r"r,rr1s rr4r4>s@ # $ $ 4 4#    rr5r6cLt|||||Sr%)rCr8)r5rr*r!r6s rr8r8Ys. # $ $ @ @Ix2C  r%total_dispatch_payload_size_per_tokencombine_payload_size_per_tokenct||}d}||d|||z|zdz|||z|zdzS)a Get the workspace size per rank for the MoeAlltoAll operation. Args: ep_size: Total expert parallel size max_num_tokens: Maximum number of tokens across all ranks total_dispatch_payload_size_per_token: The size of the payload per token in the dispatch phase. This should be the sum of all payloads. combine_payload_size_per_token: The size of the payload per token in the combine phase. Returns: workspace_size_per_rank: Size of the workspace per rank in bytes c||zdz |z|zS)Nrr)xys rpad_upz3moe_a2a_get_workspace_size_per_rank..pad_up~sQq A%%r)rCr<)r"r#rfrg aux_data_sizerls r#moe_a2a_get_workspace_size_per_rankrofs&,--GGM &&& }c"" &>),QQSV W W X &>),JJC P P Qrc(eZdZUdZiZeeeeeefefed<e dedededede def d Z e e d"ded eded ed edef dZdZeeed<e dZ d#de ded ededed edeefdZdZe d$dejdeejdedeedeedeejf dZe d%dejdededejfdZe ded ed ejdejfd!ZdS)& MoeAlltoAlla  Manages MoE All-to-All operations with proper workspace allocation and synchronization. This class provides the throughput-optimized backend that supports multiple payloads per collective operation, explicit dispatch/combine phases, and workspace-backed tensors. Example: >>> moe_a2a = MoeAlltoAll(mapping, max_num_tokens=2048, top_k=2, num_experts=8) >>> recv = moe_a2a.dispatch(experts, [hidden, ids, scales], batch_size) >>> output = moe_a2a.combine(processed, batch_size) _WORKSPACE_CACHEworkspace_size_per_rankr!r"r#mappingr2c||||f}||jvr |j|St||}|tj}t ||||} ||||||| d|j|<|j|S)N)rsr#r!r" mnnvl_memrr*)rrr as_torch_strided_tensorr?rOr&) clsrsr!r"r#rtkeyrvrr*s r get_workspacezMoeAlltoAll.get_workspaces'.I #& & &', ,#G-DEEI!99%+FFI) H,C"0""&&$))C  %', ,rrr, hidden_sizeextra_payload_bytes_per_tokencd}t||z|dzz|dzz|z}t||z}t||||S)a Convenience wrapper to calculate the workspace size per rank for the MoeAlltoAll operation. Automatically calculates the size of the dispatch and combine payloads when using default values. This allocates space assuming 16-bit float, which may overallocate for quantized models. For a tighter bound, use the base function `moe_a2a_get_workspace_size_per_rank` directly. Args: ep_size: Total expert parallel size top_k: Number of experts per token max_num_tokens: Maximum number of tokens across all ranks hidden_size: Hidden dimension size extra_payload_bytes_per_token: Extra size per token in the payload Returns: workspace_size_per_rank: Size of the workspace per rank in bytes r)rro)r"r,r#r{r| element_sizerfrgs rget_moe_workspace_size_per_rankz+MoeAlltoAll.get_moe_workspace_size_per_ranksu.   l* + +ai ai , , .*-[<-G)H)H&2   1 *    rN_METAINFO_INDEXc(|jt}|\}}i|_t||dD]K\}}|dr|ddn|}t ||j|<JdSdS)z2Initialize constants from C++ if not already done.NTrZMOE_A2A_)rrCr:r\ startswithreplacer)rxr'namesvaluesnamevalue clean_names r_init_constantszMoeAlltoAll._init_constantss   &,..F"CCEEME6#%C "5&>>> = = ez22DLLR000 36e**#J// ' & = =rr- mnnvl_configc||)| Jd||j|||}tj|rtj||||_||_|j|_|j |_ ||_ ||_ t|j tr |j dkrtdt|j tr |j dkrtd|||j |j|j||_|jd|ks Jd|jd|jks Jd |jd |j ks Jd |jd |jks Jd |jd|_|jd|_|jd|_t+|_dS)a Initialize MoeAlltoAll with workspace allocation. Args: mapping: Mapping object containing rank information max_num_tokens: Maximum number of tokens supported top_k: Number of experts per token num_experts: Total number of experts workspace_size_per_rank: Size of workspace per rank in bytes, if None hidden_size must be provided hidden_size: Hidden dimension size used when calculating the workspace size, if workspace_size_per_rank is not provided mnnvl_config: Used to configure the communication backend for the MNNVL memory object NzGhidden_size must be provided if workspace_size_per_rank is not providedrztop_k must be a positive intz"num_experts must be a positive intrszWorkspace size mismatchr#zMax tokens mismatchr!zEP rank mismatchr"zEP size mismatchrvrr*)rr moe_ep_sizer initializeset_comm_from_configrsr#r" moe_ep_rankr!r,r- isinstancer ValueErrorrz _WORKSPACErvrr*r_state)selfrtr#r,r-rsr{rs r__init__zMoeAlltoAll.__init__s.  " ***Y+**'+&J&J#UNK'' #     D  ,Wl C C C'>$,* *  &$*c** =djAoo;<< <$*C00 CD4D4I4IABB B,, # L L      89=TTTT %UTT/0D4GGGG !HGGy)T\999;M999y)T\999;M99955 3 kk rctj|`|j|j|j|j|jf=d|j _ dS)zReset the workspace to free up its state. This is mainly used for testing. Use this with caution. This object is no longer usable after this.deletedN) r?cuda synchronizerrrrsr!r"r#rr)rs r_reset_workspacezMoeAlltoAll._reset_workspaceEsS    O  !,  #   & rr(r)r+invalid_token_expert_idexpert_id_payload_indexc |jjdks Jd||jks Jdt|||j|j||j|j|j|j \}}| d|j_ ||j_ d|j_|6| Jd||}t||j|j|j||S)aN Perform MoE all-to-all dispatch operation. Args: token_selected_experts: [local_num_tokens, top_k] expert indices input_payloads: List of [local_num_tokens, *] tensors to dispatch runtime_max_tokens_per_rank: Max tokens per rank in this batch invalid_token_expert_id: If set, sanitize invalid tokens to this ID expert_id_payload_index: Index of expert IDs in input_payloads (required if invalid_token_expert_id is set) Returns: recv_tensors: List of [ep_size, max_tokens, *] tensors rz%dispatch called twice without combine2runtime_max_tokens_per_rank exceeds max_num_tokensr dispatchedNzDexpert_id_payload_index required when invalid_token_expert_id is set)rrr#r/rr*r!r"r,r-rcrrr8) rr(r)r+rr recv_tensorsrrecv_expert_idss rdispatchzMoeAlltoAll.dispatchSs ,{ F***,S****d.AAAA @BAA0@ "  N M ' L L J   0 0 , ,(>'B'B1'E'E $-C *(  # .*66V766++BCO '  '    rFr0r1c |jjdks Jd||jks Jdt||jj|j|j||j|j|j |jj | }t|_|S)a Perform MoE all-to-all combine operation. Args: payload: [ep_size, max_tokens, elements_per_token] tensor runtime_max_tokens_per_rank: Max tokens per rank in this batch payload_in_workspace: If True, payload is workspace-backed (skip staging) Returns: output: [local_num_tokens, elements_per_token] tensor rz)combine called before successful dispatchr) rrr#r4rrr*r!r"r,rr)rr0r+r1outputs rcombinezMoeAlltoAll.combines${ L000 7100+d.AAAA @BAA!  K ( N M ' L L J K .     kk  rrHc6|jjdkrtdtjg|}t |j|jddf|j |g|jj |jj |j |z|z|zz|S)a Get combine payload tensor backed by workspace (zero-copy). This tensor can be written to directly by expert processing, avoiding a staging copy in the combine operation. Args: runtime_max_tokens_per_rank: Max tokens per rank in this batch hidden_size: Hidden dimension size dtype: Data type for the tensor Returns: tensor: [ep_size, max_tokens, hidden_size] workspace-backed tensor rzIget_combine_payload_tensor_in_workspace called before successful dispatchrJN) rr RuntimeErrorr?tensorrrXrr!r"r)rr+r{rHrs r'get_combine_payload_tensor_in_workspacez3MoeAlltoAll.get_combine_payload_tensor_in_workspaces* ;  , ,[ |Be444AACC 7 N4<? + \6 7 K . K .l88;FU V     r)r)NNN)NNr=)rrrrrrdicttuplerr classmethodr rz staticmethodrrrrrr rrr?r@rArrBrrHrrrrrqrqs  ?Ad5c3!34d:;@@@-!$-- -  -  - ---[-> ./ ' ' ' ' '  ' (+ ' ' ' ' ^\' V'+OXd^***==[=0(,.2G"G"G"G" G"  G" "% G"G"{+G"G"G"G"R & & & 2615 99 % 9U\*9&) 9 "*# 9 "*# 9 el 999^9v &+ (((&)(# (  (((^(T! %(! ! { !  ! ! ! ^! ! ! rrq)rqr4r/ror&r8rXr=)$r dataclassesrtypesrtypingrr? functools api_loggingrmnnvlr r rtr jit.commr utilsr rcacherCr@rr&rArHrXr/rBr4r8rorq__all__rrrrs"!!!!!!!!!!! ((((((++++++++......&&&&&& 1111111 1 eeeP|  '|'9'' ' ; '  \ ''''T:3!L:3&:3|:3l :3 "% :3  :3:3 :3:3:3:3:3z"' \|l  "%      \4   | l        ,/%( DS S S S S S S S l    r