o ۾i= @s>ddlZddlZddlZddlZddlZddlmZmZddlmZddl m Z ddlm Z ddl m Z ddlmZmZmZmZmZmZddlZddlmZddlmZdd lmZdd lmZmZmZdd l m!Z!m"Z"m#Z#dd l$m%Z%dd l&m'Z'erddl(Z)ddl*m+Z+ddl,m-Z-m.Z.ee/Z0edZ1Gdddee1eZ2GdddZ3d:de4de5de6de5fddZ7GdddZ8  d;de8deddBd d!ddfd"d#Z9d$e:e fd%d&Z;d'ejfd,eddfd-d.Z?da@d/e5defd0d1ZAd2ejr.zConstantList.clearrr,startstopcCs$|j|||dur |St|jSr )r"indexlen)r#r,r5r6r$r$r%r7As$zConstantList.indexcCdSr r$r+r$r$r% __getitem__DzConstantList.__getitem__scCr9r r$)r#r<r$r$r%r:Gr;cCs |j|Sr r!r+r$r$r%r:Jr'valuecCr9r r$r#r,r=r$r$r% __setitem__Mr;zConstantList.__setitem__cCr9r r$)r#r<r=r$r$r%r?Pr;cCr()Nz"Cannot set item in a constant listr)r>r$r$r%r?Sr.cCr()Nz'Cannot delete item from a constant listr)r+r$r$r% __delitem__Vr.zConstantList.__delitem__cC t|jSr )iterr"r3r$r$r%__iter__Yr'zConstantList.__iter__cCs ||jvSr r!r+r$r$r% __contains__\r'zConstantList.__contains__cCrAr )r8r"r3r$r$r%__len___r'zConstantList.__len__cCsd|jdS)Nz ConstantList()r!r3r$r$r%__repr__bszConstantList.__repr__cCs |jSr )r"copyr3r$r$r%rHer'zConstantList.copy)rN)__name__ __module__ __qualname__listrr&r-r/r0r1r2r4intr7r r:slicer?r@rCrDrErGrHr$r$r$r%r+s2 rc @sveZdZdZdddeejBdejdejde de d d f d d Z dd ed Bd ej fddZ dd ed Bd ej fddZ d S) CpuGpuBufferz2Buffer to easily copy tensors between CPU and GPU.T) with_numpysizedtypedevice pin_memoryrPrNcGsVtj||d|d|_tj|j|d|_||r)|tjkr!td|j|_dSdS)Ncpu)rRrSrT)rSzkBfloat16 torch tensors cannot be directly cast to a numpy array, so call CpuGpuBuffer with with_numpy=False) torchzerosrU zeros_likegpubfloat16 ValueErrornumpynp)r#rRrSrTrPrQr$r$r%r&ls zCpuGpuBuffer.__init__ncCs<|dur |jj|jddS|jd|j|jd|ddS)NT non_blocking)rYcopy_rUr#r^r$r$r% copy_to_gpus"zCpuGpuBuffer.copy_to_gpucCs<|dur |jj|jddS|jd|j|jd|ddS)zzNOTE: Because this method is non-blocking, explicit synchronization is needed to ensure the data is copied to CPU.NTr_)rUrarYrbr$r$r% copy_to_cpus"zCpuGpuBuffer.copy_to_cpur )rIrJrK__doc__rMrVSymIntrRrSboolr&Tensorrcrdr$r$r$r%rOis$ rO local_onlyhostportrcCs|rtSt||p tS)a Assign a new ZMQ socket address. If local_only is True, participants are colocated and so a unique IPC address will be returned. Otherwise, the provided host and port will be used to construct a TCP address (port == 0 means assign an available port).)rrr)rirjrkr$r$r%get_engine_client_zmq_addrs  rlc@sVeZdZdZ ddedededejde de ed e ed edBfd d Z dddZ dS)APIServerProcessManagerzManages a group of API server processes. Handles creation, monitoring, and termination of API server worker processes. Also monitors extra processes to check if they are healthy. Ntarget_server_fnlisten_addresssockargs num_serversinput_addressesoutput_addressesstats_update_addressc Cs||_||_||_td} g|_tt|||D]-\} } } | | || d} |dur-|| d<| j|d| |||| fd}|j || qt dt |jt|t|j|_dS)a7Initialize and start API server worker processes. Args: target_server_fn: Function to call for each API server process listen_address: Address to listen for client connections sock: Socket for client connections args: Command line arguments num_servers: Number of API server processes to start input_addresses: Input addresses for each API server output_addresses: Output addresses for each API server stats_update_address: Optional stats update address spawn) input_addressoutput_address client_count client_indexNru ApiServer_)targetnamerqzStarted %d API server processes)rorprqmultiprocessing get_context processesziprangeProcessr-r5loggerinfor8weakreffinalizeshutdown _finalizer)r#rnrorprqrrrsrtru spawn_contextiin_addrout_addr client_configprocr$r$r%r&s0     z APIServerProcessManager.__init__rcCs |dSr )rr3r$r$r%closes zAPIServerProcessManager.closer )rN) rIrJrKrerstrr argparse NamespacerMrLr&rr$r$r$r%rms* 9rmapi_server_managerengine_manager)rr coordinatorzDPCoordinator | Nonec Csddlm}m}zzqtddd|jD}|r |j||jj<g}t||r3|jD]}|||j<q*n t||r<| }|s@|ryt j |dd}|D]} | | }|j dkretd|jd |jd |j qI|rudd l} | j |dd\} }|s@|s@Wn"tytd Ynty} z td t| d } ~ wwWtd||r||r|d Sd Std||r||r|ww)aWait for all processes to complete or detect if any fail. Raises an exception if any process exits with a non-zero status. Args: api_server_manager: The manager for API servers. engine_manager: The manager for engine processes. If CoreEngineProcManager, it manages local engines; if CoreEngineActorManager, it manages all engines. coordinator: The coordinator for data parallel. rrz'Waiting for API servers to complete ...cSsi|]}|j|qSr$)sentinel).0rr$r$r% sz2wait_for_completion_or_failure..)timeoutzProcess z (PID: z) died with exit code Nz8Received KeyboardInterrupt, shutting down API servers...z0Exception occurred while running API servers: %sz#Terminating remaining processes ...)vllm.v1.engine.utilsrrrrrrr isinstance get_run_refsrwaitr1exitcode RuntimeErrorr}pidrayKeyboardInterrupt Exception exceptionrr) rrrrrsentinel_to_procactor_run_refsrready_sentinelsrr_er$r$r%wait_for_completion_or_failuresj            rprocscCs|D] }|r |qtd}|D]}|t}|dkr#n |r,||q|D]}|r@|j}dur@t|q/dS)Nrr)is_alive terminatetime monotonicjoinrr)rrdeadline remainingrr$r$r%r.s"   r from_tensor to_tensorlengthcCs|d|j|d|ddS)z Copy the first length elements of a tensor into another tensor in a non-blocking manner. Used to copy pinned CPU tensor data to pre-allocated GPU tensors. Returns the sliced target tensor. NTr_)ra)rrrr$r$r% copy_sliceBs r usage_contextcCstsdSddlm}|j}d}|jdur|jj}tj||j|idt |jj d|j j d|j j d|j jd|jjd t |j jd t|jd |j jd |jjd |jd|jd|jd|jd|jd|jd|ddS)z#Report usage statistics if enabled.Nr)get_architecture_class_namerR block_sizegpu_memory_utilizationkv_cache_memory_bytes quantizationkv_cache_dtype enable_loraenable_prefix_caching enforce_eagerdisable_custom_all_reducetensor_parallel_sizedata_parallel_sizepipeline_parallel_sizeenable_expert_parallelall2all_backend kv_connector) extra_kvs)r vllm.model_executor.model_loaderrparallel_configkv_transfer_configrr report_usage model_configrrR cache_configrrrr cache_dtyperg lora_configrrrrrrrr) vllm_configrrrrr$r$r%report_usage_statsPsX         rr}cCsBtdurt|Stj}tjrt}n tjrddl}|j}|a||S)Nr) _PROFILER_FUNC contextlib nullcontextenvs VLLM_CUSTOM_SCOPES_FOR_PROFILINGrVLLM_NVTX_SCOPES_FOR_PROFILINGnvtxannotate)r}funcrr$r$r%record_function_or_nullcontextsrtensorcCs|tjjS)zGet the raw data of a tensor as a uint8 memoryview, useful for serializing and hashing. Args: tensor: The input tensor. Returns: A memoryview of the tensor data as uint8. )flatten contiguousviewrVuint8r\data)rr$r$r% tensor_datas rc@s<eZdZUeed<eed<eed<eed<defddZdS) IterationDetailsnum_ctx_requestsnum_ctx_tokensnum_generation_requestsnum_generation_tokensrc Cs&d|jd|jd|jd|jd S)Nz"IterationDetails(num_ctx_requests=z!, num_ctx_tokens=z+, num_generation_requests=z), num_generation_tokens=rF)rrrrr3r$r$r%rGs zIterationDetails.__repr__N)rIrJrKrM__annotations__rrGr$r$r$r%rs rscheduler_outputcCsxd}d}d}d}dd|jD}|jD]\}}|j|s#||vr,|d7}||7}q|d7}||7}qt||||S)a Compute the number of context/generation requests and tokens for the current iteration's scheduler output. A requests is regarded as a context request if its output tokens are still 0, an extended chunk of chunked prefill falls into this category. Args: scheduler_output: The scheduler output for the current iteration. Returns: An IterationDetails object containing the number of context/generation requests and tokens. rcSsh|]}|jqSr$)req_id)rnew_reqr$r$r% sz,compute_iteration_details..)scheduled_new_reqsnum_scheduled_tokensitemsscheduled_cached_reqsis_context_phaser)rnum_context_requestsnum_context_tokensrr new_req_idsr num_tokensr$r$r%compute_iteration_detailss$   r)r)NN)Frrr~rrcollections.abcrrr dataclassesrrmultiprocessing.processrtypingrr r r r r rVtorch.autograd.profilerr vllm.envsr vllm.loggerrvllm.usage.usage_librrrvllm.utils.network_utilsrrrvllm.utils.system_utilsrvllm.v1.core.sched.outputrr\r]vllm.v1.engine.coordinatorrrrrrIrrrrOrgrrMrlrmrrLrrhrENGINE_CONTEXTrrr memoryviewrrrr$r$r$r%s           >&G K  /