[TRTLLM-6904][feat] add timing metrics for input processing and multimodal encoding

docs/source/features/multi-modality.md

@@ -74,6 +74,52 @@ You can then send OpenAI-compatible requests, such as via curl or API clients, t
 
 `trtllm-serve` renders the chat template itself and marks the resulting `TextPrompt` with `chat_template_applied`, as do the multimodal prompt builders in `tensorrt_llm.inputs.utils`. An input processor that would otherwise apply the chat template to a raw `prompt` should use such a prompt verbatim; see `TextPrompt` in `tensorrt_llm/inputs/data.py`.
 
+### Input processing and encoder metrics
+
+Enable the existing metrics options in the serving YAML:
+
+```yaml
+return_perf_metrics: true
+enable_iter_perf_stats: true
+```
+
+`/prometheus/metrics` exposes these histograms; `/metrics` serves iteration statistics as JSON.
+
+| Metric | Measurement |
+| --- | --- |
+| `trtllm_input_processing_seconds{stage="total"}` | Wall time from entry into a chat, completion, encoder, Harmony or Responses handler through synchronous input preparation and request submission, before waiting for generation. Includes message parsing, rendering, media resolution, preprocessing and their waits. |
+| `trtllm_input_processing_seconds{stage="media_load"}` | Time awaiting the common media resolver, including concurrent downloads, decodes and worker waits. |
+| `trtllm_input_processing_seconds{stage="chat_template"}` | Time awaiting chat template rendering, when rendering is needed. |
+| `trtllm_input_processing_seconds{stage="processor"}` | Input processor wall time inside its worker, including tokenization, model preprocessing and hashing. |
+| `trtllm_input_processing_queue_seconds` | Wait for an input processor worker, separate from processor execution. |
+| `trtllm_media_processing_seconds{stage="download", modality="image"}` | Per-item HTTP(S) download wall time, excluding shared HTTP session initialization. Also supports `audio` and `video`. Local and inline data inputs do not create download observations. |
+| `trtllm_media_processing_seconds{stage="decode", modality="image"}` | Per-item decode execution in the common media worker, separate from queue time. Local file decoding includes file IO; inline decoding includes base64 conversion. |
+| `trtllm_media_processing_seconds{stage="decode_queue", modality="image"}` | Per-item wait for a media decode worker. |
+| `trtllm_multimodal_encoder_gpu_time_seconds` | Rank-0 CUDA-event elapsed time per encoder call through `MultimodalModelMixin`, including item scheduling, inline encoding and side-stream prefetch. |
+
+Input timings are captured in a request context independently of Prometheus. When a route emits a completed executor metrics record, it also carries them in `perf_metrics.input_processing_metrics` in the existing JSONL output (`perf_metrics_output_dir`). Values are seconds: request-level stages use keys such as `total`, `media_load`, `chat_template`, `processor` and `processor_queue`. Per-item times are summed by stage and modality within each request, for example `media_download_image` and `media_decode_image`; these sums can exceed elapsed request time when items overlap. Batched completion inputs share the HTTP request's frontend timings. Anthropic Message Batch items have separate request contexts and JSONL records.
+
+`/health_generate` probes do not contribute to the input-processing histograms; their request-local timings and existing executor metrics remain available. For Anthropic Messages, `total` starts in the delegated chat handler and excludes Anthropic request conversion and validation. Shared HTTP session initialization remains included in `media_load` and `total`.
+
+Encoder-only servers expose the same Prometheus endpoint and completed-request input timings when metrics are enabled. Applications that configure a `ProcessPoolExecutor` for media loading retain the direct worker submission path; per-worker `decode` and `decode_queue` timings are available only for thread executors.
+
+For selected requests, send `X-TRTLLM-return-metrics: 1` to receive the same values in `Server-Timing`, for example `server_input_total;dur=150.000000` (milliseconds). Streaming responses use the existing `trtllm.perf_metrics` SSE event. These fields survive context/generation worker handoff and remain separate from executor timing and batch-level encoder GPU measurements. Stage histograms include failed attempts; request JSONL records follow the existing completed-request reporting path.
+
+Processor metrics cover the chat and text completion paths that offload preprocessing to the input processor pool. Encoder metrics exclude warmup, cache-only lookups and H2D transfers performed before the encoder call. Calls using precomputed embeddings or legacy paths outside `MultimodalModelMixin` do not produce encoder observations. The count is encoder calls, not requests or media items. The timings describe rank 0; they are not a sum or maximum across GPUs. Completed encoder observations may appear in a later iteration's JSON field, `multimodalEncoderStats.gpuTimesMS`, because collection does not wait for the GPU.
+
+Media loading and chat rendering can overlap. Encoder work can overlap LLM execution. Do not add these histogram averages to reconstruct TTFT or divide a shared encoder call across requests. Use client TTFT and throughput alongside the stage distributions to identify bottlenecks.
+
+[AIPerf server metrics](https://docs.nvidia.com/aiperf/server-metrics/server-metrics-collection) can scrape these metrics without a TRT-LLM-specific adapter. Add the explicit endpoint to your profile command:
+
+```bash
+aiperf profile --model MODEL --url http://localhost:8000 \
+  --endpoint-type chat --endpoint /v1/chat/completions \
+  --server-metrics http://localhost:8000/prometheus/metrics \
+  --server-metrics-formats json csv jsonl
+```
+
+The histogram sum and count provide total measured time and mean; percentile values are bucket estimates. Grafana can query the same Prometheus series.
+
 ### Run with [`trtllm-bench`](../../source/commands/trtllm-bench.rst)
 
 Evaluate offline inference performance with multimodal inputs using the `trtllm-bench` tool. For detailed instructions, see the [benchmarking guide](../developer-guide/perf-benchmarking.md).

tensorrt_llm/_torch/models/modeling_multimodal_mixin.py

@@ -433,6 +433,7 @@ class MultimodalModelMixin:
 
     model_config: ModelConfig
     _multimodal_encoder_cache: Optional[TensorLRUCache] = None
+    _encoder_forward_context: Callable[[], contextlib.AbstractContextManager[None]] | None = None
 
     @classmethod
     def _cast_multimodal_encoder_dtype(
@@ -482,6 +483,16 @@ def validate_multimodal_request_data(self, mm_data: dict) -> None:
         """
         return None
 
+    def _run_multimodal_encoder(
+        self,
+        multimodal_params: Sequence[MultimodalParams],
+    ) -> torch.Tensor:
+        """Run encoder work with optional runtime instrumentation after H2D."""
+        if self._encoder_forward_context is None:
+            return self.encode_multimodal_inputs(multimodal_params)
+        with self._encoder_forward_context():
+            return self.encode_multimodal_inputs(multimodal_params)
+
     def prepare_multimodal_encoder_inputs(
         self,
         selected_items: Sequence[tuple[MultimodalParams, int]],
@@ -599,7 +610,7 @@ def forward_multimodal_encoder_items(
         def flush_group() -> None:
             if not group_params:
                 return
-            embeddings = self.encode_multimodal_inputs(group_params)
+            embeddings = self._run_multimodal_encoder(group_params)
             expected_length = sum(group_lengths)
             if embeddings.shape[0] != expected_length:
                 raise MultimodalEncoderContractError(
@@ -1133,7 +1144,7 @@ def _get_or_encode_multimodal_embeddings(
             self._encode_with_partial_cache(partial_hits, encoder_cache)
 
         embeddings = get_multimodal_embeddings(
-            encoder_forward_fn=self.encode_multimodal_inputs,
+            encoder_forward_fn=self._run_multimodal_encoder,
             multimodal_params=list(multimodal_params),
         )
         if encoder_cache is not None:
@@ -1435,7 +1446,7 @@ def _encode_with_partial_cache(
         )
         # Keep one call across requests and modalities so EncoderGroup can
         # batch modalities that share an encoder.
-        batched_output = self.encode_multimodal_inputs([param for param, _, _ in encoder_inputs])
+        batched_output = self._run_multimodal_encoder([param for param, _, _ in encoder_inputs])
         output_lengths = [length for _, lengths, _ in encoder_inputs for length in lengths]
         encoded_items = iter(torch.split(batched_output, output_lengths, dim=0))
 
@@ -1734,7 +1745,7 @@ def _dispatch_cross_iter_prefetch(
                 if partial_hits and encoder_cache is not None:
                     model._encode_with_partial_cache(partial_hits, encoder_cache)
                 if cache_misses:
-                    encoder_output = model.encode_multimodal_inputs(cache_misses).clone()
+                    encoder_output = model._run_multimodal_encoder(cache_misses).clone()
                     _store_chunked_prefill_embeddings(cache_misses, [encoder_output])
                     if encoder_cache is not None:
                         for param in cache_misses:

tensorrt_llm/_torch/pyexecutor/perf_metrics_manager.py

@@ -8,6 +8,7 @@
 readability and separation of concerns.
 """
 
+from collections.abc import Iterator
 from contextlib import contextmanager
 from types import SimpleNamespace
 from typing import Optional
@@ -33,6 +34,8 @@ def __init__(self, enabled: bool):
         self._perf_events = None
         self._perf_event_idx = 0
         self._forward_event_pool = []
+        self._encoder_events: list[tuple[torch.cuda.Event, torch.cuda.Event]] = []
+        self._encoder_gpu_times_ms: list[float] = []
 
     # ------------------------------------------------------------------
     # GPU event helpers
@@ -74,6 +77,44 @@ def release_forward_timing_events(self, start_event, end_event) -> None:
         if start_event is not None and end_event is not None:
             self._forward_event_pool.append((start_event, end_event))
 
+    @contextmanager
+    def record_encoder_forward(self) -> Iterator[None]:
+        """Time one encoder call on the current stream, including graph replay."""
+        self._collect_encoder_gpu_times()
+        start_event, end_event = self.borrow_forward_timing_events()
+        start_event.record()
+        try:
+            yield
+        finally:
+            end_event.record()
+            self._encoder_events.append((start_event, end_event))
+
+    def take_encoder_gpu_times_ms(self) -> list[float]:
+        """Drain completed encoder calls without waiting for a side stream.
+
+        Pending events stay owned here until they complete; they must not be
+        reused by a later encoder or LLM forward before their timing is read.
+        """
+        self._collect_encoder_gpu_times()
+        completed, self._encoder_gpu_times_ms = self._encoder_gpu_times_ms, []
+        return completed
+
+    def _collect_encoder_gpu_times(self) -> None:
+        # Reclaim completed events even when iteration stats are downsampled.
+        pending = []
+        for start_event, end_event in self._encoder_events:
+            try:
+                if not end_event.query():
+                    pending.append((start_event, end_event))
+                    continue
+                elapsed_ms = start_event.elapsed_time(end_event)
+            except RuntimeError as error:
+                logger.warning("Failed to compute encoder GPU elapsed time: %s", error)
+            else:
+                self._encoder_gpu_times_ms.append(float(elapsed_ms))
+                self.release_forward_timing_events(start_event, end_event)
+        self._encoder_events = pending
+
     @contextmanager
     def record_perf_events(
         self, start_event: Optional[torch.cuda.Event], end_event: Optional[torch.cuda.Event]

tensorrt_llm/_torch/pyexecutor/py_executor.py

@@ -73,8 +73,8 @@
 from ..disaggregation.orchestration.transfer_manager import AsyncTransferManager
 from ..distributed import Distributed
 from ..distributed.communicator import ReduceOp
-from ..models.modeling_multimodal_mixin import \
-    maybe_prefetch_mm_encoder_for_next_iter
+from ..models.modeling_multimodal_mixin import (
+    MultimodalModelMixin, maybe_prefetch_mm_encoder_for_next_iter)
 from ..models.modeling_utils import DecoderModelForCausalLM
 from ..modules.decoder_layer import DecoderLayer
 from ..modules.dwdp.dense.manager import DenseDwdpManager
@@ -1372,6 +1372,9 @@ def _initialize_execution(self, resources: PyExecutorResources) -> None:
             from tensorrt_llm._torch.late_compile import mark_server_ready
             mark_server_ready()
 
+        # Install only after warmup/capture so dummy encoder work is excluded.
+        self._setup_encoder_perf_metrics()
+
         # Snapshot some cumulative KV cache counters so that stats reported to
         # users exclude blocks reused and missed during warmup dummy requests.
         if hasattr(self.kv_cache_manager, 'snapshot_warmup_baseline'):
@@ -1642,6 +1645,15 @@ def rebind_resources(self, resources: PyExecutorResources) -> None:
             self.shutdown()
             raise
 
+    def _setup_encoder_perf_metrics(self) -> None:
+        """Install optional encoder timing on the module that owns encoder methods."""
+        # torch.compile delegates encoder methods with self bound to _orig_mod.
+        model = getattr(self.model_engine.model, "_orig_mod",
+                        self.model_engine.model)
+        if (self.enable_iter_perf_stats and self.dist.rank == 0
+                and isinstance(model, MultimodalModelMixin)):
+            model._encoder_forward_context = self.perf_manager.record_encoder_forward
+
     @property
     def is_generation_model(self) -> bool:
         """Whether requests generate tokens, rather than finish after one forward."""
@@ -3580,6 +3592,16 @@ def _append_iter_stats(self,
                            or self.dist.pp_size > 1)
         scheduler_mode = "overlap" if is_overlap_like else "non_overlap"
 
+        # These are completed rank-0 encoder calls, including side-stream
+        # prefetch. They can finish after the iteration that launched them.
+        # Attach them once, rather than copying them to every ADP fanout row.
+        encoder_gpu_times_ms = (self.perf_manager.take_encoder_gpu_times_ms()
+                                if attention_dp_rank in (None, 0) else [])
+
+        mm_encoder_stats = {
+            "gpuTimesMS": encoder_gpu_times_ms
+        } if encoder_gpu_times_ms else None
+
         tp_size = getattr(self.dist, "tp_size", 1)
         gather_all_ranks = os.environ.get("TLLM_METRICS_ALL_RANKS", "0") == "1"
         if (gather_all_ranks and self.enable_iter_perf_stats and tp_size > 1
@@ -3597,6 +3619,8 @@ def _append_iter_stats(self,
                 local_dict["prevDeviceStepTimeMS"] = prev_device_step_time_ms
             if gpu_forward_time_ms is not None:
                 local_dict["gpuForwardTimeMS"] = gpu_forward_time_ms
+            if mm_encoder_stats is not None:
+                local_dict["multimodalEncoderStats"] = mm_encoder_stats
             local_dict["schedulerMode"] = scheduler_mode
             local_dict["rank"] = self.dist.tp_rank
             # Buffer for _flush_iter_stats_synced; in-place tp_allgather would
@@ -3615,14 +3639,15 @@ def _append_iter_stats(self,
         #   [5] prev_device_step_time_ms: Optional[float]
         #   [6] scheduler_mode: "overlap" | "non_overlap"
         #   [7] gpu_forward_time_ms: Optional[float]
+        #   [8] mm_encoder_stats: optional completed rank-0 encoder observations
         with self.stats_lock:
             if (not _stats_buffer_is_unbounded(self.max_stats_len)
                     and len(self.stats) > self.max_stats_len):
                 self.stats.pop(0)
             self.stats.append(
                 (stats, req_stats, kv_iter_stats, attention_dp_rank,
                  host_step_time_ms, prev_device_step_time_ms, scheduler_mode,
-                 gpu_forward_time_ms))
+                 gpu_forward_time_ms, mm_encoder_stats))
 
     def _process_iter_stats(
         self,

tensorrt_llm/executor/base_worker.py

@@ -1301,6 +1301,7 @@ def _stats_serializer(stats) -> str:
         prev_device_step_time_ms = stats[5] if len(stats) > 5 else None
         scheduler_mode = stats[6] if len(stats) > 6 else None
         gpu_forward_time_ms = stats[7] if len(stats) > 7 else None
+        mm_encoder_stats = stats[8] if len(stats) > 8 else None
 
         stats_dict = json.loads(iteration_stats.to_json_str())
         # Always tag the row so Dynamo's adapter can read
@@ -1337,6 +1338,8 @@ def _stats_serializer(stats) -> str:
         # ForwardPassMetrics wall_time.
         if gpu_forward_time_ms is not None:
             stats_dict["gpuForwardTimeMS"] = gpu_forward_time_ms
+        if mm_encoder_stats is not None:
+            stats_dict["multimodalEncoderStats"] = mm_encoder_stats
         # Scheduler mode for this record. "overlap" means iterLatencyMS
         # spans ~2 loops (use hostStepTimeMS for clean per-loop cost);
         # "non_overlap" means iterLatencyMS is itself the clean per-loop