Add corrupted request logic with IterationStats and removal from SchedulerStats

Signed-off-by: atalhens <[email protected]>

Author: atalhens

examples/online_serving/prometheus_grafana/grafana.json

@@ -593,22 +593,6 @@
           "range": true,
           "refId": "C",
           "useBackend": false
-        },
-        {
-          "datasource": {
-            "type": "prometheus",
-            "uid": "${DS_PROMETHEUS}"
-          },
-          "disableTextWrap": false,
-          "editorMode": "builder",
-          "expr": "vllm:num_requests_corrupted{model_name=\"$model_name\"}",
-          "fullMetaSearch": false,
-          "includeNullMetadata": true,
-          "instant": false,
-          "legendFormat": "Num Corrupted",
-          "range": true,
-          "refId": "D",
-          "useBackend": false
         }
       ],
       "title": "Scheduler State",

vllm/config/scheduler.py

@@ -137,12 +137,6 @@ class SchedulerConfig:
     structured outputs, speculative decoding, and pipeline parallelism.
     """
 
-    include_corrupted_requests: bool = False
-    """If set to True, include corrupted requests in scheduler statistics.
-    This adds computational overhead but provides more detailed metrics for
-    monitoring and debugging purposes.
-    """
-
     def compute_hash(self) -> str:
         """
         WARNING: Whenever a new field is added to this config,

vllm/engine/arg_utils.py

@@ -550,7 +550,6 @@ class EngineArgs:
     """Custom logitproc types"""
 
     async_scheduling: bool = SchedulerConfig.async_scheduling
-    include_corrupted_requests: bool = SchedulerConfig.include_corrupted_requests
 
     kv_sharing_fast_prefill: bool = CacheConfig.kv_sharing_fast_prefill
 
@@ -1042,10 +1041,6 @@ def add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser:
         scheduler_group.add_argument(
             "--async-scheduling", **scheduler_kwargs["async_scheduling"]
         )
-        scheduler_group.add_argument(
-            "--include-corrupted-requests",
-            **scheduler_kwargs["include_corrupted_requests"],
-        )
 
         # Compilation arguments
         compilation_kwargs = get_kwargs(CompilationConfig)
@@ -1591,7 +1586,6 @@ def create_engine_config(
             long_prefill_token_threshold=self.long_prefill_token_threshold,
             disable_hybrid_kv_cache_manager=self.disable_hybrid_kv_cache_manager,
             async_scheduling=self.async_scheduling,
-            include_corrupted_requests=self.include_corrupted_requests,
         )
 
         if not model_config.is_multimodal_model and self.default_mm_loras:

vllm/v1/core/sched/interface.py

@@ -136,8 +136,8 @@ def reset_prefix_cache(self) -> bool:
         raise NotImplementedError
 
     @abstractmethod
-    def get_request_counts(self) -> tuple[int, int, int]:
-        """Returns (num_running_reqs, num_waiting_reqs, num_corrupted_reqs)."""
+    def get_request_counts(self) -> tuple[int, int]:
+        """Returns (num_running_reqs, num_waiting_reqs)."""
         raise NotImplementedError
 
     @abstractmethod

vllm/v1/core/sched/scheduler.py

@@ -124,7 +124,6 @@ def __init__(
         # Priority queues for requests.
         self.waiting = create_request_queue(self.policy)
         self.running: list[Request] = []
-        self.corrupted: list[Request] = []
 
         # The request IDs that are finished in between the previous and the
         # current steps. This is used to notify the workers about the finished
@@ -1043,6 +1042,7 @@ def update_from_output(
                         kv_transfer_params=kv_transfer_params,
                         trace_headers=request.trace_headers,
                         num_cached_tokens=request.num_cached_tokens,
+                        num_nans_in_logits=request.num_nans_in_logits,
                     )
                 )
             else:
