o )i C @sddlZddlZddlmZddlmZmZmZmZm Z m Z ddl Z ddl Z ddlmZddlmZmZddlmZddl mZddlmZmZdd lmZdd lmZerbdd lmZdd lm Z ee!Z"d Z#z7ddl$Z$ddl%m&Z&ddl'm(Z(zddl)m*Z*Wne+yddl)m,Z-e-j.Z*YnwGdddeZ/dZ0Wne+yZ1zdZ$e2e1Z0dZ/WYdZ1[1ndZ1[1wwde3fddZ4ddZ5dddede2fddZ6d+d d!Z7d+d"d#Z8 d,ded$ee2fd%d&Z9de:fd'd(Z;de:fd)d*Ztj|i|d|_tjjttd|_tjj t d|_ dS)NF)Zdec_hook)Zenc_hook) super__init__compiled_dag_cuda_device_setmsgspecmsgpackZDecoderrr input_decoderZEncoderr output_encoder)selfargskwargs __class__c/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/vllm/executor/ray_utils.pyr*s zRayWorkerWrapper.__init__cCstSNrr#r(r(r) get_node_ip6szRayWorkerWrapper.get_node_ipcCsBt}tjjj}|stdtjjjt |}||fS)Nz)current platform %s does not support ray.) rayget_runtime_context get_node_idvllm platformsrray_device_key RuntimeError device_nameZget_accelerator_ids)r#node_idZ device_keyZgpu_idsr(r(r)get_node_and_gpu_ids9s   z%RayWorkerWrapper.get_node_and_gpu_ids req_or_tuplecCsxt|tr |d}}n|\}}|j|}|js"t|jjd|_|j ||}t|t r4||f}|S|j |}|S)aExecute model in SPMD fashion: used only when SPMD worker and compiled DAG are both enabled. Args: req_or_tuple: A request or a tuple containing the request and intermediate tensors. Intermediate tensors are None unless if it is provided because it is > 0 pipeline stage. The request is serialized by msgspec. NT) isinstancebytesr!decoderr set_deviceworkerdeviceZ_execute_model_spmdrr"encode)r#r7Zserialized_reqintermediate_tensorsZexecute_model_reqoutputr(r(r)execute_model_spmdCs     z#RayWorkerWrapper.execute_model_spmdcCs>|jdus Jd|jstrnt|jjd|_dSdS)NWorker is not initializedT)r<rrZis_tpur;r=r+r(r(r)setup_device_if_necessaryis z*RayWorkerWrapper.setup_device_if_necessaryscheduler_outputr)rrrcCs^||jdus Jdt|tr|\}}n|d}}|jj||}t|tr-||f}|S)NrB)rCr<r8tupleZ model_runnerZ execute_modelr)r#rDr?r@r(r(r)execute_model_rayxs     z"RayWorkerWrapper.execute_model_rayvarscCstj|dSr*)osenvironupdate)r#rGr(r(r)override_env_varssz"RayWorkerWrapper.override_env_vars)rN)__name__ __module__ __qualname____doc__rstrr,rrintr6rr9rrrArCrFrrK __classcell__r(r(r&r)r&s.   & rrcCstduS)z!Returns True if Ray is available.N)r-r(r(r(r)ray_is_availablesrScCstdur tdtddS)z+Raise an exception if Ray is not available.NzFailed to import Ray: z+.Please install Ray with `pip install ray`.)r- ValueErrorray_import_errr(r(r(r)assert_ray_availablesrVplacement_grouprparallel_config device_strc CstsJdt|}|d}|d}tt}|D] \}}||||qt} | |vrCt d| d|j d|d|D]\}}t ||j krat d|j |t ||||j qGd S) zVerify a given placement group has bundles located in the right place. There are 2 rules. - Warn if all tensor parallel workers cannot fit in a single node. - Fail if driver node is not included in a placement group. zDRay is not initialized although distributed-executor-backend is ray.bundles_to_node_idbundleszdriver node id z& is not included in a placement group z. Node id -> bundles z. You don't have enough GPUs available in a current node. Check `ray status` and `ray list nodes` to see if you have available GPUs in a node `{driver_node_id}` before starting an vLLM engine.aCtensor_parallel_size=%d is bigger than a reserved number of %ss (%d %ss) in a node %s. Tensor parallel workers can be spread out to 2+ nodes which can degrade the performance unless you have fast interconnect across nodes, like Infiniband. To resolve this issue, make sure you have more than %d GPUs available at each node.N)r-is_initializedrrlistitemsappendr.r/r3idlenZtensor_parallel_sizeloggerwarning) rWrXrYZpg_dataZbundle_to_node_idsr[Znode_id_to_bundleZ bundle_idxr5Zdriver_node_idr(r(r)_verify_bundless6    rdcurrent_placement_groupc Cs|j}t}|}d}t|tkr?tj|g|d\}}t|dkr&n|d9}tdt t||t|tksz tj |ddWd Stj j y^t d|dtdd w) zWait until a placement group is ready. It prints the informative log messages if the placement group is not created within time. )timeoutra6Waiting for creating a placement group of specs for %d seconds. specs=%s. Check `ray status` and `ray list nodes` to see if you have enough resources, and make sure the IP addresses used by ray cluster are the same as VLLM_HOST_IP environment variable specified in each node if you are running on a multi-node.z:Cannot provide a placement group of placement_group_specs=z within z^ seconds. See `ray status` and `ray list nodes` to make sure the cluster has enough resources.N) bundle_specstimereadyPG_WAIT_TIMEOUTr-waitrarbinforQget exceptionsZGetTimeoutErrorrT)replacement_group_specssZ pg_ready_ref wait_intervalrk_r(r(r)_wait_until_pg_readys6  rucCstj|t}d}t|tkr>tj}|durdS|d9}tdtt|t |t|tksdSdS)Nrfrhz?Waiting for removing a placement group of specs for %d seconds.) r-utilZremove_placement_grouprjrlget_current_placement_grouprbrnrQsleep)rerrrspgr(r(r)_wait_until_pg_removeds   rz ray_addressc s.tddlm}trtdn0|s|r;zt dWn t y:t dtj ||j |j dYn wtj ||j d|jsQtd|jd |jrX|j}ntj}|rtd |j}d}|D]}|d}|d krtd d |r|d 7}qk|j |krtddd|j d|d ngtdtd}|j |krt dfddt|j D} t} t} t| } | dd krtdd| ddd| d| d d| dd| <tjj| dd}t||d us Jt||||_d S)!aInitialize the distributed cluster with Ray. it will connect to the Ray cluster and create a placement group for the workers, which includes the specification of the resources for each distributed worker. Args: parallel_config: The configurations for parallel execution. ray_address: The address of the Ray cluster. If None, uses the default Ray cluster address. rr z8Ray is already initialized. Skipping Ray initialization.autoz_No existing RAY instance detected. A new instance will be launched with current node resources.)addressZnum_gpus runtime_env)r}r~zcurrent platform z does not support ray.z"Using the existing placement groupz/Placement group bundle cannot have more than 1 .zThe number of required z(s exceeds the total number of available z6s in the placement group. Required number of devices: z. Total number of devices: zANo current placement group found. Creating a new placement group.z\The number of required %ss exceeds the total number of available %ss in the placement group.csg|]}diqS)g?r().0rtrYr(r) _sz*initialize_ray_cluster..zCurrent node has no z" available. current_node_resource=z#. vLLM engine cannot start without z . Make sure you have at least 1 z% available in a node current_node_id=z current_ip=gMbP?znode:ZPACK)ZstrategyN)rVvllm.platformsrr-r\rbrnZis_rocmZis_xpuinitConnectionErrorrcZ world_sizeZray_runtime_envr2rTr4rWrvrwrirocluster_resourcesrangerr.r/rrurd) rXr{rrer[Zdevice_bundlesZbundleZbundle_devicesZnum_devices_in_clusterrqZ current_ipZcurrent_node_idZcurrent_node_resourcer(rr)initialize_ray_clusters                rcCs@ddlm}t}t|d}|}||dksJ||S)Nr)TPUAcceleratorManagerZTPU)Zray._private.acceleratorsrr-rrQZ!get_current_node_num_accelerators)rrZ total_tpusZ tpus_per_noder(r(r)get_num_tpu_nodes~s  rcCsntj}tj}d}|r5t}|D]\}}||jkr0|dD] \}}||q&qt |}|S)NrrZ) r-rvrrwsetr^r`hexaddra)Zpg_tableZ current_pgZ num_nodesZ nodes_in_pgZpg_keyryrtnoder(r(r) get_num_nodes_in_placement_groups   r)rerr*)=rHrj collectionsrtypingrrrrrrrrr0Z vllm.configr Zvllm.executor.msgspec_utilsr r Z vllm.loggerr rZ vllm.sequencerrZ vllm.utilsrZvllm.worker.worker_baserZvllm.v1.core.sched.outputrZvllm.v1.outputsrrLrbrlr-Zray.utilrZray.util.placement_grouprZray._private.stater ImportErrorr_stateZ_available_resources_per_noderrUerPboolrSrVrdrurzrrQrrr(r(r(r)sj              j . ) p