Decouple scheduler log printing from metrics collection (#20107)
This commit is contained in:
@@ -535,9 +535,8 @@ class SchedulerDisaggregationPrefillMixin:
|
|||||||
self.send_kv_chunk(req, last_chunk=False, end_idx=req.tmp_end_idx)
|
self.send_kv_chunk(req, last_chunk=False, end_idx=req.tmp_end_idx)
|
||||||
req.time_stats.set_last_chunked_prefill_finish_time()
|
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)
|
can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
|
||||||
self.log_prefill_stats(
|
self.report_prefill_stats(
|
||||||
prefill_stats=batch.prefill_stats,
|
prefill_stats=batch.prefill_stats,
|
||||||
can_run_cuda_graph=can_run_cuda_graph,
|
can_run_cuda_graph=can_run_cuda_graph,
|
||||||
dp_cooperation_info=batch.dp_cooperation_info,
|
dp_cooperation_info=batch.dp_cooperation_info,
|
||||||
|
|||||||
@@ -92,9 +92,8 @@ class SchedulerDllmMixin:
|
|||||||
self.stream_output(batch.reqs, batch.return_logprob)
|
self.stream_output(batch.reqs, batch.return_logprob)
|
||||||
self.token_to_kv_pool_allocator.free_group_end()
|
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)
|
can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
|
||||||
self.log_prefill_stats(
|
self.report_prefill_stats(
|
||||||
prefill_stats=batch.prefill_stats,
|
prefill_stats=batch.prefill_stats,
|
||||||
can_run_cuda_graph=can_run_cuda_graph,
|
can_run_cuda_graph=can_run_cuda_graph,
|
||||||
dp_cooperation_info=batch.dp_cooperation_info,
|
dp_cooperation_info=batch.dp_cooperation_info,
|
||||||
|
|||||||
@@ -310,10 +310,6 @@ class Scheduler(
|
|||||||
self.enable_overlap = not server_args.disable_overlap_schedule
|
self.enable_overlap = not server_args.disable_overlap_schedule
|
||||||
self.enable_pdmux = server_args.enable_pdmux
|
self.enable_pdmux = server_args.enable_pdmux
|
||||||
self.skip_tokenizer_init = server_args.skip_tokenizer_init
|
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.stream_interval = server_args.stream_interval
|
||||||
self.spec_algorithm = SpeculativeAlgorithm.from_string(
|
self.spec_algorithm = SpeculativeAlgorithm.from_string(
|
||||||
server_args.speculative_algorithm
|
server_args.speculative_algorithm
|
||||||
|
|||||||
@@ -306,13 +306,8 @@ class SchedulerOutputProcessorMixin:
|
|||||||
|
|
||||||
self.stream_output(batch.reqs, batch.return_logprob, skip_stream_req)
|
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)
|
can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
|
||||||
if self.enable_metrics:
|
self.report_prefill_stats(
|
||||||
self.metrics_collector.increment_prefill_cuda_graph_pass(
|
|
||||||
value=can_run_cuda_graph
|
|
||||||
)
|
|
||||||
self.log_prefill_stats(
|
|
||||||
prefill_stats=batch.prefill_stats,
|
prefill_stats=batch.prefill_stats,
|
||||||
can_run_cuda_graph=can_run_cuda_graph,
|
can_run_cuda_graph=can_run_cuda_graph,
|
||||||
dp_cooperation_info=batch.dp_cooperation_info,
|
dp_cooperation_info=batch.dp_cooperation_info,
|
||||||
@@ -485,11 +480,10 @@ class SchedulerOutputProcessorMixin:
|
|||||||
self.token_to_kv_pool_allocator.free_group_end()
|
self.token_to_kv_pool_allocator.free_group_end()
|
||||||
|
|
||||||
self.forward_ct_decode = (self.forward_ct_decode + 1) % (1 << 30)
|
self.forward_ct_decode = (self.forward_ct_decode + 1) % (1 << 30)
|
||||||
if self.current_scheduler_metrics_enabled:
|
self.report_decode_stats(
|
||||||
if self.forward_ct_decode % self.server_args.decode_log_interval == 0:
|
can_run_cuda_graph,
|
||||||
self.log_decode_stats(can_run_cuda_graph, running_batch=batch)
|
running_batch=batch,
|
||||||
self.log_decode_stats_every_iteration(
|
num_accepted_tokens=result.num_accepted_tokens,
|
||||||
batch, num_accepted_tokens=result.num_accepted_tokens
|
|
||||||
)
|
)
|
||||||
|
|
||||||
def _mamba_prefix_cache_update(
|
def _mamba_prefix_cache_update(
|
||||||
|
|||||||
@@ -271,8 +271,7 @@ class SchedulerRuntimeCheckerMixin:
|
|||||||
self._check_req_pool()
|
self._check_req_pool()
|
||||||
|
|
||||||
if (
|
if (
|
||||||
self.enable_metrics
|
self.current_scheduler_metrics_enabled
|
||||||
and self.current_scheduler_metrics_enabled
|
|
||||||
and time.perf_counter() > self.metrics_collector.last_log_time + 30
|
and time.perf_counter() > self.metrics_collector.last_log_time + 30
|
||||||
):
|
):
|
||||||
# During idle time, also collect metrics every 30 seconds.
|
# During idle time, also collect metrics every 30 seconds.
|
||||||
|
|||||||
@@ -114,8 +114,10 @@ class SchedulerMetricsMixin:
|
|||||||
self.stats = SchedulerStats()
|
self.stats = SchedulerStats()
|
||||||
|
|
||||||
# Metrics
|
# Metrics
|
||||||
self.current_scheduler_metrics_enabled = (
|
self.enable_metrics = self.server_args.enable_metrics
|
||||||
self.attn_tp_rank == 0 or self.enable_metrics_for_all_schedulers
|
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.enable_metrics:
|
||||||
if self.server_args.disaggregation_mode == DisaggregationMode.PREFILL.value:
|
if self.server_args.disaggregation_mode == DisaggregationMode.PREFILL.value:
|
||||||
@@ -158,7 +160,9 @@ class SchedulerMetricsMixin:
|
|||||||
if self.enable_kv_cache_events:
|
if self.enable_kv_cache_events:
|
||||||
self.init_kv_events(self.server_args.kv_events_config)
|
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]):
|
def init_kv_events(self: Scheduler, kv_events_config: Optional[str]):
|
||||||
if self.enable_kv_cache_events:
|
if self.enable_kv_cache_events:
|
||||||
@@ -179,12 +183,18 @@ class SchedulerMetricsMixin:
|
|||||||
self.spec_total_num_accepted_tokens = 0
|
self.spec_total_num_accepted_tokens = 0
|
||||||
self.spec_total_num_forward_ct = 0
|
self.spec_total_num_forward_ct = 0
|
||||||
|
|
||||||
def log_prefill_stats(
|
def report_prefill_stats(
|
||||||
self: Scheduler,
|
self: Scheduler,
|
||||||
prefill_stats: PrefillStats,
|
prefill_stats: PrefillStats,
|
||||||
can_run_cuda_graph: bool,
|
can_run_cuda_graph: bool,
|
||||||
dp_cooperation_info: Optional[DPCooperationInfo] = None,
|
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
|
gap_latency = time.perf_counter() - self.last_prefill_stats_tic
|
||||||
self.last_prefill_stats_tic = time.perf_counter()
|
self.last_prefill_stats_tic = time.perf_counter()
|
||||||
self.last_input_throughput = self.last_prefill_tokens / gap_latency
|
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}"
|
msg += f"{graph_backend[self.device]}: {can_run_cuda_graph}"
|
||||||
|
|
||||||
|
if self.is_stats_logging_rank:
|
||||||
logger.info(msg)
|
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(
|
self.metrics_collector.increment_realtime_tokens(
|
||||||
prefill_compute_tokens=prefill_stats.log_input_tokens,
|
prefill_compute_tokens=prefill_stats.log_input_tokens,
|
||||||
prefill_cache_tokens=prefill_stats.log_hit_tokens,
|
prefill_cache_tokens=prefill_stats.log_hit_tokens,
|
||||||
@@ -332,11 +346,34 @@ class SchedulerMetricsMixin:
|
|||||||
self._emit_kv_metrics()
|
self._emit_kv_metrics()
|
||||||
self._publish_kv_events()
|
self._publish_kv_events()
|
||||||
|
|
||||||
def log_decode_stats(
|
def report_decode_stats(
|
||||||
self: Scheduler, can_run_cuda_graph: bool, running_batch: ScheduleBatch = None
|
self: Scheduler,
|
||||||
|
can_run_cuda_graph: bool,
|
||||||
|
running_batch: ScheduleBatch = None,
|
||||||
|
num_accepted_tokens: int = 0,
|
||||||
):
|
):
|
||||||
batch = running_batch or self.running_batch
|
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
|
gap_latency = time.perf_counter() - self.last_decode_stats_tic
|
||||||
self.last_decode_stats_tic = time.perf_counter()
|
self.last_decode_stats_tic = time.perf_counter()
|
||||||
self.last_gen_throughput = self.num_generated_tokens / gap_latency
|
self.last_gen_throughput = self.num_generated_tokens / gap_latency
|
||||||
@@ -455,8 +492,9 @@ class SchedulerMetricsMixin:
|
|||||||
f"#queue-req: {len(self.waiting_queue)}"
|
f"#queue-req: {len(self.waiting_queue)}"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if self.is_stats_logging_rank:
|
||||||
logger.info(msg)
|
logger.info(msg)
|
||||||
if self.enable_metrics:
|
if self.current_scheduler_metrics_enabled:
|
||||||
priority_enabled = self.enable_priority_scheduling
|
priority_enabled = self.enable_priority_scheduling
|
||||||
# Basics
|
# Basics
|
||||||
self.stats.num_running_reqs = QueueCount.from_reqs(
|
self.stats.num_running_reqs = QueueCount.from_reqs(
|
||||||
@@ -524,19 +562,6 @@ class SchedulerMetricsMixin:
|
|||||||
self._emit_kv_metrics()
|
self._emit_kv_metrics()
|
||||||
self._publish_kv_events()
|
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(
|
def log_batch_result_stats(
|
||||||
self: Scheduler,
|
self: Scheduler,
|
||||||
batch: ScheduleBatch,
|
batch: ScheduleBatch,
|
||||||
|
|||||||
@@ -3905,7 +3905,7 @@ class ServerArgs:
|
|||||||
"--decode-log-interval",
|
"--decode-log-interval",
|
||||||
type=int,
|
type=int,
|
||||||
default=ServerArgs.decode_log_interval,
|
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(
|
parser.add_argument(
|
||||||
"--enable-request-time-stats-logging",
|
"--enable-request-time-stats-logging",
|
||||||
|
|||||||
@@ -20,11 +20,17 @@ class SchedulerStatusLogger:
|
|||||||
self.rank = dist.get_rank() if dist.is_initialized() else 0
|
self.rank = dist.get_rank() if dist.is_initialized() else 0
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def maybe_create() -> Optional["SchedulerStatusLogger"]:
|
def maybe_create(enable_metrics: bool) -> Optional["SchedulerStatusLogger"]:
|
||||||
target = envs.SGLANG_LOG_SCHEDULER_STATUS_TARGET.get()
|
target = envs.SGLANG_LOG_SCHEDULER_STATUS_TARGET.get()
|
||||||
if not target:
|
if not target:
|
||||||
return None
|
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(
|
return SchedulerStatusLogger(
|
||||||
targets=[t.strip() for t in target.split(",") if t.strip()],
|
targets=[t.strip() for t in target.split(",") if t.strip()],
|
||||||
dump_interval=envs.SGLANG_LOG_SCHEDULER_STATUS_INTERVAL.get(),
|
dump_interval=envs.SGLANG_LOG_SCHEDULER_STATUS_INTERVAL.get(),
|
||||||
|
|||||||
@@ -33,7 +33,7 @@ class TestSchedulerStatusLogger(CustomTestCase):
|
|||||||
"Qwen/Qwen3-0.6B",
|
"Qwen/Qwen3-0.6B",
|
||||||
DEFAULT_URL_FOR_TEST,
|
DEFAULT_URL_FOR_TEST,
|
||||||
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||||
other_args=["--skip-server-warmup"],
|
other_args=["--skip-server-warmup", "--enable-metrics"],
|
||||||
env=env,
|
env=env,
|
||||||
)
|
)
|
||||||
cls.addClassCleanup(kill_process_tree, cls.process.pid)
|
cls.addClassCleanup(kill_process_tree, cls.process.pid)
|
||||||
|
|||||||
Reference in New Issue
Block a user