diff --git a/docs/design/metrics.md b/docs/design/metrics.md index 7b7851bb72e..93efa1c2fc4 100644 --- a/docs/design/metrics.md +++ b/docs/design/metrics.md @@ -272,8 +272,43 @@ After the wrap, every upstream `vllm:*` family — TTFT, ITL, TPOT, e2e latency, ## Logging vs. Prometheus -`OrchestratorAggregator` (in `vllm_omni/metrics/stats.py`) is the logging-oriented metrics path. It collects detailed per-request, per-stage, and per-transfer statistics and prints formatted tables to the `INFO` log. This is designed for development and debugging — individual request traces, transfer bandwidth, inter-stage timing. +`OrchestratorAggregator` (in `vllm_omni/metrics/stats.py`) is the logging-oriented metrics path. It collects detailed per-request, per-stage, and per-transfer statistics and prints formatted tables to the log (per-request tables at `DEBUG`, the `[OmniTiming]` summary at `INFO`). This is designed for development and debugging — individual request traces, transfer bandwidth, inter-stage timing. `OmniPrometheusMetrics` / `OmniModalityMetrics` / `OmniTransferMetrics` form the Prometheus-oriented path. They record aggregate counters, gauges, and histograms suitable for time-series monitoring and alerting. Both paths share the same source data (`StageRequestStats`, `TransferEdgeStats`) — `OrchestratorAggregator.record_transfer_tx/rx` in particular calls both the existing accumulator code and the Prometheus emit hook in the same method body. The two consumption models can run simultaneously without coupling. The separation follows upstream vLLM's pattern of `LoggingStatLogger` vs. `PrometheusStatLogger` — same underlying data, different consumption models. + +### LLM-stage phase split in `StageRequestStats` + +With `--log-stats`, a stage served by a vLLM engine core (AR and generation stages) adds the engine-core phase split of the finished request to its `StageRequestStats` row: + +| Field | Interval | +| ----- | -------- | +| `vllm_queued_ms` | `QUEUED` event to first `SCHEDULED` event | +| `vllm_prefill_ms` | First `SCHEDULED` event to first engine-core output (the first token for an AR stage) | +| `vllm_decode_ms` | First engine-core output to last engine-core output | +| `vllm_num_preemptions` | Number of `PREEMPTED` events | + +The intervals match upstream `FinishedRequestStats`, so time spent preempted stays inside prefill or decode. The wrapped `vllm:request_queue_time_seconds`, `vllm:request_prefill_time_seconds`, `vllm:request_decode_time_seconds`, and `vllm:request_num_preemptions` histograms, which already carry `{stage, replica}` labels, use the same intervals. Two differences: upstream subtracts an unobserved timestamp as `0.0`, and it reports a negative interval as is, while these fields leave either interval out. No Prometheus family is added. + +The first engine-core output ends prefill even when it carries no token. A generation stage that emits one output when it stops therefore reports its whole run as `vllm_prefill_ms`, with a measured `vllm_decode_ms` of `0`. + +`vllm_queued_ms` runs from the stage's engine core accepting the request to its first scheduling. For stage 0 this is scheduler wait. A downstream stage can also wait there for its input: + +- With `async_chunk`, a downstream stage that receives chunks is submitted together with stage 0, and the request stays in `WAITING_FOR_CHUNK` until the first upstream chunk arrives. Later waits for the next chunk fall inside `vllm_decode_ms`. +- A stage with `requires_full_payload_input` stays in `WAITING_FOR_INPUT` until the connector delivers the payload. + +For these stages `vllm_queued_ms` includes input wait, and with `async_chunk` so does `vllm_decode_ms`; neither measures scheduler contention alone. The split does not separate input wait. `vllm_queued_ms` is also separate from the orchestration-layer wait in `vllm_omni:request_queue_wait_s` (`pipeline_timings["queue_wait_ms"]`) and from the diffusion scheduler wait in `vllm_omni:stage_in_queue_s`. + +These intervals can overlap with other timings of the same request: the upstream stage's run, the transfer times in `TransferEdgeStats` and `vllm_omni:transfer_tx_s` / `vllm_omni:transfer_rx_s`, and stage-to-stage handoff events. Do not add them together into an end-to-end breakdown of a request unless the intervals are known not to overlap and use the same clock. + +Two more cases where an interval covers more than its name suggests: + +- A KV-transfer sender (a stage with `kv_transfer_criteria`) keeps the request running until the KV extraction is acknowledged, and the token-less `kv_ready` output it emits still advances the last-token timestamp. `vllm_decode_ms` therefore includes the wait for that acknowledgement. +- `vllm_num_preemptions` counts `PREEMPTED` events. In the default (non-windowed) path, when the chunk transfer adapter moves running requests over `max_num_seqs` back to the waiting queue, it sets `PREEMPTED` without recording an event, so that wait is not counted as a preemption and stays inside prefill or decode. + +The per-request `[StageRequestStats]` table is logged at `DEBUG` (for example `VLLM_LOGGING_LEVEL=DEBUG`); at `INFO` only the `[OmniTiming]` line is printed. + +Missing and zero are different. A field is `None` when its interval was not observed: a diffusion stage, `--log-stats` off, or an engine-core event that never arrived. A measured `0` stays `0`. In the `[StageRequestStats]` table a missing value prints as `None`, and, as for every other field, a row whose values are all zero or missing is hidden. + +For a streaming-input request, only the output that finishes the request carries the split. On an AR stage, vLLM's output processor finishes the request only after the input stream has ended; at each earlier segment stop it applies the next streaming update instead, which restarts the request stats, so the split covers the last input segment. The last segment's `QUEUED` event can arrive with the previous segment's final output, before that restart, so its `vllm_queued_ms` may be missing. A generation stage finishes on the first output that is not a segment stop and does not restart its stats between segments, so its split spans the whole request: `vllm_prefill_ms` is the first segment's, `vllm_decode_ms` runs from the first segment's first output to the last output, gaps between segments included, and `vllm_queued_ms` is missing when a later segment was queued after the first scheduling. The fields are not added to the `stage_metrics` snapshot returned to clients and benchmarks. The duplex per-response table leaves them out (`DUPLEX_STAGE_TABLE_EXCLUDE`): they describe one engine-core request, not a response turn. diff --git a/tests/engine/duplex/test_engine_session.py b/tests/engine/duplex/test_engine_session.py index a4e4b1e9d68..9fec679d148 100644 --- a/tests/engine/duplex/test_engine_session.py +++ b/tests/engine/duplex/test_engine_session.py @@ -949,6 +949,104 @@ def _capture(self: OrchestratorAggregator) -> dict[str, object]: assert [event.serving_time_to_first_output_ms for event in events] == pytest.approx([80.631, 418.829, 487.109]) +def test_end_response_omits_engine_core_phase_split_from_stage_table( + monkeypatch: pytest.MonkeyPatch, +) -> None: + phase_fields = ("vllm_queued_ms", "vllm_prefill_ms", "vllm_decode_ms", "vllm_num_preemptions") + original = OrchestratorAggregator.build_and_log_summary + captured: list[tuple[OrchestratorAggregator, dict[str, object]]] = [] + + def _capture(self: OrchestratorAggregator) -> dict[str, object]: + summary = original(self) + captured.append((self, summary)) + return summary + + monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", _capture) + session = _session(num_stages=2, log_stats=True) + response_id = session.begin_response() + for stage_id in (0, 1): + stats = _stage_stats(stage_id=stage_id) + stats.vllm_queued_ms = 1.5 + stats.vllm_prefill_ms = 20.0 + stats.vllm_decode_ms = 300.0 + stats.vllm_num_preemptions = 0 + session.observe_stage_request_stats(stage_id, stats) + session.end_response() + + assert len(captured) == 1 + aggregator, summary = captured[0] + stage_table = summary.get("stage_table") + assert isinstance(stage_table, list) and stage_table + rows = stage_table[0]["stages"] + assert isinstance(rows, list) + assert [row["stage_id"] for row in rows] == [0, 1] + assert all(field not in row for row in rows for field in phase_fields) + assert all("vllm_ttft_ms" in row for row in rows) + events = aggregator.stage_events[response_id] + assert [event.vllm_prefill_ms for event in events] == pytest.approx([20.0, 20.0]) + + +def test_end_response_fold_keeps_the_chunk_that_carries_the_phase_split( + monkeypatch: pytest.MonkeyPatch, +) -> None: + phase_fields = ("vllm_queued_ms", "vllm_prefill_ms", "vllm_decode_ms", "vllm_num_preemptions") + logged: list[OrchestratorAggregator] = [] + + def _record(self: OrchestratorAggregator) -> dict[str, object]: + logged.append(self) + return {} + + monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", _record) + session = _session(num_stages=3, log_stats=True) + response_id = session.begin_response() + + def _chunk(stage_id: int, split: tuple[float | None, float | None, float | None, int | None]): + stats = _stage_stats(stage_id=stage_id) + for field, value in zip(phase_fields, split): + setattr(stats, field, value) + return stats + + no_split = (None, None, None, None) + # Stage 0: the split arrives on the last chunk, after the first (template) chunk. + session.observe_stage_request_stats(0, _chunk(0, no_split)) + session.observe_stage_request_stats(0, _chunk(0, (1.5, 20.0, 300.0, 0))) + # Stage 1: a later chunk without a split does not erase it, and the split + # is taken from one chunk, not merged field by field. + session.observe_stage_request_stats(1, _chunk(1, (2.0, 30.0, 400.0, 1))) + session.observe_stage_request_stats(1, _chunk(1, (None, 35.0, 450.0, 0))) + session.observe_stage_request_stats(1, _chunk(1, no_split)) + # Stage 2: no chunk carries a split. + session.observe_stage_request_stats(2, _chunk(2, no_split)) + session.observe_stage_request_stats(2, _chunk(2, no_split)) + session.end_response() + + rows = logged[0].stage_events[response_id] + assert [row.stage_id for row in rows] == [0, 1, 2] + assert [tuple(getattr(row, field) for field in phase_fields) for row in rows] == [ + (1.5, 20.0, 300.0, 0), + (None, 35.0, 450.0, 0), + no_split, + ] + + +def test_http_stage_table_keeps_engine_core_phase_split() -> None: + agg = OrchestratorAggregator(num_stages=1, log_stats=True, wall_start_ts=0.0, final_stage_id_for_e2e=0) + stats = _stage_stats(stage_id=0) + stats.vllm_queued_ms = 1.5 + stats.vllm_prefill_ms = 20.0 + stats.vllm_decode_ms = 300.0 + stats.vllm_num_preemptions = 0 + agg.on_stage_metrics(0, "r1", stats) + agg.on_finalize_request(0, "r1", req_start_ts=0.0) + + summary = agg.build_and_log_summary() + stage_table = summary.get("stage_table") + assert isinstance(stage_table, list) and stage_table + row = stage_table[0]["stages"][0] + assert row["vllm_prefill_ms"] == pytest.approx(20.0) + assert row["vllm_num_preemptions"] == 0 + + def test_http_stage_table_keeps_serving_time_to_first_output() -> None: agg = OrchestratorAggregator(num_stages=2, log_stats=True, wall_start_ts=0.0, final_stage_id_for_e2e=1) agg.on_stage_metrics(0, "r1", _stage_stats(stage_id=0, serving_time_to_first_output_ms=80.0)) diff --git a/tests/engine/test_orchestrator.py b/tests/engine/test_orchestrator.py index 88aff38ca80..0393363a28e 100644 --- a/tests/engine/test_orchestrator.py +++ b/tests/engine/test_orchestrator.py @@ -1810,6 +1810,71 @@ def pop_native_text_metrics(self, request_id: str) -> dict[str, Any]: assert metrics.finish_reason == "length" +@pytest.mark.parametrize( + ("native_record", "expected"), + [ + pytest.param( + {"vllm_queued_ms": 2.5, "vllm_prefill_ms": 30.0, "vllm_decode_ms": 0.0, "vllm_num_preemptions": 0}, + {"vllm_queued_ms": 2.5, "vllm_prefill_ms": 30.0, "vllm_decode_ms": 0.0, "vllm_num_preemptions": 0}, + id="observed", + ), + pytest.param( + # An absent key is an unobserved phase and must not become 0. + {"vllm_decode_ms": 12.0}, + {"vllm_queued_ms": None, "vllm_prefill_ms": None, "vllm_decode_ms": 12.0, "vllm_num_preemptions": None}, + id="partly_unobserved", + ), + ], +) +def test_stage_pool_metrics_copy_native_phase_split(native_record, expected) -> None: + class PhaseMetricsOutputProcessor(FakeOutputProcessor): + def pop_native_text_metrics(self, request_id: str) -> dict[str, Any]: + assert request_id == "req-ar" + return {"num_generation_tokens": 2, **native_record} + + pool = StagePool(0, [FakeStageClient(stage_type="llm")], output_processor=PhaseMetricsOutputProcessor()) + + metrics = pool.build_stage_metrics( + [_build_request_output("req-ar")], + submit_ts=time.time(), + request_timestamp=time.time(), + replica_id=0, + ) + + assert {key: getattr(metrics, key) for key in expected} == expected + + +def test_stage_pool_metrics_skip_native_phase_split_when_metrics_disabled() -> None: + popped: list[str] = [] + + class PhaseMetricsOutputProcessor(FakeOutputProcessor): + def pop_native_text_metrics(self, request_id: str) -> dict[str, Any]: + popped.append(request_id) + return { + "num_generation_tokens": 2, + "vllm_queued_ms": 2.5, + "vllm_prefill_ms": 30.0, + "vllm_decode_ms": 12.0, + "vllm_num_preemptions": 1, + } + + pool = StagePool(0, [FakeStageClient(stage_type="llm")], output_processor=PhaseMetricsOutputProcessor()) + pool.collect_stage_metrics = False + + metrics = pool.build_stage_metrics( + [_build_request_output("req-ar")], + submit_ts=time.time(), + request_timestamp=time.time(), + replica_id=0, + ) + + # The record is still popped so it does not leak, but the optional split stays unset. + assert popped == ["req-ar"] + assert metrics.num_tokens_out == 2 + assert metrics.vllm_queued_ms is None and metrics.vllm_prefill_ms is None + assert metrics.vllm_decode_ms is None and metrics.vllm_num_preemptions is None + + def test_image_ttfo_preserves_request_time_and_tracks_stage_time() -> None: stage = FakeStageClient(stage_type="diffusion", final_output=True, final_output_type="image") pool = StagePool( @@ -1830,6 +1895,9 @@ def test_image_ttfo_preserves_request_time_and_tracks_stage_time() -> None: assert metrics.serving_time_to_first_output_ms == pytest.approx(12000.0) assert metrics.image_time_to_first_output_ms == pytest.approx(2000.0) + # A diffusion stage has no engine-core phase split. + assert metrics.vllm_queued_ms is None and metrics.vllm_prefill_ms is None + assert metrics.vllm_decode_ms is None and metrics.vllm_num_preemptions is None @pytest.mark.asyncio diff --git a/tests/engine/test_output_processor.py b/tests/engine/test_output_processor.py index d3e54786c56..405d31a362a 100644 --- a/tests/engine/test_output_processor.py +++ b/tests/engine/test_output_processor.py @@ -13,8 +13,15 @@ from vllm.outputs import PoolingRequestOutput from vllm.sampling_params import RequestOutputKind from vllm.v1.engine import EngineCoreEvent, EngineCoreEventType, FinishReason +from vllm.v1.engine.detokenizer import IncrementalDetokenizer +from vllm.v1.engine.logprobs import LogprobsProcessor from vllm.v1.engine.output_processor import OutputProcessor as VLLMOutputProcessor -from vllm.v1.metrics.stats import IterationStats, LoRARequestStates, PrefillStats +from vllm.v1.metrics.stats import ( + IterationStats, + LoRARequestStates, + PrefillStats, + RequestStateStats, +) from vllm.v1.serial_utils import MsgpackDecoder, MsgpackEncoder from vllm_omni.engine import OmniEngineCoreOutput @@ -945,6 +952,408 @@ def test_mm_only_outputs_update_iteration_stats(): assert finished.num_generation_tokens == 2 +_PHASE_KEYS = ("vllm_queued_ms", "vllm_prefill_ms", "vllm_decode_ms", "vllm_num_preemptions") + + +@pytest.mark.parametrize( + ("timestamps", "expected"), + [ + pytest.param( + dict(num_preemptions=1, queued_ts=100.0, scheduled_ts=100.25, first_token_ts=100.75, last_token_ts=102.0), + {"vllm_queued_ms": 250.0, "vllm_prefill_ms": 500.0, "vllm_decode_ms": 1250.0, "vllm_num_preemptions": 1}, + id="all_events", + ), + pytest.param( + # A one-token output and an immediate schedule are measured zeros. + dict(queued_ts=10.0, scheduled_ts=10.0, first_token_ts=10.5, last_token_ts=10.5), + {"vllm_queued_ms": 0.0, "vllm_prefill_ms": 500.0, "vllm_decode_ms": 0.0, "vllm_num_preemptions": 0}, + id="measured_zero", + ), + pytest.param(dict(), {}, id="nothing_observed"), + pytest.param( + # No QUEUED/SCHEDULED: "first_token_ts - 0.0" would invent a prefill. + dict(first_token_ts=50.0, last_token_ts=50.5), + {"vllm_decode_ms": 500.0}, + id="no_engine_core_events", + ), + pytest.param( + dict(scheduled_ts=5.0, first_token_ts=6.0, last_token_ts=7.0), + {"vllm_prefill_ms": 1000.0, "vllm_decode_ms": 1000.0, "vllm_num_preemptions": 0}, + id="no_queued_event", + ), + pytest.param( + # Finished (e.g. aborted) before it was ever scheduled. + dict(queued_ts=7.0), + {"vllm_num_preemptions": 0}, + id="never_scheduled", + ), + pytest.param( + # The first output was processed without stats, so no first token. + dict(queued_ts=1.0, scheduled_ts=2.0, last_token_ts=3.0), + {"vllm_queued_ms": 1000.0, "vllm_num_preemptions": 0}, + id="no_first_token", + ), + pytest.param( + dict(queued_ts=5.0, scheduled_ts=4.0, first_token_ts=6.0, last_token_ts=5.5), + {"vllm_prefill_ms": 2000.0, "vllm_num_preemptions": 0}, + id="negative_intervals", + ), + ], +) +def test_native_phase_split_omits_unobserved_intervals(timestamps, expected): + metrics = output_processor._native_phase_metrics(RequestStateStats(**timestamps)) + + assert metrics == pytest.approx(expected) + + +def _make_text_stage_state(*, stream_input: bool = False) -> OmniRequestState: + # vLLM's no-tokenizer detokenizer keeps the output on the AR text path + # (upstream process_outputs) instead of the multimodal-only path. + return OmniRequestState( + **{ + **_DEFAULT_STATE_KWARGS, + "detokenizer": IncrementalDetokenizer(), + "logprobs_processor": LogprobsProcessor( + tokenizer=None, + logprobs=None, + prompt_logprobs=None, + cumulative_logprob=None, + num_logprobs=None, + num_prompt_logprobs=None, + ), + }, + output_kind=RequestOutputKind.CUMULATIVE, + stream_input=stream_input, + ) + + +def _stats_processor(state: OmniRequestState) -> MultimodalOutputProcessor: + processor = MultimodalOutputProcessor(tokenizer=None, log_stats=True, engine_core_output_type=AUDIO) + processor.request_states[state.request_id] = state + processor.external_req_ids[state.external_req_id].append(state.request_id) + return processor + + +def _stats_output( + *, events=None, finish_reason=None, is_segment_finished=False, new_token_ids=(10,) +) -> OmniEngineCoreOutput: + return OmniEngineCoreOutput( + request_id="r", + new_token_ids=list(new_token_ids), + events=events, + finish_reason=finish_reason, + is_segment_finished=is_segment_finished, + multimodal_output={AUDIO: torch.ones(1, 4)}, + ) + + +@pytest.mark.parametrize( + "make_state", + [_make_text_stage_state, lambda: _make_no_detok_state(RequestOutputKind.CUMULATIVE)], + ids=["ar_text_path", "mm_only_path"], +) +def test_engine_core_events_reach_native_phase_split(make_state): + state = make_state() + processor = _stats_processor(state) + prefill = _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 1.0), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 1.25), + ] + ) + processor.process_outputs([prefill], engine_core_timestamp=1.75, iteration_stats=IterationStats()) + decode = _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.PREEMPTED, 2.0), + # Re-scheduling after preemption must not move the prefill start. + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 2.5), + ], + finish_reason=FinishReason.LENGTH, + ) + iteration_stats = IterationStats() + processor.process_outputs([decode], engine_core_timestamp=3.0, iteration_stats=iteration_stats) + + record = processor.pop_native_text_metrics("r") + + assert {key: record[key] for key in _PHASE_KEYS} == pytest.approx( + {"vllm_queued_ms": 250.0, "vllm_prefill_ms": 500.0, "vllm_decode_ms": 1250.0, "vllm_num_preemptions": 1} + ) + finished = iteration_stats.finished_requests[0] + assert record["vllm_queued_ms"] == pytest.approx(finished.queued_time * 1000.0) + assert record["vllm_prefill_ms"] == pytest.approx(finished.prefill_time * 1000.0) + assert record["vllm_decode_ms"] == pytest.approx(finished.decode_time * 1000.0) + assert record["vllm_num_preemptions"] == finished.num_preemptions + + +def test_native_phase_split_with_coalesced_decode_output(): + # A final audio stage folds token-only decode steps into its next emitted + # output (is_coalesced). The scheduler never holds the output that carries + # prefill stats, and a coalesced output is stamped when it is emitted, so + # the split is unchanged; only its ITL sample is dropped. + state = _make_no_detok_state(RequestOutputKind.CUMULATIVE) + processor = _stats_processor(state) + coalesced = _stats_output(new_token_ids=(11, 12, 13)) + coalesced.is_coalesced = True + steps = [ + ( + 1.75, + _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 1.0), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 1.25), + ] + ), + ), + (2.0, _stats_output()), + (2.5, coalesced), + (3.0, _stats_output(finish_reason=FinishReason.STOP)), + ] + itls = [] + for timestamp, output in steps: + iteration_stats = IterationStats() + processor.process_outputs([output], engine_core_timestamp=timestamp, iteration_stats=iteration_stats) + itls += iteration_stats.inter_token_latencies_iter + + record = processor.pop_native_text_metrics("r") + + assert _phase_split(record) == pytest.approx( + {"vllm_queued_ms": 250.0, "vllm_prefill_ms": 500.0, "vllm_decode_ms": 1250.0, "vllm_num_preemptions": 0} + ) + finished = iteration_stats.finished_requests[0] + assert record["vllm_decode_ms"] == pytest.approx(finished.decode_time * 1000.0) + assert finished.num_generation_tokens == 6 + assert itls == pytest.approx([0.25, 0.5]) + + +def test_one_shot_generation_phase_split_reports_whole_run_as_prefill(): + # A generation stage that emits a single token-less output at stop: the + # first engine-core output sets the first-token timestamp, so the whole + # run is prefill and decode is a measured 0. + state = _make_no_detok_state(RequestOutputKind.CUMULATIVE) + processor = _stats_processor(state) + only = _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 1.0), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 1.25), + ], + finish_reason=FinishReason.STOP, + new_token_ids=(), + ) + processor.process_outputs([only], engine_core_timestamp=2.0, iteration_stats=IterationStats()) + + record = processor.pop_native_text_metrics("r") + + assert {key: record[key] for key in _PHASE_KEYS} == pytest.approx( + {"vllm_queued_ms": 250.0, "vllm_prefill_ms": 750.0, "vllm_decode_ms": 0.0, "vllm_num_preemptions": 0} + ) + + +def test_first_output_without_stats_leaves_prefill_and_decode_out(): + # The orchestrator flushes a codec stage's first audio frame without + # iteration stats, so the native first-token timestamp is never set. + processor = _stats_processor(_make_no_detok_state(RequestOutputKind.CUMULATIVE)) + processor.process_outputs([_stats_output()], engine_core_timestamp=1.5, iteration_stats=None) + processor.process_outputs( + [ + _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 1.0), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 1.25), + ], + finish_reason=FinishReason.STOP, + ) + ], + engine_core_timestamp=2.0, + iteration_stats=IterationStats(), + ) + + record = processor.pop_native_text_metrics("r") + + assert {key: record[key] for key in _PHASE_KEYS if key in record} == pytest.approx( + {"vllm_queued_ms": 250.0, "vllm_num_preemptions": 0} + ) + + +@pytest.mark.parametrize( + ("stats_on", "is_segment_finished"), + [(False, False), (True, True)], + ids=["stats_off", "segment_finish"], +) +def test_no_native_phase_split_without_finished_request_stats(stats_on, is_segment_finished): + # Only a terminal finish with stats on is a finished vLLM request. + processor = _stats_processor(_make_no_detok_state(RequestOutputKind.CUMULATIVE)) + processor.process_outputs( + [ + _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 1.0), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 1.25), + ], + finish_reason=FinishReason.STOP, + is_segment_finished=is_segment_finished, + ) + ], + engine_core_timestamp=1.75, + iteration_stats=IterationStats() if stats_on else None, + ) + + record = processor.pop_native_text_metrics("r") + + assert not set(_PHASE_KEYS) & set(record) + + +def _streaming_input(arrival_time: float, *, resumable: bool = True) -> SimpleNamespace: + # The EngineCoreRequest fields the output processor reads for a request + # that is already registered: the next input segment, or (resumable=False) + # the end of the input stream. + return SimpleNamespace(request_id="r", resumable=resumable, prompt_token_ids=[1], arrival_time=arrival_time) + + +def _phase_split(record: dict) -> dict: + return {key: record[key] for key in _PHASE_KEYS if key in record} + + +@pytest.mark.parametrize("next_input_before_stop", [True, False], ids=["queued_before_stop", "after_stop"]) +def test_text_path_phase_split_covers_last_streaming_segment(next_input_before_stop): + # On the AR text path, vLLM's process_outputs applies the next streaming + # update at a segment stop instead of finishing the request, and the + # update restarts the request stats. + processor = _stats_processor(_make_text_stage_state(stream_input=True)) + processor.process_outputs( + [ + _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 1.0), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 1.25), + ] + ) + ], + engine_core_timestamp=1.75, + iteration_stats=IterationStats(), + ) + stop_events = None + if next_input_before_stop: + processor.add_request(_streaming_input(2.9), prompt=None) + # The scheduler starts the queued segment at the stop, so its QUEUED + # event rides on segment 1's last output, before the stats restart. + stop_events = [EngineCoreEvent(EngineCoreEventType.QUEUED, 3.0)] + stop_stats = IterationStats() + processor.process_outputs( + [_stats_output(events=stop_events, finish_reason=FinishReason.STOP, is_segment_finished=True)], + engine_core_timestamp=3.0, + iteration_stats=stop_stats, + ) + segment_1 = processor.pop_native_text_metrics("r") + + segment_2_events = [EngineCoreEvent(EngineCoreEventType.SCHEDULED, 4.0)] + if not next_input_before_stop: + processor.add_request(_streaming_input(3.4), prompt=None) + segment_2_events.insert(0, EngineCoreEvent(EngineCoreEventType.QUEUED, 3.5)) + processor.add_request(_streaming_input(4.1, resumable=False), prompt=None) + processor.process_outputs( + [_stats_output(events=segment_2_events)], engine_core_timestamp=4.5, iteration_stats=IterationStats() + ) + final_stats = IterationStats() + processor.process_outputs( + [_stats_output(finish_reason=FinishReason.STOP)], engine_core_timestamp=6.0, iteration_stats=final_stats + ) + segment_2 = processor.pop_native_text_metrics("r") + + assert not stop_stats.finished_requests and not _phase_split(segment_1) + expected = {"vllm_prefill_ms": 500.0, "vllm_decode_ms": 1500.0, "vllm_num_preemptions": 0} + if not next_input_before_stop: + expected["vllm_queued_ms"] = 500.0 + assert _phase_split(segment_2) == pytest.approx(expected) + assert len(final_stats.finished_requests) == 1 + + +def test_text_path_stop_after_input_end_carries_phase_split(): + # The end of input reaches the output processor before the engine core, + # so the engine core still reports the last segment's stop as a segment + # stop. vLLM's process_outputs finishes the request on it, and the split + # matches the FinishedRequestStats recorded there. + processor = _stats_processor(_make_text_stage_state(stream_input=True)) + processor.process_outputs( + [ + _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 1.0), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 1.25), + ] + ) + ], + engine_core_timestamp=1.75, + iteration_stats=IterationStats(), + ) + processor.add_request(_streaming_input(2.0, resumable=False), prompt=None) + iteration_stats = IterationStats() + processor.process_outputs( + [_stats_output(finish_reason=FinishReason.STOP, is_segment_finished=True)], + engine_core_timestamp=3.0, + iteration_stats=iteration_stats, + ) + record = processor.pop_native_text_metrics("r") + + assert "r" not in processor.request_states + assert _phase_split(record) == pytest.approx( + {"vllm_queued_ms": 250.0, "vllm_prefill_ms": 500.0, "vllm_decode_ms": 1250.0, "vllm_num_preemptions": 0} + ) + (finished,) = iteration_stats.finished_requests + assert record["vllm_prefill_ms"] == pytest.approx(finished.prefill_time * 1000.0) + assert record["vllm_decode_ms"] == pytest.approx(finished.decode_time * 1000.0) + + +def test_generation_stage_phase_split_spans_streaming_segments(): + # The multimodal-only path does not apply streaming updates, so the stats + # of a generation stage do not restart between segments. + state = OmniRequestState(**_NO_DETOK_STATE_KWARGS, output_kind=RequestOutputKind.CUMULATIVE, stream_input=True) + processor = _stats_processor(state) + processor.process_outputs( + [ + _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 1.0), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 1.25), + ] + ) + ], + engine_core_timestamp=1.75, + iteration_stats=IterationStats(), + ) + processor.process_outputs( + [_stats_output(finish_reason=FinishReason.STOP, is_segment_finished=True)], + engine_core_timestamp=3.0, + iteration_stats=IterationStats(), + ) + segment_1 = processor.pop_native_text_metrics("r") + processor.add_request(_streaming_input(3.4), prompt=None) + processor.add_request(_streaming_input(4.1, resumable=False), prompt=None) + processor.process_outputs( + [ + _stats_output( + events=[ + EngineCoreEvent(EngineCoreEventType.QUEUED, 3.5), + EngineCoreEvent(EngineCoreEventType.SCHEDULED, 4.0), + ] + ) + ], + engine_core_timestamp=4.5, + iteration_stats=IterationStats(), + ) + processor.process_outputs( + [_stats_output(finish_reason=FinishReason.STOP)], engine_core_timestamp=6.0, iteration_stats=IterationStats() + ) + record = processor.pop_native_text_metrics("r") + + assert not _phase_split(segment_1) + # Prefill is segment 1's; decode runs from segment 1's first output to the + # last output; segment 2's QUEUED comes after the first SCHEDULED, so the + # queue interval would be negative and is left out. + assert _phase_split(record) == pytest.approx( + {"vllm_prefill_ms": 500.0, "vllm_decode_ms": 4250.0, "vllm_num_preemptions": 0} + ) + + def test_ec_transfer_params_survive_both_construction_paths(): """v0.28 contract: encoder-cache transfer metadata must reach RequestOutput.ec_transfer_params through BOTH omni construction paths — diff --git a/tests/metrics/test_stats.py b/tests/metrics/test_stats.py index f72cd086772..5192513daf6 100644 --- a/tests/metrics/test_stats.py +++ b/tests/metrics/test_stats.py @@ -194,3 +194,78 @@ def record_table(*args, **kwargs): agg.log_timing_summary() assert not tables assert len(lines) == 1 and lines[0].startswith("[OmniTiming] req=r1 total=") + + +_PHASE_FIELDS = ("vllm_queued_ms", "vllm_prefill_ms", "vllm_decode_ms", "vllm_num_preemptions") + + +def _phase_split_summary(**phase_split) -> dict: + agg = OrchestratorAggregator(num_stages=2, log_stats=True, wall_start_ts=0.0, final_stage_id_for_e2e=1) + for stage_id, split in ((0, phase_split), (1, {})): # stage 1 stands in for a diffusion stage + agg.on_stage_metrics( + stage_id, + "r1", + StageRequestStats( + batch_id=1, + batch_size=1, + num_tokens_in=0, + num_tokens_out=4, + stage_gen_time_ms=20.0, + rx_transfer_bytes=0, + rx_decode_time_ms=0.0, + rx_in_flight_time_ms=0.0, + stage_stats=StageStats(), + **split, + ), + ) + agg.on_finalize_request(1, "r1", req_start_ts=0.0) + return agg.build_and_log_summary() + + +def _logged_stage_table_rows(monkeypatch, **phase_split) -> dict[str, list[str]]: + from vllm_omni.metrics import stats + + tables: list[str] = [] + monkeypatch.setattr(stats.logger, "isEnabledFor", lambda level: True) + monkeypatch.setattr(stats.logger, "debug", lambda msg, *args: tables.append(msg % args)) + monkeypatch.setattr(stats.logger, "info", lambda *args, **kwargs: None) + _phase_split_summary(**phase_split) + table = next(table for table in tables if "[StageRequestStats [request_id=r1]]" in table) + rows = {} + for line in table.splitlines(): + cells = [cell.strip() for cell in line.strip().strip("|").split("|")] + rows[cells[0]] = cells[1:] + return rows + + +def test_stage_table_summary_keeps_measured_zero_apart_from_missing_phase_split() -> None: + summary = _phase_split_summary(vllm_queued_ms=3.0, vllm_prefill_ms=40.0, vllm_decode_ms=0.0, vllm_num_preemptions=0) + + ar_row, diffusion_row = _get_request_entry(summary["stage_table"], "r1")["stages"] + assert {field: ar_row[field] for field in _PHASE_FIELDS} == { + "vllm_queued_ms": 3.0, + "vllm_prefill_ms": 40.0, + "vllm_decode_ms": 0.0, + "vllm_num_preemptions": 0, + } + assert all(diffusion_row[field] is None for field in _PHASE_FIELDS) + + +def test_logged_stage_table_renders_phase_split_rows(monkeypatch) -> None: + rows = _logged_stage_table_rows( + monkeypatch, vllm_queued_ms=3.0, vllm_prefill_ms=40.0, vllm_decode_ms=900.0, vllm_num_preemptions=2 + ) + + assert {field: rows[field] for field in _PHASE_FIELDS} == { + "vllm_queued_ms": ["3.000", "None"], + "vllm_prefill_ms": ["40.000", "None"], + "vllm_decode_ms": ["900.000", "None"], + "vllm_num_preemptions": ["2", "None"], + } + + +def test_logged_stage_table_has_no_phase_split_rows_when_unobserved(monkeypatch) -> None: + rows = _logged_stage_table_rows(monkeypatch) + + assert "stage_gen_time_ms" in rows + assert not set(_PHASE_FIELDS) & set(rows) diff --git a/vllm_omni/engine/stage_pool.py b/vllm_omni/engine/stage_pool.py index 1318f7772e0..cd0ffc38533 100644 --- a/vllm_omni/engine/stage_pool.py +++ b/vllm_omni/engine/stage_pool.py @@ -757,6 +757,11 @@ def build_stage_metrics( stage_metrics.vllm_tpot_ms = float(native_text_metrics.get("vllm_tpot_ms") or 0.0) stage_metrics.vllm_itl_ms = float(native_text_metrics.get("vllm_itl_ms") or 0.0) stage_metrics.vllm_itls_ms = list(native_text_metrics.get("vllm_itls_ms") or []) + # An absent key means the phase was not observed; keep it None, not 0. + stage_metrics.vllm_queued_ms = native_text_metrics.get(defs.VLLM_QUEUED_MS) + stage_metrics.vllm_prefill_ms = native_text_metrics.get(defs.VLLM_PREFILL_MS) + stage_metrics.vllm_decode_ms = native_text_metrics.get(defs.VLLM_DECODE_MS) + stage_metrics.vllm_num_preemptions = native_text_metrics.get(defs.VLLM_NUM_PREEMPTIONS) return stage_metrics def _infer_output_unit_type(self, request_outputs: list[Any], *, token_count: int) -> str: diff --git a/vllm_omni/metrics/definitions.py b/vllm_omni/metrics/definitions.py index 1a7db543150..18961de62b0 100644 --- a/vllm_omni/metrics/definitions.py +++ b/vllm_omni/metrics/definitions.py @@ -133,6 +133,10 @@ VLLM_TPOT_MS = "vllm_tpot_ms" VLLM_ITL_MS = "vllm_itl_ms" VLLM_ITLS_MS = "vllm_itls_ms" +VLLM_QUEUED_MS = "vllm_queued_ms" +VLLM_PREFILL_MS = "vllm_prefill_ms" +VLLM_DECODE_MS = "vllm_decode_ms" +VLLM_NUM_PREEMPTIONS = "vllm_num_preemptions" NUM_TOKENS_IN = "num_tokens_in" NUM_TOKENS_OUT = "num_tokens_out" AUDIO_SAMPLE_RATE = "audio_sample_rate" diff --git a/vllm_omni/metrics/stats.py b/vllm_omni/metrics/stats.py index 80894caf19f..5b807878a1f 100644 --- a/vllm_omni/metrics/stats.py +++ b/vllm_omni/metrics/stats.py @@ -69,6 +69,14 @@ class StageRequestStats: vllm_tpot_ms: float = 0.0 vllm_itl_ms: float = 0.0 vllm_itls_ms: list[float] | None = None + # Engine-core phase split of a finished vLLM request: queued -> scheduled + # -> first token -> last token, as in vLLM's FinishedRequestStats. + # None means not observed (diffusion stage, --log-stats off, or a missing + # engine-core event); 0 is a measured value. + vllm_queued_ms: float | None = None + vllm_prefill_ms: float | None = None + vllm_decode_ms: float | None = None + vllm_num_preemptions: int | None = None @property def rx_mbps(self) -> float: @@ -135,8 +143,25 @@ def e2e_tpt(self) -> float: # Duplex logs one StageRequestStats table per response. Chunk submits refresh # ``request_timestamp``, so serving_time_to_first_output_ms is often a clamped # 0 on the audio column and is not a useful TTFP. HTTP ``--print-stage`` keeps -# the row. -DUPLEX_STAGE_TABLE_EXCLUDE = frozenset({defs.SERVING_TIME_TO_FIRST_OUTPUT_MS}) +# the row. The engine-core phase split describes one vLLM request, not a +# response turn, so it stays out of the duplex table until turn semantics are +# defined; the non-duplex ``[StageRequestStats]`` table keeps it. +DUPLEX_STAGE_TABLE_EXCLUDE = frozenset( + { + defs.SERVING_TIME_TO_FIRST_OUTPUT_MS, + defs.VLLM_QUEUED_MS, + defs.VLLM_PREFILL_MS, + defs.VLLM_DECODE_MS, + defs.VLLM_NUM_PREEMPTIONS, + } +) +# StageRequestStats fields of the engine-core phase split. +_PHASE_SPLIT_FIELDS = ( + defs.VLLM_QUEUED_MS, + defs.VLLM_PREFILL_MS, + defs.VLLM_DECODE_MS, + defs.VLLM_NUM_PREEMPTIONS, +) TRANSFER_EXCLUDE = {"from_stage", "to_stage", "request_id", "used_shm"} E2E_EXCLUDE = {"request_id"} @@ -223,7 +248,16 @@ def _one_row_per_stage(events: list[StageRequestStats]) -> list[StageRequestStat tpot_ms = _weighted_tpot_ms(chunks_by_stage[sid]) if tpot_ms is not None: merged[defs.VLLM_TPOT_MS] = tpot_ms - rows.append(_apply_merged_stage_stats(templates[sid], merged)) + row = _apply_merged_stage_stats(templates[sid], merged) + # Only the chunk that finished an engine-core request carries its phase + # split, and the template (first chunk) usually does not. Copy the split + # of the last chunk that has one, without mixing fields across chunks. + for chunk in reversed(chunks_by_stage[sid]): + if any(getattr(chunk, field) is not None for field in _PHASE_SPLIT_FIELDS): + for field in _PHASE_SPLIT_FIELDS: + setattr(row, field, getattr(chunk, field)) + break + rows.append(row) return rows diff --git a/vllm_omni/outputs/output_processor.py b/vllm_omni/outputs/output_processor.py index 2873560892e..19cd5491d54 100644 --- a/vllm_omni/outputs/output_processor.py +++ b/vllm_omni/outputs/output_processor.py @@ -21,6 +21,7 @@ from vllm.v1.metrics.stats import IterationStats, RequestStateStats from vllm_omni.data_entry_keys import unflatten_payload +from vllm_omni.metrics import definitions as defs from vllm_omni.outputs import OmniRequestOutput from vllm_omni.outputs.mm_outputs import MultimodalCompletionOutput, MultimodalPayload from vllm_omni.outputs.multimodal_accumulation import ( @@ -86,6 +87,32 @@ def _accumulate_segment_tpot(record: dict[str, object], *, elapsed_ms: float, ne record["vllm_tpot_ms"] = total_elapsed_ms / float(total_intervals) +def _native_phase_metrics(stats: RequestStateStats) -> dict[str, float | int]: + """Return the observed queue/prefill/decode split of a finished request. + + The intervals match vLLM's ``FinishedRequestStats`` (queued_time, + prefill_time, decode_time). vLLM leaves a timestamp at 0.0 when its + engine-core event never arrived; subtracting it would invent a duration, + so such intervals, and any that would be negative, are omitted instead + of reported as 0. + """ + metrics: dict[str, float | int] = {} + queued_ts, scheduled_ts = stats.queued_ts, stats.scheduled_ts + first_ts, last_ts = stats.first_token_ts, stats.last_token_ts + if queued_ts > 0 and scheduled_ts >= queued_ts: + metrics[defs.VLLM_QUEUED_MS] = (scheduled_ts - queued_ts) * 1000.0 + if scheduled_ts > 0 and first_ts >= scheduled_ts: + metrics[defs.VLLM_PREFILL_MS] = (first_ts - scheduled_ts) * 1000.0 + if first_ts > 0 and last_ts >= first_ts: + metrics[defs.VLLM_DECODE_MS] = (last_ts - first_ts) * 1000.0 + if queued_ts > 0 or scheduled_ts > 0: + # PREEMPTED rides the same engine-core event stream as QUEUED and + # SCHEDULED, so a zero count only means something once that stream + # was observed. + metrics[defs.VLLM_NUM_PREEMPTIONS] = int(stats.num_preemptions) + return metrics + + class OmniRequestState(RequestState): """Request state for omni models, tracking multimodal outputs. @@ -889,6 +916,7 @@ def _update_stats_from_finished( req_stats=native_stats, num_cached_tokens=req_state.num_cached_tokens, ) + self._native_text_metric_record(req_state.external_req_id).update(_native_phase_metrics(native_stats)) if not iteration_stats.finished_requests: return