@tduncan / Nvidia-Optimization-Stack-VLLM Mirror · read-only / commits / 65334ef3b9
Fork 0

A high-throughput and memory-efficient inference and serving engine for LLMs

Mirrored from https://github.com/tduncan-lum/Nvidia-Optimization-Stack---VLLM.git

Synced Nov 15, 09:45 UTC

[V1][Metrics] Remove unused code (#18158)

Signed-off-by: Mark McLoughlin <markmc@redhat.com>

Mark McLoughlin committed May 15, 2025 at 04:13 UTC 65334ef3b9e4fd32ebc5c4e512debc25d5025488
4 files changed -756
.buildkite/test-pipeline.yaml
-1
@@ -216,7 +216,6 @@ steps:
216 - pytest -v -s v1/spec_decode
217 - pytest -v -s v1/kv_connector/unit
218 - pytest -v -s v1/test_serial_utils.py
219 - - pytest -v -s v1/test_stats.py
219 - pytest -v -s v1/test_utils.py
220 - pytest -v -s v1/test_oracle.py
221 # TODO: accuracy does not match, whether setting
tests/v1/test_stats.py deleted
-302
@@ -1,302 +0,0 @@
1 -# SPDX-License-Identifier: Apache-2.0
2 -
3 -import pytest
4 -
5 -from vllm.sampling_params import SamplingParams
6 -from vllm.v1.stats.common import RequestStats, RequestStatsUpdate
7 -
8 -
9 -def make_update(
10 - request_id: str,
11 - update_type: RequestStatsUpdate.Type,
12 - monotonic_ts_s: float,
13 - **kwargs,
14 -):
15 - if update_type == RequestStatsUpdate.Type.INPUT_PROCESSED:
16 - kwargs.setdefault("sampling_params", SamplingParams(n=1))
17 - kwargs.setdefault("num_prompt_tokens", 10)
18 - elif update_type == RequestStatsUpdate.Type.PREFILLING:
19 - kwargs.setdefault("num_computed_tokens", 10)
20 - kwargs.setdefault("num_cached_tokens", 10)
21 - elif update_type == RequestStatsUpdate.Type.DETOKENIZED:
22 - kwargs.setdefault("num_new_tokens", 10)
23 - elif update_type == RequestStatsUpdate.Type.FINISHED:
24 - kwargs.setdefault("finish_reason", "test_reason")
25 -
26 - return RequestStatsUpdate(
27 - request_id=request_id,
28 - type=update_type,
29 - monotonic_ts_s=monotonic_ts_s,
30 - **kwargs,
31 - )
32 -
33 -
34 -def test_invalid_request_update():
35 - request_id = "test_request"
36 - update_specific_required_fields = {
37 - RequestStatsUpdate.Type.INPUT_PROCESSED: [
38 - "sampling_params",
39 - "num_prompt_tokens",
40 - ],
41 - RequestStatsUpdate.Type.PREFILLING: [
42 - "num_computed_tokens",
43 - "num_cached_tokens",
44 - ],
45 - RequestStatsUpdate.Type.DETOKENIZED: ["num_new_tokens"],
46 - RequestStatsUpdate.Type.FINISHED: ["finish_reason"],
47 - }
48 -
49 - # Missing a required field should raise an assertion error.
50 - for update_type in RequestStatsUpdate.Type:
51 - required_fields = update_specific_required_fields.get(update_type, [])
52 -
53 - # Try to miss one of the required fields.
54 - kwargs = {field: object() for field in required_fields}
55 - for field in required_fields:
56 - copy_kwargs = kwargs.copy()
57 - copy_kwargs.pop(field)
58 - with pytest.raises(ValueError):
59 - RequestStatsUpdate(
60 - request_id=request_id,
61 - type=update_type,
62 - **copy_kwargs,
63 - )
64 -
65 -
66 -def test_invalid_request_update_transition():
67 - # Test invalid transition type.
68 - for src in RequestStatsUpdate.Type:
69 - for dst in RequestStatsUpdate.Type:
70 - if dst not in RequestStatsUpdate._VALID_TRANSITIONS[src]:
71 - with pytest.raises(AssertionError):
72 - RequestStatsUpdate.check_valid_update(
73 - make_update(
74 - update_type=dst,
75 - request_id="test_request",
76 - monotonic_ts_s=1,
77 - ),
78 - last_update_type=src,
79 - last_updated_ts_s=0,
80 - )
81 - else:
82 - RequestStatsUpdate.check_valid_update(
83 - make_update(
84 - request_id="test_request",
85 - update_type=dst,
86 - monotonic_ts_s=1,
87 - ),
88 - last_update_type=src,
89 - last_updated_ts_s=0,
90 - )
91 -
92 - # Test invalid timestamp.
93 - with pytest.raises(AssertionError):
94 - RequestStatsUpdate.check_valid_update(
95 - make_update(
96 - request_id="test_request",
97 - update_type=RequestStatsUpdate.Type.ARRIVED,
98 - monotonic_ts_s=1,
99 - ),
100 - last_update_type=None,
101 - last_updated_ts_s=2,
102 - )
103 -
104 -
105 -def test_lifecycle_updates():
106 - request_id = "test_request"
107 - stats = RequestStats(request_id=request_id)
108 -
109 - # Test the below scenario:
110 - arrived_ts = 0
111 - input_processed_ts = 1
112 - queued_ts = 2
113 - prefilling_ts = 3
114 - decoded_ts = 5
115 - detokenized_ts = 6
116 - decoded_2_ts = 7
117 - detokenized_2_ts = 8
118 - preempted_ts = 9
119 - resumed_ts = 10
120 - decoded_3_ts = 11
121 - detokenized_3_ts = 12
122 - finished_ts = 13
123 -
124 - # Test ARRIVED
125 - arrived_update = RequestStatsUpdate(
126 - request_id=request_id,
127 - type=RequestStatsUpdate.Type.ARRIVED,
128 - monotonic_ts_s=arrived_ts,
129 - )
130 - stats.update_from(arrived_update)
131 - assert stats.arrival_ts_s == arrived_ts
132 - assert stats.last_updated_ts_s == arrived_ts
133 -
134 - # Test INPUT_PROCESSED
135 - sampling_params = SamplingParams(n=1)
136 - input_processed_update = RequestStatsUpdate(
137 - request_id=request_id,
138 - type=RequestStatsUpdate.Type.INPUT_PROCESSED,
139 - monotonic_ts_s=input_processed_ts,
140 - sampling_params=sampling_params,
141 - num_prompt_tokens=6,
142 - )
143 - stats.update_from(input_processed_update)
144 - assert stats.input_processor_end_ts_s == input_processed_ts
145 - assert stats.last_updated_ts_s == input_processed_ts
146 - assert stats.num_prompt_tokens == 6
147 - assert stats.sampling_params == sampling_params
148 -
149 - assert stats.first_token_ts_s is None
150 - assert stats.prefill_ts_s is None
151 -
152 - # Test QUEUED
153 - queued_update = RequestStatsUpdate(
154 - request_id=request_id,
155 - type=RequestStatsUpdate.Type.QUEUED,
156 - monotonic_ts_s=queued_ts,
157 - )
158 - stats.update_from(queued_update)
159 - assert stats.queued_ts_s == queued_ts
160 - assert stats.last_updated_ts_s == queued_ts
161 -
162 - # Test PREFILLING
163 - prefilling_update = RequestStatsUpdate(
164 - request_id=request_id,
165 - type=RequestStatsUpdate.Type.PREFILLING,
166 - monotonic_ts_s=prefilling_ts,
167 - num_computed_tokens=3,
168 - num_cached_tokens=1,
169 - )
170 - stats.update_from(prefilling_update)
171 - assert stats.prefill_ts_s == prefilling_ts
172 - assert stats.num_computed_tokens == 3
173 - assert stats.num_cached_tokens == 1
174 - assert stats.queue_duration_s == prefilling_ts - queued_ts
175 -
176 - # Test DECODING
177 - decoded_update = RequestStatsUpdate(
178 - request_id=request_id,
179 - type=RequestStatsUpdate.Type.DECODING,
180 - monotonic_ts_s=decoded_ts,
181 - )
182 - stats.update_from(decoded_update)
183 - assert stats.last_updated_ts_s == decoded_ts
184 -
185 - # Test DETOKENIZED
186 - detokenized_update = RequestStatsUpdate(
187 - request_id=request_id,
188 - type=RequestStatsUpdate.Type.DETOKENIZED,
189 - monotonic_ts_s=detokenized_ts,
190 - num_new_tokens=1,
191 - )
192 - stats.update_from(detokenized_update)
193 - assert stats.last_updated_ts_s == detokenized_ts
194 - assert stats.num_output_tokens == 1
195 - # Since arrival
196 - assert stats.first_token_latency_s == detokenized_ts - arrived_ts
197 - # Since first scheduled
198 - assert stats.prefill_latency_s == detokenized_ts - prefilling_ts
199 -
200 - # Test another DECODING and DETOKENIZED should
201 - # yield correct inter token latency
202 - decoded_update = RequestStatsUpdate(
203 - request_id=request_id,
204 - type=RequestStatsUpdate.Type.DECODING,
205 - monotonic_ts_s=decoded_2_ts,
206 - )
207 - stats.update_from(decoded_update)
208 -
209 - detokenized_update = RequestStatsUpdate(
210 - request_id=request_id,
211 - type=RequestStatsUpdate.Type.DETOKENIZED,
212 - monotonic_ts_s=detokenized_2_ts,
213 - num_new_tokens=1,
214 - )
215 - stats.update_from(detokenized_update)
216 - assert stats.output_token_latency_s_lst == [
217 - detokenized_2_ts - detokenized_ts,
218 - ]
219 - assert stats.num_output_tokens == 2
220 -
221 - # Test PREEMPTED
222 - preempted_update = RequestStatsUpdate(
223 - request_id=request_id,
224 - type=RequestStatsUpdate.Type.PREEMPTED,
225 - monotonic_ts_s=preempted_ts,
226 - )
227 - stats.update_from(preempted_update)
228 - assert stats.last_updated_ts_s == preempted_ts
229 - assert stats.preempted_ts_s_lst == [preempted_ts]
230 - # States should be reset
231 - assert stats.num_computed_tokens == 0
232 - assert stats.num_cached_tokens == 0
233 - # These states should not be reset
234 - assert stats.num_output_tokens == 2
235 - assert stats.output_token_latency_s_lst == [
236 - detokenized_2_ts - detokenized_ts,
237 - ]
238 - assert stats.prefill_latency_s == prefilling_ts - arrived_ts
239 - assert stats.num_prompt_tokens == 6
240 - assert stats.prefill_start_ts_s_lst == [prefilling_ts]
241 -
242 - # Test resumed
243 - resumed_update = RequestStatsUpdate(
244 - request_id=request_id,
245 - type=RequestStatsUpdate.Type.PREFILLING,
246 - monotonic_ts_s=resumed_ts,
247 - num_computed_tokens=6,
248 - num_cached_tokens=2,
249 - )
250 - stats.update_from(resumed_update)
251 - # prefill timestamp should not be updated since it's a resumed prefill
252 - assert stats.prefill_ts_s == prefilling_ts
253 - assert stats.num_computed_tokens == 6
254 - assert stats.num_cached_tokens == 2
255 - assert stats.prefill_start_ts_s_lst == [
256 - prefilling_ts,
257 - resumed_ts,
258 - ]
259 - assert stats.last_updated_ts_s == resumed_ts
260 -
261 - # Test another DECODED/DETOKENIZED should yield correct first token latency.
262 - decoded_update = RequestStatsUpdate(
263 - request_id=request_id,
264 - type=RequestStatsUpdate.Type.DECODING,
265 - monotonic_ts_s=decoded_3_ts,
266 - )
267 - detokenized_update = RequestStatsUpdate(
268 - request_id=request_id,
269 - type=RequestStatsUpdate.Type.DETOKENIZED,
270 - monotonic_ts_s=detokenized_3_ts,
271 - num_new_tokens=1,
272 - )
273 - stats.update_from(decoded_update)
274 - stats.update_from(detokenized_update)
275 - assert stats.first_token_ts_s == detokenized_ts - arrived_ts
276 - assert stats.num_output_tokens == 3
277 - assert stats.output_token_latency_s_lst == [
278 - detokenized_2_ts - detokenized_ts,
279 - detokenized_3_ts - detokenized_2_ts,
280 - ]
281 -
282 - # Test FINISHED
283 - finished_update = RequestStatsUpdate(
284 - request_id=request_id,
285 - type=RequestStatsUpdate.Type.FINISHED,
286 - monotonic_ts_s=finished_ts,
287 - finish_reason="test_reason",
288 - )
289 - stats.update_from(finished_update)
290 - assert stats.last_updated_ts_s == finished_ts
291 - assert stats.e2e_latency_s == finished_ts - arrived_ts
292 - assert stats.inference_latency_s == finished_ts - prefilling_ts
293 - assert stats.prefill_latency_s == detokenized_ts - prefilling_ts
294 - assert stats.decode_latency_s == finished_ts - detokenized_ts
295 - assert stats.first_token_latency_s == detokenized_ts - arrived_ts
296 - assert stats.queue_duration_s == prefilling_ts - queued_ts
297 - assert stats.is_finished
298 - assert stats.finish_reason == "test_reason"
299 -
300 - # TODO(rickyx): Add model forward/execute time.
301 - assert stats.model_forward_duration_s == 0.0
302 - assert stats.model_execute_duration_s == 0.0
vllm/v1/stats/__init__.py
vllm/v1/stats/common.py deleted
-453
@@ -1,453 +0,0 @@
1 -# SPDX-License-Identifier: Apache-2.0
2 -
3 -import time
4 -from dataclasses import dataclass
5 -from dataclasses import field as dataclass_field
6 -from enum import IntEnum
7 -from typing import ClassVar, Optional
8 -
9 -import msgspec
10 -from msgspec import field as msgspec_field
11 -
12 -from vllm.sampling_params import SamplingParams
13 -
14 -
15 -class RequestStatsUpdate(
16 - msgspec.Struct, # type: ignore
17 - array_like=True,
18 - omit_defaults=True,
19 - gc=False):
20 - """
21 - An update to the request stats.
22 -
23 - This represents a stats update at a specific timestamp with metadata
24 - associated with the update.
25 -
26 - NOTE: since there might be multiple processes generating updates at
27 - different parts of the engine (e.g. input processor, scheduler, engine core,
28 - etc.), we use the monotonic timestamp to record the update to compute any
29 - intervals, and explicit wall-clock timestamp should be used for timestamps.
30 -
31 - WARNING: This assumes stats are generated in a single machine. If there are
32 - potentially multiple machines, one should always generate the stats updates
33 - on one single machine or use something else.
34 - """
35 -
36 - class Type(IntEnum):
37 - """See `RequestStats` for the lifecycle of a request."""
38 -
39 - # Request arrived at the engine frontend.
40 - ARRIVED = 0
41 - # Input processed by the input processor.
42 - INPUT_PROCESSED = 1
43 - # Queued on the engine core.
44 - QUEUED = 2
45 - # Scheduled running prefill by the scheduler.
46 - # A request could be running a new prefill on the prompt tokens or
47 - # a resumed prefill on the original prefill tokens + generated output
48 - # tokens before preemption.
49 - PREFILLING = 3
50 - # Preempted by the scheduler.
51 - PREEMPTED = 4
52 - # Output token is generated by the engine core.
53 - DECODING = 5
54 - # Token detokenized by the detokenizer.
55 - # We will record the timestamp for each output token, as well as the
56 - # finish reason.
57 - DETOKENIZED = 6
58 - # Request finishes (or aborts).
59 - FINISHED = 7
60 -
61 - """
62 - Valid state updates:
63 - ARRIVED
64 - │
65 - ├──────► INPUT_PROCESSED ──────► QUEUED ──────► PREFILLING ◄────┐
66 - │ │ │ │ │
67 - │ │ │ ▼ │
68 - │ │ │ -──► DECODING │
69 - │ │ │ | │ │
70 - │ │ │ | ▼ │
71 - │ │ │ └─ DETOKENIZED │
72 - │ │ │ │ │
73 - │ │ │ ▼ │
74 - │ ▼ ▼ PREEMPTED ◄──────┘
75 - │ │ │ │
76 - └──────────────┴───────────────────┴──────────────┴
77 - │
78 - ▼
79 - FINISHED (All could go to FINISHED)
80 - """
81 - _VALID_TRANSITIONS: ClassVar[dict[Type, set[Type]]] = {
82 - Type.ARRIVED: {
83 - Type.INPUT_PROCESSED,
84 - Type.FINISHED,
85 - },
86 - Type.INPUT_PROCESSED: {
87 - Type.QUEUED,
88 - Type.FINISHED,
89 - },
90 - Type.QUEUED: {
91 - Type.PREFILLING,
92 - Type.FINISHED,
93 - },
94 - Type.PREFILLING: {
95 - Type.DECODING,
96 - Type.PREEMPTED,
97 - Type.FINISHED,
98 - },
99 - Type.DECODING: {
100 - Type.DETOKENIZED,
101 - Type.FINISHED,
102 - },
103 - Type.DETOKENIZED: {
104 - Type.DECODING,
105 - Type.PREEMPTED,
106 - Type.FINISHED,
107 - },
108 - Type.PREEMPTED: {Type.PREFILLING, Type.FINISHED},
109 - Type.FINISHED: set(),
110 - }
111 -
112 - request_id: str
113 -
114 - type: Type
115 -
116 - # Timestamp when the update is recorded. This is used to record time
117 - # intervals between events rather than wall clock time.
118 - monotonic_ts_s: float = msgspec_field(
119 - default_factory=lambda: time.monotonic())
120 -
121 - ############################################################
122 - # Metadata associated with the update.
123 - ############################################################
124 - # For input_processed. Metadata needed for stats logging.
125 - num_prompt_tokens: Optional[int] = None
126 - sampling_params: Optional[SamplingParams] = None
127 -
128 - # For running.
129 - # Number of tokens computed when scheduled to run.
130 - num_computed_tokens: Optional[int] = None
131 - # Number of cached tokens when scheduled to run.
132 - num_cached_tokens: Optional[int] = None
133 -
134 - # For decoded.
135 - # The number of new output tokens generated.
136 - num_new_tokens: Optional[int] = None
137 -
138 - # For both detokenized and decoded.
139 - # Finished reason.
140 - finish_reason: Optional[str] = None
141 -
142 - # Non-optional fields for each update type.
143 - _REQUIRED_FIELDS: ClassVar[dict[Type, list[str]]] = {
144 - Type.INPUT_PROCESSED: ["num_prompt_tokens", "sampling_params"],
145 - Type.PREFILLING: ["num_computed_tokens", "num_cached_tokens"],
146 - Type.DETOKENIZED: ["num_new_tokens"],
147 - Type.FINISHED: ["finish_reason"],
148 - }
149 -
150 - def __post_init__(self):
151 - required_fields = self._REQUIRED_FIELDS.get(self.type, [])
152 - for field in required_fields:
153 - if getattr(self, field) is None:
154 - raise ValueError(
155 - f"Field {field} is required for update type {self.type}.")
156 -
157 - @staticmethod
158 - def check_valid_update(
159 - update: "RequestStatsUpdate",
160 - last_update_type: Optional[Type],
161 - last_updated_ts_s: Optional[float],
162 - ):
163 - if last_update_type is None:
164 - assert update.type == RequestStatsUpdate.Type.ARRIVED
165 - else:
166 - valid_cur_update_types = RequestStatsUpdate._VALID_TRANSITIONS[
167 - last_update_type]
168 - assert update.type in valid_cur_update_types, (
169 - f"Invalid update type: {update.type} for last_update_type: "
170 - f"{last_update_type}.")
171 -
172 - if last_updated_ts_s is not None:
173 - assert update.monotonic_ts_s >= last_updated_ts_s, (
174 - "Update timestamp must be monotonically increasing, but "
175 - f"last_updated_ts_s={last_updated_ts_s} and "
176 - f"update.monotonic_ts_s={update.monotonic_ts_s}.")
177 -
178 -
179 -@dataclass
180 -class RequestStats:
181 - """Stats associated with a request (`Request`)."""
182 -
183 - ############################################################
184 - # Metadata
185 - ############################################################
186 - request_id: str
187 - sampling_params: Optional[SamplingParams] = None
188 - num_prompt_tokens: Optional[int] = None
189 -
190 - ############################################################
191 - # Metrics and Stats
192 - ############################################################
193 - # Timestamp when the request was last updated.
194 - last_updated_ts_s: Optional[float] = None
195 -
196 - # Last update stats type.
197 - last_update_type: Optional[RequestStatsUpdate.Type] = None
198 -
199 - # Timestamp when the request arrived at the llm engine.
200 - arrival_ts_s: Optional[float] = None
201 -
202 - # Number of tokens cached. When part of the request prefix is cached,
203 - # this will be set.
204 - num_cached_tokens: int = 0
205 -
206 - # Number of tokens computed.
207 - num_computed_tokens: int = 0
208 -
209 - # The timestamp when the request become waiting in the queue.
210 - queued_ts_s: Optional[float] = None
211 -
212 - # When the input processor is completed.
213 - input_processor_end_ts_s: Optional[float] = None
214 -
215 - # A sorted list of timestamps when the request was scheduled to prefill.
216 - # This could be when:
217 - # 1. the request is newly scheduled, so it's a new prefill.
218 - # 2. the request was preempted and resumed. It is equivalent to running
219 - # a prefill of the original prefill tokens + generated output tokens
220 - # before preemption.
221 - prefill_start_ts_s_lst: list[float] = dataclass_field(default_factory=list)
222 -
223 - # A list of timestamps when a token is decoded by the engine core.
224 - decoding_ts_s_lst: list[float] = dataclass_field(default_factory=list)
225 -
226 - # A sorted list of timestamps for each output token.
227 - output_token_ts_s_lst: list[float] = dataclass_field(default_factory=list)
228 -
229 - # First token's timestamp.
230 - first_token_ts_s: Optional[float] = None
231 -
232 - # TODO(rickyx): we need model runner to surface these.
233 - model_forward_duration_s: float = 0.0
234 - # Includes model forward, block/sync across workers, cpu-gpu sync time
235 - # and sampling time.
236 - model_execute_duration_s: float = 0.0
237 -
238 - # A sorted list of timestamps when the request was preempted at the
239 - # scheduler.
240 - # TODO(rickyx): right now, we don't actually have a good high-level
241 - # metric to measure the impact of preemption other than observation of
242 - # large P99 TPOT. Ideally we could quantify the impact of preemption by
243 - # measuring the number of tokens re-computed due to preemption.
244 - preempted_ts_s_lst: list[float] = dataclass_field(default_factory=list)
245 -
246 - # Timestamp when the request was finished at the engine core.
247 - finished_ts_s: Optional[float] = None
248 -
249 - # Finish reason.
250 - finish_reason: Optional[str] = None
251 -
252 - ############################################################
253 - # Derived properties.
254 - ############################################################
255 - @property
256 - def prefill_ts_s(self) -> Optional[float]:
257 - """The timestamp when the request started prefilling.
258 - Since a request could be preempted in decoding and later resumed
259 - to prefill the decoded tokens, we use the first prefill start timestamp.
260 - """
261 - return (self.prefill_start_ts_s_lst[0]
262 - if self.prefill_start_ts_s_lst else None)
263 -
264 - @property
265 - def e2e_latency_s(self) -> Optional[float]:
266 - if self.finished_ts_s is None or self.arrival_ts_s is None:
267 - return None
268 - assert self.finished_ts_s >= self.arrival_ts_s
269 - return self.finished_ts_s - self.arrival_ts_s
270 -
271 - @property
272 - def queue_duration_s(self) -> Optional[float]:
273 - """How long the request was waiting to run."""
274 - if self.queued_ts_s is None or self.prefill_ts_s is None:
275 - # Either not queued or not running yet.
276 - return None
277 - assert self.queued_ts_s <= self.prefill_ts_s
278 - return self.prefill_ts_s - self.queued_ts_s
279 -
280 - @property
281 - def inference_latency_s(self) -> Optional[float]:
282 - """How long the request was running inference
283 - (prefill and decode)."""
284 - if self.finished_ts_s is None or self.prefill_ts_s is None:
285 - return None
286 - assert self.finished_ts_s >= self.prefill_ts_s
287 - return self.finished_ts_s - self.prefill_ts_s
288 -
289 - @property
290 - def first_token_latency_s(self) -> Optional[float]:
291 - if self.first_token_ts_s is None or self.arrival_ts_s is None:
292 - return None
293 - assert self.first_token_ts_s >= self.arrival_ts_s
294 - return self.first_token_ts_s - self.arrival_ts_s
295 -
296 - @property
297 - def prefill_latency_s(self) -> Optional[float]:
298 - if self.first_token_ts_s is None or self.prefill_ts_s is None:
299 - return None
300 - assert self.first_token_ts_s >= self.prefill_ts_s
301 - return self.first_token_ts_s - self.prefill_ts_s
302 -
303 - @property
304 - def decode_latency_s(self) -> Optional[float]:
305 - if self.e2e_latency_s is None or self.first_token_latency_s is None:
306 - return None
307 - assert self.e2e_latency_s >= self.first_token_latency_s
308 - return self.e2e_latency_s - self.first_token_latency_s
309 -
310 - @property
311 - def output_token_latency_s_lst(self) -> list[float]:
312 - if len(self.output_token_ts_s_lst) == 0:
313 - return []
314 - latency_s_lst = []
315 - for i in range(1, len(self.output_token_ts_s_lst)):
316 - assert (self.output_token_ts_s_lst[i]
317 - >= self.output_token_ts_s_lst[i - 1])
318 - latency_s = (self.output_token_ts_s_lst[i] -
319 - self.output_token_ts_s_lst[i - 1])
320 - latency_s_lst.append(latency_s)
321 - return latency_s_lst
322 -
323 - @property
324 - def num_output_tokens(self) -> int:
325 - return len(self.output_token_ts_s_lst)
326 -
327 - @property
328 - def is_finished(self) -> bool:
329 - return self.finished_ts_s is not None
330 -
331 - def update_from(self, update: "RequestStatsUpdate"):
332 - RequestStatsUpdate.check_valid_update(update, self.last_update_type,
333 - self.last_updated_ts_s)
334 - ts = update.monotonic_ts_s
335 - self.last_updated_ts_s = ts
336 - self.last_update_type = update.type
337 - if update.type == RequestStatsUpdate.Type.ARRIVED:
338 - self.arrival_ts_s = ts
339 - elif update.type == RequestStatsUpdate.Type.INPUT_PROCESSED:
340 - self.input_processor_end_ts_s = ts
341 - self.sampling_params = update.sampling_params
342 - self.num_prompt_tokens = update.num_prompt_tokens
343 - elif update.type == RequestStatsUpdate.Type.QUEUED:
344 - self.queued_ts_s = ts
345 - elif update.type == RequestStatsUpdate.Type.PREFILLING:
346 - self.prefill_start_ts_s_lst.append(ts)
347 - self.num_cached_tokens = update.num_cached_tokens or 0
348 - self.num_computed_tokens = update.num_computed_tokens or 0
349 - elif update.type == RequestStatsUpdate.Type.PREEMPTED:
350 - self._reset_for_preemption(ts)
351 - elif update.type == RequestStatsUpdate.Type.DECODING:
352 - self.decoding_ts_s_lst.append(ts)
353 - elif update.type == RequestStatsUpdate.Type.DETOKENIZED:
354 - self._record_detokenized_output(
355 - ts,
356 - update.num_new_tokens or 0,
357 - )
358 - elif update.type == RequestStatsUpdate.Type.FINISHED:
359 - self.finished_ts_s = ts
360 - self.finish_reason = update.finish_reason
361 - else:
362 - raise ValueError(f"Unknown update type: {update.type}")
363 -
364 - def _record_detokenized_output(
365 - self,
366 - ts_s: float,
367 - num_new_tokens: int,
368 - ):
369 - # Update if first output token is generated.
370 - if len(self.output_token_ts_s_lst) == 0:
371 - self.first_token_ts_s = ts_s
372 - assert (
373 - self.prefill_ts_s is not None
374 - ), "Request must be running before generating output tokens."
375 -
376 - # Some X new tokens were generated at the ts.
377 - self.output_token_ts_s_lst.extend([ts_s] * num_new_tokens)
378 -
379 - def _reset_for_preemption(self, ts_s: float):
380 - self.preempted_ts_s_lst.append(ts_s)
381 - # Reset the computed tokens since it might restart the prefill.
382 - self.num_computed_tokens = 0
383 - # Cached token count might also change when resumed.
384 - self.num_cached_tokens = 0
385 - # These stats don't change since they happen before request running.
386 - # - arrival_ts_s
387 - # - input_processor_end_ts_s
388 - # - sampling_params
389 - # - num_prompt_tokens
390 - # - first_token_ts_s
391 - #
392 - # These stats are accumulated over preemptions:
393 - # - output_token_ts_s_lst
394 - # - prefill_start_ts_s_lst (after preemption, it will prefill the
395 - # original prefill tokens and any output tokens generated before
396 - # preemption.)
397 -
398 -
399 -@dataclass
400 -class KVCacheStats:
401 - # KV Cache Usage in %
402 - gpu_cache_usage_sys: float = 0.0
403 - gpu_prefix_cache_hit_rate: float = 0.0
404 -
405 -
406 -@dataclass
407 -class SchedulerStats:
408 - """Stats associated with the scheduler."""
409 -
410 - # Number of requests currently running.
411 - num_running_reqs: int = 0
412 - # Number of requests currently waiting.
413 - num_waiting_reqs: int = 0
414 -
415 - kv_cache_stats: KVCacheStats = dataclass_field(
416 - default_factory=KVCacheStats)
417 -
418 -
419 -@dataclass
420 -class EngineCoreProcessStats:
421 - """Stats associated with the engine core process."""
422 -
423 - # Number of requests currently in the input queue. None if the engine core
424 - # is not running in multiprocess mode.
425 - input_queue_size: Optional[int] = None
426 - # Number of outputs currently in the output queue. None if the engine core
427 - # is not running in multiprocess mode.
428 - output_queue_size: Optional[int] = None
429 -
430 -
431 -class EngineCoreStatsSnapshot(
432 - msgspec.Struct, # type: ignore
433 - array_like=True,
434 - omit_defaults=True,
435 - gc=False):
436 - """
437 - A snapshot of the EngineCore's current stats over a period of time.
438 - """
439 -
440 - # Snapshot of the scheduler stats.
441 - scheduler_stats: SchedulerStats = msgspec_field(
442 - default_factory=SchedulerStats)
443 -
444 - # Per request stats updates.
445 - requests_stats_updates: list[RequestStatsUpdate] = msgspec_field(
446 - default_factory=list)
447 -
448 - # Engine core's queue stats.
449 - engine_core_process_stats: EngineCoreProcessStats = msgspec_field(
450 - default_factory=EngineCoreProcessStats)
451 -
452 - # TODO(rickyx): Add other components' stats,
453 - # e.g. model runner/worker and etc.