| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import time |
| 5 | from collections import defaultdict, deque |
| 6 | from dataclasses import dataclass, field |
| 7 | from typing import TYPE_CHECKING, Any |
| 8 | |
| 9 | import vllm.envs as envs |
| 10 | from vllm.compilation.cuda_graph import CUDAGraphStat |
| 11 | from vllm.v1.metrics.perf import PerfStats |
| 12 | from vllm.v1.spec_decode.metrics import SpecDecodingStats |
| 13 | |
| 14 | if TYPE_CHECKING: |
| 15 | from vllm.v1.engine import EngineCoreEvent, EngineCoreOutput, FinishReason |
| 16 | |
| 17 | |
| 18 | @dataclass |
| 19 | class BaseCacheStats: |
| 20 | """Stores cache hit statistics.""" |
| 21 | |
| 22 | reset: bool = False |
| 23 | """Whether the cache was reset.""" |
| 24 | |
| 25 | requests: int = 0 |
| 26 | """The number of requests in this update.""" |
| 27 | |
| 28 | queries: int = 0 |
| 29 | """The number of queries in these requests.""" |
| 30 | |
| 31 | hits: int = 0 |
| 32 | """The number of hits in these requests.""" |
| 33 | |
| 34 | |
| 35 | class CachingMetrics: |
| 36 | """Metrics for caching with a hit rate of the most recent N requests. |
| 37 | |
| 38 | Args: |
| 39 | max_recent_requests: The number of the most recent requests to aggregate. |
| 40 | |
| 41 | """ |
| 42 | |
| 43 | def __init__(self, max_recent_requests: int = 1000) -> None: |
| 44 | super().__init__() |
| 45 | |
| 46 | self.max_recent_requests = max_recent_requests |
| 47 | # The current aggregated values. |
| 48 | self.aggregated_requests = 0 |
| 49 | self.aggregated_query_total = 0 |
| 50 | self.aggregated_query_hit = 0 |
| 51 | |
| 52 | # A deque of (requests, queries, hits) for the most recent requests. |
| 53 | self.query_queue = deque[tuple[int, int, int]]() |
| 54 | |
| 55 | def observe(self, stats: BaseCacheStats): |
| 56 | """Observe the prefix caching for a set of requests. |
| 57 | |
| 58 | This function is called with information gathered when new requests |
| 59 | are being scheduled and are looking for computed blocks. |
| 60 | |
| 61 | When there are more than `max_recent_requests` requests, the oldest set |
| 62 | of requests are removed from the metrics. |
| 63 | |
| 64 | Args: |
| 65 | stats: The prefix cache stats. |
| 66 | |
| 67 | """ |
| 68 | # reset_prefix_cache was invoked before the current update. |
| 69 | # Reset the metrics before aggregating the current stats. |
| 70 | if stats.reset: |
| 71 | self.reset() |
| 72 | |
| 73 | # DO NOT appending empty stats to avoid helpful info get kicked out |
| 74 | # due to sliding window. |
| 75 | if stats.requests == 0: |
| 76 | return |
| 77 | |
| 78 | # Update the metrics. |
| 79 | self.query_queue.append((stats.requests, stats.queries, stats.hits)) |
| 80 | self.aggregated_requests += stats.requests |
| 81 | self.aggregated_query_total += stats.queries |
| 82 | self.aggregated_query_hit += stats.hits |
| 83 | |
| 84 | # Remove the oldest stats until number of requests does not exceed |
| 85 | # the limit. |
| 86 | # NOTE: We preserve the latest added stats regardless. |
| 87 | while ( |
| 88 | len(self.query_queue) > 1 |
| 89 | and self.aggregated_requests > self.max_recent_requests |
| 90 | ): |
| 91 | old_requests, old_queries, old_hits = self.query_queue.popleft() |
| 92 | self.aggregated_requests -= old_requests |
| 93 | self.aggregated_query_total -= old_queries |
| 94 | self.aggregated_query_hit -= old_hits |
| 95 | |
| 96 | def reset(self): |
| 97 | """Reset the metrics.""" |
| 98 | self.aggregated_requests = 0 |
| 99 | self.aggregated_query_total = 0 |
| 100 | self.aggregated_query_hit = 0 |
| 101 | self.query_queue.clear() |
| 102 | |
| 103 | @property |
| 104 | def empty(self) -> bool: |
| 105 | """Return true if no requests have been observed.""" |
| 106 | return self.aggregated_requests == 0 |
| 107 | |
| 108 | @property |
| 109 | def hit_rate(self) -> float: |
| 110 | """Calculate the hit rate for the past N requests.""" |
| 111 | if self.aggregated_query_total == 0: |
| 112 | return 0.0 |
| 113 | return self.aggregated_query_hit / self.aggregated_query_total |
| 114 | |
| 115 | |
| 116 | @dataclass |
| 117 | class PrefixCacheStats(BaseCacheStats): |
| 118 | """Stores prefix cache hit statistics. |
| 119 | - `reset`: Whether `reset_prefix_cache` was invoked. |
| 120 | - `queries`: Refers to the number of tokens that were queried. |
| 121 | """ |
| 122 | |
| 123 | preempted_requests: int = 0 |
| 124 | """The number of previously preempted requests in this update.""" |
| 125 | |
| 126 | preempted_queries: int = 0 |
| 127 | """The `queries` number for preempted requests.""" |
| 128 | |
| 129 | preempted_hits: int = 0 |
| 130 | """The `hits` number for preempted requests.""" |
| 131 | |
| 132 | def record(self, num_tokens: int, num_hits: int, preempted: bool) -> None: |
| 133 | """Aggregate request information into the stats.""" |
| 134 | if preempted: |
| 135 | # Previously preempted request |
| 136 | self.preempted_requests += 1 |
| 137 | self.preempted_queries += num_tokens |
| 138 | self.preempted_hits += num_hits |
| 139 | else: |
| 140 | # New request |
| 141 | self.requests += 1 |
| 142 | self.queries += num_tokens |
| 143 | self.hits += num_hits |
| 144 | |
| 145 | |
| 146 | @dataclass |
| 147 | class MultiModalCacheStats(BaseCacheStats): |
| 148 | """Stores multi-modal cache hit statistics. |
| 149 | - `reset`: Whether `reset_mm_cache` was invoked. |
| 150 | - `queries`: Refers to the number of multi-modal data items |
| 151 | that were queried. |
| 152 | """ |
| 153 | |
| 154 | def record(self, num_queries: int, num_hits: int) -> None: |
| 155 | """Aggregate request information into the stats.""" |
| 156 | self.requests += 1 |
| 157 | self.queries += num_queries |
| 158 | self.hits += num_hits |
| 159 | |
| 160 | |
| 161 | @dataclass |
| 162 | class KVCacheEvictionEvent: |
| 163 | """Single KV cache block eviction sample.""" |
| 164 | |
| 165 | lifetime_seconds: float |
| 166 | idle_seconds: float |
| 167 | reuse_gaps_seconds: tuple[float, ...] |
| 168 | |
| 169 | |
| 170 | @dataclass |
| 171 | class SchedulerIterationDetails: |
| 172 | """Scheduler-side details for one engine iteration.""" |
| 173 | |
| 174 | iteration_index: int |
| 175 | num_ctx_requests: int |
| 176 | num_ctx_tokens: int |
| 177 | num_generation_requests: int |
| 178 | num_generation_tokens: int |
| 179 | elapsed_ms: float |
| 180 | num_encoder_inputs: int = 0 |
| 181 | num_encoder_output_tokens: int = 0 |
| 182 | is_dummy: bool = False |
| 183 | |
| 184 | |
| 185 | KV_FETCH_WAITING_TO_START = "waiting_to_start" |
| 186 | KV_FETCH_IN_PROGRESS = "in_progress" |
| 187 | KV_FETCH_COMPLETED_WAITING = "completed_waiting" |
| 188 | KV_FETCH_STAGES = ( |
| 189 | KV_FETCH_WAITING_TO_START, |
| 190 | KV_FETCH_IN_PROGRESS, |
| 191 | KV_FETCH_COMPLETED_WAITING, |
| 192 | ) |
| 193 | |
| 194 | |
| 195 | @dataclass |
| 196 | class SchedulerStats: |
| 197 | """Stats associated with the scheduler.""" |
| 198 | |
| 199 | num_running_reqs: int = 0 |
| 200 | |
| 201 | num_waiting_reqs: int = 0 # waiting requests not deferred |
| 202 | # waiting requests deferred by transient constraints or a blocked status |
| 203 | num_skipped_waiting_reqs: int = 0 |
| 204 | |
| 205 | num_kv_fetch_reqs_by_stage: dict[str, int] = field(default_factory=dict) |
| 206 | |
| 207 | # These are used for internal DP load-balancing. |
| 208 | step_counter: int = 0 |
| 209 | current_wave: int = 0 |
| 210 | |
| 211 | kv_cache_usage: float = 0.0 |
| 212 | iteration_details: SchedulerIterationDetails | None = None |
| 213 | |
| 214 | prefix_cache_stats: PrefixCacheStats = field(default_factory=PrefixCacheStats) |
| 215 | connector_prefix_cache_stats: PrefixCacheStats | None = None |
| 216 | |
| 217 | kv_cache_eviction_events: list[KVCacheEvictionEvent] = field(default_factory=list) |
| 218 | |
| 219 | spec_decoding_stats: SpecDecodingStats | None = None |
| 220 | kv_connector_stats: dict[str, Any] | None = None |
| 221 | ec_connector_stats: dict[str, Any] | None = None |
| 222 | |
| 223 | waiting_lora_adapters: dict[str, int] = field(default_factory=dict) |
| 224 | running_lora_adapters: dict[str, int] = field(default_factory=dict) |
| 225 | |
| 226 | cudagraph_stats: CUDAGraphStat | None = None |
| 227 | |
| 228 | perf_stats: PerfStats | None = None |
| 229 | |
| 230 | |
| 231 | @dataclass |
| 232 | class RequestStateStats: |
| 233 | """Stats that need to be tracked across delta updates.""" |
| 234 | |
| 235 | num_generation_tokens: int = 0 |
| 236 | num_preemptions: int = 0 |
| 237 | |
| 238 | # This is an engine frontend timestamp (wall-clock) |
| 239 | arrival_time: float = 0.0 |
| 240 | |
| 241 | # These are engine core timestamps (monotonic) |
| 242 | queued_ts: float = 0.0 |
| 243 | scheduled_ts: float = 0.0 |
| 244 | first_token_ts: float = 0.0 |
| 245 | last_token_ts: float = 0.0 |
| 246 | |
| 247 | # first token latency |
| 248 | first_token_latency: float = 0.0 |
| 249 | |
| 250 | # Track if this request is corrupted (NaNs in logits) |
| 251 | is_corrupted: bool = False |
| 252 | |
| 253 | |
| 254 | @dataclass |
| 255 | class FinishedRequestStats: |
| 256 | """Stats associated with a finished request.""" |
| 257 | |
| 258 | finish_reason: "FinishReason" |
| 259 | request_id: str | None = None |
| 260 | e2e_latency: float = 0.0 |
| 261 | num_prompt_tokens: int = 0 |
| 262 | num_preemptions: int = 0 |
| 263 | num_generation_tokens: int = 0 |
| 264 | max_tokens_param: int | None = None |
| 265 | queued_time: float = 0.0 |
| 266 | prefill_time: float = 0.0 |
| 267 | inference_time: float = 0.0 |
| 268 | decode_time: float = 0.0 |
| 269 | mean_time_per_output_token: float = 0.0 |
| 270 | is_corrupted: bool = False |
| 271 | num_cached_tokens: int = 0 |
| 272 | |
| 273 | |
| 274 | @dataclass |
| 275 | class PrefillStats: |
| 276 | """Breakdown of a scheduled prefill computation. |
| 277 | |
| 278 | Fields: |
| 279 | num_prompt_tokens: Total number of tokens to be prefilled. |
| 280 | num_computed_tokens: Tokens to be prefilled locally (actual compute work). |
| 281 | num_cached_tokens: Tokens to be prefilled without actual compute work. |
| 282 | num_local_cached_tokens: Tokens to be prefilled from local prefix cache. |
| 283 | num_external_cached_tokens: Tokens to be prefilled from external KV transfer. |
| 284 | num_cache_creation_tokens: Tokens computed and written to the prefix cache. |
| 285 | """ |
| 286 | |
| 287 | num_prompt_tokens: int = 0 |
| 288 | num_computed_tokens: int = 0 |
| 289 | num_cached_tokens: int = 0 |
| 290 | num_local_cached_tokens: int = 0 |
| 291 | num_external_cached_tokens: int = 0 |
| 292 | num_cache_creation_tokens: int = 0 |
| 293 | |
| 294 | def set( |
| 295 | self, |
| 296 | num_prompt_tokens: int, |
| 297 | num_local_cached_tokens: int, |
| 298 | num_external_cached_tokens: int, |
| 299 | ): |
| 300 | num_cached_tokens = num_local_cached_tokens + num_external_cached_tokens |
| 301 | assert num_cached_tokens <= num_prompt_tokens |
| 302 | |
| 303 | self.num_prompt_tokens = num_prompt_tokens |
| 304 | self.num_computed_tokens = num_prompt_tokens - num_cached_tokens |
| 305 | self.num_cached_tokens = num_cached_tokens |
| 306 | self.num_local_cached_tokens = num_local_cached_tokens |
| 307 | self.num_external_cached_tokens = num_external_cached_tokens |
| 308 | |
| 309 | def finalize(self, num_cached_tokens: int) -> None: |
| 310 | assert num_cached_tokens >= 0 |
| 311 | self.num_cache_creation_tokens = max( |
| 312 | 0, min(num_cached_tokens, self.num_prompt_tokens) - self.num_cached_tokens |
| 313 | ) |
| 314 | |
| 315 | |
| 316 | @dataclass |
| 317 | class RequestSpecDecodeMetrics: |
| 318 | """Per-output-sequence speculative-decoding statistics accumulator. |
| 319 | |
| 320 | Accumulates, over one sequence's verify steps, a histogram of accepted |
| 321 | draft-token counts (``j``, draft-only) and the total number of proposed |
| 322 | draft tokens. When ``detailed`` is requested it also records the ordered |
| 323 | per-step accepted/proposed sequences (``summary`` omits them). Tracked per |
| 324 | engine ``Request`` (one per sampled sequence, so ``n > 1`` yields one per |
| 325 | child), surfaced via ``EngineCoreOutput`` and the response |
| 326 | ``metrics.speculative_decoding`` for single-sequence requests (see |
| 327 | ``to_dict``). |
| 328 | |
| 329 | Fields: |
| 330 | num_spec_tokens: Configured ``num_speculative_tokens`` (the max ``k``); |
| 331 | also the histogram's upper bound. |
| 332 | histogram: Dense counts indexed by accepted draft tokens ``j`` |
| 333 | (length ``num_spec_tokens + 1``). |
| 334 | num_draft_tokens: Total proposed draft tokens, after the |
| 335 | grammar-invalidated (``num_invalid_spec_tokens``) adjustment. |
| 336 | per_step_accepted: Ordered accepted-draft count per verify step |
| 337 | (``detailed`` only; empty otherwise). |
| 338 | per_step_drafted: Ordered proposed-draft count per verify step |
| 339 | (``detailed`` only; empty otherwise). |
| 340 | """ |
| 341 | |
| 342 | num_spec_tokens: int |
| 343 | histogram: list[int] = field(default_factory=list) |
| 344 | num_draft_tokens: int = 0 |
| 345 | per_step_accepted: list[int] = field(default_factory=list) |
| 346 | per_step_drafted: list[int] = field(default_factory=list) |
| 347 | |
| 348 | @classmethod |
| 349 | def new(cls, num_spec_tokens: int) -> "RequestSpecDecodeMetrics": |
| 350 | return cls( |
| 351 | num_spec_tokens=num_spec_tokens, |
| 352 | histogram=[0] * (num_spec_tokens + 1), |
| 353 | ) |
| 354 | |
| 355 | def observe( |
| 356 | self, num_draft_tokens: int, num_accepted: int, detailed: bool = False |
| 357 | ) -> None: |
| 358 | self.histogram[num_accepted] += 1 |
| 359 | self.num_draft_tokens += num_draft_tokens |
| 360 | if detailed: |
| 361 | self.per_step_accepted.append(num_accepted) |
| 362 | self.per_step_drafted.append(num_draft_tokens) |
| 363 | |
| 364 | def to_dict(self) -> dict[str, Any]: |
| 365 | """Payload matching ``SpeculativeDecodingMetrics`` for the response. |
| 366 | |
| 367 | ``acceptance_histogram`` is a dense list indexed by accepted draft count |
| 368 | ``j`` (length ``num_spec_tokens + 1``). ``mean_acceptance_length`` |
| 369 | includes the bonus token (``j + 1``); ``draft_acceptance_rate`` is |
| 370 | draft-only, full precision. Per-step arrays are included only when |
| 371 | populated (``detailed`` level). |
| 372 | """ |
| 373 | num_spec_steps = sum(self.histogram) |
| 374 | num_accepted = sum(j * count for j, count in enumerate(self.histogram)) |
| 375 | mean_al = 1.0 + num_accepted / num_spec_steps if num_spec_steps else 1.0 |
| 376 | rate = num_accepted / self.num_draft_tokens if self.num_draft_tokens else 0.0 |
| 377 | result: dict[str, Any] = { |
| 378 | "mean_acceptance_length": mean_al, |
| 379 | "draft_acceptance_rate": rate, |
| 380 | "acceptance_histogram": list(self.histogram), |
| 381 | "num_spec_steps": num_spec_steps, |
| 382 | "num_accepted_draft_tokens": num_accepted, |
| 383 | "num_draft_tokens": self.num_draft_tokens, |
| 384 | "num_spec_tokens": self.num_spec_tokens, |
| 385 | } |
| 386 | if self.per_step_accepted: |
| 387 | result["per_step_accepted"] = self.per_step_accepted |
| 388 | result["per_step_drafted"] = self.per_step_drafted |
| 389 | return result |
| 390 | |
| 391 | |
| 392 | @dataclass |
| 393 | class PromptTokenStats: |
| 394 | """Breakdown of prompt tokens by source. |
| 395 | |
| 396 | Fields: |
| 397 | computed: Tokens prefilled locally (actual compute work). |
| 398 | local_cache_hit: Tokens from local prefix cache. |
| 399 | external_kv_transfer: Tokens from external KV transfer. |
| 400 | cached_tokens: Tokens skipped during prefill (from scheduler). |
| 401 | total: Total prompt tokens. |
| 402 | |
| 403 | Invariants: |
| 404 | computed + local_cache_hit + external_kv_transfer = total |
| 405 | local_cache_hit + external_kv_transfer = cached_tokens |
| 406 | """ |
| 407 | |
| 408 | ALL_SOURCES: tuple[str, ...] = ( |
| 409 | "local_compute", |
| 410 | "local_cache_hit", |
| 411 | "external_kv_transfer", |
| 412 | ) |
| 413 | |
| 414 | computed: int = 0 |
| 415 | local_cache_hit: int = 0 |
| 416 | external_kv_transfer: int = 0 |
| 417 | cached_tokens: int = 0 |
| 418 | total: int = 0 |
| 419 | |
| 420 | def update_from_output(self, prefill_stats: PrefillStats) -> None: |
| 421 | """Update stats from a prefill output.""" |
| 422 | self.computed += prefill_stats.num_computed_tokens |
| 423 | self.cached_tokens += prefill_stats.num_cached_tokens |
| 424 | self.total += prefill_stats.num_prompt_tokens |
| 425 | |
| 426 | self.local_cache_hit += prefill_stats.num_local_cached_tokens |
| 427 | self.external_kv_transfer += prefill_stats.num_external_cached_tokens |
| 428 | |
| 429 | def get_by_source(self, source: str) -> int: |
| 430 | """Get token count by source label.""" |
| 431 | source_map = { |
| 432 | "local_compute": self.computed, |
| 433 | "local_cache_hit": self.local_cache_hit, |
| 434 | "external_kv_transfer": self.external_kv_transfer, |
| 435 | } |
| 436 | if source not in source_map: |
| 437 | raise ValueError(f"Unknown source: {source}") |
| 438 | return source_map[source] |
| 439 | |
| 440 | |
| 441 | class IterationStats: |
| 442 | """Stats associated with a single set of EngineCoreOutputs.""" |
| 443 | |
| 444 | def __init__(self): |
| 445 | self.iteration_timestamp = time.time() |
| 446 | self.num_generation_tokens = 0 |
| 447 | self.prompt_token_stats = PromptTokenStats() |
| 448 | self.num_preempted_reqs = 0 |
| 449 | self.finished_requests: list[FinishedRequestStats] = [] |
| 450 | self.max_num_generation_tokens_iter: list[int] = [] |
| 451 | self.n_params_iter: list[int] = [] |
| 452 | self.time_to_first_tokens_iter: list[float] = [] |
| 453 | self.inter_token_latencies_iter: list[float] = [] |
| 454 | self.num_corrupted_reqs: int = 0 |
| 455 | |
| 456 | def __repr__(self) -> str: |
| 457 | field_to_value_str = ", ".join(f"{k}={v}" for k, v in vars(self).items()) |
| 458 | return f"{self.__class__.__name__}({field_to_value_str})" |
| 459 | |
| 460 | @property |
| 461 | def num_prompt_tokens(self) -> int: |
| 462 | """Total prompt tokens (for backward compatibility).""" |
| 463 | return self.prompt_token_stats.total |
| 464 | |
| 465 | def _time_since(self, start: float) -> float: |
| 466 | """Calculate an interval relative to this iteration's timestamp.""" |
| 467 | return self.iteration_timestamp - start |
| 468 | |
| 469 | def update_from_output( |
| 470 | self, |
| 471 | output: "EngineCoreOutput", |
| 472 | engine_core_timestamp: float, |
| 473 | is_prefilling: bool, |
| 474 | req_stats: RequestStateStats, |
| 475 | lora_states: "LoRARequestStates", |
| 476 | lora_name: str | None, |
| 477 | ): |
| 478 | num_new_generation_tokens = len(output.new_token_ids) |
| 479 | |
| 480 | self.num_generation_tokens += num_new_generation_tokens |
| 481 | if is_prefilling: |
| 482 | if output.prefill_stats is not None: |
| 483 | self.prompt_token_stats.update_from_output(output.prefill_stats) |
| 484 | |
| 485 | first_token_latency = self._time_since(req_stats.arrival_time) |
| 486 | self.time_to_first_tokens_iter.append(first_token_latency) |
| 487 | req_stats.first_token_latency = first_token_latency |
| 488 | |
| 489 | req_stats.num_generation_tokens += num_new_generation_tokens |
| 490 | |
| 491 | # Track if this request is corrupted (only check once per request) |
| 492 | # Early exit if already marked as corrupted to avoid redundant checks |
| 493 | if ( |
| 494 | envs.VLLM_COMPUTE_NANS_IN_LOGITS |
| 495 | and not req_stats.is_corrupted |
| 496 | and output.num_nans_in_logits > 0 |
| 497 | ): |
| 498 | req_stats.is_corrupted = True |
| 499 | |
| 500 | # Process request-level engine core events |
| 501 | if output.events is not None: |
| 502 | self.update_from_events( |
| 503 | output.request_id, |
| 504 | output.events, |
| 505 | is_prefilling, |
| 506 | req_stats, |
| 507 | lora_states, |
| 508 | lora_name, |
| 509 | ) |
| 510 | |
| 511 | # Process the batch-level "new tokens" engine core event |
| 512 | if is_prefilling: |
| 513 | req_stats.first_token_ts = engine_core_timestamp |
| 514 | else: |
| 515 | itl = engine_core_timestamp - req_stats.last_token_ts |
| 516 | self.inter_token_latencies_iter.append(itl) |
| 517 | |
| 518 | req_stats.last_token_ts = engine_core_timestamp |
| 519 | |
| 520 | def update_from_events( |
| 521 | self, |
| 522 | req_id: str, |
| 523 | events: list["EngineCoreEvent"], |
| 524 | is_prefilling: bool, |
| 525 | req_stats: RequestStateStats, |
| 526 | lora_states: "LoRARequestStates", |
| 527 | lora_name: str | None, |
| 528 | ): |
| 529 | # Avoid circular dependency |
| 530 | from vllm.v1.engine import EngineCoreEventType |
| 531 | |
| 532 | for event in events: |
| 533 | if event.type == EngineCoreEventType.QUEUED: |
| 534 | req_stats.queued_ts = event.timestamp |
| 535 | lora_states.request_waiting(req_id, lora_name) |
| 536 | elif event.type == EngineCoreEventType.SCHEDULED: |
| 537 | if req_stats.scheduled_ts == 0.0: # ignore preemptions |
| 538 | req_stats.scheduled_ts = event.timestamp |
| 539 | lora_states.request_running(req_id, lora_name) |
| 540 | elif event.type == EngineCoreEventType.PREEMPTED: |
| 541 | self.num_preempted_reqs += 1 |
| 542 | req_stats.num_preemptions += 1 |
| 543 | lora_states.request_waiting(req_id, lora_name) |
| 544 | |
| 545 | def update_from_finished_request( |
| 546 | self, |
| 547 | finish_reason: "FinishReason", |
| 548 | request_id: str, |
| 549 | num_prompt_tokens: int, |
| 550 | max_tokens_param: int | None, |
| 551 | req_stats: RequestStateStats, |
| 552 | num_cached_tokens: int = 0, |
| 553 | ): |
| 554 | e2e_latency = self._time_since(req_stats.arrival_time) |
| 555 | |
| 556 | # Queued interval is from first QUEUED event to first SCHEDULED |
| 557 | queued_time = req_stats.scheduled_ts - req_stats.queued_ts |
| 558 | |
| 559 | # Prefill interval is from first SCHEDULED to first NEW_TOKEN |
| 560 | # Any preemptions during prefill is included in the interval |
| 561 | prefill_time = req_stats.first_token_ts - req_stats.scheduled_ts |
| 562 | |
| 563 | # Decode interval is from first NEW_TOKEN to last NEW_TOKEN |
| 564 | # Any preemptions during decode are included |
| 565 | decode_time = req_stats.last_token_ts - req_stats.first_token_ts |
| 566 | |
| 567 | # Inference interval is from first SCHEDULED to last NEW_TOKEN |
| 568 | # Any preemptions during prefill or decode are included |
| 569 | inference_time = req_stats.last_token_ts - req_stats.scheduled_ts |
| 570 | |
| 571 | # Do not count the token generated by the prefill phase |
| 572 | mean_time_per_output_token = ( |
| 573 | decode_time / (req_stats.num_generation_tokens - 1) |
| 574 | if req_stats.num_generation_tokens - 1 > 0 |
| 575 | else 0 |
| 576 | ) |
| 577 | |
| 578 | finished_req = FinishedRequestStats( |
| 579 | finish_reason=finish_reason, |
| 580 | request_id=request_id, |
| 581 | e2e_latency=e2e_latency, |
| 582 | num_prompt_tokens=num_prompt_tokens, |
| 583 | num_preemptions=req_stats.num_preemptions, |
| 584 | num_generation_tokens=req_stats.num_generation_tokens, |
| 585 | max_tokens_param=max_tokens_param, |
| 586 | queued_time=queued_time, |
| 587 | prefill_time=prefill_time, |
| 588 | inference_time=inference_time, |
| 589 | decode_time=decode_time, |
| 590 | mean_time_per_output_token=mean_time_per_output_token, |
| 591 | is_corrupted=req_stats.is_corrupted, |
| 592 | num_cached_tokens=num_cached_tokens, |
| 593 | ) |
| 594 | self.finished_requests.append(finished_req) |
| 595 | |
| 596 | # Count corrupted requests when they finish (only once per request) |
| 597 | if req_stats.is_corrupted: |
| 598 | self.num_corrupted_reqs += 1 |
| 599 | |
| 600 | |
| 601 | class LoRAStats: |
| 602 | """Tracks waiting and running request IDs for a single LoRA.""" |
| 603 | |
| 604 | def __init__(self): |
| 605 | self.waiting: set[str] = set() |
| 606 | self.running: set[str] = set() |
| 607 | |
| 608 | def update(self, req_id: str, waiting: bool, running: bool): |
| 609 | assert not (waiting and running) |
| 610 | if waiting: |
| 611 | self.waiting.add(req_id) |
| 612 | else: |
| 613 | self.waiting.discard(req_id) |
| 614 | |
| 615 | if running: |
| 616 | self.running.add(req_id) |
| 617 | else: |
| 618 | self.running.discard(req_id) |
| 619 | |
| 620 | @property |
| 621 | def empty(self) -> bool: |
| 622 | return not (self.waiting or self.running) |
| 623 | |
| 624 | |
| 625 | class LoRARequestStates: |
| 626 | """A per-LoRA count of running and waiting requests.""" |
| 627 | |
| 628 | def __init__(self, log_stats: bool = False): |
| 629 | self.log_stats = log_stats |
| 630 | self.requests: defaultdict[str, LoRAStats] = defaultdict(LoRAStats) |
| 631 | |
| 632 | def _request_update( |
| 633 | self, req_id: str, lora_name: str | None, waiting: bool, running: bool |
| 634 | ): |
| 635 | if not self.log_stats or lora_name is None: |
| 636 | return |
| 637 | |
| 638 | lora_stats = self.requests[lora_name] |
| 639 | lora_stats.update(req_id, waiting, running) |
| 640 | if lora_stats.empty: |
| 641 | del self.requests[lora_name] |
| 642 | |
| 643 | def request_waiting(self, req_id: str, lora_name: str | None): |
| 644 | self._request_update(req_id, lora_name, waiting=True, running=False) |
| 645 | |
| 646 | def request_running(self, req_id: str, lora_name: str | None): |
| 647 | self._request_update(req_id, lora_name, waiting=False, running=True) |
| 648 | |
| 649 | def request_finished(self, req_id: str, lora_name: str | None): |
| 650 | self._request_update(req_id, lora_name, waiting=False, running=False) |
| 651 | |
| 652 | def update_scheduler_stats(self, scheduler_stats: SchedulerStats | None): |
| 653 | if not self.log_stats or scheduler_stats is None: |
| 654 | return |
| 655 | for lora_name, stats in self.requests.items(): |
| 656 | scheduler_stats.waiting_lora_adapters[lora_name] = len(stats.waiting) |
| 657 | scheduler_stats.running_lora_adapters[lora_name] = len(stats.running) |