o پi@sddlZddlmZmZddlZddlmZddlm Z ddl m Z ddl m Z eeZdZdae rMz ddlmZeZWneyLedYnwGdddZea d#d ed edefddZ    d$dejdejdejded edeedededeejejffddZe gded   d%dejdejdejded edeedededeejejffdd Zd!d"ZdS)&N)OptionalTuple)$get_tensor_model_parallel_world_size)is_flashinfer_available)register_custom_opzIflashinfer.comm is not available, falling back to standard implementationc @s@eZdZddZ  ddedededed ef d d Zd d ZdS)FlashInferWorkspaceManagercCs"d|_d|_d|_d|_d|_dSNF)workspace_tensor ipc_handles world_sizerank initialized)selfr\/home/ubuntu/.local/lib/python3.10/site-packages/sglang/srt/layers/flashinfer_comm_fusion.py__init__s  z#FlashInferWorkspaceManager.__init__NFr r max_token_num hidden_dimuse_fp32_lamportcCs||jr |j|kr dStdurtddS|tj||||||d\|_|_ ||_||_ d|_t d|d|dS)zInitialize workspaceNz@FlashInfer comm not available, skipping workspace initialization)grouprTz*FlashInfer workspace initialized for rank z , world_size ) r r _flashinfer_commloggerwarningcleanupcomm1trtllm_create_ipc_workspace_for_all_reduce_fusionr r r info)rr r rrrrrrr initialize$s2   z%FlashInferWorkspaceManager.initializec Cs|jrR|jdurTz?z tj|jtjjdWnty/}zt d|WYd}~nd}~wwWd|_ d|_d|_dSWd|_ d|_d|_dSd|_ d|_d|_wdSdS)zClean up workspaceN)rz(Failed to cleanup FlashInfer workspace: F) r r r+trtllm_destroy_ipc_workspace_for_all_reducedistrWORLD Exceptionrrr )rerrrrMs*    z"FlashInferWorkspaceManager.cleanupr)__name__ __module__ __qualname__rintboolrrrrrrrs   )rFrrrcCsVtrtdur dSt}|dkrdSt}tjrtj|kr(tj|||||dtjS)zEnsure workspace is initializedNF)r r rrr) rrrrget_rank_workspace_managerr r r)rrrr r rrrensure_workspace_initialized_s" r-ư>@ input_tensorresidualweighteps use_oneshottrigger_completion_at_endfp32_accreturnc Cst|}t|} | |fSN)torch empty_like) r0r1r2r3rr4r5r6 residual_outnorm_outrrr*fake_flashinfer_allreduce_residual_rmsnorm{s r=)r0r1r2) mutates_args fake_implc Cs.trtdurtddSt}|dkrtddS|jd|ks%Jt||jd|jtj kds:td dS|j\} } t |} t |} tj d id |d |d t d | d| dtjddd|d|d|dtjjddd|d| d| ddddd|d|dddd| | fS)!a Use FlashInfer's fused allreduce + residual + RMS norm operation Args: input_tensor: Input tensor that needs allreduce residual: Residual tensor weight: RMS norm weight eps: RMS norm epsilon max_token_num: Maximum token number use_oneshot: Whether to use oneshot mode trigger_completion_at_end: Whether to trigger completion at end fp32_acc: Whether to use fp32 precision Returns: Tuple[torch.Tensor, torch.Tensor]: (norm_output, residual_output) NzAFlashInfer not available, falling back to standard implementation)NNr*z(Single GPU, no need for allreduce fusionr)rrrz"FlashInfer workspace not available allreduce_inr world_rank token_numrworkspace_ptrslaunch_with_pdlTr4r5r6 pattern_code allreduce_out residual_inr;r< quant_out scale_out rms_gammarms_eps scale_factor layout_coder)rrrdebugrshaper-dtyper9float32r:trtllm_allreduce_fusionrr+r,r AllReduceFusionPatternkARResidualRMSNorm) r0r1r2r3rr4r5r6r rCrr;r<rrr%flashinfer_allreduce_residual_rmsnorms            rVcCstdur tdSdSr8)r,rrrrrcleanup_flashinfer_workspaces rW)r(r)F)r.r/NFF)r.r(NFF) loggingtypingrrr9torch.distributed distributedrsglang.srt.distributedrsglang.srt.utilsrsglang.srt.utils.custom_opr getLoggerr#rrr,flashinfer.commr ImportErrorrrr&r'r-Tensorfloatr=rVrWrrrrs       @    O