diff --git a/docs/references/environment_variables.md b/docs/references/environment_variables.md index 0352dfe3b..73a11be23 100644 --- a/docs/references/environment_variables.md +++ b/docs/references/environment_variables.md @@ -12,7 +12,6 @@ SGLang supports various environment variables that can be used to configure its | `SGLANG_HOST_IP` | Host IP address for the server | `0.0.0.0` | | `SGLANG_PORT` | Port for the server | auto-detected | | `SGLANG_LOGGING_CONFIG_PATH` | Custom logging configuration path | Not set | -| `SGLANG_DISABLE_REQUEST_LOGGING` | Disable request logging | `false` | | `SGLANG_LOG_REQUEST_HEADERS` | Comma-separated list of additional HTTP headers to log when `--log-requests` is enabled. Appends to the default `x-smg-routing-key`. | Not set | | `SGLANG_HEALTH_CHECK_TIMEOUT` | Timeout for health check in seconds | `20` | | `SGLANG_EPLB_HEATMAP_COLLECTION_INTERVAL` | The interval of passes to collect the metric of selected count of physical experts on each layer and GPU rank. 0 means disabled. | `0` | diff --git a/python/sglang/srt/environ.py b/python/sglang/srt/environ.py index f8dbcca85..4c201d8f8 100644 --- a/python/sglang/srt/environ.py +++ b/python/sglang/srt/environ.py @@ -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) diff --git a/python/sglang/srt/layers/quantization/fp8_utils.py b/python/sglang/srt/layers/quantization/fp8_utils.py index d4073c29b..b39da5267 100755 --- a/python/sglang/srt/layers/quantization/fp8_utils.py +++ b/python/sglang/srt/layers/quantization/fp8_utils.py @@ -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 diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index 31d3aa022..2c3d98647 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -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: diff --git a/python/sglang/srt/managers/scheduler_runtime_checker_mixin.py b/python/sglang/srt/managers/scheduler_runtime_checker_mixin.py index a88473be6..72b01cfd8 100644 --- a/python/sglang/srt/managers/scheduler_runtime_checker_mixin.py +++ b/python/sglang/srt/managers/scheduler_runtime_checker_mixin.py @@ -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 ( diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index e2830217c..8aa79b6b4 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -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, ) diff --git a/python/sglang/srt/model_executor/cuda_graph_runner.py b/python/sglang/srt/model_executor/cuda_graph_runner.py index 5dcc968e9..05b64778a 100644 --- a/python/sglang/srt/model_executor/cuda_graph_runner.py +++ b/python/sglang/srt/model_executor/cuda_graph_runner.py @@ -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): diff --git a/python/sglang/srt/model_executor/model_runner.py b/python/sglang/srt/model_executor/model_runner.py index 005754b71..a833dac43 100644 --- a/python/sglang/srt/model_executor/model_runner.py +++ b/python/sglang/srt/model_executor/model_runner.py @@ -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[ 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}]" diff --git a/python/sglang/srt/observability/metrics_collector.py b/python/sglang/srt/observability/metrics_collector.py index 54472a2fe..3e149c0f0 100644 --- a/python/sglang/srt/observability/metrics_collector.py +++ b/python/sglang/srt/observability/metrics_collector.py @@ -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(",")] diff --git a/python/sglang/srt/observability/scheduler_metrics_mixin.py b/python/sglang/srt/observability/scheduler_metrics_mixin.py index d9d9e3418..78925a440 100644 --- a/python/sglang/srt/observability/scheduler_metrics_mixin.py +++ b/python/sglang/srt/observability/scheduler_metrics_mixin.py @@ -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 diff --git a/python/sglang/srt/speculative/eagle_draft_cuda_graph_runner.py b/python/sglang/srt/speculative/eagle_draft_cuda_graph_runner.py index a7e56eee7..baab551b8 100644 --- a/python/sglang/srt/speculative/eagle_draft_cuda_graph_runner.py +++ b/python/sglang/srt/speculative/eagle_draft_cuda_graph_runner.py @@ -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) diff --git a/python/sglang/srt/speculative/eagle_draft_extend_cuda_graph_runner.py b/python/sglang/srt/speculative/eagle_draft_extend_cuda_graph_runner.py index bccd0a09d..a5ae5b5b3 100644 --- a/python/sglang/srt/speculative/eagle_draft_extend_cuda_graph_runner.py +++ b/python/sglang/srt/speculative/eagle_draft_extend_cuda_graph_runner.py @@ -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) diff --git a/python/sglang/srt/utils/device_timer.py b/python/sglang/srt/utils/device_timer.py index 4cd5817ad..eaa44b7f4 100644 --- a/python/sglang/srt/utils/device_timer.py +++ b/python/sglang/srt/utils/device_timer.py @@ -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): diff --git a/python/sglang/srt/utils/request_logger.py b/python/sglang/srt/utils/request_logger.py index 4ce4585ad..75baba4f9 100644 --- a/python/sglang/srt/utils/request_logger.py +++ b/python/sglang/srt/utils/request_logger.py @@ -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 diff --git a/test/registered/observability/test_metrics.py b/test/registered/observability/test_metrics.py index 2b6c65987..9d9f9d056 100644 --- a/test/registered/observability/test_metrics.py +++ b/test/registered/observability/test_metrics.py @@ -62,12 +62,12 @@ class TestEnableMetrics(CustomTestCase): {"mode": "decode"}, ), ( - "sglang:dp_cooperation_gpu_execution_seconds_total", - {"category": "forward_extend"}, + "sglang:dp_cooperation_forward_execution_seconds_total", + {"category": "extend"}, ), ( - "sglang:dp_cooperation_gpu_execution_seconds_total", - {"category": "forward_decode"}, + "sglang:dp_cooperation_forward_execution_seconds_total", + {"category": "decode"}, ), ] _check_metrics_positive(self, metrics, metrics_to_check) @@ -129,15 +129,17 @@ class TestEnableMetrics(CustomTestCase): for _ in response.iter_lines(decode_unicode=False): pass - response = requests.post( - f"{DEFAULT_URL_FOR_TEST}/generate", - json={ - "text": "Hello", - "sampling_params": {"temperature": 0, "max_new_tokens": 5}, - }, - headers={"x-smg-routing-key": "test-key"}, - ) - self.assertEqual(response.status_code, 200) + for i in range(2): + # Send the request twice to trigger cached token metrics + response = requests.post( + f"{DEFAULT_URL_FOR_TEST}/generate", + json={ + "text": "Hello, " * 100, + "sampling_params": {"temperature": 0, "max_new_tokens": 5}, + }, + headers={"x-smg-routing-key": "test-key"}, + ) + self.assertEqual(response.status_code, 200) # Get metrics metrics_response = requests.get(f"{DEFAULT_URL_FOR_TEST}/metrics") @@ -209,8 +211,8 @@ class TestEnableMetrics(CustomTestCase): metrics_to_check = [ ("sglang:realtime_tokens_total", {"mode": "prefill_compute"}), ("sglang:realtime_tokens_total", {"mode": "decode"}), - ("sglang:gpu_execution_seconds_total", {"category": "forward_extend"}), - ("sglang:gpu_execution_seconds_total", {"category": "forward_decode"}), + ("sglang:forward_execution_seconds_total", {"category": "extend"}), + ("sglang:forward_execution_seconds_total", {"category": "decode"}), ("sglang:process_cpu_seconds_total", {"component": "tokenizer"}), ] _check_metrics_positive(self, metrics, metrics_to_check) diff --git a/test/registered/perf/test_bench_one_batch_1gpu.py b/test/registered/perf/test_bench_one_batch_1gpu.py index 13f16dc92..fcd7dc6aa 100644 --- a/test/registered/perf/test_bench_one_batch_1gpu.py +++ b/test/registered/perf/test_bench_one_batch_1gpu.py @@ -1,12 +1,17 @@ +import os +import re +import subprocess import unittest +import numpy as np + from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci from sglang.test.test_utils import ( DEFAULT_MODEL_NAME_FOR_TEST, DEFAULT_SMALL_MODEL_NAME_FOR_TEST, CustomTestCase, is_in_ci, - run_bench_offline_throughput, + kill_process_tree, run_bench_one_batch, write_github_step_summary, ) @@ -24,10 +29,46 @@ class TestBenchOneBatch1GPU(CustomTestCase): self.assertGreater(output_throughput, 50) def test_bs1_default(self): - output_throughput = run_bench_offline_throughput( - DEFAULT_MODEL_NAME_FOR_TEST, ["--cuda-graph-max-bs", "2"] + env = os.environ.copy() + env["SGLANG_ENABLE_METRICS_DEVICE_TIMER"] = "1" + + command = [ + "python3", + "-m", + "sglang.bench_offline_throughput", + "--num-prompts", + "1", + "--dataset-name", + "random", + "--random-input-len", + "256", + "--random-output-len", + "1024", + "--model-path", + DEFAULT_MODEL_NAME_FOR_TEST, + "--cuda-graph-max-bs", + "2", + ] + + print(f"command={' '.join(command)}") + process = subprocess.Popen( + command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env ) + try: + stdout, stderr = process.communicate() + output = stdout.decode(errors="backslashreplace") + error = stderr.decode(errors="backslashreplace") + print(f"Output: {output}", flush=True) + print(f"Error: {error}", flush=True) + + output_throughput = -1 + for line in output.split("\n"): + if "Last generation throughput (tok/s):" in line: + output_throughput = float(line.split(":")[-1]) + finally: + kill_process_tree(process.pid) + if is_in_ci(): write_github_step_summary( f"### test_bs1_default (llama-3.1-8b)\n" @@ -35,6 +76,23 @@ class TestBenchOneBatch1GPU(CustomTestCase): ) self.assertGreater(output_throughput, 135) + fwd_occupancy_values = [] + for line in error.split("\n"): + match = re.search(r"fwd occupancy:\s*([\d.]+|nan)%", line) + if match: + val = match.group(1) + if val != "nan": + fwd_occupancy_values.append(float(val)) + + print(f"{fwd_occupancy_values=}", flush=True) + self.assertGreater( + len(fwd_occupancy_values), 0, "No fwd occupancy values found in logs" + ) + + fwd_occupancy_p90 = float(np.percentile(fwd_occupancy_values, 90)) + print(f"{fwd_occupancy_p90=}", flush=True) + self.assertGreater(fwd_occupancy_p90, 97.5) + if __name__ == "__main__": unittest.main()