o پioR@sddlmZddlmZddlmZmZmZmZddl Z ddl m m Z ddlZddlmZddlmZddlmZmZmZmZmZmZmZddlmZddlmZm Z er]dd l!m"Z"d e#fd d Z$d dZ%ddZ&ddZ'd=ddZ(dee j)effddZ*d=ddZ+d=ddZ,eGdddZ-d e#d!e#d"e.fd#d$Z/de j)fd%d&Z0d'e j)fd(d)Z1ej2d*ej3d!ej3d+ej3fd,d-Z4d.d/Z5d0eee j)ee j)ffd1d2Z6d>d3d4Z7de j)fd5d6Z8d7d8Z9d9d:Z:d;d<Z;dS)?) dataclass) accumulate) TYPE_CHECKINGListTupleUnionN)use_symmetric_memory) DpPaddingModeattn_cp_all_gather_into_tensorget_attention_cp_groupget_attention_cp_rankget_attention_cp_sizeget_attention_dp_rankis_allocation_symmetric)get_global_server_args) ceil_alignceil_div) ForwardBatchnsa_index_topkcCs |j|dS)N)max)clamp)original_seq_lensrrY/home/ubuntu/.local/lib/python3.10/site-packages/sglang/srt/layers/attention/nsa/utils.pycompute_nsa_seqlenss rcCstjSN)r#enable_nsa_prefill_context_parallelrrrris_nsa_enable_prefill_cp"srcCtotjdkS)Nz in-seq-splitrrnsa_prefill_cp_moderrrris_nsa_prefill_cp_in_seq_split& r!cCr)Nzround-robin-splitrrrrr#is_nsa_prefill_cp_round_robin_split-r"r# forward_batchrcCs4|jsdSt}t|j}to|dko|dkS)NFr) forward_modeis_context_parallel_extendr sumextend_seq_lens_cpur#)r$cp_sizeseq_lenrrr$can_nsa_prefill_cp_round_robin_split4s  r,input_cCst}t}t|ttfrt|t||}||St|}||dkrM|||||k}|dkr?|jdg|jddRSt j ||||j d}||S|j d|g|jddRdd|f S)aO # for round-robin-split, split the tokens evenly according to the rule of token_idx % cp_size. | +-----------before split------------+| | token0, token1, token2, token3, token4, token5, token6, token7, ... | | +--------------result-------------------+ | dp_atten_tp0: token0, token4, token8, token12, token16, ... | | dp_atten_tp1: token1, token5, token9, token13, token17, ... | | dp_atten_tp2: token2, token6, token10, token14, token18, ... | | dp_atten_tp3: token3, token7, token11, token15, token19, ... | | +-------------------------+ rr%N)device)r r isinstancetuplelistrangelen new_emptyshapetorcharanger.view contiguous)r-r*cp_rankindicestokenscur_lenrrrnsa_cp_round_robin_split_data<s  .r?cCs|j}t|}t}t|D] }t|||||<qt|j|}| r,t |}nt|dkr8|t }n|d}t |rEt ||}|SNr%r)global_num_tokens_cpucopyr4r r3rr get_dp_padding_modeis_extend_in_batch is_max_lenrrr,r)r$global_num_tokenssync_group_size attn_cp_sizeidp_padding_moder=rrrcal_padded_tokens[s      rKcCstt}|dko t|}|jdu}|s|s|St|}||jd}|dkr8t||j|g|jddRg}|Sr@)r r,rArKr6r7cat new_zeros)r$nsa_cache_seqlensrH needs_cp_pad needs_dp_padr=pad_lenrrrpad_nsa_cache_seqlensqs   rRc@seZdZUdZeeed<dZeeed<dZeeed<dZ eeed<dZ eeed<dZ eeed<dZ eed <dZ eed <dZeed <dZeed <dZejed <dZejed<dZejed<dZejed<dZejed<dS)NSAContextParallelMetadataN split_list max_rank_len zigzag_indexper_rank_actual_tokenreverse_split_lencp_reverse_indexr/ kv_len_prev kv_len_nextactual_seq_q_prevactual_seq_q_nextkv_len_prev_tensorkv_len_next_tensoractual_seq_q_prev_tensoractual_seq_q_next_tensortotal_seq_lens)__name__ __module__ __qualname__rTrint__annotations__rUrVrWrXrYrZr[r\r]r^r7Tensorr_r`rarbrrrrrSs     rSr+r*use_nsacCsjtr||}||dksJd|d|dn||d}|dkr3|dkr3|r3|jr3tr3dSdS) Nrzseq_len z is not divisible by cp_size z. when nsa_prefill_cp_mode is round-robin-splitr%TF)r#r&r'r)r+r*rir$cur_cp_seq_lenrrr can_cp_splits  rlcstrt}|jd|dksJd|jd|t|Sttj||jjddtj fdd|jj Ddd d|jd}|S)NrzAExpect input shape 0 can divided by cp size, but got input shape , cp size dimcg|]}|qSrr.0rI input_listrr z-cp_split_and_rebuild_data..r/) r#r r6r?r2r7splitnsa_cp_metadatarTrLrVr9)r$r-r*resultrrsrcp_split_and_rebuild_datasrz positionscsxtrt}|jd|dksJd|jd|t|Sttj||jjddtj fdd|jj Ddd}|S)NrzIExpect positions shape 0 can divided by cp size, but got positions shape rmr/rncrprrrqposition_id_listrrrurvz1cp_split_and_rebuild_position..) r#r r6r?r2r7rwrxrTrLrV)r$r{r*rr|rcp_split_and_rebuild_positions  r~r=r;c Csd}d}t|D]5}t||} | |7} | || ||k} | dkr7t|||t||| |d7}| | |}qdS)Nrr%)r3tlloadstore) in_seqs_ptr out_seqs_ptr bs_idx_ptrr=r*r; extra_seqbs_idxbsr>cur_seqrrr&nsa_cp_round_robin_split_q_seqs_kernels  rc Cst}t}d}g}t|D]\}}||7}||t|||k}|||||}qtddt|D}dd|D}||fS)NrcSsg|] \}}|dkr|qSrr)rrrIxrrrrusz7nsa_cp_round_robin_split_q_seqs_cpu..cSsg|]}|dkr|qSrr)rrq_lenrrrrus)r r enumeraterfappendr2) extend_seqsr*r;rq_seqsrr>rrrrr#nsa_cp_round_robin_split_q_seqs_cpus rreturnc Csvt}t}t|\}}tjt|f|j|jd}tjt|f|jtjd}d}t ||||t|||||||fS)a round-robin-split distributes tokens across ranks based on token_idx % cp_size. Return: ret_q_lens_cpu(List) and ret_q_lens(torch.Tensor): the partitioned length (excluding zeros) on the current cp rank for each sequence after distribution across cp ranks. bs_idx_cpu(List) and bs_idx(torch.Tensor): marks which sequences are ultimately selected, i.e., those with a partitioned length greater than zero. r.dtyper%) r r rr7emptyr4r.rint32r) extend_seqs_cpurr*r;ret_q_lens_cpu bs_idx_cpu ret_q_lensrgridrrrnsa_cp_round_robin_split_q_seqss   rcCs.|durt}|jdur|r|jrdSdS)NTF)rrxr&r')r$nsa_enable_prefill_cprrrnsa_use_prefill_cps rc s||d|}||jd}|dkr tj|ddd|fddd}ttt dtj|||jd|j|j d}Wdn1sDwYt |||t tj ||j jddtjfd d t|j jDdd}|S) a Allgather communication for context_parallel(kv_cache, index_k, hidden_states). This implementation mainly consists of three parts: Step 1, padding the input shape to unify the shape for allgather communication (the shape must be the same). Step 2, allgather communication(async). Step 3, removing the padding and reassembling the data according to the actual tokens. r%rconstant)modevaluedisabledrNrncs g|] \}}|d|qSrr)rrindex per_rank_lenoutputs_list_maxrrruDszBcp_attn_tp_all_gather_reorganazied_into_tensor..)r6Fpadrr rr7rr.rcp_all_gather_into_tensor_asyncr2rwrxrUrLrrW) r- total_len attn_tp_sizer$ stream_opmax_lenpad_sizeinput_tensor_alloutputsrrr.cp_attn_tp_all_gather_reorganazied_into_tensor#s:     rcstrNttt d||jd|g|jddR}Wdn1s*wYt|||j}|j|dg|ddRdd |}|S|j\}}t ||j j |||}t tj||j jddtjfdd|j jDdd}|d|}|S) a, # for in-seq-split | +-----------before allgather------------+| | | dp_atten_tp0: block0, block7 | | | dp_atten_tp1: block1, block6 | | | dp_atten_tp2: block2, block5 | | | dp_atten_tp3: block3, block4 | | | +----------before rerange---------------+| | block0 | block7 | block1 | block6 | block2 | block5 | block3 | block4 | | | +--------------result-------------------+ | block0 | block1 | block2 | block3 | block4 | block5 | block6 | block7 | | +-------------------------+ # for round-robin-split | +-----------before allgather------------+| | dp_atten_tp0: token0, token4, token8, token12, token16, ... | | dp_atten_tp1: token1, token5, token9, token13, token17, ... | | dp_atten_tp2: token2, token6, token10, token14, token18, ... | | dp_atten_tp3: token3, token7, token11, token15, token19, ... | | | +--------------result-------------------+ | token0, token1, token2, token3, token4, token5, token6, token7, ... | +-------------------------+ rrr%Nr/rncrprrrq outputs_listrrrurvz0cp_all_gather_rerange_output..)r#rr rr5r6r r9 transposereshaperrxrbr2r7rwrXrLrY) input_tensorr*r$stream output_tensor out_shape bs_seq_len hidden_sizerrrcp_all_gather_rerange_outputOsH    rcCs.d}g}i}tt|D]}i}g}||}|t|krr||} | |kr9|} | || <|| d| || <|d7}n9| |krT|} ||| <|| d||| <| |||<n|} | || <|| d| || <|| 8}|d7}|t|ks|D]\} } || d| } | | } || | | fqv||q |S)zUsed to obtain the index of the seq corresponding to each cp block in the forwardbatch, and the starting and ending positions of the corresponding seq in the cp blockrr%)r3r4getitemsr) cp_chunks_lenseqs_lenj tuple_len cumulativerI current_dictcurrent_tuplesc_vals_validxvalprev_cum current_cumrrrcalculate_cp_seq_idxsB    rcstrdS t|}d}|}d}||}||||}|dkr7ddd|Dd|<|d} | } tt|||||tt||d|||} tfddtD} fd dtD} g}t|D]!}|tt|||d|tt|d||dd |qtt }||}|d|d}|}d|d}t|j d tj d }t|j d tj d }t|j d tj d }t|j d tj d }t | | | | ||||||||||d }|S)NTr%rjrcSsg|]}|dqSrr)rrrrrrrurvz0prepare_input_dp_with_cp_dsa..c3s,|]}|d|dVqdS)rjr%Nrrqr*rTrr s z/prepare_input_dp_with_cp_dsa..cs2g|]}|d|dfD]}|qqS)rjr%r)rrrIelementrrrru scudar)rTrUrVrWrXrYrZr[r\r]r^r_r`rarb) r#r7tensorrepeat_interleaverftolistr2r3extendrtorrS)kv_lenr;r*rbs_per_cp_group kv_len_origincp_segment_num seq_per_batch remainderseq_max_rank_lenrUrVrWrXrYbatch_idprefix_sum_listrZr[r\r]r^r_r`rarxrrrprepare_input_dp_with_cp_dsas &       r)r$rr)< dataclassesr itertoolsrtypingrrrrr7torch.nn.functionalnn functionalrtritontriton.languagelanguagersZ    $         ,B 0