diff --git a/tests/v1/core/test_scheduler.py b/tests/v1/core/test_scheduler.py index 157170400b0..da1e0c5e76c 100644 --- a/tests/v1/core/test_scheduler.py +++ b/tests/v1/core/test_scheduler.py @@ -43,6 +43,65 @@ from .utils import EOS_TOKEN_ID, create_requests, create_scheduler, mock_kv pytestmark = pytest.mark.cpu_test +def test_make_scheduled_encoder_input_stats_output_embeddings(): + scheduler = create_scheduler() + mm_features = [ + MultiModalFeatureSpec( + data=MultiModalKwargsItem.dummy(), + modality="image", + identifier="image-0", + mm_position=PlaceholderRange(offset=0, length=196), + ), + MultiModalFeatureSpec( + data=MultiModalKwargsItem.dummy(), + modality="video", + identifier="video-0", + mm_position=PlaceholderRange(offset=200, length=196), + ), + MultiModalFeatureSpec( + data=MultiModalKwargsItem.dummy(), + modality="audio", + identifier="audio-0", + mm_position=PlaceholderRange(offset=400, length=49), + ), + ] + scheduler.requests["req"] = Mock(mm_features=mm_features) + + stats = scheduler._make_scheduled_encoder_input_stats({"req": [0, 1, 2]}) + + assert stats is not None + assert stats.num_inputs == 3 + assert stats.output_tokens == 441 + + +def test_scheduled_encoder_input_stats_disabled_without_iteration_logging( + monkeypatch: pytest.MonkeyPatch, +): + scheduler = create_scheduler() + make_stats = Mock(side_effect=AssertionError("stats should not be computed")) + monkeypatch.setattr(scheduler, "_make_scheduled_encoder_input_stats", make_stats) + + scheduler_output = scheduler.schedule() + + make_stats.assert_not_called() + assert scheduler_output.scheduled_encoder_input_stats is None + + +def test_scheduled_encoder_input_stats_disabled_without_log_stats( + monkeypatch: pytest.MonkeyPatch, +): + scheduler = create_scheduler() + scheduler.log_stats = False + scheduler.observability_config.enable_logging_iteration_details = True + make_stats = Mock(side_effect=AssertionError("stats should not be computed")) + monkeypatch.setattr(scheduler, "_make_scheduled_encoder_input_stats", make_stats) + + scheduler_output = scheduler.schedule() + + make_stats.assert_not_called() + assert scheduler_output.scheduled_encoder_input_stats is None + + def test_add_requests(): scheduler = create_scheduler() requests = create_requests(num_requests=10) @@ -108,6 +167,29 @@ def test_schedule(enable_prefix_caching: bool, prompt_logprobs: int | None): assert scheduler.running[i] == request +def test_scheduler_stats_route_to_existing_output_client(): + scheduler = create_scheduler() + request = create_requests(num_requests=1)[0] + request.client_index = 1 + scheduler.add_request(request) + + scheduler_output = scheduler.schedule() + model_output = ModelRunnerOutput( + req_ids=[request.request_id], + req_id_to_index={request.request_id: 0}, + sampled_token_ids=[[1000]], + logprobs=None, + prompt_logprobs_dict={}, + pooler_output=[], + ) + + engine_core_outputs = scheduler.update_from_output(scheduler_output, model_output) + + assert 0 not in engine_core_outputs + assert engine_core_outputs[1].scheduler_stats is not None + assert len(engine_core_outputs[1].outputs) == 1 + + def test_schedule_multimodal_requests(): scheduler = create_scheduler(model="llava-hf/llava-1.5-7b-hf") mm_positions = [[PlaceholderRange(offset=i, length=100)] for i in range(10)] diff --git a/tests/v1/engine/test_iteration_logging.py b/tests/v1/engine/test_iteration_logging.py new file mode 100644 index 00000000000..5a08308163e --- /dev/null +++ b/tests/v1/engine/test_iteration_logging.py @@ -0,0 +1,88 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +import time +from types import SimpleNamespace + +from vllm.v1.engine import EngineCoreOutputs +from vllm.v1.engine.core import EngineCore +from vllm.v1.metrics.stats import SchedulerIterationDetails, SchedulerStats + + +class FakeEngineCore: + def _make_iteration_details_stats( + self, iteration_details: SchedulerIterationDetails + ) -> SchedulerStats: + return SchedulerStats(iteration_details=iteration_details) + + +def make_iteration_details() -> SchedulerIterationDetails: + return SchedulerIterationDetails( + iteration_index=1, + num_ctx_requests=2, + num_ctx_tokens=3, + num_generation_requests=4, + num_generation_tokens=5, + elapsed_ms=6.7, + ) + + +def make_fake_engine(log_stats: bool = True) -> SimpleNamespace: + return SimpleNamespace( + log_stats=log_stats, + vllm_config=SimpleNamespace( + observability_config=SimpleNamespace( + enable_logging_iteration_details=True, + ) + ), + ) + + +def test_capture_iteration_details_disabled_without_log_stats(): + engine = make_fake_engine(log_stats=False) + + with EngineCore.capture_iteration_details(engine, None) as iteration_details: + assert iteration_details is None + + assert not hasattr(engine, "_iteration_index") + + +def test_capture_iteration_details_fills_elapsed_time(): + engine = make_fake_engine() + + with EngineCore.capture_iteration_details(engine, None) as iteration_details: + assert iteration_details is not None + assert iteration_details.elapsed_ms == 0.0 + assert iteration_details.is_dummy + time.sleep(0.001) + + assert iteration_details is not None + assert iteration_details.elapsed_ms > 0.0 + assert engine._iteration_index == 1 + + +def test_attach_iteration_details_uses_existing_output(): + iteration_details = make_iteration_details() + outputs = { + 2: EngineCoreOutputs(scheduler_stats=SchedulerStats()), + 1: EngineCoreOutputs(scheduler_stats=SchedulerStats()), + } + + EngineCore._attach_iteration_details(FakeEngineCore(), outputs, iteration_details) + + assert 0 not in outputs + assert outputs[2].scheduler_stats is not None + assert outputs[2].scheduler_stats.iteration_details == iteration_details + assert outputs[1].scheduler_stats is not None + assert outputs[1].scheduler_stats.iteration_details is None + + +def test_attach_iteration_details_falls_back_to_client_zero_without_outputs(): + iteration_details = make_iteration_details() + outputs: dict[int, EngineCoreOutputs] = {} + + EngineCore._attach_iteration_details(FakeEngineCore(), outputs, iteration_details) + + assert set(outputs) == {0} + assert outputs[0].scheduler_stats is not None + assert outputs[0].scheduler_stats.iteration_details == iteration_details diff --git a/tests/v1/metrics/test_stats.py b/tests/v1/metrics/test_stats.py index 21f496ea4ae..0de74f0faaa 100644 --- a/tests/v1/metrics/test_stats.py +++ b/tests/v1/metrics/test_stats.py @@ -1,12 +1,17 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project -from vllm.v1.engine import FinishReason +from vllm.v1.core.sched.output import ScheduledEncoderInputStats, SchedulerOutput +from vllm.v1.engine import EngineCoreOutputs, FinishReason from vllm.v1.metrics.stats import ( IterationStats, PrefillStats, PromptTokenStats, RequestStateStats, + SchedulerIterationDetails, + SchedulerStats, ) +from vllm.v1.serial_utils import MsgpackDecoder, MsgpackEncoder +from vllm.v1.utils import compute_iteration_details def test_iteration_stats_repr(): @@ -14,6 +19,45 @@ def test_iteration_stats_repr(): assert repr(iteration_stats).startswith("IterationStats(") +def test_scheduler_iteration_details_serialization(): + iteration_details = SchedulerIterationDetails( + iteration_index=1, + num_ctx_requests=2, + num_ctx_tokens=3, + num_generation_requests=4, + num_generation_tokens=5, + elapsed_ms=6.7, + num_encoder_inputs=2, + num_encoder_output_tokens=392, + ) + outputs = EngineCoreOutputs( + scheduler_stats=SchedulerStats( + kv_cache_usage=0.5, + iteration_details=iteration_details, + ) + ) + + encoded = MsgpackEncoder().encode(outputs) + decoded = MsgpackDecoder(EngineCoreOutputs).decode(encoded) + + assert decoded.scheduler_stats is not None + assert decoded.scheduler_stats.kv_cache_usage == 0.5 + assert decoded.scheduler_stats.iteration_details == iteration_details + + +def test_compute_iteration_details_includes_encoder_stats(): + scheduler_output = SchedulerOutput.make_empty() + scheduler_output.scheduled_encoder_input_stats = ScheduledEncoderInputStats( + num_inputs=2, + output_tokens=392, + ) + + iteration_details = compute_iteration_details(scheduler_output) + + assert iteration_details.num_encoder_inputs == 2 + assert iteration_details.num_encoder_output_tokens == 392 + + def test_prefill_kv_computed_with_cache(): """Test that prefill KV compute correctly excludes cached tokens.""" iteration_stats = IterationStats() diff --git a/vllm/v1/core/sched/output.py b/vllm/v1/core/sched/output.py index 4401fb050b3..5667482e948 100644 --- a/vllm/v1/core/sched/output.py +++ b/vllm/v1/core/sched/output.py @@ -179,6 +179,14 @@ class CachedRequestData: ) +@dataclass +class ScheduledEncoderInputStats: + """Stats for encoder inputs scheduled in one iteration.""" + + num_inputs: int = 0 + output_tokens: int = 0 + + @dataclass class SchedulerOutput: # list of the requests that are scheduled for the first time. @@ -216,6 +224,8 @@ class SchedulerOutput: # freed from the encoder cache. free_encoder_mm_hashes: list[str] + scheduled_encoder_input_stats: ScheduledEncoderInputStats | None = None + # Request IDs that are preempted in this step. # Only used for v2 model runner. preempted_req_ids: set[str] | None = None diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py index 371139fa722..bde6af0e66b 100644 --- a/vllm/v1/core/sched/scheduler.py +++ b/vllm/v1/core/sched/scheduler.py @@ -44,6 +44,7 @@ from vllm.v1.core.sched.output import ( CachedRequestData, GrammarOutput, NewRequestData, + ScheduledEncoderInputStats, SchedulerOutput, ) from vllm.v1.core.sched.request_queue import ( @@ -1121,6 +1122,15 @@ class Scheduler(SchedulerInterface): len(num_scheduled_tokens) ] + scheduled_encoder_input_stats = None + if ( + self.log_stats + and self.observability_config.enable_logging_iteration_details + ): + scheduled_encoder_input_stats = self._make_scheduled_encoder_input_stats( + scheduled_encoder_inputs + ) + scheduler_output = SchedulerOutput( scheduled_new_reqs=new_reqs_data, scheduled_cached_reqs=cached_reqs_data, @@ -1128,6 +1138,7 @@ class Scheduler(SchedulerInterface): total_num_scheduled_tokens=total_num_scheduled_tokens, scheduled_spec_decode_tokens=scheduled_spec_decode_tokens, scheduled_encoder_inputs=scheduled_encoder_inputs, + scheduled_encoder_input_stats=scheduled_encoder_input_stats, num_common_prefix_blocks=num_common_prefix_blocks, preempted_req_ids=self.reset_preempted_req_ids, # finished_req_ids is an existing state in the scheduler, @@ -1506,6 +1517,23 @@ class Scheduler(SchedulerInterface): external_load_encoder_input, ) + def _make_scheduled_encoder_input_stats( + self, scheduled_encoder_inputs: dict[str, list[int]] + ) -> ScheduledEncoderInputStats | None: + stats = ScheduledEncoderInputStats() + + for req_id, input_ids in scheduled_encoder_inputs.items(): + request = self.requests.get(req_id) + if request is None: + continue + + for input_id in input_ids: + mm_feature = request.mm_features[input_id] + stats.num_inputs += 1 + stats.output_tokens += mm_feature.mm_position.get_num_embeds() + + return stats if stats.num_inputs else None + def get_grammar_bitmask( self, scheduler_output: SchedulerOutput ) -> GrammarOutput | None: @@ -1877,7 +1905,10 @@ class Scheduler(SchedulerInterface): if ( stats := self.make_stats( - spec_decoding_stats, kv_connector_stats, cudagraph_stats, perf_stats + spec_decoding_stats, + kv_connector_stats, + cudagraph_stats, + perf_stats, ) ) is not None: # Return stats to only one of the front-ends. diff --git a/vllm/v1/engine/core.py b/vllm/v1/engine/core.py index 301c5892a58..383853807db 100644 --- a/vllm/v1/engine/core.py +++ b/vllm/v1/engine/core.py @@ -79,16 +79,17 @@ from vllm.v1.engine.utils import ( ) from vllm.v1.executor import Executor from vllm.v1.kv_cache_interface import KVCacheConfig, get_kv_cache_spec_kind -from vllm.v1.metrics.stats import SchedulerStats +from vllm.v1.metrics.stats import SchedulerIterationDetails, SchedulerStats from vllm.v1.outputs import ModelRunnerOutput from vllm.v1.request import Request, RequestStatus from vllm.v1.serial_utils import MsgpackDecoder, MsgpackEncoder from vllm.v1.structured_output import StructuredOutputManager -from vllm.v1.utils import IterationDetails, compute_iteration_details +from vllm.v1.utils import compute_iteration_details from vllm.version import __version__ as VLLM_VERSION logger = init_logger(__name__) + HANDSHAKE_TIMEOUT_MINS = 5 _R = TypeVar("_R") # Return type for collective_rpc @@ -498,45 +499,74 @@ class EngineCore: raise err @contextmanager - def log_iteration_details(self, scheduler_output: SchedulerOutput | None): - if not self.vllm_config.observability_config.enable_logging_iteration_details: - yield + def capture_iteration_details( + self, scheduler_output: SchedulerOutput | None + ) -> Generator[SchedulerIterationDetails | None, None, None]: + enable_details = ( + self.vllm_config.observability_config.enable_logging_iteration_details + ) + if not self.log_stats or not enable_details: + yield None return # 0-token step: let the dummy_batch wrapper log it (avoids double-log). - if scheduler_output and scheduler_output.total_num_scheduled_tokens == 0: - yield + if ( + scheduler_output is not None + and scheduler_output.total_num_scheduled_tokens == 0 + ): + yield None return - self._iteration_index = getattr(self, "_iteration_index", 0) + + iteration_index = getattr(self, "_iteration_index", 0) # scheduler_output=None marks a DP dummy iteration. if scheduler_output is None: - iteration_details = IterationDetails(0, 0, 0, 0) - is_dummy = True - else: - iteration_details = compute_iteration_details(scheduler_output) - is_dummy = False - before = time.monotonic() - yield - logger.info( - "".join( - [ - "Iteration(", - str(self._iteration_index), - "): ", - str(iteration_details.num_ctx_requests), - " context requests, ", - str(iteration_details.num_ctx_tokens), - " context tokens, ", - str(iteration_details.num_generation_requests), - " generation requests, ", - str(iteration_details.num_generation_tokens), - " generation tokens, iteration elapsed time: ", - format((time.monotonic() - before) * 1000, ".2f"), - " ms", - " (dummy)" if is_dummy else "", - ] + iteration_details = SchedulerIterationDetails( + iteration_index=iteration_index, + num_ctx_requests=0, + num_ctx_tokens=0, + num_generation_requests=0, + num_generation_tokens=0, + elapsed_ms=0.0, + is_dummy=True, ) - ) - self._iteration_index += 1 + else: + details = compute_iteration_details(scheduler_output) + iteration_details = SchedulerIterationDetails( + iteration_index=iteration_index, + num_ctx_requests=details.num_ctx_requests, + num_ctx_tokens=details.num_ctx_tokens, + num_generation_requests=details.num_generation_requests, + num_generation_tokens=details.num_generation_tokens, + elapsed_ms=0.0, + num_encoder_inputs=details.num_encoder_inputs, + num_encoder_output_tokens=details.num_encoder_output_tokens, + ) + + start_time = time.monotonic() + yield iteration_details + iteration_details.elapsed_ms = (time.monotonic() - start_time) * 1000 + self._iteration_index = iteration_index + 1 + + def _make_iteration_details_stats( + self, iteration_details: SchedulerIterationDetails + ) -> SchedulerStats: + stats = self.scheduler.make_stats() or SchedulerStats() + stats.iteration_details = iteration_details + return stats + + def _attach_iteration_details( + self, + outputs: dict[int, EngineCoreOutputs], + iteration_details: SchedulerIterationDetails | None, + ) -> None: + if iteration_details is None: + return + + if (eco := next(iter(outputs.values()), None)) is None: + outputs[0] = eco = EngineCoreOutputs() + if eco.scheduler_stats is None: + eco.scheduler_stats = self._make_iteration_details_stats(iteration_details) + else: + eco.scheduler_stats.iteration_details = iteration_details def _should_throttle_prefills(self) -> bool: """Whether to defer new prefills this step (DP prefill balancing). @@ -558,8 +588,8 @@ class EngineCore: future = self.model_executor.execute_model(scheduler_output, non_block=True) grammar_output = self.scheduler.get_grammar_bitmask(scheduler_output) with ( + self.capture_iteration_details(scheduler_output) as iteration_details, self.log_error_detail(scheduler_output), - self.log_iteration_details(scheduler_output), ): model_output = future.result() if model_output is None: @@ -571,6 +601,7 @@ class EngineCore: engine_core_outputs = self.scheduler.update_from_output( scheduler_output, model_output ) + self._attach_iteration_details(engine_core_outputs, iteration_details) return engine_core_outputs, scheduler_output.total_num_scheduled_tokens > 0 @@ -656,8 +687,8 @@ class EngineCore: # Block until the next result is available. future, scheduler_output, exec_model_fut = batch_queue.pop() with ( + self.capture_iteration_details(scheduler_output) as iteration_details, self.log_error_detail(scheduler_output), - self.log_iteration_details(scheduler_output), ): model_output = future.result() if model_output is None: @@ -672,6 +703,7 @@ class EngineCore: engine_core_outputs = self.scheduler.update_from_output( scheduler_output, model_output ) + self._attach_iteration_details(engine_core_outputs, iteration_details) # NOTE(nick): We can either handle the deferred tasks here or save # in a field and do it immediately once step_with_batch_queue is @@ -2019,8 +2051,13 @@ class DPEngineCoreProc(EngineCoreProc): # Execute a dummy pass when no ready requests ran, unless the # engine is sleeping. elif not self.model_executor.is_sleeping: - with self.log_iteration_details(None): + with self.capture_iteration_details(None) as iteration_details: self.execute_dummy_batch() + if iteration_details is not None and not self.has_coordinator: + stats = self._make_iteration_details_stats(iteration_details) + self.output_queue.put_nowait( + (0, EngineCoreOutputs(scheduler_stats=stats)) + ) # 3) All-reduce operation to determine global unfinished reqs. self.engines_running = self._has_global_unfinished_reqs( diff --git a/vllm/v1/metrics/loggers.py b/vllm/v1/metrics/loggers.py index 021019dc1cd..692106cf396 100644 --- a/vllm/v1/metrics/loggers.py +++ b/vllm/v1/metrics/loggers.py @@ -160,6 +160,42 @@ class LoggingStatLogger(StatLoggerBase): def log_prefix(self): return "Engine {:03d}: ".format(self.engine_index) + def _log_prefix_for_engine(self, engine_idx: int) -> str: + if self.engine_index == engine_idx: + return self.log_prefix + return "Engine {:03d}: ".format(engine_idx) + + def _log_iteration_details( + self, scheduler_stats: SchedulerStats, engine_idx: int + ) -> None: + details = scheduler_stats.iteration_details + if details is None: + return + + encoder_msg = "" + if details.num_encoder_inputs: + encoder_msg = ( + f", encoder inputs: {details.num_encoder_inputs}, " + f"encoder output embeddings: {details.num_encoder_output_tokens}" + ) + + logger.info( + "%sIteration(%d): %d context requests, %d context tokens, " + "%d generation requests, %d generation tokens, " + "iteration elapsed time: %.2f ms%s, " + "GPU KV cache usage: %.1f%%%s", + self._log_prefix_for_engine(engine_idx), + details.iteration_index, + details.num_ctx_requests, + details.num_ctx_tokens, + details.num_generation_requests, + details.num_generation_tokens, + details.elapsed_ms, + " (dummy)" if details.is_dummy else "", + scheduler_stats.kv_cache_usage * 100, + encoder_msg, + ) + def record( self, scheduler_stats: SchedulerStats | None, @@ -172,6 +208,7 @@ class LoggingStatLogger(StatLoggerBase): self._track_iteration_stats(iteration_stats) if scheduler_stats is not None: + self._log_iteration_details(scheduler_stats, engine_idx) self.prefix_caching_metrics.observe(scheduler_stats.prefix_cache_stats) if scheduler_stats.connector_prefix_cache_stats is not None: diff --git a/vllm/v1/metrics/stats.py b/vllm/v1/metrics/stats.py index a7a5fb7a2d2..20bb3e1caa6 100644 --- a/vllm/v1/metrics/stats.py +++ b/vllm/v1/metrics/stats.py @@ -167,6 +167,21 @@ class KVCacheEvictionEvent: reuse_gaps_seconds: tuple[float, ...] +@dataclass +class SchedulerIterationDetails: + """Scheduler-side details for one engine iteration.""" + + iteration_index: int + num_ctx_requests: int + num_ctx_tokens: int + num_generation_requests: int + num_generation_tokens: int + elapsed_ms: float + num_encoder_inputs: int = 0 + num_encoder_output_tokens: int = 0 + is_dummy: bool = False + + @dataclass class SchedulerStats: """Stats associated with the scheduler.""" @@ -181,6 +196,7 @@ class SchedulerStats: current_wave: int = 0 kv_cache_usage: float = 0.0 + iteration_details: SchedulerIterationDetails | None = None prefix_cache_stats: PrefixCacheStats = field(default_factory=PrefixCacheStats) connector_prefix_cache_stats: PrefixCacheStats | None = None diff --git a/vllm/v1/utils.py b/vllm/v1/utils.py index b485d838f3e..e17ffed9340 100644 --- a/vllm/v1/utils.py +++ b/vllm/v1/utils.py @@ -782,12 +782,16 @@ class IterationDetails: num_ctx_tokens: int num_generation_requests: int num_generation_tokens: int + num_encoder_inputs: int = 0 + num_encoder_output_tokens: int = 0 def __repr__(self) -> str: return f"IterationDetails(num_ctx_requests={self.num_ctx_requests},\ num_ctx_tokens={self.num_ctx_tokens}, \ num_generation_requests={self.num_generation_requests}, \ - num_generation_tokens={self.num_generation_tokens})" + num_generation_tokens={self.num_generation_tokens}, \ + num_encoder_inputs={self.num_encoder_inputs}, \ + num_encoder_output_tokens={self.num_encoder_output_tokens})" def compute_iteration_details(scheduler_output: SchedulerOutput) -> IterationDetails: @@ -818,9 +822,18 @@ def compute_iteration_details(scheduler_output: SchedulerOutput) -> IterationDet else: num_generation_requests += 1 num_generation_tokens += num_tokens + scheduled_encoder_input_stats = scheduler_output.scheduled_encoder_input_stats + num_encoder_inputs = 0 + num_encoder_output_tokens = 0 + if scheduled_encoder_input_stats is not None: + num_encoder_inputs = scheduled_encoder_input_stats.num_inputs + num_encoder_output_tokens = scheduled_encoder_input_stats.output_tokens + return IterationDetails( num_context_requests, num_context_tokens, num_generation_requests, num_generation_tokens, + num_encoder_inputs, + num_encoder_output_tokens, )