minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 14 Packages Projects Releases Wiki Activity

Add SchedulerMetricsReporter and route metrics state through it (#25629)

Browse Source
This commit is contained in:
fzyzcjy
2026-05-18 18:41:35 +08:00
committed by GitHub
parent 8357d07569
commit 780d969699
10 changed files with 406 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/sglang/srt/disaggregation/decode.py
+3 -3
View File
@@ -637,7 +637,7 @@ class DecodePreallocQueue:
error_message,
status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
)
if self.scheduler.enable_metrics:
if self.scheduler.metrics_reporter.enable_metrics:
self.scheduler.metrics_collector.increment_bootstrap_failed_reqs()
else:
raise ValueError(f"Unexpected poll case: {poll}")
@@ -1516,7 +1516,7 @@ class DecodeTransferQueue:
# release pre-allocated kv cache, but don't insert into the tree since it's failed
release_kv_cache(decode_req.req, self.tree_cache, is_insert=False)
indices_to_remove.add(i)
if self.scheduler.enable_metrics:
if self.scheduler.metrics_reporter.enable_metrics:
self.scheduler.metrics_collector.increment_transfer_failed_reqs()
continue
elif poll == KVPoll.Success:
@@ -1535,7 +1535,7 @@ class DecodeTransferQueue:
release_kv_cache(
decode_req.req, self.tree_cache, is_insert=False
)
if self.scheduler.enable_metrics:
if self.scheduler.metrics_reporter.enable_metrics:
self.scheduler.metrics_collector.increment_transfer_failed_reqs()
else:
transferred_reqs.append(decode_req.req)
python/sglang/srt/disaggregation/encode_receiver.py
+1 -1
View File
@@ -1031,7 +1031,7 @@ class MMReceiverBase(ABC):
priority=recv_req.priority,
metrics_collector=(
self.scheduler.metrics_collector
if self.scheduler.enable_metrics
if self.scheduler.metrics_reporter.enable_metrics
else None
),
http_worker_ipc=recv_req.http_worker_ipc,
python/sglang/srt/disaggregation/prefill.py
+5 -4
View File
@@ -312,7 +312,7 @@ class PrefillBootstrapQueue:
self.scheduler.stream_output([req], req.return_logprob)
indices_to_remove.add(i)
failed_reqs.append(req)
if self.scheduler.enable_metrics:
if self.scheduler.metrics_reporter.enable_metrics:
self.scheduler.metrics_collector.increment_bootstrap_failed_reqs()
if self.scheduler.enable_hicache_storage:
# to release prefetch events associated with the request
@@ -588,6 +588,7 @@ class SchedulerDisaggregationPrefillMixin:
can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
self.report_prefill_stats(
self.metrics_reporter,
batch=batch,
prefill_stats=batch.prefill_stats,
can_run_cuda_graph=can_run_cuda_graph,
@@ -659,7 +660,7 @@ class SchedulerDisaggregationPrefillMixin:
req, error_message, status_code=HTTPStatus.INTERNAL_SERVER_ERROR
)
done_reqs.append(req)
if self.enable_metrics:
if self.metrics_reporter.enable_metrics:
self.metrics_collector.increment_transfer_failed_reqs()
else:
logger.warning_once(
@@ -685,9 +686,9 @@ class SchedulerDisaggregationPrefillMixin:
if metrics:
# Update last-value for REST API
if "latency_ms" in metrics:
self.kv_transfer_latency_ms = metrics["latency_ms"]
self.metrics_reporter.kv_transfer_latency_ms = metrics["latency_ms"]
if "speed_gb_s" in metrics:
self.kv_transfer_speed_gb_s = metrics["speed_gb_s"]
self.metrics_reporter.kv_transfer_speed_gb_s = metrics["speed_gb_s"]
# Stream requests which have finished transfer
self.stream_output(
python/sglang/srt/dllm/mixin/scheduler.py
+2 -1
View File
@@ -80,7 +80,7 @@ class SchedulerDllmMixin:
continue
req.fill_ids[-new_tokens:] = next_token_ids[:]
self.num_generated_tokens += new_tokens
self.metrics_reporter.num_generated_tokens += new_tokens
req.output_ids.extend(next_token_ids)
req.check_finished(new_accepted_len=new_tokens)
@@ -94,6 +94,7 @@ class SchedulerDllmMixin:
can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
self.report_prefill_stats(
self.metrics_reporter,
batch=batch,
prefill_stats=batch.prefill_stats,
can_run_cuda_graph=can_run_cuda_graph,
python/sglang/srt/managers/scheduler.py
+67 -30
View File
@@ -176,6 +176,9 @@ from sglang.srt.managers.scheduler_components.kv_events_publisher import (
from sglang.srt.managers.scheduler_components.load_inquirer import (
SchedulerLoadInquirer,
)
from sglang.srt.managers.scheduler_components.metrics_reporter import (
SchedulerMetricsReporter,
)
from sglang.srt.managers.scheduler_components.pool_stats_observer import (
SchedulerPoolStatsObserver,
)
@@ -203,6 +206,7 @@ from sglang.srt.mem_cache.common import maybe_cache_unfinished_req, release_kv_c
from sglang.srt.model_executor.forward_batch_info import ForwardMode, PPProxyTensors
from sglang.srt.model_loader.utils import get_resolved_model_impl
from sglang.srt.multiplex.multiplexing_mixin import SchedulerMultiplexMixin
from sglang.srt.observability.metrics_collector import SchedulerMetricsCollector
from sglang.srt.observability.req_time_stats import (
real_time,
set_schedule_time_batch,
@@ -452,7 +456,17 @@ class Scheduler(
self.init_model_config()
# Init metrics stats
self.init_metrics(tp_rank, pp_rank, dp_rank)
self.metrics_collector_context = SchedulerMetricsCollector.init_new(
server_args=self.server_args,
ps=self.ps,
tp_rank=tp_rank,
pp_rank=pp_rank,
dp_rank=dp_rank,
enable_priority_scheduling=self.enable_priority_scheduling,
enable_lora=self.enable_lora,
enable_hierarchical_cache=self.enable_hierarchical_cache,
)
self.metrics_collector = self.metrics_collector_context.collector
# Init inter-process communication
self.init_ipc_channels(port_args)
@@ -481,7 +495,6 @@ class Scheduler(
# Launch a model worker and draft model worker if using speculative decoding
self.init_model_worker()
self.install_device_timer_on_runners()
if (t := envs.SGLANG_TEST_STUCK_SCHEDULER_INIT.get()) > 0:
time.sleep(t)
@@ -496,7 +509,7 @@ class Scheduler(
attn_tp_cpu_group=self.attn_tp_cpu_group,
tp_cpu_group=self.tp_cpu_group,
attn_cp_cpu_group=self.attn_cp_cpu_group,
enable_metrics=self.enable_metrics,
enable_metrics=self.server_args.enable_metrics,
enable_kv_cache_events=bool(
self.server_args.kv_events_config
and self.ps.attn_tp_rank == 0
@@ -559,6 +572,15 @@ class Scheduler(
# Init diffusion LLM
self.init_diffusion_llm()
self.metrics_reporter = SchedulerMetricsReporter(
scheduler=self,
tp_rank=tp_rank,
pp_rank=pp_rank,
dp_rank=dp_rank,
metrics_collector_context=self.metrics_collector_context,
metrics_collector=self.metrics_collector,
)
# Init schedule policy and new token estimation
self.init_schedule_policy()
@@ -695,7 +717,7 @@ class Scheduler(
send_metrics_from_scheduler=self.send_metrics_from_scheduler,
max_running_requests=self.max_running_requests,
max_total_num_tokens=self.max_total_num_tokens,
get_stats=lambda: self.stats,
get_stats=lambda: self.metrics_reporter.stats,
)
self.load_inquirer = SchedulerLoadInquirer(
@@ -710,14 +732,14 @@ class Scheduler(
spec_algorithm=self.spec_algorithm,
get_running_batch=lambda: self.running_batch,
get_waiting_queue=lambda: self.waiting_queue,
get_stats=lambda: self.stats,
get_stats=lambda: self.metrics_reporter.stats,
get_chunked_req=lambda: self.chunked_req,
get_disagg_prefill_bootstrap_queue=lambda: self.disagg_prefill_bootstrap_queue,
get_disagg_prefill_inflight_queue=lambda: self.disagg_prefill_inflight_queue,
get_disagg_decode_prealloc_queue=lambda: self.disagg_decode_prealloc_queue,
get_disagg_decode_transfer_queue=lambda: self.disagg_decode_transfer_queue,
get_spec_total_num_accept_tokens=lambda: self.spec_total_num_accept_tokens,
get_spec_total_num_forward_ct=lambda: self.spec_total_num_forward_ct,
get_spec_total_num_accept_tokens=lambda: self.metrics_reporter.spec_total_num_accept_tokens,
get_spec_total_num_forward_ct=lambda: self.metrics_reporter.spec_total_num_forward_ct,
)
self.is_initializing = False
@@ -800,7 +822,10 @@ class Scheduler(
self.send_to_tokenizer = SenderWrapper(None)
self.send_to_detokenizer = SenderWrapper(None)
if self.current_scheduler_metrics_enabled:
if self.server_args.enable_metrics and (
self.ps.attn_tp_rank == 0
or self.server_args.enable_metrics_for_all_schedulers
):
self.send_metrics_from_scheduler = get_zmq_socket(
context, zmq.PUSH, port_args.metrics_ipc_name, False
)
@@ -1010,7 +1035,7 @@ class Scheduler(
f"{'available_cpu_mem' if self.device == 'cpu' else 'available_gpu_mem'}={avail_mem:.2f} GB"
)
if self.enable_metrics:
if self.server_args.enable_metrics:
self.metrics_collector.emit_constants(
max_total_num_tokens=self.max_total_num_tokens,
# TODO: max_running_requests_under_SLO has no setter — dead chain.
@@ -1036,8 +1061,6 @@ class Scheduler(
self.forward_ct = 0
self.return_health_check_ipcs: Deque[Optional[str]] = deque()
self._pending_flush: Optional[Tuple[FlushCacheReqInput, float]] = None
self.num_retracted_reqs: int = 0
self.num_paused_reqs: int = 0
self.session_controller = SessionController(self.tree_cache)
self.forward_sleep_time = None
self._engine_paused = False
@@ -1114,7 +1137,9 @@ class Scheduler(
device_group=self.tp_group.device_group,
server_args=self.server_args,
metrics_collector=(
self.metrics_collector if self.enable_metrics else None
self.metrics_collector
if self.metrics_reporter.enable_metrics
else None
),
max_delay_passes=self.server_args.prefill_delayer_max_delay_passes,
token_usage_low_watermark=self.server_args.prefill_delayer_token_usage_low_watermark,
@@ -1860,7 +1885,9 @@ class Scheduler(
vocab_size=self.model_config.vocab_size,
priority=recv_req.priority,
metrics_collector=(
self.metrics_collector if self.enable_metrics else None
self.metrics_collector
if self.metrics_reporter.enable_metrics
else None
),
routing_key=recv_req.routing_key,
extra_key=recv_req.extra_key,
@@ -1902,7 +1929,7 @@ class Scheduler(
eos_token_ids=self.model_config.hf_eos_token_id,
)
# TODO: set trace context
if self.enable_metrics:
if self.metrics_reporter.enable_metrics:
req.time_stats.set_metrics_collector(self.metrics_collector)
if isinstance(req.finished_reason, FINISH_ABORT):
self.init_req_max_new_tokens(req)
@@ -2752,9 +2779,9 @@ class Scheduler(
else None
)
self.num_retracted_reqs = len(retracted_reqs)
if self.enable_metrics and len(retracted_reqs) > 0:
self.metrics_collector.increment_retracted_reqs(
self.metrics_reporter.num_retracted_reqs = len(retracted_reqs)
if self.metrics_reporter.enable_metrics and len(retracted_reqs) > 0:
self.metrics_reporter.metrics_collector.increment_retracted_reqs(
num_retracted_reqs=len(retracted_reqs),
num_retracted_input_tokens=sum(
len(r.origin_input_ids) for r in retracted_reqs
@@ -3040,15 +3067,15 @@ class Scheduler(
elif batch.forward_mode.is_idle():
self.process_batch_result_idle(batch, result)
self.log_batch_result_stats(batch, result)
self.log_batch_result_stats(self.metrics_reporter, batch, result)
# Emit forward pass metrics (every iteration when enabled)
if self.enable_fpm:
self._emit_forward_pass_metrics(batch, result)
self._emit_forward_pass_metrics(self.metrics_reporter, batch, result)
self._maybe_clear_mm_inputs(batch)
self.maybe_send_health_check_signal()
self.update_device_timer()
self.update_device_timer(self.metrics_reporter)
def maybe_send_health_check_signal(self):
if self.return_health_check_ipcs:
@@ -3174,7 +3201,7 @@ class Scheduler(
self.new_token_ratio = self.init_new_token_ratio
# reset device timer window so idle time isn't counted
self.reset_device_timer_window()
self.reset_device_timer_window(self.metrics_reporter)
# sleep until next event
self.maybe_sleep_on_idle()
@@ -3335,7 +3362,7 @@ class Scheduler(
self.req_to_token_pool.clear()
self.token_to_kv_pool_allocator.clear()
self.grammar_manager.clear()
self.reset_metrics()
self.reset_metrics(self.metrics_reporter)
if self.draft_worker:
self.draft_worker.clear_cache_pool()
@@ -3358,7 +3385,7 @@ class Scheduler(
def get_internal_state(self, recv_req: GetInternalStateReq):
ret = dict(vars(get_global_server_args())) # vars returns a ref to obj.__dict__
ret["last_gen_throughput"] = self.last_gen_throughput
ret["last_gen_throughput"] = self.metrics_reporter.last_gen_throughput
ret["memory_usage"] = {
"weight": round(self.tp_worker.model_runner.weight_load_mem_usage, 2),
"kvcache": round(
@@ -3369,13 +3396,17 @@ class Scheduler(
}
ret["effective_max_running_requests_per_dp"] = self.max_running_requests
if not self.spec_algorithm.is_none() and self.spec_total_num_forward_ct > 0:
if (
not self.spec_algorithm.is_none()
and self.metrics_reporter.spec_total_num_forward_ct > 0
):
ret["avg_spec_accept_length"] = (
self.spec_total_num_accept_tokens / self.spec_total_num_forward_ct
self.metrics_reporter.spec_total_num_accept_tokens
/ self.metrics_reporter.spec_total_num_forward_ct
)
if RECORD_STEP_TIME:
ret["step_time_dict"] = self.step_time_dict
ret["step_time_dict"] = self.metrics_reporter.step_time_dict
# This field is not serializable.
ret.pop("model_config", None)
@@ -3408,12 +3439,18 @@ class Scheduler(
break
if if_success:
if not self.spec_algorithm.is_none() and self.spec_total_num_forward_ct > 0:
if (
not self.spec_algorithm.is_none()
and self.metrics_reporter.spec_total_num_forward_ct > 0
):
avg_spec_accept_length = (
self.spec_total_num_accept_tokens / self.spec_total_num_forward_ct
self.metrics_reporter.spec_total_num_accept_tokens
/ self.metrics_reporter.spec_total_num_forward_ct
)
logger.info(f"{avg_spec_accept_length=}")
self.spec_total_num_accept_tokens = self.spec_total_num_forward_ct = 0
self.metrics_reporter.spec_total_num_accept_tokens = (
self.metrics_reporter.spec_total_num_forward_ct
) = 0
for k, v in server_args_dict.items():
setattr(get_global_server_args(), k, v)
logger.info(f"Global server args updated! {get_global_server_args()=}")
@@ -3944,4 +3981,4 @@ def run_scheduler_process(
if scheduler is not None:
# FPM has a background ZMQ publisher thread that needs explicit
# teardown to flush queued metrics and close the socket cleanly.
scheduler._shutdown_fpm()
scheduler._shutdown_fpm(scheduler.metrics_reporter)
python/sglang/srt/managers/scheduler_components/metrics_reporter.py
+47
View File
@@ -0,0 +1,47 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING, Optional
from sglang.srt.observability.metrics_collector import (
SchedulerMetricsCollector,
SchedulerMetricsCollectorContext,
)
from sglang.srt.observability.scheduler_metrics_mixin import (
SchedulerMetricsMixin,
)
if TYPE_CHECKING:
from sglang.srt.managers.scheduler import Scheduler
logger = logging.getLogger(__name__)
@dataclass(kw_only=True)
class SchedulerMetricsReporter:
scheduler: "Scheduler"
tp_rank: int
pp_rank: int
dp_rank: Optional[int]
metrics_collector_context: SchedulerMetricsCollectorContext
metrics_collector: Optional[SchedulerMetricsCollector]
num_retracted_reqs: int = 0
num_paused_reqs: int = 0
def __post_init__(self) -> None:
self.enable_metrics = self.metrics_collector_context.enable_metrics
self.is_stats_logging_rank = (
self.metrics_collector_context.is_stats_logging_rank
)
self.current_scheduler_metrics_enabled = (
self.metrics_collector_context.current_scheduler_metrics_enabled
)
self.enable_kv_cache_events = (
self.metrics_collector_context.enable_kv_cache_events
)
SchedulerMetricsMixin._init_metrics(
self, self.tp_rank, self.pp_rank, self.dp_rank
)
SchedulerMetricsMixin._install_device_timer_on_runners(self)
python/sglang/srt/managers/scheduler_output_processor_mixin.py
+10 -4
View File
@@ -400,6 +400,7 @@ class SchedulerOutputProcessorMixin:
can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
self.report_prefill_stats(
self.metrics_reporter,
batch=batch,
prefill_stats=batch.prefill_stats,
can_run_cuda_graph=can_run_cuda_graph,
@@ -511,10 +512,12 @@ class SchedulerOutputProcessorMixin:
# else: Spec V1 — output_ids, check_finished, grammar, and reasoning tokens
# are already handled in the verify phase (eagle_info.py / ngram_info.py).
self.num_generated_tokens += len(batch.reqs)
self.metrics_reporter.num_generated_tokens += len(batch.reqs)
if not batch.spec_algorithm.is_none():
self.update_spec_metrics(batch.batch_size(), result.num_correct_drafts)
if self.enable_metrics:
self.update_spec_metrics(
self.metrics_reporter, batch.batch_size(), result.num_correct_drafts
)
if self.metrics_reporter.enable_metrics:
self.metrics_collector.increment_decode_cuda_graph_pass(
value=can_run_cuda_graph
)
@@ -624,8 +627,11 @@ class SchedulerOutputProcessorMixin:
self.stream_output(batch.reqs, batch.return_logprob)
self.token_to_kv_pool_allocator.free_group_end()
self.forward_ct_decode = (self.forward_ct_decode + 1) % (1 << 30)
self.metrics_reporter.forward_ct_decode = (
self.metrics_reporter.forward_ct_decode + 1
) % (1 << 30)
self.report_decode_stats(
self.metrics_reporter,
can_run_cuda_graph,
running_batch=batch,
num_correct_drafts=result.num_correct_drafts,
python/sglang/srt/observability/metrics_collector.py
+66
View File
@@ -23,6 +23,7 @@ from collections import Counter
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union
from sglang.srt.disaggregation.utils import DisaggregationMode
from sglang.srt.environ import envs
from sglang.srt.model_executor.forward_batch_info import ForwardMode
from sglang.srt.observability.utils import exponential_buckets, generate_buckets
@@ -182,6 +183,15 @@ class DPCooperationInfo:
return dataclasses.asdict(self)
@dataclass(kw_only=True, frozen=True, slots=True)
class SchedulerMetricsCollectorContext:
enable_metrics: bool
is_stats_logging_rank: bool
current_scheduler_metrics_enabled: bool
enable_kv_cache_events: bool
collector: Optional["SchedulerMetricsCollector"]
class SchedulerMetricsCollector:
def __init__(
@@ -938,6 +948,62 @@ class SchedulerMetricsCollector:
multiprocess_mode="mostrecent",
)
@classmethod
def init_new(
cls,
*,
server_args: "ServerArgs",
ps: Any,
tp_rank: int,
pp_rank: int,
dp_rank: Optional[int],
enable_priority_scheduling: bool,
enable_lora: bool,
enable_hierarchical_cache: bool,
) -> "SchedulerMetricsCollectorContext":
enable_metrics = server_args.enable_metrics
is_stats_logging_rank = ps.attn_tp_rank == 0
current_scheduler_metrics_enabled = enable_metrics and (
is_stats_logging_rank or server_args.enable_metrics_for_all_schedulers
)
enable_kv_cache_events = bool(
server_args.kv_events_config
and ps.attn_tp_rank == 0
and ps.attn_cp_rank == 0
)
collector: Optional["SchedulerMetricsCollector"] = None
if enable_metrics:
engine_type = DisaggregationMode.to_engine_type(
server_args.disaggregation_mode
)
labels = {
"model_name": server_args.served_model_name,
"engine_type": engine_type,
"tp_rank": tp_rank,
"pp_rank": pp_rank,
"moe_ep_rank": ps.moe_ep_rank,
}
if enable_priority_scheduling:
labels["priority"] = ""
if dp_rank is not None:
labels["dp_rank"] = dp_rank
if server_args.extra_metric_labels:
labels.update(server_args.extra_metric_labels)
collector = cls(
labels=labels,
enable_lora=enable_lora,
enable_hierarchical_cache=enable_hierarchical_cache,
enable_streaming_session=server_args.enable_streaming_session,
server_args=server_args,
)
return SchedulerMetricsCollectorContext(
enable_metrics=enable_metrics,
is_stats_logging_rank=is_stats_logging_rank,
current_scheduler_metrics_enabled=current_scheduler_metrics_enabled,
enable_kv_cache_events=enable_kv_cache_events,
collector=collector,
)
def _log_gauge(self, gauge: Gauge, data: Union[int, float]) -> None:
# Convenience function for logging a scalar to gauge.
gauge.labels(**self.labels).set(data)
python/sglang/srt/observability/scheduler_metrics_mixin.py
+204 -177
View File
@@ -14,7 +14,6 @@ from sglang.srt.managers.utils import GenerationBatchResult
from sglang.srt.observability.metrics_collector import (
DPCooperationInfo,
QueueCount,
SchedulerMetricsCollector,
SchedulerStats,
compute_routing_key_stats,
)
@@ -24,7 +23,7 @@ from sglang.srt.utils.scheduler_status_logger import SchedulerStatusLogger
if TYPE_CHECKING:
from sglang.srt.managers.schedule_batch import Req
from sglang.srt.managers.schedule_policy import PrefillAdder
from sglang.srt.managers.scheduler import EmbeddingBatchResult, Scheduler
from sglang.srt.managers.scheduler import EmbeddingBatchResult
logger = logging.getLogger(__name__)
@@ -67,8 +66,12 @@ class PrefillStats:
class SchedulerMetricsMixin:
enable_fpm: bool = False
def init_metrics(
self: Scheduler, tp_rank: int, pp_rank: int, dp_rank: Optional[int]
@staticmethod
def _init_metrics(
self: "SchedulerMetricsReporter",
tp_rank: int,
pp_rank: int,
dp_rank: Optional[int],
):
# Basic stats
self.forward_ct_decode = 0
@@ -83,7 +86,7 @@ class SchedulerMetricsMixin:
"cpu": "cpu graph",
"npu": "npu graph",
"musa": "musa graph",
}.get(getattr(self, "device", ""), "cuda graph")
}.get(getattr(self.scheduler, "device", ""), "cuda graph")
# Cumulative spec-decoding counters (reset every decode_log_interval).
# Each update adds (num_correct_drafts + bs, bs).
@@ -97,43 +100,12 @@ class SchedulerMetricsMixin:
self.kv_transfer_speed_gb_s: float = 0.0
self.kv_transfer_latency_ms: float = 0.0
# Metrics
self.enable_metrics = self.server_args.enable_metrics
self.is_stats_logging_rank = self.ps.attn_tp_rank == 0
self.current_scheduler_metrics_enabled = self.enable_metrics and (
self.is_stats_logging_rank
or self.server_args.enable_metrics_for_all_schedulers
)
self.enable_mfu_metrics = False
if self.enable_metrics:
engine_type = DisaggregationMode.to_engine_type(
self.server_args.disaggregation_mode
)
labels = {
"model_name": self.server_args.served_model_name,
"engine_type": engine_type,
"tp_rank": tp_rank,
"pp_rank": pp_rank,
"moe_ep_rank": self.ps.moe_ep_rank,
}
if self.enable_priority_scheduling:
labels["priority"] = ""
if dp_rank is not None:
labels["dp_rank"] = dp_rank
if self.server_args.extra_metric_labels:
labels.update(self.server_args.extra_metric_labels)
self.metrics_collector = SchedulerMetricsCollector(
labels=labels,
enable_lora=self.enable_lora,
enable_hierarchical_cache=self.enable_hierarchical_cache,
enable_streaming_session=self.server_args.enable_streaming_session,
server_args=self.server_args,
)
self.enable_mfu_metrics = self.server_args.enable_mfu_metrics
self.enable_mfu_metrics = self.scheduler.server_args.enable_mfu_metrics
if self.enable_mfu_metrics:
self._init_estimated_perf_constants()
SchedulerMetricsMixin._init_estimated_perf_constants(self)
self._mfu_log_flops = 0.0
self._mfu_log_read_bytes = 0.0
self._mfu_log_write_bytes = 0.0
@@ -156,54 +128,62 @@ class SchedulerMetricsMixin:
reporter=_wrap_execution_reporter,
)
self._init_fpm()
SchedulerMetricsMixin._init_fpm(self)
self.scheduler_status_logger = SchedulerStatusLogger.maybe_create(
enable_metrics=self.enable_metrics
)
def install_device_timer_on_runners(self: Scheduler):
@staticmethod
def _install_device_timer_on_runners(self: "SchedulerMetricsReporter"):
if self.forward_pass_device_timer is None:
return
timer = self.forward_pass_device_timer
self.tp_worker.model_runner.device_timer = timer
if self.draft_worker is not None:
dw = getattr(self.draft_worker, "draft_worker", None)
self.scheduler.tp_worker.model_runner.device_timer = timer
if self.scheduler.draft_worker is not None:
dw = getattr(self.scheduler.draft_worker, "draft_worker", None)
if dw is not None:
if hasattr(dw, "draft_runner"):
dw.draft_runner.device_timer = timer
for r in getattr(dw, "draft_runner_list", []):
r.device_timer = timer
def _init_fpm(self: Scheduler):
@staticmethod
def _init_fpm(self: "SchedulerMetricsReporter"):
"""Initialize Forward Pass Metrics (FPM) publisher if configured."""
self.enable_fpm = False
self.scheduler.enable_fpm = False
if (
self.server_args.enable_forward_pass_metrics
and self.ps.attn_tp_rank == 0
and self.ps.pp_rank == self.ps.pp_size - 1
self.scheduler.server_args.enable_forward_pass_metrics
and self.scheduler.ps.attn_tp_rank == 0
and self.scheduler.ps.pp_rank == self.scheduler.ps.pp_size - 1
):
from sglang.srt.observability.forward_pass_metrics import (
_FpmPublisherThread,
)
self._fpm_dp_rank = self.ps.dp_rank if self.ps.dp_rank is not None else 0
self._fpm_worker_id = self.server_args.forward_pass_metrics_worker_id
base_endpoint = self.server_args.forward_pass_metrics_ipc_name
self.scheduler._fpm_dp_rank = (
self.scheduler.ps.dp_rank
if self.scheduler.ps.dp_rank is not None
else 0
)
self.scheduler._fpm_worker_id = (
self.scheduler.server_args.forward_pass_metrics_worker_id
)
base_endpoint = self.scheduler.server_args.forward_pass_metrics_ipc_name
if base_endpoint is None:
ipc_path = tempfile.NamedTemporaryFile(delete=False).name
base_endpoint = f"ipc://{ipc_path}"
self.server_args.forward_pass_metrics_ipc_name = base_endpoint
endpoint = f"{base_endpoint}.{self._fpm_dp_rank}"
self._fpm_publisher = _FpmPublisherThread(
self.scheduler.server_args.forward_pass_metrics_ipc_name = base_endpoint
endpoint = f"{base_endpoint}.{self.scheduler._fpm_dp_rank}"
self.scheduler._fpm_publisher = _FpmPublisherThread(
endpoint,
worker_id=self._fpm_worker_id,
dp_rank=self._fpm_dp_rank,
worker_id=self.scheduler._fpm_worker_id,
dp_rank=self.scheduler._fpm_dp_rank,
)
self._fpm_gpu_time_acc = 0.0
self.scheduler._fpm_gpu_time_acc = 0.0
def _fpm_device_timer_reporter(t, **_kwargs):
self._fpm_gpu_time_acc += t
self.scheduler._fpm_gpu_time_acc += t
if self.forward_pass_device_timer is not None:
self.forward_pass_device_timer.add_reporter(_fpm_device_timer_reporter)
@@ -211,16 +191,19 @@ class SchedulerMetricsMixin:
self.forward_pass_device_timer = DeviceTimer(
reporter=_fpm_device_timer_reporter,
)
self._fpm_uses_device_timer = True
self.enable_fpm = True
self.scheduler._fpm_uses_device_timer = True
self.scheduler.enable_fpm = True
logger.info(
"FPM: ZMQ PUB bound on %s (dp_rank=%d, device_timer=%s)",
endpoint,
self._fpm_dp_rank,
self._fpm_uses_device_timer,
self.scheduler._fpm_dp_rank,
self.scheduler._fpm_uses_device_timer,
)
def _build_scheduled_request_metrics(self: Scheduler, batch: ScheduleBatch):
@staticmethod
def _build_scheduled_request_metrics(
self: "SchedulerMetricsReporter", batch: ScheduleBatch
):
from sglang.srt.observability.forward_pass_metrics import (
ScheduledRequestMetrics,
WelfordAccumulator,
@@ -265,7 +248,8 @@ class SchedulerMetricsMixin:
var_decode_kv_tokens=decode_kv.variance(),
)
def _build_queued_request_metrics(self: Scheduler):
@staticmethod
def _build_queued_request_metrics(self: "SchedulerMetricsReporter"):
from sglang.srt.observability.forward_pass_metrics import (
QueuedRequestMetrics,
WelfordAccumulator,
@@ -273,16 +257,16 @@ class SchedulerMetricsMixin:
prefill_q = WelfordAccumulator()
decode_q = WelfordAccumulator()
if self.disaggregation_mode == DisaggregationMode.PREFILL:
for req in self.disagg_prefill_bootstrap_queue.queue:
if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
for req in self.scheduler.disagg_prefill_bootstrap_queue.queue:
prefill_q.add(len(req.origin_input_ids))
elif self.disaggregation_mode == DisaggregationMode.DECODE:
for req in self.disagg_decode_prealloc_queue.queue:
elif self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
for req in self.scheduler.disagg_decode_prealloc_queue.queue:
decode_q.add(req.seqlen)
for req in self.disagg_decode_transfer_queue.queue:
for req in self.scheduler.disagg_decode_transfer_queue.queue:
decode_q.add(req.seqlen)
else:
for req in self.waiting_queue:
for req in self.scheduler.waiting_queue:
if len(req.output_ids) > 0:
decode_q.add(req.seqlen)
else:
@@ -297,22 +281,28 @@ class SchedulerMetricsMixin:
var_decode_kv_tokens=decode_q.variance(),
)
def update_spec_metrics(self: Scheduler, bs: int, num_correct_drafts: int):
@staticmethod
def update_spec_metrics(
self: "SchedulerMetricsReporter", bs: int, num_correct_drafts: int
):
self.spec_num_accept_tokens += num_correct_drafts + bs
self.spec_num_forward_ct += bs
# Bonus tokens updated elsewhere
self.num_generated_tokens += num_correct_drafts
def _init_estimated_perf_constants(self: Scheduler) -> None:
model_config = self.model_config
@staticmethod
def _init_estimated_perf_constants(self: "SchedulerMetricsReporter") -> None:
model_config = self.scheduler.model_config
hf_text_config = model_config.hf_text_config
hidden_size = float(model_config.hidden_size)
num_layers = float(getattr(model_config, "num_attention_layers", 0))
head_dim = float(getattr(model_config, "head_dim", 0))
num_attn_heads = float(model_config.get_num_attention_heads(self.ps.tp_size))
num_kv_heads = float(model_config.get_num_kv_heads(self.ps.tp_size))
num_attn_heads = float(
model_config.get_num_attention_heads(self.scheduler.ps.tp_size)
)
num_kv_heads = float(model_config.get_num_kv_heads(self.scheduler.ps.tp_size))
intermediate_size = getattr(hf_text_config, "intermediate_size", None)
if intermediate_size is None:
intermediate_size = getattr(hf_text_config, "ffn_hidden_size", 0)
@@ -386,8 +376,9 @@ class SchedulerMetricsMixin:
num_attn_heads * head_dim * act_bytes * num_layers
)
@staticmethod
def _estimate_prefill_perf(
self: Scheduler, num_tokens: int
self: "SchedulerMetricsReporter", num_tokens: int
) -> Tuple[float, float, float]:
tokens = max(0, int(num_tokens))
if tokens == 0:
@@ -412,8 +403,9 @@ class SchedulerMetricsMixin:
)
return flops, read_bytes, write_bytes
@staticmethod
def _estimate_decode_perf(
self: Scheduler, batch: ScheduleBatch, num_tokens: int
self: "SchedulerMetricsReporter", batch: ScheduleBatch, num_tokens: int
) -> Tuple[float, float, float]:
tokens = max(0, int(num_tokens))
if tokens == 0:
@@ -437,7 +429,8 @@ class SchedulerMetricsMixin:
)
return flops, read_bytes, write_bytes
def reset_metrics(self: Scheduler):
@staticmethod
def reset_metrics(self: "SchedulerMetricsReporter"):
self.forward_ct_decode = 0
self.num_generated_tokens = 0
self.spec_num_accept_tokens = 0
@@ -445,8 +438,9 @@ class SchedulerMetricsMixin:
self.spec_total_num_accept_tokens = 0
self.spec_total_num_forward_ct = 0
@staticmethod
def report_prefill_stats(
self: Scheduler,
self: "SchedulerMetricsReporter",
batch: Optional[ScheduleBatch],
prefill_stats: PrefillStats,
can_run_cuda_graph: bool,
@@ -465,14 +459,14 @@ class SchedulerMetricsMixin:
prefill_stats.log_input_tokens / gap_latency if gap_latency > 0 else 0.0
)
pool_stats = self.pool_stats_observer.get_pool_stats()
pool_stats = self.scheduler.pool_stats_observer.get_pool_stats()
token_usage_msg = ", ".join(pool_stats.get_prefill_usage_msg_parts()) + ", "
self.stats.new_token_ratio = prefill_stats.new_token_ratio
batch_iter = (
batch.forward_iter
if batch is not None and batch.forward_iter is not None
else self.forward_ct
else self.scheduler.forward_ct
)
iter_msg = f" [{batch_iter}]" if LOG_FORWARD_ITERS else ""
@@ -483,25 +477,32 @@ class SchedulerMetricsMixin:
f"#cached-token: {prefill_stats.log_hit_tokens}, "
f"{token_usage_msg}"
f"#running-req: {prefill_stats.num_running_reqs.total}, "
f"#queue-req: {len(self.waiting_queue)}, "
f"#queue-req: {len(self.scheduler.waiting_queue)}, "
f"#pending-token: {prefill_stats.num_pending_tokens}, "
)
if self.disaggregation_mode == DisaggregationMode.PREFILL:
msg += f"#bootstrap-req: {len(self.disagg_prefill_bootstrap_queue.queue)}, "
msg += f"#inflight-req: {len(self.disagg_prefill_inflight_queue)}, "
if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
msg += f"#bootstrap-req: {len(self.scheduler.disagg_prefill_bootstrap_queue.queue)}, "
msg += (
f"#inflight-req: {len(self.scheduler.disagg_prefill_inflight_queue)}, "
)
if (
self.server_args.language_only
and self.server_args.encoder_transfer_backend == "zmq_to_scheduler"
self.scheduler.server_args.language_only
and self.scheduler.server_args.encoder_transfer_backend
== "zmq_to_scheduler"
):
msg += f"waiting-image-req: {len(self.mm_receiver.waiting_list)}, "
msg += (
f"waiting-image-req: {len(self.scheduler.mm_receiver.waiting_list)}, "
)
msg += f"{self._graph_backend_label}: {can_run_cuda_graph}, "
msg += f"input throughput (token/s): {self.last_input_throughput:.2f}"
if self.enable_mfu_metrics and gap_latency > 0:
flops, _, _ = self._estimate_prefill_perf(prefill_stats.log_input_tokens)
flops, _, _ = SchedulerMetricsMixin._estimate_prefill_perf(
self, prefill_stats.log_input_tokens
)
tflops_per_s = flops / gap_latency / 1e12
msg += f", est. prefill TFLOPS/s (per GPU): {tflops_per_s:.2f}"
@@ -520,8 +521,10 @@ class SchedulerMetricsMixin:
dp_cooperation_info=dp_cooperation_info,
)
if self.enable_mfu_metrics:
flops, read_bytes, write_bytes = self._estimate_prefill_perf(
prefill_stats.log_input_tokens
flops, read_bytes, write_bytes = (
SchedulerMetricsMixin._estimate_prefill_perf(
self, prefill_stats.log_input_tokens
)
)
self.metrics_collector.increment_estimated_perf(
num_flops_per_gpu=flops,
@@ -529,7 +532,7 @@ class SchedulerMetricsMixin:
num_write_bytes_per_gpu=write_bytes,
)
priority_enabled = self.enable_priority_scheduling
priority_enabled = self.scheduler.enable_priority_scheduling
total_tokens = prefill_stats.log_input_tokens + prefill_stats.log_hit_tokens
cache_hit_rate = (
prefill_stats.log_hit_tokens / total_tokens if total_tokens > 0 else 0.0
@@ -538,9 +541,9 @@ class SchedulerMetricsMixin:
# Basics
self.stats.num_running_reqs = prefill_stats.num_running_reqs
self.stats.num_queue_reqs = QueueCount.from_reqs(
self.waiting_queue, priority_enabled
self.scheduler.waiting_queue, priority_enabled
)
self.stats.num_grammar_queue_reqs = len(self.grammar_manager)
self.stats.num_grammar_queue_reqs = len(self.scheduler.grammar_manager)
self.stats.cache_hit_rate = cache_hit_rate
# Memory pool usage ratios / Absolute token counts
@@ -552,39 +555,41 @@ class SchedulerMetricsMixin:
self.num_retracted_reqs = self.num_paused_reqs = 0
# PD disaggregation
if self.disaggregation_mode == DisaggregationMode.PREFILL:
if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
self.stats.num_prefill_bootstrap_queue_reqs = QueueCount.from_reqs(
self.disagg_prefill_bootstrap_queue.queue, priority_enabled
self.scheduler.disagg_prefill_bootstrap_queue.queue,
priority_enabled,
)
self.stats.num_prefill_inflight_queue_reqs = QueueCount.from_reqs(
self.disagg_prefill_inflight_queue, priority_enabled
self.scheduler.disagg_prefill_inflight_queue, priority_enabled
)
self.stats.kv_transfer_speed_gb_s = self.kv_transfer_speed_gb_s
self.stats.kv_transfer_latency_ms = self.kv_transfer_latency_ms
elif self.disaggregation_mode == DisaggregationMode.DECODE:
elif self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
self.stats.num_decode_prealloc_queue_reqs = QueueCount.from_reqs(
self.disagg_decode_prealloc_queue.queue, priority_enabled
self.scheduler.disagg_decode_prealloc_queue.queue, priority_enabled
)
self.stats.num_decode_transfer_queue_reqs = QueueCount.from_reqs(
self.disagg_decode_transfer_queue.queue, priority_enabled
self.scheduler.disagg_decode_transfer_queue.queue, priority_enabled
)
# Utilization / LoRA / HiCache
self.calculate_utilization()
SchedulerMetricsMixin._calculate_utilization(self)
self.stats.fwd_occupancy = self.fwd_occupancy
self.update_lora_metrics()
self._log_hicache_stats()
SchedulerMetricsMixin._update_lora_metrics(self)
SchedulerMetricsMixin._log_hicache_stats(self)
self.metrics_collector.log_stats(self.stats)
self.kv_events_publisher.emit_kv_metrics()
self.kv_events_publisher.publish_kv_events()
self.scheduler.kv_events_publisher.emit_kv_metrics()
self.scheduler.kv_events_publisher.publish_kv_events()
@staticmethod
def report_decode_stats(
self: Scheduler,
self: "SchedulerMetricsReporter",
can_run_cuda_graph: bool,
running_batch: ScheduleBatch = None,
num_correct_drafts: int = 0,
):
batch = running_batch or self.running_batch
batch = running_batch or self.scheduler.running_batch
# Every-iteration work: realtime token counting + status logger
if self.current_scheduler_metrics_enabled:
@@ -595,8 +600,10 @@ class SchedulerMetricsMixin:
dp_cooperation_info=batch.dp_cooperation_info,
)
if self.enable_mfu_metrics:
flops, read_bytes, write_bytes = self._estimate_decode_perf(
batch, decode_tokens
flops, read_bytes, write_bytes = (
SchedulerMetricsMixin._estimate_decode_perf(
self, batch, decode_tokens
)
)
self.metrics_collector.increment_estimated_perf(
num_flops_per_gpu=flops,
@@ -608,10 +615,10 @@ class SchedulerMetricsMixin:
self._mfu_log_write_bytes += write_bytes
if x := self.scheduler_status_logger:
x.maybe_dump(batch, self.waiting_queue)
x.maybe_dump(batch, self.scheduler.waiting_queue)
# Periodic work: log + heavy metrics at decode_log_interval
if self.forward_ct_decode % self.server_args.decode_log_interval != 0:
if self.forward_ct_decode % self.scheduler.server_args.decode_log_interval != 0:
return
if (
not self.is_stats_logging_rank
@@ -626,32 +633,34 @@ class SchedulerMetricsMixin:
self.num_generated_tokens = 0
num_running_reqs = len(batch.reqs)
pool_stats = self.pool_stats_observer.get_pool_stats()
pool_stats = self.scheduler.pool_stats_observer.get_pool_stats()
token_usage_msg = ", ".join(pool_stats.get_decode_usage_msg_parts()) + ", "
if RECORD_STEP_TIME:
self.step_time_dict[num_running_reqs].append(
gap_latency / self.server_args.decode_log_interval
gap_latency / self.scheduler.server_args.decode_log_interval
)
batch_iter = (
batch.forward_iter
if batch is not None and batch.forward_iter is not None
else self.forward_ct
else self.scheduler.forward_ct
)
iter_msg = f" [{batch_iter}]" if LOG_FORWARD_ITERS else ""
msg = f"Decode batch{iter_msg}, #running-req: {num_running_reqs}, {token_usage_msg}"
if self.spec_algorithm.is_none():
if self.scheduler.spec_algorithm.is_none():
spec_accept_length = 0
spec_accept_rate = 0
else:
spec_accept_length = self.spec_num_accept_tokens / self.spec_num_forward_ct
num_correct_drafts = self.spec_num_accept_tokens - self.spec_num_forward_ct
if self.server_args.speculative_num_draft_tokens:
draft_per_round = self.server_args.speculative_num_draft_tokens - 1
if self.scheduler.server_args.speculative_num_draft_tokens:
draft_per_round = (
self.scheduler.server_args.speculative_num_draft_tokens - 1
)
else:
draft_per_round = self.server_args.speculative_num_steps or 0
draft_per_round = self.scheduler.server_args.speculative_num_steps or 0
total_draft_tokens = self.spec_num_forward_ct * draft_per_round
spec_accept_rate = (
num_correct_drafts / total_draft_tokens if total_draft_tokens > 0 else 0
@@ -662,22 +671,25 @@ class SchedulerMetricsMixin:
msg += f"accept len: {spec_accept_length:.2f}, accept rate: {spec_accept_rate:.2f}, "
cache_hit_rate = 0.0
if self.disaggregation_mode == DisaggregationMode.DECODE:
msg += f"pre-allocated usage: {self.disagg_decode_prealloc_queue.num_tokens_pre_allocated / self.max_total_num_tokens:.2f}, "
msg += f"#prealloc-req: {len(self.disagg_decode_prealloc_queue.queue)}, "
msg += f"#transfer-req: {len(self.disagg_decode_transfer_queue.queue)}, "
msg += f"#retracted-req: {len(self.disagg_decode_prealloc_queue.retracted_queue)}, "
if self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
msg += f"pre-allocated usage: {self.scheduler.disagg_decode_prealloc_queue.num_tokens_pre_allocated / self.scheduler.max_total_num_tokens:.2f}, "
msg += f"#prealloc-req: {len(self.scheduler.disagg_decode_prealloc_queue.queue)}, "
msg += f"#transfer-req: {len(self.scheduler.disagg_decode_transfer_queue.queue)}, "
msg += f"#retracted-req: {len(self.scheduler.disagg_decode_prealloc_queue.retracted_queue)}, "
if (
self.server_args.language_only
and self.server_args.encoder_transfer_backend == "zmq_to_scheduler"
self.scheduler.server_args.language_only
and self.scheduler.server_args.encoder_transfer_backend
== "zmq_to_scheduler"
):
msg += f"waiting-image-req: {len(self.mm_receiver.waiting_list)}, "
msg += (
f"waiting-image-req: {len(self.scheduler.mm_receiver.waiting_list)}, "
)
msg += (
f"{self._graph_backend_label}: {can_run_cuda_graph}, "
f"gen throughput (token/s): {self.last_gen_throughput:.2f}, "
f"#queue-req: {len(self.waiting_queue)}"
f"#queue-req: {len(self.scheduler.waiting_queue)}"
)
if self.enable_mfu_metrics and gap_latency > 0:
@@ -702,16 +714,16 @@ class SchedulerMetricsMixin:
if self.is_stats_logging_rank:
logger.info(msg)
if self.current_scheduler_metrics_enabled:
priority_enabled = self.enable_priority_scheduling
priority_enabled = self.scheduler.enable_priority_scheduling
# Basics
self.stats.num_running_reqs = QueueCount.from_reqs(
batch.reqs, priority_enabled
)
self.stats.num_queue_reqs = QueueCount.from_reqs(
self.waiting_queue, priority_enabled
self.scheduler.waiting_queue, priority_enabled
)
self.stats.num_grammar_queue_reqs = len(self.grammar_manager)
self.stats.num_grammar_queue_reqs = len(self.scheduler.grammar_manager)
self.stats.gen_throughput = self.last_gen_throughput
self.stats.cache_hit_rate = cache_hit_rate
self.stats.decode_sum_seq_lens = batch.seq_lens_cpu.sum().item()
@@ -729,34 +741,37 @@ class SchedulerMetricsMixin:
self.num_retracted_reqs = self.num_paused_reqs = 0
# PD disaggregation
if self.disaggregation_mode == DisaggregationMode.PREFILL:
if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
self.stats.num_prefill_bootstrap_queue_reqs = QueueCount.from_reqs(
self.disagg_prefill_bootstrap_queue.queue, priority_enabled
self.scheduler.disagg_prefill_bootstrap_queue.queue,
priority_enabled,
)
self.stats.num_prefill_inflight_queue_reqs = QueueCount.from_reqs(
self.disagg_prefill_inflight_queue, priority_enabled
self.scheduler.disagg_prefill_inflight_queue, priority_enabled
)
elif self.disaggregation_mode == DisaggregationMode.DECODE:
elif self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
self.stats.num_decode_prealloc_queue_reqs = QueueCount.from_reqs(
self.disagg_decode_prealloc_queue.queue, priority_enabled
self.scheduler.disagg_decode_prealloc_queue.queue, priority_enabled
)
self.stats.num_decode_transfer_queue_reqs = QueueCount.from_reqs(
self.disagg_decode_transfer_queue.queue, priority_enabled
self.scheduler.disagg_decode_transfer_queue.queue, priority_enabled
)
# Streaming session metrics
self.stats.num_streaming_sessions = (
self.pool_stats_observer.streaming_session_count()
self.scheduler.pool_stats_observer.streaming_session_count()
)
self.stats.streaming_session_held_tokens = (
self.pool_stats_observer.session_held_tokens()
self.scheduler.pool_stats_observer.session_held_tokens()
)
# Routing key metrics
# (to reduce the overhead, we only compute this when all requests have routing_key)
if all(r.routing_key is not None for r in batch.reqs):
running_routing_keys = [r.routing_key for r in batch.reqs]
waiting_routing_keys = [r.routing_key for r in self.waiting_queue]
waiting_routing_keys = [
r.routing_key for r in self.scheduler.waiting_queue
]
(
self.stats.num_unique_running_routing_keys,
self.stats.routing_key_running_req_counts,
@@ -766,16 +781,17 @@ class SchedulerMetricsMixin:
)
# Utilization / LoRA / HiCache
self.calculate_utilization()
SchedulerMetricsMixin._calculate_utilization(self)
self.stats.fwd_occupancy = self.fwd_occupancy
self.update_lora_metrics()
self._log_hicache_stats()
SchedulerMetricsMixin._update_lora_metrics(self)
SchedulerMetricsMixin._log_hicache_stats(self)
self.metrics_collector.log_stats(self.stats)
self.kv_events_publisher.emit_kv_metrics()
self.kv_events_publisher.publish_kv_events()
self.scheduler.kv_events_publisher.emit_kv_metrics()
self.scheduler.kv_events_publisher.publish_kv_events()
@staticmethod
def log_batch_result_stats(
self: Scheduler,
self: "SchedulerMetricsReporter",
batch: ScheduleBatch,
result: Union[GenerationBatchResult, EmbeddingBatchResult],
):
@@ -790,8 +806,9 @@ class SchedulerMetricsMixin:
balancedness=m.eplb_balancedness.item(),
)
@staticmethod
def _emit_forward_pass_metrics(
self: Scheduler,
self: "SchedulerMetricsReporter",
batch: ScheduleBatch,
result=None,
):
@@ -801,61 +818,66 @@ class SchedulerMetricsMixin:
model_runner.forward / cuda_graph.replay via PR #24197).
Falls back to monotonic clock when DeviceTimer is not enabled.
"""
if not self.enable_fpm:
if not self.scheduler.enable_fpm:
return
from sglang.srt.observability.forward_pass_metrics import (
ForwardPassMetrics,
)
if self._fpm_uses_device_timer:
if self.scheduler._fpm_uses_device_timer:
self.forward_pass_device_timer._report()
wall_time = self._fpm_gpu_time_acc
self._fpm_gpu_time_acc = 0.0
wall_time = self.scheduler._fpm_gpu_time_acc
self.scheduler._fpm_gpu_time_acc = 0.0
if wall_time == 0.0:
return
else:
wall_time = max(0.0, time.monotonic() - batch.fpm_start_time)
fpm = ForwardPassMetrics(
worker_id=self._fpm_worker_id,
dp_rank=self._fpm_dp_rank,
worker_id=self.scheduler._fpm_worker_id,
dp_rank=self.scheduler._fpm_dp_rank,
wall_time=wall_time,
scheduled_requests=self._build_scheduled_request_metrics(batch),
queued_requests=self._build_queued_request_metrics(),
scheduled_requests=SchedulerMetricsMixin._build_scheduled_request_metrics(
self, batch
),
queued_requests=SchedulerMetricsMixin._build_queued_request_metrics(self),
)
self._fpm_publisher.publish(fpm)
self.scheduler._fpm_publisher.publish(fpm)
def _shutdown_fpm(self: Scheduler):
@staticmethod
def _shutdown_fpm(self: "SchedulerMetricsReporter"):
"""Shut down the FPM publisher thread."""
if self.enable_fpm:
self._fpm_publisher.shutdown()
if self.scheduler.enable_fpm:
self.scheduler._fpm_publisher.shutdown()
def _log_hicache_stats(self: Scheduler):
@staticmethod
def _log_hicache_stats(self: "SchedulerMetricsReporter"):
"""Populate HiCache host-tier stats on self.stats.
These are pushed to Prometheus by SchedulerMetricsCollector.log_stats().
"""
if not self.enable_hierarchical_cache:
if not self.scheduler.enable_hierarchical_cache:
return
host_pool = getattr(self.tree_cache, "token_to_kv_pool_host", None) or getattr(
self.tree_cache, "full_kv_pool_host", None
)
host_pool = getattr(
self.scheduler.tree_cache, "token_to_kv_pool_host", None
) or getattr(self.scheduler.tree_cache, "full_kv_pool_host", None)
assert host_pool is not None, "Host pool not found"
self.stats.hicache_host_used_tokens = (
host_pool.size - host_pool.available_size()
)
self.stats.hicache_host_total_tokens = host_pool.size
def update_lora_metrics(self: Scheduler):
@staticmethod
def _update_lora_metrics(self: "SchedulerMetricsReporter"):
"""Update LoRA pool metrics for monitoring and autoscaling."""
if not self.enable_lora:
if not self.scheduler.enable_lora:
return
try:
# Get LoRA memory pool stats
lora_manager = self.tp_worker.model_runner.lora_manager
lora_manager = self.scheduler.tp_worker.model_runner.lora_manager
if lora_manager is None or lora_manager.memory_pool is None:
return
@@ -867,16 +889,16 @@ class SchedulerMetricsMixin:
active_lora_ids = set()
# For PP mode, check all running micro batches
if self.server_args.pp_size > 1:
for batch in self.running_mbs:
if self.scheduler.server_args.pp_size > 1:
for batch in self.scheduler.running_mbs:
if batch and hasattr(batch, "reqs"):
for req in batch.reqs:
if hasattr(req, "lora_id") and req.lora_id is not None:
active_lora_ids.add(req.lora_id)
# For normal mode, check running_batch
elif self.running_batch:
if hasattr(self.running_batch, "reqs"):
for req in self.running_batch.reqs:
elif self.scheduler.running_batch:
if hasattr(self.scheduler.running_batch, "reqs"):
for req in self.scheduler.running_batch.reqs:
if hasattr(req, "lora_id") and req.lora_id is not None:
active_lora_ids.add(req.lora_id)
@@ -892,19 +914,23 @@ class SchedulerMetricsMixin:
except Exception as e:
logger.warning(f"Failed to update LoRA metrics: {e}")
def calculate_utilization(self: Scheduler):
if self.disaggregation_mode == DisaggregationMode.PREFILL:
@staticmethod
def _calculate_utilization(self: "SchedulerMetricsReporter"):
if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
self.stats.utilization = -1
else:
# TODO: max_running_requests_under_SLO has no setter — sglang:utilization stuck at 0 (regressed #22713).
max_under_slo = getattr(self, "max_running_requests_under_SLO", None)
max_under_slo = getattr(
self.scheduler, "max_running_requests_under_SLO", None
)
if max_under_slo is not None and max_under_slo > 0:
self.stats.utilization = max(
self.stats.num_running_reqs.total / max_under_slo,
self.stats.token_usage / 0.9,
)
def update_device_timer(self: Scheduler):
@staticmethod
def update_device_timer(self: "SchedulerMetricsReporter"):
if not ENABLE_METRICS_DEVICE_TIMER:
return
self.forward_pass_device_timer._report()
@@ -924,11 +950,12 @@ class SchedulerMetricsMixin:
self._device_timer_window_batch_count += 1
if (
self._device_timer_window_batch_count
>= self.server_args.decode_log_interval
>= self.scheduler.server_args.decode_log_interval
):
self._device_timer_window_batch_count = 0
def reset_device_timer_window(self: Scheduler):
@staticmethod
def reset_device_timer_window(self: "SchedulerMetricsReporter"):
if ENABLE_METRICS_DEVICE_TIMER:
self._device_timer_window_batch_count = 0
self.fwd_occupancy = float("nan")