o پiݖ@s0UdZddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z mZmZddlmZddlZddlZdd lmZddlmmZdd lmZdd lmZd d lmZddlmZm Z m!Z!m"Z"ee#Z$da%eee&d<da'eee&d<da(ee!e&d<da)ee e&d<da*eee&d<da+eee&d<da,eee&d<da-eee&d<edgdZ.de/e0ej1e Bfde2e3e2e0e fe3ej1ffddZ4iZ5e/e0egedffe&d<dd!d"Z6d#ej1d$e0dej1fd%d&Z7d#ej1d$e0dej1fd'd(Z8da%edBe&d<dZ9edBe&d)<defd*d+Z:d,e3e;d-e;d.e0defd/d0Zd7a?d8e@fd9d:ZA   ;  <  dd=e;d>e;d?e0d-e;d.e0d@ejBdBdAe;dBfdBdCZCda(edBe&d<de!fdDdEZDda+edBe&d<defdFdGZE         ddHe;dIe;dJee;dKe;dLe;dMe;dNe;dOe;d.ee0ddfdPdQZFde;fdRdSZGde;fdTdUZHde;fdVdWZIde;fdXdYZJde;fdZd[ZKde;fd\d]ZL    ; dd^e;d_e;d`e@dKe;dLe;dae;d?e0dbe;dBfdcddZMde@fdedfZNdgaOe dhefdidjZPde;fdkdlZQde;fdmdnZRddodpZSddqe@fdrdsZT ddteeBdue;de3e;fdvdwZU   xddye;d.e0dBdze0defd{d|ZV   xdd}e;d.e0dBdze0defd~dZWdefdd+Z:defdd6Z>ddZXddZYddZZddZ[ddZ\ddZ]ddZ^ddZ_de fddZ`ddZaddZbddZcddZddefddZeddZfddZgdefddGZEddZhddZiddZjddZkdefddZlddZmddZnd,ee;d-e;d.e0defdd0Z._DummyProcessGroupN) __name__ __module__ __qualname__r$rrr0WORLD ULYSSES_PGRING_PGr,r,r,r-_DummyProcessGroups rsp)sp_ulysses_degreesp_ring_degreerb sp_groupsrQ)rHrErFrN ulysses_group ring_groupz,Tensor parallel group is already initializedtpr7)rlrg!get_torch_distributed_backend_strr$rrrsr get_backendrB device_group RuntimeErrorrrrW get_ranksrErrrparallel_groupsrr ImportErrorrbrrrinit_vae_groupinit_dit_group)rrrrrrrrrFrgradit_parallel_sizerank_generator_YC_PROCESS_GROUP!_set_seq_parallel_pg_by_sp_groupsrrrr,r,r-initialize_model_parallel6s3          rcCtjS)z8Return world size for the sequence model parallel group.rrar,r,r,r-get_sp_world_size-r^rcCr)z5Return my rank for the sequence model parallel group.r rank_in_groupr,r,r,r-get_sp_parallel_rank2r^rcCr)z&Return world size for the world group.)rBrar,r,r,r-r7r^rcCr)z#Return my rank for the world group.)rBrbr,r,r,r-get_world_rank<r^rcCrz.Return world size for the data parallel group.rrar,r,r,r-get_dp_world_sizeAr^rcCrz+Return my rank for the data parallel group.rrr,r,r,r- get_dp_rankFr^rtp_sizesp_sizeenable_cfg_paralleldp_size dist_timeoutc Csddlm}tdur/tr/t|ksJd|dtt|ks-Jd|dtdSttj dd} ttj dd} ttj dd} t } t j d | | |d d t | | | || ||d t||rjd nd||||d|rtd| } tj| dSdS)Nrrfz=You are trying to initialize model parallel groups with size z-, but they are already initialized with size r| WORLD_SIZErRANKzNInitializing distributed environment with world_size=%d, device=%s, timeout=%sF)main_process_only)rarbrErcrdrFrer )rrrrrrzcuda:)rlrgrmodel_parallel_is_initializedget_tp_world_sizerintosenvirongetrrnrorrrrmr$rcuda set_device) rrrrrrrcrrgrErarbrr,r,r-5maybe_init_distributed_environment_and_model_parallelKsR      rcCs tduotduotduotduS)z:Check if tensor, sequence parallel groups are initialized.N)rrrrr,r,r,r-rs rFtp_groupccs<trJddat}|az dVWda|adSda|aw)zPatch 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. z)Should not call when it's already patchedTNF)_TP_STATE_PATCHEDrZr)r old_tp_groupr,r,r-patch_tensor_parallel_groups rcCrz6Return world size for the tensor model parallel group.rZrar,r,r,r-rr^rcCrz3Return my rank for the tensor model parallel group.rZrr,r,r,r- get_tp_rankr^rcC,trtdatjrtjdSdSr1rdestroyr$rrrsdestroy_process_groupr,r,r,r-destroy_distributed_environment  r shutdown_raycCs\tttt tjWdn1swY|r,ddl}| dSdS)Nr) destroy_model_parallelr contextlibsuppressAssertionErrorr$rrrrayshutdown)rrr,r,r-cleanup_dist_env_and_memorys   rpg source_rankc Cst|tr)tj|tjjjksJdtjj|d}tjj|d}tj |}n |j }|j }t t |}tjdg|tjd}d}d}zztt||krtjddd }||jdt|<t|trstjj|jg|||d n|j|j|d d ||<nKt|trdg}tjj||||d |d} n|jd|d } td ddtj| d}Wdn1swY|jdt||krd ||<Wdn1swYWnty} z td| WYd} ~ nd} ~ wwW|r|n |r|wwt|trtjj|dn|tt||kr)|r)| Wdn 1s4wYt|trJtjj!||d|} nt"|} t |D]} |j|| d } | | 7} qSdd| #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.r5r)rs magic_messageNT)creater)srcr0)rrz)multiprocessing.resource_tracker.registerc_sdSr1r,)argsrVr,r,r-sz%is_the_same_node_as..)namez(Error ignored in is_in_the_same_node: %scSsg|]}|dkqS)rr,).0xr,r,r- sz'is_the_same_node_as..)$r#r r$rrrBackendNCCLget_rankrget_process_group_ranksrbrar}r~r7int32rrOSErrorr SharedMemorybuflenbroadcast_object_listr broadcast_objr Exceptionrnerrorclosebarrierunlinkr; zeros_liketolist)rrrbrarDis_in_the_same_node magic_messageshmrecvreaggregated_datai rank_datar,r,r-is_the_same_node_ass                r tensor_model_parallel_sizegroup_name_suffixc CstjsJtj}|ptjtj}||dks'Jd|d|d||}g}t|D]}tt|||d|}| |q1|rMd|nd}t |tj |d|d } | S) a9Initialize a tensor parallel group for a specific model. This function creates a tensor parallel group that can be used with the patch_tensor_parallel_group context manager. It allows different models to use different tensor parallelism configurations. Arguments: tensor_model_parallel_size: number of GPUs used for tensor model parallelism. backend: communication backend to use. group_name_suffix: optional suffix to make the group name unique. Returns: A GroupCoordinator for tensor parallelism that can be used with the patch_tensor_parallel_group context manager. Example usage: ```python # Initialize tensor parallel group for model1 tp_group_model1 = initialize_tensor_parallel_group( tensor_model_parallel_size=4, group_name_suffix="model1" ) # Use tensor parallelism for model1 with patch_tensor_parallel_group(tp_group_model1): # Run model1 with tensor parallelism output1 = model1(input1) ``` r World size (z3) must be divisible by tensor_model_parallel_size (rrtp_rT)use_message_queue_broadcasterr8 r$rrrsrrrBrr~r}r'rWrE) r rFr ra num_tensor_model_parallel_groupstp_group_ranksrrDr8rr,r,r- initialize_tensor_parallel_groups,#   rsequence_model_parallel_sizec CstjsJtj}|ptjtj}||dks'Jd|d|d||}g}t|D]}tt|||d|}| |q1|rMd|nd}t |tj ||d} | S) aUInitialize a sequence parallel group for a specific model. This function creates a sequence parallel group that can be used with the patch_sequence_parallel_group context manager. It allows different models to use different sequence parallelism configurations. Arguments: sequence_model_parallel_size: number of GPUs used for sequence model parallelism. backend: communication backend to use. group_name_suffix: optional suffix to make the group name unique. Returns: A GroupCoordinator for sequence parallelism that can be used with the patch_sequence_parallel_group context manager. Example usage: ```python # Initialize sequence parallel group for model2 sp_group_model2 = initialize_sequence_parallel_group( sequence_model_parallel_size=2, group_name_suffix="model2" ) # Use sequence parallelism for model2 with patch_sequence_parallel_group(sp_group_model2): # Run model2 with sequence parallelism output2 = model2(input2) ``` rrz5) must be divisible by sequence_model_parallel_size (rrsp_r)r8r) rrFr ra"num_sequence_model_parallel_groupssp_group_ranksrrDr8rTr,r,r-"initialize_sequence_parallel_group]s(#    rcCr?r@rAr,r,r,r-rBrCcCr?rXrYr,r,r,r-rZrCcCrrrr,r,r,r-$get_tensor_model_parallel_world_sizer^rcCrrrr,r,r,r-get_tensor_model_parallel_rankr^rcCr)z2Return world size for the sequence parallel group.rr,r,r,r- get_sequence_parallel_world_sizer^rcCr)z/Return my rank for the sequence parallel group.rr,r,r,r-get_sequence_parallel_rankr^rcCrr1)rulysses_world_sizer,r,r,r-get_ulysses_parallel_world_sizercCrr1)r ulysses_rankr,r,r,r-get_ulysses_parallel_rankr r"cCrr1)rring_world_sizer,r,r,r-get_ring_parallel_world_sizer r$cCrr1)r ring_rankr,r,r,r-get_ring_parallel_rankr r&cCr?r)rr,r,r,r- get_pp_grouprCr'cCr)z8Return world size for the pipeline model parallel group.)r'rar,r,r,r- get_pipeline_parallel_world_sizer^r(cCr)z5Return my rank for the pipeline model parallel group.)r'rr,r,r,r-get_pipeline_parallel_rankr^r)cCs tdkS)zKReturn True if in the first pipeline model parallel stage, False otherwise.r)r)r,r,r,r-is_pipeline_first_stages r*cCsttdkS)zJReturn True if in the last pipeline model parallel stage, False otherwise.r)r)r(r,r,r,r-is_pipeline_last_stagesr+cCr?)Nz:classifier_free_guidance parallel group is not initialized)rr,r,r,r- get_cfg_groups r,cCr)zBReturn world size for the classifier_free_guidance parallel group.)r,rar,r,r,r-'get_classifier_free_guidance_world_sizer^r-cCr)z?Return my rank for the classifier_free_guidance parallel group.)r,rr,r,r,r-!get_classifier_free_guidance_rankr^r.cCr?rrr,r,r,r-rrCcCrrrr,r,r,r-get_data_parallel_world_sizer^r/cCrrrr,r,r,r-get_data_parallel_rankr^r0cCs0ttdkottdkottdkS)z@Return True if in the last data parallel group, False otherwise.r)rrr.r-r)r(r,r,r,r-is_dp_last_group sr1cCstttttS)z4Return world size for the DiT model (excluding VAE).)r/r-rr(rr,r,r,r-get_dit_world_sizesr2cCr?)Nz%VAE parallel group is not initialized)rr,r,r,r-get_vae_parallel_group!rCr3cCr)z-Return world size for the VAE parallel group.)r3rar,r,r,r-get_vae_parallel_world_size&r^r4cCr)z*Return my rank for the VAE parallel group.)r3rr,r,r,r-get_vae_parallel_rank+r^r5cCst|g||dS)N)rHrErIrKrLr,r,r-rM3s cCs(tduotduotduotduotduS)z=Check if tensor and pipeline parallel groups are initialized.N)rrrrrr,r,r,r-r=srcCstjjtt||dadS)NrDrF)r$rr new_groupr}r~r)rrFr,r,r-rHs  rcCr?)NzDIT group is not initialized)rr,r,r,r- get_dit_groupRrCr8cCs6tdusJdtt|||}tjj||dadS)Nz)VAE parallel group is already initializedr6)rr}r~r$rrr7)rrrF vae_ranksr,r,r-rWsrcCs4trtdatrtdatrtdadS)z(Set the groups to none and destroy them.N)rrrrr,r,r,r-rcsrcCrr1rr,r,r,r-rr)r0rr!N)rrr_rr`NN) rrNrrrrrN)rrrr_N)r!N)F)r)rNr )s__doc__rrxrr2 collectionsrcollections.abcrrmultiprocessingrtypingrrr unittest.mockr r$torch.distributedr sglang.multimodal_gen.envsmultimodal_genr{/sglang.multimodal_gen.runtime.distributed.utilsr 1sglang.multimodal_gen.runtime.utils.logging_utilsr utils.distributedrgroup_coordinatorrrrrrrnr__annotations__rrrrrrrrrwstrr%tupler}r.r/r6r;r=r>rBrrMrWrZr\boolr]rrrrrrrrrrrrrrrrrrrr rrrrrrrr"r$r&r'r(r)r*r+r,r-r.r/r0r1r2r3r4r5rr8rrr,r,r,r-s              ; M  x  9    [ C C        4