| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import enum |
| 5 | import time |
| 6 | from collections.abc import Mapping |
| 7 | from dataclasses import dataclass |
| 8 | from typing import Any, Literal |
| 9 | |
| 10 | import msgspec |
| 11 | import numpy as np |
| 12 | import torch |
| 13 | |
| 14 | from vllm.config.kv_events import KVEventsConfig |
| 15 | from vllm.lora.request import LoRARequest |
| 16 | from vllm.multimodal.inputs import MultiModalFeatureSpec |
| 17 | from vllm.pooling_params import PoolingParams |
| 18 | from vllm.sampling_params import SamplingParams |
| 19 | from vllm.v1.kv_hints import KvHintsEnvelope |
| 20 | from vllm.v1.metrics.stats import ( |
| 21 | PrefillStats, |
| 22 | RequestSpecDecodeMetrics, |
| 23 | SchedulerStats, |
| 24 | ) |
| 25 | from vllm.v1.outputs import LogprobsLists, LogprobsTensors, SamplingMaskLists |
| 26 | from vllm.v1.serial_utils import UtilityResult |
| 27 | |
| 28 | # Type for pause_generation mode parameter. |
| 29 | # - "abort": Abort all in-flight requests immediately (default). |
| 30 | # - "wait": Wait for in-flight requests to complete before pausing. |
| 31 | # - "keep": Freeze requests in queue; they resume on resume_generation(). |
| 32 | PauseMode = Literal["abort", "wait", "keep"] |
| 33 | |
| 34 | # These are possible values of RequestOutput.finish_reason, |
| 35 | # so form part of the external API. |
| 36 | FINISH_REASON_STRINGS = ("stop", "length", "abort", "error", "repetition") |
| 37 | |
| 38 | EEP_NOTIFICATION_CALL_ID = -1 |
| 39 | |
| 40 | FT_STATUS_CALL_ID = -2 |
| 41 | |
| 42 | |
| 43 | class EEPNotificationType(enum.Enum): |
| 44 | RECONFIGURE_FINISHED = "RECONFIGURE_FINISHED" |
| 45 | SHUTDOWN_COMPLETE = "SHUTDOWN_COMPLETE" |
| 46 | |
| 47 | |
| 48 | class FinishReason(enum.IntEnum): |
| 49 | """Reason a request finished - stop, length, abort, error, or repetition. |
| 50 | |
| 51 | Int rather than Str for more compact serialization. |
| 52 | |
| 53 | stop - a stop string was emitted |
| 54 | length - max_tokens was consumed, or max_model_len was reached |
| 55 | abort - aborted by client |
| 56 | error - retryable request-level internal error (e.g., KV load failure). |
| 57 | Invariant: always converted to 500 Internal Server Error. |
| 58 | repetition - repetitive token pattern detected (hallucination) |
| 59 | |
| 60 | """ |
| 61 | |
| 62 | STOP = 0 |
| 63 | LENGTH = 1 |
| 64 | ABORT = 2 |
| 65 | ERROR = 3 |
| 66 | REPETITION = 4 |
| 67 | |
| 68 | def __str__(self): |
| 69 | return FINISH_REASON_STRINGS[self.value] |
| 70 | |
| 71 | |
| 72 | @dataclass |
| 73 | class EngineCoreReadyResponse: |
| 74 | """Sent from EngineCore to each frontend at the end of engine startup. |
| 75 | |
| 76 | Contains post-initialization config that may differ from the original |
| 77 | values (e.g. max_model_len after KV cache auto-fitting). |
| 78 | """ |
| 79 | |
| 80 | max_model_len: int |
| 81 | num_gpu_blocks: int |
| 82 | block_size: int |
| 83 | dp_stats_address: str | None |
| 84 | dtype: str |
| 85 | vllm_version: str |
| 86 | world_size: int |
| 87 | data_parallel_size: int |
| 88 | tensor_parallel_size: int |
| 89 | pipeline_parallel_size: int |
| 90 | decode_context_parallel_size: int |
| 91 | data_parallel_rank: int |
| 92 | max_num_seqs: int |
| 93 | max_num_batched_tokens: int |
| 94 | instance_id: str |
| 95 | supports_lora: bool |
| 96 | max_loras: int |
| 97 | mamba_block_size: int | None = None |
| 98 | # KV cache capacity (None for encoder-only/attention-free models). |
| 99 | kv_cache_size_tokens: int | None = None |
| 100 | kv_cache_max_concurrency: float | None = None |
| 101 | kv_events_config: KVEventsConfig | None = None |
| 102 | weight_transfer_backend: str | None = None |
| 103 | enable_sleep_mode: bool = False |
| 104 | supports_draft_weight_updates: bool = False |
| 105 | # Full-attention block size in tokens after initialization, or unavailable. |
| 106 | effective_attention_block_size: int | None = None |
| 107 | |
| 108 | |
| 109 | class EngineCoreRequest( |
| 110 | msgspec.Struct, |
| 111 | array_like=True, # type: ignore[call-arg] |
| 112 | omit_defaults=True, # type: ignore[call-arg] |
| 113 | gc=False, |
| 114 | ): # type: ignore[call-arg] |
| 115 | request_id: str |
| 116 | prompt_token_ids: list[int] | None |
| 117 | mm_features: list[MultiModalFeatureSpec] | None |
| 118 | sampling_params: SamplingParams | None |
| 119 | pooling_params: PoolingParams | None |
| 120 | arrival_time: float |
| 121 | lora_request: LoRARequest | None |
| 122 | cache_salt: str | None |
| 123 | data_parallel_rank: int | None |
| 124 | prompt_embeds: torch.Tensor | None = None |
| 125 | |
| 126 | # Per-position mask for mixed-mode inputs (e.g chat completion with |
| 127 | # prompt_embeds content parts). `True` means the position is a real |
| 128 | # token ID; `False` means the position uses a pre-computed entry from |
| 129 | # `prompt_embeds`. `None` for pure-tokens and pure-embeds requests. |
| 130 | prompt_is_token_ids: list[bool] | None = None |
| 131 | |
| 132 | # Index of the client, used to ensure outputs are sent back to the same |
| 133 | # client for this request when scaling out the front-end. |
| 134 | client_index: int = 0 |
| 135 | |
| 136 | # Used in DP case to indicate which wave of requests this is expected to |
| 137 | # belong to, to cover a race condition where the request is sent before |
| 138 | # a wave finished notification is received. |
| 139 | current_wave: int = 0 |
| 140 | priority: int = 0 |
| 141 | |
| 142 | trace_headers: Mapping[str, str] | None = None |
| 143 | resumable: bool = False |
| 144 | |
| 145 | # The user-provided request ID. This field is set internally, |
| 146 | # copied from the provided request_id that's originally assigned |
| 147 | # to the request_id field, see InputProcessor.assign_request_id(). |
| 148 | # Used in outputs and to support abort(req_id, internal=False). |
| 149 | external_req_id: str | None = None |
| 150 | |
| 151 | reasoning_ended: bool | None = None |
| 152 | reasoning_parser_kwargs: dict[str, Any] | None = None |
| 153 | |
| 154 | # If True, the request should be added to the scheduler's waiting queue |
| 155 | # and immediately aborted, so connector-side cleanup runs via the standard |
| 156 | # request_finished hook. Used to free P-side prefill blocks when a |
| 157 | # KV-transfer request is rejected on the D node before engine admission. |
| 158 | abort_immediately: bool = False |
| 159 | |
| 160 | session_id: str | None = None |
| 161 | kv_hints: KvHintsEnvelope | None = None |
| 162 | |
| 163 | @property |
| 164 | def params(self) -> SamplingParams | PoolingParams: |
| 165 | """Return the processed params (sampling or pooling).""" |
| 166 | if self.sampling_params is not None: |
| 167 | return self.sampling_params |
| 168 | assert self.pooling_params is not None |
| 169 | return self.pooling_params |
| 170 | |
| 171 | |
| 172 | class EngineCoreEventType(enum.IntEnum): |
| 173 | """The type of engine core request event.""" |
| 174 | |
| 175 | QUEUED = 1 |
| 176 | SCHEDULED = 2 |
| 177 | PREEMPTED = 3 |
| 178 | |
| 179 | |
| 180 | class EngineCoreEvent(msgspec.Struct): |
| 181 | """A timestamped engine core event associated with a request. |
| 182 | |
| 183 | The timestamp is a monotonic timestamp and is used by the engine |
| 184 | frontend to calculate intervals between engine core events. These |
| 185 | timestamps should not be compared with timestamps from other processes. |
| 186 | """ |
| 187 | |
| 188 | type: EngineCoreEventType |
| 189 | timestamp: float |
| 190 | |
| 191 | @classmethod |
| 192 | def new_event( |
| 193 | cls, event_type: EngineCoreEventType, timestamp: float | None = None |
| 194 | ) -> "EngineCoreEvent": |
| 195 | timestamp = time.monotonic() if timestamp is None else timestamp |
| 196 | return cls(event_type, timestamp) |
| 197 | |
| 198 | |
| 199 | class EngineCoreOutput( |
| 200 | msgspec.Struct, |
| 201 | array_like=True, # type: ignore[call-arg] |
| 202 | omit_defaults=True, # type: ignore[call-arg] |
| 203 | gc=False, |
| 204 | ): # type: ignore[call-arg] |
| 205 | request_id: str |
| 206 | new_token_ids: list[int] |
| 207 | |
| 208 | new_logprobs: LogprobsLists | None = None |
| 209 | new_prompt_logprobs_tensors: LogprobsTensors | None = None |
| 210 | |
| 211 | pooling_output: torch.Tensor | None = None |
| 212 | |
| 213 | finish_reason: FinishReason | None = None |
| 214 | stop_reason: int | str | None = None |
| 215 | events: list[EngineCoreEvent] | None = None |
| 216 | kv_transfer_params: dict[str, Any] | None = None |
| 217 | ec_transfer_params: dict[str, Any] | None = None |
| 218 | |
| 219 | trace_headers: Mapping[str, str] | None = None |
| 220 | |
| 221 | prefill_stats: PrefillStats | None = None |
| 222 | |
| 223 | routed_experts: np.ndarray | None = None |
| 224 | # The number of NaNs in logits. |
| 225 | # A value greater than 0 indicates that the output is corrupted. |
| 226 | num_nans_in_logits: int = 0 |
| 227 | # Multi-modal hashes missing from the P1 receiver cache (P0/P1 drift; see |
| 228 | # `MultiModalCacheMissError`). Non-empty => retryable: the frontend drops these |
| 229 | # from its sender cache and the request is resent with the data. Appended last |
| 230 | # so `array_like` positional serialization stays backward compatible. |
| 231 | mm_cache_miss_hashes: list[str] | None = None |
| 232 | |
| 233 | new_sampling_mask: SamplingMaskLists | None = None |
| 234 | |
| 235 | # Per-request spec-decode acceptance; attached only on the final output. |
| 236 | # Appended last so `array_like` positional serialization stays compatible. |
| 237 | spec_decode_metrics: RequestSpecDecodeMetrics | None = None |
| 238 | |
| 239 | # [num_scored_rows, num_token_ids]; appended last for array_like compatibility. |
| 240 | prompt_token_id_logprobs: torch.Tensor | None = None |
| 241 | |
| 242 | @property |
| 243 | def finished(self) -> bool: |
| 244 | return self.finish_reason is not None |
| 245 | |
| 246 | |
| 247 | class UtilityOutput( |
| 248 | msgspec.Struct, |
| 249 | array_like=True, # type: ignore[call-arg] |
| 250 | gc=False, |
| 251 | ): # type: ignore[call-arg] |
| 252 | call_id: int |
| 253 | |
| 254 | # Non-None implies the call failed, result should be None. |
| 255 | failure_message: str | None = None |
| 256 | result: UtilityResult | None = None |
| 257 | |
| 258 | |
| 259 | class EngineCoreOutputs( |
| 260 | msgspec.Struct, |
| 261 | array_like=True, # type: ignore[call-arg] |
| 262 | omit_defaults=True, # type: ignore[call-arg] |
| 263 | gc=False, |
| 264 | ): # type: ignore[call-arg] |
| 265 | # NOTE(Nick): We could consider ways to make this more compact, |
| 266 | # e.g. columnwise layout |
| 267 | |
| 268 | engine_index: int = 0 |
| 269 | |
| 270 | # [num_reqs] |
| 271 | outputs: list[EngineCoreOutput] = [] |
| 272 | scheduler_stats: SchedulerStats | None = None |
| 273 | timestamp: float = 0.0 |
| 274 | |
| 275 | utility_output: UtilityOutput | None = None |
| 276 | finished_requests: set[str] | None = None |
| 277 | |
| 278 | # In DP case, used to signal that the current wave of requests |
| 279 | # has finished and the engines are paused. |
| 280 | wave_complete: int | None = None |
| 281 | # In DP case, used to signal that a request was received for an |
| 282 | # "old" wave, so the next wave needs to be started in other engines. |
| 283 | start_wave: int | None = None |
| 284 | |
| 285 | def __post_init__(self): |
| 286 | if self.timestamp == 0.0: |
| 287 | self.timestamp = time.monotonic() |
| 288 | |
| 289 | |
| 290 | class EngineCoreRequestType(enum.Enum): |
| 291 | """Request types defined as hex byte strings, so it can be sent over sockets |
| 292 | without separate encoding step. |
| 293 | """ |
| 294 | |
| 295 | ADD = b"\x00" |
| 296 | ABORT = b"\x01" |
| 297 | START_DP_WAVE = b"\x02" |
| 298 | UTILITY = b"\x03" |
| 299 | # Sentinel used within EngineCoreProc. |
| 300 | EXECUTOR_FAILED = b"\x04" |
| 301 | # Sentinel to wake up input_queue.get() during shutdown. |
| 302 | WAKEUP = b"\x05" |
| 303 | |
| 304 | |
| 305 | class ReconfigureDistributedRequest(msgspec.Struct): |
| 306 | new_data_parallel_size: int |
| 307 | new_data_parallel_rank: int |
| 308 | new_data_parallel_rank_local: int |
| 309 | new_data_parallel_master_ip: str |
| 310 | new_data_parallel_master_port: int |
| 311 | new_data_parallel_master_port_list: list[int] |
| 312 | coord_store_port: int |
| 313 | |
| 314 | |
| 315 | class ReconfigureRankType(enum.IntEnum): |
| 316 | """Rank type for reconfiguring distributed request.""" |
| 317 | |
| 318 | KEEP_CURRENT_RANK = -1 |
| 319 | SHUTDOWN_CURRENT_RANK = -2 |
| 320 | |
| 321 | |
| 322 | class EngineStatusType(enum.IntEnum): |
| 323 | HEALTHY = 0 |
| 324 | DEAD = 1 |
| 325 | UNHEALTHY = 2 |