o پiJ@sUdZddlZddlZddlZddlZddlZddlZddlmZddlm Z m Z ddl m Z ddl mZddlmZddlmZmZmZmZmZmZmZdd lmZddlZddlZdd lmZmZdd lm Z dd l!m"Z"dd l#m$Z$ddl%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0ddl1m2Z2e.Z3e*Z4e0Z5e-Z6edgdZ7e8ej9j:j;Zdee?eej@effdeeee?efeej@ffddZAiZBee?e8feCd<de?de?fddZDiZEee?egedffeCd<dd!d"ZFe2d#gd$e d#ej@d%e?ddfd&d'ZGe2d#d(d#ej@d%e?d)e?dej@fd*d+ZHe2d,gd$d,ej@d-ej@d%e?ddfd.d/ZIe2d,gd$d,ej@d-ej@d%e?ddfd0d1ZJGd2ddZKdaLeeKeCd3<deKfd4d5ZMd6ee8d7e8d8e?deKfd9d:ZN   ;   < dd=eee8d7e8d8e?d>eeOd?eeOd@eOd%ee?dAeeOdBeOdCeeOdeKfdDdEZPdaQeeKeCdF<daReeKeCdG<daSeeKeCdH<daTeeKeCdI<d;aUeOeCdJ<dKeOfdLdMZVdeKfdNdOZWdeKfdPdQZXdeKfdRdSZYdaZeeKeCdT<da[eeKeCdU<da\eeKeCdV<deKfdWdXZ]deKfdYdZZ^deKfd[d\Z_eWZ`daaeeKeCd]<deKfd^d_ZbebZcd`daZde ddbeejejffdcddZgeheiZjd "tp:0" _get_unique_name("tp") -> "tp:1" r:)rA)rBnewnamer.r.r/_get_unique_nameps rFGroupCoordinator_groupsgroupcCst|t|j<dSN)weakrefrefrH unique_namerIr.r.r/_register_groupsrOtensor) mutates_args group_namecCsH|tvs Jd|dt|}|durtd|d||dSNzGroup z is not found.z is destroyed.)rH ValueError_all_reduce_in_place)rPrRrIr.r.r/inplace_all_reduces  rV) out_shapeoutplace_all_reduce_methodcCsF|tvs Jd|dt|}|durtd|d|||SrS)rHrT_all_reduce_out_place)rPrRrXrIr.r.r/outplace_all_reduces   rZoutputinputcCJ|tvs Jd|dt|}|durtd|d|||dSrS)rHrT_all_gather_into_tensorr[r\rRrIr.r.r/reg_all_gather_into_tensor  r`cCr]rS)rHrT_reduce_scatter_tensorr_r.r.r/reg_reduce_scatter_tensorrarcc@seZdZUdZeed<eeed<eed<eed<eed<eed<eed<eed <eed <eed <eed <eed <e e ed<e e ed<e e ed<e e ed<ddde ddfdeeedede e efd ed ed ed edededed ede e dede fddZd d!Zed"d#Zed$d%Zed&d'Zed(d)Zed*d+Zed,d-Ze  d}d.e ed/e ejjfd0d1Zd2ejd3ejfd4d5Zd2ejd6e d3ejfd7d8Z d2ejd3dfd9d:Z!d;ejdZ"d;ejdd|_?|r|jdkrz|}||j |jd|_>Wnt@y} ztABd| dWYd} ~ nd} ~ wwt8rz|r||j |jd|_?Wn+t@y} ztABd| WYd} ~ nd} ~ wwn|jdkrt8rtACdd|_D|j&r|jdkr||j |jd|_Ddd lEmF}!dd!lGmH}"dd"lImJ}#d|_K|r|jdkr|!|jd#|_Kd|_L| r,|jdkr,|#|jd#|_Ld|_M| r?|jdkr?|"|jd#|_Mdd$lNmO}$d|_P| r\|jdkr^|$Q|j d%d&|_PdSdSdS)'N anonymous LOCAL_SIZErzcuda:znpu:zxpu:zmusa:cpur#r"r#mooncake)MooncakeBackendOptions)backend pg_optionsz mooncake-cpu)rgloo)rtimeout)dispatch_custom_allreduce)PyMscclppCommunicator)PyNcclCommunicator)is_symmetric_memory_enableduse_symmetric_memory)TorchSymmMemCommunicator)is_allocation_symmetric)QuickAllReduceqr_rocm_arch_availablerD)rIr"use_current_stream)rIr"z#Setup Custom allreduce failed with zJ. To silence this warning, specify --disable-custom-all-reduce explicitly.z%Failed to initialize QuickAllReduce: z4[AR] All-reduce call path: NCCL (custom AR disabled))HpuCommunicator)NpuCommunicator)XpuCommunicatorrN) MessageQueuei@)RrFrMrOr*r3get_rankrdrgrjrir local_sizerr%SGLANG_ONE_VISIBLE_DEVICE_PER_PROCESSgetr"_is_npu_is_xpu_is_musar+ device_moduleoneslenint32 mooncake.epr new_grouprerfindexrh active_ranksactive_ranks_cpurkr{rlrmrnrxryrzro=sglang.srt.distributed.device_communicators.custom_all_reducer5sglang.srt.distributed.device_communicators.pymscclppr2sglang.srt.distributed.device_communicators.pyncclrReturn the global rank of the process that precedes the callerrDrrr.r.r/ prev_rankrzGroupCoordinator.prev_rankgraph_capture_contextr&c csT|dur|dur|j}t|}n|j}|j}|dur tn|}t}||kr0|||j|j|O|j }|sCt}n|j dtd}|j }|sTt} n|j dd} || |VWdn1smwYWdn1s|wYWdn1swYWddSWddS1swYdS)NTenabler&r) rr,r%r&rqrcapturer wait_streamrp change_stater) rrr&rqmaybe_ca_context curr_streamrpmaybe_pynccl_contextrmaybe_pymscclpp_contextr.r.r/ graph_captures:     PzGroupCoordinator.graph_captureinput_r7cCsV|jdkr|Stjrtj}ntj}to|}|rJ|jsJ|jdurJ| | }||jj kr<|jj |ddS|j ||jj |ddS|jrkt|j|j|jr`tjj|t|Stjj||jd|S|jdurz|jjsz|j|S|jdur|jjs|j|S|jdur|jjs|j|S|jdur|r|jjdt d|j||WdS1swYd}|jdur|jjs|j!|rd}n?|j"dur|j"js|j"#|rd }n-|j$dur|j$js|j$%|rd }n|j&dur|j&js|j&'|rd }nt(rd }|dur"t)||j*|d St+||j*d|S)as User-facing all-reduce function before we actually call the all-reduce operation. We need this because Dynamo does not support passing an arbitrary object (`self` in this case) to a custom op. We need to pass the group name as a string, and then look up the group coordinator from the group name, dispatch the all-reduce operation to the group coordinator. In addition, PyTorch custom ops do not support mutation or returning a new tensor in the same op. So we need to figure out if the op is in-place or out-of-place ahead of time. rDNF) registeredTrNrcaqr pymscclpptorch_symm_mempynccl)rRrXrR),rfrSGLANG_USE_1STAGE_ALLREDUCEis_setr%SGLANG_ENABLE_DETERMINISTIC_INFERENCErrrqnumel element_sizemax_sizedeterministic_all_reduceregister_bufferrr#rr*ops sgl_kernel shm_allreduce REDUCE_OP_SUMr3 all_reducerjrdisabledrrrprrrshould_custom_arrshould_quick_allreducershould_mscclpp_allreducerrshould_torch_symm_mem_allreducerrZrMrV)rr use_1stage_aruse_deterministic_arinp_sizerXr.r.r/rs                     zGroupCoordinator.all_reducerXc Cs|j}|j}|j}|j}|j}t|||||gsJ|dkr)|jr#J||}nN|dkr8|jr2J||}n?|dkrG|jrAJ| |}n0|dkrV|jrPJ| |}n!|dkrw|j dt d | |}Wdn1srwY|dus}J|S)NrrrrrTr) rqrrrrrpanyrcustom_all_reducequick_all_reducerrrrZ) rrrXrqrrrrrpoutr.r.r/rYus4          z&GroupCoordinator._all_reduce_out_placecCsZ|j}|j}|dur|js||dS|dur"|js"||dStjj||jddSNrN)rprrrrr*r3rj)rrrprrr.r.r/rUsz%GroupCoordinator._all_reduce_in_placer[r\cCsv|j}|dur/|jr|r/|jdtd|||Wd|S1s(wY|Stjj|||j d|SNTrrN) rprrrrreduce_scatterr*r3reduce_scatter_tensorrjrr[r\rpr.r.r/rbs$ z'GroupCoordinator._reduce_scatter_tensorcCs(tr |||dSt|||jddSNr)rrbrcrMrr[r\r.r.r/rsz&GroupCoordinator.reduce_scatter_tensor input_listcCstjj|||jd|Sr)r*r3rrj)rr[rr.r.r/rszGroupCoordinator.reduce_scattersizescCs|j}|j}|jdtdl|dur|jrJd|dur7t||ks&J|jdt|ks1J||j}n|jd|dksBJ|jd|}|f|jdd}|durbt j ||j |j d}n|j|ksiJ|j |||d|WdS1s}wYdS)NTrz&pynccl is required for reduce_scattervrrDrr)rfrprrrrshapesumrhr*emptyr#r"r)rrr[rrfrp chunk_size output_shaper.r.r/reduce_scattervs.  $z GroupCoordinator.reduce_scattervcCsv|j}|dur/|jr|r/|jdtd|||WddS1s(wYdStjj|||j ddSr) rprrrr all_gatherr*r3all_gather_into_tensorrjrr.r.r/r^s" z(GroupCoordinator._all_gather_into_tensorcCs,tstr |||dSt|||jddSr)rrr^r`rMrr.r.r/rsz'GroupCoordinator.all_gather_into_tensorcCsP|dus Jd|d|j}|dus|jr|||dS|j|||ddS)a Implement an asynchronous `allgather` operation on a specified stream. (the default `torch.distributed.all_gather_into_tensor` will trigger event synchronization), eliminating the CPU-side launch-kernel blocking issue caused by synchronization problems. The specific implementation uses the interface provided by pynccl to remove the synchronization logic of events. NzInvalid params stream (zQ, Please specify the stream to use when calling cp_all_gather_into_tensor_async.)r&)rprrcp_all_gather_into_tensor)rr[r\r&rpr.r.r/cp_all_gather_into_tensor_asyncs z0GroupCoordinator.cp_all_gather_into_tensor_asyncrdimoutput_tensor_listc Cs|j}|dkr|durtd|d|dS|S|dur)tjj|||jdS| |kr8|ksEnJd|d| |j }|durU|j sU|||S|j }|dure|j se|||S|dkro||7}| }|d|f|dd}|j || dtj||j|jd} Wdn1swY|jrt|j|j|jrtjj||Stjj| ||jdn|| || |f|} | d|} | |d||||f||dd} | S) NrDz|Performing in-place all-gather with a group size of 1. This may be unnecessary; consider bypassing it for better efficiency.rrN Invalid dim () for input tensor with shape rr)rfrrcopy_r*r3rrjrr$rrrrrr r#r"rrrrr shm_allgatherrreshapemovedim) rrrrrfhpu_commnpu_comm input_size output_size output_tensorr.r.r/rs\"       (zGroupCoordinator.all_gatherc sjj}|jdtdi|dur|jrJd d dtjdttt ffdd }t |tjr6|g}g}g}|D]}|||d \}} | || | q<| t |D]\} }|j|| ||| d qY||WdS1szwYdS) z Supports varying sizes per rank and input tensor list. `sizes`: a list of len(world_size) with the number of items per rank to gather. TrNz"pynccl is required for all_gathervrrcs|}dur5tksJ|jdjksJtf|dd}tfddDr4dn |df|dd}jdudtj||j |j d}Wd|fS1scwY|fS)NrrDc3s|] }|dkVqdSrNr.).0srr.r/ mszTGroupCoordinator.all_gatherv.._all_gather_allocate_output..rr) r$rr rhr allrr*r r#r")rrrr r!rrfrr/_all_gather_allocate_outputds"  zAGroupCoordinator.all_gatherv.._all_gather_allocate_outputrrJ)rfrprrrr*r5r r intr9r; group_start enumerater group_end) rrrrpr( output_list size_listinpr!r$ir.r'r/ all_gathervQs8     $zGroupCoordinator.all_gathervrdstcs|j}|dkr S |krks%nJd|d|dkr/|7}|jdurB|jjsB|j|j||S|j|krSfddt|D}nd}tj j||j ||j d|j|krqtj ||d }|Sd}|S) z NOTE: We assume that the input tensor is on the same device across all the ranks. NOTE: `dst` is the local rank of the destination rank. rDrrrNcsg|]}tqSr.)r* empty_liker#_rr.r/ sz+GroupCoordinator.gather..)r2rIr) rfrr$rrgatherrhranger*r3rerjcat)rrr2rrf gather_listr!r.r6r/r9s*"   zGroupCoordinator.gathersrccCF||jks Jd|d|jdkr|Stjj||j||jd|S)z^Broadcast the input tensor. NOTE: `src` is the local rank of the source rank. Invalid src rank ()rDr=rI)rfr*r3 broadcastrerj)rrr=r.r.r/rBs zGroupCoordinator.broadcastobjcCs||jks Jd|d|jdkr|S|jdur'|dks!Jd|j|S|j|kr)zcBroadcast the input object list. NOTE: `src` is the local rank of the source rank. r?r@rDrA)rfr*r3rErerj)rrGr=rIr.r.r/rEs z&GroupCoordinator.broadcast_object_listcCs$dg|j}tjj|||jd|Sr)rfr*r3all_gather_objectri)rrCobjsr.r.r/rHs z"GroupCoordinator.all_gather_object async_sendc Cs||jks Jd|d||jksJd|rtjjntjj}tjt|tj d}tj | gtj dd}g}|||j ||jd}|rN|t|||||j ||jd} |rc|t| ||S)a Send the input object list to the destination rank. This function uses the CPU group for all communications. TODO: If you want to use GPU communication, please add a new argument (e.g., data_group, group), use other functions (e.g., send), or implement a new function (e.g., send_object_device). NOTE: `dst` is the local rank of the destination rank. Invalid dst rank (r@zKInvalid destination rank. Destination rank is the same as the current rank.rrrrN)rfrhr*r3isendsend frombufferpickledumpsuint8rPrlongrerir;r0) rrCr2rJ send_func object_tensor size_tensorp2p_work size_work object_workr.r.r/ send_objects2 zGroupCoordinator.send_objectcCs ||jksJd|d||jksJdtjdtjdd}tjj||j||jd}| tj| tj dd}tjj||j||jd}| t |}|S)z3Receive the input object list from the source rank.r?r@zAInvalid source rank. Source rank is the same as the current rank.rDrrrA)rfrhr*r rRr3irecvreriwaititemrQrOloadsnumpy)rr=rUr1rTrCr.r.r/ recv_objects* zGroupCoordinator.recv_objectr6metadata_groupcCstjr |jdkr |S|j}|j}||jksJd|d|j}||krg}t|ts6Jdt |t |\}}|j ||dg}|D],} | dkrPqG| j ratjj| |j||dd} n tjj| |j||dd} || qG|D]} | qv|S|j d |d}i}g}|D]K\} } t| trtj| j| j| jd } | dkr| || <q| j rtjj| |j||dd} n tjj| |j||dd} || | || <q| || <q|D]} | q|S) ziBroadcast the input tensor dictionary. NOTE: `src` is the local rank of the source rank. rDr?r@Expecting a dictionary, got r=rT)r=rIasync_opNr)r*r3is_initializedrfrjrirhr9dictr:r@rDrrrBrer;r[r!r r$r#r")rr6r=rIr`rhr<r= async_handlesrPhandle async_handler>r?r.r.r/broadcast_tensor_dict>sr      "        z&GroupCoordinator.broadcast_tensor_dictall_gather_groupcCs6|jdkr|S|dur dn|j}|durdn|j}|j}|j}|dur+|jd|j}||jks8Jd|dt|tsFJdt|t|\} } |rRtj j ntj j } |j | ||d} | D]8} | dkriq`|dur}| |dkr}| |d|} | jr|n|}| | |j||d }|r| t|| q`| S) zdSend the input tensor dictionary. NOTE: `dst` is the local rank of the source rank. rDNrrKr@ra)r2rJrrN)rfrhrjrir9rer:r@r*r3rLrMrYrrrrer;r0)rr6r2rjrJall_gather_sizeall_gather_rankrIr`r<r=rS p2p_worksrP comm_groupr1r.r.r/send_tensor_dicts:    z!GroupCoordinator.send_tensor_dictcCs`tjr |jdkr dS|durdn|j}|durdn|j}|j}|j}|dur0|jd|j}||jks=Jd|d|j|d}i}|D]f\} } t| t rtj | j | j | j d} | dkrf| || <qG|duoq| |dk} | r| j} | |d|} | jr|n|}tjj| |j||d }|| r|j| dd } | | } | || <qG| || <qG|S) zdRecv the input tensor dictionary. NOTE: `src` is the local rank of the source rank. rDNrr?r@rbrrrAr8)r*r3rdrfrhrjrir_r9r!r r$r#r"rr rrrZrer[r)rr=rjrkrlrIr`recv_metadata_listr6r>r?rPuse_all_gather orig_shapernr1r.r.r/recv_tensor_dictsF         z!GroupCoordinator.recv_tensor_dictcCstjj|jddS)a+Barrier synchronization among the group. NOTE: don't use `device_group` here! `barrier` in NCCL is terrible because it is internally a broadcast operation with secretly created GPU tensors. It is easy to mess up the current device. Use the CPU group instead. rNN)r*r3barrierrirr.r.r/rtszGroupCoordinator.barrierrPcCsZ |dur |jd|j}|j}|dur|js|||dStj||j||jdS)zJdg} t |D]}t t |||} | | qDt | tj |dddadS) a Initialize model parallel groups. Arguments: tensor_model_parallel_size: number of GPUs used for tensor model parallelism. expert_model_parallel_size: number of GPUs used for expert model parallelism. pipeline_model_parallel_size: number of GPUs used for pipeline model parallelism. attention_data_parallel_size: number of GPUs used for attention data parallelism. attention_context_model_parallel_size: number of GPUs used for attention context parallelism. moe_data_model_parallel_size: number of GPUs used for moe data parallelism. Let's say we have a total of 8 GPUs denoted by g0 ... g7 and we use 2 GPUs to parallelize the model tensor, and 4 GPUs to parallelize the model pipeline. The present function will create 4 tensor model-parallel groups and 2 pipeline model-parallel groups: 4 tensor model-parallel groups: [g0, g1], [g2, g3], [g4, g5], [g6, g7] 2 pipeline model-parallel groups: [g0, g2, g4, g6], [g1, g3, g5, g7] Let's say we use 2 GPUs for attention context parallelism (attn_cp_size=2) and 4 GPUs for attention tensor parallelism (attn_tp_size=4). As for MoE part, we use 2 GPUs for moe data parallelism (moe_dp_size=2) and 4 GPUs for moe expert parallelism (moe_ep_size=4). The present function will create the following groups: 2 tensor model-parallel groups: [g0, g1, g2, g3], [g4, g5, g6, g7] 4 attention context-parallel groups: [g0, g4], [g1, g5], [g2, g6], [g3, g7] 2 moe expert-parallel groups: [g0, g1, g2, g3], [g4, g5, g6, g7] 4 moe data-parallel groups: [g0, g4], [g1, g5], [g2, g6], [g3, g7] Note that for efficiency, the caller should make sure adjacent ranks are on the same DGX box. For example if we are using 2 DGX-1 boxes with a total of 16 GPUs, rank 0 to 7 belong to the first box and ranks 8 to 15 belong to the second box. z world_size (z.) is not equal to tensor_model_parallel_size (z") x pipeline_model_parallel_size (r@Nz2tensor model parallel group is already initializedrD$SGLANG_USE_MESSAGE_QUEUE_BROADCASTERtruetp)rorRr{zNtensor model parallel group for PD-Multiplexing Prefill is already initializedpdmux_prefill_tpTFz=attention context model parallel group is already initializedattn_cprr)SYNC_TOKEN_IDS_ACROSS_TPz.cSs|dSrxr.)xr.r.r/osz.create_custom_parallel_group..)r>N)rerzRank z+ successfully created/joined custom group: )r*r3rdrrsortedrsetr:rHtupleaddr;sortrrr) rvrrfrd local_configgathered_configs unique_groupsseen_signaturesconfig config_tuple my_new_groupg_ranksrIr.r.r/create_custom_parallel_groupNs2   rcCst|p tjtj}tst||||dSt|ks&Jdtd|tj }||ks8Jd|d|dS)zHelper to initialize model parallel groups if they are not initialized, or ensure tensor-parallel and pipeline-parallel sizes are equal to expected values if the model parallel groups are initialized. Nzjtensor parallel group already initialized, but of unexpected size: get_tensor_model_parallel_world_size()=z vs. tensor_model_parallel_size=zSpipeline parallel group already initialized, but of unexpected size: pp_world_size=z" vs. pipeline_model_parallel_size=) r*r3rrrjmodel_parallel_is_initializedr$get_tensor_model_parallel_world_sizerrf)rrrr pp_world_sizer.r.r/!ensure_model_parallel_initializeds0   rcCstduotduS)z=Check if tensor and pipeline parallel groups are initialized.N)rrr.r.r.r/rsrtp_groupccs<trJddat}|az dVWda|adSda|aw)aPatch the tp group temporarily until this function ends. This method is for draft workers of speculative decoding to run draft model with different tp degree from that of target model workers. Args: tp_group (GroupCoordinator): the tp group coordinator z)Should not call when it's already patchedTNF)_TP_STATE_PATCHEDrr)r old_tp_groupr.r.r/patch_tensor_parallel_groups rcCtjS)z&Return world size for the world group.)rrfr.r.r.r/rrrcCr)z#Return my rank for the world group.)rrhr.r.r.r/get_world_rankrrcCr)z6Return world size for the tensor model parallel group.)rrfr.r.r.r/rrrcCr)z3Return my rank for the tensor model parallel group.)rrhr.r.r.r/get_tensor_model_parallel_rankrrcCr)z@Return world size for the attention tensor model parallel group.)rrfr.r.r.r/)get_attn_tensor_model_parallel_world_sizerrcCr)z=Return my rank for the attention tensor model parallel group.)rrhr.r.r.r/#get_attn_tensor_model_parallel_rankrrcCr)zAReturn world size for the attention context model parallel group.)rrfr.r.r.r/*get_attn_context_model_parallel_world_sizerrcCr)z>Return my rank for the attention context model parallel group.)rrhr.r.r.r/$get_attn_context_model_parallel_rankrrcCr)z8Return world size for the pipeline model parallel group.)rrfr.r.r.r/&get_pipeline_model_parallel_world_sizerrcCr)z5Return my rank for the pipeline model parallel group.)rrhr.r.r.r/ get_pipeline_model_parallel_rankrrcCr)z2Return world size for the moe data parallel group.)rrfr.r.r.r/ get_moe_data_parallel_world_sizerrcCr)z/Return my rank for the moe data parallel group.)rrhr.r.r.r/get_moe_data_parallel_rankrrcCr)z4Return world size for the moe expert parallel group.)rrfr.r.r.r/"get_moe_expert_parallel_world_sizerrcCr)z1Return my rank for the moe expert parallel group.)rrhr.r.r.r/get_moe_expert_parallel_rankrrcCr)z4Return world size for the moe tensor parallel group.)rrfr.r.r.r/"get_moe_tensor_parallel_world_size rrcCr)z1Return my rank for the moe tensor parallel group.)rrhr.r.r.r/get_moe_tensor_parallel_rankrr cCsttrtdatrtdatrtdatrtdatr&tdatr.tdatr6tdadS)z(Set the groups to none and destroy them.N)rrvrrrrrrr.r.r.r/destroy_model_parallels*r cCs,trtdatjrtjdSdSrJ)rrvr*r3rdrur.r.r.r/destroy_distributed_environment<s  r  shutdown_raycCstttt tjWdn1swY|r*ddl}| t t st tdrStjrStjt tjdrLtjdStddSt tdrdtjrdtjdSt tdrutjrutjdSt tdrtjrtjdSdSdSdS)Nrr~_host_emptyCachez;torch._C._host_emptyCache() only available in Pytorch >=2.5xpunpumusa)r r  contextlibsuppressAssertionErrorr*r3rurayshutdowngccollect_is_cpuhasattrr~ is_available empty_cache_Cr rrrrr)r rr.r.r/cleanup_dist_env_and_memoryEs2     rpg source_rankc Cstj|tjjjksJdtjj|d}tjj|d}tjdg|tjd}tj |}d}d}zz{t t l||kr`t jddd }||jdt|<tjj|jg|||d d ||<n>dg}tjj||||d |d} td d dt j| d}Wdn1swY|jdt||krd ||<Wdn1swYWnty} z td| WYd} ~ nd} ~ wwW|r|n|r|wwtjj|dt t ||kr|r|Wdn1swYtjj||ddd|DS)z This is a collective operation that returns if each rank is in the same node as the source rank. It tests if processes are attached to the same memory system (shared access to shared memory). z;in_the_same_node_as should be tested with a non-NCCL group.rNrrs magic_messageNT)creater$rArDz)multiprocessing.resource_tracker.registerc_sdSrJr.)argskwargsr.r.r/rsz%in_the_same_node_as..)rBz(Error ignored in is_in_the_same_node: %scSsg|]}|dkqS)rDr.)r#rr.r.r/r7sz'in_the_same_node_as..)r*r3rrNCCLrrrPrget_process_group_ranksrrOSErrorr SharedMemorybufrrErBrrrerrorclosertunlinkrtolist) rrrdrfis_in_the_same_nodere magic_messageshmrFrBrr.r.r/in_the_same_node_as`sj       r0reversecCsz ddlmm}Wn tyYdSwtdur#|ja|ja|ja |r9t |dtt |dtt |dt dSt |dtt |dtt |dtdS)Nrrrr) vllm.distributed.parallel_stater3parallel_statervllm_get_pp_grouprrvllm_get_tp_grouprvllm_get_world_groupsetattr)r1vllm_parrlel_stater.r.r/ monkey_patch_vllm_parallel_states      r9)rIrGr7N)NNFNNTNrJ)rrrrrN)rDrDrDrDrDrDNF)rryrw)rzrrloggingrrOrK collectionsrrr dataclassesrdatetimermultiprocessingrtypingrr r r r r r unittest.mockrr*torch.distributedrr)sglang.srt.compilation.compilation_configr0sglang.srt.compilation.piecewise_context_managerrsglang.srt.environrsglang.srt.utilsrrrrrrrrrrrsglang.srt.utils.custom_opr rrrrr!r)r3ReduceOpSUMrr%r0r|r5r@rAr-rFrHrOrVrZr`rcrGrrrr{rrrrrrrrrrrrrrrrget_tensor_model_parallel_grouprr!get_pipeline_model_parallel_grouprr~r,r getLoggerr'rrrrrrrrrrrrrrrrrrrrrrrrrrrrrr r r rr0r4r5r6r9r.r.r.r/s    $    4             '     C   5 !& B