@@ -1162,22 +1162,9 @@ def update_draft_token_ids(
             else:
                 request.spec_token_ids = spec_token_ids
 
-    def _get_corrupted_requests_count(self, include_corrupted: bool) -> int:
-        """Get the count of corrupted requests if enabled, otherwise return 0.
-
-        This method centralizes the corrupted requests counting logic to avoid
-        code duplication and improve performance.
-        """
-        if not include_corrupted:
-            return 0
-        return sum(req.is_output_corrupted for req in self.running)
-
-    def get_request_counts(self) -> tuple[int, int, int]:
-        """Returns (num_running_reqs, num_waiting_reqs, num_corrupted_reqs)."""
-        num_corrupted_reqs = self._get_corrupted_requests_count(
-            self.scheduler_config.include_corrupted_requests
-        )
-        return len(self.running), len(self.waiting), num_corrupted_reqs
+    def get_request_counts(self) -> tuple[int, int]:
+        """Returns (num_running_reqs, num_waiting_reqs)."""
+        return len(self.running), len(self.waiting)
 
     def add_request(self, request: Request) -> None:
         self.waiting.add_request(request)
@@ -1271,7 +1258,6 @@ def make_stats(
         return SchedulerStats(
             num_running_reqs=len(self.running),
             num_waiting_reqs=len(self.waiting),
-            num_corrupted_reqs=num_corrupted_reqs,
             kv_cache_usage=self.kv_cache_manager.usage,
             prefix_cache_stats=prefix_cache_stats,
             connector_prefix_cache_stats=connector_prefix_cache_stats,

vllm/v1/engine/__init__.py

@@ -121,6 +121,8 @@ class EngineCoreOutput(
     trace_headers: Mapping[str, str] | None = None
     # The number of tokens with prefix cache hits.
     num_cached_tokens: int = 0
+    # The number of NaNs in logits for this request.
+    num_nans_in_logits: int = 0
 
     @property
     def finished(self) -> bool:

vllm/v1/engine/core.py

@@ -98,9 +98,6 @@ def __init__(
             )
 
         self.log_stats = log_stats
-        self.include_corrupted_requests = (
-            vllm_config.scheduler_config.include_corrupted_requests
-        )
 
         # Setup Model.
         self.model_executor = executor_class(vllm_config)
@@ -1057,7 +1054,7 @@ def __init__(
         # finished with DP peers every N steps.
         self.step_counter = 0
         self.current_wave = 0
-        self.last_counts = (0, 0, 0)
+        self.last_counts = (0, 0)
 
         # Initialize the engine.
         dp_rank = vllm_config.parallel_config.data_parallel_rank

vllm/v1/metrics/loggers.py

@@ -9,6 +9,7 @@
 
 from prometheus_client import Counter, Gauge, Histogram
 
+import vllm.envs as envs
 from vllm.config import SupportsMetricsInfo, VllmConfig
 from vllm.distributed.kv_transfer.kv_connector.v1.metrics import KVConnectorLogging
 from vllm.logger import init_logger
@@ -104,6 +105,9 @@ def __init__(self, vllm_config: VllmConfig, engine_index: int = 0):
         self.engine_is_idle = False
         self.aggregated = False
 
+        # Track the number of corrupted requests, never reset.
+        self.num_corrupted_reqs: int = 0
+
     def _reset(self, now):
         self.last_log_time = now
 
@@ -115,6 +119,7 @@ def _track_iteration_stats(self, iteration_stats: IterationStats):
         # Save tracked stats for token counters.
         self.num_prompt_tokens += iteration_stats.num_prompt_tokens
         self.num_generation_tokens += iteration_stats.num_generation_tokens
+        self.num_corrupted_reqs += iteration_stats.num_corrupted_reqs
 
     def _get_throughput(self, tracked_stats: int, now: float) -> float:
         # Compute summary metrics for tracked stats
@@ -187,7 +192,6 @@ def log(self):
             "Avg generation throughput: %.1f tokens/s",
             "Running: %d reqs",
             "Waiting: %d reqs",
-            "Corrupted: %d reqs",
             "GPU KV cache usage: %.1f%%",
             "Prefix cache hit rate: %.1f%%",
         ]
@@ -196,13 +200,15 @@ def log(self):
             self.last_generation_throughput,
             self.last_scheduler_stats.num_running_reqs,
             self.last_scheduler_stats.num_waiting_reqs,
-            self.last_scheduler_stats.num_corrupted_reqs,
             self.last_scheduler_stats.kv_cache_usage * 100,
             self.prefix_caching_metrics.hit_rate * 100,
         ]
         if not self.connector_prefix_caching_metrics.empty:
             log_parts.append("External prefix cache hit rate: %.1f%%")
             log_args.append(self.connector_prefix_caching_metrics.hit_rate * 100)
+        if envs.VLLM_COMPUTE_NANS_IN_LOGITS:
+            log_parts.append("Corrupted: %d reqs")
+            log_args.append(self.num_corrupted_reqs)
         if not self.mm_caching_metrics.empty:
             log_parts.append("MM cache hit rate: %.1f%%")
             log_args.append(self.mm_caching_metrics.hit_rate * 100)
@@ -271,9 +277,6 @@ def aggregate_scheduler_stats(self):
             self.last_scheduler_stats.num_running_reqs += (
                 last_scheduler_stats.num_running_reqs
             )
-            self.last_scheduler_stats.num_corrupted_reqs += (
-                last_scheduler_stats.num_corrupted_reqs
-            )
             self.last_scheduler_stats.kv_cache_usage += (
                 last_scheduler_stats.kv_cache_usage
             )
@@ -387,16 +390,6 @@ def __init__(
             gauge_scheduler_waiting, engine_indexes, model_name
         )
 
-        gauge_scheduler_corrupted = self._gauge_cls(
-            name="vllm:num_requests_corrupted",
-            documentation="Number of requests corrupted.",
-            multiprocess_mode="mostrecent",
-            labelnames=labelnames,
-        )
-        self.gauge_scheduler_corrupted = make_per_engine(
-            gauge_scheduler_corrupted, engine_indexes, model_name
-        )
-
         #
         # GPU cache
         #
@@ -458,6 +451,16 @@ def __init__(
             gauge_kv_cache_usage, engine_indexes, model_name
         )
 
+        if envs.VLLM_COMPUTE_NANS_IN_LOGITS:
+            counter_corrupted_requests = self._counter_cls(
+                name="vllm:corrupted_requests",
+                documentation="Number of requests corrupted out of running requests.",
+                labelnames=labelnames,
+            )
+            self.counter_corrupted_requests = make_per_engine(
+                counter_corrupted_requests, engine_indexes, model_name
+            )
+
         counter_prefix_cache_queries = self._counter_cls(
             name="vllm:prefix_cache_queries",
             documentation=(
@@ -910,10 +913,6 @@ def record(
             self.gauge_scheduler_waiting[engine_idx].set(
                 scheduler_stats.num_waiting_reqs
             )
-            self.gauge_scheduler_corrupted[engine_idx].set(
-                scheduler_stats.num_corrupted_reqs
-            )
-
             if self.show_hidden_metrics:
                 self.gauge_gpu_cache_usage[engine_idx].set(
                     scheduler_stats.kv_cache_usage
@@ -958,6 +957,10 @@ def record(
         self.counter_num_preempted_reqs[engine_idx].inc(
             iteration_stats.num_preempted_reqs
         )
+        if envs.VLLM_COMPUTE_NANS_IN_LOGITS and iteration_stats.num_corrupted_reqs > 0:
+            self.counter_corrupted_requests[engine_idx].inc(
+                iteration_stats.num_corrupted_reqs
+            )
         self.counter_prompt_tokens[engine_idx].inc(iteration_stats.num_prompt_tokens)
         self.counter_generation_tokens[engine_idx].inc(
             iteration_stats.num_generation_tokens

vllm/v1/metrics/stats.py

@@ -156,7 +156,6 @@ class SchedulerStats:
 
     num_running_reqs: int = 0
     num_waiting_reqs: int = 0
-    num_corrupted_reqs: int = 0
 
     # These are used for internal DP load-balancing.
     step_counter: int = 0
@@ -195,6 +194,9 @@ class RequestStateStats:
     # first token latency
     first_token_latency: float = 0.0
 
+    # Track if output has NaNs in logits
+    num_nans_in_logits: int = 0
+
 
 @dataclass
 class FinishedRequestStats:
@@ -210,6 +212,7 @@ class FinishedRequestStats:
     inference_time: float = 0.0
     decode_time: float = 0.0
     mean_time_per_output_token: float = 0.0
+    is_corrupted: bool = False
 
 
 class IterationStats:
@@ -220,6 +223,7 @@ def __init__(self):
         self.num_generation_tokens = 0
         self.num_prompt_tokens = 0
         self.num_preempted_reqs = 0
+        self.num_corrupted_reqs = 0
         self.finished_requests: list[FinishedRequestStats] = []
         self.max_num_generation_tokens_iter: list[int] = []
         self.n_params_iter: list[int] = []
@@ -257,6 +261,10 @@ def update_from_output(
 
         req_stats.num_generation_tokens += num_new_generation_tokens
 
+        # Track NaNs in logits if present
+        if output.num_nans_in_logits > 0:
+            req_stats.num_nans_in_logits += output.num_nans_in_logits
+
         # Process request-level engine core events
         if output.events is not None:
             self.update_from_events(
@@ -327,6 +335,10 @@ def update_from_finished_request(
             else 0
         )
 
+        # Check if output was corrupted based on NaN count
+        # (will be 0 if VLLM_COMPUTE_NANS_IN_LOGITS was not enabled)
+        is_corrupted = req_stats.num_nans_in_logits > 0
+
         finished_req = FinishedRequestStats(
             finish_reason=finish_reason,
             e2e_latency=e2e_latency,
@@ -338,7 +350,10 @@ def update_from_finished_request(
             inference_time=inference_time,
             decode_time=decode_time,
             mean_time_per_output_token=mean_time_per_output_token,
+            is_corrupted=is_corrupted,
         )
+        if is_corrupted:
+            self.num_corrupted_reqs += 1
         self.finished_requests.append(finished_req)