From d02c515ee8e25b3126a4ae9804f880f839c7fb2c Mon Sep 17 00:00:00 2001 From: Liangsheng Yin Date: Sat, 7 Mar 2026 22:09:10 -0800 Subject: [PATCH] Decouple scheduler log printing from metrics collection (#20107) --- python/sglang/srt/disaggregation/prefill.py | 13 ++-- python/sglang/srt/dllm/mixin/scheduler.py | 13 ++-- python/sglang/srt/managers/scheduler.py | 4 -- .../scheduler_output_processor_mixin.py | 28 +++----- .../scheduler_runtime_checker_mixin.py | 3 +- .../observability/scheduler_metrics_mixin.py | 71 +++++++++++++------ python/sglang/srt/server_args.py | 2 +- .../srt/utils/scheduler_status_logger.py | 8 ++- .../utils/test_scheduler_status_logger.py | 2 +- 9 files changed, 81 insertions(+), 63 deletions(-) diff --git a/python/sglang/srt/disaggregation/prefill.py b/python/sglang/srt/disaggregation/prefill.py index fa9964026..381e92590 100644 --- a/python/sglang/srt/disaggregation/prefill.py +++ b/python/sglang/srt/disaggregation/prefill.py @@ -535,13 +535,12 @@ class SchedulerDisaggregationPrefillMixin: self.send_kv_chunk(req, last_chunk=False, end_idx=req.tmp_end_idx) req.time_stats.set_last_chunked_prefill_finish_time() - if self.current_scheduler_metrics_enabled: - can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False) - self.log_prefill_stats( - prefill_stats=batch.prefill_stats, - can_run_cuda_graph=can_run_cuda_graph, - dp_cooperation_info=batch.dp_cooperation_info, - ) + can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False) + self.report_prefill_stats( + prefill_stats=batch.prefill_stats, + can_run_cuda_graph=can_run_cuda_graph, + dp_cooperation_info=batch.dp_cooperation_info, + ) def process_disagg_prefill_inflight_queue( self: Scheduler, rids_to_check: Optional[List[str]] = None diff --git a/python/sglang/srt/dllm/mixin/scheduler.py b/python/sglang/srt/dllm/mixin/scheduler.py index 2b852507c..3524d40aa 100644 --- a/python/sglang/srt/dllm/mixin/scheduler.py +++ b/python/sglang/srt/dllm/mixin/scheduler.py @@ -92,13 +92,12 @@ class SchedulerDllmMixin: self.stream_output(batch.reqs, batch.return_logprob) self.token_to_kv_pool_allocator.free_group_end() - if self.current_scheduler_metrics_enabled: - can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False) - self.log_prefill_stats( - prefill_stats=batch.prefill_stats, - can_run_cuda_graph=can_run_cuda_graph, - dp_cooperation_info=batch.dp_cooperation_info, - ) + can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False) + self.report_prefill_stats( + prefill_stats=batch.prefill_stats, + can_run_cuda_graph=can_run_cuda_graph, + dp_cooperation_info=batch.dp_cooperation_info, + ) def _fetch_waiting_reqs(self: Scheduler): # Calculate how many requests can be added to DLLM manager diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index f7641cc34..ec3a254ad 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -310,10 +310,6 @@ class Scheduler( self.enable_overlap = not server_args.disable_overlap_schedule self.enable_pdmux = server_args.enable_pdmux self.skip_tokenizer_init = server_args.skip_tokenizer_init - self.enable_metrics = server_args.enable_metrics - self.enable_metrics_for_all_schedulers = ( - server_args.enable_metrics_for_all_schedulers - ) self.stream_interval = server_args.stream_interval self.spec_algorithm = SpeculativeAlgorithm.from_string( server_args.speculative_algorithm diff --git a/python/sglang/srt/managers/scheduler_output_processor_mixin.py b/python/sglang/srt/managers/scheduler_output_processor_mixin.py index f3b2d01ee..c56a92217 100644 --- a/python/sglang/srt/managers/scheduler_output_processor_mixin.py +++ b/python/sglang/srt/managers/scheduler_output_processor_mixin.py @@ -306,17 +306,12 @@ class SchedulerOutputProcessorMixin: self.stream_output(batch.reqs, batch.return_logprob, skip_stream_req) - if self.current_scheduler_metrics_enabled: - can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False) - if self.enable_metrics: - self.metrics_collector.increment_prefill_cuda_graph_pass( - value=can_run_cuda_graph - ) - self.log_prefill_stats( - prefill_stats=batch.prefill_stats, - can_run_cuda_graph=can_run_cuda_graph, - dp_cooperation_info=batch.dp_cooperation_info, - ) + can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False) + self.report_prefill_stats( + prefill_stats=batch.prefill_stats, + can_run_cuda_graph=can_run_cuda_graph, + dp_cooperation_info=batch.dp_cooperation_info, + ) def _resolve_spec_overlap_token_ids( self: Scheduler, result: GenerationBatchResult, batch: ScheduleBatch @@ -485,12 +480,11 @@ class SchedulerOutputProcessorMixin: self.token_to_kv_pool_allocator.free_group_end() self.forward_ct_decode = (self.forward_ct_decode + 1) % (1 << 30) - if self.current_scheduler_metrics_enabled: - if self.forward_ct_decode % self.server_args.decode_log_interval == 0: - self.log_decode_stats(can_run_cuda_graph, running_batch=batch) - self.log_decode_stats_every_iteration( - batch, num_accepted_tokens=result.num_accepted_tokens - ) + self.report_decode_stats( + can_run_cuda_graph, + running_batch=batch, + num_accepted_tokens=result.num_accepted_tokens, + ) def _mamba_prefix_cache_update( self, req: Req, batch: ScheduleBatch, result: GenerationBatchResult, i: int diff --git a/python/sglang/srt/managers/scheduler_runtime_checker_mixin.py b/python/sglang/srt/managers/scheduler_runtime_checker_mixin.py index 0441be4c5..78eaef316 100644 --- a/python/sglang/srt/managers/scheduler_runtime_checker_mixin.py +++ b/python/sglang/srt/managers/scheduler_runtime_checker_mixin.py @@ -271,8 +271,7 @@ class SchedulerRuntimeCheckerMixin: self._check_req_pool() if ( - self.enable_metrics - and self.current_scheduler_metrics_enabled + self.current_scheduler_metrics_enabled and time.perf_counter() > self.metrics_collector.last_log_time + 30 ): # During idle time, also collect metrics every 30 seconds. diff --git a/python/sglang/srt/observability/scheduler_metrics_mixin.py b/python/sglang/srt/observability/scheduler_metrics_mixin.py index 33ce9cc9a..60390fe47 100644 --- a/python/sglang/srt/observability/scheduler_metrics_mixin.py +++ b/python/sglang/srt/observability/scheduler_metrics_mixin.py @@ -114,8 +114,10 @@ class SchedulerMetricsMixin: self.stats = SchedulerStats() # Metrics - self.current_scheduler_metrics_enabled = ( - self.attn_tp_rank == 0 or self.enable_metrics_for_all_schedulers + self.enable_metrics = self.server_args.enable_metrics + self.is_stats_logging_rank = self.attn_tp_rank == 0 + self.current_scheduler_metrics_enabled = self.enable_metrics and ( + self.attn_tp_rank == 0 or self.server_args.enable_metrics_for_all_schedulers ) if self.enable_metrics: if self.server_args.disaggregation_mode == DisaggregationMode.PREFILL.value: @@ -158,7 +160,9 @@ class SchedulerMetricsMixin: if self.enable_kv_cache_events: self.init_kv_events(self.server_args.kv_events_config) - self.scheduler_status_logger = SchedulerStatusLogger.maybe_create() + self.scheduler_status_logger = SchedulerStatusLogger.maybe_create( + enable_metrics=self.enable_metrics + ) def init_kv_events(self: Scheduler, kv_events_config: Optional[str]): if self.enable_kv_cache_events: @@ -179,12 +183,18 @@ class SchedulerMetricsMixin: self.spec_total_num_accepted_tokens = 0 self.spec_total_num_forward_ct = 0 - def log_prefill_stats( + def report_prefill_stats( self: Scheduler, prefill_stats: PrefillStats, can_run_cuda_graph: bool, dp_cooperation_info: Optional[DPCooperationInfo] = None, ): + if ( + not self.is_stats_logging_rank + and not self.current_scheduler_metrics_enabled + ): + return + gap_latency = time.perf_counter() - self.last_prefill_stats_tic self.last_prefill_stats_tic = time.perf_counter() self.last_input_throughput = self.last_prefill_tokens / gap_latency @@ -267,9 +277,13 @@ class SchedulerMetricsMixin: msg += f"{graph_backend[self.device]}: {can_run_cuda_graph}" - logger.info(msg) + if self.is_stats_logging_rank: + logger.info(msg) - if self.enable_metrics: + if self.current_scheduler_metrics_enabled: + self.metrics_collector.increment_prefill_cuda_graph_pass( + value=can_run_cuda_graph + ) self.metrics_collector.increment_realtime_tokens( prefill_compute_tokens=prefill_stats.log_input_tokens, prefill_cache_tokens=prefill_stats.log_hit_tokens, @@ -332,11 +346,34 @@ class SchedulerMetricsMixin: self._emit_kv_metrics() self._publish_kv_events() - def log_decode_stats( - self: Scheduler, can_run_cuda_graph: bool, running_batch: ScheduleBatch = None + def report_decode_stats( + self: Scheduler, + can_run_cuda_graph: bool, + running_batch: ScheduleBatch = None, + num_accepted_tokens: int = 0, ): batch = running_batch or self.running_batch + # Every-iteration work: realtime token counting + status logger + if self.current_scheduler_metrics_enabled: + self.metrics_collector.increment_realtime_tokens( + # TODO unify this w/ the bumping logic in `Scheduler.num_generated_tokens` accumulator + decode_tokens=batch.batch_size() + num_accepted_tokens, + dp_cooperation_info=batch.dp_cooperation_info, + ) + + if x := self.scheduler_status_logger: + x.maybe_dump(batch, self.waiting_queue) + + # Periodic work: log + heavy metrics at decode_log_interval + if self.forward_ct_decode % self.server_args.decode_log_interval != 0: + return + if ( + not self.is_stats_logging_rank + and not self.current_scheduler_metrics_enabled + ): + return + gap_latency = time.perf_counter() - self.last_decode_stats_tic self.last_decode_stats_tic = time.perf_counter() self.last_gen_throughput = self.num_generated_tokens / gap_latency @@ -455,8 +492,9 @@ class SchedulerMetricsMixin: f"#queue-req: {len(self.waiting_queue)}" ) - logger.info(msg) - if self.enable_metrics: + if self.is_stats_logging_rank: + logger.info(msg) + if self.current_scheduler_metrics_enabled: priority_enabled = self.enable_priority_scheduling # Basics self.stats.num_running_reqs = QueueCount.from_reqs( @@ -524,19 +562,6 @@ class SchedulerMetricsMixin: self._emit_kv_metrics() self._publish_kv_events() - def log_decode_stats_every_iteration( - self: Scheduler, batch: ScheduleBatch, num_accepted_tokens: int - ): - if self.enable_metrics: - self.metrics_collector.increment_realtime_tokens( - # TODO unify this w/ the bumping logic in `Scheduler.num_generated_tokens` accumulator - decode_tokens=batch.batch_size() + num_accepted_tokens, - dp_cooperation_info=batch.dp_cooperation_info, - ) - - if x := self.scheduler_status_logger: - x.maybe_dump(batch, self.waiting_queue) - def log_batch_result_stats( self: Scheduler, batch: ScheduleBatch, diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index a0ce0a435..f941545ba 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -3905,7 +3905,7 @@ class ServerArgs: "--decode-log-interval", type=int, default=ServerArgs.decode_log_interval, - help="The log interval of decode batch.", + help="The log and metrics reporting interval (in decode iterations) for decode batches.", ) parser.add_argument( "--enable-request-time-stats-logging", diff --git a/python/sglang/srt/utils/scheduler_status_logger.py b/python/sglang/srt/utils/scheduler_status_logger.py index 0d76c9e8c..5f1f03caa 100644 --- a/python/sglang/srt/utils/scheduler_status_logger.py +++ b/python/sglang/srt/utils/scheduler_status_logger.py @@ -20,11 +20,17 @@ class SchedulerStatusLogger: self.rank = dist.get_rank() if dist.is_initialized() else 0 @staticmethod - def maybe_create() -> Optional["SchedulerStatusLogger"]: + def maybe_create(enable_metrics: bool) -> Optional["SchedulerStatusLogger"]: target = envs.SGLANG_LOG_SCHEDULER_STATUS_TARGET.get() if not target: return None + if not enable_metrics: + raise ValueError( + "SGLANG_LOG_SCHEDULER_STATUS_TARGET is set but --enable-metrics " + "is not active. Status dumps require --enable-metrics to work." + ) + return SchedulerStatusLogger( targets=[t.strip() for t in target.split(",") if t.strip()], dump_interval=envs.SGLANG_LOG_SCHEDULER_STATUS_INTERVAL.get(), diff --git a/test/registered/utils/test_scheduler_status_logger.py b/test/registered/utils/test_scheduler_status_logger.py index 1e366d008..2247afbd4 100644 --- a/test/registered/utils/test_scheduler_status_logger.py +++ b/test/registered/utils/test_scheduler_status_logger.py @@ -33,7 +33,7 @@ class TestSchedulerStatusLogger(CustomTestCase): "Qwen/Qwen3-0.6B", DEFAULT_URL_FOR_TEST, timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, - other_args=["--skip-server-warmup"], + other_args=["--skip-server-warmup", "--enable-metrics"], env=env, ) cls.addClassCleanup(kill_process_tree, cls.process.pid)