o .i= @sFddlZddlZddlZddlZddlZddlmZmZddlmZddl m Z ddlm Z ddl m Z ddlmZmZmZmZmZmZmZddlZddlmZddlmZdd lmZdd lmZmZm Z dd 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/ee0Z1edZ2Gdddee2eZ3GdddZ4d:de5de6de7de6fddZ8GdddZ9  d;de9deddBd ed!ddfd"d#Z:d$e;e fd%d&Zej?fd,eddfd-d.Z@daAd/e6defd0d1ZBd2ej=deCfd3d4ZDe Gd5d6d6ZEd7e(deEfd8d9ZFdS)<N)CallableSequence)AbstractContextManager) dataclass) connection) BaseProcess) TYPE_CHECKINGAnyGenericOptionalTypeVarUnionoverload)record_function) init_logger) UsageContextis_usage_stats_enabled usage_message) get_open_portget_open_zmq_ipc_path get_tcp_uri)kill_process_tree)SchedulerOutput) DPCoordinatorCoreEngineActorManagerCoreEngineProcManagerTc @sDeZdZdeeddfddZddZdd Zd d Zd d Z ddZ ddZ d.dede de dBde fddZ ede defddZededeefddZde eBdeeeBfddZede defddZededefd dZde eBdeeeBfd!dZd"d#Zd$d%Zd&d'Zd(d)Zd*d+Zdeefd,d-ZdS)/ ConstantListxreturnNcCs ||_dSN_x)selfrr%J/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/vllm/v1/utils.py__init__- zConstantList.__init__cCtd)Nz Cannot append to a constant list TypeErrorr$itemr%r%r&append0zConstantList.appendcCr))NzCannot extend a constant listr*r,r%r%r&extend3r/zConstantList.extendcCr))Nz"Cannot insert into a constant listr*r,r%r%r&insert6r/zConstantList.insertcCr))NzCannot pop from a constant listr*r,r%r%r&pop9r/zConstantList.popcCr))Nz"Cannot remove from a constant listr*r,r%r%r&remove<r/zConstantList.removecCr))NzCannot clear a constant listr*r$r%r%r&clear?r/zConstantList.clearrr-startstopcCs$|j|||dur |St|jSr!)r#indexlen)r$r-r6r7r%r%r&r8Bs$zConstantList.indexcCdSr!r%r,r%r%r& __getitem__EzConstantList.__getitem__scCr:r!r%)r$r=r%r%r&r;Hr<cCs |j|Sr!r"r,r%r%r&r;Kr(valuecCr:r!r%r$r-r>r%r%r& __setitem__Nr<zConstantList.__setitem__cCr:r!r%)r$r=r>r%r%r&r@Qr<cCr))Nz"Cannot set item in a constant listr*r?r%r%r&r@Tr/cCr))Nz'Cannot delete item from a constant listr*r,r%r%r& __delitem__Wr/zConstantList.__delitem__cC t|jSr!)iterr#r4r%r%r&__iter__Zr(zConstantList.__iter__cCs ||jvSr!r"r,r%r%r& __contains__]r(zConstantList.__contains__cCrBr!)r9r#r4r%r%r&__len__`r(zConstantList.__len__cCsd|jdS)Nz ConstantList()r"r4r%r%r&__repr__cszConstantList.__repr__cCs |jSr!)r#copyr4r%r%r&rIfr(zConstantList.copy)rN)__name__ __module__ __qualname__listrr'r.r0r1r2r3r5intr8rr;slicer@rArDrErFrHrIr%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_memoryrQr NcGsVtj||d|d|_tj|j|d|_||r)|tjkr!td|j|_dSdS)Ncpu)rSrTrU)rTzkBfloat16 torch tensors cannot be directly cast to a numpy array, so call CpuGpuBuffer with with_numpy=False) torchzerosrV zeros_likegpubfloat16 ValueErrornumpynp)r$rSrTrUrQrRr%r%r&r'ms zCpuGpuBuffer.__init__ncCs<|dur |jj|jddS|jd|j|jd|ddS)NT non_blocking)rZcopy_rVr$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`)rVrbrZrcr%r%r& copy_to_cpus"zCpuGpuBuffer.copy_to_cpur!)rJrKrL__doc__rNrWSymIntrSrTboolr'Tensorrdrer%r%r%r&rPjs$ rP local_onlyhostportr cCs|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)rjrkrlr%r%r&get_engine_client_zmq_addrs  rmc@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_indexNrv ApiServer_)targetnamerrzStarted %d API server processes)rprqrrmultiprocessing get_context processesziprangeProcessr.r6loggerinfor9weakreffinalizeshutdown _finalizer)r$rorprqrrrsrtrurv spawn_contextiin_addrout_addr client_configprocr%r%r&r's0     z APIServerProcessManager.__init__r cCs |dSr!)rr4r%r%r&closes zAPIServerProcessManager.closer!)r N) rJrKrLrfrstrr argparse NamespacerNrMr'rr%r%r%r&rns* 9rnapi_server_managerengine_manager)rr coordinatorrc 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_refsrwaitr2exitcode 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`)rb)rrrr%r%r& copy_sliceCs 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_namerS 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_configrrS cache_configrrrr cache_dtyperh lora_configrrrrrrrr) vllm_configrrrrr%r%r&report_usage_statsQsX         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 contiguousviewrWuint8r]data)rr%r%r& tensor_datas rc@s<eZdZUeed<eed<eed<eed<defddZdS) IterationDetailsnum_ctx_requestsnum_ctx_tokensnum_generation_requestsnum_generation_tokensr c Cs&d|jd|jd|jd|jd S)Nz"IterationDetails(num_ctx_requests=z!, num_ctx_tokens=z+, num_generation_requests=z), num_generation_tokens=rG)rrrrr4r%r%r&rHs zIterationDetails.__repr__N)rJrKrLrN__annotations__rrHr%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)Grrrrrcollections.abcrrr dataclassesrrmultiprocessing.processrtypingrr r r r r rrWtorch.autograd.profilerr vllm.envsr vllm.loggerrvllm.usage.usage_librrrvllm.utils.network_utilsrrrvllm.utils.system_utilsrvllm.v1.core.sched.outputrr]r^vllm.v1.engine.coordinatorrrrrrJrrrrPrhrrNrmrnrrMrrirENGINE_CONTEXTrrr memoryviewrrrr%r%r%r&s    $      >&G K  /