Merge branch 'main' into zihaok_fix_misc_pyxisruntime_issue

Author: arekay-nv

docs/metrics/report_design.md

@@ -100,6 +100,10 @@ text prompt with the configured reference tokenizer. This records the workload's
 distribution as early as possible. `input_sequence_lengths` is empty when no ISL values were
 recorded (for example, no tokenizer is available for text-only inputs).
 
+Set `settings.metrics_isl: false` in YAML, or pass `--no-metrics-isl`, to skip ISL collection.
+This avoids all ISL tokenizer work and leaves `input_sequence_lengths` empty. OSL, TPOT,
+full-run OSL, and output-derived TPS reporting are unchanged.
+
 **QPS/TPS window selection.** The snapshot always carries both a native window
 (`tracked_duration_ns`) and the MLPerf LoadGen "completed" window
 (`legacy_loadgen_window_duration_ns`, poisson only), so it stays reinterpretable either way. The

src/inference_endpoint/async_utils/services/metrics_aggregator/__main__.py

@@ -188,6 +188,12 @@ async def main() -> None:
             "sharded pool — one worker process per 8-core block."
         ),
     )
+    parser.add_argument(
+        "--metrics-isl",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help="Collect input sequence length (ISL) metrics (default: enabled).",
+    )
     parser.add_argument(
         "--streaming",
         action="store_true",
@@ -280,6 +286,7 @@ async def main() -> None:
                     args.publish_interval if args.tokenizer_workers > 0 else None
                 ),
                 streaming=args.streaming,
+                enable_isl=args.metrics_isl,
                 shutdown_event=shutdown_event,
                 drain_timeout_s=args.drain_timeout,
             )

src/inference_endpoint/async_utils/services/metrics_aggregator/aggregator.py

@@ -135,6 +135,7 @@ def __init__(
         streaming: bool = False,
         shutdown_event: asyncio.Event | None = None,
         drain_timeout_s: float | None = None,
+        enable_isl: bool = True,
         **kwargs,
     ):
         # drain_timeout_s is injected (not derived) because the right
