From 1bd4f94598a621cf5e8c27686311e92134e9edb0 Mon Sep 17 00:00:00 2001 From: Liangsheng Yin Date: Wed, 20 May 2026 03:34:37 -0700 Subject: [PATCH] [Test] Add fwd_occupancy sanity kit (#25886) --- python/sglang/test/kits/fwd_occupancy_kit.py | 209 ++++++++++++++++++ python/sglang/test/kits/hellaswag_kit.py | 29 +++ python/sglang/test/test_programs.py | 4 +- test/registered/core/test_basic_sanity.py | 39 ++-- .../core/test_basic_sanity_eagle3.py | 79 +++++++ 5 files changed, 333 insertions(+), 27 deletions(-) create mode 100644 python/sglang/test/kits/fwd_occupancy_kit.py create mode 100644 python/sglang/test/kits/hellaswag_kit.py create mode 100644 test/registered/core/test_basic_sanity_eagle3.py diff --git a/python/sglang/test/kits/fwd_occupancy_kit.py b/python/sglang/test/kits/fwd_occupancy_kit.py new file mode 100644 index 000000000..0f66eaed5 --- /dev/null +++ b/python/sglang/test/kits/fwd_occupancy_kit.py @@ -0,0 +1,209 @@ +"""Single-batch decode GPU occupancy sanity kit. + +Probes ``sglang:fwd_occupancy`` (a 0-100 percentage averaged over the +last ``decode_log_interval`` batches; resets to NaN at window +boundaries) under one long single-batch ``/generate`` request, and +asserts median above a threshold. Single-batch is where CPU overhead +dominates -- overlap scheduler / cuda graph regressions surface here +before batched throughput moves. + +Prerequisites on the consuming server: + env: SGLANG_ENABLE_METRICS_DEVICE_TIMER=1 + server flag: --enable-metrics + +Mix into a ``CustomTestCase`` subclass exposing ``self.base_url``. +""" + +import re +import statistics +import threading +import time + +import requests +import tabulate + +_FWD_OCCUPANCY_RE = re.compile( + r"^sglang:fwd_occupancy(?:\{[^}]*\})?\s+(\S+)", re.MULTILINE +) +_GENERATE_REQUEST_TIMEOUT = 600 +_METRICS_REQUEST_TIMEOUT = 10 + + +class FwdOccupancyMixin: + """Assert single-batch ``sglang:fwd_occupancy`` median > threshold.""" + + fwd_occupancy_threshold: float = 95.0 + fwd_occupancy_min_samples: int = 5 + fwd_occupancy_scrape_interval: float = 0.5 + + # Spec-decoding accept-length floor. Only enforced when the server + # is running with a spec algorithm (avg_spec_accept_length present + # in /server_info); silently skipped otherwise. EAGLE3 3/1/4 on + # 5090 + Llama-3.1-8B measured ~2.0 in CI; 1.8 leaves a small + # buffer while still catching silent fallback to vanilla (~1.0). + spec_accept_length_threshold: float = 1.8 + + # Warmup: one short request to fill cuda graphs + get the + # device-timer past its first NaN window. + fwd_occupancy_warmup_max_new_tokens: int = 64 + fwd_occupancy_warmup_settle_seconds: float = 1.0 + + # Measurement: one long single-batch request -- max_new_tokens must + # span several decode_log_interval windows for enough samples. + fwd_occupancy_max_new_tokens: int = 2048 + fwd_occupancy_prompt: str = "Write a long, detailed, multi-paragraph story about " + + def _scrape_fwd_occupancy(self): + """Max non-NaN gauge value across exposed labels (e.g. dp ranks); + None on transient scrape failure.""" + try: + resp = requests.get( + self.base_url + "/metrics", timeout=_METRICS_REQUEST_TIMEOUT + ) + except requests.RequestException: + return None + if resp.status_code != 200: + return None + vals = [] + for raw in _FWD_OCCUPANCY_RE.findall(resp.text): + try: + v = float(raw) + except ValueError: + continue + if v == v: # NaN filter (gauge resets to NaN on window boundary) + vals.append(v) + return max(vals) if vals else None + + def _assert_metrics_device_timer_enabled(self): + """Fail loudly on missing flag/env -- otherwise a NaN-only gauge + looks like a real occupancy regression.""" + resp = requests.get( + self.base_url + "/metrics", timeout=_METRICS_REQUEST_TIMEOUT + ) + assert resp.status_code == 200, ( + f"/metrics returned {resp.status_code}; the test class's server " + "must be launched with --enable-metrics" + ) + assert "sglang:fwd_occupancy" in resp.text, ( + "sglang:fwd_occupancy gauge not exposed; set " + "SGLANG_ENABLE_METRICS_DEVICE_TIMER=1 in the server's env and " + "pass --enable-metrics" + ) + + def _fwd_occupancy_fire(self, prompt: str, max_new_tokens: int): + """Fire one /generate. Must not be called concurrently -- that + would break the single-batch invariant.""" + try: + requests.post( + self.base_url + "/generate", + json={ + "text": prompt, + "sampling_params": { + "temperature": 0.0, + "max_new_tokens": max_new_tokens, + }, + }, + timeout=_GENERATE_REQUEST_TIMEOUT, + ) + except requests.RequestException: + # Final stats-vs-threshold is the signal; individual fire + # failure isn't. + pass + + def _fwd_occupancy_warmup(self): + """Fill cuda graphs + step the device-timer past its first NaN + window before measurement starts.""" + self._fwd_occupancy_fire( + "warmup " + self.fwd_occupancy_prompt, + self.fwd_occupancy_warmup_max_new_tokens, + ) + time.sleep(self.fwd_occupancy_warmup_settle_seconds) + + def _fwd_occupancy_measure(self): + """Background-fire one long single-batch request, scrape + /metrics on the foreground; return non-NaN samples.""" + samples = [] + request_done = threading.Event() + + def fire_one(): + try: + self._fwd_occupancy_fire( + self.fwd_occupancy_prompt, + self.fwd_occupancy_max_new_tokens, + ) + finally: + request_done.set() + + firer = threading.Thread(target=fire_one, daemon=True) + firer.start() + + while not request_done.is_set(): + v = self._scrape_fwd_occupancy() + if v is not None: + samples.append(v) + time.sleep(self.fwd_occupancy_scrape_interval) + + firer.join(timeout=_GENERATE_REQUEST_TIMEOUT) + return samples + + def test_fwd_occupancy(self): + self._assert_metrics_device_timer_enabled() + self._fwd_occupancy_warmup() + samples = self._fwd_occupancy_measure() + + self.assertGreaterEqual( + len(samples), + self.fwd_occupancy_min_samples, + f"only {len(samples)} non-NaN occupancy samples collected " + f"(need >= {self.fwd_occupancy_min_samples}); the measurement " + "window may be too short or the gauge stuck at NaN", + ) + + # Median is the steady-state signal; peak / p10 included in the + # assertion message for triage. + samples_sorted = sorted(samples) + median = statistics.median(samples_sorted) + peak = samples_sorted[-1] + p10 = samples_sorted[max(0, len(samples_sorted) // 10 - 1)] + print( + "\n" + + tabulate.tabulate( + [ + ["samples (n)", len(samples)], + ["median", f"{median:.2f}"], + ["peak", f"{peak:.2f}"], + ["p10", f"{p10:.2f}"], + ["threshold", f"{self.fwd_occupancy_threshold:.2f}"], + ], + headers=["fwd_occupancy", "value"], + tablefmt="github", + ) + ) + + self.assertGreater( + median, + self.fwd_occupancy_threshold, + f"sglang:fwd_occupancy median={median:.2f} did not exceed " + f"threshold {self.fwd_occupancy_threshold} " + f"(peak={peak:.2f}, p10={p10:.2f}, n={len(samples)})", + ) + + # The 2048-token decode above populates the spec running average + # if a spec algorithm is enabled; absent otherwise (vanilla + # decode skips this check). + try: + info = requests.get( + self.base_url + "/server_info", timeout=_METRICS_REQUEST_TIMEOUT + ).json() + avg_accept = info["internal_states"][0].get("avg_spec_accept_length") + except (requests.RequestException, KeyError, IndexError): + avg_accept = None + if avg_accept is not None: + print(f"avg_spec_accept_length = {avg_accept:.3f}") + self.assertGreater( + avg_accept, + self.spec_accept_length_threshold, + f"avg_spec_accept_length={avg_accept:.3f} did not exceed " + f"threshold {self.spec_accept_length_threshold} -- spec " + "barely accepted, possibly degraded to vanilla decode", + ) diff --git a/python/sglang/test/kits/hellaswag_kit.py b/python/sglang/test/kits/hellaswag_kit.py new file mode 100644 index 000000000..cc79533f4 --- /dev/null +++ b/python/sglang/test/kits/hellaswag_kit.py @@ -0,0 +1,29 @@ +"""Hellaswag sanity kit. + +Runs hellaswag via the sgl frontend DSL bound to ``self.base_url`` and +asserts accuracy above a threshold. Catches systematic regressions that +pass every cheap single-prompt probe but tank multi-choice reasoning. + +Mix into a ``CustomTestCase`` subclass exposing ``self.base_url``. +""" + + +class HellaswagMixin: + """Assert hellaswag accuracy > threshold.""" + + hellaswag_accuracy_threshold: float = 0.60 + + def test_accuracy_floor(self): + import sglang as sgl + from sglang.test.test_programs import test_hellaswag_select + + sgl.set_default_backend(sgl.RuntimeEndpoint(self.base_url)) + try: + accuracy, _ = test_hellaswag_select() + finally: + sgl.set_default_backend(None) + self.assertGreater( + accuracy, + self.hellaswag_accuracy_threshold, + f"hellaswag accuracy floor breached: {accuracy:.3f}", + ) diff --git a/python/sglang/test/test_programs.py b/python/sglang/test/test_programs.py index 8ff8f9d68..6779814a0 100644 --- a/python/sglang/test/test_programs.py +++ b/python/sglang/test/test_programs.py @@ -582,9 +582,9 @@ def test_hellaswag_select(): # Compute accuracy accuracy_gen = np.mean(np.array(preds_gen) == np.array(labels)) - print(f"{accuracy=}, {accuracy_gen=}") + print(f"{accuracy=}, {accuracy_gen=} {latency=:.2f}s {latency_gen=:.2f}s") assert np.abs(accuracy_gen - accuracy) < 0.1 - assert np.abs(latency_gen - latency) < 1 if not _is_hip else 2 + # No latency assert: the 2nd run hits the radix cache the 1st filled. return accuracy, latency diff --git a/test/registered/core/test_basic_sanity.py b/test/registered/core/test_basic_sanity.py index a6d77cdd1..1b58ed833 100644 --- a/test/registered/core/test_basic_sanity.py +++ b/test/registered/core/test_basic_sanity.py @@ -1,6 +1,7 @@ -"""Basic sanity: small-but-broad server smoke that downstream stages -depend on. Three sanity kits, one shared server, covering protocol -contract, decode correctness, and scheduler stress paths.""" +"""Stage-a basic sanity: small-but-broad server smoke that downstream +stages depend on. Multiple sanity-kit mixins driving one shared server, +covering protocol, decode correctness, scheduler stress, occupancy, and +hellaswag accuracy.""" import unittest @@ -9,6 +10,8 @@ from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci from sglang.test.kits.basic_api_contract_kit import BasicAPIContractMixin from sglang.test.kits.basic_decode_correctness_kit import BasicDecodeCorrectnessMixin from sglang.test.kits.basic_scheduler_stress_kit import BasicSchedulerStressMixin +from sglang.test.kits.fwd_occupancy_kit import FwdOccupancyMixin +from sglang.test.kits.hellaswag_kit import HellaswagMixin from sglang.test.test_utils import ( DEFAULT_MODEL_NAME_FOR_TEST, DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, @@ -17,17 +20,22 @@ from sglang.test.test_utils import ( popen_launch_server, ) -register_cuda_ci(est_time=120, stage="base-a", runner_config="1-gpu-small") -register_amd_ci(est_time=120, suite="stage-a-test-1-gpu-small-amd") +register_cuda_ci(est_time=160, stage="base-a", runner_config="1-gpu-small") +register_amd_ci(est_time=160, suite="stage-a-test-1-gpu-small-amd") class TestBasicSanity( BasicAPIContractMixin, BasicDecodeCorrectnessMixin, BasicSchedulerStressMixin, + FwdOccupancyMixin, + HellaswagMixin, CustomTestCase, ): served_model_name = DEFAULT_MODEL_NAME_FOR_TEST + # 5090 + Llama-3.1-8B single-batch decode with overlap scheduler + + # cuda graph measured ~99 median in CI; keep ~2pp headroom. + fwd_occupancy_threshold = 97.0 @classmethod def setUpClass(cls): @@ -43,32 +51,13 @@ class TestBasicSanity( "0.7", "--enable-metrics", ], + env={"SGLANG_ENABLE_METRICS_DEVICE_TIMER": "1"}, ) @classmethod def tearDownClass(cls): kill_process_tree(cls.process.pid) - def test_accuracy_floor(self): - # Stage-a-private accuracy guard: hellaswag via the frontend DSL - # bound to this server. Catches systematic regressions that pass - # every cheap probe in the mixed-in kits but tank multi-choice - # reasoning. Not part of any reusable mixin -- accuracy gating - # is the gate test's own responsibility. - import sglang as sgl - from sglang.test.test_programs import test_hellaswag_select - - sgl.set_default_backend(sgl.RuntimeEndpoint(self.base_url)) - try: - accuracy, _ = test_hellaswag_select() - finally: - sgl.set_default_backend(None) - self.assertGreater( - accuracy, - 0.60, - f"hellaswag accuracy floor breached: {accuracy:.3f}", - ) - if __name__ == "__main__": unittest.main() diff --git a/test/registered/core/test_basic_sanity_eagle3.py b/test/registered/core/test_basic_sanity_eagle3.py new file mode 100644 index 000000000..17f9f8784 --- /dev/null +++ b/test/registered/core/test_basic_sanity_eagle3.py @@ -0,0 +1,79 @@ +"""Stage-a basic sanity with EAGLE3 spec decoding enabled. Mirrors +test_basic_sanity.py with the spec-decoding path active.""" + +import unittest + +from sglang.srt.utils import kill_process_tree +from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci +from sglang.test.kits.basic_api_contract_kit import BasicAPIContractMixin +from sglang.test.kits.basic_decode_correctness_kit import BasicDecodeCorrectnessMixin +from sglang.test.kits.basic_scheduler_stress_kit import BasicSchedulerStressMixin +from sglang.test.kits.fwd_occupancy_kit import FwdOccupancyMixin +from sglang.test.kits.hellaswag_kit import HellaswagMixin +from sglang.test.test_utils import ( + DEFAULT_DRAFT_MODEL_EAGLE3, + DEFAULT_TARGET_MODEL_EAGLE3, + DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + DEFAULT_URL_FOR_TEST, + CustomTestCase, + popen_launch_server, +) + +register_cuda_ci(est_time=200, stage="base-a", runner_config="1-gpu-small") +register_amd_ci(est_time=200, suite="stage-a-test-1-gpu-small-amd") + + +class TestBasicSanityEagle3( + BasicAPIContractMixin, + BasicDecodeCorrectnessMixin, + BasicSchedulerStressMixin, + FwdOccupancyMixin, + HellaswagMixin, + CustomTestCase, +): + served_model_name = DEFAULT_TARGET_MODEL_EAGLE3 + # Match vanilla gate at 97; EAGLE3 spec should sustain similar + # single-batch occupancy. Adjust per CI calibration. + fwd_occupancy_threshold = 97.0 + + @classmethod + def setUpClass(cls): + cls.base_url = DEFAULT_URL_FOR_TEST + cls.process = popen_launch_server( + DEFAULT_TARGET_MODEL_EAGLE3, + cls.base_url, + timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + other_args=[ + # Canonical EAGLE3 sglang config: fp16 + triton attention. + # bf16 + flashinfer cutlass RMSNorm hits a SM120 dtype + # mismatch on the draft model's input_layernorm. + "--dtype", + "float16", + "--attention-backend", + "triton", + "--speculative-algorithm", + "EAGLE3", + "--speculative-draft-model-path", + DEFAULT_DRAFT_MODEL_EAGLE3, + "--speculative-num-steps", + "3", + "--speculative-eagle-topk", + "1", + "--speculative-num-draft-tokens", + "4", + "--cuda-graph-max-bs", + "4", + "--mem-fraction-static", + "0.7", + "--enable-metrics", + ], + env={"SGLANG_ENABLE_METRICS_DEVICE_TIMER": "1"}, + ) + + @classmethod + def tearDownClass(cls): + kill_process_tree(cls.process.pid) + + +if __name__ == "__main__": + unittest.main()