o ۾iM6@s ddlmZddlmZmZmZddlZddlmZddl m Z m Z ddl m Z ddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZeredd lmZmZddlm Z m!Z!ne"Ze"Ze"Z e"Z!e e#Z$edZ%GdddZ&GdddZ'dS))Callable) TYPE_CHECKINGAnyTypeVarN) VllmConfigset_current_vllm_config) init_logger) LoRARequest)MULTIMODAL_REGISTRY) instrument)resolve_obj_by_qualname)update_environment_variables) KVCacheSpec) run_method) GrammarOutputSchedulerOutput)AsyncModelRunnerOutputModelRunnerOutput_Rc @sreZdZdZ d9dedededededd f d d Zde ee ffd d Z d:ddZ d:ddZ d:ddZdededd fddZd:ddZdejfddZdeejgefdefddZdefdd Zd:d!d"Zd#edeeBd Bfd$d%Zd&edeeBfd'd(Zdefd)d*Zd+e defd,d-Z!d.edefd/d0Z"d.edefd1d2Z#de$efd3d4Z%e&defd5d6Z'd:d7d8Z(d S); WorkerBasezWorker interface that allows vLLM to cleanly separate implementations for different hardware. Also abstracts control plane communication, e.g., to communicate request metadata to other workers. F vllm_config local_rankrankdistributed_init_methodis_driver_workerreturnNcCs||_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j |_ |j |_ |j |_ ddl m }||_ ||j_||_||_||_||_d|_d|_dS)a Initialize common worker components. Args: vllm_config: Complete vLLM configuration local_rank: Local device index rank: Global rank in distributed setup distributed_init_method: Distributed initialization method is_driver_worker: Whether this worker handles driver responsibilities r)current_platformN)r model_config cache_config lora_config load_configparallel_configscheduler_config device_configspeculative_configobservability_configkv_transfer_configcompilation_configvllm.platformsrrrrrdevice model_runner)selfrrrrrrr,N/home/ubuntu/.local/lib/python3.10/site-packages/vllm/v1/worker/worker_base.py__init__(s*  zWorkerBase.__init__cCt)z/Get specifications for KV cache implementation.NotImplementedErrorr+r,r,r-get_kv_cache_specVzWorkerBase.get_kv_cache_speccCr/)z7Prepare model for execution through compilation/warmup.r0r2r,r,r-compile_or_warm_up_modelZr4z#WorkerBase.compile_or_warm_up_modelcCdS)z9Basic health check (override for device-specific checks).Nr,r2r,r,r- check_health^r4zWorkerBase.check_healthcCr/)zjInitialize device state, such as loading the model or other on-device memory allocations. r0r2r,r,r- init_devicebzWorkerBase.init_devicenum_gpu_blocksnum_cpu_blockscCr/)z6Initialize the KV cache with the given size in blocks.r0)r+r:r;r,r,r-initialize_cachehr4zWorkerBase.initialize_cachecCs$t|jdd}t|r|dSdS)Nreset_mm_cache)getattrr*callable)r+reset_fnr,r,r-r=ls zWorkerBase.reset_mm_cachecCr/Nr0r2r,r,r- get_modelqzWorkerBase.get_modelfncCs ||S)z1Apply a function on the model inside this worker.)rB)r+rDr,r,r- apply_modelts zWorkerBase.apply_modelcCsddlm}||S)z;Return a transformers-style hierarchical view of the model.r)format_model_inspection)vllm.model_inspectionrFrB)r+rFr,r,r-get_model_inspectionxs  zWorkerBase.get_model_inspectioncCr/)zLoad model onto target device.r0r2r,r,r- load_model~r4zWorkerBase.load_modelscheduler_outputcCr/)zIf this method returns None, sample_tokens should be called immediately after to obtain the ModelRunnerOutput. Note that this design may be changed in future if/when structured outputs parallelism is re-architected. r0r+rJr,r,r- execute_models zWorkerBase.execute_modelgrammar_outputcCr/)zFShould be called immediately after execute_model iff it returned None.r0)r+rMr,r,r- sample_tokensr9zWorkerBase.sample_tokenscCr/)zaReturn the size of a single cache block, in bytes. Used in speculative decoding. r0r2r,r,r-get_cache_block_size_bytesr9z%WorkerBase.get_cache_block_size_bytes lora_requestcCr/rAr0)r+rPr,r,r-add_lorarCzWorkerBase.add_loralora_idcCr/rAr0r+rRr,r,r- remove_lorarCzWorkerBase.remove_loracCr/rAr0rSr,r,r-pin_lorarCzWorkerBase.pin_loracCr/rAr0r2r,r,r- list_lorasrCzWorkerBase.list_lorascCs |jS)z-Get vocabulary size from model configuration.)rget_vocab_sizer2r,r,r- vocab_sizes zWorkerBase.vocab_sizecCr6)z&Clean up resources held by the worker.Nr,r2r,r,r-shutdownr4zWorkerBase.shutdown)FrN))__name__ __module__ __qualname____doc__rintstrboolr.dictrr3r5r7r8r<r=nnModulerBrrrErHrIrrrrLrrNrOr rQrTrUsetrVpropertyrXrYr,r,r,r-r"sX  .         rc@seZdZdZ  d(dededBddfddZd)d d Zd eeefddfd d Zde ee e fddfddZ e ddde ee e fddfddZde e ddfddZddZde eBfddZde fdd Zd!eddfd"d#Zd!edeeBdBfd$d%Zd)d&d'ZdS)*WorkerWrapperBaseao This class represents one process in an executor/engine. It is responsible for lazily initializing the worker and handling the worker's lifecycle. We first instantiate the WorkerWrapper, which remembers the worker module and class name. Then, when we call `update_environment_variables`, and the real initialization happens in `init_worker`. rNrpc_rank global_rankrcCs&||_|dur |jn||_||dS)a# Initialize the worker wrapper with the given vllm_config and rpc_rank. Note: rpc_rank is the rank of the worker in the executor. In most cases, it is also the rank of the worker in the distributed group. However, when multiple executors work together, they can be different. e.g. in the case of SPMD-style offline inference with TP=2, users can launch 2 engines/executors, each with only 1 worker. All workers have rpc_rank=0, but they have different ranks in the TP group. N)rhri)r+rhrir,r,r-r.szWorkerWrapperBase.__init__cCs|jdur |jdSdSrA)workerrYr2r,r,r-rYs zWorkerWrapperBase.shutdown rank_mappingcCs|j|vr ||j|_dSdS)z Adjust the rpc_rank based on the given mapping. It is only used during the initialization of the executor, to adjust the rpc_rank of workers after we create all workers. N)rh)r+rkr,r,r- adjust_ranks zWorkerWrapperBase.adjust_rank envs_listcCs||j}t|dSrA)rhr )r+rmenvsr,r,r-r s  z.WorkerWrapperBase.update_environment_variablesz Worker init) span_name all_kwargsc Cs||j}|d}|dusJd||_|ddlm}||j}t|jt r1t |j}nt d|j rt |j }g}||j vrt|D]'} | drPqHt|| rcJd|d | d |d tt|| ro|| qH|j |f|_ td ||||d d} | durd} |jj} | r| jdkrt | t| d|_nt|| |_t|j|di||_WddS1swYdS)z Here we inject some common logic before initializing the worker. Arguments are passed to the worker class constructor. rNz0vllm_config is required to initialize the workerr)load_general_pluginszpassing worker_cls is no longer supported. Please pass keep the class in a separate module and pass the qualified name of the class as a string.__z Worker class z already has an attribute z2, which conflicts with the worker extension class .z8Injected %s into %s for extended collective_rpc calls %sshared_worker_lockzoMissing `shared_worker_lock` argument from executor. This argument is needed for mm_processor_cache_type='shm'.shmr,) rhgetr%enable_trace_function_call_for_thread vllm.pluginsrqr! isinstance worker_clsr`r ValueErrorworker_extension_cls __bases__dir startswithhasattrr?r>appendloggerinfopoprmultimodal_configmm_processor_cache_type warning_oncemm_receiver_cacher !worker_receiver_cache_from_configrrj) r+rpkwargsrrqr! worker_classr|extended_callsattrrtmsg mm_configr,r,r- init_workers|             "zWorkerWrapperBase.init_workerkv_cache_configscCsT||j}|jdus Jt|j|j|WddS1s#wYdSrA)rirrrjinitialize_from_config)r+rkv_cache_configr,r,r-r8s  "z(WorkerWrapperBase.initialize_from_configcCsH|jdusJt|j|jWddS1swYdSrA)rrrjr8r2r,r,r-r8>s  "zWorkerWrapperBase.init_devicemethodc OsDzt||||WSty!}z d|d}t||d}~ww)NzError executing method z5. This might cause deadlock in distributed execution.)r Exceptionr exception)r+rargsrerr,r,r-execute_methodDs  z WorkerWrapperBase.execute_methodrcCs t|j|SrA)r>rj)r+rr,r,r- __getattr__Ws zWorkerWrapperBase.__getattr__rJcCs0|j}|dur dS|jD] }||j|_q dSrA)rscheduled_new_reqsget_and_update_features mm_features)r+rJmm_cachereq_datar,r,r-_apply_mm_cacheZs z!WorkerWrapperBase._apply_mm_cachecCs|||j|SrA)rrjrLrKr,r,r-rLds  zWorkerWrapperBase.execute_modelcCs$|j}|dur ||jdSrA)r clear_cacherjr=)r+rr,r,r-r=ksz WorkerWrapperBase.reset_mm_cache)rNrZ)r[r\r]r^r_r.rYrbrllistr`r r rrrr8bytesrrrrrrrLr=r,r,r,r-rgs>     U   rg)(collections.abcrtypingrrrtorchtorch.nnrc vllm.configrr vllm.loggerrvllm.lora.requestr vllm.multimodalr vllm.tracingr vllm.utils.import_utilsr vllm.utils.system_utilsr vllm.v1.kv_cache_interfacervllm.v1.serial_utilsrvllm.v1.core.sched.outputrrvllm.v1.outputsrrobjectr[rrrrgr,r,r,r-s2