@@ -179,17 +180,21 @@ def __init__(
 
         # Pre-register all metrics on the registry. Tests can introspect via
         # registry.has_counter / has_series.
-        self._register_metrics(streaming, sig_figs, n_histogram_buckets)
+        self._register_metrics(streaming, enable_isl, sig_figs, n_histogram_buckets)
 
         self._table = MetricsTable(self._registry)
-        self._register_triggers(streaming)
+        self._register_triggers(streaming, enable_isl)
 
     # ------------------------------------------------------------------
     # Registration helpers
     # ------------------------------------------------------------------
 
     def _register_metrics(
-        self, streaming: bool, sig_figs: int, n_histogram_buckets: int
+        self,
+        streaming: bool,
+        enable_isl: bool,
+        sig_figs: int,
+        n_histogram_buckets: int,
     ) -> None:
         """Register all counters and series on the registry."""
         for key in MetricCounterKey:
@@ -204,13 +209,14 @@ def _register_metrics(
             sig_figs=sig_figs,
             n_histogram_buckets=n_histogram_buckets,
         )
-        self._registry.register_series(
-            MetricSeriesKey.ISL.value,
-            hdr_low=TOKEN_HDR_LOW,
-            hdr_high=TOKEN_HDR_HIGH,
-            sig_figs=sig_figs,
-            n_histogram_buckets=n_histogram_buckets,
-        )
+        if enable_isl:
+            self._registry.register_series(
+                MetricSeriesKey.ISL.value,
+                hdr_low=TOKEN_HDR_LOW,
+                hdr_high=TOKEN_HDR_HIGH,
+                sig_figs=sig_figs,
+                n_histogram_buckets=n_histogram_buckets,
+            )
         self._registry.register_series(
             MetricSeriesKey.OSL.value,
             hdr_low=TOKEN_HDR_LOW,
@@ -246,18 +252,19 @@ def _register_metrics(
                 dtype=float,
             )
 
-    def _register_triggers(self, streaming: bool) -> None:
+    def _register_triggers(self, streaming: bool, enable_isl: bool) -> None:
         """Register metric triggers on the table.
 
         Streaming-only triggers (TTFT, chunk_delta, TPOT) are only registered
-        when ``streaming=True``.
+        when ``streaming=True``; the ISL trigger only when ``enable_isl=True``.
         """
         table = self._table
         registry = self._registry
         queue = self._token_queue
 
-        # Always registered
-        table.add_trigger(SampleField.ISSUED_NS, IslTrigger(registry, queue))
+        # Latency and OSL are always registered; ISL is opt-out.
+        if enable_isl:
+            table.add_trigger(SampleField.ISSUED_NS, IslTrigger(registry, queue))
         table.add_trigger(SampleField.COMPLETE_NS, SampleLatencyTrigger(registry))
         table.add_trigger(SampleField.COMPLETE_NS, OslTrigger(registry, queue))
 

src/inference_endpoint/commands/benchmark/pipeline.py

@@ -102,6 +102,7 @@ def _build_aggregator_args(
     tokenizer_name: str | None,
     drain_timeout_s: float | None,
     tokenizer_workers: int,
+    enable_isl: bool,
     early_stopping: bool,
 ) -> list[str]:
     """CLI args for the metrics_aggregator subprocess."""
@@ -124,6 +125,7 @@ def _build_aggregator_args(
     if drain_timeout_s is not None:
         args.extend(["--drain-timeout", str(drain_timeout_s)])
     args.extend(["--tokenizer-workers", str(tokenizer_workers)])
+    args.append("--metrics-isl" if enable_isl else "--no-metrics-isl")
     return args
 
 
@@ -294,6 +296,7 @@ async def start(self) -> None:
                 tokenizer_name=self._tokenizer_name,
                 drain_timeout_s=timeouts.metrics_drain_timeout_s,
                 tokenizer_workers=self._config.settings.metrics_tokenizer_workers,
+                enable_isl=self._config.settings.metrics_isl,
                 early_stopping=self._config.settings.early_stopping.enabled,
             )
             event_logger_args = _build_event_logger_args(

src/inference_endpoint/config/schema.py

@@ -973,6 +973,16 @@ class Settings(WithUpdatesMixin, BaseModel):
             "(default: 4; 0 = defer everything to the end-of-run drain)."
         ),
     )
+    metrics_isl: Annotated[
+        bool,
+        cyclopts.Parameter(
+            alias="--metrics-isl",
+            help="Collect input sequence length (ISL) metrics",
+        ),
+    ] = Field(
+        True,
+        description="Collect input sequence length (ISL) metrics (default: true).",
+    )
 
     @model_validator(mode="after")
     def _min_issue_duration_requires_qps(self) -> Self:

src/inference_endpoint/config/templates/concurrency_template_full.yaml

@@ -105,6 +105,7 @@ settings:
   early_stopping:  # MLPerf early-stopping percentile estimates (on by default; enabled: false opts out)
     enabled: true  # Early-stopping percentile estimates (default on)
   metrics_tokenizer_workers: 4  # In-process tokenizer threads for live (mid-run) ISL/OSL/TPOT (default: 4; 0 = defer everything to the end-of-run drain).
+  metrics_isl: true  # Collect input sequence length (ISL) metrics (default: true).
 endpoint_config:
   endpoints:  # Endpoint URL(s). Must include scheme, e.g. 'http://host:port'.
   - http://localhost:8000

src/inference_endpoint/config/templates/offline_template_full.yaml

@@ -105,6 +105,7 @@ settings:
   early_stopping:  # MLPerf early-stopping percentile estimates (on by default; enabled: false opts out)
     enabled: true  # Early-stopping percentile estimates (default on)
   metrics_tokenizer_workers: 4  # In-process tokenizer threads for live (mid-run) ISL/OSL/TPOT (default: 4; 0 = defer everything to the end-of-run drain).
+  metrics_isl: true  # Collect input sequence length (ISL) metrics (default: true).
 endpoint_config:
   endpoints:  # Endpoint URL(s). Must include scheme, e.g. 'http://host:port'.
   - http://localhost:8000

src/inference_endpoint/config/templates/online_template_full.yaml

@@ -106,6 +106,7 @@ settings:
   early_stopping:  # MLPerf early-stopping percentile estimates (on by default; enabled: false opts out)
     enabled: true  # Early-stopping percentile estimates (default on)
   metrics_tokenizer_workers: 4  # In-process tokenizer threads for live (mid-run) ISL/OSL/TPOT (default: 4; 0 = defer everything to the end-of-run drain).
+  metrics_isl: true  # Collect input sequence length (ISL) metrics (default: true).
 endpoint_config:
   endpoints:  # Endpoint URL(s). Must include scheme, e.g. 'http://host:port'.
   - http://localhost:8000

tests/unit/async_utils/services/metrics_aggregator/conftest.py

@@ -185,6 +185,7 @@ def make_aggregator(
     tokenizer=None,
     live_flush_interval_s: float | None = None,
     streaming: bool = True,
+    enable_isl: bool = True,
     shutdown_event: asyncio.Event | None = None,
     drain_timeout_s: float | None = None,
 ) -> tuple[MetricsAggregatorService, MetricsRegistry, MagicMock]:
@@ -218,6 +219,7 @@ def make_aggregator(
         tokenizer=tokenizer,
         live_flush_interval_s=live_flush_interval_s,
         streaming=streaming,
+        enable_isl=enable_isl,
         shutdown_event=shutdown_event,
         drain_timeout_s=drain_timeout_s,
     )

tests/unit/async_utils/services/metrics_aggregator/test_aggregator.py

@@ -428,6 +428,58 @@ async def test_issued_without_data_no_isl(self, tmp_path):
             finally:
                 agg.close()
 
+    @pytest.mark.asyncio
+    async def test_disabled_isl_skips_prompt_tokenization_keeps_osl(self, tmp_path):
+        """enable_isl=False: no ISL series, no prompt enqueued; OSL/TPOT intact."""
+        loop = asyncio.get_event_loop()
+        messages = (
+            {"role": "system", "content": "follow instructions"},
+            {"role": "user", "content": "summarize this"},
+        )
+        with ManagedZMQContext.scoped(socket_dir=str(tmp_path)) as ctx:
+            agg, registry, _ = make_aggregator(
+                ctx,
+                loop,
+                "agg_isl_disabled",
+                tokenizer=MockBatchTokenizer(),
+                enable_isl=False,
+            )
+            try:
+                await agg.process(
+                    [
+                        session_event(
+                            SessionEventType.START_PERFORMANCE_TRACKING, ts=0
+                        ),
+                        sample_event(
+                            SampleEventType.ISSUED,
+                            "s1",
+                            ts=1000,
+                            data=PromptData(messages=messages),
+                        ),
+                    ]
+                )
+                assert not registry.has_series(MetricSeriesKey.ISL.value)
+                assert agg.pending_tokens == 0
+
+                await agg.process(
+                    [
+                        sample_event(SampleEventType.RECV_FIRST, "s1", ts=2000),
+                        sample_event(
+                            SampleEventType.COMPLETE,
+                            "s1",
+                            ts=5000,
+                            data=streaming_text("hello", " world", " foo"),
+                        ),
+                        session_event(SessionEventType.ENDED, ts=5000),
+                    ]
+                )
+                assert snapshot_series_total(registry, MetricSeriesKey.OSL.value) == 3
+                assert snapshot_series_total(
+                    registry, MetricSeriesKey.TPOT_NS.value
+                ) == pytest.approx(1500.0)
+            finally:
+                agg.close()
+
 
 # ---------------------------------------------------------------------------
 # Edge cases and event ordering