Add scheduler prefill queue wait breakdown

docs_new/docs/references/production_metrics.mdx

@@ -7,6 +7,8 @@ SGLang exposes the following metrics via Prometheus. You can enable it by adding
 
 An example of the monitoring dashboard is available in [examples/monitoring/grafana.json](https://github.com/sgl-project/sglang/blob/main/examples/monitoring/grafana/dashboards/json/sglang-dashboard.json).
 
+The `sglang:per_stage_req_latency_seconds` histogram includes scheduler-side prefill queue stages. `prefill_queue_idle` measures time from queue entry until the scheduler first attempts prefill admission. `prefill_budget_wait` measures time from that first attempt until the request enters forward. Use these stages to separate backlog from token budget or chunking pressure when TTFT rises.
+
 Here is an example of the metrics:
 
 ```text Output

docs_new/docs/references/production_request_trace.mdx

@@ -7,6 +7,8 @@ SGLang exports request trace data based on the OpenTelemetry Collector. You can
 
 You can find example screenshots of the visualization in https://github.com/sgl-project/sglang/issues/8965.
 
+Scheduler traces include `prefill_queue_idle` and `prefill_budget_wait` slices for prefill requests. The first slice shows time spent waiting before the scheduler first tries to admit the request. The second slice shows time spent after that first attempt before forward starts, which usually points to token budget, chunking, or retry pressure.
+
 ## Setup Guide
 This section explains how to configure the request tracing and export the trace data.
 1. Install the required packages and tools

python/sglang/srt/managers/schedule_policy.py

@@ -858,6 +858,8 @@ def add_req_state(r, insert_sort=False):
     def add_one_req(
         self, req: Req, has_chunked_req: bool, truncation_align_size: Optional[int]
     ):
+        req.time_stats.set_first_prefill_attempt_time()
+
         if (self.prefill_delayer_single_pass is not None) and (
             not self.prefill_delayer_single_pass.negotiate_should_allow_prefill(
                 local_prefillable=True,

python/sglang/srt/observability/req_time_stats.py

@@ -126,6 +126,16 @@ class RequestStage:
         # equal to "observe_queue_time"
         metrics_is_observed=False,
     )
+    PREFILL_QUEUE_IDLE = RequestStageConfig(
+        "prefill_queue_idle",
+        level=1,
+        metrics_is_observed=True,
+    )
+    PREFILL_BUDGET_WAIT = RequestStageConfig(
+        "prefill_budget_wait",
+        level=1,
+        metrics_is_observed=True,
+    )
     DECODE_FORWARD = RequestStageConfig(
         "decode_forward",
         level=1,
@@ -578,6 +588,7 @@ class SchedulerReqTimeStats(ReqTimeStatsBase):
 
     # common, get by time.perf_counter()
     wait_queue_entry_time: float = 0.0
+    first_prefill_attempt_time: float = 0.0
     forward_entry_time: float = 0.0
     prefill_finished_time: float = 0.0
     completion_time: float = 0.0
@@ -623,6 +634,7 @@ def __getstate__(self) -> object:
 
         state = {
             "wait_queue_entry_time": self.wait_queue_entry_time,
+            "first_prefill_attempt_time": self.first_prefill_attempt_time,
             "forward_entry_time": self.forward_entry_time,
             "prefill_finished_time": self.prefill_finished_time,
             "diff_realtime_monotonic": global_diff_realtime_monotonic,
@@ -693,12 +705,14 @@ def set_retract_time(self, ts=None):
         self.last_chunked_prefill_finish_time = 0.0
         self.last_decode_finish_time = 0.0
         self.last_decode_scheduled_time = 0.0
+        self.first_prefill_attempt_time = 0.0
 
         if self.trace_ctx.tracing_enable:
             self.trace_ctx.trace_event("retract", 1, convert_time_to_realtime_ns(ts))
 
     def reset_prefill_retry_time(self):
         self.wait_queue_entry_time = 0.0
+        self.first_prefill_attempt_time = 0.0
         self.forward_entry_time = 0.0
         self.prefill_finished_time = 0.0
         self.completion_time = 0.0
@@ -729,6 +743,10 @@ def set_wait_queue_entry_time(self, ts=None):
 
         self.wait_queue_entry_time = ts
 
+    def set_first_prefill_attempt_time(self, ts=None):
+        if self.first_prefill_attempt_time == 0.0:
+            self.first_prefill_attempt_time = ts or time.perf_counter()
+
     def set_forward_entry_time(self, ts=None):
         ts = ts or time.perf_counter()
         if self.forward_entry_time == 0.0:
@@ -747,6 +765,8 @@ def set_forward_entry_time(self, ts=None):
 
                 self.observe_per_stage_req_latency(stage, ts - slice_start_time)
                 self.trace_slice(stage, slice_start_time, ts)
+                if self.disagg_mode != DisaggregationMode.DECODE:
+                    self.observe_prefill_queue_breakdown(ts)
 
                 if self.disagg_mode == DisaggregationMode.DECODE:
                     self.trace_ctx.trace_slice_start(
@@ -1020,6 +1040,43 @@ def set_decode_prebuilt_finish_time(self, ts=None):
     def get_queueing_time(self) -> float:
         return self.forward_entry_time - self.wait_queue_entry_time
 
+    def get_idle_in_queue_time(self) -> float:
+        return self.duration_between(
+            self.wait_queue_entry_time, self.first_prefill_attempt_time
+        )
+
+    def get_prefill_budget_wait_time(self) -> float:
+        return self.duration_between(
+            self.first_prefill_attempt_time, self.forward_entry_time
+        )
+
+    def observe_prefill_queue_breakdown(self, forward_entry_time: float):
+        if self.first_prefill_attempt_time <= 0.0:
+            return
+
+        idle_in_queue_time = self.get_idle_in_queue_time()
+        prefill_budget_wait_time = self.duration_between(
+            self.first_prefill_attempt_time, forward_entry_time
+        )
+
+        self.observe_per_stage_req_latency(
+            RequestStage.PREFILL_QUEUE_IDLE, idle_in_queue_time
+        )
+        self.trace_slice(
+            RequestStage.PREFILL_QUEUE_IDLE,
+            self.wait_queue_entry_time,
+            self.first_prefill_attempt_time,
+        )
+
+        self.observe_per_stage_req_latency(
+            RequestStage.PREFILL_BUDGET_WAIT, prefill_budget_wait_time
+        )
+        self.trace_slice(
+            RequestStage.PREFILL_BUDGET_WAIT,
+            self.first_prefill_attempt_time,
+            forward_entry_time,
+        )
+
     def convert_to_duration(self) -> str:
         if self.disagg_mode == DisaggregationMode.NULL:
             queue_duration = self.duration_between(
@@ -1034,7 +1091,12 @@ def convert_to_duration(self) -> str:
                     queue_duration >= 0 and forward_duration >= 0
                 ), f"queue_duration={queue_duration} < 0 or forward_duration={forward_duration} < 0"
 
-            return f"queue_duration={self.format_duration(queue_duration)}, forward_duration={self.format_duration(forward_duration)}, entry_time={self.format_wallclock(self.wait_queue_entry_time)}"
+            return (
+                f"{self.format_prefill_queue_breakdown()}"
+                f"queue_duration={self.format_duration(queue_duration)}, "
+                f"forward_duration={self.format_duration(forward_duration)}, "
+                f"entry_time={self.format_wallclock(self.wait_queue_entry_time)}"
+            )
         elif self.disagg_mode == DisaggregationMode.PREFILL:
             bootstrap_queue_duration = self.duration_between(
                 self.prefill_bootstrap_queue_entry_time, self.wait_queue_entry_time
@@ -1075,6 +1137,7 @@ def convert_to_duration(self) -> str:
 
             return (
                 f"{bootstrap_fields}"
+                f"{self.format_prefill_queue_breakdown()}"
                 f"queue_duration={self.format_duration(queue_duration)}, "
                 f"forward_duration={self.format_duration(forward_duration)}, "
                 f"entry_time={self.format_wallclock(self.prefill_bootstrap_queue_entry_time)}, "
@@ -1153,8 +1216,25 @@ def convert_to_output_meta_info(self):
                 "queue_time": self.get_queueing_time(),
             }
         )
+        if self.first_prefill_attempt_time > 0.0:
+            meta_data.update(
+                {
+                    "idle_in_queue_time": self.get_idle_in_queue_time(),
+                    "prefill_budget_wait_time": self.get_prefill_budget_wait_time(),
+                }
+            )
         return meta_data
 
+    def format_prefill_queue_breakdown(self) -> str:
+        if self.first_prefill_attempt_time <= 0.0:
+            return ""
+        return (
+            f"idle_in_queue_duration="
+            f"{self.format_duration(self.get_idle_in_queue_time())}, "
+            f"prefill_budget_wait_duration="
+            f"{self.format_duration(self.get_prefill_budget_wait_time())}, "
+        )
+
     def format_duration(self, duration: float) -> str:
         return f"{duration * 1e3:.2f}ms"
 

test/registered/unit/managers/test_prefill_adder.py

@@ -80,7 +80,10 @@ def create_mock_req(self, rid, priority, max_new_tokens, output_len=0, wait_time
         req.extend_logprob_start_len = 0
         req.output_ids = [0] * output_len
         req.sampling_params = SimpleNamespace(max_new_tokens=max_new_tokens)
-        req.time_stats = SimpleNamespace(wait_queue_entry_time=wait_time)
+        req.time_stats = SimpleNamespace(
+            wait_queue_entry_time=wait_time,
+            set_first_prefill_attempt_time=MagicMock(),
+        )
         req.retracted_stain = False
         req.finished.return_value = False
         req.needs_host_load_back.return_value = False

test/registered/unit/observability/test_req_time_stats.py

@@ -0,0 +1,65 @@
+import unittest
+from unittest.mock import MagicMock
+
+from sglang.srt.observability.req_time_stats import (
+    RequestStage,
+    SchedulerReqTimeStats,
+)
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=1, suite="base-b-test-cpu")
+
+
+class TestSchedulerReqTimeStats(unittest.TestCase):
+    def test_prefill_queue_breakdown_from_first_attempt(self):
+        collector = MagicMock()
+        stats = SchedulerReqTimeStats()
+        stats.set_metrics_collector(collector)
+
+        stats.set_wait_queue_entry_time(ts=10.0)
+        stats.set_first_prefill_attempt_time(ts=13.0)
+        stats.set_forward_entry_time(ts=21.0)
+
+        self.assertEqual(stats.get_queueing_time(), 11.0)
+        self.assertEqual(stats.get_idle_in_queue_time(), 3.0)
+        self.assertEqual(stats.get_prefill_budget_wait_time(), 8.0)
+
+        collector.observe_queue_time.assert_called_once_with(11.0)
+        collector.observe_per_stage_req_latency.assert_any_call(
+            RequestStage.PREFILL_QUEUE_IDLE.stage_name, 3.0
+        )
+        collector.observe_per_stage_req_latency.assert_any_call(
+            RequestStage.PREFILL_BUDGET_WAIT.stage_name, 8.0
+        )
+
+        meta = stats.convert_to_output_meta_info()
+        self.assertEqual(meta["queue_time"], 11.0)
+        self.assertEqual(meta["idle_in_queue_time"], 3.0)
+        self.assertEqual(meta["prefill_budget_wait_time"], 8.0)
+
+        duration = stats.convert_to_duration()
+        self.assertIn("idle_in_queue_duration=3000.00ms", duration)
+        self.assertIn("prefill_budget_wait_duration=8000.00ms", duration)
+
+    def test_first_prefill_attempt_is_set_once_and_serialized(self):
+        stats = SchedulerReqTimeStats()
+
+        stats.set_first_prefill_attempt_time(ts=13.0)
+        stats.set_first_prefill_attempt_time(ts=14.0)
+
+        self.assertEqual(stats.first_prefill_attempt_time, 13.0)
+        self.assertEqual(stats.__getstate__()["first_prefill_attempt_time"], 13.0)
+
+    def test_retry_reset_clears_first_attempt(self):
+        stats = SchedulerReqTimeStats()
+
+        stats.set_wait_queue_entry_time(ts=10.0)
+        stats.set_first_prefill_attempt_time(ts=13.0)
+        stats.reset_prefill_retry_time()
+
+        self.assertEqual(stats.wait_queue_entry_time, 0.0)
+        self.assertEqual(stats.first_prefill_attempt_time, 0.0)
+
+
+if __name__ == "__main__":
+    unittest.main()