o )iH @s2UdZddlZddlZddlZddlZddlmZddlmZmZddl m Z ddl m Z ddl mZmZmZmZddlmZddlZddlZdd lmZmZdd lmZddlmZdd lmZdd lmZdd l m!Z!ddl"m#Z#m$Z$m%Z%m&Z&e GdddZ'edgdZ(de)e*eej+effde,e-e,e*efe-ej+ffddZ.iZ/e)e*e0fe1d<de*de*fddZ2iZ3e)e*egedffe1d<dddZ4d ej+d!e*dej+fd"d#Z5d ej+d!e*dej+fd$d%Z6d ej+d&e0d'e0d!e*dej+f d(d)Z7d ej+d&e0d'e0d!e*dej+f d*d+Z8d ej+d&e0d'e0d!e*dej+f d,d-Z9d ej+d&e0d'e0d!e*dej+f d.d/Z:e&rVdd0l;mda?ee>e1d3<da@ee0e1d4<de>fd5d6ZAd7e-e0d8e0d9e*de>fd:d;ZB < dd=e-e-e0d8e0d9e*d>eCd!ee*de>f d?d@ZDdaEee>e1dA<de>fdBdCZFedDdEdFZGdaHee>e1dG<daIee>e1dH<de>fdIdJZJdaKee>e1dK<de>fdLdMZLde>fdNdOZMedPdQdRZNedSejOfdTdUZPe!eQZRdVaSdWeCfdXdYZT Z Z [ Z \dd'e0d]e0d^e*d8e0d9e*f d_d`ZU a a ddbe0dce0d9ee*ddfdddeZV ddbe0dce0d9ee*ddfdfdgZWdhejXjYfdidjZZdkdlZ[dfdndoZ]dpdqZ^drdsZ_de0fdtduZ`dvdwZadxdyZbddzeCfd{d|Zc dd}eeefd~e0de-eCfddZddeCfddZed}eeefde0fddZfdS)avLLM distributed state. It takes over the control of the distributed environment from PyTorch. The typical workflow is: - call `init_distributed_environment` to initialize the distributed environment. - call `initialize_model_parallel` or `ensure_model_parallel_initialized` to initialize the model parallel groups. - any code dealing with the distributed stuff - call `destroy_model_parallel` to destroy the model parallel groups. - call `destroy_distributed_environment` to destroy the distributed environment. If you only need to use the distributed environment without model/pipeline parallelism, you can skip the model parallel initialization and destruction steps. N) namedtuple)contextmanager nullcontext) dataclass) shared_memory)AnyCallableOptionalUnion)patch)Backend ProcessGroup) deprecated)DeviceCommunicatorBase)StatelessProcessGroup) init_logger)direct_register_custom_opget_distributed_init_methodresolve_obj_by_qualnamesupports_custom_opc@seZdZUejjed<dS)GraphCaptureContextstreamN)__name__ __module__ __qualname__torchcudaStream__annotations__rrk/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/vllm/distributed/parallel_state.pyr2s rTensorMetadata)devicedtypesize tensor_dictreturnc Cslg}g}|D])\}}t|tjr*|jj}||t||j| f||q|||fq||fS)zSplit the tensor dictionary into two parts: 1. A list of (key, value) pairs. If the value is a tensor, it is replaced by its metadata. 2. A list of tensors. ) items isinstancerTensorr"typeappendr!r#r$)r% metadata_list tensor_listkeyvaluer"rrr _split_tensor_dict:s  r0_group_name_counternamecCs6|tvrdt|<|dt|}t|d7<|S)z|Get a unique name for the group. Example: _get_unique_name("tp") -> "tp:0" _get_unique_name("tp") -> "tp:1" r:)r1)r2Znewnamerrr _get_unique_nameVs r5GroupCoordinator_groupsgroupcCst|t|j<dSN)weakrefrefr7 unique_namer8rrr _register_groupfsr>tensor group_namecCsD|tvs Jd|dt|}|durtd|d||SNzGroup z is not found.z is destroyed.)r7 ValueError_all_reduce_out_place)r?r@r8rrr all_reducejs   rDcCs t|Sr9)rZ empty_like)r?r@rrr all_reduce_fakers rEdim world_sizecCF|tvs Jd|dt|}|durtd|d|||SrA)r7rB_reduce_scatter_out_placer?rFrGr@r8rrr reduce_scatterv   rKcCs0t|j}|j||||<tj||j|jdSNr#r"listshaperemptyr#r"r?rFrGr@Z new_shaperrr reduce_scatter_fake rTcCrHrA)r7rB_all_gather_out_placerJrrr all_gatherrLrWcCs0t|j}|j||||<tj||j|jdSrMrOrSrrr all_gather_fakerUrXcurrent_platform)Zop_nameZop_funcZ mutates_argsZ fake_impl dispatch_keyc@s$eZdZUdZeed<eeed<eed<eed<eed<eed<eed<ee ed <ee ed < dgd eeedede e e fdededee f ddZeddZeddZeddZeddZeddZeddZe dhd eefd!d"Zd#ejd$ejfd%d&Zd#ejd$ejfd'd(Zdid#ejd*ed$ejfd+d,Zd#ejd*ed$ejfd-d.Z / djd#e ejeejfd*ed0eeefd1d2Z )did#ejd*ed$ejfd3d4Z! ) dkd#ejd*ed0eeed$ejfd5d6Z"d#ejd*ed$ejfd7d8Z# / )dld#ejd9ed*ed$eejfd:d;Z$dmd#ejdZ%dnd?ee dr distributedget_rankr\r^Z new_groupr]lenrGindexr_r`ravllm.platformsrZZ is_cuda_aliker"Zis_xpuZis_out_of_treeZ device_namerfrbrZget_device_communicator_clsZ3vllm.distributed.device_communicators.shm_broadcastrkrcZcreate_from_process_groupZis_tpuuse_custom_op_callis_cpuhasattrops_Cuse_cpu_custom_send_recv)selfrdr^rerfrgr@Zself_device_groupZself_cpu_groupr]rar`rZZdevice_comm_clsrkrrr __init__st            zGroupCoordinator.__init__cC |jdS)z8Return the global rank of the first process in the grouprr]rxrrr first_rank zGroupCoordinator.first_rankcCrz)z7Return the global rank of the last process in the groupr{r|rrr last_rankr~zGroupCoordinator.last_rankcC |j|jkS)z;Return whether the caller is the first process in the group)r\r}r|rrr is_first_rank$ zGroupCoordinator.is_first_rankcCr)z:Return whether the caller is the last process in the group)r\rr|rrr is_last_rank)rzGroupCoordinator.is_last_rankcCs|j}|j}|j|d|S)z=Return the global rank of the process that follows the callerr4r_rGr]rxr_rGrrr next_rank.zGroupCoordinator.next_rankcCs|j}|j}|j|d|S)z>Return the global rank of the process that precedes the callerr4rrrrr prev_rank5rzGroupCoordinator.prev_rankgraph_capture_contextc cs|durtj}t|}n|j}t}ddlm}|jdur4t |j|s(J|jj }|dur4| }tj }||krB| |tj|&| |VWdn1sYwYWddSWddS1sqwYdS)Nr)CudaCommunicator)rrrrrrZ7vllm.distributed.device_communicators.cuda_communicatorrrbr(ca_commcaptureZcurrent_streamZ wait_stream)rxrrZmaybe_ca_contextrrZ curr_streamrrr graph_capture<s$      PzGroupCoordinator.graph_captureinput_r&cCs2|jdkr|S|jrtjjj||jdS||S)a^ 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 always make the all-reduce operation out-of-place. r4r@)rGrrrruvllmrDr<rCrxrrrr rDYs   zGroupCoordinator.all_reducecCs|jdur td|j|SNNo device communicator found)rbrBrDrrrr rCrs  z&GroupCoordinator._all_reduce_out_placerrFcCt|j}|dkr |S| |kr|ks%nJd|d||jr4tjjj||||jdS| ||SNr4z Invalid dim (z) for input tensor with shape r) rGrFr$rrrrurrWr<rVrxrrFrGrrr rWws"  zGroupCoordinator.all_gathercC |jdur td|j||Sr)rbrBrWrxrrFrrr rV z&GroupCoordinator._all_gather_out_placersizescC"|jdur td|j|||Sr)rbrB all_gathervrxrrFrrrr r zGroupCoordinator.all_gathervcCrr) rGrFr$rrrrurrKr<rIrrrr rKs"  zGroupCoordinator.reduce_scattercCrr)rbrBreduce_scattervrrrr rrz GroupCoordinator.reduce_scattervcCrr)rbrBrKrrrr rIrz*GroupCoordinator._reduce_scatter_out_placedstcCs4|j}|dkr |S|jdurtd|j|||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. r4Nr)rGrbrBgather)rxrrrFrGrrr rs   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 ()r4rr8)rGrrm broadcastr]ra)rxrrrrr rs zGroupCoordinator.broadcastobjcCs||jks Jd|d|jdkr|S|jdur'|dks!Jd|j|S|j|kr|jjdkr>|j}|j||}|j}|j}| } t || }t d|||t j sl|dusKJdt j |sbtd|t j s`Jdd }t j j||||d |d krz|d krxtj}n|}tdurttt j } t| ||attjatd tdStjt j ksJddS)NzIworld_size=%d rank=%d local_rank=%d distributed_init_method=%s backend=%srget_current_vllm_configr4zAAdjusting world_size=%d rank=%d distributed_init_method=%s for DPzRdistributed_init_method must be provided when initializing distributed environmentz>Distributed backend %s is not available; falling back to gloo.z'Fallback Gloo backend is not available.ri)rhZ init_methodrGr\rrz0Detected %d nodes in the distributed environmentz;world group already initialized with a different world size)loggerdebug vllm.configrparallel_configdata_parallel_sizeZdata_parallel_rankZworld_size_across_dpZdata_parallel_master_ipZget_next_dp_init_portrinforrmrZis_backend_availablewarningZis_gloo_availableZinit_process_groupenvsZ LOCAL_RANKrrPrangeget_world_sizer _node_countr`rrG) rGr\rr^rhrconfigripportr]rrr init_distributed_environmentsf        r r4tensor_model_parallel_sizepipeline_model_parallel_sizec CstjsJtj}tj}|ptjtj}d}ddlm }|}|dur.|j j }t | d|||}tdusAJd|d|d} dd| D} t| tj|d d d atdusdJd |d d d|d} dd| D} t| tj|ddatdusJd|dd d|d} dd| D} t| tj|ddatdusJd|dd  d||d} dd| D} t| tj|ddatd||tjtjtjtjdS)a  Initialize model parallel groups. Arguments: tensor_model_parallel_size: number of GPUs used for tensor model parallelism. pipeline_model_parallel_size: number of GPUs used for pipeline model parallelism. backend: name of torch distributed communication backend. 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] 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. r4rrNrz2tensor model parallel group is already initializedcSg|]}|qSrtolist.0xrrr Dz-initialize_model_parallel..Ttp)rgr@z4pipeline model parallel group is already initializedcSrrrrrrr rSrpprz*data parallel group is already initializedcSrrrrrrr r^rZdpz,expert parallel group is already initializedcSrrrrrrr rhrepzVrank %s in world size %s is assigned as DP rank %s, PP rank %s, TP rank %s, EP rank %s)rrmrrrn get_backendrrarrrrZarangerrviewZunbindrr^rZ transposerrrrr_) rrrhrGr\rrr Z all_ranksrdrrr initialize_model_parallel s         rcCsr|p tjtj}tst|||dSt|ks%Jdtd|tj }||ks7Jd|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. Nzotensor parallel group already initialized, but of unexpected size. got: get_tensor_model_parallel_world_size()=z( vs. wanted: tensor_model_parallel_size=zXpipeline parallel group already initialized, but of unexpected size. got: pp_world_size=z* vs. wanted: pipeline_model_parallel_size=) rrmrrramodel_parallel_is_initializedr$get_tensor_model_parallel_world_sizerrG)rrrhZ pp_world_sizerrr !ensure_model_parallel_initializedus.   r"rcCsPtdur t|tdurt|tdurt|tdur&t|dSdS)a1Prepare the communication buffer for the model. Traditional communication libraries like NCCL are almost model agnostic. However, emerging new communication libraries like MoE all2all (DeepEP) usually allocate the communication buffer based on the model shape for optimal performance. N)rrrrr)rrrr rs   rcCstduotduS)z=Check if tensor and pipeline parallel groups are initialized.N)rrrrrr r sr tp_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#Z old_tp_grouprrr patch_tensor_parallel_groups r%cCtjS)z6Return world size for the tensor model parallel group.)rrGrrrr r!rr!cCr&)z3Return my rank for the tensor model parallel group.)rr_rrrr get_tensor_model_parallel_rankrr'cCstdusJdtS)zAReturn the total number of nodes in the distributed environment. Nz*distributed environment is not initialized)rrrrr get_node_counts r(cCsDtrtdatrtdatrtdatrtdadS)z(Set the groups to none and destroy them.N)rrrrrrrrr destroy_model_parallelsr)cCs0trtdadatjrtjdSdSr9)rrrrrmrrrrrr destroy_distributed_environments r* shutdown_raycCstt|rddl}|tddlm}|j}|dur$|z| s1t j WdSWdSt yBtdYdSw)NrrYz;torch._C._host_emptyCache() only available in Pytorch >=2.5)r)r*rayshutdowngcZcollectrqrZ empty_cachersrrvZ_host_emptyCacheAttributeErrorrr)r+r,rZr/rrr cleanup_dist_env_and_memorys&   r1pg 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.r=rrs magic_messageNT)creater$rrr4z)multiprocessing.resource_tracker.registerc_sdSr9r)argskwargsrrr >sz%in_the_same_node_as..)r2z(Error ignored in is_in_the_same_node: %scSsg|]}|dkqS)r4rrrrr r[rz'in_the_same_node_as..)$r(r rrmrr ZNCCLrnrZget_process_group_ranksr\rGrPrr?Zint32 contextlibsuppressOSErrorrZ SharedMemorybufrorr2Z broadcast_objr ExceptionrerrorcloserunlinkrDZ zeros_liker)r2r3r\rGr]Zis_in_the_same_nodeZ magic_messageZshmrr2eZaggregated_dataiZ rank_datarrr in_the_same_node_as s               rCcCsFztdur tjWStjsWdStjdkWSty"YdSw)a Check if the current process is the first rank globally across all parallelism strategies (PP, TP, DP, EP, etc.). Unlike group-specific checks like `get_tensor_model_parallel_rank() == 0` or `get_pp_group().is_first_rank`, this function checks the global rank across all parallelism dimensions. Returns: bool: True if this is the global first rank (rank 0), False otherwise. Returns True if distributed is not initialized (single process). NTr)rrrrmrrnr=rrrr is_global_first_rank^s   rDcCst|tr tjj|d}n|j}|dkrdSdg|}d}t|D]+}||dkr*q!|d7}|||<t||}t|D]\}}|rK||dkrK|||<q;q!|S)z Returns the total number of nodes in the process group. Args: pg: The process group to analyze Returns: int: The total number of nodes r=r4r) r(r rrmrrGrrC enumerate)r2rGZnode_assignmentZ next_node_idZ current_rankZsame_node_flagsZ other_rankZ is_same_noderrr r }s&     r )r8r6r&Nr)rrrrr)r4r4Nr9)Fr)grr9r.rr: collectionsrrr dataclassesrmultiprocessingrtypingrrr r Z unittest.mockr rZtorch.distributedr r Ztyping_extensionsrZ vllm.envsrZ>vllm.distributed.device_communicators.base_device_communicatorrZvllm.distributed.utilsrZ vllm.loggerrZ vllm.utilsrrrrrr!rrr)rrPr0r1rrr5r7r>rDrErKrTrWrXrqrZr[r6rrrrrrrrrrrrrrrrr"rrrrrr rr"rrrr r$r%r!r'r(r)r*r1rCrDr rrrr sh                   *      B n   U