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

Refactor device timer, clean up metrics collector, and add fwd occupancy metric (#24197)

Browse Source
This commit is contained in:
Lianmin Zheng
2026-05-01 10:25:25 -07:00
committed by GitHub
parent 4a50cd781e
commit ece8a1a788
16 changed files with 711 additions and 574 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/sglang/srt/environ.py
+1 -5
View File
@@ -169,7 +169,6 @@ class Envs:
SGLANG_LOG_GC = EnvBool(False)
SGLANG_LOG_FORWARD_ITERS = EnvBool(False)
SGLANG_LOG_MS = EnvBool(False)
SGLANG_DISABLE_REQUEST_LOGGING = EnvBool(False)
SGLANG_LOG_REQUEST_EXCEEDED_MS = EnvInt(-1)
SGLANG_LOG_REQUEST_HEADERS = EnvTuple(tuple())
SGLANG_LOG_SCHEDULER_STATUS_TARGET = EnvStr("")
@@ -187,7 +186,6 @@ class Envs:
SGLANG_GRAMMAR_MAX_POLL_ITERATIONS = EnvInt(10000)
SGLANG_DISABLE_OUTLINES_DISK_CACHE = EnvBool(False)
# Test & Debug
SGLANG_DETECT_SLOW_RANK = EnvBool(False)
SGLANG_TEST_STUCK_DETOKENIZER = EnvFloat(0)
@@ -425,9 +423,7 @@ class Envs:
# Flash Attention
SGLANG_USE_SGL_FA3_KERNEL = EnvBool(True)
# vLLM dependencies (TODO: they have been deprecated, we can remove them safely)
USE_VLLM_CUTLASS_W8A8_FP8_KERNEL = EnvBool(False)
# Kernels
USE_TRITON_W8A8_FP8_KERNEL = EnvBool(False)
SGLANG_RETURN_ORIGINAL_LOGPROB = EnvBool(False)
SGLANG_ALLOW_OVERWRITE_LONGER_CONTEXT_LEN = EnvBool(False)
python/sglang/srt/layers/quantization/fp8_utils.py
-1
View File
@@ -119,7 +119,6 @@ if _is_cuda:
return mat_a.new_empty((M, N), dtype=out_dtype)
use_vllm_cutlass_w8a8_fp8_kernel = get_bool_env_var("USE_VLLM_CUTLASS_W8A8_FP8_KERNEL")
use_triton_w8a8_fp8_kernel = get_bool_env_var("USE_TRITON_W8A8_FP8_KERNEL")
# Input scaling factors are no longer optional in _scaled_mm starting
python/sglang/srt/managers/scheduler.py
+36 -23
View File
@@ -699,6 +699,18 @@ class Scheduler(
else:
self.model_worker = self.draft_worker
# Install device timer on model runners for fwd occupancy tracking
if hasattr(self, "forward_pass_device_timer"):
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)
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
# Get token and memory info from the model worker
(
self.max_total_num_tokens,
@@ -744,10 +756,10 @@ class Scheduler(
set_random_seed(self.random_seed)
# Print debug info
avail_mem = get_available_gpu_memory(
self.device, self.gpu_id, empty_cache=False
)
if self.tp_rank == 0:
avail_mem = get_available_gpu_memory(
self.device, self.gpu_id, empty_cache=False
)
logger.info(
f"max_total_num_tokens={self.max_total_num_tokens}, "
f"chunked_prefill_size={self.server_args.chunked_prefill_size}, "
@@ -758,8 +770,17 @@ class Scheduler(
)
if self.enable_metrics and hasattr(self, "metrics_collector"):
self.metrics_collector.emit_cache_config_info(
self.page_size, self.max_total_num_tokens // self.page_size
self.metrics_collector.emit_constants(
max_total_num_tokens=self.max_total_num_tokens,
max_running_requests_under_SLO=getattr(
self, "max_running_requests_under_SLO", None
),
engine_startup_time=0.0,
engine_load_weights_time=0.0,
page_size=self.page_size,
num_pages=self.max_total_num_tokens // self.page_size,
context_len=self.model_config.context_len,
startup_available_gpu_memory_gb=avail_mem,
)
def init_cache_with_memory_pool(self):
@@ -1491,7 +1512,6 @@ class Scheduler(
recv_reqs = self.recv_requests()
self.process_input_requests(recv_reqs)
if self._engine_paused:
self.cancel_bubble_timer()
continue
# Get the next batch to run
@@ -1546,7 +1566,6 @@ class Scheduler(
self.result_queue.append((batch.copy(), batch_result))
else:
batch_result = None
self.cancel_bubble_timer()
# Process the last batch
if self.last_batch:
@@ -2922,14 +2941,13 @@ class Scheduler(
bs = len(model_worker_batch.seq_lens)
future_indices = self.future_map.alloc_future_indices(bs)
with self.forward_stream_ctx, self.record_bubble_metrics(batch):
with self.forward_stream_ctx:
self.forward_stream.wait_stream(self.schedule_stream)
self.future_map.resolve_future(model_worker_batch)
with self.record_forward_metrics(batch):
batch_result = self.model_worker.forward_batch_generation(
model_worker_batch
# here pp is not compatible with overlap
)
batch_result = self.model_worker.forward_batch_generation(
model_worker_batch
# here pp is not compatible with overlap
)
# FIXME(lsyin): maybe move this to forward_batch_generation
batch_result.copy_done = self.device_module.Event()
if batch_result.delay_sample_func is None:
@@ -2948,11 +2966,6 @@ class Scheduler(
batch.spec_info = batch_result.next_draft_input
batch.spec_info.future_indices = future_indices
# batch.spec_info = EagleDraftInput(
# future_indices=future_indices,
# verify_done=batch_result.next_draft_input.verify_done,
# )
# The future value, usually for next batch preparation
# Current implementation strictly synchronizes the seq_lens
batch.seq_lens = batch_result.next_draft_input.new_seq_lens
@@ -2965,10 +2978,9 @@ class Scheduler(
if self.spec_algorithm.is_none()
else {}
)
with self.record_forward_metrics(batch):
batch_result = self.model_worker.forward_batch_generation(
worker_batch_or_batch, **kwargs
)
batch_result = self.model_worker.forward_batch_generation(
worker_batch_or_batch, **kwargs
)
future_indices_or_next_token_ids = batch_result.next_token_ids
self.update_cache_from_scheduler(batch, batch_result)
@@ -2998,7 +3010,7 @@ class Scheduler(
if self.enable_overlap:
self.record_batch_in_overlap(model_worker_batch)
with self.forward_stream_ctx, self.record_bubble_metrics(batch):
with self.forward_stream_ctx:
self.forward_stream.wait_stream(self.schedule_stream)
pooler_output = self.tp_worker.forward_batch_embedding(
model_worker_batch
@@ -3083,6 +3095,7 @@ class Scheduler(
self.log_batch_result_stats(batch, result)
self._maybe_clear_mm_inputs(batch)
self.maybe_send_health_check_signal()
self.update_device_timer()
def maybe_send_health_check_signal(self):
if self.return_health_check_ipcs:
python/sglang/srt/managers/scheduler_runtime_checker_mixin.py
+4 -2
View File
@@ -10,7 +10,6 @@ from sglang.srt.disaggregation.utils import DisaggregationMode
from sglang.srt.environ import envs
from sglang.srt.observability.metrics_collector import QueueCount
from sglang.srt.utils.common import ceil_align, raise_error_or_warn
from sglang.srt.utils.request_logger import disable_request_logging
from sglang.srt.utils.watchdog import WatchdogRaw
if TYPE_CHECKING:
@@ -556,6 +555,9 @@ class SchedulerRuntimeCheckerMixin:
# reset token ratio
self.new_token_ratio = self.init_new_token_ratio
# reset device timer window so idle time isn't counted
self.reset_device_timer_window()
# sleep until next event
self.maybe_sleep_on_idle()
@@ -564,7 +566,7 @@ def create_scheduler_watchdog(
scheduler: Scheduler, watchdog_timeout: float, soft: bool = False
) -> WatchdogRaw:
def dump_info() -> str:
if scheduler.is_initializing or disable_request_logging():
if scheduler.is_initializing:
return ""
_, messages = scheduler._check_all_pools(scheduler.get_pool_stats())
return (
python/sglang/srt/managers/tokenizer_manager.py
-8
View File
@@ -2161,13 +2161,6 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
state.last_completion_tokens = completion_tokens
if state.finished:
retraction_count = (
recv_obj.retraction_counts[i]
if getattr(recv_obj, "retraction_counts", None)
and i < len(recv_obj.retraction_counts)
else 0
)
# Get detailed cache breakdown if available
cached_tokens_details = None
if (
@@ -2183,7 +2176,6 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
recv_obj.cached_tokens[i],
state.time_stats.get_e2e_latency(),
self._request_has_grammar(state.obj),
retraction_count,
cached_tokens_details,
)
python/sglang/srt/model_executor/cuda_graph_runner.py
+16 -4
View File
@@ -16,6 +16,7 @@
from __future__ import annotations
import bisect
import contextlib
import gc
import inspect
import logging
@@ -896,9 +897,10 @@ class CudaGraphRunner:
else:
set_pdmux_status(False)
for i, sg in enumerate(self.stream_groups):
with graph_capture(
stream=sg[1]
) as graph_capture_context, profile_context as prof:
with (
graph_capture(stream=sg[1]) as graph_capture_context,
profile_context as prof,
):
self.stream = graph_capture_context.stream
_capture_one_stream(i)
@@ -1313,7 +1315,17 @@ class CudaGraphRunner:
variant_label = self._resolve_lora_variant(forward_batch)
stream_idx = get_current_stream_idx() if self.enable_pdmux else None
graph_key = self._make_graph_key(self.bs, stream_idx, variant_label)
self.graphs[graph_key].replay()
ctx = (
self.model_runner.device_timer.wrap(
metadata={
"category": forward_batch.forward_mode.name.lower(),
}
)
if self.model_runner.device_timer
else contextlib.nullcontext()
)
with ctx:
self.graphs[graph_key].replay()
output = self.output_buffers[graph_key]
if isinstance(output, LogitsProcessorOutput):
python/sglang/srt/model_executor/model_runner.py
+109 -84
View File
@@ -348,6 +348,7 @@ class ModelRunner(ModelRunnerKVCacheMixin):
self.is_draft_worker = is_draft_worker
self.memory_pool_config = memory_pool_config
self.is_generation = model_config.is_generation
self.device_timer = None
self.is_multimodal = model_config.is_multimodal
self.is_multimodal_chunked_prefill_supported = (
model_config.is_multimodal_chunked_prefill_supported
@@ -362,6 +363,13 @@ class ModelRunner(ModelRunnerKVCacheMixin):
self.is_hybrid_swa_compress = model_config.is_hybrid_swa_compress
self.use_mla_backend = self.model_config.attention_arch == AttentionArch.MLA
self.attention_chunk_size = model_config.attention_chunk_size
rope_scaling = getattr(
model_config.hf_text_config, "rope_parameters", None
) or getattr(model_config.hf_text_config, "rope_scaling", {})
self.model_is_mrope = (
rope_scaling is not None and "mrope_section" in rope_scaling
)
self.enable_elastic_ep = server_args.elastic_ep_backend is not None
self.forward_pass_id = 0
self.init_new_workspace = False
self.draft_model_idx = draft_model_idx
@@ -2962,6 +2970,7 @@ class ModelRunner(ModelRunnerKVCacheMixin):
skip_attn_backend_init: bool = False,
pp_proxy_tensors=None,
) -> Union[LogitsProcessorOutput, PPProxyTensors]:
# Set extra arguments
if not skip_attn_backend_init:
if hasattr(self.model, "prepare_forward_batch"):
# Prepare model-specific attention metadata before planning,
@@ -2976,12 +2985,20 @@ class ModelRunner(ModelRunnerKVCacheMixin):
kwargs = {}
if self.support_pp:
kwargs["pp_proxy_tensors"] = pp_proxy_tensors
return self.model.forward(
forward_batch.input_ids,
forward_batch.positions,
forward_batch,
**kwargs,
# Launch forward
ctx = (
self.device_timer.wrap(metadata={"category": "decode"})
if self.device_timer
else contextlib.nullcontext()
)
with ctx:
return self.model.forward(
forward_batch.input_ids,
forward_batch.positions,
forward_batch,
**kwargs,
)
def forward_extend(
self,
@@ -2991,6 +3008,7 @@ class ModelRunner(ModelRunnerKVCacheMixin):
) -> Tuple[
Union[LogitsProcessorOutput, PPProxyTensors, EmbeddingPoolerOutput], bool
]:
# Setup extra arguments
kwargs = {}
if self.support_pp:
kwargs["pp_proxy_tensors"] = pp_proxy_tensors
@@ -3010,17 +3028,25 @@ class ModelRunner(ModelRunnerKVCacheMixin):
if not self.is_generation:
kwargs["get_embedding"] = True
# Check piecewies cuda graph
can_run_graph = (
self.piecewise_cuda_graph_runner is not None
and self.piecewise_cuda_graph_runner.can_run(forward_batch)
)
if can_run_graph:
return (
self.piecewise_cuda_graph_runner.replay(forward_batch, **kwargs),
can_run_graph,
# TODO: device_timer.wrap is too broad here — it also includes
# replay_prepare time. Move timing into the piecewise cuda graph
# runner to capture only the model.forward part.
ctx = (
self.device_timer.wrap(metadata={"category": "extend"})
if self.device_timer
else contextlib.nullcontext()
)
with ctx:
ret = self.piecewise_cuda_graph_runner.replay(forward_batch, **kwargs)
return (ret, can_run_graph)
# Launch model forward
if not skip_attn_backend_init:
if hasattr(self.model, "prepare_forward_batch"):
# Prepare model-specific attention metadata before planning,
@@ -3028,15 +3054,19 @@ class ModelRunner(ModelRunnerKVCacheMixin):
self.model.prepare_forward_batch(forward_batch)
self.attn_backend.init_forward_metadata(forward_batch)
return (
self.model.forward(
ctx = (
self.device_timer.wrap(metadata={"category": "extend"})
if self.device_timer
else contextlib.nullcontext()
)
with ctx:
ret = self.model.forward(
forward_batch.input_ids,
forward_batch.positions,
forward_batch,
**kwargs,
),
can_run_graph,
)
)
return (ret, can_run_graph)
def forward_idle(
self, forward_batch: ForwardBatch, pp_proxy_tensors=None
@@ -3050,12 +3080,18 @@ class ModelRunner(ModelRunnerKVCacheMixin):
kwargs = {}
if self.support_pp:
kwargs["pp_proxy_tensors"] = pp_proxy_tensors
return self.model.forward(
forward_batch.input_ids,
forward_batch.positions,
forward_batch,
**kwargs,
ctx = (
self.device_timer.wrap(metadata={"category": "idle"})
if self.device_timer
else contextlib.nullcontext()
)
with ctx:
return self.model.forward(
forward_batch.input_ids,
forward_batch.positions,
forward_batch,
**kwargs,
)
def forward_split_prefill(
self,
@@ -3069,12 +3105,18 @@ class ModelRunner(ModelRunnerKVCacheMixin):
forward_batch.split_index + forward_count,
self.model_config.num_hidden_layers,
)
ret = self.model.forward_split_prefill(
forward_batch.input_ids,
forward_batch.positions,
forward_batch,
(forward_batch.split_index, next_split_index),
ctx = (
self.device_timer.wrap(metadata={"category": "split_prefill"})
if self.device_timer
else contextlib.nullcontext()
)
with ctx:
ret = self.model.forward_split_prefill(
forward_batch.input_ids,
forward_batch.positions,
forward_batch,
(forward_batch.split_index, next_split_index),
)
forward_batch.split_index = next_split_index
return ret
@@ -3088,12 +3130,14 @@ class ModelRunner(ModelRunnerKVCacheMixin):
) -> ModelRunnerOutput:
self.forward_pass_id += 1
# Try msprob debugger
if self.msprobe_debugger is not None:
rank_id = (
self.gpu_id if self.dp_size is not None and self.dp_size > 1 else None
)
self.msprobe_debugger.start(model=self.model, rank_id=rank_id)
# Step span
step_span_ctx = (
torch.profiler.record_function(_build_step_span_name(forward_batch))
if torch.autograd._profiler_enabled()
@@ -3113,21 +3157,9 @@ class ModelRunner(ModelRunnerKVCacheMixin):
reinit_attn_backend,
split_forward_count,
)
elastic_ep_state = ElasticEPStateManager.instance()
if (
elastic_ep_state is not None
and not elastic_ep_state.is_active_equal_last()
):
elastic_ep_state.snapshot_active_to_last()
elastic_ep_state.sync_active_to_cpu()
logging.info("EPLB due to rank faults")
gen = self.eplb_manager.rebalance()
while True:
try:
next(gen)
except StopIteration:
break
output = self._forward_raw(
if self.enable_elastic_ep:
output = self._maybe_rebalance_after_rank_fault(
output,
forward_batch,
skip_attn_backend_init,
pp_proxy_tensors,
@@ -3167,6 +3199,7 @@ class ModelRunner(ModelRunnerKVCacheMixin):
reinit_attn_backend: bool = False,
split_forward_count: int = 1,
) -> ModelRunnerOutput:
# Check whether can run cuda graph
mode_check = (
forward_batch.forward_mode.is_cpu_graph
if self.device == "cpu"
@@ -3178,12 +3211,14 @@ class ModelRunner(ModelRunnerKVCacheMixin):
and self.graph_runner.can_run(forward_batch)
)
# Hisparse coordinator
if (
self.hisparse_coordinator is not None
and forward_batch.forward_mode.is_decode()
):
self.hisparse_coordinator.wait_for_pending_backup()
# Replay cuda graph if applicable
if can_run_graph:
ret = self.graph_runner.replay(
forward_batch,
@@ -3213,10 +3248,12 @@ class ModelRunner(ModelRunnerKVCacheMixin):
if forward_batch.out_cache_loc_swa is not None:
self.token_to_kv_pool.set_swa_loc(forward_batch.out_cache_loc_swa)
# Hisparse coordinator
forward_batch.hisparse_coordinator = self.hisparse_coordinator
if self.hisparse_coordinator is not None:
self.hisparse_coordinator.num_real_reqs.fill_(forward_batch.batch_size)
# Forward without cuda graph
if forward_batch.forward_mode.is_decode():
ret = self.forward_decode(
forward_batch,
@@ -3279,14 +3316,8 @@ class ModelRunner(ModelRunnerKVCacheMixin):
Returns:
A list of next_token_ids
"""
# For duplex models with multiple output streams.
if isinstance(logits_output, tuple):
return torch.stack(
[self.sample(values, forward_batch) for values in logits_output],
axis=-1,
)
self._preprocess_logits(logits_output, forward_batch.sampling_info)
# Sample the next tokens
next_token_ids = self.sampler(
logits_output,
@@ -3336,18 +3367,6 @@ class ModelRunner(ModelRunnerKVCacheMixin):
forward_batch.token_ids_logprobs,
)
@property
def model_is_mrope(self) -> bool:
"""Detect if the model has "mrope" rope_scaling type.
mrope requires keep "rope_deltas" between prompt and decoding phases."""
rope_scaling = getattr(
self.model_config.hf_text_config, "rope_parameters", None
) or getattr(self.model_config.hf_text_config, "rope_scaling", {})
if rope_scaling is None:
return False
is_mrope_enabled = "mrope_section" in rope_scaling
return is_mrope_enabled
def save_remote_model(self, url: str):
from sglang.srt.model_loader.loader import RemoteModelLoader
@@ -3405,6 +3424,35 @@ class ModelRunner(ModelRunnerKVCacheMixin):
device=self.device,
)
def _maybe_rebalance_after_rank_fault(
self,
output: ModelRunnerOutput,
forward_batch: ForwardBatch,
skip_attn_backend_init: bool,
pp_proxy_tensors: Optional[PPProxyTensors],
reinit_attn_backend: bool,
split_forward_count: int,
) -> ModelRunnerOutput:
elastic_ep_state = ElasticEPStateManager.instance()
if elastic_ep_state is not None and not elastic_ep_state.is_active_equal_last():
elastic_ep_state.snapshot_active_to_last()
elastic_ep_state.sync_active_to_cpu()
logging.info("EPLB due to rank faults")
gen = self.eplb_manager.rebalance()
while True:
try:
next(gen)
except StopIteration:
break
output = self._forward_raw(
forward_batch,
skip_attn_backend_init,
pp_proxy_tensors,
reinit_attn_backend,
split_forward_count,
)
return output
def _model_load_weights_direct(model, named_tensors: List[Tuple[str, torch.Tensor]]):
params_dict = dict(model.named_parameters())
@@ -3419,35 +3467,12 @@ def _unwrap_tensor(tensor, tp_rank, device):
def _build_step_span_name(forward_batch: ForwardBatch) -> str:
"""Build a profile-trace span name for one forward step.
Format:
step[decode bs=N] — decode-only batch
step[prefill bs=N toks=T] — extend-only (prefill) batch
step[mixed bs=N ext=T dec=D] — extend+decode mixed batch
step[idle] — idle/padding step
step[<MODE> bs=N] — other modes (target-verify, etc.)
Used by ModelRunner.forward to wrap each step in a torch.profile
record_function so Chrome traces show labeled step boundaries.
"""
"""Build a profile-trace span name for one forward step."""
mode = forward_batch.forward_mode
bs = forward_batch.batch_size
if mode.is_idle():
return "step[idle]"
if mode.is_decode():
return f"step[decode bs={bs}]"
if mode.is_extend():
if mode == ForwardMode.EXTEND:
ext_toks = forward_batch.extend_num_tokens or 0
ext_seqs = (
forward_batch.extend_seq_lens.shape[0]
if forward_batch.extend_seq_lens is not None
else bs
)
dec_seqs = bs - ext_seqs
if dec_seqs > 0:
return f"step[mixed bs={bs} ext={ext_toks} dec={dec_seqs}]"
return f"step[prefill bs={bs} toks={ext_toks}]"
return f"step[EXTEND bs={bs} toks={ext_toks}]"
return f"step[{mode.name} bs={bs}]"
python/sglang/srt/observability/metrics_collector.py
+349 -314
View File
@@ -40,20 +40,6 @@ SGLANG_TEST_REQUEST_TIME_STATS = get_bool_env_var("SGLANG_TEST_REQUEST_TIME_STAT
logger = logging.getLogger(__name__)
def get_histogram_conf_from_env(env_var_name: str) -> Optional[List[float]]:
"""
Get the histogram configuration from the environment variable.
env value should be like "0.1,0.2,0.5,1,2"
"""
if env_var_name not in os.environ:
return None
# if the env var is not set or empty, return None
env_var_value = os.environ[env_var_name]
if not env_var_value:
return None
return [float(x) for x in env_var_value.split(",")]
@dataclass
class QueueCount:
"""Holds both the total count and optional per-priority breakdown for a queue."""
@@ -78,22 +64,36 @@ class QueueCount:
class SchedulerStats:
# Basics
num_running_reqs: QueueCount = field(default_factory=QueueCount)
num_used_tokens: int = 0
# FIXME: token_usage is actually max usage across all pools (KV, SWA, mamba),
# not just KV token usage. Rename requires API deprecation.
token_usage: float = 0.0
full_token_usage: float = 0.0
pending_prealloc_token_usage: float = 0.0
swa_token_usage: float = 0.0
mamba_usage: float = 0.0
decode_sum_seq_lens: int = 0
gen_throughput: float = 0.0
num_queue_reqs: QueueCount = field(default_factory=QueueCount)
num_grammar_queue_reqs: int = 0
num_running_reqs_offline_batch: int = 0
gen_throughput: float = 0.0
cache_hit_rate: float = 0.0
decode_sum_seq_lens: int = 0
max_total_num_tokens: int = 0
# Memory pool usage ratios (0.0–1.0).
# Each pool tracks: used = total - available - evictable, usage = used / total.
#
# token_usage: max(full, swa, mamba) — the bottleneck across all pools.
# FIXME: misleadingly named "token_usage"; rename requires API deprecation.
# full_token_usage: full-attention KV cache pool usage (always active).
# swa_token_usage: sliding-window attention KV cache pool usage (hybrid SWA models only, e.g. Gemma2).
# mamba_usage: Mamba SSM state pool usage (hybrid SSM models only, e.g. Jamba).
token_usage: float = 0.0
full_token_usage: float = 0.0
swa_token_usage: float = 0.0
mamba_usage: float = 0.0
# Absolute token counts for the full-attention KV cache pool.
# Invariant: kv_available_tokens + kv_evictable_tokens + kv_used_tokens <= max_total_num_tokens
# (the gap accounts for protected/session-held tokens not exposed here).
# max_total_num_tokens is emitted once at startup via emit_constants.
#
# kv_available_tokens: free (unallocated) slots in the pool.
# kv_evictable_tokens: slots holding radix-cached KV data that can be evicted for new requests.
# kv_used_tokens: actively used slots (locked by running requests). Equals full_num_used.
# num_used_tokens: max(full_num_used, swa_num_used) for hybrid-SWA models, else full_num_used.
# Does NOT include the mamba pool.
num_used_tokens: int = 0
kv_available_tokens: int = 0
kv_evictable_tokens: int = 0
kv_used_tokens: int = 0
@@ -113,18 +113,16 @@ class SchedulerStats:
num_decode_transfer_queue_reqs: QueueCount = field(default_factory=QueueCount)
kv_transfer_speed_gb_s: float = 0.0
kv_transfer_latency_ms: float = 0.0
pending_prealloc_token_usage: float = 0.0
# Utilization
utilization: float = 0.0
max_running_requests_under_SLO: Optional[int] = None
# Engine startup
engine_startup_time: float = 0.0
engine_load_weights_time: float = 0.0
# Scheduler policy
new_token_ratio: float = 0.0
# CUDA graph
is_cuda_graph: float = 0.0
is_cuda_graph: int = 0
# LoRA pool metrics
lora_pool_slots_used: int = 0
@@ -196,60 +194,15 @@ class SchedulerMetricsCollector:
self.last_log_time = time.perf_counter()
self._known_priorities: Set[int] = set()
# =================================================================
# Basics
# =================================================================
self.num_running_reqs = Gauge(
name="sglang:num_running_reqs",
documentation="The number of running requests.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.num_used_tokens = Gauge(
name="sglang:num_used_tokens",
documentation="The number of used tokens.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.token_usage = Gauge(
name="sglang:token_usage",
documentation="The token usage.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.full_token_usage = Gauge(
name="sglang:full_token_usage",
documentation="The token usage for full attention layers.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.pending_prealloc_token_usage = Gauge(
name="sglang:pending_prealloc_token_usage",
documentation="The token usage for pending preallocated tokens (not preallocated yet).",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.swa_token_usage = Gauge(
name="sglang:swa_token_usage",
documentation="The token usage for SWA layers.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.mamba_usage = Gauge(
name="sglang:mamba_usage",
documentation="The token usage for Mamba layers.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.decode_sum_seq_lens = Gauge(
name="sglang:decode_sum_seq_lens",
documentation="The sum of all sequence lengths in decode.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.gen_throughput = Gauge(
name="sglang:gen_throughput",
documentation="The generation throughput (token/s).",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.num_queue_reqs = Gauge(
name="sglang:num_queue_reqs",
documentation="The number of requests in the waiting queue.",
@@ -262,9 +215,9 @@ class SchedulerMetricsCollector:
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.num_running_reqs_offline_batch = Gauge(
name="sglang:num_running_reqs_offline_batch",
documentation="The number of running low-priority offline batch requests(label is 'batch').",
self.gen_throughput = Gauge(
name="sglang:gen_throughput",
documentation="The generation throughput (token/s).",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
@@ -274,14 +227,50 @@ class SchedulerMetricsCollector:
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.max_total_num_tokens = Gauge(
name="sglang:max_total_num_tokens",
documentation="Maximum total number of tokens in the KV cache pool.",
self.decode_sum_seq_lens = Gauge(
name="sglang:decode_sum_seq_lens",
documentation="The sum of all sequence lengths in decode.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
# =================================================================
# Memory pool usage ratios
# =================================================================
self.token_usage = Gauge(
name="sglang:token_usage",
documentation="The token usage.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.full_token_usage = Gauge(
name="sglang:full_token_usage",
documentation="The token usage for full attention layers.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.swa_token_usage = Gauge(
name="sglang:swa_token_usage",
documentation="The token usage for SWA layers.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.mamba_usage = Gauge(
name="sglang:mamba_usage",
documentation="The token usage for Mamba layers.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
# =================================================================
# Absolute token counts
# =================================================================
self.num_used_tokens = Gauge(
name="sglang:num_used_tokens",
documentation="The number of used tokens.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.kv_available_tokens = Gauge(
name="sglang:kv_available_tokens",
documentation="Number of free token slots in the KV cache pool.",
@@ -301,7 +290,9 @@ class SchedulerMetricsCollector:
multiprocess_mode="mostrecent",
)
# =================================================================
# Speculative decoding
# =================================================================
self.spec_accept_length = Gauge(
name="sglang:spec_accept_length",
documentation="Mean acceptance length of speculative decoding (accepted drafts + bonus token per forward).",
@@ -315,7 +306,9 @@ class SchedulerMetricsCollector:
multiprocess_mode="mostrecent",
)
# =================================================================
# Retract
# =================================================================
# TODO maybe remove this old gauge in favor of the new counter
self.num_retracted_reqs = Gauge(
name="sglang:num_retracted_reqs",
@@ -344,7 +337,9 @@ class SchedulerMetricsCollector:
labelnames=labels.keys(),
)
# =================================================================
# PD disaggregation
# =================================================================
self.num_prefill_prealloc_queue_reqs = Gauge(
name="sglang:num_prefill_prealloc_queue_reqs",
documentation="The number of requests in the prefill prealloc queue.",
@@ -369,6 +364,24 @@ class SchedulerMetricsCollector:
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.kv_transfer_speed_gb_s = Histogram(
name="sglang:kv_transfer_speed_gb_s",
documentation="Histogram of KV cache transfer speed in GB/s.",
labelnames=labels.keys(),
buckets=(0.1, 0.5, 1, 5, 10, 25, 50, 100, 200, 400),
)
self.kv_transfer_latency_ms = Histogram(
name="sglang:kv_transfer_latency_ms",
documentation="Histogram of KV cache transfer latency in ms.",
labelnames=labels.keys(),
buckets=(1, 2, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000),
)
self.pending_prealloc_token_usage = Gauge(
name="sglang:pending_prealloc_token_usage",
documentation="The token usage for pending preallocated tokens (not preallocated yet).",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.num_bootstrap_failed_reqs = Counter(
name="sglang:num_bootstrap_failed_reqs_total",
documentation="The number of bootstrap failed requests.",
@@ -384,18 +397,6 @@ class SchedulerMetricsCollector:
documentation="Total number of prefill retries.",
labelnames=labels.keys(),
)
self.kv_transfer_speed_gb_s = Histogram(
name="sglang:kv_transfer_speed_gb_s",
documentation="Histogram of KV cache transfer speed in GB/s.",
labelnames=labels.keys(),
buckets=(0.1, 0.5, 1, 5, 10, 25, 50, 100, 200, 400),
)
self.kv_transfer_latency_ms = Histogram(
name="sglang:kv_transfer_latency_ms",
documentation="Histogram of KV cache transfer latency in ms.",
labelnames=labels.keys(),
buckets=(1, 2, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000),
)
self.kv_transfer_bootstrap_ms = Histogram(
name="sglang:kv_transfer_bootstrap_ms",
documentation="Histogram of KV transfer bootstrap time in ms.",
@@ -415,35 +416,124 @@ class SchedulerMetricsCollector:
buckets=(1, 5, 10, 50, 100, 500, 1000, 5000, 10000),
)
# =================================================================
# Utilization
# =================================================================
self.utilization = Gauge(
name="sglang:utilization",
documentation="The utilization.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.max_running_requests_under_SLO = Gauge(
name="sglang:max_running_requests_under_SLO",
documentation="The maximum number of running requests under SLO.",
# =================================================================
# Scheduler policy
# =================================================================
self.new_token_ratio = Gauge(
name="sglang:new_token_ratio",
documentation="The new token ratio.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
# Engine startup
self.engine_startup_time = Gauge(
name="sglang:engine_startup_time",
documentation="The time taken for the engine to start up.",
# =================================================================
# CUDA graph
# =================================================================
# TODO maybe remove this old gauge in favor of the new counter
self.is_cuda_graph = Gauge(
name="sglang:is_cuda_graph",
documentation="Whether the batch is using CUDA graph.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.engine_load_weights_time = Gauge(
name="sglang:engine_load_weights_time",
documentation="The time taken for the engine to load weights.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
self.cuda_graph_passes_total = Counter(
name="sglang:cuda_graph_passes_total",
documentation="Total number of forward passes categorized by CUDA graph.",
labelnames=list(labels.keys()) + ["mode"],
)
# Additional queueing time histogram
# =================================================================
# LoRA pool metrics (only created when LoRA is enabled)
# =================================================================
if self.enable_lora:
self.lora_pool_slots_used = Gauge(
name="sglang:lora_pool_slots_used",
documentation="Number of LoRA adapter slots currently occupied in GPU memory.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.lora_pool_slots_total = Gauge(
name="sglang:lora_pool_slots_total",
documentation="Total number of LoRA adapter slots available (max_loras_per_batch).",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.lora_pool_utilization = Gauge(
name="sglang:lora_pool_utilization",
documentation="LoRA pool utilization ratio (used/total). 1.0 means pool is full.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
# =================================================================
# HiCache metrics (only created when hierarchical cache is enabled)
# =================================================================
if self.enable_hierarchical_cache:
self.hicache_host_used_tokens = Gauge(
name="sglang:hicache_host_used_tokens",
documentation="Number of tokens currently used in the host KV cache.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.hicache_host_total_tokens = Gauge(
name="sglang:hicache_host_total_tokens",
documentation="Total capacity of the host KV cache in tokens.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
# =================================================================
# Streaming session metrics (only created when streaming sessions are enabled)
# =================================================================
if self.enable_streaming_session:
self.num_streaming_sessions = Gauge(
name="sglang:num_streaming_sessions",
documentation="The number of streaming sessions.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.streaming_session_held_tokens = Gauge(
name="sglang:streaming_session_held_tokens",
documentation="The number of KV tokens currently held by streaming session slots.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
# =================================================================
# Routing key metrics
# =================================================================
self.num_unique_running_routing_keys = Gauge(
name="sglang:num_unique_running_routing_keys",
documentation="Number of unique routing keys in running batch.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.routing_key_running_req_count = GaugeHistogram(
name="sglang:routing_key_running_req_count",
documentation="Distribution of routing keys by running request count (gt < count <= le).",
labelnames=list(labels.keys()),
bucket_bounds=ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS,
)
self.routing_key_all_req_count = GaugeHistogram(
name="sglang:routing_key_all_req_count",
documentation="Distribution of routing keys by running+waiting request count (gt < count <= le).",
labelnames=list(labels.keys()),
bucket_bounds=ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS,
)
# =================================================================
# Request latency
# =================================================================
self.queue_time = Histogram(
name="sglang:queue_time_seconds",
documentation="Histogram of queueing time in seconds.",
@@ -487,8 +577,17 @@ class SchedulerMetricsCollector:
3000,
],
)
self.per_stage_req_latency_seconds = Histogram(
name="sglang:per_stage_req_latency_seconds",
documentation="The latency of each stage of requests.",
# captures latency in range [1ms - ~1191s]
buckets=exponential_buckets(start=0.001, width=1.62, length=30),
labelnames=list(labels.keys()) + ["stage"],
)
# Grammar metrics
# =================================================================
# Grammar
# =================================================================
self.grammar_compilation_time = Histogram(
name="sglang:grammar_compilation_time_seconds",
documentation="Histogram of grammar compilation time in seconds.",
@@ -616,27 +715,9 @@ class SchedulerMetricsCollector:
buckets=tree_traversal_time_buckets,
)
self.per_stage_req_latency_seconds = Histogram(
name="sglang:per_stage_req_latency_seconds",
documentation="The latency of each stage of requests.",
# captures latency in range [1ms - ~1191s]
buckets=exponential_buckets(start=0.001, width=1.62, length=30),
labelnames=list(labels.keys()) + ["stage"],
)
# TODO maybe remove this old gauge in favor of the new counter
self.is_cuda_graph = Gauge(
name="sglang:is_cuda_graph",
documentation="Whether the batch is using CUDA graph.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.cuda_graph_passes_total = Counter(
name="sglang:cuda_graph_passes_total",
documentation="Total number of forward passes categorized by CUDA graph.",
labelnames=list(labels.keys()) + ["mode"],
)
# =================================================================
# Execution
# =================================================================
if (
labels["moe_ep_rank"] == 0
) and envs.SGLANG_ENABLE_EPLB_BALANCEDNESS_METRIC.get():
@@ -646,83 +727,6 @@ class SchedulerMetricsCollector:
labelnames=list(labels.keys()) + ["forward_mode"],
)
# LoRA pool metrics (only created when LoRA is enabled)
if self.enable_lora:
self.lora_pool_slots_used = Gauge(
name="sglang:lora_pool_slots_used",
documentation="Number of LoRA adapter slots currently occupied in GPU memory.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.lora_pool_slots_total = Gauge(
name="sglang:lora_pool_slots_total",
documentation="Total number of LoRA adapter slots available (max_loras_per_batch).",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.lora_pool_utilization = Gauge(
name="sglang:lora_pool_utilization",
documentation="LoRA pool utilization ratio (used/total). 1.0 means pool is full.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
# HiCache host-tier metrics (only created when hierarchical cache is enabled)
if self.enable_hierarchical_cache:
self.hicache_host_used_tokens = Gauge(
name="sglang:hicache_host_used_tokens",
documentation="Number of tokens currently used in the host KV cache.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.hicache_host_total_tokens = Gauge(
name="sglang:hicache_host_total_tokens",
documentation="Total capacity of the host KV cache in tokens.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
# Streaming session metrics (only created when streaming sessions are enabled)
if self.enable_streaming_session:
self.num_streaming_sessions = Gauge(
name="sglang:num_streaming_sessions",
documentation="The number of streaming sessions.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.streaming_session_held_tokens = Gauge(
name="sglang:streaming_session_held_tokens",
documentation="The number of KV tokens currently held by streaming session slots.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.num_unique_running_routing_keys = Gauge(
name="sglang:num_unique_running_routing_keys",
documentation="Number of unique routing keys in running batch.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.routing_key_running_req_count = GaugeHistogram(
name="sglang:routing_key_running_req_count",
documentation="Distribution of routing keys by running request count (gt < count <= le).",
labelnames=list(labels.keys()),
bucket_bounds=ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS,
)
self.routing_key_all_req_count = GaugeHistogram(
name="sglang:routing_key_all_req_count",
documentation="Distribution of routing keys by running+waiting request count (gt < count <= le).",
labelnames=list(labels.keys()),
bucket_bounds=ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS,
)
self.new_token_ratio = Gauge(
name="sglang:new_token_ratio",
documentation="The new token ratio.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.realtime_tokens_total = Counter(
name="sglang:realtime_tokens_total",
documentation=(
@@ -731,22 +735,14 @@ class SchedulerMetricsCollector:
),
labelnames=list(labels.keys()) + ["mode"],
)
self.gpu_execution_seconds_total = Counter(
name="sglang:gpu_execution_seconds_total",
self.forward_execution_seconds_total = Counter(
name="sglang:forward_execution_seconds_total",
documentation=(
"Total time that GPU is busy executing a workload. "
"Total time that GPU is busy executing model forward passes. "
"Refer to ForwardMode for category labels."
),
labelnames=list(labels.keys()) + ["category"],
)
self.gpu_overlap_wait_seconds_total = Counter(
name="sglang:gpu_overlap_wait_seconds_total",
documentation=(
"Total time that GPU forward stream was idle waiting for "
"the CPU schedule stream (overlap bubble)."
),
labelnames=list(labels.keys()) + ["category"],
)
self.estimated_flops_per_gpu_total = Counter(
name="sglang:estimated_flops_per_gpu_total",
documentation=(
@@ -780,15 +776,19 @@ class SchedulerMetricsCollector:
),
labelnames=list(labels.keys()) + ["mode", "num_prefill_ranks"],
)
self.dp_cooperation_gpu_execution_seconds_total = Counter(
name="sglang:dp_cooperation_gpu_execution_seconds_total",
self.dp_cooperation_forward_execution_seconds_total = Counter(
name="sglang:dp_cooperation_forward_execution_seconds_total",
documentation=(
"Total time that GPU is busy executing a workload with labels about DP cooperation. "
"Total time that GPU is busy executing model forward passes, "
"with labels about DP cooperation. "
"Refer to ForwardMode for category labels."
),
labelnames=list(labels.keys()) + ["category", "num_prefill_ranks"],
)
# =================================================================
# Prefill delayer
# =================================================================
max_delay = server_args.prefill_delayer_max_delay_passes
self.prefill_delayer_wait_forward_passes = Histogram(
name="sglang:prefill_delayer_wait_forward_passes",
@@ -832,13 +832,55 @@ class SchedulerMetricsCollector:
],
)
# This is a work-around Info metric since Info metrics are not supported in Prometheus.
# Similar to vLLM, https://github.com/vllm-project/vllm/blob/main/vllm/v1/metrics/loggers.py
# If more Info metrics are needed, we can create a common _log_info function.
self.cache_config_info = Gauge(
name="sglang:cache_config_info",
documentation="Cache configuration information.",
labelnames=["page_size", "num_pages"],
# =================================================================
# Constants (set once at startup via emit_constants)
# =================================================================
self.max_total_num_tokens = Gauge(
name="sglang:max_total_num_tokens",
documentation="Maximum total number of tokens in the KV cache pool.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.max_running_requests_under_SLO = Gauge(
name="sglang:max_running_requests_under_SLO",
documentation="The maximum number of running requests under SLO.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.engine_startup_time = Gauge(
name="sglang:engine_startup_time",
documentation="The time taken for the engine to start up.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.engine_load_weights_time = Gauge(
name="sglang:engine_load_weights_time",
documentation="The time taken for the engine to load weights.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.page_size = Gauge(
name="sglang:page_size",
documentation="KV cache page size in tokens.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.num_pages = Gauge(
name="sglang:num_pages",
documentation="Number of KV cache pages.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.context_len = Gauge(
name="sglang:context_len",
documentation="Maximum context length.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
self.startup_available_gpu_memory_gb = Gauge(
name="sglang:startup_available_gpu_memory_gb",
documentation="Available GPU memory in GB at startup.",
labelnames=labels.keys(),
multiprocess_mode="mostrecent",
)
@@ -972,26 +1014,17 @@ class SchedulerMetricsCollector:
**dp_cooperation_info.to_labels(),
).inc(delta)
def increment_gpu_overlap_wait_seconds(
def increment_forward_execution_seconds(
self,
category: str,
t: float,
dp_cooperation_info: Optional[DPCooperationInfo],
dp_cooperation_info: Optional[DPCooperationInfo] = None,
):
self.gpu_overlap_wait_seconds_total.labels(
self.forward_execution_seconds_total.labels(
**self.labels, category=category
).inc(t)
def increment_gpu_execution_seconds(
self,
category: str,
t: float,
dp_cooperation_info: Optional[DPCooperationInfo],
):
logger.debug(f"GPU execution seconds: {category=} {t=:.3f}")
self.gpu_execution_seconds_total.labels(**self.labels, category=category).inc(t)
if dp_cooperation_info is not None:
self.dp_cooperation_gpu_execution_seconds_total.labels(
self.dp_cooperation_forward_execution_seconds_total.labels(
**self.labels,
category=category,
**dp_cooperation_info.to_labels(),
@@ -1017,25 +1050,22 @@ class SchedulerMetricsCollector:
)
def log_stats(self, stats: SchedulerStats) -> None:
# Basics
self._log_gauge_queue_count(self.num_running_reqs, stats.num_running_reqs)
self._log_gauge(self.num_used_tokens, stats.num_used_tokens)
self._log_gauge(self.token_usage, stats.token_usage)
self._log_gauge(self.full_token_usage, stats.full_token_usage)
self._log_gauge(
self.pending_prealloc_token_usage, stats.pending_prealloc_token_usage
)
self._log_gauge(self.swa_token_usage, stats.swa_token_usage)
self._log_gauge(self.mamba_usage, stats.mamba_usage)
self._log_gauge(self.decode_sum_seq_lens, stats.decode_sum_seq_lens)
self._log_gauge(self.gen_throughput, stats.gen_throughput)
self._log_gauge_queue_count(self.num_queue_reqs, stats.num_queue_reqs)
self._log_gauge(self.num_grammar_queue_reqs, stats.num_grammar_queue_reqs)
self._log_gauge(
self.num_running_reqs_offline_batch, stats.num_running_reqs_offline_batch
)
self._log_gauge(self.gen_throughput, stats.gen_throughput)
self._log_gauge(self.cache_hit_rate, stats.cache_hit_rate)
self._log_gauge(self.decode_sum_seq_lens, stats.decode_sum_seq_lens)
self._log_gauge(self.max_total_num_tokens, stats.max_total_num_tokens)
# Memory pool usage ratios
self._log_gauge(self.token_usage, stats.token_usage)
self._log_gauge(self.full_token_usage, stats.full_token_usage)
self._log_gauge(self.swa_token_usage, stats.swa_token_usage)
self._log_gauge(self.mamba_usage, stats.mamba_usage)
# Absolute token counts
self._log_gauge(self.num_used_tokens, stats.num_used_tokens)
self._log_gauge(self.kv_available_tokens, stats.kv_available_tokens)
self._log_gauge(self.kv_evictable_tokens, stats.kv_evictable_tokens)
self._log_gauge(self.kv_used_tokens, stats.kv_used_tokens)
@@ -1044,6 +1074,10 @@ class SchedulerMetricsCollector:
self._log_gauge(self.spec_accept_length, stats.spec_accept_length)
self._log_gauge(self.spec_accept_rate, stats.spec_accept_rate)
# Retract
self._log_gauge(self.num_retracted_reqs, stats.num_retracted_reqs)
self._log_gauge(self.num_paused_reqs, stats.num_paused_reqs)
# PD disaggregation
self._log_gauge_queue_count(
self.num_prefill_prealloc_queue_reqs, stats.num_prefill_prealloc_queue_reqs
@@ -1057,36 +1091,26 @@ class SchedulerMetricsCollector:
self._log_gauge_queue_count(
self.num_decode_transfer_queue_reqs, stats.num_decode_transfer_queue_reqs
)
# Retract
self._log_gauge(self.num_retracted_reqs, stats.num_retracted_reqs)
self._log_gauge(self.num_paused_reqs, stats.num_paused_reqs)
self._log_gauge(
self.pending_prealloc_token_usage, stats.pending_prealloc_token_usage
)
# Utilization
self._log_gauge(self.utilization, stats.utilization)
if stats.max_running_requests_under_SLO is not None:
self._log_gauge(
self.max_running_requests_under_SLO,
stats.max_running_requests_under_SLO,
)
# Engine startup time
self._log_gauge(self.engine_startup_time, stats.engine_startup_time)
if stats.engine_load_weights_time is not None:
self._log_gauge(
self.engine_load_weights_time, stats.engine_load_weights_time
)
# Scheduler policy
self._log_gauge(self.new_token_ratio, stats.new_token_ratio)
# CUDA graph
self._log_gauge(self.is_cuda_graph, stats.is_cuda_graph)
# LoRA pool metrics (only logged if LoRA is enabled)
# LoRA pool metrics
if self.enable_lora:
self._log_gauge(self.lora_pool_slots_used, stats.lora_pool_slots_used)
self._log_gauge(self.lora_pool_slots_total, stats.lora_pool_slots_total)
self._log_gauge(self.lora_pool_utilization, stats.lora_pool_utilization)
# HiCache host-tier metrics (only logged if hierarchical cache is enabled)
# HiCache metrics
if self.enable_hierarchical_cache:
self._log_gauge(
self.hicache_host_used_tokens, stats.hicache_host_used_tokens
@@ -1095,13 +1119,14 @@ class SchedulerMetricsCollector:
self.hicache_host_total_tokens, stats.hicache_host_total_tokens
)
# Streaming session metrics (only logged if streaming sessions are enabled)
# Streaming session metrics
if self.enable_streaming_session:
self._log_gauge(self.num_streaming_sessions, stats.num_streaming_sessions)
self._log_gauge(
self.streaming_session_held_tokens, stats.streaming_session_held_tokens
)
# Routing key metrics
self._log_gauge(
self.num_unique_running_routing_keys, stats.num_unique_running_routing_keys
)
@@ -1139,8 +1164,30 @@ class SchedulerMetricsCollector:
)
self.num_grammar_total.labels(**self.labels).inc(1)
def emit_cache_config_info(self, page_size: int, num_pages: int) -> None:
self.cache_config_info.labels(page_size=page_size, num_pages=num_pages).set(1)
def emit_constants(
self,
max_total_num_tokens: int,
max_running_requests_under_SLO: Optional[int],
engine_startup_time: float,
engine_load_weights_time: float,
page_size: int,
num_pages: int,
context_len: int,
startup_available_gpu_memory_gb: float,
) -> None:
self._log_gauge(self.max_total_num_tokens, max_total_num_tokens)
if max_running_requests_under_SLO is not None:
self._log_gauge(
self.max_running_requests_under_SLO, max_running_requests_under_SLO
)
self._log_gauge(self.engine_startup_time, engine_startup_time)
self._log_gauge(self.engine_load_weights_time, engine_load_weights_time)
self._log_gauge(self.page_size, page_size)
self._log_gauge(self.num_pages, num_pages)
self._log_gauge(self.context_len, context_len)
self._log_gauge(
self.startup_available_gpu_memory_gb, startup_available_gpu_memory_gb
)
class TokenizerMetricsCollector:
@@ -1162,7 +1209,6 @@ class TokenizerMetricsCollector:
documentation="Number of prefill tokens processed.",
labelnames=labels.keys(),
)
self.generation_tokens_total = Counter(
name="sglang:generation_tokens_total",
documentation="Number of generation tokens processed.",
@@ -1185,20 +1231,25 @@ class TokenizerMetricsCollector:
8000,
9000,
10000,
12000,
12500,
15000,
17500,
20000,
22000,
22500,
25000,
27500,
30000,
35000,
40000,
66000,
99000,
132000,
60000,
80000,
100000,
200000,
300000,
400000,
600000,
900000,
800000,
1000000,
1100000,
]
self.prompt_tokens_histogram = Histogram(
@@ -1339,34 +1390,6 @@ class TokenizerMetricsCollector:
buckets=bucket_e2e_request_latency,
)
# Retraction count histogram
self.num_retractions = Histogram(
name="sglang:num_retractions",
documentation="Histogram of retraction counts per request.",
labelnames=labels.keys(),
buckets=[
0,
1,
2,
3,
4,
5,
6,
7,
8,
9,
10,
15,
20,
25,
30,
40,
50,
75,
100,
],
)
def observe_one_finished_request(
self,
labels: Dict[str, str],
@@ -1375,7 +1398,6 @@ class TokenizerMetricsCollector:
cached_tokens: int,
e2e_latency: float,
has_grammar: bool,
retraction_count: int,
cached_tokens_details: Optional[Dict[str, Any]] = None,
):
self.prompt_tokens_total.labels(**labels).inc(prompt_tokens)
@@ -1414,7 +1436,6 @@ class TokenizerMetricsCollector:
self.generation_tokens_histogram.labels(**labels).observe(
float(generation_tokens)
)
self.num_retractions.labels(**labels).observe(retraction_count)
def observe_time_to_first_token(self, labels: Dict[str, str], value: float):
self.histogram_time_to_first_token.labels(**labels).observe(value)
@@ -1661,3 +1682,17 @@ class RadixCacheMetricsCollector:
def observe_load_back_duration(self, duration_seconds: float) -> None:
self.load_back_duration_seconds.labels(**self.labels).observe(duration_seconds)
def get_histogram_conf_from_env(env_var_name: str) -> Optional[List[float]]:
"""
Get the histogram configuration from the environment variable.
env value should be like "0.1,0.2,0.5,1,2"
"""
if env_var_name not in os.environ:
return None
# if the env var is not set or empty, return None
env_var_value = os.environ[env_var_name]
if not env_var_value:
return None
return [float(x) for x in env_var_value.split(",")]
python/sglang/srt/observability/scheduler_metrics_mixin.py
+99 -104
View File
@@ -4,7 +4,6 @@ import dataclasses
import logging
import time
from collections import defaultdict
from contextlib import contextmanager
from typing import TYPE_CHECKING, List, Optional, Tuple, Union
from sglang.srt.disaggregation.kv_events import EventPublisherFactory, KVEventBatch
@@ -28,7 +27,7 @@ from sglang.srt.observability.metrics_collector import (
SchedulerStats,
compute_routing_key_stats,
)
from sglang.srt.utils.device_timer import DeviceTimer, GapTimer
from sglang.srt.utils.device_timer import DeviceTimer
from sglang.srt.utils.scheduler_status_logger import SchedulerStatusLogger
if TYPE_CHECKING:
@@ -74,16 +73,16 @@ class PrefillStats:
)
@dataclasses.dataclass
class KvMetrics:
def __init__(self):
self.request_active_slots = None
self.request_total_slots = None
self.kv_active_blocks = None
self.kv_total_blocks = None
self.num_requests_waiting = None
self.gpu_cache_usage_perc = None
self.gpu_prefix_cache_hit_rate = None
self.data_parallel_rank = None
request_active_slots: int = 0
request_total_slots: int = 0
kv_active_blocks: int = 0
kv_total_blocks: int = 0
num_requests_waiting: int = 0
gpu_cache_usage_perc: float = 0.0
gpu_prefix_cache_hit_rate: float = 0.0
data_parallel_rank: int = 0
class SchedulerMetricsMixin:
@@ -98,6 +97,12 @@ class SchedulerMetricsMixin:
self.last_gen_throughput: float = 0.0
self.last_input_throughput: float = 0.0
self.step_time_dict = defaultdict(list) # Dict[batch size -> step time]
self.stats = SchedulerStats()
self._graph_backend_label = {
"cpu": "cpu graph",
"npu": "npu graph",
"musa": "musa graph",
}.get(getattr(self, "device", ""), "cuda graph")
# Cumulative spec-decoding counters (reset every decode_log_interval).
# Each update adds (num_accepted_drafts + bs, bs).
@@ -111,15 +116,15 @@ class SchedulerMetricsMixin:
self.kv_transfer_speed_gb_s: float = 0.0
self.kv_transfer_latency_ms: float = 0.0
self.stats = SchedulerStats()
# Metrics
self.enable_mfu_metrics = False
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
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
@@ -145,20 +150,27 @@ class SchedulerMetricsMixin:
enable_streaming_session=self.server_args.enable_streaming_session,
server_args=self.server_args,
)
self.enable_mfu_metrics = bool(self.server_args.enable_mfu_metrics)
self.enable_mfu_metrics = self.server_args.enable_mfu_metrics
if self.enable_mfu_metrics:
self._init_estimated_perf_constants()
self._mfu_log_flops = 0.0
self._mfu_log_read_bytes = 0.0
self._mfu_log_write_bytes = 0.0
if ENABLE_METRICS_DEVICE_TIMER:
self.forward_pass_device_timer = DeviceTimer(
reporter=self.metrics_collector.increment_gpu_execution_seconds,
)
self.bubble_timer = GapTimer(
reporter=self.metrics_collector.increment_gpu_overlap_wait_seconds,
)
if ENABLE_METRICS_DEVICE_TIMER:
self._device_timer_window_batch_count = 0
self._device_timer_window_gpu_time = 0.0
self._device_timer_window_start = None
self.fwd_occupancy = float("nan")
def _wrap_execution_reporter(**kwargs):
self._device_timer_window_gpu_time += kwargs["t"]
if self.enable_metrics:
self.metrics_collector.increment_forward_execution_seconds(**kwargs)
self.forward_pass_device_timer = DeviceTimer(
reporter=_wrap_execution_reporter,
)
self.init_kv_events(self.server_args.kv_events_config)
@@ -369,16 +381,8 @@ class SchedulerMetricsMixin:
and self.server_args.encoder_transfer_backend == "zmq_to_scheduler"
):
msg += f"waiting-image-req: {len(self.mm_receiver.waiting_list)}, "
graph_backend = defaultdict(
lambda: "cuda graph",
{
"cpu": "cpu graph",
"npu": "npu graph",
"musa": "musa graph",
},
)
msg += f"{graph_backend[self.device]}: {can_run_cuda_graph}, "
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:
@@ -386,9 +390,11 @@ class SchedulerMetricsMixin:
tflops_per_s = flops / gap_latency / 1e12
msg += f", est. prefill TFLOPS/s (per GPU): {tflops_per_s:.2f}"
if ENABLE_METRICS_DEVICE_TIMER:
msg += f", fwd occupancy: {self.fwd_occupancy:.2f}%"
if self.is_stats_logging_rank:
logger.info(msg)
if self.current_scheduler_metrics_enabled:
self.metrics_collector.increment_prefill_cuda_graph_pass(
value=can_run_cuda_graph
@@ -408,24 +414,22 @@ class SchedulerMetricsMixin:
num_write_bytes_per_gpu=write_bytes,
)
# Basics
priority_enabled = self.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
)
# Basics
self.stats.num_running_reqs = prefill_stats.num_running_reqs
self.stats.num_running_reqs_offline_batch = 0
pool_stats.update_scheduler_stats(self.stats)
priority_enabled = self.enable_priority_scheduling
self.stats.num_queue_reqs = QueueCount.from_reqs(
self.waiting_queue, priority_enabled
)
self.stats.num_grammar_queue_reqs = len(self.grammar_manager)
self.stats.cache_hit_rate = cache_hit_rate
self.stats.max_total_num_tokens = self.max_total_num_tokens
# Memory pool usage ratios / Absolute token counts
pool_stats.update_scheduler_stats(self.stats)
# Retract
self.stats.num_retracted_reqs = self.num_retracted_reqs
@@ -450,7 +454,7 @@ class SchedulerMetricsMixin:
self.disagg_decode_transfer_queue.queue, priority_enabled
)
# Others
# Utilization / LoRA / HiCache
self.calculate_utilization()
self.update_lora_metrics()
self._log_hicache_stats()
@@ -505,7 +509,6 @@ class SchedulerMetricsMixin:
self.num_generated_tokens = 0
num_running_reqs = len(batch.reqs)
num_running_reqs_offline_batch = 0
pool_stats = self.get_pool_stats()
token_usage_msg = ", ".join(pool_stats.get_decode_usage_msg_parts()) + ", "
@@ -556,16 +559,8 @@ class SchedulerMetricsMixin:
):
msg += f"waiting-image-req: {len(self.mm_receiver.waiting_list)}, "
graph_backend = defaultdict(
lambda: "cuda graph",
{
"cpu": "cpu graph",
"npu": "npu graph",
"musa": "musa graph",
},
)
msg += (
f"{graph_backend[self.device]}: {can_run_cuda_graph}, "
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)}"
)
@@ -586,31 +581,32 @@ class SchedulerMetricsMixin:
self._mfu_log_read_bytes = 0.0
self._mfu_log_write_bytes = 0.0
if ENABLE_METRICS_DEVICE_TIMER:
msg += f", fwd occupancy: {self.fwd_occupancy:.2f}%"
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(
batch.reqs, priority_enabled
)
self.stats.num_running_reqs_offline_batch = num_running_reqs_offline_batch
pool_stats.update_scheduler_stats(self.stats)
self.stats.decode_sum_seq_lens = batch.seq_lens_cpu.sum().item()
self.stats.gen_throughput = self.last_gen_throughput
self.stats.num_queue_reqs = QueueCount.from_reqs(
self.waiting_queue, priority_enabled
)
self.stats.num_grammar_queue_reqs = len(self.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()
self.stats.max_total_num_tokens = self.max_total_num_tokens
self.stats.num_streaming_sessions = self._streaming_session_count()
self.stats.streaming_session_held_tokens = self._session_held_tokens()
# Memory pool usage ratios / Absolute token counts
pool_stats.update_scheduler_stats(self.stats)
# Speculative decoding
self.stats.spec_accept_rate = spec_accept_rate
self.stats.spec_accept_length = spec_accept_length
self.stats.spec_accept_rate = spec_accept_rate
# Retract
self.stats.num_retracted_reqs = self.num_retracted_reqs
@@ -632,17 +628,25 @@ class SchedulerMetricsMixin:
self.stats.num_decode_transfer_queue_reqs = QueueCount.from_reqs(
self.disagg_decode_transfer_queue.queue, priority_enabled
)
running_routing_keys = [r.routing_key for r in batch.reqs]
waiting_routing_keys = [r.routing_key for r in self.waiting_queue]
(
self.stats.num_unique_running_routing_keys,
self.stats.routing_key_running_req_counts,
) = compute_routing_key_stats(running_routing_keys)
_, self.stats.routing_key_all_req_counts = compute_routing_key_stats(
running_routing_keys + waiting_routing_keys
)
# Others
# Streaming session metrics
self.stats.num_streaming_sessions = self._streaming_session_count()
self.stats.streaming_session_held_tokens = self._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]
(
self.stats.num_unique_running_routing_keys,
self.stats.routing_key_running_req_counts,
) = compute_routing_key_stats(running_routing_keys)
_, self.stats.routing_key_all_req_counts = compute_routing_key_stats(
running_routing_keys + waiting_routing_keys
)
# Utilization / LoRA / HiCache
self.calculate_utilization()
self.update_lora_metrics()
self._log_hicache_stats()
@@ -759,13 +763,10 @@ class SchedulerMetricsMixin:
if self.disaggregation_mode == DisaggregationMode.PREFILL:
self.stats.utilization = -1
else:
if (
self.stats.max_running_requests_under_SLO is not None
and self.stats.max_running_requests_under_SLO > 0
):
max_under_slo = getattr(self, "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
/ self.stats.max_running_requests_under_SLO,
self.stats.num_running_reqs.total / max_under_slo,
self.stats.token_usage / 0.9,
)
@@ -918,36 +919,30 @@ class SchedulerMetricsMixin:
queues=queues,
)
@contextmanager
def record_forward_metrics(self: Scheduler, batch: ScheduleBatch):
if not (self.enable_metrics and ENABLE_METRICS_DEVICE_TIMER):
yield
def update_device_timer(self: Scheduler):
if not ENABLE_METRICS_DEVICE_TIMER:
return
category = "forward_" + batch.forward_mode.name.lower()
with self.forward_pass_device_timer.wrap(
metadata=dict(
category=category,
dp_cooperation_info=batch.dp_cooperation_info,
),
self.forward_pass_device_timer._report()
now = time.perf_counter()
if self._device_timer_window_batch_count == 0:
self._device_timer_window_start = now
self._device_timer_window_gpu_time = 0.0
cpu_time = 0
self.fwd_occupancy = float("nan")
else:
cpu_time = now - self._device_timer_window_start
self.fwd_occupancy = min(
self._device_timer_window_gpu_time / cpu_time * 100, 100
)
# ratio = self._device_timer_window_gpu_time / cpu_time if cpu_time > 0 else float("nan")
# print(f"{self._device_timer_window_batch_count=} {self.fwd_occupancy=}, {self._device_timer_window_gpu_time=}, {cpu_time=}, {ratio=}")
self._device_timer_window_batch_count += 1
if (
self._device_timer_window_batch_count
>= self.server_args.decode_log_interval
):
yield
self._device_timer_window_batch_count = 0
@contextmanager
def record_bubble_metrics(self: Scheduler, batch: ScheduleBatch):
if not (self.enable_metrics and ENABLE_METRICS_DEVICE_TIMER):
yield
return
category = "forward_" + batch.forward_mode.name.lower()
with self.bubble_timer.wrap(
metadata=dict(
category=category,
dp_cooperation_info=batch.dp_cooperation_info,
),
):
yield
def cancel_bubble_timer(self: Scheduler):
if self.enable_metrics and ENABLE_METRICS_DEVICE_TIMER:
self.bubble_timer.cancel()
def reset_device_timer_window(self: Scheduler):
if ENABLE_METRICS_DEVICE_TIMER:
self._device_timer_window_batch_count = 0
python/sglang/srt/speculative/eagle_draft_cuda_graph_runner.py
+8 -1
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import bisect
import contextlib
from dataclasses import dataclass
from typing import TYPE_CHECKING, Callable, Optional
@@ -216,7 +217,13 @@ class EAGLEDraftCudaGraphRunner:
return out
def _replay(self, forward_batch: ForwardBatch):
self.graphs[self.bs].replay()
ctx = (
self.model_runner.device_timer.wrap(metadata={"category": "eagle_draft"})
if self.model_runner.device_timer
else contextlib.nullcontext()
)
with ctx:
self.graphs[self.bs].replay()
def capture(self):
CudaGraphRunner.capture(self)
python/sglang/srt/speculative/eagle_draft_extend_cuda_graph_runner.py
+10 -1
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import bisect
import contextlib
from dataclasses import dataclass
from typing import TYPE_CHECKING, Callable, Optional
@@ -282,7 +283,15 @@ class EAGLEDraftExtendCudaGraphRunner:
return out
def _replay(self, forward_batch: ForwardBatch):
self.graphs[self.bs].replay()
ctx = (
self.model_runner.device_timer.wrap(
metadata={"category": "eagle_draft_extend"}
)
if self.model_runner.device_timer
else contextlib.nullcontext()
)
with ctx:
self.graphs[self.bs].replay()
def capture(self):
CudaGraphRunner.capture(self)
python/sglang/srt/utils/device_timer.py
+1
View File
@@ -28,6 +28,7 @@ class DeviceTimer:
self._intervals.popleft()
self._reporter(t=interval.elapsed_time() / 1000.0, **interval.metadata)
# print(f"{interval.elapsed_time()=:.6f}, {interval.metadata=}")
class GapTimer(DeviceTimer):
python/sglang/srt/utils/request_logger.py
-8
View File
@@ -15,11 +15,9 @@ from __future__ import annotations
import dataclasses
import logging
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Union
from sglang.srt.environ import envs
from sglang.srt.utils.common import get_bool_env_var
from sglang.srt.utils.log_utils import create_log_targets, log_json
if TYPE_CHECKING:
@@ -237,12 +235,6 @@ class RequestLogger:
target.info(msg)
# TODO remove this?
@lru_cache(maxsize=2)
def disable_request_logging() -> bool:
return get_bool_env_var("SGLANG_DISABLE_REQUEST_LOGGING")
# TODO unify this w/ `_transform_data_for_logging` if we find performance enough
def _dataclass_to_string_truncated(
data: Any, max_length: int = 2048, skip_names: Optional[Set[str]] = None