o ۾i@sddlZddlZddlZddlZddlZddlmZddlmZm Z ddl m Z ddl m Z mZddlmZmZddlmZddlmZmZmZddlZddlZdd lmZmZdd lmZdd lm Z dd l!m"Z"dd l#m$Z$ddl%m&Z&ddl'm(Z(ddl)m*Z*m+Z+ddl,m-Z-m.Z.ddl/m0Z0ddl1m2Z2m3Z3ddl4m5Z5ddl6m7Z7ddl8m9Z9m:Z:ddl;mZ>m?Z?m@Z@ddlAmBZBddlCmDZDddlEmFZFmGZGmHZHmIZImJZJmKZKmLZLmMZMmNZNddlOmPZPmQZQmRZRddlSmTZTddlUmVZVddlWmXZXddlYmZZZdd l[m\Z\m]Z]dd!l^m_Z_m`Z`dd"lambZbdd#lcmdZddd$lemfZge"ehZid%Zjd&Zked'ZlGd(d)d)ZmGd*d+d+emZnGd,d-d-enZoGd.d/d/ZpGd0d1d1epeoZqGd2d3d3epenZrdS)4N)deque)Callable Generator)Future) ExitStackcontextmanager)isclass signature)DEBUG)AnyTypeVarcast)ParallelConfig VllmConfig)1stateless_destroy_torch_distributed_process_group)enable_envs_cache) init_logger)dump_engine_exception) LoRARequest)MULTIMODAL_REGISTRY) POOLING_TASKS SupportedTask) instrumentmaybe_init_worker_tracer)(maybe_register_config_serialize_by_value)freeze_gc_heapmaybe_attach_gc_debug_callback)get_hash_fn_by_name)make_zmq_socket) decorate_logsset_process_title) BlockHash"generate_scheduler_kv_cache_configget_kv_cache_configsget_request_block_hasherinit_none_hash)SchedulerInterface)SchedulerOutput) EngineCoreOutputEngineCoreOutputsEngineCoreRequestEngineCoreRequestType FinishReasonReconfigureDistributedRequestReconfigureRankType UtilityOutput UtilityResult)EngineHandshakeMetadataEngineZmqAddressesget_device_indices)Executor) KVCacheConfig)SchedulerStats)ModelRunnerOutput)Request RequestStatus)MsgpackDecoderMsgpackEncoder)StructuredOutputManager)compute_iteration_details) __version__g@_Rc @seZdZdZ  d^dedeedededBdef d d Z e d d ded e e e e ffddZd e edffddZd_dede fddZdeefddZd`ddZd`ddZedefd d!Zedefd"d#Zd e ee efeffd$d%Zd&ed dfd'd(Zd e ee efdBeffd)d*Z d+d,Z!d-d.Z"dad0efd1d2Z#d3d4Z$ dbd5ed6ed efd7d8Z%d`d9d:Z&dcdZ'ddd?eedBfd@dAZ(d efdBdCZ)dDdEZ*dFe+d efdGdHZ,dIe d efdJdKZ-d e.e fdLdMZ/dIe d efdNdOZ0  dedPedQedBdRe dBd dfdSdTZ1  U dfdVeede2fBdWe3dBdXe dYeee4fdBd ee2f dZd[Z5de6d e ee ffd\d]Z7dS)g EngineCorezInner loop of vLLM's Engine.NF vllm_configexecutor_class log_statsexecutor_fail_callbackinclude_finished_setcCsbddlm}|||_|jjstdt|||_|||_ |dur)|j |d|_ | |\}}} ||j _||j _|jd||fdt||_|j} t| jdkrc|jjrctdd|j_|j j|jj|jj} | || |j||j| d |_|jdu|_|jjdur|j |jjt!|_"} | #||_$|j%} | dur|j &}|ri}|D] }|dur|'|q| (||j j)|_*d|_+|j*d krt,d |j*t-|j*d |_+|j.duo|j.j/|_/|j0j1d k|_2d|_3|j j4s| durt5|j j6}t7|t8| ||_3|j+dur|j9n|j:|_;|jj<|_t?t@|_Ad|_BtCtDtEdS)Nr)load_general_pluginsz2Initializing a V1 LLM engine (v%s) with config: %sinitialize_cacheargsz3Disabling chunked prefill for model without KVCacheF)rBkv_cache_configstructured_output_managerrFrD block_sizez#Batch queue is enabled with size %d)maxlenpooling)F vllm.pluginsrGrBparallel_configdata_parallel_rank_localloggerinfo VLLM_VERSIONrDmodel_executorregister_failure_callback!available_gpu_memory_for_kv_cache_initialize_kv_caches cache_confignum_gpu_blocksnum_cpu_blockscollective_rpcr<rMscheduler_configget_scheduler_clslenkv_cache_groupsenable_chunked_prefillwarningrNdecode_context_parallel_sizeprefill_context_parallel_size schedulerspeculative_configuse_spec_decode connectorinit_kv_output_aggregatorr mm_registry!engine_receiver_cache_from_configmm_receiver_cacheget_kv_connector#get_kv_connector_handshake_metadataupdateset_xfer_handshake_metadatamax_concurrent_batchesbatch_queue_size batch_queuedebugrec_transfer_configis_ec_producer model_config runner_typeis_pooling_modelrequest_block_hasherenable_prefix_cachingrprefix_caching_hash_algor%r$stepstep_with_batch_queuestep_fnasync_schedulingqueueQueueliststr aborts_queue_scheduler_pausedrrr)selfrBrCrDrErFrGr]r^rL Schedulerscheduler_block_sizerm kv_connectorxfer_handshake_metadatacontent worker_dictcaching_hash_fnrG/home/ubuntu/.local/lib/python3.10/site-packages/vllm/v1/engine/core.py__init__Rs                  zEngineCore.__init__z Prepare model span_namereturncCs t}|j}tdd|D}|rCtjddkr8t|dd}|dus(Jt |d|_ |j gt |}n|j }|d|_ ndgt |}t |t |ksTJ|j j}t|||}|j j} | |krn|jd| fd t|} | j} d} |j|t|} tjd | d d | | | fS) Ncss|]}|VqdSNr).0 kv_cache_specrrr sz3EngineCore._initialize_kv_caches..VLLM_ELASTIC_EP_SCALE_UP_LAUNCH1dp_grouprHrupdate_max_model_lenrJzFinit engine (profile, create kv cache, warmup model) took %.2f secondslocal)scope)timerXget_kv_cache_specsanyosenvirongetgetattrrsync_kv_cache_memory_sizerZrbdetermine_available_memoryrz max_model_lenr#r_r" num_blocksinitialize_from_configrU info_once)rrBstartkv_cache_specs has_kv_cacheravailable_gpu_memorymax_model_len_beforekv_cache_configsmax_model_len_afterscheduler_kv_cache_configr]r^elapsedrrrr[sD         z EngineCore._initialize_kv_caches.cC|jjSr)rXsupported_tasksrrrrget_supported_taskszEngineCore.get_supported_tasksrrequest request_wavecCst|jtstdt|j|j}r.dd|D}|j|vr.td|jd||j dur=|j s=t d|j |dS)zAdd request to the scheduler. `request_wave`: indicate which wave of requests this is expected to belong to in DP case z!request_id must be a string, got cSsg|]}|tvr|qSr)r)rtaskrrr -sz*EngineCore.add_request..zUnsupported task: z Supported tasks: NzXGot kv_transfer_params, but no KVConnector found. Disabling KVTransfer for this request.) isinstance request_idr TypeErrortypepooling_paramsrr ValueErrorkv_transfer_paramsrhrprUre add_request)rrrrsupported_pooling_tasksrrrr s*     zEngineCore.add_request request_idscCs|j|tjdS)z"Abort requests from the scheduler.N)rhfinish_requestsr9FINISHED_ABORTED)rrrrrabort_requestsAszEngineCore.abort_requestscC d|_dS)zPause the scheduler, keeping requests frozen in queue. Requests are kept frozen in queue and can be resumed later. TNrrrrrpause_schedulerI zEngineCore.pause_schedulercCr)ziResume the scheduler after a pause. Resumes processing of frozen requests in the queue. FNrrrrrresume_schedulerPrzEngineCore.resume_schedulerscheduler_outputc cs@zdVWdSty}z t|j||j|d}~ww)z3Execute the model and log detailed info on failure.N) ExceptionrrBrh make_stats)rrerrrrrlog_error_detailWs zEngineCore.log_error_detailccs|jjjs dVdSt|dd|_t|}t}dVt d dt |jdt |j dt |j dt |jdt |jd tt|d d d g |jd 7_dS)N_iteration_indexrz Iteration(z): z context requests, z context tokens, z generation requests, z, generation tokens, iteration elapsed time: iz.2fz msrO)rBobservability_config enable_logging_iteration_detailsrrr=r monotonicrUrVjoinrnum_ctx_requestsnum_ctx_tokensnum_generation_requestsnum_generation_tokensformat)rriteration_detailsbeforerrrlog_iteration_detailsgs6 z EngineCore.log_iteration_detailsc Cs|jridfS|jsidfS|j}|jj|dd}|j|}||+||| }|dur=|j |}Wdn1sGwYWdn1sVwY| |j ||}||j dkfS)zSchedule, execute, and make output. Returns tuple of outputs and a flag indicating whether the model was executed. FT non_blockNr)rrh has_requestsschedulerX execute_modelget_grammar_bitmaskrrresult sample_tokens_process_aborts_queueupdate_from_outputtotal_num_scheduled_tokens)rrfuturegrammar_output model_outputengine_core_outputsrrrrs.     zEngineCore.stepmodel_executedcCsB|js|jr|r|j}|dur|j|dSdSdSdSdSr)rrjrXtake_draft_token_idsrhupdate_draft_token_ids)rrdraft_token_idsrrr post_steps   zEngineCore.post_stepc Cs|jridfS|j}|dusJt||jksJd}d}|jrv|j}|jj|dd}|j s7|j dk}|j s<|sDt t t|}n|jsV|j|}|jj|dd}n|}|su||||f|rut||jkru|ddsudSn|szdS|\}}}||-|||} | dur|td Wdn1swYWdn1swY||j|| } |r|jr|j} | dusJ|j| ||j|}|jj|dd}||||f| |fS) aSchedule and execute batches with the batch queue. Note that if nothing to output in this step, None is returned. The execution flow is as follows: 1. Try to schedule a new batch if the batch queue is not full. If a new batch is scheduled, directly return an empty engine core output. In other words, fulfilling the batch queue has a higher priority than getting model outputs. 2. If there is no new scheduled batch, meaning that the batch queue is full or no other requests can be scheduled, we block until the first batch in the job queue is finished. 3. Update the scheduler from the output. FNTrrrHNTNFzunexpected error)rrvrbrurhrrrXrryrr|r rr7 pending_structured_output_tokensrr appendleftdonepoprrr RuntimeErrorrrrjr update_draft_token_ids_in_output) rrvrdeferred_scheduler_outputr exec_futurerrexec_model_futrrrrrrrs        z EngineCore.step_with_batch_queuecCsX|js*g}|js#|j}|t|tr|fn||jr ||dSdSr)rempty get_nowaitextendrrr)rridsrrrr*s    z EngineCore._process_aborts_queuecCs2|j|jr |j|jr|jdSdSr)rM clear_backendrXshutdownrhrrrrr 4s  zEngineCore.shutdownTis_startcC|j|dSr)rXprofile)rr rrrr ;zEngineCore.profilecCs6|jr td|jdur|j|jdS)NzcResetting the multi-modal cache when requests are in progress may lead to desynced internal caches.)rhhas_unfinished_requestsrUrero clear_cacherXreset_mm_cacherrrrr>s   zEngineCore.reset_mm_cachereset_running_requestsreset_connectorcCs|j||Sr)rhreset_prefix_cache)rrrrrrrMszEngineCore.reset_prefix_cachecCs,|jr td|j|jdS)a0Reset the encoder cache to invalidate all cached encoder outputs. This should be called when model weights are updated to ensure stale vision embeddings computed with old weights are not reused. Clears both the scheduler's cache manager and the GPU model runner's cache. z_Resetting the encoder cache when requests are in progress may lead to desynced internal caches.N)rhrrUrereset_encoder_cacherXrrrrrTs  zEngineCore.reset_encoder_cacherOlevelcCr r)rXsleep)rrrrrrhrzEngineCore.sleeptagscCr r)rXwake_up)rrrrrrkrzEngineCore.wake_upcCrr)rX is_sleepingrrrrrnrzEngineCore.is_sleepingcCs|jdSr)rXexecute_dummy_batchrrrrrqszEngineCore.execute_dummy_batch lora_requestcC |j|Sr)rXadd_lora)rrrrrrt zEngineCore.add_loralora_idcCrr)rX remove_lorarr rrrr!wrzEngineCore.remove_loracCs |jSr)rX list_lorasrrrrr#zs zEngineCore.list_lorascCrr)rXpin_lorar"rrrr$}rzEngineCore.pin_lorapathpatternmax_sizecCs|jj|||ddS)N)r%r&r')rXsave_sharded_state)rr%r&r'rrrr(s zEngineCore.save_sharded_statermethodtimeoutrKkwargscCs|j||||Sr)rXr_)rr)r*rKr+rrrr_szEngineCore.collective_rpccCsJ|jdur|jr|j|j|_t||j}|jr |j|||j fS)zPreprocess the request. This function could be directly used in input processing thread to allow request initialization running in parallel with Model forward N) ro mm_featuresget_and_update_featuresr8from_engine_core_requestr}use_structured_outputrM grammar_init current_wave)rrreqrrrpreprocess_add_requests   z!EngineCore.preprocess_add_requestrr)rN)T)FF)rOr)NN)NrN)8__name__ __module__ __qualname____doc__rrr4boolrrrtupleintr5r[rrr8rrrrrrrr'rrdictr)rrrrr r rrrrrrrrrr!setr#r$r(r@floatr r_r*r3rrrrrAOs   ;!  #  x      rAcseZdZdZdZedd d>ddded ed ed e e d ed edBde ffddZ e d eded eded edBdeeddff ddZe d>dejd eded ededededBdeeddffddZe d>dejd edededBdef ddZedddde d e fd!d"Zdefd#d$Zd%d&Zd'd(Zdefd)d*Zd+ed,eddfd-d.Z ed/d0Z!d1d2Z"d3e#ed4edBded5e$j%fd6d7Z&d8e#ed9edBde fd:d;Z'd,e(ddfdsz)EngineCoreProc.__init__..little)length byteorderFz1Has DP Coordinator: %s, stats publish address: %sT)targetrKdaemon r*z'Input socket thread died during startupz0Waiting for READY message from DP Coordinator...))rrr:r+r rFr;r)bytes output_queuerAto_bytesengines_running_perform_handshakesrboutputs client_countcoordinator_outputhas_coordinatorfrontend_stats_publish_addressrUrwrSdata_parallel_external_lbpublish_dp_lb_stats_init_data_parallelsuperr threadingEventThreadprocess_input_socketsinputscoordinator_inputrprocess_output_sockets output_threadwaitis_aliverrV)rrBrBrCrCrDrDrArEidentity addressesinternal_dp_balancing ready_event input_thread __class__rrrs         D"zEngineCoreProc.__init__rjrc cst}|o |du}| }||||||||j} |dur4| } | VWdn1s.wYn=|s8J||||dd|} | %} | } | j| _| j| _| VWdn1s]wYWdn1slwY|dS)a Perform startup handshakes. For DP=1 or offline mode, this is with the colocated front-end process. For DP>1 with internal load-balancing this is with the shared front-end process which may reside on a different node. For DP>1 with external or hybrid load-balancing, two handshakes are performed: - With the rank 0 front-end process which retrieves the DP Coordinator ZMQ addresses and DP process group address. - With the colocated front-end process which retrieves the client input/output socket addresses. with the exception of the rank 0 and colocated engines themselves which don't require the second handshake. Here, "front-end" process can mean the process containing the engine core client (which is the API server process in the case the API server is not scaled out), OR the launcher process running the run_multi_api_server() function in serve.py. NTF)zmqContext_perform_handshakerSrdrW __post_init__) rrCrjrBrBrD input_ctxis_localheadless handshakerklocal_handshakeclient_addressesrrrrVs<     z"EngineCoreProc._perform_handshakesctxrwparallel_config_to_updatec cst||tj|ddd9}|||||} | V|jj} |j} d||| | d} |jjdkr4|j | d<| t j | WddS1sHwYdS)NiF)rjlingerbindREADY)statusrrwr]dp_stats_addressrOparallel_config_hash)rrqDEALERstartup_handshaker\r]r[rSdata_parallel_size compute_hashsendmsgspecmsgpackencode) rr{rCrjrBrwrBr|handshake_socketrkr]r ready_msgrrrrsLs6  "z!EngineCoreProc._perform_handshakerrScCs|tjd||dtd|jtdds"tdtd| }tjj |t d}td ||durH|j D] \}}t|||q=|jS) NHELLO)rrrwz(Waiting for init message from front-end.i`rQz7Did not receive response from front-end process within z minutesrzReceived init message: %s)rrrrrUrwpollHANDSHAKE_TIMEOUT_MINSrrecvdecoder1rSitemssetattrrk)rrBrwrS init_bytes init_messagekeyvaluerrrr{s0  z EngineCoreProc.startup_handshake)dp_rank local_dp_rankrrc sdtfdd}ttj|ttj|d}zz|d}|j}|jdkp,|dk}|rE||_tdd d |d td d |n tdd d d td t |rm|j durm|j j d||j _ t d|j j ||_|r|jjr||_t|i|}nd|_d|_d|_t|d|i|}|dusJ|Wn.tyt dty} z|durt d| t d|| d} ~ wwW|dur|dSdS|dur|ww)z2Launch EngineCore busy loop in background process.FcssdtdSr) SystemExit)signumframeshutdown_requestedrrsignal_handlersz6EngineCoreProc.run_engine_core..signal_handlerNrBrOrvllm.engine_core engine_core EngineCore_DPinstrumenting_module_name process_kind process_namerADP_dpz*Setting kv_transfer_config.engine_id to %srAEngineCore exiting.zEngineCore failed to start.%EngineCore encountered a fatal error.)rsignalSIGTERMSIGINTrSrrTrr rkv_transfer_config engine_idrUrwdata_parallel_indexrzis_moedata_parallel_rankDPEngineCoreProcdata_parallel_size_localr? run_busy_looprr exception_send_engine_deadr ) rrrKr+rrrBrS data_parallelerrrrun_engine_coresx          zEngineCoreProc.run_engine_corecCsdSrr)rrBrrrr^sz"EngineCoreProc._init_data_parallelcCs ||q)z!Core busy loop of the EngineCore.)_process_input_queue_process_engine_steprrrrrszEngineCoreProc.run_busy_loopcCsd}|jsS|jsS|jsS|jsS|jr;|jj|jj Wdn1s*wYt t r;t dd}|j}|j||jsS|jsS|jsS|jr|rZt d|jsp|j}|j||jr_dSdS)z0Exits when an engine step needs to be performed.FNzEngineCore waiting for work.TzEngineCore loop active.)rUrhrrvrrFrrmutexrclearrU isEnabledForr rwr_handle_client_requestr)rwaitedr2rrrrs@          z#EngineCoreProc._process_input_queuecCsT|\}}|r |ndD]}|j|q|||s(|jr(td|S)z5Called only when there are unfinished local requests.rgMbP?) rrrSrGrrhrrr)rrWroutputrrrr s   z#EngineCoreProc._process_engine_step request_typerc Cs|tjkr|\}}|||dS|tjkr||dS|tjkrq|\}}}}t|} zt||} | || |} t | | _ Wn#t yc} zt d|d|dt| | _WYd} ~ nd} ~ ww|j|t| dfdS|tjkrztdt d|dS)zDispatch request from client.zInvocation of %s method failedzCall to z method failed: N)utility_outputzExecutor failed.z/Unrecognized input request type encountered: %s)r+ADDrABORTrUTILITYr/r_convert_msgspec_argsr0r BaseExceptionrUrrfailure_messagerSrGr)rHrerror) rrrr2r client_idxcall_id method_namerKrr)rrrrrr4s4        z%EngineCoreProc._handle_client_requestcCsB|s|St|j}t|t|ksJtddt||DS)ztIf a provided arg type doesn't match corresponding target method arg type, try converting to msgspec object.cssL|]!\}}t|jr t|jtjr t||js tj||jdn|VqdS)rN)r annotation issubclassrStructrconvert)rvprrrr\s   z7EngineCoreProc._convert_msgspec_args..)r parametersvaluesrbr:zip)r)rK arg_typesrrrrTsz$EngineCoreProc._convert_msgspec_argscCs8|jtj|jjdd|jrtddSdS)z/Send EngineDead status to the EngineCoreClient.g@rQzNvLLM shutdown signal from EngineCore failed to send. Please report this issue.N) rSrGr?ENGINE_CORE_DEADrgrrirUfatalrrrrres z EngineCoreProc._send_engine_deadinput_addressescoord_input_addressrmc stt}t}ttfdd|D}|dur"d}nt|tjdd}|dt } |D]} | d| | tj q:|dur\| dksUJ| |tj | ~ | D]K\} } | jdd ^} } tt| j}|tjkr|| }z||}Wnty||Yqfw|| }|tjkr|j||j||fqfqb1swYWddS1swYdS) zInput socket IO thread.c s&g|]}t|tjddqS)Frjr~) enter_contextrrqr)r input_addressr{rjstackrrrs z8EngineCoreProc.process_input_sockets..NFrrEsREADYT)copy)r:r*rrqrrrrXSUBrPollerregisterPOLLINrr=rrecv_multipartr+rRbufferrrr3r_handle_request_preproc_errorrrrGrF)rrrrjrmadd_request_decodergeneric_decoder input_sockets coord_socketpoller input_socket_ type_frame data_framesrr2rrrrrcssZ          4z$EngineCoreProc.process_input_sockets output_pathscoord_output_pathc st}g}tttjttf}tՉtfdd|D}|dur4 t |tj dddnd}t |d} |j } | tjkrR|D]} | | qInwt| trYJ| \} } || _| d krs|dusjJ||| q<|r|d d jr||d |r|d d js||r|nt}|| |}|| j|ddd }|jst |dkr| nd}||||fn t || kr||q=Wdn1swYWddSWddS1swYdS) zOutput socket IO thread.c s$g|]}t|tjddqS))r})rrrqPUSH)r output_pathr{rrrrs z9EngineCoreProc.process_output_sockets..NFr)r~r}rOTrHrrJ)rtrack)r;rr:rqMessageTrackerr bytearrayrrrrrrrbrSrr?rrrrRrAsend_multipartrrappendr encode_intor)rrrrAencoder reuse_bufferspendingsocketsrmax_reuse_bufsrsocket client_indexrWrbufferstrackerrefrrrrfsZ            Pz%EngineCoreProc.process_output_socketsc CsDtd|j|j|jt|j|jht|jgt j dgdfdS)zLog and return a request-scoped error response for exceptions raised from the add request preprocessing in the input socket processing thread. z*Unexpected error pre-processing request %s)r new_token_ids finish_reason)rAfinished_requestsrWN) rUrrrSrGrr)rAr(r,ERROR)rrrrrr s"z,EngineCoreProc._handle_request_preproc_errorr)+r5r6r7r8rrrr9rrr4r;rrrRrr2rVrqrrrrs staticmethodSocketrrr^rrrr+r rrrrr`rarcrfr*r __classcell__rrrorr?s  _  :  . %U    Q Fr?cseZdZdZ d"dedededeedededBf fd d Z defd d Z fd dZ d#de de ffdd Zdededdffdd ZddZddZdedefddZdeddfd d!ZZS)$rzXZMQ-wrapper for running EngineCore in background process in a data parallel context.NrBrBrCrCrDrDc sH|jjsJdd|_d|_d|_|jj}tj|||||||ddS)Nz3DPEngineCoreProc should only be used for MoE modelsrrrr@) rzr step_counterr1 last_countsrSrr_r)rrBrBrCrCrDrDrrorrr's  zDPEngineCoreProc.__init__cCsl|jj}|jj}|jj}|dksJ|dusJd|kr(|kr(|ks+JJ||_|j|_dS)NrOr)rSrrrTrstateless_init_dp_groupr)rrBrdp_sizerrrrr^Fs  &z$DPEngineCoreProc._init_data_parallelcs*tt|dd}rt|dSdS)Nr)r_r rr)rrrorrr Ss  zDPEngineCoreProc.shutdownrrrcsR|jr ||jkr ||jkr||_n|js |jdt|jdft||dS)NrH) start_wave)rZr1rUrSrGr)r_r)rrrrorrrXs zDPEngineCoreProc.add_requestrrcsf|tjkr*|\}}||jkr$||jkr&||_|js(td|d|_dSdSdSdSt||dS)Nz*EngineCore starting idle loop for wave %d.T) r+ START_DP_WAVErAr1rUrUrwr_r)rrrnew_waveexclude_eng_indexrorrres     z'DPEngineCoreProc._handle_client_requestcCsV|jsdS|j}||jkr)||_t||j|jd}|jdt |dfdSdS)N)rr1rH)scheduler_stats) r]rhget_request_countsrr6rr1rSrGr))rcountsstatsrrr_maybe_publish_request_countsts   z.DPEngineCoreProc._maybe_publish_request_countscCs ||}||j}|s|s|jsq||||_|jsS|jdks/|j sIt d|j |j r;dnd}|j |t|j df|j d7_ d|_q)z8Core busy loop of the EngineCore for data parallel case.Trz&Wave %d finished, pausing engine loop.rH) wave_completerO)rrrrhrrUr_has_global_unfinished_reqsrrZrUrwr1rSrGr)r)rexecutedlocal_unfinished_reqsrrrrrs4   zDPEngineCoreProc.run_busy_looplocal_unfinishedcCs.|jd7_|jddkrdSt|j|S)NrO rT)rrhas_unfinished_dpr)rr$rrrr!sz,DPEngineCoreProc._has_global_unfinished_reqsreconfig_requestcCst|j||jj}|j}|j|_|jdkr|j|_|j t j ks%J|j |_ |j|_|jdkr;|j|_||_|j|_|j||j|kr_|jdksQJt|j|j|jd|jt jkrr|td|jdStd|jdS)NrHrcompile_or_warm_up_modelzDPEngineCoreProc %s shutdownz4Distributed environment reinitialized for DP rank %s)rrr rBrSrnew_data_parallel_sizenew_data_parallel_rankrnew_data_parallel_rank_localr.KEEP_CURRENT_RANKnew_data_parallel_master_ipdata_parallel_master_ipnew_data_parallel_master_portdata_parallel_master_portrrrXreinitialize_distributedrZrrr_SHUTDOWN_CURRENT_RANKrUrV)rr'rS old_dp_sizerrrr2sD       z)DPEngineCoreProc.reinitialize_distributedrr4)r5r6r7r8rr9rrr4rr^r r8r;rr+r rrrr!r-r2rrrrorr#sD    /rc @seZdZdZ  ddedededefddZdedefd d Zdeded e fd d Z e de de de dede dBf ddZddZddZdS)EngineCoreActorMixinzE Ray actor for running EngineCore in a data parallel context rrBrkrrcCs:tddd|d||_||j_||j_|||dS)NrrDPEngineCoreActor_DPr)rrkrSrrT_set_visible_devices)rrBrkrrrrrrszEngineCoreActorMixin.__init__cCs0ddlm}|r dS|j}||||dS)Nr)current_platform)vllm.platformsr8is_xpudevice_control_env_var_set_cuda_visible_devices)rrBrr8r;rrrr7s z)EngineCoreActorMixin._set_visible_devicesr;cCst|jj}zt|||}|tj|<WdSty9}ztd|d||d|d|dt|d |d}~ww)NzError setting z: local range: [z, rOz) base value: "")rS world_sizer3rr IndexErrorrgetenv)rrBrr;r>rrrrrr<"s( z.EngineCoreActorMixin._set_cuda_visible_devicesrCrjrBrDNccs|jVdS)z For Ray, we don't need to actually perform handshake. All addresses information is known before the actor creation. Therefore, we simply yield these addresses. N)rk)rrCrjrBrBrDrrrrV4s z(EngineCoreActorMixin._perform_handshakescCsdS)a Wait until the engine core is initialized. This is just an empty method. When ray.get() on this method (or any other method of the actor) returns, it is guaranteed that actor creation (i.e., __init__) is complete. Nrrrrr wait_for_initDsz"EngineCoreActorMixin.wait_for_initcCsZz'z|Wntytdty tdwW|dS|w)z0 Run the engine core busy loop. rrN)rrrUrwrrr rrrrrunNs     zEngineCoreActorMixin.runr)r5r6r7r8rr2r;rr7rr<rrRr9rVrArBrrrrr5sF %    r5c@@eZdZdZ  d dedededeedede d e fd d Z d S)DPMoEEngineCoreActorz'Used for MoE model data parallel cases.rrBrBrkrCrDrrcCs2||j_t|||||t|||d||dS)Nr)rSrr5rrrrBrBrkrCrDrrrrrras   zDPMoEEngineCoreActor.__init__Nr r5r6r7r8rr9r2rr4r;rrrrrrD^& rDc@rC)EngineCoreActorz%Used for non-MoE and/or non-DP cases.rrBrBrkrCrDrrc CsFd|j_d|j_d|j_t|||||tj|||d|||ddS)NrOrrr@)rSrrrr5rr?rErrrrxs   zEngineCoreActor.__init__NrrFrrrrrHurGrH)srrrr`r collectionsrcollections.abcrrconcurrent.futuresr contextlibrrinspectrr loggingr typingr r r rrq vllm.configrrvllm.distributedr vllm.envsr vllm.loggerrvllm.logging_utils.dump_inputrvllm.lora.requestrvllm.multimodalr vllm.tasksrr vllm.tracingrrvllm.transformers_utils.configrvllm.utils.gc_utilsrrvllm.utils.hashingrvllm.utils.network_utilsrvllm.utils.system_utilsrr vllm.v1.core.kv_cache_utilsr!r"r#r$r%vllm.v1.core.sched.interfacer&vllm.v1.core.sched.outputr'vllm.v1.enginer(r)r*r+r,r-r.r/r0vllm.v1.engine.utilsr1r2r3vllm.v1.executorr4vllm.v1.kv_cache_interfacer5vllm.v1.metrics.statsr6vllm.v1.outputsr7vllm.v1.requestr8r9vllm.v1.serial_utilsr:r;vllm.v1.structured_outputr< vllm.v1.utilsr= vllm.versionr>rWr5rUPOLLING_TIMEOUT_Srr@rAr?rr5rDrHrrrrsz              ,       a{Kq